(DON'T MERGE) EAGLE-1104, Eagle 1.0 preparation

eagle-core/eagle-alert-parent/eagle-alert-app/src/main/resources/application.conf

@@ -13,30 +13,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 {
-  "appId" : "HBaseAuditLogApp",
+  "appId" : "EaglePolicyEngine",
   "mode" : "LOCAL",
   "siteId" : "testsite",
   "topology" : {
     "name" : "alertUnitTopology_1",
     "numOfTotalWorkers" : 2,
     "numOfSpoutTasks" : 1,
-    "numOfRouterBolts" : 4,
-    "numOfAlertBolts" : 10,
+    "numOfRouterBolts" : 1,
+    "numOfAlertBolts" : 1,
     "numOfPublishExecutors" : 1,
     "numOfPublishTasks" : 1,
     "messageTimeoutSecs": 3600,
     "localMode" : "true"
   },
   "spout" : {
-    "kafkaBrokerZkQuorum": "server.eagle.apache.org:2181",
+    "kafkaBrokerZkQuorum": "127.0.0.1:2000",
     "kafkaBrokerZkBasePath": "/kafka",
     "stormKafkaUseSameZkQuorumWithKafkaBroker": true,
     "stormKafkaTransactionZkQuorum": "",
     "stormKafkaTransactionZkPath": "/consumers",
     "stormKafkaEagleConsumer": "eagle_consumer"
   },
   "zkConfig" : {
-    "zkQuorum" : "server.eagle.apache.org:2181",
+    "zkQuorum" : "127.0.0.1:2000",
     "zkRoot" : "/alert"
   },
   "metadataService": {

eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java

@@ -65,6 +65,7 @@ public class CorrelationSpout extends BaseRichSpout implements SpoutSpecListener
 
     private SpoutSpec cachedSpoutSpec;
 
+
     private transient KafkaSpoutMetric kafkaSpoutMetric;
 
     @SuppressWarnings("rawtypes")

eagle-examples/eagle-app-example/src/main/java/org/apache/eagle/app/example/ExampleStormApplication.java

@@ -35,7 +35,7 @@ public class ExampleStormApplication extends StormApplication{
     @Override
     public StormTopology execute(Config config, StormEnvironment environment) {
         TopologyBuilder builder = new TopologyBuilder();
-        builder.setSpout("metric_spout", new RandomEventSpout(), config.getInt("spoutNum"));
+        builder.setSpout("metric_spout", new RandomEventSpout(), config.hasPath("spoutNum") ? config.getInt("spoutNum") : 1);
         builder.setBolt("sink_1",environment.getStreamSink("SAMPLE_STREAM_1",config)).fieldsGrouping("metric_spout",new Fields("metric"));
         builder.setBolt("sink_2",environment.getStreamSink("SAMPLE_STREAM_2",config)).fieldsGrouping("metric_spout",new Fields("metric"));
         return builder.createTopology();

eagle-examples/eagle-app-example/src/main/resources/application.conf

@@ -0,0 +1,45 @@
+# 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.
+
+{
+  "appId" : "ExampleMonitorApp",
+  "mode" : "LOCAL",
+  "siteId" : "testsite",
+  "topology" : {
+    "numOfTotalWorkers" : 2,
+    "numOfSpoutTasks" : 2,
+    "numOfParserTasks" : 2,
+    "numOfJoinTasks" : 2,
+    "numOfSinkTasks" : 2
+  },
+  "dataSourceConfig": {
+    "topic" : "nonexistingtopic",
+    "zkConnection" : "127.0.0.1:2000",
+    "txZkServers" : "127.0.0.1:2000",
+    "schemeCls" : "storm.kafka.StringScheme"
+  },
+  "eagleService": {
+    "host": "localhost",
+    "port": 9090
+    "username": "admin",
+    "password": "secret"
+  },
+  "dataSinkConfig": {
+    "topic" : "myexampletopic",
+    "brokerList" : "127.0.0.1:2000",
+    "serializerClass" : "kafka.serializer.StringEncoder",
+    "keySerializerClass" : "kafka.serializer.StringEncoder"
+  }
+}

eagle-flink/README.md

@@ -0,0 +1,22 @@
+## Design goals
+
+### 1. execute rules on one or multiple streams
+
+### 2. dynamically inject new rules on existing streams
+
+### 3. reuse streams as much as possible
+
+## Primivite operations
+
+### 1. rules on single stream keyed by some fields
+ avg(cpu) > 0.8 [1m] group by host 
+
+ sum(failed_requests) > 60 [1m] group by host
+
+ avg(failure_ratio) > 0.1 [1m] group by host
+
+### 2. rules on multiple streams joined by some fields
+
+
+
+

eagle-flink/pom.xml

@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>eagle-parent</artifactId>
+        <groupId>org.apache.eagle</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>eagle-flink</artifactId>
+    <properties>
+        <flink.version>1.10.0</flink.version>
+        <scala.binary.version>2.11</scala.binary.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.16.20</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.wso2.siddhi</groupId>
+            <artifactId>siddhi-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

eagle-flink/src/main/java/org/apache/eagle/flink/AlertDeduplication.java

@@ -0,0 +1,82 @@
+/*
+ *  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.eagle.flink;
+
+import org.apache.commons.collections.ListUtils;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+
+import java.util.List;
+import java.util.Objects;
+
+public class AlertDeduplication {
+    private String outputStreamId;
+    private String dedupIntervalMin;
+    private List<String> dedupFields;
+
+    public String getOutputStreamId() {
+        return outputStreamId;
+    }
+
+    public void setOutputStreamId(String outputStreamId) {
+        this.outputStreamId = outputStreamId;
+    }
+
+    public String getDedupIntervalMin() {
+        return dedupIntervalMin;
+    }
+
+    public void setDedupIntervalMin(String dedupIntervalMin) {
+        this.dedupIntervalMin = dedupIntervalMin;
+    }
+
+    public List<String> getDedupFields() {
+        return dedupFields;
+    }
+
+    public void setDedupFields(List<String> dedupFields) {
+        this.dedupFields = dedupFields;
+    }
+
+    @Override
+    public int hashCode() {
+        return new HashCodeBuilder()
+                .append(outputStreamId)
+                .append(dedupFields)
+                .append(dedupIntervalMin)
+                .build();
+    }
+
+    @Override
+    public boolean equals(Object that) {
+        if (that == this) {
+            return true;
+        }
+        if (!(that instanceof AlertDeduplication)) {
+            return false;
+        }
+        AlertDeduplication another = (AlertDeduplication) that;
+        if (ListUtils.isEqualList(another.dedupFields, this.dedupFields)
+                && Objects.equals(another.dedupIntervalMin, this.dedupIntervalMin)
+                && Objects.equals(another.outputStreamId, this.outputStreamId)) {
+            return true;
+        }
+        return false;
+    }
+
+
+}

eagle-flink/src/main/java/org/apache/eagle/flink/AlertDefinition.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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.eagle.flink;
+
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+
+import java.util.Objects;
+
+public class AlertDefinition {
+    private TemplateType templateType = TemplateType.TEXT;
+    private String subject;
+    private String body;
+
+    private AlertSeverity severity;
+    private String category;
+
+    public String getBody() {
+        return body;
+    }
+
+    public void setBody(String templateResource) {
+        this.body = templateResource;
+    }
+
+    public TemplateType getTemplateType() {
+        return templateType;
+    }
+
+    public void setTemplateType(TemplateType type) {
+        this.templateType = type;
+    }
+
+    public String getSubject() {
+        return subject;
+    }
+
+    public void setSubject(String subject) {
+        this.subject = subject;
+    }
+
+    public AlertSeverity getSeverity() {
+        return severity;
+    }
+
+    public void setSeverity(AlertSeverity severity) {
+        this.severity = severity;
+    }
+
+    public String getCategory() {
+        return category;
+    }
+
+    public void setCategory(String category) {
+        this.category = category;
+    }
+
+    public enum TemplateType {
+        TEXT,
+        // FILE,
+        // HTTP
+    }
+
+    @Override
+    public int hashCode() {
+        return new HashCodeBuilder()
+                .append(templateType)
+                .append(this.body)
+                .append(this.category)
+                .append(this.severity)
+                .append(this.subject)
+                .build();
+    }
+
+    @Override
+    public boolean equals(Object that) {
+        if (that == this) {
+            return true;
+        }
+        if (!(that instanceof AlertDefinition)) {
+            return false;
+        }
+        AlertDefinition another = (AlertDefinition) that;
+        if (Objects.equals(another.templateType, this.templateType)
+                && Objects.equals(another.body, this.body)
+                && Objects.equals(another.category, this.category)
+                && Objects.equals(another.severity, this.severity)
+                && Objects.equals(another.subject, this.subject)) {
+            return true;
+        }
+        return false;
+    }
+}