alluxio-master-2.x/stress/common/src/main/java/alluxio/stress/client/ClientIOTaskResult.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.stress.client;

import alluxio.Constants;
import alluxio.collections.Pair;
import alluxio.stress.BaseParameters;
import alluxio.stress.Parameters;
import alluxio.stress.Summary;
import alluxio.stress.TaskResult;
import alluxio.stress.common.SummaryStatistics;
import alluxio.stress.graph.Graph;
import alluxio.stress.graph.LineGraph;

import com.fasterxml.jackson.annotation.JsonIgnore;
import com.google.common.base.Splitter;

import java.text.DateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * The task result for the master stress tests.
 */
public final class ClientIOTaskResult implements TaskResult, Summary {
  private long mRecordStartMs;
  private long mEndMs;
  private Map<Integer, ThreadCountResult> mThreadCountResults;
  private BaseParameters mBaseParameters;
  private ClientIOParameters mParameters;

  private Map<Integer, Map<String, SummaryStatistics>> mTimeToFirstByte;

  /**
   * Creates an instance.
   */
  public ClientIOTaskResult() {
    // Default constructor required for json deserialization
    mThreadCountResults = new HashMap<>();
    mTimeToFirstByte = new HashMap<>();
  }

  @Override
  public BaseParameters getBaseParameters() {
    return mBaseParameters;
  }

  /**
   * @param baseParameters the base parameters
   */
  public void setBaseParameters(BaseParameters baseParameters) {
    mBaseParameters = baseParameters;
  }

  /**
   * @return the parameters
   */
  public ClientIOParameters getParameters() {
    return mParameters;
  }

  /**
   * @param parameters the parameters
   */
  public void setParameters(ClientIOParameters parameters) {
    mParameters = parameters;
  }

  /**
   * @return the start time (in ms)
   */
  public long getRecordStartMs() {
    return mRecordStartMs;
  }

  /**
   * @param recordStartMs the start time (in ms)
   */
  public void setRecordStartMs(long recordStartMs) {
    mRecordStartMs = recordStartMs;
  }

  /**
   * @return client IO statistics per method
   */
  public Map<Integer, Map<String, SummaryStatistics>> getTimeToFirstBytePerThread() {
    return mTimeToFirstByte;
  }

  /**
   * @param timeToFirstByte time to first statistics
   */
  public void setTimeToFirstBytePerThread(Map<Integer, Map<String,
      SummaryStatistics>> timeToFirstByte) {
    mTimeToFirstByte = timeToFirstByte;
  }

  /**
   * @param numThreads thread count
   * @param statistics ClientIOTaskResultStatistics
   */
  public void putTimeToFirstBytePerThread(Integer numThreads,
      Map<String, SummaryStatistics> statistics) {
    mTimeToFirstByte.put(numThreads, statistics);
  }

  /**
   * @return the end time (in ms)
   */
  public long getEndMs() {
    return mEndMs;
  }

  /**
   * @param endMs the end time (in ms)
   */
  public void setEndMs(long endMs) {
    mEndMs = endMs;
  }

  /**
   * @return the map of thread counts to results
   */
  public Map<Integer, ThreadCountResult> getThreadCountResults() {
    return mThreadCountResults;
  }

  /**
   * @param threadCountResults the map of thread counts to results
   */
  public void setThreadCountResults(Map<Integer, ThreadCountResult> threadCountResults) {
    mThreadCountResults = threadCountResults;
  }

  /**
   * @param threadCount the thread count of the results
   * @param threadCountResult the results to add
   */
  public void addThreadCountResults(int threadCount, ThreadCountResult threadCountResult) {
    mThreadCountResults.put(threadCount, threadCountResult);
  }

  private long computeLastEndMs() {
    long endMs = 0;
    for (ThreadCountResult result : mThreadCountResults.values()) {
      endMs = Math.max(endMs, result.getEndMs());
    }
    return endMs;
  }

  private LineGraph.Data getThroughputData() {
    LineGraph.Data data = new LineGraph.Data();
    for (Map.Entry<Integer, ThreadCountResult> entry : mThreadCountResults.entrySet()) {
      data.addData(entry.getKey(), entry.getValue().getIOMBps());
    }
    return data;
  }

  private void getNumSuccessData(String series, LineGraph lineGraph) {
    Map<String, LineGraph.Data> data = new HashMap<>();

    for (Map.Entry<Integer, Map<String, SummaryStatistics>> threadEntry :
        mTimeToFirstByte.entrySet()) {
      for (Map.Entry<String, SummaryStatistics> methodEntry :
          threadEntry.getValue().entrySet()) {
        String prefix = series + ", method: " + methodEntry.getKey();
        LineGraph.Data currentData = data.getOrDefault(prefix, new LineGraph.Data());
        currentData.addData(threadEntry.getKey(), methodEntry.getValue().mNumSuccesses);
        data.put(prefix, currentData);
      }
    }

    for (Map.Entry<String, LineGraph.Data> entry : data.entrySet()) {
      lineGraph.addDataSeries(entry.getKey(), entry.getValue());
    }
  }

  private void getTimeToFistByteData(String series, LineGraph lineGraph) {
    for (Map.Entry<Integer, Map<String, SummaryStatistics>> threadEntry :
        mTimeToFirstByte.entrySet()) {
      for (Map.Entry<String, SummaryStatistics> methodEntry :
          threadEntry.getValue().entrySet()) {
        lineGraph.addDataSeries(series
            + ", method: " + methodEntry.getKey()
            + ", thread: " + threadEntry.getKey(), methodEntry.getValue().computeTimeData());
      }
    }
  }

  @Override
  public List<String> getErrors() {
    List<String> errors = new ArrayList<>();
    for (Map.Entry<Integer, ThreadCountResult> entry : mThreadCountResults.entrySet()) {
      // add all the errors for this thread count, with the thread count appended to prefix
      errors.addAll(
          entry.getValue().getErrors().stream().map(err -> entry.getKey().toString() + ": " + err)
              .collect(Collectors.toList()));
    }
    return errors;
  }

  @Override
  public TaskResult.Aggregator aggregator() {
    return new Aggregator();
  }

  private static final class Aggregator implements TaskResult.Aggregator<ClientIOTaskResult> {
    @SuppressWarnings("checkstyle:OperatorWrap")
    @Override
    public ClientIOSummary aggregate(Iterable<ClientIOTaskResult> results) throws Exception {
      long recordStartMs = 0;
      long endMs = 0;
      ClientIOParameters clientIOParameters = null;
      BaseParameters baseParameters = null;
      Map<String, ClientIOTaskResult> nodes = new HashMap<>();
      for (ClientIOTaskResult taskResult: results) {
        clientIOParameters = taskResult.getParameters();
        baseParameters = taskResult.getBaseParameters();
        String jobWorkerUniqueId = taskResult.getBaseParameters().mId;
        nodes.put(jobWorkerUniqueId, taskResult);
      }

      Map<Integer, Long> threadCountIOBytes = new HashMap<>();
      Map<Integer, Long> threadCountRecordedStartMs = new HashMap<>();
      Map<Integer, Long> threadCountEndMs = new HashMap<>();
      for (ClientIOTaskResult taskResult: results) {
        for (Map.Entry<Integer, ThreadCountResult> entry :
            taskResult.getThreadCountResults().entrySet()) {
          int numThreads = entry.getKey();
          ThreadCountResult result = entry.getValue();
          long ioBytes = result.getIOBytes();
          threadCountIOBytes.put(numThreads,
              threadCountIOBytes.getOrDefault(numThreads, (long) 0) + ioBytes);
          if (threadCountRecordedStartMs.containsKey(numThreads)) {
            threadCountRecordedStartMs.put(numThreads,
                Math.min(threadCountRecordedStartMs.get(numThreads), result.getRecordStartMs()));
          } else {
            threadCountRecordedStartMs.put(numThreads, result.getRecordStartMs());
          }
          if (threadCountEndMs.containsKey(numThreads)) {
            threadCountEndMs.put(numThreads,
                Math.max(threadCountEndMs.get(numThreads), result.getEndMs()));
          } else {
            threadCountEndMs.put(numThreads, result.getEndMs());
          }
        }
      }
      Map<Integer, Float> threadCountIoMbps = new HashMap<>();
      for (Map.Entry<Integer, Long> threadCountIOBytesEntry: threadCountIOBytes.entrySet()) {
        int numThreads = threadCountIOBytesEntry.getKey();
        long ioBytes = threadCountIOBytesEntry.getValue();
        threadCountIoMbps.put(numThreads, (float) ioBytes
            / (threadCountEndMs.get(numThreads) - threadCountRecordedStartMs.get(numThreads))
            * 1000.0f / Constants.MB);
      }

      return new ClientIOSummary(clientIOParameters, baseParameters, nodes, threadCountIoMbps);
    }
  }

  @Override
  public alluxio.stress.GraphGenerator graphGenerator() {
    return new GraphGenerator();
  }

  /**
   * The graph generator for this summary.
   */
  public static final class GraphGenerator extends alluxio.stress.GraphGenerator {
    @Override
    public List<Graph> generate(List<? extends Summary> results) {
      List<Graph> graphs = new ArrayList<>();
      // expecting ClientIOTaskResult, or will throw ClassCastException
      List<ClientIOTaskResult> summaries =
          results.stream().map(x -> (ClientIOTaskResult) x).collect(Collectors.toList());

      if (!summaries.isEmpty()) {
        // first() is the list of common field names, second() is the list of unique field names
        Pair<List<String>, List<String>> fieldNames = Parameters.partitionFieldNames(
            summaries.stream().map(x -> x.mParameters).collect(Collectors.toList()));

        // Split up common description into 100 character chunks, for the subtitle
        List<String> subTitle = new ArrayList<>(Splitter.fixedLength(100).splitToList(
            summaries.get(0).mParameters.getDescription(fieldNames.getFirst())));

        for (ClientIOTaskResult summary : summaries) {
          String series = summary.mParameters.getDescription(fieldNames.getSecond());
          subTitle.add(series + ": " + DateFormat.getDateTimeInstance()
              .format(summary.computeLastEndMs()));
        }
        ClientIOOperation operation = summaries.get(0).getParameters().mOperation;
        LineGraph responseTimeGraph = new LineGraph(String
            .format("%s - %s - Throughput", operation,
                summaries.get(0).mParameters.getDescription(
                    Collections.singletonList(ClientIOParameters.FIELD_READ_RANDOM))),
            subTitle, "# Threads", "Throughput (MB/s)");

        LineGraph numSuccessGraph = new LineGraph(String
            .format("%s - %s - API calls", operation,
                summaries.get(0).mParameters.getDescription(
                    Collections.singletonList(ClientIOParameters.FIELD_READ_RANDOM))),
            subTitle, "# Threads", "# API calls");

        LineGraph timeToFirstByteGraph = new LineGraph(String
            .format("%s - %s - Time To First Byte", operation,
                summaries.get(0).mParameters.getDescription(
                    Collections.singletonList(ClientIOParameters.FIELD_READ_RANDOM))),
            subTitle, "# Threads", "Time To First Byte (Ms)");

        for (ClientIOTaskResult summary : summaries) {
          String series = summary.mParameters.getDescription(fieldNames.getSecond());
          responseTimeGraph.addDataSeries(series, summary.getThroughputData());
          responseTimeGraph.setErrors(series, summary.getErrors());

          summary.getNumSuccessData(series, numSuccessGraph);

          summary.getTimeToFistByteData(series, timeToFirstByteGraph);
        }
        graphs.add(responseTimeGraph);
        graphs.add(numSuccessGraph);
        graphs.add(timeToFirstByteGraph);
      }
      return graphs;
    }
  }

  /**
   * A result for a single thread count test.
   */
  public static final class ThreadCountResult {
    private long mRecordStartMs;
    private long mEndMs;
    private long mIOBytes;
    private List<String> mErrors;

    /**
     * Creates an instance.
     */
    public ThreadCountResult() {
      // Default constructor required for json deserialization
      mErrors = new ArrayList<>();
    }

    /**
     * Merges (updates) a result with this result.
     *
     * @param result  the result to merge
     */
    public void merge(ClientIOTaskResult.ThreadCountResult result) {
      mRecordStartMs = Math.min(mRecordStartMs, result.mRecordStartMs);
      mEndMs = Math.max(mEndMs, result.mEndMs);
      mIOBytes += result.mIOBytes;
      mErrors.addAll(result.mErrors);
    }

    /**
     * @return the duration (in ms)
     */
    public long getDurationMs() {
      return mEndMs - mRecordStartMs;
    }

    /**
     * @param durationMs the duration (in ms)
     */
    @JsonIgnore
    public void setDurationMs(long durationMs) {
      // ignore
    }

    /**
     * @return bytes of IO
     */
    public long getIOBytes() {
      return mIOBytes;
    }

    /**
     * Increments the bytes of IO an amount.
     *
     * @param ioBytes the amount to increment by
     */
    public void incrementIOBytes(long ioBytes) {
      mIOBytes += ioBytes;
    }

    /**
     * @param ioBytes bytes of IO
     */
    public void setIOBytes(long ioBytes) {
      mIOBytes = ioBytes;
    }

    /**
     * @return the start time (in ms)
     */
    public long getRecordStartMs() {
      return mRecordStartMs;
    }

    /**
     * @param recordStartMs the start time (in ms)
     */
    public void setRecordStartMs(long recordStartMs) {
      mRecordStartMs = recordStartMs;
    }

    /**
     * @return the end time (in ms)
     */
    public long getEndMs() {
      return mEndMs;
    }

    /**
     * @param endMs the end time (in ms)
     */
    public void setEndMs(long endMs) {
      mEndMs = endMs;
    }

    /**
     * @return the list of errors
     */
    public List<String> getErrors() {
      return mErrors;
    }

    /**
     * @param errors the list of errors
     */
    public void setErrors(List<String> errors) {
      mErrors = errors;
    }

    /**
     * @param errMesssage the error message to add
     */
    public void addErrorMessage(String errMesssage) {
      mErrors.add(errMesssage);
    }

    /**
     * @return the throughput (MB/s)
     */
    public float getIOMBps() {
      return ((float) mIOBytes / getDurationMs()) * 1000.0f / Constants.MB;
    }

    /**
     * @param ioMBps the throughput (MB / s)
     */
    @JsonIgnore
    public void setIOMBps(float ioMBps) {
      // ignore
    }
  }
}
