Java API Proposal

logstash-core/build.gradle

@@ -121,6 +121,7 @@ dependencies {
     compile "com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:${jacksonVersion}"
     compile "org.jruby:jruby-complete:${jrubyVersion}"
     compile 'com.google.googlejavaformat:google-java-format:1.5'
+    compile 'org.reflections:reflections:0.9.11'
     testCompile 'org.apache.logging.log4j:log4j-core:2.9.1:tests'
     testCompile 'junit:junit:4.12'
     testCompile 'net.javacrumbs.json-unit:json-unit:1.9.0'

logstash-core/src/main/java/org/logstash/execution/DiscoverPlugins.java

@@ -0,0 +1,32 @@
+package org.logstash.execution;
+
+import java.lang.reflect.Constructor;
+import java.util.Set;
+import org.reflections.Reflections;
+
+/**
+ * Quick demo of plugin discovery showing that the solution wouldn't require anything beyond
+ * the plugin classes on the classpath.
+ */
+public final class DiscoverPlugins {
+
+    public static void main(final String... args) throws NoSuchMethodException {
+        Reflections reflections = new Reflections("org.logstash");
+        Set<Class<?>> annotated = reflections.getTypesAnnotatedWith(LogstashPlugin.class);
+        for (final Class<?> cls : annotated) {
+            System.out.println(cls.getName());
+            System.out.println(((LogstashPlugin) cls.getAnnotations()[0]).name());
+            final Constructor<?> ctor = cls.getConstructor(LsConfiguration.class, LsContext.class);
+            System.out.println("Found Ctor at : " + ctor.getName());
+            if (Filter.class.isAssignableFrom(cls)) {
+                System.out.println("Filter");
+            }
+            if (Output.class.isAssignableFrom(cls)) {
+                System.out.println("Output");
+            }
+            if (Input.class.isAssignableFrom(cls)) {
+                System.out.println("Input");
+            }
+        }
+    }
+}

logstash-core/src/main/java/org/logstash/execution/Filter.java

@@ -0,0 +1,132 @@
+package org.logstash.execution;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import org.logstash.Event;
+
+/**
+ * A Filter is simply a mapping of {@link QueueReader} to a new {@link QueueReader}.
+ */
+public interface Filter extends LsPlugin {
+
+    QueueReader filter(QueueReader reader);
+
+    @LogstashPlugin(name = "mutate")
+    final class Mutate implements Filter {
+
+        private static final PluginConfigSpec<String> FIELD_CONFIG =
+            LsConfiguration.requiredStringSetting("field");
+
+        private static final PluginConfigSpec<String> VALUE_CONFIG =
+            LsConfiguration.requiredStringSetting("value");
+
+        private final String field;
+
+        private final String value;
+
+        /**
+         * Required Constructor Signature only taking a {@link LsConfiguration}.
+         * @param configuration Logstash Configuration
+         * @param context Logstash Context
+         */
+        public Mutate(final LsConfiguration configuration, final LsContext context) {
+            this.field = configuration.get(FIELD_CONFIG);
+            this.value = configuration.get(VALUE_CONFIG);
+        }
+
+        @Override
+        public QueueReader filter(final QueueReader reader) {
+            return new QueueReader() {
+                @Override
+                public long poll(final Event event) {
+                    final long seq = reader.poll(event);
+                    if (seq > -1L) {
+                        event.setField(field, value);
+                    }
+                    return seq;
+                }
+
+                @Override
+                public long poll(final Event event, final long millis) {
+                    final long seq = reader.poll(event, millis);
+                    if (seq > -1L) {
+                        event.setField(field, value);
+                    }
+                    return seq;
+                }
+
+                @Override
+                public void acknowledge(final long sequenceNum) {
+                    reader.acknowledge(sequenceNum);
+                }
+            };
+        }
+
+        @Override
+        public Collection<PluginConfigSpec<?>> configSchema() {
+            return Arrays.asList(FIELD_CONFIG, VALUE_CONFIG);
+        }
+    }
+
+    @LogstashPlugin(name = "clone")
+    final class Clone implements Filter {
+
+        private Event clone;
+
+        private long lastSeq = -1L;
+
+        /**
+         * Required Constructor Signature only taking a {@link LsConfiguration}.
+         * @param configuration Logstash Configuration
+         * @param context Logstash Context
+         */
+        public Clone(final LsConfiguration configuration, final LsContext context) {
+        }
+
+        @Override
+        public QueueReader filter(final QueueReader reader) {
+            return new QueueReader() {
+                @Override
+                public long poll(final Event event) {
+                    if (clone != null) {
+                        event.overwrite(clone);
+                        clone = null;
+                        return lastSeq;
+                    }
+                    final long seq = reader.poll(event);
+                    lastSeq = seq;
+                    if (seq > -1L) {
+                        clone = event.clone();
+                    }
+                    return seq;
+                }
+
+                @Override
+                public long poll(final Event event, final long millis) {
+                    if (clone != null) {
+                        event.overwrite(clone);
+                        clone = null;
+                        return lastSeq;
+                    }
+                    final long seq = reader.poll(event, millis);
+                    lastSeq = seq;
+                    if (seq > -1L) {
+                        clone = event.clone();
+                    }
+                    return seq;
+                }
+
+                @Override
+                public void acknowledge(final long sequenceNum) {
+                    reader.acknowledge(sequenceNum);
+                }
+            };
+        }
+
+        @Override
+        public Collection<PluginConfigSpec<?>> configSchema() {
+            return Collections.emptyList();
+        }
+    }
+}

logstash-core/src/main/java/org/logstash/execution/Input.java

@@ -0,0 +1,79 @@
+package org.logstash.execution;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Scanner;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * A Logstash Pipeline Input pushes to a {@link QueueWriter}.
+ */
+public interface Input extends LsPlugin {
+
+    /**
+     * Start pushing {@link org.logstash.Event} to given {@link QueueWriter}.
+     * @param writer Queue Writer to Push to
+     */
+    void start(QueueWriter writer);
+
+    /**
+     * Stop the input.
+     * Stopping happens asynchronously, use {@link #awaitStop()} to make sure that the input has
+     * finished.
+     */
+    void stop();
+
+    /**
+     * Blocks until the input execution has finished.
+     * @throws InterruptedException On Interrupt
+     */
+    void awaitStop() throws InterruptedException;
+
+    @LogstashPlugin(name = "stream")
+    final class StreamInput implements Input {
+
+        private Scanner inpt;
+
+        private final CountDownLatch done = new CountDownLatch(1);
+
+        private volatile boolean stopped;
+
+        /**
+         * Required Constructor Signature only taking a {@link LsConfiguration}.
+         * @param configuration Logstash Configuration
+         * @param context Logstash Context
+         */
+        public StreamInput(final LsConfiguration configuration, final LsContext context) {
+            // Do whatever
+        }
+
+        @Override
+        public void start(final QueueWriter writer) {
+            inpt = new Scanner(System.in, "\n");
+            try {
+                while (!stopped && inpt.hasNext()) {
+                    final String message = inpt.next();
+                    writer.push(Collections.singletonMap("message", message));
+                }
+            } finally {
+                stopped = true;
+                done.countDown();
+            }
+        }
+
+        @Override
+        public void stop() {
+            stopped = true;
+        }
+
+        @Override
+        public void awaitStop() throws InterruptedException {
+            done.await();
+        }
+
+        @Override
+        public Collection<PluginConfigSpec<?>> configSchema() {
+            return Collections.emptyList();
+        }
+    }
+}

logstash-core/src/main/java/org/logstash/execution/LogstashPlugin.java

@@ -0,0 +1,16 @@
+package org.logstash.execution;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Logstash plugin annotation for finding plugins on the classpath and setting their name as used
+ * in the configuration syntax.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface LogstashPlugin {
+    String name();
+}

logstash-core/src/main/java/org/logstash/execution/LsConfiguration.java

@@ -0,0 +1,75 @@
+package org.logstash.execution;
+
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * LS Configuration example. Should be implemented like Spark config or Hadoop job config classes.
+ */
+public final class LsConfiguration {
+
+    /**
+     * @param raw Configuration Settings Map. Values are serialized.
+     */
+    public LsConfiguration(final Map<String, String> raw) {
+
+    }
+
+    public <T> T get(final PluginConfigSpec<T> configSpec) {
+        // TODO: Implement
+        return null;
+    }
+
+    public boolean contains(final PluginConfigSpec<?> configSpec) {
+        // TODO: Implement
+        return false;
+    }
+
+    public Collection<String> allKeys() {
+        return null;
+    }
+
+    public static PluginConfigSpec<String> stringSetting(final String name) {
+        return new PluginConfigSpec<>(
+            name, String.class, null, false, false
+        );
+    }
+
+    public static PluginConfigSpec<String> requiredStringSetting(final String name) {
+        return new PluginConfigSpec<>(name, String.class, null, false, true);
+    }
+
+    public static PluginConfigSpec<Long> numSetting(final String name) {
+        return new PluginConfigSpec<>(
+            name, Long.class, null, false, false
+        );
+    }
+
+    public static PluginConfigSpec<Long> numSetting(final String name, final long defaultValue) {
+        return new PluginConfigSpec<>(
+            name, Long.class, defaultValue, false, false
+        );
+    }
+
+    public static PluginConfigSpec<Path> pathSetting(final String name) {
+        return new PluginConfigSpec<>(name, Path.class, null, false, false);
+    }
+
+    public static PluginConfigSpec<Boolean> booleanSetting(final String name) {
+        return new PluginConfigSpec<>(name, Boolean.class, null, false, false);
+    }
+
+    @SuppressWarnings("unchecked")
+    public static PluginConfigSpec<Map<String, String>> hashSetting(final String name) {
+        return new PluginConfigSpec(name, Map.class, null, false, false);
+    }
+
+    @SuppressWarnings("unchecked")
+    public static PluginConfigSpec<Map<String, LsConfiguration>> requiredHashSetting(
+        final String name, final Collection<PluginConfigSpec<?>> spec) {
+        return new PluginConfigSpec(
+            name, Map.class, null, false, true
+        );
+    }
+}

logstash-core/src/main/java/org/logstash/execution/LsContext.java

@@ -0,0 +1,15 @@
+package org.logstash.execution;
+
+import org.logstash.common.io.DeadLetterQueueWriter;
+
+/**
+ * Holds Logstash Environment.
+ */
+public final class LsContext {
+
+    // TODO: Add getters for metrics, logger etc.
+
+    public DeadLetterQueueWriter dlqWriter() {
+        return null;
+    }
+}

logstash-core/src/main/java/org/logstash/execution/LsPlugin.java

@@ -0,0 +1,8 @@
+package org.logstash.execution;
+
+import java.util.Collection;
+
+public interface LsPlugin {
+
+    Collection<PluginConfigSpec<?>> configSchema();
+}