add aggregator for parallel requests

ParallelRequestsAggregator generates a line plot that shows the number
of parallel requests among the plotted events.
This plot has two issues:
1. It only considers events that are plotted. Events that occur later,
   but were started within the plotted time frame are not considered.
2. For performance reasons we are only plotting points when a value
   changed. This leads to diagonal lines.

pdb-plotting/src/main/java/org/lucares/pdb/plot/api/AggregateHandler.java

@@ -21,6 +21,6 @@ public interface AggregateHandler {
 
 	void addPlots(StringBuilder result, Collection<DataSeries> dataSeries);
 
-	CustomAggregator createCustomAggregator(Path tmpDir);
+	CustomAggregator createCustomAggregator(Path tmpDir, long fromEpochMilli, long toEpochMilli);
 
 }

pdb-plotting/src/main/java/org/lucares/pdb/plot/api/NullAggregate.java

@@ -8,18 +8,19 @@ import org.lucares.recommind.logs.DataSeries;
 public class NullAggregate implements AggregateHandler {
 
 	@Override
-	public void addGnuplotDefinitions(StringBuilder result, String separator,
-			Collection<DataSeries> dataSeries) {
+	public void addGnuplotDefinitions(final StringBuilder result, final String separator,
+			final Collection<DataSeries> dataSeries) {
 		// nothing to do; this is a Null-Object
 	}
 
 	@Override
-	public void addPlots(StringBuilder result, Collection<DataSeries> dataSeries) {
+	public void addPlots(final StringBuilder result, final Collection<DataSeries> dataSeries) {
 		// nothing to do; this is a Null-Object
 	}
 
 	@Override
-	public CustomAggregator createCustomAggregator(Path tmpDir) {
+	public CustomAggregator createCustomAggregator(final Path tmpDir, final long fromEpochMilli,
+			final long toEpochMilli) {
 		return new NullCustomAggregator();
 	}
 

pdb-plotting/src/main/java/org/lucares/pdb/plot/api/ParallelRequestsAggregate.java

@@ -0,0 +1,39 @@
+package org.lucares.pdb.plot.api;
+
+import java.nio.file.Path;
+import java.util.Collection;
+
+import org.lucares.recommind.logs.DataSeries;
+
+public class ParallelRequestsAggregate implements AggregateHandler {
+
+	@Override
+	public void addGnuplotDefinitions(final StringBuilder result, final String separator,
+			final Collection<DataSeries> dataSeries) {
+		appendln(result, "set y2label \"Parallel Requests\"");
+		appendln(result, "set y2tics");
+	}
+
+	@Override
+	public void addPlots(final StringBuilder result, final Collection<DataSeries> dataSeries) {
+		for (final DataSeries dataSerie : dataSeries) {
+			final AggregatedData aggregatedData = dataSerie.getAggregatedData();
+			if (aggregatedData != null) {
+				appendfln(result, "'%s' using 1:2 notitle with lines axes x1y2 lw 1 %s, \\", //
+						aggregatedData.getDataFile().getAbsolutePath(), //
+						dataSerie.getStyle()//
+				);
+			}
+		}
+	}
+
+	@Override
+	public CustomAggregator createCustomAggregator(final Path tmpDir, final long fromEpochMilli,
+			final long toEpochMilli) {
+		if ((toEpochMilli - fromEpochMilli) <= 3600 * 1000) {
+			return new ParallelRequestsAggregator(tmpDir, fromEpochMilli, toEpochMilli);
+		} else {
+			return new NullCustomAggregator();
+		}
+	}
+}

pdb-plotting/src/main/java/org/lucares/pdb/plot/api/ParallelRequestsAggregator.java

@@ -0,0 +1,85 @@
+package org.lucares.pdb.plot.api;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ParallelRequestsAggregator implements CustomAggregator {
+
+	private static final Logger METRICS_LOGGER = LoggerFactory
+			.getLogger("org.lucares.metrics.aggregator.parallelRequests");
+
+	private final Path tmpDir;
+
+	private final int[] increments;
+
+	private final long fromEpochMilli;
+
+	public ParallelRequestsAggregator(final Path tmpDir, final long fromEpochMilli, final long toEpochMilli) {
+		this.tmpDir = tmpDir;
+		this.fromEpochMilli = fromEpochMilli;
+
+		if ((toEpochMilli - fromEpochMilli) > 3600 * 1000) {
+			throw new IllegalArgumentException("The " + ParallelRequestsAggregator.class.getSimpleName()
+					+ " must only be active for periods shorter than one hour, due to memory concerns.");
+		}
+
+		final int milliseconds = (int) (toEpochMilli - fromEpochMilli);
+		increments = new int[milliseconds];
+	}
+
+	@Override
+	public void addValue(final long epochMilli, final long value) {
+
+		final int endPos = (int) (epochMilli - fromEpochMilli);
+		increments[endPos]--;
+
+		final int startPos = Math.max(0, (int) (endPos - value));
+		increments[startPos]++;
+
+	}
+
+	@Override
+	public AggregatedData getAggregatedData() throws IOException {
+
+		final long start = System.nanoTime();
+		final char separator = ',';
+		final char newline = '\n';
+
+		final File dataFile = File.createTempFile("data", ".dat", tmpDir.toFile());
+		try (final Writer output = new BufferedWriter(
+				new OutputStreamWriter(new FileOutputStream(dataFile), StandardCharsets.US_ASCII));) {
+
+			final StringBuilder data = new StringBuilder();
+
+			int value = 0;
+			for (int i = 0; i < increments.length; i++) {
+				final int increment = increments[i];
+				if (increment != 0) {
+					value += increment;
+					data.append(String.format("%.3f", (fromEpochMilli + i) / 1000.0));
+					data.append(separator);
+					data.append(value);
+					data.append(newline);
+				}
+			}
+
+			output.write(data.toString());
+
+		}
+
+		final String title = String.format("parallelRequests");
+		METRICS_LOGGER.debug("wrote parallelRequests csv in: {}ms file={}", (System.nanoTime() - start) / 1_000_000.0,
+				dataFile);
+		return new AggregatedData(title, dataFile);
+	}
+
+}

pdb-plotting/src/main/java/org/lucares/pdb/plot/api/PercentileAggregate.java

@@ -8,7 +8,8 @@ import org.lucares.recommind.logs.DataSeries;
 public class PercentileAggregate implements AggregateHandler {
 
 	@Override
-	public CustomAggregator createCustomAggregator(final Path tmpDir) {
+	public CustomAggregator createCustomAggregator(final Path tmpDir, final long fromEpochMilli,
+			final long toEpochMilli) {
 		return new PercentileCustomAggregator(tmpDir);
 	}
 

pdb-plotting/src/main/java/org/lucares/recommind/logs/GnuplotFileGenerator.java

@@ -65,13 +65,16 @@ public class GnuplotFileGenerator {
 		appendfln(result, "set key font \",10\"");
 
 		if (!settings.isRenderLabels()) {
-			appendfln(result, "set format x \"\"", xAxis.getFormatX());
-			appendfln(result, "set xlabel \"\"", xAxis.getXlabel());
+			appendfln(result, "set format x \"\"");
+			appendfln(result, "set xlabel \"\"");
 			appendfln(result, "set x2label \"\"");
 			appendln(result, "set format x2 \"\"");
 
-			appendfln(result, "set ylabel \"\"", settings.getYlabel());
+			appendfln(result, "set ylabel \"\"");
 			appendln(result, "set format y \"\"");
+			appendln(result, "set y2label \"\"");
+			appendln(result, "set format y2 \"\"");
+
 			appendln(result, "set nokey");
 		}
 

pdb-plotting/src/main/java/org/lucares/recommind/logs/ScatterPlot.java

@@ -210,7 +210,8 @@ public class ScatterPlot {
 		final long maxValue = plotSettings.getYRangeUnit() == TimeRangeUnitInternal.AUTOMATIC ? Long.MAX_VALUE
 				: plotSettings.getYRangeUnit().toMilliSeconds(plotSettings.getYRangeMax());
 
-		final CustomAggregator aggregator = plotSettings.getAggregate().createCustomAggregator(tmpDir);
+		final CustomAggregator aggregator = plotSettings.getAggregate().createCustomAggregator(tmpDir, fromEpochMilli,
+				toEpochMilli);
 
 		int count = 0; // number of values in the x-axis range (used to compute stats)
 		int plottedValues = 0;