content: /**
   * 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.apex.examples.machinedata.operator;
  
  import java.io.Serializable;
  import java.math.BigDecimal;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  
  import javax.validation.constraints.Max;
  import javax.validation.constraints.Min;
  
  import org.apache.apex.examples.machinedata.data.MachineInfo;
  import org.apache.apex.examples.machinedata.data.MachineKey;
  import org.apache.apex.examples.machinedata.data.ResourceType;
  import org.apache.apex.examples.machinedata.util.DataTable;
  import org.apache.apex.malhar.lib.codec.KryoSerializableStreamCodec;
  import org.apache.apex.malhar.lib.util.KeyValPair;
  
  import com.google.common.base.Objects;
  import com.google.common.collect.Lists;
  import com.google.common.collect.Maps;
  
  import com.datatorrent.api.DefaultInputPort;
  import com.datatorrent.api.DefaultOutputPort;
  import com.datatorrent.api.StreamCodec;
  import com.datatorrent.common.util.BaseOperator;
  
  /**
   * <p>
   * CalculatorOperator class.
   * </p>
   *
   * @since 0.3.5
   */
  public class CalculatorOperator extends BaseOperator
  {
  
    private final DataTable<MachineKey, ResourceType, List<Integer>> data = new DataTable<>();
  
    @Min(1)
    @Max(99)
    private int kthPercentile = 95; // kth percentile
    private boolean computePercentile;
    private boolean computeSD;
    private boolean computeMax;
  
    private int percentileThreshold = 80;
    private int sdThreshold = 70;
    private int maxThreshold = 99;
  
    public final transient DefaultInputPort<MachineInfo> dataPort = new DefaultInputPort<MachineInfo>()
    {
      @Override
      public void process(MachineInfo tuple)
      {
        addDataToCache(tuple);
      }
  
      /**
       * Stream codec used for partitioning.
       */
      @Override
      public StreamCodec<MachineInfo> getStreamCodec()
      {
        return new MachineInfoStreamCodec();
      }
    };
  
    public final transient DefaultOutputPort<KeyValPair<MachineKey, Map<ResourceType, Double>>> percentileOutputPort = new DefaultOutputPort<>();
  
    public final transient DefaultOutputPort<KeyValPair<MachineKey, Map<ResourceType, Double>>> sdOutputPort = new DefaultOutputPort<>();
  
    public final transient DefaultOutputPort<KeyValPair<MachineKey, Map<ResourceType, Integer>>> maxOutputPort = new DefaultOutputPort<>();
  
    public transient DefaultOutputPort<String> smtpAlert = new DefaultOutputPort<>();
  
    private void addDataToCache(MachineInfo tuple)
    {
      MachineKey machineKey = tuple.getMachineKey();
      if (!data.containsRow(machineKey)) {
        data.put(machineKey, ResourceType.CPU, Lists.<Integer>newArrayList());
        data.put(machineKey, ResourceType.RAM, Lists.<Integer>newArrayList());
        data.put(machineKey, ResourceType.HDD, Lists.<Integer>newArrayList());
      }
      data.get(machineKey, ResourceType.CPU).add(tuple.getCpu());
      data.get(machineKey, ResourceType.RAM).add(tuple.getRam());
      data.get(machineKey, ResourceType.HDD).add(tuple.getHdd());
    }
  
    @Override
    public void endWindow()
    {
  
      if (computePercentile) {
        for (MachineKey machineKey : data.rowKeySet()) {
          Collections.sort(data.get(machineKey, ResourceType.CPU));
          Collections.sort(data.get(machineKey, ResourceType.RAM));
          Collections.sort(data.get(machineKey, ResourceType.HDD));
  
          Map<ResourceType, Double> percentileData = Maps.newHashMap();
          percentileData.put(ResourceType.CPU, getKthPercentile(data.get(machineKey, ResourceType.CPU)));
          percentileData.put(ResourceType.RAM, getKthPercentile(data.get(machineKey, ResourceType.RAM)));
          percentileData.put(ResourceType.HDD, getKthPercentile(data.get(machineKey, ResourceType.HDD)));
          percentileOutputPort.emit(new KeyValPair<>(machineKey, percentileData));
  
          for (ResourceType resourceType : percentileData.keySet()) {
            double percentileValue = percentileData.get(resourceType);
            if (percentileValue > percentileThreshold) {
              emitAlert(resourceType, machineKey, percentileValue, "Percentile");
            }
          }
        }
      }
      if (computeSD) {
        for (MachineKey machineKey : data.rowKeySet()) {
  
          Map<ResourceType, Double> sdData = Maps.newHashMap();
  
          for (ResourceType resourceType : ResourceType.values()) {
            sdData.put(resourceType, getSD(data.get(machineKey, resourceType)));
          }
          sdOutputPort.emit(new KeyValPair<>(machineKey, sdData));
  
          for (ResourceType resourceType : sdData.keySet()) {
            double sdValue = sdData.get(resourceType);
            if (sdValue > sdThreshold) {
              emitAlert(resourceType, machineKey, sdValue, "SD");
            }
          }
        }
      }
      if (computeMax) {
        for (MachineKey machineKey : data.rowKeySet()) {
  
          Map<ResourceType, Integer> maxData = Maps.newHashMap();
          maxData.put(ResourceType.CPU, Collections.max(data.get(machineKey, ResourceType.CPU)));
          maxData.put(ResourceType.RAM, Collections.max(data.get(machineKey, ResourceType.RAM)));
          maxData.put(ResourceType.HDD, Collections.max(data.get(machineKey, ResourceType.HDD)));
  
          maxOutputPort.emit(new KeyValPair<>(machineKey, maxData));
  
          for (ResourceType resourceType : maxData.keySet()) {
            double sdValue = maxData.get(resourceType).doubleValue();
            if (sdValue > maxThreshold) {
              emitAlert(resourceType, machineKey, sdValue, "Max");
            }
          }
        }
      }
      data.clear();
    }
  
    private void emitAlert(ResourceType type, MachineKey machineKey, double alertVal, String prefix)
    {
      BigDecimal decimalVal = new BigDecimal(alertVal);
      decimalVal = decimalVal.setScale(2, BigDecimal.ROUND_HALF_UP);
      String alertTime = machineKey.getDay() + machineKey.getTimeKey();
      smtpAlert.emit(prefix + "-" + type.toString().toUpperCase() + " alert at " + alertTime + " " + type + " usage breached current usage: " + decimalVal.doubleValue() + "% threshold: " + percentileThreshold + "%\n\n" + machineKey);
    }
  
    private double getKthPercentile(List<Integer> sorted)
    {
  
      double val = (kthPercentile * sorted.size()) / 100.0;
      if (val == (int)val) {
        // Whole number
        int idx = (int)val - 1;
        return (sorted.get(idx) + sorted.get(idx + 1)) / 2.0;
      } else {
        int idx = (int)Math.round(val) - 1;
        return sorted.get(idx);
      }
    }
  
    private double getSD(List<Integer> data)
    {
      int sum = 0;
      for (int i : data) {
        sum += i;
      }
      double avg = sum / (data.size() * 1.0);
      double sd = 0;
      for (Integer point : data) {
        sd += Math.pow(point - avg, 2);
      }
      return Math.sqrt(sd);
    }
  
    /**
     * @param kVal the percentile which will be emitted by this operator
     */
    public void setKthPercentile(int kVal)
    {
      this.kthPercentile = kVal;
    }
  
    /**
     * @param doCompute when true percentile will be computed
     */
    public void setComputePercentile(boolean doCompute)
    {
      this.computePercentile = doCompute;
    }
  
    /**
     * @param doCompute when true standard deviation will be computed
     */
    public void setComputeSD(boolean doCompute)
    {
      this.computeSD = doCompute;
    }
  
    /**
     * @param doCompute when true max will be computed
     */
    public void setComputeMax(boolean doCompute)
    {
      this.computeMax = doCompute;
    }
  
    /**
     * @param threshold for percentile when breached will cause alert
     */
    public void setPercentileThreshold(int threshold)
    {
      this.percentileThreshold = threshold;
    }
  
    /**
     * @param threshold for standard deviation when breached will cause alert
     */
    public void setSDThreshold(int threshold)
    {
      this.sdThreshold = threshold;
    }
  
    /**
     * @param threshold for Max when breached will cause alert
     */
    public void setMaxThreshold(int threshold)
    {
      this.maxThreshold = threshold;
    }
  
    public static class MachineInfoStreamCodec extends KryoSerializableStreamCodec<MachineInfo> implements Serializable
    {
      public MachineInfoStreamCodec()
      {
        super();
      }
  
      @Override
      public int getPartition(MachineInfo o)
      {
        return Objects.hashCode(o.getMachineKey().getCustomer(), o.getMachineKey().getOs(), o.getMachineKey().getProduct(), o.getMachineKey().getSoftware1(), o.getMachineKey().getSoftware2(), o.getMachineKey().getSoftware3());
      }
  
      private static final long serialVersionUID = 201411031403L;
    }
  }
  