HADOOP-5087.

- Fixed the regex to correctly parse the chukwa agent ADD command.
- Added a test case to validate the chukwa agent ADD command.



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

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

@@ -271,8 +271,15 @@ public class ChukwaAgent
   // but can be arbitrarily many space
   // delimited agent specific params )
   // 4) offset
-  Pattern addCmdPattern = Pattern.compile("[aA][dD][dD]\\s+(\\S+)\\s+(\\S+)\\s+(.*\\s)?\\s*(\\d+)\\s*");
-  // FIXME: should handle bad lines here
+  Pattern addCmdPattern = Pattern.compile(
+      "[aA][dD][dD]\\s+"  //command "add", any case, plus at least one space
+      + "(\\S+)\\s+" //the adaptor classname, plus at least one space
+      + "(\\S+)\\s+" //datatype, plus at least one space
+      + "(?:"    //start a non-capturing group, for the parameters
+      +         "(.*?)\\s+"    //capture the actual parameters reluctantly, followed by whitespace
+      +       ")?"    //end non-matching group for params; group is optional
+      + "(\\d+)\\s*");  // finally, an offset and some trailing whitespace
+
   public long processCommand(String cmd)
   {
     Matcher m = addCmdPattern.matcher(cmd);

src/contrib/chukwa/src/test/org/apache/hadoop/chukwa/datacollection/adaptor/ChukwaTestAdaptor.java

@@ -0,0 +1,112 @@
+/*
+ * 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.datacollection.adaptor;
+
+import org.apache.hadoop.chukwa.datacollection.ChunkReceiver;
+
+public class ChukwaTestAdaptor implements Adaptor {
+
+  private long adaptorId = 0l;
+  private String type = null;
+  private String params = null;
+  private long startOffset = 0l;
+  private ChunkReceiver dest = null;
+  
+  @Override
+  public String getCurrentStatus() throws AdaptorException {
+    // TODO Auto-generated method stub
+    return null;
+  }
+
+  @Override
+  public String getStreamName() {
+    // TODO Auto-generated method stub
+    return null;
+  }
+
+  @Override
+  public void hardStop() throws AdaptorException {
+    // TODO Auto-generated method stub
+    
+  }
+
+  @Override
+  public long shutdown() throws AdaptorException {
+    // TODO Auto-generated method stub
+    return 0;
+  }
+
+  @Override
+  public void start(long adaptorID, String type, String params, long offset,
+      ChunkReceiver dest) throws AdaptorException {
+    this.adaptorId = adaptorID;
+    this.type = type;
+    this.params = params;
+    this.startOffset = offset;
+    this.dest = dest;
+    System.out.println("adaptorId [" +adaptorId + "]");
+    System.out.println("type [" +type+ "]");
+    System.out.println("params [" +params+ "]");
+    System.out.println("startOffset [" +startOffset+ "]");
+    
+    
+  }
+  
+
+  public String getType() {
+    return type;
+  }
+
+  public void setType(String type) {
+    this.type = type;
+  }
+
+  public String getParams() {
+    return params;
+  }
+
+  public void setParams(String params) {
+    this.params = params;
+  }
+
+  public long getStartOffset() {
+    return startOffset;
+  }
+
+  public void setStartOffset(long startOffset) {
+    this.startOffset = startOffset;
+  }
+
+  public long getAdaptorId() {
+    return adaptorId;
+  }
+
+  public void setAdaptorId(long adaptorId) {
+    this.adaptorId = adaptorId;
+  }
+
+  public ChunkReceiver getDest() {
+    return dest;
+  }
+
+  public void setDest(ChunkReceiver dest) {
+    this.dest = dest;
+  }
+  
+}

src/contrib/chukwa/src/test/org/apache/hadoop/chukwa/datacollection/agent/TestCmd.java

@@ -0,0 +1,106 @@
+/*
+ * 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.datacollection.agent;
+
+import org.apache.hadoop.chukwa.datacollection.adaptor.Adaptor;
+import org.apache.hadoop.chukwa.datacollection.adaptor.ChukwaTestAdaptor;
+import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent.AlreadyRunningException;
+import org.apache.hadoop.chukwa.datacollection.test.ConsoleOutConnector;
+
+import junit.framework.TestCase;
+
+public class TestCmd extends TestCase {
+
+  public void testAddCmdWithParam()
+  {
+    ChukwaAgent agent;
+    try {
+      agent = new ChukwaAgent();
+      ConsoleOutConnector conn = new ConsoleOutConnector(agent, true);
+      conn.start();
+      long l = agent.processCommand("ADD org.apache.hadoop.chukwa.datacollection.adaptor.ChukwaTestAdaptor  chukwaTestAdaptorType 0 my param1 param2 /var/log/messages 114027");
+      assertTrue(l != -1);
+      Adaptor adaptor = agent.getAdaptorList().get(l);
+      ChukwaTestAdaptor chukwaTestAdaptor = (ChukwaTestAdaptor) adaptor;
+      assertTrue("error in type","chukwaTestAdaptorType".intern() == chukwaTestAdaptor.getType().intern());
+      assertTrue("error in param", "0 my param1 param2 /var/log/messages".intern() == chukwaTestAdaptor.getParams().intern());
+      assertTrue("error in startOffset",114027l == chukwaTestAdaptor.getStartOffset());
+      agent.stopAdaptor(l, false);
+      agent.shutdown();
+
+      Thread.sleep(2000);
+    } catch(InterruptedException e) {
+      
+    } catch (AlreadyRunningException e) {
+      e.printStackTrace();
+      fail(e.toString());
+    }
+  }
+  
+  public void testAddCmdWithoutParam1()
+  {
+    ChukwaAgent agent;
+    try {
+      agent = new ChukwaAgent();
+      ConsoleOutConnector conn = new ConsoleOutConnector(agent, true);
+      conn.start();
+      long l = agent.processCommand("ADD org.apache.hadoop.chukwa.datacollection.adaptor.ChukwaTestAdaptor  chukwaTestAdaptorType 114027");
+      assertTrue(l != -1);
+      Adaptor adaptor = agent.getAdaptorList().get(l);
+      ChukwaTestAdaptor chukwaTestAdaptor = (ChukwaTestAdaptor) adaptor;
+      assertTrue("error in type","chukwaTestAdaptorType".intern() == chukwaTestAdaptor.getType().intern());
+      assertTrue("error in param", "".intern() == chukwaTestAdaptor.getParams().intern());
+      assertTrue("error in startOffset",114027l == chukwaTestAdaptor.getStartOffset());
+      agent.stopAdaptor(l, false);
+      agent.shutdown();
+      Thread.sleep(2000);
+    } catch(InterruptedException e) {
+      
+    } catch (AlreadyRunningException e) {
+      e.printStackTrace();
+      fail(e.toString());
+    }
+  }
+  
+  public void testAddCmdWithoutParam2()
+  {
+    ChukwaAgent agent;
+    try {
+      agent = new ChukwaAgent();
+      ConsoleOutConnector conn = new ConsoleOutConnector(agent, true);
+      conn.start();
+      long l = agent.processCommand("ADD org.apache.hadoop.chukwa.datacollection.adaptor.ChukwaTestAdaptor" 
+         + "  chukwaTestAdaptorType 0  114027");
+      assertTrue(l != -1);
+      Adaptor adaptor = agent.getAdaptorList().get(l);
+      ChukwaTestAdaptor chukwaTestAdaptor = (ChukwaTestAdaptor) adaptor;
+      assertTrue("error in type","chukwaTestAdaptorType".intern() == chukwaTestAdaptor.getType().intern());
+      assertTrue("error in param", "0".intern() == chukwaTestAdaptor.getParams().intern());
+      assertTrue("error in startOffset",114027l == chukwaTestAdaptor.getStartOffset());
+      agent.stopAdaptor(l, false);
+      agent.shutdown();
+      Thread.sleep(2000);
+    } catch(InterruptedException e) {
+      
+    } catch (AlreadyRunningException e) {
+      e.printStackTrace();
+      fail(e.toString());
+    }
+  }
+}