svn commit: r1179413 - in /incubator/flume/branches/flume-728: flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java flume-ng-node/src/main/java/org/apache/flume/node/Application.java

[es...@apache.org]

Author: esammer
Date: Wed Oct  5 20:17:32 2011
New Revision: 1179413

URL: http://svn.apache.org/viewvc?rev=1179413&view=rev
Log:
- Added a very basic event driven exec source for unix tools that stream data (e.g. tail -F).

Added:
    incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java
Modified:
    incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java

Added: incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java?rev=1179413&view=auto
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java (added)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java Wed Oct  5 20:17:32 2011
@@ -0,0 +1,135 @@
+package org.apache.flume.source;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.CounterGroup;
+import org.apache.flume.Event;
+import org.apache.flume.EventDrivenSource;
+import org.apache.flume.Transaction;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.EventBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+
+public class ExecSource extends AbstractSource implements EventDrivenSource,
+    Configurable {
+
+  private static final Logger logger = LoggerFactory
+      .getLogger(ExecSource.class);
+
+  private String command;
+
+  private CounterGroup counterGroup;
+  private ExecutorService executor;
+  private Future<?> runnerFuture;
+
+  @Override
+  public void start() {
+    logger.info("Exec source starting with command:{}", command);
+
+    executor = Executors.newSingleThreadExecutor();
+    counterGroup = new CounterGroup();
+
+    ExecRunnable runner = new ExecRunnable();
+
+    runner.command = command;
+    runner.channel = getChannel();
+    runner.counterGroup = counterGroup;
+
+    // FIXME: Use a callback-like executor / future to signal us upon failure.
+    runnerFuture = executor.submit(runner);
+
+    /*
+     * NB: This comes at the end rather than the beginning of the method because
+     * it sets our state to running. We want to make sure the executor is alive
+     * and well first.
+     */
+    super.start();
+
+    logger.debug("Exec source started");
+  }
+
+  @Override
+  public void stop() {
+    logger.info("Stopping exec source with command:{}", command);
+
+    if (runnerFuture != null) {
+      logger.debug("Stopping exec runner");
+      runnerFuture.cancel(true);
+      logger.debug("Exec runner stopped");
+    }
+
+    executor.shutdown();
+
+    while (!executor.isTerminated()) {
+      logger.debug("Waiting for exec executor service to stop");
+      try {
+        executor.awaitTermination(500, TimeUnit.MILLISECONDS);
+      } catch (InterruptedException e) {
+        logger
+            .debug("Interrupted while waiting for exec executor service to stop. Just exiting.");
+        Thread.currentThread().interrupt();
+      }
+    }
+
+    super.stop();
+
+    logger.debug("Exec source with command:{} stopped. Metrics:{}", command,
+        counterGroup);
+  }
+
+  @Override
+  public void configure(Context context) {
+    command = context.get("command", String.class);
+
+    Preconditions.checkState(command != null,
+        "The parameter command must be specified");
+  }
+
+  private static class ExecRunnable implements Runnable {
+
+    private String command;
+    private Channel channel;
+    private CounterGroup counterGroup;
+
+    @Override
+    public void run() {
+      try {
+        String[] commandArgs = command.split("\\s+");
+        Process process = new ProcessBuilder(commandArgs).start();
+        BufferedReader reader = new BufferedReader(new InputStreamReader(
+            process.getInputStream()));
+
+        String line = null;
+
+        while ((line = reader.readLine()) != null) {
+          counterGroup.incrementAndGet("exec.lines.read");
+
+          Transaction transaction = channel.getTransaction();
+
+          transaction.begin();
+          Event event = EventBuilder.withBody(line.getBytes());
+          channel.put(event);
+          transaction.commit();
+        }
+
+        reader.close();
+      } catch (IOException e) {
+        logger.error("Failed while running command:{} - Exception follows.",
+            command, e);
+      }
+    }
+
+  }
+
+}

Modified: incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java?rev=1179413&r1=1179412&r2=1179413&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java Wed Oct  5 20:17:32 2011
@@ -24,6 +24,7 @@ import org.apache.flume.sink.NullSink;
 import org.apache.flume.sink.RollingFileSink;
 import org.apache.flume.sink.hdfs.HDFSEventSink;
 import org.apache.flume.source.DefaultSourceFactory;
+import org.apache.flume.source.ExecSource;
 import org.apache.flume.source.NetcatSource;
 import org.apache.flume.source.SequenceGeneratorSource;
 import org.slf4j.Logger;
@@ -71,6 +72,7 @@ public class Application {
 
     sourceFactory.register("seq", SequenceGeneratorSource.class);
     sourceFactory.register("netcat", NetcatSource.class);
+    sourceFactory.register("exec", ExecSource.class);
 
     sinkFactory.register("null", NullSink.class);
     sinkFactory.register("logger", LoggerSink.class);