/*
 * 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.extraction.demux.processor.mapper;


import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.chukwa.datacollection.writer.hbase.Annotation.Table;
import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
import org.apache.hadoop.chukwa.extraction.engine.Record;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.log4j.Logger;

@Table(name="Mapreduce",columnFamily="JobLogHistory")
public class JobLogHistoryProcessor extends AbstractProcessor {
  static Logger log = Logger.getLogger(JobLogHistoryProcessor.class);

  private static final String recordType = "JobLogHistory";
  private static final String internalRegex = "(.*?)=\"(.*?)\"(.*)([\\n])?";
  private Pattern ip = null;

  private Matcher internalMatcher = null;

  public JobLogHistoryProcessor() {
    ip = Pattern.compile(internalRegex);
    internalMatcher = ip.matcher("-");
  }

  @Override
  protected void parse(String recordEntry,
      OutputCollector<ChukwaRecordKey, ChukwaRecord> output, Reporter reporter)
      throws Throwable {

    // log.info("JobLogHistoryProcessor record: [" + recordEntry + "] type["
    // + chunk.getDataType() + "]");

    try {

      HashMap<String, String> keys = new HashMap<String, String>();
      ChukwaRecord record = null;

      int firstSep = recordEntry.indexOf(" ");
      keys.put("RECORD_TYPE", recordEntry.substring(0, firstSep));
      // log.info("JobLogHistoryProcessor Add field: [RECORD_TYPE]["
      // + keys.get("RECORD_TYPE") + "]");

      String body = recordEntry.substring(firstSep);

      internalMatcher.reset(body);

      // String fieldName = null;
      // String fieldValue = null;

      while (internalMatcher.matches()) {

        keys.put(internalMatcher.group(1).trim(), internalMatcher.group(2)
            .trim());

        // TODO Remove debug info before production
        // fieldName = internalMatcher.group(1).trim();
        // fieldValue = internalMatcher.group(2).trim();
        // log.info("JobLogHistoryProcessor Add field: [" + fieldName +
        // "][" + fieldValue +"]" );
        // log.info("EOL : [" + internalMatcher.group(3) + "]" );
        internalMatcher.reset(internalMatcher.group(3));
      }

      if (!keys.containsKey("JOBID")) {
        // Extract JobID from taskID
        // JOBID = "job_200804210403_0005"
        // TASKID = "tip_200804210403_0005_m_000018"
        String jobId = keys.get("TASKID");
        int idx1 = jobId.indexOf('_', 0);
        int idx2 = jobId.indexOf('_', idx1 + 1);
        idx2 = jobId.indexOf('_', idx2 + 1);
        keys.put("JOBID", jobId.substring(idx1 + 1, idx2));
        // log.info("JobLogHistoryProcessor Add field: [JOBID]["
        // + keys.get("JOBID") + "]");
      } else {
        String jobId = keys.get("JOBID").replace("_", "").substring(3);
        keys.put("JOBID", jobId);
      }
      // if (keys.get("RECORD_TYPE").equalsIgnoreCase("Job") &&
      // keys.containsKey("SUBMIT_TIME"))
      // {
      // // Job JOBID="job_200804210403_0005" JOBNAME="MY_JOB"
      // USER="userxxx"
      // // SUBMIT_TIME="1208760436751"
      // JOBCONF="/mapredsystem/xxx.yyy.com/job_200804210403_0005/job.xml"
      //					
      //					
      // }
      // else if (keys.get("RECORD_TYPE").equalsIgnoreCase("Job") &&
      // keys.containsKey("LAUNCH_TIME"))
      // {
      // // Job JOBID="job_200804210403_0005" LAUNCH_TIME="1208760437110"
      // TOTAL_MAPS="5912" TOTAL_REDUCES="739"
      //					
      // }
      // else if (keys.get("RECORD_TYPE").equalsIgnoreCase("Job") &&
      // keys.containsKey("FINISH_TIME"))
      // {
      // // Job JOBID="job_200804210403_0005" FINISH_TIME="1208760906816"
      // JOB_STATUS="SUCCESS" FINISHED_MAPS="5912" FINISHED_REDUCES="739"
      // FAILED_MAPS="0" FAILED_REDUCES="0"
      // // COUNTERS="File Systems.Local bytes read:1735053407244,File
      // Systems.Local bytes written:2610106384012,File Systems.HDFS bytes
      // read:801605644910,File Systems.HDFS bytes written:44135800,
      // // Job Counters .Launched map tasks:5912,Job Counters .Launched
      // reduce tasks:739,Job Counters .Data-local map tasks:5573,Job
      // Counters .Rack-local map tasks:316,Map-Reduce Framework.
      // // Map input records:9410696067,Map-Reduce Framework.Map output
      // records:9410696067,Map-Reduce Framework.Map input
      // bytes:801599188816,Map-Reduce Framework.Map output
      // bytes:784427968116,
      // // Map-Reduce Framework.Combine input records:0,Map-Reduce
      // Framework.Combine output records:0,Map-Reduce Framework.Reduce
      // input groups:477265,Map-Reduce Framework.Reduce input
      // records:739000,
      // // Map-Reduce Framework.Reduce output records:739000"
      //					
      // }
      // else
      if (keys.get("RECORD_TYPE").equalsIgnoreCase("MapAttempt")
          && keys.containsKey("START_TIME")) {
        // MapAttempt TASK_TYPE="MAP"
        // TASKID="tip_200804210403_0005_m_000018"
        // TASK_ATTEMPT_ID="task_200804210403_0005_m_000018_0"
        // START_TIME="1208760437531"
        // HOSTNAME="tracker_xxx.yyy.com:xxx.yyy.com/xxx.xxx.xxx.xxx:53734"

        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/Map/" + keys.get("JOBID") + "/"
            + keys.get("START_TIME"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("START_TIME")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("START_TIME", keys.get("START_TIME"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/Map/S");
        output.collect(key, record);

      } else if (keys.get("RECORD_TYPE").equalsIgnoreCase("MapAttempt")
          && keys.containsKey("FINISH_TIME")) {
        // MapAttempt TASK_TYPE="MAP"
        // TASKID="tip_200804210403_0005_m_005494"
        // TASK_ATTEMPT_ID="task_200804210403_0005_m_005494_0"
        // TASK_STATUS="SUCCESS"
        // FINISH_TIME="1208760624124"
        // HOSTNAME="tracker_xxxx.yyyy.com:xxx.yyy.com/xxx.xxx.xxx.xxx:55491"

        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/Map/" + keys.get("JOBID") + "/"
            + keys.get("FINISH_TIME"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("FINISH_TIME")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("FINISH_TIME", keys.get("FINISH_TIME"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/Map/E");
        output.collect(key, record);
      }

      else if (keys.get("RECORD_TYPE").equalsIgnoreCase("ReduceAttempt")
          && keys.containsKey("START_TIME")) {
        // ReduceAttempt TASK_TYPE="REDUCE"
        // TASKID="tip_200804210403_0005_r_000138"
        // TASK_ATTEMPT_ID="task_200804210403_0005_r_000138_0"
        // START_TIME="1208760454885"
        // HOSTNAME="tracker_xxxx.yyyy.com:xxx.yyy.com/xxx.xxx.xxx.xxx:51947"

        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/SHUFFLE/" + keys.get("JOBID") + "/"
            + keys.get("START_TIME"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("START_TIME")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("START_TIME", keys.get("START_TIME"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/SHUFFLE/S");
        output.collect(key, record);

      } else if (keys.get("RECORD_TYPE").equalsIgnoreCase("ReduceAttempt")
          && keys.containsKey("FINISH_TIME")) {
        // ReduceAttempt TASK_TYPE="REDUCE"
        // TASKID="tip_200804210403_0005_r_000138"
        // TASK_ATTEMPT_ID="task_200804210403_0005_r_000138_0"
        // TASK_STATUS="SUCCESS" SHUFFLE_FINISHED="1208760787167"
        // SORT_FINISHED="1208760787354" FINISH_TIME="1208760802395"
        // HOSTNAME="tracker__xxxx.yyyy.com:xxx.yyy.com/xxx.xxx.xxx.xxx:51947"

        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/SHUFFLE/" + keys.get("JOBID") + "/"
            + keys.get("SHUFFLE_FINISHED"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("SHUFFLE_FINISHED")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("SHUFFLE_FINISHED", keys.get("SHUFFLE_FINISHED"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/SHUFFLE/E");
        output.collect(key, record);

        // SORT
        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/SORT/" + keys.get("JOBID") + "/"
            + keys.get("SHUFFLE_FINISHED"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("SHUFFLE_FINISHED")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("START_TIME", keys.get("SHUFFLE_FINISHED"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/SORT/S");
        output.collect(key, record);

        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/SORT/" + keys.get("JOBID") + "/"
            + keys.get("SORT_FINISHED"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("SORT_FINISHED")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("SORT_FINISHED", keys.get("SORT_FINISHED"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/SORT/E");
        output.collect(key, record);

        // Reduce
        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/REDUCE/" + keys.get("JOBID") + "/"
            + keys.get("SORT_FINISHED"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("SORT_FINISHED")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("START_TIME", keys.get("SORT_FINISHED"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/REDUCE/S");
        output.collect(key, record);

        key = new ChukwaRecordKey();
        key.setKey("JobLogHist/REDUCE/" + keys.get("JOBID") + "/"
            + keys.get("FINISH_TIME"));
        key.setReduceType("JobLogHistoryReduceProcessor");
        record = new ChukwaRecord();
        record.setTime(Long.parseLong(keys.get("SORT_FINISHED")));
        record.add("JOBID", keys.get("JOBID"));
        record.add("FINISH_TIME", keys.get("SORT_FINISHED"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("JobLogHist/REDUCE/E");
        output.collect(key, record);

      } else if (keys.get("RECORD_TYPE").equalsIgnoreCase("Job")) {
        // 1
        // Job JOBID="job_200809062051_0001" JOBNAME="wordcount" USER="xxx"
        // SUBMIT_TIME="1208760906812"
        // JOBCONF=
        // "/user/xxx/mapredsystem/563976.yyy.zzz.com/job_200809062051_0001/job.xml"

        // 2
        // Job JOBID="job_200809062051_0001" LAUNCH_TIME="1208760906816"
        // TOTAL_MAPS="3" TOTAL_REDUCES="7"

        // 3
        // Job JOBID="job_200804210403_0005" FINISH_TIME="1208760906826"
        // JOB_STATUS="SUCCESS" FINISHED_MAPS="5912"
        // FINISHED_REDUCES="739" FAILED_MAPS="0" FAILED_REDUCES="0"
        // COUNTERS="File Systems.Local bytes read:1735053407244,File
        // Systems.Local bytes written:2610106384012,File Systems.HDFS
        // bytes read:801605644910,File Systems.HDFS bytes
        // written:44135800,
        // Job Counters .Launched map tasks:5912,Job Counters .Launched
        // reduce tasks:739,Job Counters .Data-local map tasks:5573,Job
        // Counters .Rack-local map tasks:316,Map-Reduce Framework.
        // Map input records:9410696067,Map-Reduce Framework.Map output
        // records:9410696067,Map-Reduce Framework.Map input
        // bytes:801599188816,Map-Reduce Framework.Map output
        // bytes:784427968116,
        // Map-Reduce Framework.Combine input records:0,Map-Reduce
        // Framework.Combine output records:0,Map-Reduce
        // Framework.Reduce input groups:477265,Map-Reduce
        // Framework.Reduce input records:739000,
        // Map-Reduce Framework.Reduce output records:739000"

        record = new ChukwaRecord();
        key = new ChukwaRecordKey();
        buildGenericRecord(record, null, Long
            .parseLong(keys.get("FINISH_TIME")), "MRJob");
        if (keys.containsKey("COUNTERS")) {
          extractCounters(record, keys.get("COUNTERS"));
        }

        key = new ChukwaRecordKey();
        key.setKey("MRJob/" + keys.get("JOBID"));
        key.setReduceType("MRJobReduceProcessor");

        record = new ChukwaRecord();
        record.add(Record.tagsField, chunk.getTags());
        if (keys.containsKey("SUBMIT_TIME")) {
          record.setTime(Long.parseLong(keys.get("SUBMIT_TIME")));
        } else if (keys.containsKey("LAUNCH_TIME")) {
          record.setTime(Long.parseLong(keys.get("LAUNCH_TIME")));
        } else if (keys.containsKey("FINISH_TIME")) {
          record.setTime(Long.parseLong(keys.get("FINISH_TIME")));
        }

        for(Entry<String, String> entry : keys.entrySet()) {
          record.add(entry.getKey(), entry.getValue());
        }

        output.collect(key, record);
      }

      if (keys.containsKey("TASK_TYPE")
          && keys.containsKey("COUNTERS")
          && (keys.get("TASK_TYPE").equalsIgnoreCase("REDUCE") || keys.get(
              "TASK_TYPE").equalsIgnoreCase("MAP"))) {
        // MAP
        // Task TASKID="tip_200804210403_0005_m_000154" TASK_TYPE="MAP"
        // TASK_STATUS="SUCCESS" FINISH_TIME="1208760463883"
        // COUNTERS="File Systems.Local bytes read:159265655,File
        // Systems.Local bytes written:318531310,
        // File Systems.HDFS bytes read:145882417,Map-Reduce
        // Framework.Map input records:1706604,
        // Map-Reduce Framework.Map output records:1706604,Map-Reduce
        // Framework.Map input bytes:145882057,
        // Map-Reduce Framework.Map output bytes:142763253,Map-Reduce
        // Framework.Combine input records:0,Map-Reduce
        // Framework.Combine output records:0"

        // REDUCE
        // Task TASKID="tip_200804210403_0005_r_000524"
        // TASK_TYPE="REDUCE" TASK_STATUS="SUCCESS"
        // FINISH_TIME="1208760877072"
        // COUNTERS="File Systems.Local bytes read:1179319677,File
        // Systems.Local bytes written:1184474889,File Systems.HDFS
        // bytes written:59021,
        // Map-Reduce Framework.Reduce input groups:684,Map-Reduce
        // Framework.Reduce input records:1000,Map-Reduce
        // Framework.Reduce output records:1000"

        record = new ChukwaRecord();
        key = new ChukwaRecordKey();
        buildGenericRecord(record, null, Long
            .parseLong(keys.get("FINISH_TIME")), "SizeVsFinishTime");
        extractCounters(record, keys.get("COUNTERS"));
        record.add("JOBID", keys.get("JOBID"));
        record.add("TASKID", keys.get("TASKID"));
        record.add("TASK_TYPE", keys.get("TASK_TYPE"));
        record.add(Record.tagsField, chunk.getTags());
        // log.info("MR_Graph +1");
        output.collect(key, record);

      }
    } catch (IOException e) {
      log.warn("Unable to collect output in JobLogHistoryProcessor ["
          + recordEntry + "]", e);
      e.printStackTrace();
      throw e;
    }

  }

  protected void extractCounters(ChukwaRecord record, String input) {

    String[] data = null;
    String[] counters = input.split(",");

    for (String counter : counters) {
      data = counter.split(":");
      record.add(data[0].replaceAll(" ", "_").replaceAll("\\.", "_")
          .toUpperCase(), data[1]);
    }
  }

  public String getDataType() {
    return JobLogHistoryProcessor.recordType;
  }
}