package kg.apc.jmeter.perfmon;
import kg.apc.jmeter.JMeterPluginsUtils;
import kg.apc.jmeter.reporters.LoadosophiaUploadingNotifier;
import kg.apc.jmeter.vizualizers.CorrectedResultCollector;
import kg.apc.perfmon.PerfMonMetricGetter;
import kg.apc.perfmon.client.Transport;
import kg.apc.perfmon.client.TransportFactory;
import kg.apc.perfmon.metrics.MetricParams;
import org.apache.jmeter.samplers.SampleEvent;
import org.apache.jmeter.samplers.SampleSaveConfiguration;
import org.apache.jmeter.testelement.property.CollectionProperty;
import org.apache.jmeter.testelement.property.JMeterProperty;
import org.apache.jmeter.util.JMeterUtils;
import org.apache.jorphan.logging.LoggingManager;
import org.apache.log.Logger;
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
public class PerfMonCollector
extends CorrectedResultCollector
implements Runnable, PerfMonSampleGenerator {
private static boolean autoGenerateFiles = false;
public static final long MEGABYTE = 1024L * 1024L;
private static final String PERFMON = "PerfMon";
private static final Logger log = LoggingManager.getLoggerForClass();
public static final String DATA_PROPERTY = "metricConnections";
private int interval;
private Thread workerThread = null;
private Map<Object, PerfMonAgentConnector> connectors = new ConcurrentHashMap<Object, PerfMonAgentConnector>();
private HashMap<String, Long> oldValues = new HashMap<String, Long>();
private static String autoFileBaseName = null;
private static int counter = 0;
private LoadosophiaUploadingNotifier perfMonNotifier = LoadosophiaUploadingNotifier.getInstance();
private static String workerHost = null;
static {
autoGenerateFiles = (JMeterUtils.getPropDefault("forcePerfmonFile", "false")).trim().equalsIgnoreCase("true");
}
private static synchronized String getAutoFileName() {
String ret = "";
counter++;
if (autoFileBaseName == null) {
Calendar now = Calendar.getInstance();
SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd-HHmmss");
autoFileBaseName = "perfMon_" + formatter.format(now.getTime());
}
ret = ret + autoFileBaseName;
if (counter > 1) {
ret = ret + "_" + counter;
}
ret = ret + ".csv";
return ret;
}
public PerfMonCollector() {
// TODO: document it
interval = JMeterUtils.getPropDefault("jmeterPlugin.perfmon.interval", 1000);
}
public void setData(CollectionProperty rows) {
setProperty(rows);
}
public JMeterProperty getMetricSettings() {
return getProperty(DATA_PROPERTY);
}
public void sampleOccurred(SampleEvent event) {
// just dropping regular test samples
}
@Override
public synchronized void run() {
while (true) {
processConnectors();
try {
this.wait(interval);
} catch (InterruptedException ex) {
log.debug("Monitoring thread was interrupted", ex);
break;
}
}
}
//ensure we start only on one host (if multiple slaves)
private synchronized static boolean isWorkingHost(String host) {
if (workerHost == null) {
workerHost = host;
return true;
} else {
return host.equals(workerHost);
}
}
@Override
public void testStarted(String host) {
if (!isWorkingHost(host)) {
return;
}
//ensure the data will be saved
if (getProperty(FILENAME) == null || getProperty(FILENAME).getStringValue().trim().length() == 0) {
if (autoGenerateFiles) {
setupSaving(getAutoFileName());
} else {
try {
File tmpFile = File.createTempFile("perfmon_", ".jtl");
tmpFile.delete(); // required to have CSV header
setupSaving(tmpFile.getAbsolutePath());
} catch (IOException ex) {
log.info("PerfMon metrics will not be recorded! Please run the test with -JforcePerfmonFile=true", ex);
}
}
}
log.debug("PerfMon metrics will be stored in " + getPropertyAsString(FILENAME));
if (!getSaveConfig().saveAsXml() && getSaveConfig().saveFieldNames()) {
perfMonNotifier.addFile(getPropertyAsString(FILENAME));
} else {
log.warn("Perfmon file saving setting is not CSV with header line, cannot upload it to Loadosophia.org: " + getPropertyAsString(FILENAME));
}
initiateConnectors();
workerThread = new Thread(this);
workerThread.start();
super.testStarted(host);
}
private void setupSaving(String fileName) {
SampleSaveConfiguration config = getSaveConfig();
JMeterPluginsUtils.doBestCSVSetup(config);
setSaveConfig(config);
setFilename(fileName);
log.info("PerfMon metrics will be stored in " + new File(fileName).getAbsolutePath());
}
@Override
public void testEnded(String host) {
if (workerThread == null) {
return;
}
workerHost = null;
workerThread.interrupt();
shutdownConnectors();
//reset autoFileName for next test run
autoFileBaseName = null;
counter = 0;
super.testEnded(host);
}
private void initiateConnectors() {
oldValues.clear();
JMeterProperty prop = getMetricSettings();
connectors.clear();
if (!(prop instanceof CollectionProperty)) {
log.warn("Got unexpected property: " + prop);
return;
}
CollectionProperty rows = (CollectionProperty) prop;
for (int i = 0; i < rows.size(); i++) {
ArrayList<Object> row = (ArrayList<Object>) rows.get(i).getObjectValue();
String host = ((JMeterProperty) row.get(0)).getStringValue();
int port = ((JMeterProperty) row.get(1)).getIntValue();
String metric = ((JMeterProperty) row.get(2)).getStringValue();
String params = ((JMeterProperty) row.get(3)).getStringValue();
initiateConnector(host, port, i, metric, params);
}
Iterator<Object> it = connectors.keySet().iterator();
while (it.hasNext()) {
Object key = it.next();
try {
connectors.get(key).connect();
} catch (IOException ex) {
log.error("Error connecting to agent", ex);
connectors.put(key, new UnavailableAgentConnector(ex));
}
}
}
private void initiateConnector(String host, int port, int index, String metric, String params) {
InetSocketAddress addr = new InetSocketAddress(host, port);
String stringKey = addr.toString() + "#" + index;
// handle label parameter
MetricParams paramsParsed = MetricParams.createFromString(params);
String label;
if (paramsParsed.getLabel().isEmpty()) {
label = host + " " + metric;
if (params != null && !params.isEmpty()) {
label = label + " " + params;
}
} else {
label = host + " " + metric + " " + paramsParsed.getLabel();
String[] tokens = params.split("(?<!\\\\)" + PerfMonMetricGetter.DVOETOCHIE);
params = "";
for (int i = 0; i < tokens.length; i++) {
if (!tokens[i].startsWith("label=")) {
if (params.length() != 0) {
params = params + PerfMonMetricGetter.DVOETOCHIE;
}
params = params + tokens[i];
}
}
}
try {
if (connectors.containsKey(addr)) {
connectors.get(addr).addMetric(metric, params, label);
} else {
PerfMonAgentConnector connector = getConnector(host, port);
connector.addMetric(metric, params, label);
connectors.put(addr, connector);
}
} catch (IOException e) {
log.error("Problems creating connector", e);
connectors.put(stringKey, new UnavailableAgentConnector(e));
}
}
protected PerfMonAgentConnector getConnector(String host, int port) throws IOException {
log.debug("Trying new connector");
SocketAddress addr = new InetSocketAddress(host, port);
Transport transport = null;
try {
transport = TransportFactory.TCPInstance(addr);
if (!transport.test()) {
throw new IOException("Agent is unreachable via TCP");
}
} catch (IOException e) {
log.info("Can't connect TCP transport for host: " + addr.toString(), e);
boolean useUDP = JMeterUtils.getPropDefault("jmeterPlugin.perfmon.useUDP", false);
if (!useUDP) {
throw e;
} else {
try {
log.debug("Connecting UDP");
transport = TransportFactory.UDPInstance(addr);
if (!transport.test()) {
throw new IOException("Agent is unreachable via UDP");
}
} catch (IOException ex) {
log.info("Can't connect UDP transport for host: " + addr.toString(), ex);
throw ex;
}
}
}
NewAgentConnector conn = new NewAgentConnector();
conn.setTransport(transport);
return conn;
}
private void shutdownConnectors() {
log.debug("Shutting down connectors");
Iterator<Object> it = connectors.keySet().iterator();
while (it.hasNext()) {
Object key = it.next();
final PerfMonAgentConnector conn = connectors.get(key);
log.debug("Shutting down " + conn.toString());
//Fix ConcurrentModificationException if more than one host
//connectors.remove(key);
it.remove();
conn.disconnect();
}
}
private void processConnectors() {
Iterator<Object> it = connectors.keySet().iterator();
while (it.hasNext()) {
Object key = it.next();
PerfMonAgentConnector connector = connectors.get(key);
try {
connector.generateSamples(this);
} catch (IOException e) {
log.error(e.getMessage());
connectors.put(key, new UnavailableAgentConnector(e));
}
}
}
//need floating point precision for memory and cpu
@Override
public void generateSample(double value, String label) {
PerfMonSampleResult res = new PerfMonSampleResult();
res.setSampleLabel(label);
res.setValue(value);
res.setSuccessful(true);
SampleEvent e = new SampleEvent(res, PERFMON);
super.sampleOccurred(e);
}
@Override
public void generateErrorSample(String label, String errorMsg) {
PerfMonSampleResult res = new PerfMonSampleResult();
res.setSampleLabel(label);
res.setValue(-1L);
res.setResponseMessage(errorMsg);
res.setSuccessful(false);
SampleEvent e = new SampleEvent(res, PERFMON);
super.sampleOccurred(e);
log.error("Perfmon plugin error: " + errorMsg);
}
@Override
public void generate2Samples(long[] values, String label1, String label2) {
generate2Samples(values, label1, label2, 1d);
}
//float precision required for net io
@Override
public void generate2Samples(long[] values, String label1, String label2, double dividingFactor) {
if (oldValues.containsKey(label1) && oldValues.containsKey(label2)) {
generateSample(((double) (values[0] - oldValues.get(label1).longValue())) / dividingFactor, label1);
generateSample(((double) (values[1] - oldValues.get(label2).longValue())) / dividingFactor, label2);
}
oldValues.put(label1, new Long(values[0]));
oldValues.put(label2, new Long(values[1]));
}
}