/*
 *
 * Licensed 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 com.linkedin.drelephant.tez;

import com.linkedin.drelephant.analysis.*;
import com.linkedin.drelephant.configurations.aggregator.AggregatorConfigurationData;
import com.linkedin.drelephant.tez.data.TezApplicationData;
import org.apache.commons.io.FileUtils;
import org.apache.log4j.Logger;

/**
 * Aggregates task level metrics to application
 */

public class TezMetricsAggregator implements HadoopMetricsAggregator {

  private static final Logger logger = Logger.getLogger(TezMetricsAggregator.class);

  private static final String TEZ_CONTAINER_CONFIG = "hive.tez.container.size";
  private static final String MAP_CONTAINER_CONFIG = "mapreduce.map.memory.mb";
  private static final String REDUCER_CONTAINER_CONFIG = "mapreduce.reduce.memory.mb";
  private static final String REDUCER_SLOW_START_CONFIG = "mapreduce.job.reduce.slowstart.completedmaps";
  private static final long CONTAINER_MEMORY_DEFAULT_BYTES = 2048L * FileUtils.ONE_MB;

  private HadoopAggregatedData _hadoopAggregatedData = null;
  private TezTaskLevelAggregatedMetrics _mapTasks;
  private TezTaskLevelAggregatedMetrics _reduceTasks;

  private AggregatorConfigurationData _aggregatorConfigurationData;

  public TezMetricsAggregator(AggregatorConfigurationData _aggregatorConfigurationData) {
    this._aggregatorConfigurationData = _aggregatorConfigurationData;
    _hadoopAggregatedData = new HadoopAggregatedData();
  }

  @Override
  public void aggregate(HadoopApplicationData hadoopData) {

    TezApplicationData data = (TezApplicationData) hadoopData;

    long mapTaskContainerSize = getMapContainerSize(data);
    long reduceTaskContainerSize = getReducerContainerSize(data);

    int reduceTaskSlowStartPercentage =
        (int) (Double.parseDouble(data.getConf().getProperty(REDUCER_SLOW_START_CONFIG)) * 100);


    //overwrite reduceTaskSlowStartPercentage to 100%. TODO: make use of the slow start percent
    reduceTaskSlowStartPercentage = 100;

    _mapTasks = new TezTaskLevelAggregatedMetrics(data.getMapTaskData(), mapTaskContainerSize, data.getStartTime());

    long reduceIdealStartTime = _mapTasks.getNthPercentileFinishTime(reduceTaskSlowStartPercentage);

    // Mappers list is empty
    if(reduceIdealStartTime == -1) {
      // ideal start time for reducer is infinite since it cannot start
      reduceIdealStartTime = Long.MAX_VALUE;
    }

    _reduceTasks = new TezTaskLevelAggregatedMetrics(data.getReduceTaskData(), reduceTaskContainerSize, reduceIdealStartTime);

    _hadoopAggregatedData.setResourceUsed(_mapTasks.getResourceUsed() + _reduceTasks.getResourceUsed());
    _hadoopAggregatedData.setTotalDelay(_mapTasks.getDelay() + _reduceTasks.getDelay());
    _hadoopAggregatedData.setResourceWasted(_mapTasks.getResourceWasted() + _reduceTasks.getResourceWasted());
  }

  @Override
  public HadoopAggregatedData getResult() {
    return _hadoopAggregatedData;
  }

  private long getMapContainerSize(HadoopApplicationData data) {
    try {
      long mapContainerSize = Long.parseLong(data.getConf().getProperty(TEZ_CONTAINER_CONFIG));
      if (mapContainerSize > 0)
        return mapContainerSize;
      else
        return Long.parseLong(data.getConf().getProperty(MAP_CONTAINER_CONFIG));
    } catch ( NumberFormatException ex) {
      return CONTAINER_MEMORY_DEFAULT_BYTES;
    }
  }

  private long getReducerContainerSize(HadoopApplicationData data) {
    try {
      long reducerContainerSize = Long.parseLong(data.getConf().getProperty(TEZ_CONTAINER_CONFIG));
      if (reducerContainerSize > 0)
        return reducerContainerSize;
      else
        return Long.parseLong(data.getConf().getProperty(REDUCER_CONTAINER_CONFIG));
    } catch ( NumberFormatException ex) {
      return CONTAINER_MEMORY_DEFAULT_BYTES;
    }
  }
}

@@ -3538,31 +3538,179 @@
     
-long mapContainerSize =
+// Trying to get container size from tez config, if not found trying from MapReduce config%0A      long mapContainerSize = data.getConf().containsKey(TEZ_CONTAINER_CONFIG) ?
  Lon
@@ -3766,32 +3766,37 @@
 NER_CONFIG))
+ : -1
 ;%0A      
 if (mapConta
@@ -3783,20 +3783,23 @@
 ;%0A      
-if (
+return 
 mapConta
@@ -3810,32 +3810,18 @@
 Size %3E 0
-)%0A        return
+ ?
  mapCont
@@ -3833,35 +3833,10 @@
 Size
-;%0A      else%0A        return
+ :
  Lon
@@ -4085,35 +4085,183 @@
     
-long reducerContainerSize =
+// Trying to get container size from tez config, if not found trying from MapReduce config%0A      long reducerContainerSize = data.getConf().containsKey(TEZ_CONTAINER_CONFIG) ?
  Lon
@@ -4321,24 +4321,29 @@
 CONFIG))
+ : -1
 ;%0A      
 if (redu
@@ -4338,12 +4338,15 @@
     
-if (
+return 
 redu
@@ -4369,24 +4369,10 @@
  %3E 0
-)%0A        return
+ ?
  red
@@ -4392,35 +4392,9 @@
 Size
-;%0A      else%0A        return
+:
  Lon

Fix wrong metrics for Pig job using TEZ (#479)