Merge branch 'release/v0.13.0'

README.rst

@@ -16,6 +16,8 @@ chat with us on `gitter <https://gitter.im/logisland/logisland>`_
 
 **LogIsland is an event mining scalable platform designed to handle a high throughput of events.**
 
+It is highly inspired from DataFlow programming tools such as Apache Nifi, but with a highly scalable architecture.
+
 
 Event mining Workflow
 ---------------------
@@ -49,18 +51,20 @@ to build from the source just clone source and package with maven
 
     git clone https://github.com/Hurence/logisland.git
     cd logisland
-    mvn install
+    mvn clean install
 
-the final package is available at `logisland-assembly/target/logisland-0.12.2-bin-hdp2.5.tar.gz`
+the final package is available at `logisland-assembly/target/logisland-0.13.0-bin-hdp2.5.tar.gz`
 
 You can also download the `latest release build <https://github.com/Hurence/logisland/releases>`_
 
+Quick start
+-----------
 
 Local Setup
------------
-basically **logisland** depends on Kafka and Spark, you can deploy it on any linux server
++++++++++++
+Alternatively you can deploy **logisland** on any linux server from which Kafka and Spark are available
 
-.. code-block::
+.. code-block:: sh
 
     # install Kafka 0.10.0.0 & start a zookeeper node + a broker
     curl -s http://apache.crihan.fr/dist/kafka/0.10.0.0/kafka_2.11-0.10.0.0.tgz | tar -xz -C /usr/local/
@@ -72,16 +76,43 @@ basically **logisland** depends on Kafka and Spark, you can deploy it on any lin
     curl -s http://d3kbcqa49mib13.cloudfront.net/spark-2.1.0-bin-hadoop2.7.tgz | tar -xz -C /usr/local/
     export SPARK_HOME=/usr/local/spark-2.1.0-bin-hadoop2.7
 
-    # install Logisland 0.12.2
-    curl -s https://github.com/Hurence/logisland/releases/download/v0.10.0/logisland-0.12.2-bin-hdp2.5.tar.gz  | tar -xz -C /usr/local/
-    cd /usr/local/logisland-0.12.2
+    # install Logisland 0.13.0
+    curl -s https://github.com/Hurence/logisland/releases/download/v0.10.0/logisland-0.13.0-bin-hdp2.5.tar.gz  | tar -xz -C /usr/local/
+    cd /usr/local/logisland-0.13.0
 
     # launch a logisland job
     bin/logisland.sh --conf conf/index-apache-logs.yml
 
 you can find some **logisland** job configuration samples under `$LOGISLAND_HOME/conf` folder
 
 
+Docker setup
+++++++++++++
+The easiest way to start is the launch a docker compose stack
+
+.. code-block:: sh
+
+    # launch logisland environment
+    cd /tmp
+    curl -s https://raw.githubusercontent.com/Hurence/logisland/master/logisland-framework/logisland-resources/src/main/resources/conf/docker-compose.yml > docker-compose.yml
+    docker-compose up
+
+    # sample execution of a logisland job
+    docker exec -i -t logisland conf/index-apache-logs.yml
+
+
+Hadoop distribution setup
++++++++++++++++++++++++++
+Launching logisland streaming apps is just easy as unarchiving logisland distribution on an edge node, editing a config with YARN parameters and submitting job.
+
+.. code-block:: sh
+
+    # install Logisland 0.13.0
+    curl -s https://github.com/Hurence/logisland/releases/download/v0.10.0/logisland-0.13.0-bin-hdp2.5.tar.gz  | tar -xz -C /usr/local/
+    cd /usr/local/logisland-0.13.0
+    bin/logisland.sh --conf conf/index-apache-logs.yml
+
+
 Start a stream processing job
 -----------------------------
 
@@ -99,15 +130,15 @@ The first part is the `ProcessingEngine` configuration (here a Spark streaming e
 
 .. code-block:: yaml
 
-    version: 0.12.2
+    version: 0.13.0
     documentation: LogIsland job config file
     engine:
       component: com.hurence.logisland.engine.spark.KafkaStreamProcessingEngine
       type: engine
       documentation: Index some apache logs with logisland
       configuration:
         spark.app.name: IndexApacheLogsDemo
-        spark.master: local[4]
+        spark.master: yarn-cluster
         spark.driver.memory: 1G
         spark.driver.cores: 1
         spark.executor.memory: 2G
@@ -204,7 +235,11 @@ Once you've edited your configuration file, you can submit it to execution engin
 
 .. code-block:: bash
 
-    bin/process-stream.sh -conf conf/job-configuration.yml
+    bin/logisland.sh -conf conf/job-configuration.yml
+
+
+You should jump to the `tutorials section <http://logisland.readthedocs.io/en/latest/tutorials/index.html>`_ of the documentation.
+And then continue with `components documentation<http://logisland.readthedocs.io/en/latest/components.html>`_
 
 Contributing
 ------------

ROADMAP.rst

@@ -1,27 +1,21 @@
-Log Island Roadmap and future work
-====
+Logisland Roadmap and future work
+=================================
 
 follow the roadmap through `github issues <https://github.com/Hurence/logisland/issues>`_ too
 
-GUI
-----
-
-- manage visualy the streams
-- search kafka topics
 
 Engine
-----
+------
 
-- Add KafkaStreamEngine
-- Add autoscaler component
-- move offsets management from Zookeeper to Kafka
+- Dynamic config via REST API
+- Autoscaler
 - whole integration test framework (file => kafka topic => process stream => es => query)
 
 Components
-----
+----------
 
+- Alert & threshold managment
 - add EventField mutator based on EL
-- add an HDFS bulk loader
 - add a generic parser that infers a Regexp from a list (Streaming Deep Learning)
 
 

launch-tuto.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
 
-logisland-assembly/target/logisland-0.12.2-bin-hdp2.5/logisland-0.12.2/bin/logisland.sh \
+logisland-assembly/target/logisland-0.13.0-bin-hdp2.5/logisland-0.13.0/bin/logisland.sh \
     --conf logisland-framework/logisland-resources/src/main/resources/conf/$1

logisland-api/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <groupId>com.hurence.logisland</groupId>
         <artifactId>logisland</artifactId>
-        <version>0.12.2</version>
+        <version>0.13.0</version>
     </parent>
     <artifactId>logisland-api</artifactId>
     <packaging>jar</packaging>

logisland-api/src/main/java/com/hurence/logisland/component/AbstractPropertyValue.java

@@ -21,6 +21,9 @@
 import com.hurence.logisland.record.Record;
 import com.hurence.logisland.record.StandardRecord;
 import com.hurence.logisland.registry.VariableRegistry;
+import com.hurence.logisland.util.FormatUtils;
+
+import java.util.concurrent.TimeUnit;
 
 /**
  * Created by mathieu on 08/06/17.
@@ -70,6 +73,12 @@ public Double asDouble() {
         return (getRawValue() == null) ? null : Double.parseDouble(getRawValue().trim());
     }
 
+    @Override
+    public Long asTimePeriod(final TimeUnit timeUnit) {
+        return (rawValue == null) ? null : FormatUtils.getTimeDuration(rawValue.toString().trim(), timeUnit);
+    }
+
+
     @Override
     public boolean isSet() {
         return getRawValue() != null;

logisland-api/src/main/java/com/hurence/logisland/component/PropertyValue.java

@@ -21,6 +21,7 @@
 import com.hurence.logisland.record.Record;
 
 import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
 
 /**
  * <p>
@@ -91,7 +92,7 @@ public interface PropertyValue extends Serializable {
      * in terms of the specified TimeUnit; if the property is not set, returns
      * <code>null</code>
      */
-   // public Long asTimePeriod(TimeUnit timeUnit);
+    public Long asTimePeriod(TimeUnit timeUnit);
 
 
 

logisland-api/src/main/java/com/hurence/logisland/config/DefaultConfigValues.java

@@ -20,6 +20,7 @@
  */
 public enum DefaultConfigValues {
 
+    REDIS_CONNECTION("sandbox:6379"),
     ES_HOSTS("sandbox:9300"),
     ES_CLUSTER_NAME("es-logisland"),
     KAFKA_BROKERS("sandbox:9092"),

logisland-api/src/main/java/com/hurence/logisland/controller/AbstractControllerService.java

@@ -17,6 +17,8 @@
 package com.hurence.logisland.controller;
 
 
+import com.hurence.logisland.annotation.lifecycle.OnDisabled;
+import com.hurence.logisland.annotation.lifecycle.OnEnabled;
 import com.hurence.logisland.component.AbstractConfigurableComponent;
 import com.hurence.logisland.component.InitializationException;
 import com.hurence.logisland.logging.ComponentLog;
@@ -26,6 +28,7 @@ public abstract class AbstractControllerService extends AbstractConfigurableComp
 
     private ControllerServiceLookup serviceLookup;
     private ComponentLog logger;
+    private volatile boolean enabled = true;
 
     @Override
     public final void initialize(final ControllerServiceInitializationContext context) throws InitializationException {
@@ -62,4 +65,18 @@ protected ComponentLog getLogger() {
         return logger;
     }
 
+
+    @OnEnabled
+    public final void enabled() {
+        this.enabled = true;
+    }
+
+    @OnDisabled
+    public final void disabled() {
+        this.enabled = false;
+    }
+
+    public boolean isEnabled() {
+        return this.enabled;
+    }
 }

logisland-api/src/main/java/com/hurence/logisland/record/Field.java

@@ -17,10 +17,12 @@
 
 import com.hurence.logisland.component.PropertyValue;
 import com.hurence.logisland.controller.ControllerService;
+import com.hurence.logisland.util.FormatUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Primitive Types
@@ -208,6 +210,11 @@ public Double asDouble() {
         }
     }
 
+    @Override
+    public Long asTimePeriod(final TimeUnit timeUnit) {
+        return (rawValue == null) ? null : FormatUtils.getTimeDuration(rawValue.toString().trim(), timeUnit);
+    }
+
     @Override
     public boolean isSet() {
         return rawValue != null;

logisland-api/src/main/java/com/hurence/logisland/record/StandardRecord.java

@@ -65,6 +65,7 @@ public StandardRecord(String type) {
     }
 
     public StandardRecord(Record toClone) {
+        this();
         this.setType(toClone.getType());
         this.setTime(toClone.getTime());
         this.setId(UUID.randomUUID().toString());
@@ -159,7 +160,9 @@ public Record addFields(Map<String, Field> fields) {
 
     @Override
     public Record setType(String type) {
-        this.setField(FieldDictionary.RECORD_TYPE, FieldType.STRING, type);
+        if (type != null) {
+            this.setField(FieldDictionary.RECORD_TYPE, FieldType.STRING, type);
+        }
         return this;
     }
 

logisland-api/src/main/java/com/hurence/logisland/serializer/Deserializer.java

@@ -17,6 +17,7 @@
 
 
 import java.io.IOException;
+import java.io.InputStream;
 
 /**
  * Provides an interface for deserializing an array of bytes into an Object
@@ -28,12 +29,12 @@ public interface Deserializer<T> {
     /**
      * Deserializes the given byte array input an Object and returns that value.
      *
-     * @param input input
+     * @param objectDataInput input
      * @return returns deserialized value
      * @throws DeserializationException if a valid object cannot be deserialized
      * from the given byte array
      * @throws IOException ex
      */
-    T deserialize(byte[] input) throws DeserializationException, IOException;
+    T deserialize(InputStream objectDataInput) throws DeserializationException, IOException;
 
 }

logisland-api/src/main/java/com/hurence/logisland/serializer/RecordSerializer.java

@@ -39,7 +39,7 @@
 import java.io.Serializable;
 
 
-public interface RecordSerializer extends Serializable {
+public interface RecordSerializer extends Serializable, Serializer<Record>, Deserializer<Record> {
     void serialize(OutputStream objectDataOutput, Record record) throws RecordSerializationException;
 
     Record deserialize(InputStream objectDataInput) throws RecordSerializationException;

logisland-api/src/main/java/com/hurence/logisland/serializer/Serializer.java

@@ -34,6 +34,6 @@ public interface Serializer<T> {
      * @throws SerializationException If unable to serialize the given value
      * @throws IOException ex
      */
-    void serialize(T value, OutputStream output) throws SerializationException, IOException;
+    void serialize(OutputStream output, T value) throws SerializationException, IOException;
 
 }