src/test/java/org/apache/hadoop/chukwa/extraction/demux/processor/mapper/TestJsonProcessors.java

/*
 * 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.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.net.InetAddress;
import java.net.UnknownHostException;

import org.apache.hadoop.chukwa.ChukwaArchiveKey;
import org.apache.hadoop.chukwa.Chunk;
import org.apache.hadoop.chukwa.ChunkImpl;
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.json.simple.JSONArray;
import org.json.simple.JSONObject;

import junit.framework.TestCase;

public class TestJsonProcessors extends TestCase {

	/**
	 * Process the chunk with the passed Processor and compare with the input
	 * JSONObject
	 * 
	 * @param p
	 * @param inData
	 * @param chunk
	 * @return
	 */
	private String testProcessor(AbstractProcessor p, JSONObject inData,
			Chunk chunk) {
		ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord> output = new ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord>();
		p.process(new ChukwaArchiveKey(), chunk, output, null);
		HashMap<ChukwaRecordKey, ChukwaRecord> outData = output.data;
		
		// First get all ChukwaRecords and then get all field-data pairs within
		// each record
		Iterator<Entry<ChukwaRecordKey, ChukwaRecord>> recordIter = outData
				.entrySet().iterator();
		while (recordIter.hasNext()) {
			Entry<ChukwaRecordKey, ChukwaRecord> recordEntry = recordIter
					.next();
			ChukwaRecord value = recordEntry.getValue();
			String[] fields = value.getFields();
			for (String field : fields) {
				//ignore ctags, capps, csource
				if (field.equals(Record.tagsField)
						|| field.equals(Record.applicationField)
						|| field.equals(Record.sourceField)) {
					continue;
				}
				String data = value.getValue(field);
				String expected = String.valueOf(inData.get(field));
				/*System.out.println("Metric, expected data, received data- " +
				 field + ", " + expected + ", " +data);
				*/
				if (!expected.equals(data)) {
					StringBuilder sb = new StringBuilder(
							"Failed to verify metric - ");
					sb.append("field:").append(field);
					sb.append(", expected:").append(expected);
					sb.append(", but received:").append(data);
					return sb.toString();
				}
			}
		}
		return null;
	}

	@SuppressWarnings("unchecked")
	private JSONObject getJSONObject(){
		String csource = "localhost";
		try {
			csource = InetAddress.getLocalHost().getHostName();
		} catch (UnknownHostException e) {
			csource = "localhost";
		}
		JSONObject json = new JSONObject();
		json.put("capp", "Test");
		json.put("csource", csource);
		return json;
	}

	@SuppressWarnings("unchecked")
	public void testJobTrackerProcessor() {
		// test metric for each record type
		JSONObject json = getJSONObject();
		json.put("memHeapUsedM", "286");
		json.put("maps_killed", "3");
		json.put("waiting_maps", "1");
		json.put("RpcProcessingTime_avg_time", "0.003");
		byte[] data = json.toString().getBytes();
		JobTrackerProcessor p = new JobTrackerProcessor();
		ChunkImpl ch = new ChunkImpl("TestType", "Test", data.length, data,
				null);
		String failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);

		// test gauge metric
		json.put("maps_killed", "5");
		data = json.toString().getBytes();
		ch = new ChunkImpl("TestType", "Test", data.length, data, null);
		json.put("maps_killed", "2");
		failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);
	}

	@SuppressWarnings("unchecked")
	public void testNamenodeProcessor() {
		// test metric for each record type
		JSONObject json = getJSONObject();
		json.put("BlocksTotal", "1234");
		json.put("FilesCreated", "33");
		json.put("RpcQueueTime_avg_time", "0.001");
		json.put("gcCount", "112");
		json.put("Transactions_num_ops", "3816");
		byte[] data = json.toString().getBytes();
		NamenodeProcessor p = new NamenodeProcessor();
		ChunkImpl ch = new ChunkImpl("TestType", "Test", data.length, data,
				null);
		String failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);

		// test gauge metric
		json.put("FilesCreated", "55");
		json.put("gcCount", "115");
		data = json.toString().getBytes();
		ch = new ChunkImpl("TestType", "Test", data.length, data, null);
		json.put("FilesCreated", "22");
		json.put("gcCount", "3");
		failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);
	}

	@SuppressWarnings("unchecked")
	public void testDatanodeProcessor() {
		// test metric for each record type
		JSONObject json = getJSONObject();
		json.put("heartBeats_num_ops", "10875");
		json.put("FilesCreated", "33");
		json.put("RpcQueueTime_avg_time", "0.001");
		json.put("gcCount", "112");
		json.put("Capacity", "22926269645");
		byte[] data = json.toString().getBytes();
		DatanodeProcessor p = new DatanodeProcessor();
		ChunkImpl ch = new ChunkImpl("TestType", "Test", data.length, data,
				null);
		String failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);

		// test gauge metric
		json.put("heartBeats_num_ops", "10980");
		json.put("gcCount", "115");
		data = json.toString().getBytes();
		ch = new ChunkImpl("TestType", "Test", data.length, data, null);
		json.put("heartBeats_num_ops", "105");
		json.put("gcCount", "3");
		failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);
	}

	@SuppressWarnings("unchecked")
	public void testHBaseMasterProcessor() {
		// test metric for each record type
		JSONObject json = getJSONObject();
		json.put("splitSizeNumOps", "108");
		json.put("AverageLoad", "3.33");
		byte[] data = json.toString().getBytes();
		HBaseMasterProcessor p = new HBaseMasterProcessor();
		ChunkImpl ch = new ChunkImpl("TestType", "Test", data.length, data,
				null);
		String failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);

		// test gauge metric
		json.put("splitSizeNumOps", "109");
		data = json.toString().getBytes();
		ch = new ChunkImpl("TestType", "Test", data.length, data, null);
		json.put("splitSizeNumOps", "1");
		failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);
	}

	@SuppressWarnings("unchecked")
	public void testHBaseRegionServerProcessor() {
		// test metric for each record type
		JSONObject json = getJSONObject();
		json.put("blockCacheSize", "2681872");
		byte[] data = json.toString().getBytes();
		HBaseMasterProcessor p = new HBaseMasterProcessor();
		ChunkImpl ch = new ChunkImpl("TestType", "Test", data.length, data,
				null);
		String failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);
		// no gauge metrics yet
	}

	@SuppressWarnings("unchecked")
	public void testZookeeperProcessor() {
		// test metric for each record type
		JSONObject json = getJSONObject();
		json.put("packetsSent", "2049");
		json.put("NodeCount", "40");
		byte[] data = json.toString().getBytes();
		ZookeeperProcessor p = new ZookeeperProcessor();
		ChunkImpl ch = new ChunkImpl("TestType", "Test", data.length, data,
				null);
		String failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);

		// test gauge metric
		json.put("packetsSent", "2122");
		data = json.toString().getBytes();
		ch = new ChunkImpl("TestType", "Test", data.length, data, null);
		json.put("packetsSent", "73");
		failMsg = testProcessor(p, json, ch);
		assertNull(failMsg, failMsg);
	}
	
	@SuppressWarnings("unchecked")
	public void testSysteMetricsProcessor() {
		JSONObject system = new JSONObject();
		JSONObject memory = new JSONObject();
		JSONObject cpu1 = new JSONObject();
		JSONObject cpu2 = new JSONObject();
		JSONObject cpu3 = new JSONObject();
		JSONObject cpu4 = new JSONObject();
		JSONObject disk1 = new JSONObject();
		JSONObject disk2 = new JSONObject();
		JSONObject network1 = new JSONObject();
		JSONObject network2 = new JSONObject();

		JSONArray cpu = new JSONArray();
		JSONArray loadAvg = new JSONArray();
		JSONArray disk = new JSONArray();
		JSONArray network = new JSONArray();

		memory.put("Total", "130980773888");
		memory.put("UsedPercent", "4.493927773730516");
		memory.put("FreePercent", "95.50607222626948");
		memory.put("ActualFree", "125094592512");
		memory.put("ActualUsed", "5886181376");
		memory.put("Free", "34487599104");
		memory.put("Used", "96493174784");
		memory.put("Ram", "124920");
		system.put("memory", memory);

		system.put("timestamp", 1353981082318L);
		system.put("uptime", "495307.98");

		cpu1.put("combined", 0.607);
		cpu1.put("user", 0.49);
		cpu1.put("idle", 0.35);
		cpu1.put("sys", 0.116);
		cpu2.put("combined", 0.898);
		cpu2.put("user", 0.69);
		cpu2.put("idle", 0.06);
		cpu2.put("sys", 0.202);
		// include chunks which have null values, to simulate sigar issue on
		// pLinux
		cpu3.put("combined", null);
		cpu3.put("user", null);
		cpu3.put("idle", null);
		cpu3.put("sys", null);
		cpu4.put("combined", "null");
		cpu4.put("user", "null");
		cpu4.put("idle", "null");
		cpu4.put("sys", "null");
		cpu.add(cpu1);
		cpu.add(cpu2);
		cpu.add(cpu3);
		system.put("cpu", cpu);

		loadAvg.add("0.16");
		loadAvg.add("0.09");
		loadAvg.add("0.06");
		system.put("loadavg", loadAvg);

		disk1.put("ReadBytes", 220000000000L);
		disk1.put("Reads", 12994476L);
		disk2.put("ReadBytes", 678910987L);
		disk2.put("Reads", 276L);
		disk.add(disk1);
		disk.add(disk2);
		system.put("disk", disk);

		network1.put("RxBytes", 7234832487L);
		network2.put("RxBytes", 8123023483L);
		network.add(network1);
		network.add(network2);
		system.put("network", network);

		byte[] data = system.toString().getBytes();
		// parse with
		// org.apache.hadoop.chukwa.extraction.demux.processor.mapper.SystemMetrics
		// and verify cpu usage aggregates
		SystemMetrics p = new SystemMetrics();
		ChunkImpl ch = new ChunkImpl("TestType", "Test", data.length, data,
				null);
		ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord> output = new ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord>();
		p.process(new ChukwaArchiveKey(), ch, output, null);
		HashMap<ChukwaRecordKey, ChukwaRecord> outData = output.data;
		Iterator<Entry<ChukwaRecordKey, ChukwaRecord>> recordIter = outData
				.entrySet().iterator();
		while (recordIter.hasNext()) {
			Entry<ChukwaRecordKey, ChukwaRecord> recordEntry = recordIter
					.next();
			ChukwaRecordKey key = recordEntry.getKey();
			ChukwaRecord value = recordEntry.getValue();
			if (value.getValue("combined") != null) {
				assertEquals(Double.parseDouble(value.getValue("combined")),
						0.7525);
				assertEquals(Double.parseDouble(value.getValue("user")), 0.59);
				assertEquals(Double.parseDouble(value.getValue("sys")), 0.159);
				assertEquals(Double.parseDouble(value.getValue("idle")), 0.205);
				System.out.println("CPU metrics verified");
			}
		}

	}
}