Better charts for statistics over time (series)

* Remove IterationData hack, introduce representations *.Series
* Make PeriodicStatistics merge-able
* Rename DefaultStatistics and DefaultOperationStats to BasicStatistics and BasicOperationStats
* Add CommonStatistics that works as a handy template for periodic statistics + histogram
* Refactor Statistics.getOperationStats + OperationStats.getRepresentation -> Statistics.getOperations + Statistics.getRepresentation
* Add the series charts to HTML report

core/src/main/java/org/radargun/stages/test/AnalyzeTestStage.java

@@ -13,7 +13,6 @@
 import org.radargun.config.Stage;
 import org.radargun.reporting.Report;
 import org.radargun.stages.AbstractDistStage;
-import org.radargun.stats.OperationStats;
 import org.radargun.stats.Statistics;
 import org.radargun.stats.representation.RepresentationType;
 import org.radargun.utils.Utils;
@@ -64,7 +63,7 @@ protected Number analyze(Report.Test test) {
       double min = Double.MAX_VALUE, max = -Double.MAX_VALUE, sum = 0;
       Group minGroup = null, maxGroup = null;
       for (Group g : groups) {
-         double value = statisticsType.getValue(g.stats, g.duration);
+         double value = statisticsType.getValue(g.statistics, operation, g.duration);
          log.tracef("iteration %d, node %d, thread %d: %d threads, duration %s -> value %f",
             g.origin.iteration, g.origin.node, g.origin.thread, g.threads,
             Utils.prettyPrintTime(g.duration, TimeUnit.NANOSECONDS), value);
@@ -147,21 +146,21 @@ private List<Group> group(int iteration, Set<Map.Entry<Integer, List<Statistics>
                   }
                }
                for (Statistics s : entry.getValue()) {
-                  groups.add(new Group(s.getOperationsStats().get(operation), 1, duration(s), new Origin(iteration, entry.getKey(), threadCounter)));
+                  groups.add(new Group(s, 1, duration(s), new Origin(iteration, entry.getKey(), threadCounter)));
                }
             }
             break;
          case GROUP_BY_NODE:
             for (Map.Entry<Integer, List<Statistics>> entry : statistics) {
                entry.getValue().stream().reduce(Statistics.MERGE).map(aggregation ->
-                  groups.add(new Group(aggregation.getOperationsStats().get(operation), entry.getValue().size(), duration(aggregation), new Origin(iteration, entry.getKey(), -1)))
+                  groups.add(new Group(aggregation, entry.getValue().size(), duration(aggregation), new Origin(iteration, entry.getKey(), -1)))
                );
             }
             break;
          case GROUP_ALL:
             int threads = statistics.stream().mapToInt(e -> e.getValue().size()).sum();
             statistics.stream().flatMap(e -> e.getValue().stream()).reduce(Statistics.MERGE).map(aggregation ->
-               groups.add(new Group(aggregation.getOperationsStats().get(operation), threads, duration(aggregation), new Origin(iteration, -1, -1)))
+               groups.add(new Group(aggregation, threads, duration(aggregation), new Origin(iteration, -1, -1)))
             );
             break;
          default:
@@ -180,13 +179,13 @@ public DistStageAck executeOnSlave() {
    }
 
    protected static class Group {
-      public final OperationStats stats;
+      public final Statistics statistics;
       public final int threads;
       public final long duration;
       public final Origin origin;
 
-      public Group(OperationStats stats, int threads, long duration, Origin origin) {
-         this.stats = stats;
+      public Group(Statistics statistics, int threads, long duration, Origin origin) {
+         this.statistics = statistics;
          this.threads = threads;
          this.duration = duration;
          this.origin = origin;

core/src/main/java/org/radargun/stages/test/BaseTestStage.java

@@ -9,7 +9,7 @@
 import org.radargun.config.Stage;
 import org.radargun.reporting.Report;
 import org.radargun.stages.AbstractDistStage;
-import org.radargun.stats.DefaultStatistics;
+import org.radargun.stats.BasicStatistics;
 import org.radargun.stats.Statistics;
 import org.radargun.utils.TimeConverter;
 
@@ -30,7 +30,7 @@ public abstract class BaseTestStage extends AbstractDistStage {
 
    @Property(name = "statistics", doc = "Type of gathered statistics. Default are the 'default' statistics " +
       "(fixed size memory footprint for each operation).", complexConverter = Statistics.Converter.class)
-   public Statistics statisticsPrototype = new DefaultStatistics();
+   public Statistics statisticsPrototype = new BasicStatistics();
 
    @Property(doc = "Property, which value will be used to identify individual iterations (e.g. num-threads).")
    public String iterationProperty;

core/src/main/java/org/radargun/stages/test/PerformanceCondition.java

@@ -10,7 +10,6 @@
 import org.radargun.config.PropertyHelper;
 import org.radargun.logging.Log;
 import org.radargun.logging.LogFactory;
-import org.radargun.stats.OperationStats;
 import org.radargun.stats.Statistics;
 import org.radargun.stats.representation.DefaultOutcome;
 import org.radargun.stats.representation.OperationThroughput;
@@ -90,10 +89,8 @@ public void init() {
 
       @Override
       public boolean evaluate(Statistics statistics) {
-         OperationStats stats = statistics.getOperationsStats().get(on);
-         if (stats == null) throw new IllegalStateException("No statistics for operation " + on);
-         DefaultOutcome outcome = stats.getRepresentation(DefaultOutcome.class);
-         if (outcome == null) throw new IllegalStateException("Cannot determine mean from " + stats);
+         DefaultOutcome outcome = statistics.getRepresentation(on, DefaultOutcome.class);
+         if (outcome == null) throw new IllegalStateException("Cannot determine mean from " + statistics);
          log.info("Mean is " + Utils.prettyPrintTime((long) outcome.responseTimeMean, TimeUnit.NANOSECONDS) + PropertyHelper.toString(this));
          if (below != null) return outcome.responseTimeMean < below;
          if (over != null) return outcome.responseTimeMean > over;
@@ -116,12 +113,10 @@ public void init() {
 
       @Override
       public boolean evaluate(Statistics statistics) {
-         OperationStats stats = statistics.getOperationsStats().get(on);
-         if (stats == null) throw new IllegalStateException("No statistics for operation " + on);
-         org.radargun.stats.representation.OperationThroughput throughput = stats.getRepresentation(
+         org.radargun.stats.representation.OperationThroughput throughput = statistics.getRepresentation(on,
             org.radargun.stats.representation.OperationThroughput.class,
             TimeUnit.MILLISECONDS.toNanos(statistics.getEnd() - statistics.getBegin()));
-         if (throughput == null) throw new IllegalStateException("Cannot determine throughput from " + stats);
+         if (throughput == null) throw new IllegalStateException("Cannot determine throughput from " + statistics);
          log.info(getClass().getSimpleName() + " is " + getThroughput(throughput) + " ops/s " + PropertyHelper.toString(this));
          if (below != null) return getThroughput(throughput) < below;
          if (over != null) return getThroughput(throughput) > over;
@@ -161,10 +156,8 @@ public void init() {
 
       @Override
       public boolean evaluate(Statistics statistics) {
-         OperationStats stats = statistics.getOperationsStats().get(on);
-         if (stats == null) throw new IllegalStateException("No statistics for operation " + on);
-         DefaultOutcome outcome = stats.getRepresentation(DefaultOutcome.class);
-         if (outcome == null) throw new IllegalStateException("Cannot determine request count from " + stats);
+         DefaultOutcome outcome = statistics.getRepresentation(on, DefaultOutcome.class);
+         if (outcome == null) throw new IllegalStateException("Cannot determine request count from " + statistics);
          log.info("Executed " + outcome.requests + " reqs " + PropertyHelper.toString(this));
          if (below != null) return outcome.requests < below;
          if (over != null) return outcome.requests > over;
@@ -199,10 +192,8 @@ public void init() {
 
       @Override
       public boolean evaluate(Statistics statistics) {
-         OperationStats stats = statistics.getOperationsStats().get(on);
-         if (stats == null) throw new IllegalStateException("No statistics for operation " + on);
-         DefaultOutcome outcome = stats.getRepresentation(DefaultOutcome.class);
-         if (outcome == null) throw new IllegalStateException("Cannot determine error count from " + stats);
+         DefaultOutcome outcome = statistics.getRepresentation(on, DefaultOutcome.class);
+         if (outcome == null) throw new IllegalStateException("Cannot determine error count from " + statistics);
          log.info("Encountered " + outcome.errors + " errors " + PropertyHelper.toString(this));
          if (totalBelow != null) return outcome.errors < totalBelow;
          if (totalOver != null) return outcome.errors > totalOver;
@@ -232,11 +223,9 @@ public void init() {
 
       @Override
       public boolean evaluate(Statistics statistics) {
-         OperationStats stats = statistics.getOperationsStats().get(on);
-         if (stats == null) throw new IllegalStateException("No statistics for operation " + on);
          org.radargun.stats.representation.Percentile percentile
-            = stats.getRepresentation(org.radargun.stats.representation.Percentile.class, value);
-         if (percentile == null) throw new IllegalStateException("Cannot determine percentile from " + stats);
+            = statistics.getRepresentation(on, org.radargun.stats.representation.Percentile.class, value);
+         if (percentile == null) throw new IllegalStateException("Cannot determine percentile from " + statistics);
          log.info("Response time is " + Utils.prettyPrintTime((long) percentile.responseTimeMax, TimeUnit.NANOSECONDS) + PropertyHelper.toString(this));
          if (below != null) return percentile.responseTimeMax < below;
          if (over != null) return percentile.responseTimeMax > over;

core/src/main/java/org/radargun/stages/test/legacy/LegacyTestStage.java

@@ -1,9 +1,11 @@
 package org.radargun.stages.test.legacy;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import org.radargun.DistStageAck;
 import org.radargun.StageResult;
@@ -12,7 +14,6 @@
 import org.radargun.config.Namespace;
 import org.radargun.config.Property;
 import org.radargun.config.Stage;
-import org.radargun.reporting.IterationData;
 import org.radargun.reporting.Report;
 import org.radargun.stages.test.BaseTestStage;
 import org.radargun.stages.test.TransactionMode;
@@ -120,23 +121,17 @@ public StageResult processAckOnMaster(List<DistStageAck> acks) {
       Report.Test test = getTest(amendTest);
       testIteration = test == null ? 0 : test.getIterations().size();
       // we cannot use aggregated = createStatistics() since with PeriodicStatistics the merge would fail
-      int threads = 0;
       List<StatisticsAck> statisticsAcks = instancesOf(acks, StatisticsAck.class);
-      Statistics aggregated = statisticsAcks.stream().filter(ack -> ack.iterations != null)
-            .flatMap(ack -> ack.iterations.stream().flatMap(s -> s.stream())).reduce(null, Statistics.MERGE);
+      Statistics aggregated = statisticsAcks.stream().flatMap(ack -> ack.statistics.stream()).reduce(null, Statistics.MERGE);
       for (StatisticsAck ack : statisticsAcks) {
-         if (ack.iterations != null) {
-            int i = getTestIteration();
-            for (List<Statistics> threadStats : ack.iterations) {
-               if (test != null) {
-                  // TODO: this looks like we could get same iteration value for all iterations reported
-                  String iterationValue = resolveIterationValue();
-                  if (iterationValue != null) {
-                     test.setIterationValue(i, iterationValue);
-                  }
-                  test.addStatistics(i++, ack.getSlaveIndex(), threadStats);
+         if (ack.statistics != null) {
+            if (test != null) {
+               int testIteration = getTestIteration();
+               String iterationValue = resolveIterationValue();
+               if (iterationValue != null) {
+                  test.setIterationValue(testIteration, iterationValue);
                }
-               threads = Math.max(threads, threadStats.size());
+               test.addStatistics(testIteration, ack.getSlaveIndex(), ack.statistics);
             }
          } else {
             log.trace("No statistics received from slave: " + ack.getSlaveIndex());
@@ -235,47 +230,21 @@ protected List<LegacyStressor> startStressors() {
    }
 
    protected DistStageAck newStatisticsAck(List<LegacyStressor> stressors) {
-      List<List<Statistics>> results = gatherResults(stressors, new StatisticsResultRetriever());
+      List<Statistics> results = gatherResults(stressors, new StatisticsResultRetriever());
       return new StatisticsAck(slaveState, results);
    }
 
-   protected <T> List<List<T>> gatherResults(List<LegacyStressor> stressors, ResultRetriever<T> retriever) {
-      List<T> results = new ArrayList<>(stressors.size());
-      for (LegacyStressor stressor : stressors) {
-         T result = retriever.getResult(stressor);
-         if (result != null) { // stressor could have crashed during initialization
-            results.add(result);
-         }
-      }
-
-      List<List<T>> all = new ArrayList<>();
-      all.add(new ArrayList<T>());
-      /* expand the iteration statistics into iterations */
-      for (T result : results) {
-         if (result instanceof IterationData) {
-            int iteration = 0;
-            for (IterationData.Iteration<? extends T> it : ((IterationData<T>) result).getIterations()) {
-               while (iteration >= all.size()) {
-                  all.add(new ArrayList<T>(results.size()));
-               }
-               addResult(all.get(iteration++), it.data, retriever);
-            }
-         } else {
-            addResult(all.get(0), result, retriever);
-         }
-      }
-      return all;
-   }
-
-   private <T> void addResult(List<T> results, T result, ResultRetriever<T> retriever) {
+   protected <T> List<T> gatherResults(List<LegacyStressor> stressors, ResultRetriever<T> retriever) {
       if (mergeThreadStats) {
-         if (results.isEmpty()) {
-            results.add(result);
-         } else {
-            retriever.mergeResult(results.get(0), result);
-         }
+         return stressors.stream()
+            .map(retriever::getResult)
+            .reduce(retriever::merge)
+            .map(Collections::singletonList).orElse(Collections.emptyList());
       } else {
-         results.add(result);
+         return stressors.stream()
+            .map(retriever::getResult)
+            .filter(r -> r != null)
+            .collect(Collectors.toList());
       }
    }
 
@@ -357,7 +326,7 @@ public boolean isSingleTxType() {
    protected interface ResultRetriever<T> {
       T getResult(LegacyStressor stressor);
 
-      void mergeResult(T into, T that);
+      T merge(T stats1, T stats2);
    }
 
    protected static class StatisticsResultRetriever implements ResultRetriever<Statistics> {
@@ -369,8 +338,8 @@ public Statistics getResult(LegacyStressor stressor) {
       }
 
       @Override
-      public void mergeResult(Statistics into, Statistics that) {
-         into.merge(that);
+      public Statistics merge(Statistics stats1, Statistics stats2) {
+         return Statistics.MERGE.apply(stats1, stats2);
       }
    }
 
@@ -384,11 +353,11 @@ public TestTimeoutException(Throwable cause) {
    }
 
    protected static class StatisticsAck extends DistStageAck {
-      public final List<List<Statistics>> iterations;
+      public final List<Statistics> statistics;
 
-      public StatisticsAck(SlaveState slaveState, List<List<Statistics>> iterations) {
+      public StatisticsAck(SlaveState slaveState, List<Statistics> statistics) {
          super(slaveState);
-         this.iterations = iterations;
+         this.statistics = statistics;
       }
    }
 }

core/src/main/java/org/radargun/stats/AllRecordingOperationStats.java

@@ -107,12 +107,12 @@ public OperationStats copy() {
 
    @SuppressWarnings("unchecked")
    @Override
-   public <T> T getRepresentation(Class<T> clazz, Object... args) {
+   public <T> T getRepresentation(Class<T> clazz, Statistics ownerStatistics, Object... args) {
       long requests = full ? responseTimes.length : pos;
       if (clazz == DefaultOutcome.class) {
          return (T) new DefaultOutcome(requests, errors, getMeanDuration(), getMaxDuration());
       } else if (clazz == OperationThroughput.class) {
-         return (T) OperationThroughput.compute(requests, errors, args);
+         return (T) OperationThroughput.compute(requests, errors, ownerStatistics);
       } else if (clazz == Percentile.class) {
          double percentile = Percentile.getPercentile(args);
          int size = full ? responseTimes.length : pos;

core/src/main/java/org/radargun/stats/DefaultOperationStats.java → core/src/main/java/org/radargun/stats/BasicOperationStats.java

@@ -11,8 +11,8 @@
  *
  * @author Radim Vansa <rvansa@redhat.com>
  */
-@DefinitionElement(name = "default", doc = "Operations statistics with fixed memory footprint.")
-public class DefaultOperationStats implements OperationStats {
+@DefinitionElement(name = "basic", doc = "Operations statistics with fixed memory footprint.")
+public class BasicOperationStats implements OperationStats {
    private static final double INVERSE_NORMAL_95 = 1.96;
    private static final double INVERSE_NORMAL_50 = 0.67448;
    private long requests;
@@ -23,13 +23,13 @@ public class DefaultOperationStats implements OperationStats {
    private long errors;
 
    @Override
-   public DefaultOperationStats newInstance() {
-      return new DefaultOperationStats();
+   public BasicOperationStats newInstance() {
+      return new BasicOperationStats();
    }
 
    @Override
-   public DefaultOperationStats copy() {
-      DefaultOperationStats copy = newInstance();
+   public BasicOperationStats copy() {
+      BasicOperationStats copy = newInstance();
       copy.requests = requests;
       copy.responseTimeMax = responseTimeMax;
       copy.responseTimeSum = responseTimeSum;
@@ -41,8 +41,8 @@ public DefaultOperationStats copy() {
 
    @Override
    public void merge(OperationStats o) {
-      if (!(o instanceof DefaultOperationStats)) throw new IllegalArgumentException(o.toString());
-      DefaultOperationStats other = (DefaultOperationStats) o;
+      if (!(o instanceof BasicOperationStats)) throw new IllegalArgumentException(o.toString());
+      BasicOperationStats other = (BasicOperationStats) o;
       responseTimeM2 = mergeM2(responseTimeMean, responseTimeM2, requests, other.responseTimeMean, other.responseTimeM2, other.requests);
       responseTimeMean = mergeMean(responseTimeMean, requests, other.responseTimeMean, other.requests);
       requests += other.requests;
@@ -119,13 +119,13 @@ public MeanAndDev getMeanAndDev() {
    }
 
    @Override
-   public <T> T getRepresentation(Class<T> clazz, Object... args) {
+   public <T> T getRepresentation(Class<T> clazz, Statistics ownerStatistics, Object... args) {
       if (clazz == DefaultOutcome.class) {
          return (T) new DefaultOutcome(requests, errors, responseTimeMean, responseTimeMax);
       } else if (clazz == MeanAndDev.class) {
          return (T) getMeanAndDev();
       } else if (clazz == OperationThroughput.class) {
-         return (T) OperationThroughput.compute(requests, errors, args);
+         return (T) OperationThroughput.compute(requests, errors, ownerStatistics);
       } else if (clazz == BoxAndWhiskers.class) {
          return (T) getBoxAndWhiskers();
       } else {