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.hadoop.chukwa.caffe;
  
  import java.io.BufferedWriter;
  import java.io.PrintWriter;
  import java.util.Calendar;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Set;
  import java.util.TimeZone;
  import java.util.Timer;
  import java.util.TimerTask;
  
  import org.apache.hadoop.chukwa.datastore.ChukwaHBaseStore;
  import org.apache.hadoop.chukwa.hicc.bean.Series;
  import org.json.simple.JSONObject;
  
  //export CLASSPATH=/opt/apache/hadoop/etc/hadoop:/opt/apache/hbase/conf:/opt/apache/chukwa-0.8.0/share/chukwa/*:/opt/apache/chukwa-0.8.0/share/chukwa/lib/*:$CLASSPATH
  
  public class MetricsCollector
  {
    private Timer getMetricSnapshotTimer = null;
    private long intervalInMilli;
    private String hostname;
    private String dirName;
    
    public MetricsCollector (long intervalInMilli, String hostname, String dirName) {
      this.intervalInMilli = intervalInMilli;
      this.hostname = hostname;
      this.dirName = dirName;
      getMetricSnapshotTimer = new Timer ("GetMetricSnapshot", true);
    }
    
    public void start () {
      if (getMetricSnapshotTimer != null)
        getMetricSnapshotTimer.schedule (new GetMetricSnapshotTimerTask (hostname, intervalInMilli, dirName), 0, intervalInMilli);    
    }
    
    public void cancel ()
    {
      if (getMetricSnapshotTimer != null)
        getMetricSnapshotTimer.cancel ();
    }
    
    class GetMetricSnapshotTimerTask extends TimerTask
    {
      private String hostname = null;
      private BufferedWriter bufferedWriter = null;
      private long intervalInMilli;
      private String dirName;
  
      /**
       * Normalize the timestamp in time series data to use seconds
       */
      private final static int XSCALE = 1000;
  
      GetMetricSnapshotTimerTask (String hostname, long intervalInMilli, String dirName)
      {
        this.hostname = hostname;
        this.intervalInMilli = intervalInMilli;
        this.dirName = dirName;
      }
  
      public void run () 
      {
        TimeZone tz = TimeZone.getTimeZone("UTC");
        Calendar now = Calendar.getInstance(tz);
        long currTime=now.getTimeInMillis();
  
        System.out.println ("currTime in UTC: " + currTime);
        System.out.println ("currTime in current time zone" + System.currentTimeMillis ());
  
        long startTime = currTime - intervalInMilli;
        long endTime = currTime;
        try {
          System.out.println ("About to run");
          getHadoopMetrics (startTime, endTime);
          System.out.println ("Done run");
        } catch (Exception e) {
          e.printStackTrace ();
        }
      }
  
      private void getHadoopMetrics(long startTime, long endTime) throws Exception 
      {
        String source = hostname + ":NodeManager";
        System.out.println ("source: " + source);
        System.out.println ("startTime: " + startTime);
        System.out.println ("endTime: " + endTime); 
        Series series = ChukwaHBaseStore.getSeries ("HadoopMetrics.jvm.JvmMetrics.MemHeapUsedM", source, startTime, endTime);
        String value = series.toString ();
        System.out.println ("value: " + value);
  
        JSONObject jsonObj = (JSONObject) series.toJSONObject ();
        Set set = jsonObj.keySet ();
        Iterator iter = set.iterator ();
        List list = (List) jsonObj.get ("data");
        if (list != null) { 
          int size = list.size ();
          System.out.println ("size: " + size);
          if (size > 0 ) {
            String name = "NodeManager" + "_" + "HadoopMetrics.jvm.JvmMetrics.MemHeapUsedM" + "_" + hostname;
            generateCsv (list, name, startTime, bufferedWriter);
          }
        }
      }
  
      private void generateCsv (List list, String name, long startTime, BufferedWriter bufferedWriter) throws Exception
      {
        String fileName = dirName + "/" + name + "_" + startTime;
        PrintWriter writer = new PrintWriter(fileName + ".csv", "UTF-8");
        int size = list.size ();
        for (int i = 0; i < size; i++) {
          List point = (List) list.get (i);
          long time = (Long) point.get (0) / XSCALE;
          double val = (Double) point.get (1);
          writer.println(time + "," + val);
        }
        writer.close();
      }
    }
  }
  