/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you under the Apache License, Version 2.0 (the
    * "License"); you may not use this file except in compliance
    * with the License.  You may obtain a copy of the License at
    *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.giraph.metrics;
  
  import org.apache.giraph.graph.GraphTaskManager;
  import org.apache.giraph.ooc.OutOfCoreEngine;
  import org.apache.giraph.ooc.OutOfCoreIOCallable;
  import org.apache.giraph.worker.BspServiceWorker;
  
  import com.google.common.collect.Maps;
  
  import java.io.PrintStream;
  import java.util.Map;
  
  /**
   * Map of a bunch of aggregated metrics
   */
  public class AggregatedMetrics {
    /** Mapping from name to aggregated metric */
    private Map<String, AggregatedMetric<?>> metrics = Maps.newHashMap();
  
    /**
     * Add value from hostname for a metric.
     *
     * @param name String name of metric
     * @param value long value to track
     * @param hostnamePartitionId String host it came from
     * @return this
     */
    public AggregatedMetrics add(String name, long value,
                                 String hostnamePartitionId) {
      AggregatedMetricLong aggregatedMetric =
          (AggregatedMetricLong) metrics.get(name);
      if (aggregatedMetric == null) {
        aggregatedMetric = new AggregatedMetricLong();
        metrics.put(name, aggregatedMetric);
      }
      aggregatedMetric.addItem(value, hostnamePartitionId);
      return this;
    }
  
    /**
     * Add value from hostname for a metric.
     *
     * @param name String name of metric
     * @param value double value to track
     * @param hostnamePartitionId String host it came from
     * @return this
     */
    public AggregatedMetrics add(String name, double value,
                                 String hostnamePartitionId) {
      AggregatedMetricDouble aggregatedMetric =
          (AggregatedMetricDouble) metrics.get(name);
      if (aggregatedMetric == null) {
        aggregatedMetric = new AggregatedMetricDouble();
        metrics.put(name, aggregatedMetric);
      }
      aggregatedMetric.addItem(value, hostnamePartitionId);
      return this;
    }
  
    /**
     * Add metrics from worker.
     *
     * @param workerMetrics WorkerSuperstepMetrics from work
     * @param hostname String hostname of worker
     * @return this
     */
    public AggregatedMetrics add(WorkerSuperstepMetrics workerMetrics,
                                 String hostname) {
      add(GraphTaskManager.TIMER_SUPERSTEP_TIME,
          workerMetrics.getSuperstepTimer(), hostname);
      add(GraphTaskManager.TIMER_COMMUNICATION_TIME,
          workerMetrics.getCommTimer(), hostname);
      add(GraphTaskManager.TIMER_COMPUTE_ALL,
          workerMetrics.getComputeAllTimer(), hostname);
      add(GraphTaskManager.TIMER_TIME_TO_FIRST_MSG,
          workerMetrics.getTimeToFirstMsg(), hostname);
      add(BspServiceWorker.TIMER_WAIT_REQUESTS,
          workerMetrics.getWaitRequestsTimer(), hostname);
      add(OutOfCoreIOCallable.BYTES_LOAD_FROM_DISK,
          workerMetrics.getBytesLoadedFromDisk(), hostname);
      add(OutOfCoreIOCallable.BYTES_STORE_TO_DISK,
         workerMetrics.getBytesStoredOnDisk(), hostname);
     add(OutOfCoreEngine.GRAPH_PERCENTAGE_IN_MEMORY,
         workerMetrics.getGraphPercentageInMemory(), hostname);
     return this;
   }
 
   /**
    * Print the aggregated metrics to the stream provided.
    *
    * @param superstep long number of superstep.
    * @param out PrintStream to write to.
    * @return this
    */
   public AggregatedMetrics print(long superstep, PrintStream out) {
     AggregatedMetric superstepTime = get(GraphTaskManager.TIMER_SUPERSTEP_TIME);
     AggregatedMetric commTime = get(GraphTaskManager.TIMER_COMMUNICATION_TIME);
     AggregatedMetric computeAll = get(GraphTaskManager.TIMER_COMPUTE_ALL);
     AggregatedMetric timeToFirstMsg =
         get(GraphTaskManager.TIMER_TIME_TO_FIRST_MSG);
     AggregatedMetric waitRequestsMicros = get(
         BspServiceWorker.TIMER_WAIT_REQUESTS);
     AggregatedMetric bytesLoaded =
         get(OutOfCoreIOCallable.BYTES_LOAD_FROM_DISK);
     AggregatedMetric bytesStored =
         get(OutOfCoreIOCallable.BYTES_STORE_TO_DISK);
     AggregatedMetric graphInMem =
         get(OutOfCoreEngine.GRAPH_PERCENTAGE_IN_MEMORY);
 
     out.println();
     out.println("--- METRICS: superstep " + superstep + " ---");
     printAggregatedMetric(out, "superstep time", "ms", superstepTime);
     printAggregatedMetric(out, "compute all partitions", "ms", computeAll);
     printAggregatedMetric(out, "network communication time", "ms", commTime);
     printAggregatedMetric(out, "time to first message", "us", timeToFirstMsg);
     printAggregatedMetric(out, "wait requests time", "us", waitRequestsMicros);
     printAggregatedMetric(out, "bytes loaded from disk", "bytes", bytesLoaded);
     printAggregatedMetric(out, "bytes stored to disk", "bytes", bytesStored);
     printAggregatedMetric(out, "graph in mem", "%", graphInMem);
 
     return this;
   }
 
   /**
    * Print batch of lines for AggregatedMetric
    *
    * @param out PrintStream to write to
    * @param header String header to print.
    * @param unit String unit of metric
    * @param aggregatedMetric AggregatedMetric to write
    */
   private void printAggregatedMetric(PrintStream out, String header,
                                      String unit,
                                      AggregatedMetric aggregatedMetric) {
     if (aggregatedMetric.hasData()) {
       out.println(header);
       out.println("  mean: " + aggregatedMetric.mean() + " " + unit);
       printValueFromHost(out, "  smallest: ", unit, aggregatedMetric.min());
       printValueFromHost(out, "  largest: ", unit, aggregatedMetric.max());
     } else {
       out.println(header + ": NO DATA");
     }
   }
 
   /**
    * Print a line for a value with the host it came from.
    *
    * @param out PrintStream to write to
    * @param prefix String to write at beginning
    * @param unit String unit of metric
    * @param vh ValueWithHostname to write
    */
   private void printValueFromHost(PrintStream out, String prefix,
                                   String unit, ValueWithHostname vh) {
     out.println(prefix + vh.getValue() + ' ' + unit +
         " from " + vh.getHostname());
   }
 
   /**
    * Get AggregatedMetric with given name.
    *
    * @param name String metric to lookup.
    * @return AggregatedMetric for given metric name.
    */
   public AggregatedMetric get(String name) {
     return metrics.get(name);
   }
 
   /**
    * Get map of all aggregated metrics.
    *
    * @return Map of all the aggregated metrics.
    */
   public Map<String, AggregatedMetric<?>> getAll() {
     return metrics;
   }
 }