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.
   */
  
  /**
   * Demux parser for system metrics data collected through
   * org.apache.hadoop.chukwa.datacollection.adaptor.sigar.SystemMetrics.
   */
  package org.apache.hadoop.chukwa.extraction.demux.processor.mapper;
  
  import java.util.Calendar;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.TimeZone;
  
  import org.apache.hadoop.chukwa.datacollection.writer.hbase.Annotation.Table;
  import org.apache.hadoop.chukwa.datacollection.writer.hbase.Annotation.Tables;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
  import org.apache.hadoop.mapred.OutputCollector;
  import org.apache.hadoop.mapred.Reporter;
  import org.json.simple.JSONArray;
  import org.json.simple.JSONObject;
  import org.json.simple.JSONValue;
  
  @Tables(annotations={
      @Table(name="SystemMetrics",columnFamily="cpu"),
      @Table(name="SystemMetrics",columnFamily="system"),
      @Table(name="SystemMetrics",columnFamily="memory"),
      @Table(name="SystemMetrics",columnFamily="network"),
      @Table(name="SystemMetrics",columnFamily="disk")
      })
  public class SystemMetrics extends AbstractProcessor {
  
    @Override
    protected void parse(String recordEntry,
        OutputCollector<ChukwaRecordKey, ChukwaRecord> output, Reporter reporter)
        throws Throwable {
      JSONObject json = (JSONObject) JSONValue.parse(recordEntry);
      long timestamp = ((Long)json.get("timestamp")).longValue();
      ChukwaRecord record = new ChukwaRecord();
      Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
      cal.setTimeInMillis(timestamp);
      cal.set(Calendar.SECOND, 0);
      cal.set(Calendar.MILLISECOND, 0);
      JSONArray cpuList = (JSONArray) json.get("cpu");
      double combined = 0.0;
      double user = 0.0;
      double sys = 0.0;
      double idle = 0.0;
      int actualSize = 0;
      for(int i = 0; i< cpuList.size(); i++) {
        JSONObject cpu = (JSONObject) cpuList.get(i);
        //Work around for sigar returning null sometimes for cpu metrics on pLinux
        if(cpu.get("combined") == null){
      	  continue;
        }
        actualSize++;
        combined = combined + Double.parseDouble(cpu.get("combined").toString());
        user = user + Double.parseDouble(cpu.get("user").toString());
        sys = sys + Double.parseDouble(cpu.get("sys").toString());
        idle = idle + Double.parseDouble(cpu.get("idle").toString());
        @SuppressWarnings("unchecked")
        Iterator<Map.Entry<String, ?>> keys = cpu.entrySet().iterator();
        while(keys.hasNext()) {
          Map.Entry<String, ?> entry = keys.next();
          String key = entry.getKey();
          Object value = entry.getValue();
          record.add(key + "." + i, value.toString());
        }
      }
      combined = combined / actualSize;
      user = user / actualSize;
      sys = sys / actualSize;
      idle = idle / actualSize;
      record.add("combined", Double.toString(combined));
      record.add("user", Double.toString(user));
      record.add("idle", Double.toString(idle));    
      record.add("sys", Double.toString(sys));
      buildGenericRecord(record, null, cal.getTimeInMillis(), "cpu");
      output.collect(key, record);    
  
      record = new ChukwaRecord();
      record.add("Uptime", json.get("uptime").toString());
      JSONArray loadavg = (JSONArray) json.get("loadavg");
      record.add("LoadAverage.1", loadavg.get(0).toString());
      record.add("LoadAverage.5", loadavg.get(1).toString());
      record.add("LoadAverage.15", loadavg.get(2).toString());
      buildGenericRecord(record, null, cal.getTimeInMillis(), "system");
      output.collect(key, record);    
  
      record = new ChukwaRecord();
      JSONObject memory = (JSONObject) json.get("memory");
      @SuppressWarnings("unchecked")
      Iterator<Map.Entry<String, ?>> memKeys = memory.entrySet().iterator();
      while(memKeys.hasNext()) {
        Map.Entry<String, ?> entry = memKeys.next();
        String key = entry.getKey();
        Object value = entry.getValue();
        record.add(key, value.toString());
      }
      buildGenericRecord(record, null, cal.getTimeInMillis(), "memory");
      output.collect(key, record);    
  
      record = new ChukwaRecord();
      JSONObject swap = (JSONObject) json.get("swap");
      @SuppressWarnings("unchecked")
      Iterator<Map.Entry<String, ?>> swapKeys = swap.entrySet().iterator();
      while(swapKeys.hasNext()) {
        Map.Entry<String, ?> entry = swapKeys.next();
        String key = entry.getKey();
        Object value = entry.getValue();
        record.add(key, value.toString());
      }
      buildGenericRecord(record, null, cal.getTimeInMillis(), "swap");
      output.collect(key, record);
      
      double rxBytes = 0;
      double rxDropped = 0;
      double rxErrors = 0;
      double rxPackets = 0;
      double txBytes = 0;
      double txCollisions = 0;
      double txErrors = 0;
      double txPackets = 0;
      record = new ChukwaRecord();
      JSONArray netList = (JSONArray) json.get("network");
      for(int i = 0;i < netList.size(); i++) {
        JSONObject netIf = (JSONObject) netList.get(i);
        @SuppressWarnings("unchecked")
        Iterator<Map.Entry<String, ?>> keys = netIf.entrySet().iterator();
        while(keys.hasNext()) {
          Map.Entry<String, ?> entry = keys.next();
          String key = entry.getKey();
          Object value = entry.getValue();
          record.add(key + "." + i, value.toString());
          if(i!=0) {
            if(key.equals("RxBytes")) {
              rxBytes = rxBytes + (Long) value;
            } else if(key.equals("RxDropped")) {
              rxDropped = rxDropped + (Long) value;
            } else if(key.equals("RxErrors")) {          
              rxErrors = rxErrors + (Long) value;
            } else if(key.equals("RxPackets")) {
              rxPackets = rxPackets + (Long) value;
            } else if(key.equals("TxBytes")) {
              txBytes = txBytes + (Long) value;
            } else if(key.equals("TxCollisions")) {
              txCollisions = txCollisions + (Long) value;
            } else if(key.equals("TxErrors")) {
              txErrors = txErrors + (Long) value;
            } else if(key.equals("TxPackets")) {
              txPackets = txPackets + (Long) netIf.get(key);
            }
          }
        }
      }
      buildGenericRecord(record, null, cal.getTimeInMillis(), "network");
      record.add("RxBytes", Double.toString(rxBytes));
      record.add("RxDropped", Double.toString(rxDropped));
      record.add("RxErrors", Double.toString(rxErrors));
      record.add("RxPackets", Double.toString(rxPackets));
      record.add("TxBytes", Double.toString(txBytes));
      record.add("TxCollisions", Double.toString(txCollisions));
      record.add("TxErrors", Double.toString(txErrors));
      record.add("TxPackets", Double.toString(txPackets));
      output.collect(key, record);    
      
      double readBytes = 0;
      double reads = 0;
      double writeBytes = 0;
      double writes = 0;
      double total = 0;
      double used = 0;
      record = new ChukwaRecord();
      JSONArray diskList = (JSONArray) json.get("disk");
      for(int i = 0;i < diskList.size(); i++) {
        JSONObject disk = (JSONObject) diskList.get(i);
        @SuppressWarnings("unchecked")
        Iterator<Map.Entry<String, ?>> keys = disk.entrySet().iterator();
        while(keys.hasNext()) {
          Map.Entry<String, ?> entry = keys.next();
          String key = entry.getKey();
          Object value = entry.getValue();
          record.add(key + "." + i, value.toString());
          if(key.equals("ReadBytes")) {
            readBytes = readBytes + (Long) value;
          } else if(key.equals("Reads")) {
            reads = reads + (Long) value;
          } else if(key.equals("WriteBytes")) {
            writeBytes = writeBytes + (Long) value;
          } else if(key.equals("Writes")) {
            writes = writes + (Long) value;
          }  else if(key.equals("Total")) {
            total = total + (Long) value;
          } else if(key.equals("Used")) {
            used = used + (Long) value;
          }
        }
      }
      double percentUsed = used/total; 
      record.add("ReadBytes", Double.toString(readBytes));
      record.add("Reads", Double.toString(reads));
      record.add("WriteBytes", Double.toString(writeBytes));
      record.add("Writes", Double.toString(writes));
      record.add("Total", Double.toString(total));
      record.add("Used", Double.toString(used));
      record.add("PercentUsed", Double.toString(percentUsed));
      buildGenericRecord(record, null, cal.getTimeInMillis(), "disk");
      output.collect(key, record);
      
      record = new ChukwaRecord();
      record.add("cluster", chunk.getTag("cluster"));
      buildGenericRecord(record, null, cal.getTimeInMillis(), "tags");
      output.collect(key, record);
    }
  
  }
  