You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eagle.apache.org by ha...@apache.org on 2016/06/02 07:08:13 UTC
[34/46] incubator-eagle git commit: [EAGLE-325] Initialize next-gen
alert engine code on branch-0.5
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
new file mode 100755
index 0000000..52fcc04
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
@@ -0,0 +1,113 @@
+/**
+ * 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.alert.engine.coordinator.impl;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.eagle.alert.coordination.model.AlertBoltSpec;
+import org.apache.eagle.alert.coordination.model.PublishSpec;
+import org.apache.eagle.alert.coordination.model.RouterSpec;
+import org.apache.eagle.alert.coordination.model.SpoutSpec;
+import org.apache.eagle.alert.engine.coordinator.IMetadataChangeNotifyService;
+import org.apache.eagle.alert.engine.coordinator.MetadataType;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+import org.apache.eagle.alert.engine.publisher.AlertPublishSpecListener;
+import org.apache.eagle.alert.engine.router.AlertBoltSpecListener;
+import org.apache.eagle.alert.engine.router.SpoutSpecListener;
+import org.apache.eagle.alert.engine.router.StreamRouterBoltSpecListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+import com.typesafe.config.Config;
+
+/**
+ * notify 3 components of metadata change Spout, StreamRouterBolt and AlertBolt
+ */
+@SuppressWarnings({ "serial" })
+public abstract class AbstractMetadataChangeNotifyService implements IMetadataChangeNotifyService, Closeable,
+ Serializable {
+ private final static Logger LOG = LoggerFactory.getLogger(AbstractMetadataChangeNotifyService.class);
+ private final List<StreamRouterBoltSpecListener> streamRouterBoltSpecListeners = new ArrayList<>();
+ private final List<SpoutSpecListener> spoutSpecListeners = new ArrayList<>();
+ private final List<AlertBoltSpecListener> alertBoltSpecListeners = new ArrayList<>();
+ private final List<AlertPublishSpecListener> alertPublishSpecListeners = new ArrayList<>();
+ protected MetadataType type;
+
+ /**
+ * @param config
+ */
+ @Override
+ public void init(Config config, MetadataType type) {
+ this.type = type;
+ }
+
+ @Override
+ public void registerListener(AlertPublishSpecListener listener) {
+ synchronized (alertBoltSpecListeners) {
+ Preconditions.checkNotNull(alertPublishSpecListeners, "Not initialized yet");
+ LOG.info("Register {}", listener);
+ alertPublishSpecListeners.add(listener);
+ }
+ }
+
+ @Override
+ public void registerListener(StreamRouterBoltSpecListener listener) {
+ synchronized (streamRouterBoltSpecListeners) {
+ streamRouterBoltSpecListeners.add(listener);
+ }
+ }
+
+ @Override
+ public void registerListener(AlertBoltSpecListener listener) {
+ synchronized (alertBoltSpecListeners) {
+ alertBoltSpecListeners.add(listener);
+ }
+ }
+
+ @Override
+ public void registerListener(SpoutSpecListener listener) {
+ synchronized (spoutSpecListeners) {
+ spoutSpecListeners.add(listener);
+ }
+ }
+
+ protected void notifySpout(SpoutSpec spoutSpec, Map<String, StreamDefinition> sds) {
+ spoutSpecListeners.forEach(s -> s.onSpoutSpecChange(spoutSpec, sds));
+ }
+
+ protected void notifyStreamRouterBolt(RouterSpec routerSpec, Map<String, StreamDefinition> sds) {
+ streamRouterBoltSpecListeners.forEach(s -> s.onStreamRouteBoltSpecChange(routerSpec, sds));
+ }
+
+ protected void notifyAlertBolt(AlertBoltSpec alertBoltSpec, Map<String, StreamDefinition> sds) {
+ alertBoltSpecListeners.forEach(s -> s.onAlertBoltSpecChange(alertBoltSpec, sds));
+ }
+
+ protected void notifyAlertPublishBolt(PublishSpec alertPublishSpec, Map<String, StreamDefinition> sds) {
+ alertPublishSpecListeners.forEach(s -> s.onAlertPublishSpecChange(alertPublishSpec, sds));
+ }
+
+ public void close() throws IOException {
+ LOG.info("Closed");
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
new file mode 100755
index 0000000..1b93072
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
@@ -0,0 +1,200 @@
+/**
+ * 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.alert.engine.coordinator.impl;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.eagle.alert.config.ConfigBusConsumer;
+import org.apache.eagle.alert.config.ConfigChangeCallback;
+import org.apache.eagle.alert.config.ConfigValue;
+import org.apache.eagle.alert.config.ZKConfig;
+import org.apache.eagle.alert.coordination.model.AlertBoltSpec;
+import org.apache.eagle.alert.coordination.model.PublishSpec;
+import org.apache.eagle.alert.coordination.model.RouterSpec;
+import org.apache.eagle.alert.coordination.model.ScheduleState;
+import org.apache.eagle.alert.coordination.model.SpoutSpec;
+import org.apache.eagle.alert.coordination.model.VersionedPolicyDefinition;
+import org.apache.eagle.alert.coordination.model.VersionedStreamDefinition;
+import org.apache.eagle.alert.engine.coordinator.MetadataType;
+import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+import org.apache.eagle.alert.service.IMetadataServiceClient;
+import org.apache.eagle.alert.service.MetadataServiceClientImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.typesafe.config.Config;
+
+/**
+ * <b>TODO</b>: performance tuning: It is not JVM level service, so it may cause
+ * zookeeper burden in case of too many listeners This does not support
+ * dynamically adding topic, all topics should be available when service object
+ * is created.
+ * <p>
+ * ZK path format is as following:
+ * <ul>
+ * <li>/alert/topology_1/spout</li>
+ * <li>/alert/topology_1/router</li>
+ * <li>/alert/topology_1/alert</li>
+ * <li>/alert/topology_1/publisher</li>
+ * </ul>
+ */
+public class ZKMetadataChangeNotifyService extends AbstractMetadataChangeNotifyService implements ConfigChangeCallback {
+ private static final long serialVersionUID = -1509237694501235144L;
+ private static final Logger LOG = LoggerFactory.getLogger(ZKMetadataChangeNotifyService.class);
+ private ZKConfig zkConfig;
+ private String topologyId;
+ private ConfigBusConsumer consumer;
+
+ private transient IMetadataServiceClient client;
+
+ public ZKMetadataChangeNotifyService(ZKConfig config, String topologyId) {
+ this.zkConfig = config;
+ this.topologyId = topologyId;
+ }
+
+ @Override
+ public void init(Config config, MetadataType type) {
+ super.init(config, type);
+ client = new MetadataServiceClientImpl(config);
+ consumer = new ConfigBusConsumer(zkConfig, topologyId + "/" + getMetadataTopicSuffix(), this);
+ LOG.info("init called for client");
+ }
+
+ /**
+ * @seeAlso Coordinator
+ * @return
+ */
+ private String getMetadataTopicSuffix() {
+ switch (type) {
+ case ALERT_BOLT:
+ return "alert";
+ case ALERT_PUBLISH_BOLT:
+ return "publisher";
+ case SPOUT:
+ return "spout";
+ case STREAM_ROUTER_BOLT:
+ return "router";
+ default:
+ throw new RuntimeException(String.format("unexpected metadata type: %s !", type));
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ consumer.close();
+ LOG.info("Config consumer closed");
+ }
+
+ @Override
+ public void onNewConfig(ConfigValue value) {
+ LOG.info("Metadata changed {}",value);
+
+ if (client == null) {
+ LOG.error("OnNewConfig trigger, but metadata service client is null. Metadata type {}", type);
+ return;
+ }
+
+ // analyze config value and notify corresponding listeners
+ String version = value.getValue().toString();
+ // brute-force load all: this might introduce load's on metadata service.
+ // FIXME : after ScheduleState persisted with better normalization, load
+ // state based on type and version
+ ScheduleState state = client.getVersionedSpec(version);
+ if (state == null) {
+ LOG.error("Failed to load schedule state of version {}, this is possibly a bug, pls check coordinator log !", version);
+ return;
+ }
+ Map<String, StreamDefinition> sds = getStreams(state.getStreamSnapshots());
+ switch (type) {
+ case ALERT_BOLT:
+ // we might query metadata service query get metadata snapshot and StreamDefinition
+ AlertBoltSpec alertSpec = state.getAlertSpecs().get(topologyId);
+ if (alertSpec == null) {
+ LOG.error(" alert spec for version {} not found for topology {} !", version, topologyId);
+ } else {
+ prePopulate(alertSpec, state.getPolicySnapshots());
+ notifyAlertBolt(alertSpec, sds);
+ }
+ break;
+ case ALERT_PUBLISH_BOLT:
+ PublishSpec pubSpec = state.getPublishSpecs().get(topologyId);
+ if (pubSpec == null) {
+ LOG.error(" alert spec for version {} not found for topology {} !", version, topologyId);
+ } else {
+ notifyAlertPublishBolt(pubSpec, sds);
+ }
+ break;
+ case SPOUT:
+ SpoutSpec spoutSpec = state.getSpoutSpecs().get(topologyId);
+ if (spoutSpec == null) {
+ LOG.error(" alert spec for version {} not found for topology {} !", version, topologyId);
+ } else {
+ notifySpout(spoutSpec, sds);
+ }
+ break;
+ case STREAM_ROUTER_BOLT:
+ RouterSpec gSpec = state.getGroupSpecs().get(topologyId);
+ if (gSpec == null) {
+ LOG.error(" alert spec for version {} not found for topology {} !", version, topologyId);
+ } else {
+ notifyStreamRouterBolt(gSpec, sds);
+ }
+ break;
+ default:
+ LOG.error("unexpected metadata type: {} ", type);
+ }
+ }
+
+ private void prePopulate(AlertBoltSpec alertSpec, List<VersionedPolicyDefinition> list) {
+ Map<String, PolicyDefinition> policyMap = listToMap(list);
+ for (Entry<String, List<String>> policyEntry : alertSpec.getBoltPolicyIdsMap().entrySet()) {
+ List<PolicyDefinition> pds = alertSpec.getBoltPoliciesMap().get(policyEntry.getKey());
+ if (pds == null) {
+ pds = new ArrayList<PolicyDefinition>();
+ alertSpec.getBoltPoliciesMap().put(policyEntry.getKey(), pds);
+ }
+ for (String policyName : policyEntry.getValue()) {
+ if (policyMap.containsKey(policyName)) {
+ pds.add(policyMap.get(policyName));
+ }
+ }
+ }
+ }
+
+ private Map<String, StreamDefinition> getStreams(List<VersionedStreamDefinition> streamSnapshots) {
+ Map<String, StreamDefinition> result = new HashMap<String, StreamDefinition>();
+ for (VersionedStreamDefinition vsd : streamSnapshots) {
+ result.put(vsd.getDefinition().getStreamId(), vsd.getDefinition());
+ }
+ return result;
+ }
+
+ private Map<String, PolicyDefinition> listToMap(List<VersionedPolicyDefinition> listStreams) {
+ Map<String, PolicyDefinition> result = new HashMap<String, PolicyDefinition>();
+ for (VersionedPolicyDefinition sd : listStreams) {
+ result.put(sd.getDefinition().getName(), sd.getDefinition());
+ }
+ return result;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyChangeListener.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyChangeListener.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyChangeListener.java
new file mode 100644
index 0000000..22f1408
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyChangeListener.java
@@ -0,0 +1,29 @@
+package org.apache.eagle.alert.engine.evaluator;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+
+/**
+ * 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.
+ */
+public interface PolicyChangeListener {
+ void onPolicyChange(List<PolicyDefinition> added,
+ List<PolicyDefinition> removed,
+ List<PolicyDefinition> modified, Map<String, StreamDefinition> sds);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyGroupEvaluator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyGroupEvaluator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyGroupEvaluator.java
new file mode 100644
index 0000000..dd3ca24
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyGroupEvaluator.java
@@ -0,0 +1,45 @@
+package org.apache.eagle.alert.engine.evaluator;
+
+import java.io.Serializable;
+
+import org.apache.eagle.alert.engine.AlertStreamCollector;
+import org.apache.eagle.alert.engine.StreamContext;
+import org.apache.eagle.alert.engine.model.PartitionedEvent;
+
+/**
+ * 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.
+ */
+
+/**
+ * policy group refers to the policies which belong to the same MonitoredStream
+ * 3 lifecycle steps are involved in PolicyGroupEvaluator
+ * Step 1: create object. Be aware that in distributed environment, this object may be serialized and transferred across network
+ * Step 2: init. This normally is invoked only once before nextEvent is invoked
+ * Step 3: nextEvent
+ * Step 4: close
+ */
+public interface PolicyGroupEvaluator extends PolicyChangeListener, Serializable{
+ void init(StreamContext context, AlertStreamCollector collector);
+
+ /**
+ * Evaluate event
+ */
+ void nextEvent(PartitionedEvent event);
+
+ String getName();
+
+ void close();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyHandlerContext.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyHandlerContext.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyHandlerContext.java
new file mode 100644
index 0000000..2898ebc
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyHandlerContext.java
@@ -0,0 +1,60 @@
+package org.apache.eagle.alert.engine.evaluator;
+
+import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
+
+import backtype.storm.metric.api.MultiCountMetric;
+
+/**
+ * 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.
+ */
+public class PolicyHandlerContext {
+ private PolicyDefinition policyDefinition;
+ private PolicyGroupEvaluator parentEvaluator;
+ private MultiCountMetric policyCounter;
+ private String policyEvaluatorId;
+
+ public PolicyDefinition getPolicyDefinition() {
+ return policyDefinition;
+ }
+
+ public void setPolicyDefinition(PolicyDefinition policyDefinition) {
+ this.policyDefinition = policyDefinition;
+ }
+
+ public PolicyGroupEvaluator getParentEvaluator() {
+ return parentEvaluator;
+ }
+
+ public void setParentEvaluator(PolicyGroupEvaluator parentEvaluator) {
+ this.parentEvaluator = parentEvaluator;
+ }
+
+ public void setPolicyCounter(MultiCountMetric metric) {
+ this.policyCounter = metric;
+ }
+
+ public MultiCountMetric getPolicyCounter() {
+ return policyCounter;
+ }
+
+ public String getPolicyEvaluatorId() {
+ return policyEvaluatorId;
+ }
+
+ public void setPolicyEvaluatorId(String policyEvaluatorId) {
+ this.policyEvaluatorId = policyEvaluatorId;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandler.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandler.java
new file mode 100755
index 0000000..069d321
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandler.java
@@ -0,0 +1,27 @@
+package org.apache.eagle.alert.engine.evaluator;
+
+import org.apache.eagle.alert.engine.Collector;
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+import org.apache.eagle.alert.engine.model.StreamEvent;
+
+/**
+ * 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.
+ */
+public interface PolicyStreamHandler {
+ void prepare(Collector<AlertStreamEvent> collector, PolicyHandlerContext context) throws Exception;
+ void send(StreamEvent event) throws Exception;
+ void close() throws Exception;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandlers.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandlers.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandlers.java
new file mode 100644
index 0000000..2aa70e8
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/PolicyStreamHandlers.java
@@ -0,0 +1,33 @@
+/**
+ * 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.alert.engine.evaluator;
+
+import java.util.Map;
+
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+import org.apache.eagle.alert.engine.evaluator.impl.SiddhiPolicyHandler;
+
+public class PolicyStreamHandlers {
+ public static final String SIDDHI_ENGINE ="siddhi";
+
+ public static PolicyStreamHandler createHandler(String type, Map<String, StreamDefinition> sds){
+ if(SIDDHI_ENGINE.equals(type)) {
+ return new SiddhiPolicyHandler(sds);
+ }
+ throw new IllegalArgumentException("Illegal policy stream handler type: "+type);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java
new file mode 100755
index 0000000..3aa079e
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java
@@ -0,0 +1,118 @@
+/**
+ * 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.alert.engine.evaluator.impl;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.eagle.alert.engine.AlertStreamCollector;
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import backtype.storm.task.OutputCollector;
+
+/**
+ * <h2>Thread Safe Mechanism</h2>
+ * <ul>
+ * <li>
+ * emit() method is thread-safe enough to be called anywhere asynchronously in multi-thread
+ * </li>
+ * <li>
+ * flush() method must be called synchronously, because Storm OutputCollector is not thread-safe
+ * </li>
+ * </ul>
+ */
+public class AlertBoltOutputCollectorThreadSafeWrapper implements AlertStreamCollector {
+ private final OutputCollector delegate;
+ private final LinkedBlockingQueue<AlertStreamEvent> queue;
+ private final static Logger LOG = LoggerFactory.getLogger(AlertBoltOutputCollectorThreadSafeWrapper.class);
+ private final AtomicLong lastFlushTime = new AtomicLong(System.currentTimeMillis());
+ private final AutoAlertFlusher flusher;
+ private final static int MAX_ALERT_DELAY_SECS = 10;
+
+ public AlertBoltOutputCollectorThreadSafeWrapper(OutputCollector outputCollector){
+ this.delegate = outputCollector;
+ this.queue = new LinkedBlockingQueue<>();
+ this.flusher = new AutoAlertFlusher(this);
+ this.flusher.setName(Thread.currentThread().getName()+"-alertFlusher");
+ this.flusher.start();
+ }
+
+ private static class AutoAlertFlusher extends Thread{
+ private final AlertBoltOutputCollectorThreadSafeWrapper collector;
+ private boolean stopped = false;
+ private final static Logger LOG = LoggerFactory.getLogger(AutoAlertFlusher.class);
+
+ private AutoAlertFlusher(AlertBoltOutputCollectorThreadSafeWrapper collector){
+ this.collector = collector;
+ }
+
+ @Override
+ public void run() {
+ LOG.info("Starting");
+ while(!this.stopped){
+ if(System.currentTimeMillis() - collector.lastFlushTime.get() >= MAX_ALERT_DELAY_SECS * 1000L){
+ this.collector.flush();
+ }
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException ignored) {}
+ }
+ LOG.info("Stopped");
+ }
+ public void shutdown(){
+ LOG.info("Stopping");
+ this.stopped = true;
+ }
+ }
+
+ /**
+ * Emit method can be called in multi-thread
+ * @param event
+ */
+ @Override
+ public void emit(AlertStreamEvent event) {
+ try {
+ queue.put(event);
+ } catch (InterruptedException e) {
+ LOG.error(e.getMessage(),e);
+ }
+ }
+
+ /**
+ * Flush will be called in synchronous way like StormBolt.execute() as Storm OutputCollector is not thread-safe
+ */
+ @Override
+ public void flush() {
+ if(!queue.isEmpty()) {
+ List<AlertStreamEvent> events = new ArrayList<>();
+ queue.drainTo(events);
+ events.forEach((event) -> delegate.emit(Arrays.asList(event.getStreamId(), event)));
+ LOG.info("Flushed {} alerts", events.size());
+ }
+ lastFlushTime.set(System.currentTimeMillis());
+ }
+
+ @Override
+ public void close() {
+ this.flusher.shutdown();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java
new file mode 100755
index 0000000..8d8a4d2
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java
@@ -0,0 +1,55 @@
+/**
+ * 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.alert.engine.evaluator.impl;
+
+import java.util.Arrays;
+
+import org.apache.eagle.alert.engine.AlertStreamCollector;
+import org.apache.eagle.alert.engine.StreamContext;
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+
+import backtype.storm.task.OutputCollector;
+
+public class AlertBoltOutputCollectorWrapper implements AlertStreamCollector {
+ private final OutputCollector delegate;
+ private final Object outputLock;
+ private final StreamContext streamContext;
+
+ public AlertBoltOutputCollectorWrapper(OutputCollector outputCollector, Object outputLock, StreamContext streamContext){
+ this.delegate = outputCollector;
+ this.outputLock = outputLock;
+ this.streamContext = streamContext;
+ }
+
+ @Override
+ public void emit(AlertStreamEvent event) {
+ synchronized (outputLock) {
+ streamContext.counter().scope("alert_count").incr();
+ delegate.emit(Arrays.asList(event.getStreamId(), event));
+ }
+ }
+
+ @Override
+ public void flush() {
+ // do nothing
+ }
+
+ @Override
+ public void close() {
+
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java
new file mode 100644
index 0000000..f063618
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java
@@ -0,0 +1,172 @@
+/**
+ * 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.alert.engine.evaluator.impl;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.eagle.alert.engine.AlertStreamCollector;
+import org.apache.eagle.alert.engine.StreamContext;
+import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+import org.apache.eagle.alert.engine.evaluator.PolicyGroupEvaluator;
+import org.apache.eagle.alert.engine.evaluator.PolicyHandlerContext;
+import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandler;
+import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandlers;
+import org.apache.eagle.alert.engine.model.PartitionedEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PolicyGroupEvaluatorImpl implements PolicyGroupEvaluator {
+ private static final long serialVersionUID = -5499413193675985288L;
+
+ private final static Logger LOG = LoggerFactory.getLogger(PolicyGroupEvaluatorImpl.class);
+
+ private AlertStreamCollector collector;
+ // mapping from policy name to PolicyDefinition
+ private volatile Map<String,PolicyDefinition> policyDefinitionMap = new HashMap<>();
+ // mapping from policy name to PolicyStreamHandler
+ private volatile Map<String,PolicyStreamHandler> policyStreamHandlerMap = new HashMap<>();
+ private String policyEvaluatorId;
+ private StreamContext context;
+
+ public PolicyGroupEvaluatorImpl(String policyEvaluatorId){
+ this.policyEvaluatorId = policyEvaluatorId;
+ }
+
+ public void init(StreamContext context, AlertStreamCollector collector) {
+ this.collector = collector;
+ this.policyStreamHandlerMap = new HashMap<>();
+ this.context = context;
+ Thread.currentThread().setName(policyEvaluatorId);
+ }
+
+ public void nextEvent(PartitionedEvent event) {
+ this.context.counter().scope("receive_count").incr();
+ dispatch(event);
+ }
+
+ @Override
+ public String getName() {
+ return this.policyEvaluatorId;
+ }
+
+ public void close() {
+ for(PolicyStreamHandler handler: policyStreamHandlerMap.values()){
+ try {
+ handler.close();
+ } catch (Exception e) {
+ LOG.error("Failed to close handler {}",handler.toString(),e);
+ }
+ }
+ }
+
+ /**
+ * fixme make selection of PolicyStreamHandler to be more efficient
+ * @param partitionedEvent PartitionedEvent
+ */
+ private void dispatch(PartitionedEvent partitionedEvent){
+ boolean handled = false;
+ for(Map.Entry<String,PolicyStreamHandler> policyStreamHandler: policyStreamHandlerMap.entrySet()){
+ if(isAcceptedByPolicy(partitionedEvent,policyDefinitionMap.get(policyStreamHandler.getKey()))){
+ try {
+ handled = true;
+ this.context.counter().scope("eval_count").incr();
+ policyStreamHandler.getValue().send(partitionedEvent.getEvent());
+ } catch (Exception e) {
+ this.context.counter().scope("fail_count").incr();
+ LOG.error("{} failed to handle {}",policyStreamHandler.getValue(), partitionedEvent.getEvent());
+ }
+ }
+ }
+ if(!handled){
+ this.context.counter().scope("drop_count").incr();
+ LOG.warn("Drop stream non-matched event {}, which should not be sent to evaluator", partitionedEvent);
+ } else {
+ this.context.counter().scope("accept_count").incr();
+ }
+ }
+
+ private static boolean isAcceptedByPolicy(PartitionedEvent event, PolicyDefinition policy){
+ return policy.getInputStreams() != null
+ && policy.getInputStreams().contains(event.getEvent().getStreamId())
+ && policy.getPartitionSpec().contains(event.getPartition());
+ }
+
+ @Override
+ public void onPolicyChange(List<PolicyDefinition> added, List<PolicyDefinition> removed, List<PolicyDefinition> modified, Map<String, StreamDefinition> sds) {
+ Map<String,PolicyDefinition> copyPolicies = new HashMap<>(policyDefinitionMap);
+ Map<String,PolicyStreamHandler> copyHandlers = new HashMap<>(policyStreamHandlerMap);
+ for(PolicyDefinition pd : added){
+ inplaceAdd(copyPolicies, copyHandlers, pd, sds);
+ }
+ for(PolicyDefinition pd : removed){
+ inplaceRemove(copyPolicies, copyHandlers, pd);
+ }
+ for(PolicyDefinition pd : modified){
+ inplaceRemove(copyPolicies, copyHandlers, pd);
+ inplaceAdd(copyPolicies, copyHandlers, pd, sds);
+ }
+
+ // logging
+ LOG.info("Policy metadata updated with added={}, removed={}, modified={}", added, removed, modified);
+
+ // switch reference
+ this.policyDefinitionMap = copyPolicies;
+ this.policyStreamHandlerMap = copyHandlers;
+ }
+
+ private void inplaceAdd(Map<String, PolicyDefinition> policies, Map<String, PolicyStreamHandler> handlers, PolicyDefinition policy, Map<String, StreamDefinition> sds) {
+ if(handlers.containsKey(policy.getName())){
+ LOG.error("metadata calculation error, try to add existing PolicyDefinition " + policy);
+ }else {
+ policies.put(policy.getName(), policy);
+ PolicyStreamHandler handler = PolicyStreamHandlers.createHandler(policy.getDefinition().type, sds);
+ try {
+ PolicyHandlerContext context = new PolicyHandlerContext();
+ context.setPolicyCounter(this.context.counter());
+ context.setPolicyDefinition(policy);
+ context.setParentEvaluator(this);
+ context.setPolicyEvaluatorId(policyEvaluatorId);
+ handler.prepare(collector, context);
+ handlers.put(policy.getName(), handler);
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ policies.remove(policy.getName());
+ handlers.remove(policy.getName());
+ }
+ }
+ }
+
+ private void inplaceRemove(Map<String, PolicyDefinition> policies, Map<String, PolicyStreamHandler> handlers, PolicyDefinition policy) {
+ if(handlers.containsKey(policy.getName())) {
+ PolicyStreamHandler handler = handlers.get(policy.getName());
+ try {
+ handler.close();
+ } catch (Exception e) {
+ LOG.error("Failed to close handler {}",handler,e);
+ }finally {
+ policies.remove(policy.getName());
+ handlers.remove(policy.getName());
+ LOG.info("Removed policy: {}",policy);
+ }
+ } else {
+ LOG.error("metadata calculation error, try to remove nonexisting PolicyDefinition: "+policy);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java
new file mode 100644
index 0000000..5b83abb
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java
@@ -0,0 +1,142 @@
+/**
+ * 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.alert.engine.evaluator.impl;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eagle.alert.engine.coordinator.StreamColumn;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.wso2.siddhi.query.api.definition.AbstractDefinition;
+import org.wso2.siddhi.query.api.definition.Attribute;
+
+import com.google.common.base.Preconditions;
+
+public class SiddhiDefinitionAdapter {
+ private final static Logger LOG = LoggerFactory.getLogger(SiddhiDefinitionAdapter.class);
+ public final static String DEFINE_STREAM_TEMPLATE = "define stream %s ( %s );";
+
+ public static String buildStreamDefinition(StreamDefinition streamDefinition){
+ List<String> columns = new ArrayList<>();
+ Preconditions.checkNotNull(streamDefinition,"StreamDefinition is null");
+ if(streamDefinition.getColumns()!=null) {
+ for (StreamColumn column : streamDefinition.getColumns()) {
+ columns.add(String.format("%s %s", column.getName(), convertToSiddhiAttributeType(column.getType()).toString().toLowerCase()));
+ }
+ }else{
+ LOG.warn("No columns found for stream {}"+streamDefinition.getStreamId());
+ }
+ return String.format(DEFINE_STREAM_TEMPLATE, streamDefinition.getStreamId(),StringUtils.join(columns,","));
+ }
+
+ public static Attribute.Type convertToSiddhiAttributeType(StreamColumn.Type type){
+ if(_EAGLE_SIDDHI_TYPE_MAPPING.containsKey(type)){
+ return _EAGLE_SIDDHI_TYPE_MAPPING.get(type);
+ }
+
+ throw new IllegalArgumentException("Unknown stream type: "+type);
+ }
+
+ public static Class<?> convertToJavaAttributeType(StreamColumn.Type type){
+ if(_EAGLE_JAVA_TYPE_MAPPING.containsKey(type)){
+ return _EAGLE_JAVA_TYPE_MAPPING.get(type);
+ }
+
+ throw new IllegalArgumentException("Unknown stream type: "+type);
+ }
+
+ public static StreamColumn.Type convertFromJavaAttributeType(Class<?> type){
+ if(_JAVA_EAGLE_TYPE_MAPPING.containsKey(type)){
+ return _JAVA_EAGLE_TYPE_MAPPING.get(type);
+ }
+
+ throw new IllegalArgumentException("Unknown stream type: "+type);
+ }
+
+ public static StreamColumn.Type convertFromSiddhiAttributeType(Attribute.Type type){
+ if(_SIDDHI_EAGLE_TYPE_MAPPING.containsKey(type)){
+ return _SIDDHI_EAGLE_TYPE_MAPPING.get(type);
+ }
+
+ throw new IllegalArgumentException("Unknown siddhi type: "+type);
+ }
+
+ /**
+ * public enum Type {
+ * STRING, INT, LONG, FLOAT, DOUBLE, BOOL, OBJECT
+ * }
+ */
+ private final static Map<StreamColumn.Type,Attribute.Type> _EAGLE_SIDDHI_TYPE_MAPPING = new HashMap<>();
+ private final static Map<StreamColumn.Type,Class<?>> _EAGLE_JAVA_TYPE_MAPPING = new HashMap<>();
+ private final static Map<Class<?>,StreamColumn.Type> _JAVA_EAGLE_TYPE_MAPPING = new HashMap<>();
+ private final static Map<Attribute.Type,StreamColumn.Type> _SIDDHI_EAGLE_TYPE_MAPPING = new HashMap<>();
+
+ static {
+ _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.STRING,Attribute.Type.STRING);
+ _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.INT,Attribute.Type.INT);
+ _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.LONG,Attribute.Type.LONG);
+ _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.FLOAT,Attribute.Type.FLOAT);
+ _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.DOUBLE,Attribute.Type.DOUBLE);
+ _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.BOOL,Attribute.Type.BOOL);
+ _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.OBJECT,Attribute.Type.OBJECT);
+
+ _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.STRING,String.class);
+ _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.INT,Integer.class);
+ _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.LONG,Long.class);
+ _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.FLOAT,Float.class);
+ _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.DOUBLE,Double.class);
+ _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.BOOL,Boolean.class);
+ _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.OBJECT,Object.class);
+
+ _JAVA_EAGLE_TYPE_MAPPING.put(String.class,StreamColumn.Type.STRING);
+ _JAVA_EAGLE_TYPE_MAPPING.put(Integer.class,StreamColumn.Type.INT);
+ _JAVA_EAGLE_TYPE_MAPPING.put(Long.class,StreamColumn.Type.LONG);
+ _JAVA_EAGLE_TYPE_MAPPING.put(Float.class,StreamColumn.Type.FLOAT);
+ _JAVA_EAGLE_TYPE_MAPPING.put(Double.class,StreamColumn.Type.DOUBLE);
+ _JAVA_EAGLE_TYPE_MAPPING.put(Boolean.class,StreamColumn.Type.BOOL);
+ _JAVA_EAGLE_TYPE_MAPPING.put(Object.class,StreamColumn.Type.OBJECT);
+
+ _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.STRING,StreamColumn.Type.STRING);
+ _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.INT,StreamColumn.Type.INT);
+ _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.LONG,StreamColumn.Type.LONG);
+ _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.FLOAT,StreamColumn.Type.FLOAT);
+ _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.DOUBLE,StreamColumn.Type.DOUBLE);
+ _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.BOOL,StreamColumn.Type.BOOL);
+ _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.OBJECT,StreamColumn.Type.OBJECT);
+ }
+
+ public static StreamDefinition convertFromSiddiDefinition(AbstractDefinition siddhiDefinition){
+ StreamDefinition streamDefinition = new StreamDefinition();
+ streamDefinition.setStreamId(siddhiDefinition.getId());
+ List<StreamColumn> columns = new ArrayList<>(siddhiDefinition.getAttributeNameArray().length);
+ for(Attribute attribute:siddhiDefinition.getAttributeList()){
+ StreamColumn column = new StreamColumn();
+ column.setType(convertFromSiddhiAttributeType(attribute.getType()));
+ column.setName(attribute.getName());
+ columns.add(column);
+ }
+ streamDefinition.setColumns(columns);
+ streamDefinition.setTimeseries(true);
+ streamDefinition.setDescription("Auto-generated stream schema from siddhi for "+siddhiDefinition.getId());
+ return streamDefinition;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
new file mode 100755
index 0000000..dfa5612
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
@@ -0,0 +1,148 @@
+/**
+ * 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.alert.engine.evaluator.impl;
+
+import java.util.Map;
+
+import org.apache.eagle.alert.engine.Collector;
+import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinitionNotFoundException;
+import org.apache.eagle.alert.engine.evaluator.PolicyHandlerContext;
+import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandler;
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+import org.apache.eagle.alert.engine.model.StreamEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.wso2.siddhi.core.ExecutionPlanRuntime;
+import org.wso2.siddhi.core.SiddhiManager;
+import org.wso2.siddhi.core.event.Event;
+import org.wso2.siddhi.core.stream.input.InputHandler;
+import org.wso2.siddhi.core.stream.output.StreamCallback;
+
+public class SiddhiPolicyHandler implements PolicyStreamHandler {
+ private final static Logger LOG = LoggerFactory.getLogger(SiddhiPolicyHandler.class);
+ private ExecutionPlanRuntime executionRuntime;
+ private SiddhiManager siddhiManager;
+ private Map<String, StreamDefinition> sds;
+ private PolicyDefinition policy;
+ private PolicyHandlerContext context;
+
+ public SiddhiPolicyHandler(Map<String, StreamDefinition> sds){
+ this.sds = sds;
+ }
+
+ private static String generateExecutionPlan(PolicyDefinition policyDefinition, Map<String, StreamDefinition> sds) throws StreamDefinitionNotFoundException {
+ StringBuilder builder = new StringBuilder();
+ for(String inputStream:policyDefinition.getInputStreams()) {
+ builder.append(SiddhiDefinitionAdapter.buildStreamDefinition(sds.get(inputStream)));
+ builder.append("\n");
+ }
+ builder.append(policyDefinition.getDefinition().value);
+ if(LOG.isDebugEnabled()) LOG.debug("Generated siddhi execution plan: {} from policy: {}", builder.toString(),policyDefinition);
+ return builder.toString();
+ }
+
+ private static class AlertStreamCallback extends StreamCallback{
+ private final String outputStream;
+ private final Collector<AlertStreamEvent> collector;
+ private final PolicyHandlerContext context;
+ private final StreamDefinition definition;
+
+ public AlertStreamCallback(String outputStream, StreamDefinition streamDefinition, Collector<AlertStreamEvent> collector, PolicyHandlerContext context){
+ this.outputStream = outputStream;
+ this.collector = collector;
+ this.context = context;
+ this.definition = streamDefinition;
+ }
+
+ /**
+ * Possibly more than one event will be triggered for alerting
+ * @param events
+ */
+ @Override
+ public void receive(Event[] events) {
+ LOG.info("Generated {} alerts from policy '{}' in {}", events.length,context.getPolicyDefinition().getName(), context.getPolicyEvaluatorId());
+ for(Event e : events) {
+ AlertStreamEvent event = new AlertStreamEvent();
+ event.setTimestamp(e.getTimestamp());
+ event.setData(e.getData());
+ event.setStreamId(outputStream);
+ event.setPolicy(context.getPolicyDefinition());
+ if (this.context.getParentEvaluator() != null) {
+ event.setCreatedBy(context.getParentEvaluator().getName());
+ }
+ event.setCreatedTime(System.currentTimeMillis());
+ event.setSchema(definition);
+ if(LOG.isDebugEnabled())
+ LOG.debug("Generate new alert event: {}", event);
+ collector.emit(event);
+ }
+ context.getPolicyCounter().scope(String.format("%s.%s",this.context.getPolicyDefinition().getName(),"alert_count")).incrBy(events.length);
+ }
+ }
+
+ @Override
+ public void prepare(final Collector<AlertStreamEvent> collector, PolicyHandlerContext context) throws Exception {
+ LOG.info("Initializing handler for policy {}: {}",context.getPolicyDefinition(),this);
+ this.policy = context.getPolicyDefinition();
+ this.siddhiManager = new SiddhiManager();
+ String plan = generateExecutionPlan(policy, sds);
+ try {
+ this.executionRuntime = siddhiManager.createExecutionPlanRuntime(plan);
+ LOG.info("Created siddhi runtime {}",executionRuntime.getName());
+ }catch (Exception parserException){
+ LOG.error("Failed to create siddhi runtime for policy: {}, siddhi plan: \n\n{}\n",context.getPolicyDefinition().getName(),plan,parserException);
+ throw parserException;
+ }
+ for(final String outputStream:policy.getOutputStreams()){
+ if(executionRuntime.getStreamDefinitionMap().containsKey(outputStream)) {
+ this.executionRuntime.addCallback(outputStream,
+ new AlertStreamCallback(
+ outputStream, SiddhiDefinitionAdapter.convertFromSiddiDefinition(executionRuntime.getStreamDefinitionMap().get(outputStream))
+ ,collector, context));
+ } else {
+ throw new IllegalStateException("Undefined output stream "+outputStream);
+ }
+ }
+ this.executionRuntime.start();
+ this.context = context;
+ LOG.info("Initialized policy handler for policy: {}",policy.getName());
+ }
+
+ public void send(StreamEvent event) throws Exception {
+ context.getPolicyCounter().scope(String.format("%s.%s",this.context.getPolicyDefinition().getName(),"receive_count")).incr();
+ String streamId = event.getStreamId();
+ InputHandler inputHandler = executionRuntime.getInputHandler(streamId);
+ if(inputHandler != null){
+ context.getPolicyCounter().scope(String.format("%s.%s",this.context.getPolicyDefinition().getName(),"eval_count")).incr();
+ inputHandler.send(event.getTimestamp(),event.getData());
+ }else{
+ context.getPolicyCounter().scope(String.format("%s.%s",this.context.getPolicyDefinition().getName(),"drop_count")).incr();
+ LOG.warn("No input handler found for stream {}",streamId);
+ }
+ }
+
+ public void close() {
+ LOG.info("Closing handler for policy {}",this.policy.getName());
+ this.executionRuntime.shutdown();
+ LOG.info("Shutdown siddhi runtime {}",this.executionRuntime.getName());
+ this.siddhiManager.shutdown();
+ LOG.info("Shutdown siddhi manager {}",this.siddhiManager);
+ LOG.info("Closed handler for policy {}",this.policy.getName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java
new file mode 100644
index 0000000..b894cc0
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java
@@ -0,0 +1,31 @@
+/*
+ * 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.alert.engine.publisher;
+
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+
+/**
+ * Dedup Eagle entities.
+ */
+public interface AlertDeduplicator {
+
+ AlertStreamEvent dedup(AlertStreamEvent event);
+
+ void setDedupIntervalMin(String intervalMin);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java
new file mode 100644
index 0000000..8f1c248
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java
@@ -0,0 +1,28 @@
+package org.apache.eagle.alert.engine.publisher;
+
+import java.util.List;
+
+import org.apache.eagle.alert.engine.coordinator.Publishment;
+
+/**
+ * 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.
+ */
+public interface AlertPublishListener {
+ void onPublishChange(List<Publishment> added,
+ List<Publishment> removed,
+ List<Publishment> afterModified,
+ List<Publishment> beforeModified);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java
new file mode 100644
index 0000000..644fe2b
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java
@@ -0,0 +1,50 @@
+/*
+ * 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.alert.engine.publisher;
+
+import org.apache.eagle.alert.engine.coordinator.Publishment;
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+import org.apache.eagle.alert.engine.publisher.impl.PublishStatus;
+
+import com.typesafe.config.Config;
+
+import java.io.Closeable;
+import java.util.Map;
+
+/**
+ * Created on 2/10/16.
+ * Notification Plug-in interface which provide abstraction layer to notify to different system
+ */
+public interface AlertPublishPlugin extends Closeable {
+ /**
+ * for initialization
+ * @throws Exception
+ */
+ void init(Config config, Publishment publishment) throws Exception;
+
+ void update(String dedupIntervalMin, Map<String, String> pluginProperties);
+
+ void close();
+
+ void onAlert(AlertStreamEvent event) throws Exception;
+
+ AlertStreamEvent dedup(AlertStreamEvent event);
+
+ PublishStatus getStatus();
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
new file mode 100644
index 0000000..fc1fc28
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
@@ -0,0 +1,26 @@
+package org.apache.eagle.alert.engine.publisher;
+
+import java.util.Map;
+
+import org.apache.eagle.alert.coordination.model.PublishSpec;
+import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+
+/**
+ * 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.
+ */
+public interface AlertPublishSpecListener {
+ void onAlertPublishSpecChange(PublishSpec spec, Map<String, StreamDefinition> sds);
+}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java
new file mode 100644
index 0000000..7a44009
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java
@@ -0,0 +1,31 @@
+package org.apache.eagle.alert.engine.publisher;
+
+
+import java.io.Serializable;
+
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+
+import com.typesafe.config.Config;
+
+/**
+ * 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.
+ */
+public interface AlertPublisher extends AlertPublishListener, Serializable {
+ void init(Config config);
+ String getName();
+ void nextEvent(AlertStreamEvent event);
+ void close();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java
new file mode 100644
index 0000000..91c9296
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java
@@ -0,0 +1,33 @@
+package org.apache.eagle.alert.engine.publisher;
+
+import java.io.Serializable;
+
+/**
+ * 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.
+ */
+public interface AlertSink extends Serializable {
+ /**
+ *
+ * @throws Exception
+ */
+ void open() throws Exception;
+
+ /**
+ *
+ * @throws Exception
+ */
+ void close() throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java
new file mode 100644
index 0000000..54a9afa
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java
@@ -0,0 +1,48 @@
+/*
+ * 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.alert.engine.publisher;
+
+public class PublishConstants {
+ public static final String NOTIFICATION_TYPE = "type";
+ public static final String EMAIL_NOTIFICATION = "email";
+ public static final String KAFKA_STORE = "kafka";
+ public static final String EAGLE_STORE = "eagleStore";
+
+ // email specific constants
+ public static final String SUBJECT = "subject";
+ public static final String SENDER = "sender";
+ public static final String RECIPIENTS = "recipients";
+ public static final String TEMPLATE = "template";
+
+ // kafka specific constants
+ public static final String TOPIC = "topic";
+ public static final String BROKER_LIST = "kafka_broker";
+
+ public static final String ALERT_EMAIL_TIME_PROPERTY = "timestamp";
+ public static final String ALERT_EMAIL_COUNT_PROPERTY = "count";
+ public static final String ALERT_EMAIL_ALERTLIST_PROPERTY = "alertList";
+ public static final String ALERT_EMAIL_ORIGIN_PROPERTY = "alertEmailOrigin";
+
+ public static final String ALERT_EMAIL_MESSAGE = "alertMessage";
+ public static final String ALERT_EMAIL_STREAM = "streamId";
+ public static final String ALERT_EMAIL_TIMESTAMP = "alertTime";
+ public static final String ALERT_EMAIL_POLICY = "policyId";
+ public static final String ALERT_EMAIL_CREATOR = "creator";
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java
new file mode 100644
index 0000000..7d31bb5
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java
@@ -0,0 +1,39 @@
+/*
+ * 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.alert.engine.publisher.email;
+
+public class AlertEmailConstants {
+
+ public static final String MAIL_AUTH = "mail.smtp.auth";
+ public static final String MAIL_HOST = "mail.smtp.host";
+ public static final String MAIL_PORT = "mail.smtp.port";
+
+ public static final String CONN_PLAINTEXT = "plaintext";
+ public static final String CONN_TLS = "tls";
+ public static final String CONN_SSL = "ssl";
+
+ public static final String CONF_MAIL_SERVER = "smtp.server";
+ public static final String CONF_MAIL_PORT = "smtp.port";
+ public static final String CONF_MAIL_CONN = "connection";
+ public static final String CONF_MAIL_DEBUG = "mailDebug";
+ public static final String CONF_MAIL_AUTH = "smtp.auth.enable";
+ public static final String CONF_AUTH_USER = "auth.username";
+ public static final String CONF_AUTH_PASSWORD = "auth.password";
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java
new file mode 100644
index 0000000..8723283
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java
@@ -0,0 +1,71 @@
+/*
+ * 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.alert.engine.publisher.email;
+
+import java.util.Map;
+
+/**
+ * alert email bean
+ * one email consists of a list of email component
+ */
+public class AlertEmailContext {
+ private Map<String, String> alertContext;
+ private String sender;
+ private String subject;
+ private String recipients;
+ private String velocityTplFile;
+ private String cc;
+
+ public Map<String, String> getAlertContext() {
+ return alertContext;
+ }
+
+ public void setAlertContext(Map<String, String> alertContext) {
+ this.alertContext = alertContext;
+ }
+ public String getVelocityTplFile() {
+ return velocityTplFile;
+ }
+ public void setVelocityTplFile(String velocityTplFile) {
+ this.velocityTplFile = velocityTplFile;
+ }
+ public String getRecipients() {
+ return recipients;
+ }
+ public void setRecipients(String recipients) {
+ this.recipients = recipients;
+ }
+ public String getSender() {
+ return sender;
+ }
+ public void setSender(String sender) {
+ this.sender = sender;
+ }
+ public String getSubject() {
+ return subject;
+ }
+ public void setSubject(String subject) {
+ this.subject = subject;
+ }
+ public String getCc() {
+ return cc;
+ }
+ public void setCc(String cc) {
+ this.cc = cc;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java
new file mode 100644
index 0000000..0c68629
--- /dev/null
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java
@@ -0,0 +1,143 @@
+/*
+ * 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.alert.engine.publisher.email;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+import org.apache.eagle.alert.engine.publisher.PublishConstants;
+import org.apache.eagle.alert.utils.DateTimeUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AlertEmailGenerator {
+ private String tplFile;
+ private String sender;
+ private String recipients;
+ private String subject;
+ private Map<String, String> properties;
+
+ private ThreadPoolExecutor executorPool;
+
+ private final static Logger LOG = LoggerFactory.getLogger(AlertEmailGenerator.class);
+
+ private final static long MAX_TIMEOUT_MS =60000;
+
+ public boolean sendAlertEmail(AlertStreamEvent entity) {
+ return sendAlertEmail(entity, recipients, null);
+ }
+
+ public boolean sendAlertEmail(AlertStreamEvent entity, String recipients) {
+ return sendAlertEmail(entity, recipients, null);
+ }
+
+ public boolean sendAlertEmail(AlertStreamEvent event, String recipients, String cc) {
+ AlertEmailContext email = new AlertEmailContext();
+ Map<String, String> alertContext = buildAlertContext(event);
+ email.setAlertContext(alertContext);
+ email.setVelocityTplFile(tplFile);
+ email.setSubject(subject);
+ email.setSender(sender);
+ email.setRecipients(recipients);
+ email.setCc(cc);
+
+ /** asynchronized email sending */
+ AlertEmailSender thread = new AlertEmailSender(email, properties);
+
+ if(this.executorPool == null) throw new IllegalStateException("Invoking thread executor pool but it's is not set yet");
+
+ LOG.info("Sending email in asynchronous to: " + recipients + ", cc: " + cc);
+ Future<?> future = this.executorPool.submit(thread);
+ Boolean status;
+ try {
+ future.get(MAX_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ status = true;
+ //LOG.info(String.format("Successfully send email to %s", recipients));
+ } catch (InterruptedException | ExecutionException e) {
+ status = false;
+ LOG.error(String.format("Failed to send email to %s, due to:%s", recipients, e),e);
+ } catch (TimeoutException e) {
+ status = false;
+ LOG.error(String.format("Failed to send email to %s due to timeout exception, max timeout: %s ms ", recipients, MAX_TIMEOUT_MS),e);
+ }
+ return status;
+ }
+
+ private Map<String, String> buildAlertContext(AlertStreamEvent event) {
+ Map<String, String> alertContext = new HashMap<>();
+ alertContext.put(PublishConstants.ALERT_EMAIL_MESSAGE, event.toString());
+ alertContext.put(PublishConstants.ALERT_EMAIL_POLICY, event.getPolicyId());
+ alertContext.put(PublishConstants.ALERT_EMAIL_TIMESTAMP, DateTimeUtil.millisecondsToHumanDateWithSeconds(event.getCreatedTime()));
+ alertContext.put(PublishConstants.ALERT_EMAIL_STREAM, event.getStreamId());
+ alertContext.put(PublishConstants.ALERT_EMAIL_CREATOR, event.getCreatedBy());
+ return alertContext;
+ }
+
+ public String getTplFile() {
+ return tplFile;
+ }
+
+ public void setTplFile(String tplFile) {
+ this.tplFile = tplFile;
+ }
+
+ public String getSender() {
+ return sender;
+ }
+
+ public void setSender(String sender) {
+ this.sender = sender;
+ }
+
+ public String getRecipients() {
+ return recipients;
+ }
+
+ public void setRecipients(String recipients) {
+ this.recipients = recipients;
+ }
+
+ public String getSubject() {
+ return subject;
+ }
+
+ public void setSubject(String subject) {
+ this.subject = subject;
+ }
+
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
+ public void setProperties(Map<String, String> properties) {
+ this.properties = properties;
+ }
+
+ public void setExecutorPool(ThreadPoolExecutor executorPool) {
+ this.executorPool = executorPool;
+ }
+}
\ No newline at end of file