HADOOP-4431. Add versionning/tags to Chukwa Chunk.

(Jerome Boulon via Johan)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@707328 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -45,6 +45,9 @@ Trunk (unreleased changes)
 
     HADOOP-4284. Support filters that apply to all requests, or global filters,
     to HttpServer. (Kan Zhang via cdouglas)
+    
+    HADOOP-4431. Add versionning/tags to Chukwa Chunk. 
+    (Jerome Boulon via Johan)
 
   OPTIMIZATIONS
 

src/contrib/chukwa/build.xml

@@ -19,22 +19,48 @@
 
 <project name="chukwa" default="main">
 
+  <import file="../build-contrib.xml"/>
+
+
 	<property name="chukwaVersion" value="0.0.1"/>
 
 	<property name="lib.dir" value="${basedir}/lib"/>
 	<property name="src.dir" value="${basedir}/src"/>
+        <property name="build.classes" value="${build.dir}/classes"/>
 	<property name="conf.dir" value="${basedir}/conf"/>
 	<property name="docs.dir" value="${basedir}/docs"/>
 	<property name="tools.dir" value="${basedir}/tools"/>
 	<property name="dist.dir" value="${basedir}/dist"/>
 	<property name="opt.dir" value="${basedir}/opt"/>
         <property name="javac.debug" value="on"/>
-
-	<import file="../build-contrib.xml"/>
+        <property name="javac.version" value="1.6"/>
+       
+        <property name="test.src.dir" value="${basedir}/src/test"/>
+        <property name="test.lib.dir" value="${basedir}/src/test/lib"/>
+        <property name="test.build.dir" value="${build.dir}/test"/>
+        <property name="test.generated.dir" value="${test.build.dir}/src"/>
+        <property name="test.build.data" value="${test.build.dir}/data"/>
+        <property name="test.cache.data" value="${test.build.dir}/cache"/>
+        <property name="test.debug.data" value="${test.build.dir}/debug"/>
+        <property name="test.log.dir" value="${test.build.dir}/logs"/>
+        <property name="test.build.classes" value="${test.build.dir}/classes"/>
+        <property name="test.build.testjar" value="${test.build.dir}/testjar"/>
+        <property name="test.include" value="Test*"/>
+        <property name="test.classpath.id" value="test.classpath"/>
+        <property name="test.output" value="yes"/>
+        <property name="test.timeout" value="900000"/>
+        <property name="test.junit.output.format" value="plain"/>
+        <property name="test.junit.fork.mode" value="perTest" />
+        <property name="test.junit.printsummary" value="yes" />
+        <property name="test.junit.haltonfailure" value="yes" />
+        <property name="test.junit.maxmemory" value="256m" />
+ 
 
 	<target name="init">
 		<mkdir dir="${build.dir}"/>
+		<mkdir dir="${build.classes}"/>
 		<mkdir dir="${build.dir}/test"/>
+
 		<!-- 
     	<copy todir="${build.dir}">
             <fileset dir="${basedir}">
@@ -59,17 +85,50 @@
 	</target>
 
 	<target name="compile" depends="init" description="Compilation target">
-		<javac srcdir="src/java/org/apache/hadoop/chukwa" destdir="${build.dir}" excludes="**/ChukwaTTInstru.java" debug="${javac.debug}">
+		<javac srcdir="src/java/org/apache/hadoop/chukwa" destdir="${build.classes}" excludes="**/ChukwaTTInstru.java" debug="${javac.debug}">
 			<classpath refid="chukwaHadoopClasspath" />
 		</javac>
 	</target>
 
-	<target name="test" depends="init" description="Test target">
-		<javac srcdir="src/test/org/apache/hadoop/chukwa" destdir="${build.dir}" excludes="**/TestDemux.java" debug="${javac.debug}">
-			<classpath refid="chukwaHadoopClasspath" />
-		</javac>
+	<target name="compile-test" depends="init" description="Test target">
+	                      
+           <delete dir="${test.build.dir}"/>
+           <mkdir dir="${test.build.dir}"/>
+           <delete dir="${test.log.dir}"/>
+           <mkdir dir="${test.log.dir}"/>
+           <delete dir="${test.build.classes}"/>
+           <mkdir dir="${test.build.classes}"/>
+
+            <javac srcdir="${test.src.dir}/org/apache/hadoop/chukwa" destdir="${test.build.dir}/classes" excludes="**/TestDemux.java" debug="${javac.debug}">
+                 <classpath refid="chukwaTestClasspath" />
+            </javac>
+        </target>
+
+
+
+	<target name="test" depends="compile,compile-test,test-chunk" description="Chukwa Tests">
 	</target>
 
+	<target name="test-chunk" depends="compile,compile-test" description="Test chunk">
+               
+          <junit showoutput="yes"
+            fork="yes"
+            printsummary="${test.junit.printsummary}"
+            forkmode="${test.junit.fork.mode}"
+            maxmemory="${test.junit.maxmemory}"
+            dir="${test.build.dir}/classes/" timeout="${test.timeout}"
+            errorProperty="tests.failed" failureProperty="tests.failed">
+           <classpath refid="chukwaTestClasspath"/>
+           <sysproperty key="test.src.dir" value="${test.src.dir}"/>
+           <formatter type="${test.junit.output.format}" />
+           <batchtest todir="${test.build.dir}" >
+            <fileset dir="${test.src.dir}" includes="org/apache/hadoop/chukwa/*.java"/>
+           </batchtest>
+         </junit>
+         <fail if="tests.failed">Tests failed!</fail>
+
+        </target>
+
 	<target name="collector" depends="compile" description="Prepare collector.war">
 		<mkdir dir="${build.dir}/collector"/>
 		<mkdir dir="${build.dir}/collector/WEB-INF"/>
@@ -359,5 +418,18 @@
 			<exclude name="**/excluded/" />
 		</fileset>
 	</path>
+        <path id="chukwaTestClasspath">
+                <pathelement location="${hadoop.root}/build/classes"/>
+                <pathelement location="${hadoop.root}/build/contrib/chukwa/classes/"/>
+                <pathelement location="${hadoop.root}/build/contrib/chukwa/test/classes/"/>
+                <fileset dir="${lib.dir}">
+                        <include name="**/*.jar" />
+                        <exclude name="**/excluded/" />
+                </fileset>
+                <fileset dir="${hadoop.root}/lib">
+                        <include name="**/*.jar" />
+                        <exclude name="**/excluded/" />
+                </fileset>
+        </path>
 
 </project>

src/contrib/chukwa/conf/chukwa-agent-conf.xml

@@ -17,4 +17,11 @@
     <description>The hostname of the agent on this node. Usually localhost, this is used by the chukwa instrumentation agent-control interface library</description>
   </property>
 
+  <property>
+    <name>chukwaAgent.tags</name>
+    <value>cluster="demo"</value>
+    <description>The cluster's name for this agent</description>
+  </property>
+
+
 </configuration>

src/contrib/chukwa/conf/chukwa-agent-conf.xml.template

@@ -17,6 +17,12 @@
     <description>The hostname of the agent on this node. Usually localhost, this is used by the chukwa instrumentation agent-control interface library</description>
   </property>
 
+  <property>
+    <name>chukwaAgent.tags</name>
+    <value>cluster="demo"</value>
+    <description>The cluster's name for this agent</description>
+  </property>
+
   <property>
     <name>chukwaAgent.checkpoint.name</name>
     <value>chukwa_agent_checkpoint</value>

src/contrib/chukwa/src/java/org/apache/hadoop/chukwa/Chunk.java

@@ -92,5 +92,17 @@ public interface Chunk {
    */
   public int getSerializedSizeEstimate();
   
+/**
+ * @return tags from this chunk.
+ * 
+ */
+  public String getTags();  
+  
+  /**
+   * Set tags to this chunk.
+   * 
+   */
+    public void setTags(String tags);  
+  
   public void write(DataOutput data) throws IOException;
 }

src/contrib/chukwa/src/java/org/apache/hadoop/chukwa/ChunkImpl.java

@@ -23,25 +23,28 @@ import java.io.DataOutput;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
-import java.util.ArrayList;
 
 import org.apache.hadoop.chukwa.datacollection.adaptor.Adaptor;
+import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
 
 public class ChunkImpl implements org.apache.hadoop.io.Writable, Chunk 
 {
-  
+  public static int PROTOCOL_VERSION=1;
+	
   private String source = "";
   private String application = "";
   private String dataType = "";
+  private String tags = "";
   private byte[] data = null;
   private int[] recordEndOffsets;
+  private int protocolVersion=1;
+  private long seqID;
   
   private String debuggingInfo="";
-  
   private transient Adaptor initiator;
-  long seqID;
   
   ChunkImpl() {
+    this.tags = ChukwaAgent.getTags();
   }
   
   public static ChunkImpl getBlankChunk() {
@@ -50,11 +53,12 @@ public class ChunkImpl implements org.apache.hadoop.io.Writable, Chunk
   
   public ChunkImpl(String dataType, String streamName, long seq, byte[] data, Adaptor source) {
     this.seqID = seq;
+    this.source = localHostAddr;
+    this.tags = ChukwaAgent.getTags();
     this.application = streamName;
     this.dataType = dataType;
     this.data = data;
     this.initiator = source;
-    this.source = localHostAddr;
   }
   
   /**
@@ -109,6 +113,13 @@ public class ChunkImpl implements org.apache.hadoop.io.Writable, Chunk
     seqID=l;
   }
   
+  public int getProtocolVersion() {
+	  return protocolVersion;
+  }
+  
+  public void setProtocolVersion(int pv) {
+	  this.protocolVersion = pv;
+  }
   public String getApplication(){
     return application;
   }
@@ -149,12 +160,30 @@ public class ChunkImpl implements org.apache.hadoop.io.Writable, Chunk
     dataType = t;
   }
   
+  @Override
+  public void setTags(String tags)
+  {
+  	this.tags = tags;
+  }
+  
+/**
+ * @see org.apache.hadoop.chukwa.Chunk#getTags()
+ */
+  public String getTags() {
+    return tags;
+  }
+  
   /**
    * @see org.apache.hadoop.io.Writable#readFields(java.io.DataInput)
    */
   public void readFields(DataInput in) throws IOException {
+	setProtocolVersion(in.readInt());
+	if(protocolVersion!=PROTOCOL_VERSION) {
+		throw new IOException("Protocol version mismatched, drop data.  source version: "+protocolVersion+", collector version:"+PROTOCOL_VERSION);
+	}
     setSeqID(in.readLong());
     setSource(in.readUTF());
+    tags =  in.readUTF();    //no public set method here
     setApplication(in.readUTF());
     setDataType(in.readUTF());
     setDebugInfo(in.readUTF());
@@ -167,12 +196,15 @@ public class ChunkImpl implements org.apache.hadoop.io.Writable, Chunk
     in.readFully(data);
     
   }
+
   /**
    * @see org.apache.hadoop.io.Writable#write(java.io.DataOutput)
    */
   public void write(DataOutput out) throws IOException {
+	out.writeInt(PROTOCOL_VERSION);
     out.writeLong(seqID);
     out.writeUTF(source);
+    out.writeUTF(tags);
     out.writeUTF(application);
     out.writeUTF(dataType);
     out.writeUTF(debuggingInfo);

src/contrib/chukwa/src/java/org/apache/hadoop/chukwa/datacollection/agent/ChukwaAgent.java

@@ -40,7 +40,8 @@ public class ChukwaAgent
 {
   boolean DO_CHECKPOINT_RESTORE = false;
   boolean WRITE_CHECKPOINTS = false;
-
+ 
+  static String tags = "";
   static Logger log = Logger.getLogger(ChukwaAgent.class);
 
   //doesn't need an equals(), comparator, etc
@@ -444,6 +445,8 @@ public class ChukwaAgent
     }
   //  String initialAdaptorsStr = conf.get("initial_adaptors_file");
     
+    tags = conf.get("chukwaAgent.tags", "cluster=\"unknown\"");
+    
     initialAdaptors = new File(chukwaHome + "conf/initial_adaptors");
   }
   
@@ -495,4 +498,8 @@ public class ChukwaAgent
     return controlSock;
   }
 
+  public static String getTags() {
+	    return tags;
+  }
+
 }

src/contrib/chukwa/src/java/org/apache/hadoop/chukwa/extraction/engine/datasource/database/DatabaseDS.java

@@ -51,7 +51,7 @@ public class DatabaseDS implements DataSource
 		
 		 if(cluster==null)
 		 {
-			   cluster="mithrilgold";
+			   cluster="demo";
 		  }
 		
 		if (dataSource.equalsIgnoreCase("MRJob"))

src/contrib/chukwa/src/test/org/apache/hadoop/chukwa/ChunkImplTest.java

@@ -0,0 +1,58 @@
+package org.apache.hadoop.chukwa;
+
+import java.io.IOException;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+
+public class ChunkImplTest extends TestCase
+{
+	public void testVersion()
+	{
+		ChunkBuilder cb = new ChunkBuilder();
+		cb.addRecord("foo".getBytes());
+		cb.addRecord("bar".getBytes());
+		cb.addRecord("baz".getBytes());
+		Chunk c = cb.getChunk();
+		DataOutputBuffer ob = new DataOutputBuffer(c.getSerializedSizeEstimate());
+		try
+		{
+			c.write(ob);
+			DataInputBuffer ib = new DataInputBuffer();
+			ib.reset(ob.getData(), c.getSerializedSizeEstimate());
+			int version = ib.readInt();
+			assertEquals(version,ChunkImpl.PROTOCOL_VERSION);
+		} 
+		catch (IOException e)
+		{
+			e.printStackTrace();
+			fail("Should nor raise any exception"); 
+		}
+	}
+
+	public void testWrongVersion()
+	{
+		ChunkBuilder cb = new ChunkBuilder();
+		cb.addRecord("foo".getBytes());
+		cb.addRecord("bar".getBytes());
+		cb.addRecord("baz".getBytes());
+		Chunk c = cb.getChunk();
+		DataOutputBuffer ob = new DataOutputBuffer(c.getSerializedSizeEstimate());
+		try
+		{
+			c.write(ob);
+			DataInputBuffer ib = new DataInputBuffer();
+			ib.reset(ob.getData(), c.getSerializedSizeEstimate());
+			//change current chunkImpl version
+			ChunkImpl.PROTOCOL_VERSION = ChunkImpl.PROTOCOL_VERSION+1;
+			ChunkImpl.read(ib);
+			fail("Should have raised an IOexception"); 
+		} 
+		catch (IOException e)
+		{
+			// right behavior, do nothing
+		}
+	}
+}

src/contrib/chukwa/src/test/org/apache/hadoop/chukwa/TestChunkBuilder.java

@@ -35,5 +35,4 @@ public class TestChunkBuilder extends TestCase {
     assertEquals(5, chunk.getRecordOffsets()[1]);
     assertEquals(8, chunk.getRecordOffsets()[2]);
   }
-
 }

src/contrib/chukwa/src/test/org/apache/hadoop/chukwa/inputtools/log4j/TestChukwaAppender.java

@@ -29,7 +29,8 @@ import org.apache.log4j.spi.*;
 
 public class TestChukwaAppender extends TestCase {
   
-  public void testChukwaAppender() {
+  @SuppressWarnings("deprecation")
+public void testChukwaAppender() {
     try {
     
     ChukwaAgent agent = new ChukwaAgent();