1/*2 * Licensed to the Apache Software Foundation (ASF) under one3 * or more contributor license agreements. See the NOTICE file4 * distributed with this work for additional information5 * regarding copyright ownership. The ASF licenses this file6 * to you under the Apache License, Version 2.0 (the7 * "License"); you may not use this file except in compliance8 * with the License. You may obtain a copy of the License at9 *10 * http://www.apache.org/licenses/LICENSE-2.011 *12 * Unless required by applicable law or agreed to in writing, software13 * distributed under the License is distributed on an "AS IS" BASIS,14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.15 * See the License for the specific language governing permissions and16 * limitations under the License.17 */1819package org.apache.hadoop.chukwa.extraction.demux.processor.mapper;
2021import java.util.Iterator;
22import java.util.Map;
23import java.util.Map.Entry;
24import java.util.Set;
2526import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
27import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
28import org.apache.hadoop.mapred.OutputCollector;
29import org.apache.hadoop.mapred.Reporter;
30import org.apache.log4j.Logger;
31import org.json.simple.JSONObject;
32import org.json.simple.JSONValue;
3334publicclassLog4JMetricsContextProcessorextendsAbstractProcessor {
3536static Logger log = Logger.getLogger(Log4JMetricsContextProcessor.class);
3738 @Override
39protectedvoid parse(String recordEntry,
40 OutputCollector<ChukwaRecordKey, ChukwaRecord> output, Reporter reporter)
41throws Throwable
42 {
43Log4JMetricsContextChukwaRecord record = newLog4JMetricsContextChukwaRecord(recordEntry);
44ChukwaRecord chukwaRecord = record.getChukwaRecord();
45this.buildGenericRecord(chukwaRecord, null, record.getTimestamp(), record.getRecordType());
46 output.collect(key, chukwaRecord);
47 }
4849// create a static class to cove most of the code for unit test 50staticclassLog4JMetricsContextChukwaRecord {
51private String recordType = null;
52privatelong timestamp = 0;
53privateChukwaRecord chukwaRecord = newChukwaRecord();
5455 @SuppressWarnings("unchecked")
56publicLog4JMetricsContextChukwaRecord(String recordEntry) throws Throwable {
57LogEntry log = newLogEntry(recordEntry);
58 JSONObject json = (JSONObject) JSONValue.parse(log.getBody());
5960// round timestamp61 timestamp = (Long) json.get("timestamp");
62 timestamp = (timestamp / 60000) * 60000;
6364// get record type65 String contextName = (String) json.get("contextName");
66 String recordName = (String) json.get("recordName");
67 recordType = contextName;
68if (!contextName.equals(recordName)) {
69 recordType += "_" + recordName;
70 }
7172for(Entry<String, Object> entry : (Set<Map.Entry>) json.entrySet()) {
73 String key = entry.getKey();
74 String value = String.valueOf(entry.getValue());
75if(value != null) {
76 chukwaRecord.add(key, value);
77 }
78 }
79 }
8081public String getRecordType() {
82return recordType;
83 }
8485publiclong getTimestamp() {
86return timestamp;
87 }
8889publicChukwaRecord getChukwaRecord() {
90return chukwaRecord;
91 }
92 }
93 }
94