eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [48/55] [abbrv] [partial] incubator-eagle git commit: [EAGLE-46] Rename package name as "org.apache.eagle"
Date Thu, 19 Nov 2015 10:47:55 GMT
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamEntity.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamEntity.java b/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamEntity.java
new file mode 100644
index 0000000..f0f81c5
--- /dev/null
+++ b/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamEntity.java
@@ -0,0 +1,51 @@
+/*
+ * 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.entity;
+
+import org.apache.eagle.alert.common.AlertConstants;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import org.apache.eagle.log.base.taggedlog.TaggedLogAPIEntity;
+import org.apache.eagle.log.entity.meta.Column;
+import org.apache.eagle.log.entity.meta.ColumnFamily;
+import org.apache.eagle.log.entity.meta.Prefix;
+import org.apache.eagle.log.entity.meta.Service;
+import org.apache.eagle.log.entity.meta.Table;
+import org.apache.eagle.log.entity.meta.Tags;
+import org.apache.eagle.log.entity.meta.TimeSeries;
+
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@Table("alertStream")
+@ColumnFamily("f")
+@Prefix("alertStream")
+@Service(AlertConstants.ALERT_STREAM_SERVICE_ENDPOINT_NAME)
+@JsonIgnoreProperties(ignoreUnknown = true)
+@TimeSeries(false)
+@Tags({"dataSource", "streamName"})
+public class AlertStreamEntity extends TaggedLogAPIEntity{
+	@Column("a")
+	private String desc;	
+
+	public String getDesc() {
+		return desc;
+	}
+	public void setDesc(String desc) {
+		this.desc = desc;
+		valueChanged("desc");
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamSchemaEntity.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamSchemaEntity.java b/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamSchemaEntity.java
new file mode 100644
index 0000000..76b6097
--- /dev/null
+++ b/eagle-core/eagle-alert/eagle-alert-base/src/main/java/org/apache/eagle/alert/entity/AlertStreamSchemaEntity.java
@@ -0,0 +1,111 @@
+/*
+ * 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.entity;
+
+import org.apache.eagle.alert.common.AlertConstants;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import org.apache.eagle.log.base.taggedlog.TaggedLogAPIEntity;
+import org.apache.eagle.log.entity.meta.Column;
+import org.apache.eagle.log.entity.meta.ColumnFamily;
+import org.apache.eagle.log.entity.meta.Prefix;
+import org.apache.eagle.log.entity.meta.Service;
+import org.apache.eagle.log.entity.meta.Table;
+import org.apache.eagle.log.entity.meta.Tags;
+import org.apache.eagle.log.entity.meta.TimeSeries;
+
+/**
+ * ddl to create streammetadata table
+ * 
+ * create 'alertStreamSchema', {NAME => 'f', BLOOMFILTER => 'ROW', VERSIONS => '1', COMPRESSION => 'SNAPPY'}
+ */
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@Table("alertStreamSchema")
+@ColumnFamily("f")
+@Prefix("alertStreamSchema")
+@Service(AlertConstants.ALERT_STREAM_SCHEMA_SERVICE_ENDPOINT_NAME)
+@JsonIgnoreProperties(ignoreUnknown = true)
+@TimeSeries(false)
+@Tags({"dataSource", "streamName", "attrName"})
+public class AlertStreamSchemaEntity extends TaggedLogAPIEntity{
+	@Column("a")
+	private String attrType;
+	@Column("b")
+	private String category;
+	@Column("c")
+	private String attrValueResolver;
+	/* all tags form the key for alert de-duplication */
+	@Column("d")
+	private Boolean usedAsTag;
+	@Column("e")
+	private String attrDescription;
+	@Column("f")
+	private String attrDisplayName;	
+	@Column("g")
+	private String defaultValue;
+
+	public String getAttrType() {
+		return attrType;
+	}
+	public void setAttrType(String attrType) {
+		this.attrType = attrType;
+		valueChanged("attrType");
+	}
+	public String getCategory() {
+		return category;
+	}
+	public void setCategory(String category) {
+		this.category = category;
+		valueChanged("category");
+	}
+	public String getAttrValueResolver() {
+		return attrValueResolver;
+	}
+	public void setAttrValueResolver(String attrValueResolver) {
+		this.attrValueResolver = attrValueResolver;
+		valueChanged("attrValueResolver");
+	}
+	public Boolean getUsedAsTag() {
+		return usedAsTag;
+	}
+	public void setUsedAsTag(Boolean usedAsTag) {
+		this.usedAsTag = usedAsTag;
+		valueChanged("usedAsTag");
+	}
+	public String getAttrDescription() {
+		return attrDescription;
+	}
+	public void setAttrDescription(String attrDescription) {
+		this.attrDescription = attrDescription;
+		valueChanged("attrDescription");
+	}
+	public String getAttrDisplayName() {
+		return attrDisplayName;
+	}
+	public void setAttrDisplayName(String attrDisplayName) {
+		this.attrDisplayName = attrDisplayName;
+		valueChanged("attrDisplayName");
+	}
+	public String getDefaultValue() {
+		return defaultValue;
+	}
+	public void setDefaultValue(String defaultValue) {
+		this.defaultValue = defaultValue;
+		valueChanged("defaultValue");
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/AbstractPolicyDefinition.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/AbstractPolicyDefinition.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/AbstractPolicyDefinition.java
deleted file mode 100644
index c235225..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/AbstractPolicyDefinition.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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 eagle.alert.config;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
-/**
- * base fields for all policy definition
- */
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", visible=true)
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class AbstractPolicyDefinition {
-	private String type;
-    /**
-     * @return type in string
-     */
-	public String getType() {
-		return type;
-	}
-
-    /**
-     * @param type set type value
-     */
-	public void setType(String type) {
-		this.type = type;
-	}
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/DeduplicatorConfig.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/DeduplicatorConfig.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/DeduplicatorConfig.java
deleted file mode 100644
index 04f20ff..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/DeduplicatorConfig.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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 eagle.alert.config;
-
-import java.io.Serializable;
-
-public class DeduplicatorConfig implements Serializable{
-	private static final long serialVersionUID = 1L;
-	public int getAlertDedupIntervalMin() {
-		return alertDedupIntervalMin;
-	}
-	public void setAlertDedupIntervalMin(int alertDedupIntervalMin) {
-		this.alertDedupIntervalMin = alertDedupIntervalMin;
-	}
-	public int getEmailDedupIntervalMin() {
-		return emailDedupIntervalMin;
-	}
-	public void setEmailDedupIntervalMin(int emailDedupIntervalMin) {
-		this.emailDedupIntervalMin = emailDedupIntervalMin;
-	}
-	private int alertDedupIntervalMin;
-	private int emailDedupIntervalMin;
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/EmailNotificationConfig.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/EmailNotificationConfig.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/EmailNotificationConfig.java
deleted file mode 100644
index fae24d5..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/EmailNotificationConfig.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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 eagle.alert.config;
-
-public class EmailNotificationConfig extends NotificationConfig{
-	private static final long serialVersionUID = 1L;
-	private String sender;
-	private String recipients;
-	private String tplFileName;
-	private String subject;
-	public String getSubject() {
-		return subject;
-	}
-	public void setSubject(String subject) {
-		this.subject = subject;
-	}
-	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 getTplFileName() {
-		return tplFileName;
-	}
-	public void setTplFileName(String tplFileName) {
-		this.tplFileName = tplFileName;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/NotificationConfig.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/NotificationConfig.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/NotificationConfig.java
deleted file mode 100644
index 6903a57..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/NotificationConfig.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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 eagle.alert.config;
-
-import java.io.Serializable;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "flavor", visible=true)
-public class NotificationConfig implements Serializable{
-	private static final long serialVersionUID = 1L;
-	private String id;
-	private String flavor;
-
-	public String getId() {
-		return id;
-	}
-
-	public void setId(String id) {
-		this.id = id;
-	}
-
-	public String getFlavor() {
-		return flavor;
-	}
-
-	public void setFlavor(String flavor) {
-		this.flavor = flavor;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/Remediation.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/Remediation.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/Remediation.java
deleted file mode 100644
index c644a31..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/config/Remediation.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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 eagle.alert.config;
-
-import java.io.Serializable;
-
-public class Remediation implements Serializable{
-	private static final long serialVersionUID = 1L;
-	private String id;
-
-	public String getId() {
-		return id;
-	}
-
-	public void setId(String id) {
-		this.id = id;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertDeduplicationExecutorBase.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertDeduplicationExecutorBase.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertDeduplicationExecutorBase.java
deleted file mode 100644
index b459130..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertDeduplicationExecutorBase.java
+++ /dev/null
@@ -1,191 +0,0 @@
-/*
- * 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 eagle.alert.dedup;
-
-import eagle.alert.common.AlertConstants;
-import eagle.alert.config.DeduplicatorConfig;
-import eagle.alert.dao.AlertDefinitionDAO;
-import eagle.alert.entity.AlertAPIEntity;
-import eagle.alert.entity.AlertDefinitionAPIEntity;
-import eagle.alert.policy.DynamicPolicyLoader;
-import eagle.alert.policy.PolicyLifecycleMethods;
-import eagle.common.config.EagleConfigConstants;
-import eagle.datastream.Collector;
-import eagle.datastream.JavaStormStreamExecutor2;
-import eagle.datastream.Tuple2;
-import com.sun.jersey.client.impl.CopyOnWriteHashMap;
-import com.typesafe.config.Config;
-import eagle.dataproc.core.JsonSerDeserUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public abstract class AlertDeduplicationExecutorBase extends JavaStormStreamExecutor2<String, AlertAPIEntity> implements PolicyLifecycleMethods {
-	private static final long serialVersionUID = 1L;
-	private static final Logger LOG = LoggerFactory.getLogger(AlertDeduplicationExecutorBase.class);
-	protected Config config;
-	protected DEDUP_TYPE dedupType;
-
-	private List<String> alertExecutorIdList;
-	private volatile CopyOnWriteHashMap<String, DefaultDeduplicator<AlertAPIEntity>> alertDedups;
-	private AlertDefinitionDAO dao;
-
-	public enum DEDUP_TYPE {
-		ENTITY,
-		EMAIL
-	}
-
-	public AlertDeduplicationExecutorBase(List<String> alertExecutorIdList, DEDUP_TYPE dedupType, AlertDefinitionDAO dao){
-		this.alertExecutorIdList = alertExecutorIdList;
-		this.dedupType = dedupType;
-		this.dao = dao;
-	}
-	
-	@Override
-	public void prepareConfig(Config config) {
-		this.config = config;
-	}
-	
-	public DefaultDeduplicator<AlertAPIEntity> createAlertDedup(AlertDefinitionAPIEntity alertDef) {
-		DeduplicatorConfig dedupConfig = null;
-		try {
-			dedupConfig = JsonSerDeserUtils.deserialize(alertDef.getDedupeDef(), DeduplicatorConfig.class);
-		}
-		catch (Exception ex) {
-			LOG.warn("Initial dedupConfig error, " + ex.getMessage());
-		}
-
-        if (dedupConfig != null) {
-			if (dedupType.equals(DEDUP_TYPE.ENTITY)) {
-				return new DefaultDeduplicator<>(dedupConfig.getAlertDedupIntervalMin());
-			} else if (dedupType.equals(DEDUP_TYPE.EMAIL)) {
-				return new DefaultDeduplicator<>(dedupConfig.getEmailDedupIntervalMin());
-			}
-		}
-
-		return null;
-	}
-	
-	@Override
-	public void init() {		
-        String site = config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.SITE);
-        String dataSource = config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.DATA_SOURCE);
-	    Map<String, Map<String, AlertDefinitionAPIEntity>> initialAlertDefs;	    	    
-	    try {
-	 		initialAlertDefs = dao.findActiveAlertDefsGroupbyAlertExecutorId(site, dataSource);
-	    }
-	    catch (Exception ex) {
- 			LOG.error("fail to initialize initialAlertDefs: ", ex);
-	        throw new IllegalStateException("fail to initialize initialAlertDefs: ", ex);
-        }
-	    Map<String, DefaultDeduplicator<AlertAPIEntity>> tmpDeduplicators = new HashMap<String, DefaultDeduplicator<AlertAPIEntity>>();
-        if(initialAlertDefs == null || initialAlertDefs.isEmpty()){
-            LOG.warn("No alert definitions was found for site: "+site+", dataSource: "+dataSource);
-        } else {
-		    for (String alertExecutorId: alertExecutorIdList) {
-			    if(initialAlertDefs.containsKey(alertExecutorId)){
-                    for(AlertDefinitionAPIEntity alertDef : initialAlertDefs.get(alertExecutorId).values()){
-                       try {
-                          DefaultDeduplicator<AlertAPIEntity> deduplicator = createAlertDedup(alertDef);
-                          if (deduplicator != null)
-                              tmpDeduplicators.put(alertDef.getTags().get(AlertConstants.POLICY_ID), deduplicator);
-                          else LOG.warn("The dedup interval is not set, alertDef: " + alertDef);
-                        }
-                        catch (Throwable t) {
-                            LOG.error("Got an exception when initial dedup config, probably dedup config is not set: " + t.getMessage() + "," + alertDef);
-                        }
-                    }
-                } else {
-                    LOG.info(String.format("No alert definitions found for site: %s, dataSource: %s, alertExecutorId: %s",site,dataSource,alertExecutorId));
-                }
-		    }
-        }
-
-		alertDedups = new CopyOnWriteHashMap<>();
-		alertDedups.putAll(tmpDeduplicators);
-		DynamicPolicyLoader policyLoader = DynamicPolicyLoader.getInstance();
-		policyLoader.init(initialAlertDefs, dao, config);
-		for (String alertExecutorId : alertExecutorIdList) {
-		 	policyLoader.addPolicyChangeListener(alertExecutorId, this);
-		}
-	}
-
-    @Override
-    public void flatMap(java.util.List<Object> input, Collector<Tuple2<String, AlertAPIEntity>> outputCollector){
-        String policyId = (String) input.get(0);
-        AlertAPIEntity alertEntity = (AlertAPIEntity) input.get(1);
-        DefaultDeduplicator<AlertAPIEntity> dedup;
-        synchronized(alertDedups) {
-            dedup = alertDedups.get(policyId);
-        }
-
-        List<AlertAPIEntity> ret = Arrays.asList(alertEntity);
-        if (dedup == null) {
-            LOG.warn("Dedup config for policyId " + policyId + " is not set or is not a valid config");
-        } else {
-            if (dedup.getDedupIntervalMin() == -1) {
-                LOG.warn("the dedup interval is set as -1, which mean all alerts should be deduped(skipped)");
-                return;
-            }
-            ret = dedup.dedup(ret);
-        }
-        for (AlertAPIEntity entity : ret) {
-            outputCollector.collect(new Tuple2(policyId, entity));
-        }
-    }
-
-	public void onPolicyCreated(Map<String, AlertDefinitionAPIEntity> added) {
-		if(LOG.isDebugEnabled()) LOG.debug("Alert dedup config to be added : " + added);
-		for(AlertDefinitionAPIEntity alertDef : added.values()){
-			LOG.info("Alert dedup config really added " + alertDef);
-			DefaultDeduplicator<AlertAPIEntity> dedup = createAlertDedup(alertDef);
-			if (dedup != null) {
-				synchronized(alertDedups) {		
-					alertDedups.put(alertDef.getTags().get(AlertConstants.POLICY_ID), dedup);
-				}
-			}
-		}
-	}
-	
-	public void onPolicyChanged(Map<String, AlertDefinitionAPIEntity> changed) {
-		LOG.info("Alert dedup config changed : " + changed);
-		for(AlertDefinitionAPIEntity alertDef : changed.values()){
-			LOG.info("Alert dedup config really changed " + alertDef);
-			DefaultDeduplicator<AlertAPIEntity> dedup = createAlertDedup(alertDef);
-			if (dedup != null) {
-				synchronized(alertDedups) {
-					alertDedups.put(alertDef.getTags().get(AlertConstants.POLICY_ID), dedup);
-				}
-			}
-		}
-	}
-	
-	public void onPolicyDeleted(Map<String, AlertDefinitionAPIEntity> deleted) {
-		LOG.info("alert dedup config deleted : " + deleted);
-		for(AlertDefinitionAPIEntity alertDef : deleted.values()){
-			LOG.info("alert dedup config deleted " + alertDef);
-			// no cleanup to do, just remove it
-			synchronized(alertDedups) {		
-				alertDedups.remove(alertDef.getTags().get(AlertConstants.POLICY_ID));
-			}
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEmailDeduplicationExecutor.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEmailDeduplicationExecutor.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEmailDeduplicationExecutor.java
deleted file mode 100644
index 7dd0ddc..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEmailDeduplicationExecutor.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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 eagle.alert.dedup;
-
-import eagle.alert.dao.AlertDefinitionDAO;
-
-import java.util.List;
-
-public class AlertEmailDeduplicationExecutor extends AlertDeduplicationExecutorBase {
-
-	private static final long serialVersionUID = 1L;
-
-	public AlertEmailDeduplicationExecutor(List<String> alertExecutorIdList, AlertDefinitionDAO dao){
-		super(alertExecutorIdList, DEDUP_TYPE.EMAIL, dao);
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEntityDeduplicationExecutor.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEntityDeduplicationExecutor.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEntityDeduplicationExecutor.java
deleted file mode 100644
index b6050a0..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/AlertEntityDeduplicationExecutor.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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 eagle.alert.dedup;
-
-import eagle.alert.dao.AlertDefinitionDAO;
-
-import java.util.List;
-
-public class AlertEntityDeduplicationExecutor extends AlertDeduplicationExecutorBase {
-
-	private static final long serialVersionUID = 1L;
-
-	public AlertEntityDeduplicationExecutor(List<String> alertExecutorIdList, AlertDefinitionDAO dao){
-		super(alertExecutorIdList, DEDUP_TYPE.ENTITY, dao);
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/DefaultDeduplicator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/DefaultDeduplicator.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/DefaultDeduplicator.java
deleted file mode 100644
index 97e58ac..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/DefaultDeduplicator.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * 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 eagle.alert.dedup;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import org.apache.commons.lang.time.DateUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import eagle.log.base.taggedlog.TaggedLogAPIEntity;
-
-public class DefaultDeduplicator<T extends TaggedLogAPIEntity> implements EntityDeduplicator<T>{
-	protected long dedupIntervalMin;
-	protected Map<EntityTagsUniq, Long> entites = new HashMap<EntityTagsUniq, Long>();
-	public static Logger LOG = LoggerFactory.getLogger(DefaultDeduplicator.class);
-	
-	public static enum AlertDeduplicationStatus{
-		NEW,
-		DUPLICATED,
-		IGNORED
-	}
-	
-	public DefaultDeduplicator() {
-		this.dedupIntervalMin = 0;
-	}
-	
-	public DefaultDeduplicator(long intervalMin) {
-		this.dedupIntervalMin = intervalMin;
-	}
-	
-	public void clearOldCache() {
-		List<EntityTagsUniq> removedkeys = new ArrayList<EntityTagsUniq>();
-		for (Entry<EntityTagsUniq, Long> entry : entites.entrySet()) {
-			EntityTagsUniq entity = entry.getKey();
-			if (System.currentTimeMillis() - 7 * DateUtils.MILLIS_PER_DAY > entity.createdTime) {
-				removedkeys.add(entry.getKey());
-			}
-		}
-		for (EntityTagsUniq alertKey : removedkeys) {
-			entites.remove(alertKey);
-		}
-	}
-	
-	public AlertDeduplicationStatus checkDedup(EntityTagsUniq key){
-		long current = key.timestamp;
-		if(!entites.containsKey(key)){
-			entites.put(key, current);
-			return AlertDeduplicationStatus.NEW;
-		}
-		
-		long last = entites.get(key);
-		if(current - last >= dedupIntervalMin * DateUtils.MILLIS_PER_MINUTE){
-			entites.put(key, current);
-			return AlertDeduplicationStatus.DUPLICATED;
-		}
-		
-		return AlertDeduplicationStatus.IGNORED;
-	}
-	
-	public List<T> dedup(List<T> list) {
-		clearOldCache();
-		List<T> dedupList = new ArrayList<T>();
-        int totalCount = list.size();
-        int dedupedCount = 0;
-		for(T entity: list) {
-			if (entity.getTags() == null) {
-				if(LOG.isDebugEnabled()) LOG.debug("Tags is null, don't know how to deduplicate, do nothing");
-			} else {
-                AlertDeduplicationStatus status = checkDedup(new EntityTagsUniq(entity.getTags(), entity.getTimestamp()));
-                if (!status.equals(AlertDeduplicationStatus.IGNORED)) {
-                    dedupList.add(entity);
-                } else {
-                    dedupedCount++;
-                    if (LOG.isDebugEnabled())
-                        LOG.debug(String.format("Entity is skipped because it's duplicated: " + entity.toString()));
-                }
-            }
-		}
-
-        if(dedupedCount>0){
-            LOG.info(String.format("Skipped %s of %s alerts because they are duplicated",dedupedCount,totalCount));
-        }else if(LOG.isDebugEnabled()){
-            LOG.debug(String.format("Skipped %s of %s duplicated alerts",dedupedCount,totalCount));
-        }
-
-		return dedupList;
-	}
-
-	public EntityDeduplicator<T> setDedupIntervalMin(long dedupIntervalMin) {
-		this.dedupIntervalMin = dedupIntervalMin;
-		return this;
-	}
-	
-	public long getDedupIntervalMin() {
-		return dedupIntervalMin;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityDeduplicator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityDeduplicator.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityDeduplicator.java
deleted file mode 100644
index f24d31a..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityDeduplicator.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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 eagle.alert.dedup;
-
-import java.util.List;
-
-import eagle.log.base.taggedlog.TaggedLogAPIEntity;
-
-/**
- * Dedup Eagle entities.
- *
- * @param <T> Eagle entity
- */
-public interface EntityDeduplicator<T extends TaggedLogAPIEntity> {
-	
-	EntityDeduplicator<T> setDedupIntervalMin(long intervalMin);
-	
-	long getDedupIntervalMin();
-	
-	List<T> dedup(List<T> list);
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityTagsUniq.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityTagsUniq.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityTagsUniq.java
deleted file mode 100644
index b0e5dcf..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/dedup/EntityTagsUniq.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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 eagle.alert.dedup;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * @since Mar 19, 2015
- */
-public class EntityTagsUniq {
-	public Map<String, String> tags;
-	public Long timestamp;	 // entity's timestamp
-	public long createdTime; // entityTagsUniq's created time, for cache removal;
-	
-	private static final Logger LOG = LoggerFactory.getLogger(EntityTagsUniq.class);
-	
-	public EntityTagsUniq(Map<String, String> tags, long timestamp) {
-		this.tags = new HashMap<String, String>(tags);
-		this.timestamp = timestamp;
-		this.createdTime = System.currentTimeMillis();
-	}
-	
-	@Override	
-	public boolean equals(Object obj) {		
-		if (obj instanceof EntityTagsUniq) {
-			EntityTagsUniq au = (EntityTagsUniq) obj;
-			if (tags.size() != au.tags.size()) return false;
-			for (Entry<String, String> keyValue : au.tags.entrySet()) {
-				boolean keyExist = tags.containsKey(keyValue.getKey());
-				// sanity check
-				if (tags.get(keyValue.getKey()) == null || keyValue.getValue() == null) {
-					return true;
-				}
-				if ( !keyExist || !tags.get(keyValue.getKey()).equals(keyValue.getValue())) {				
-					return false;
-				}
-			}
-			return true; 
-		}
-		return false;
-	}
-	
-	@Override
-	public int hashCode() {	
-		int hashCode = 0;
-		for (Map.Entry<String,String> entry : tags.entrySet()) {
-            if(entry.getValue() == null) {
-                LOG.warn("Tag value for key ["+entry.getKey()+"] is null, skipped for hash code");
-            }else {
-                try {
-                    hashCode ^= entry.getValue().hashCode();
-                } catch (Throwable t) {
-                    LOG.info("Got exception because of entry: " + entry, t);
-                }
-            }
-		}
-		return hashCode;
-	}
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGenerator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGenerator.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGenerator.java
deleted file mode 100644
index 58ae733..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGenerator.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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 eagle.alert.notification;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.*;
-
-import eagle.alert.common.AlertConstants;
-import eagle.alert.common.AlertEmailSender;
-import eagle.alert.email.AlertEmailComponent;
-import eagle.alert.email.AlertEmailContext;
-import eagle.alert.entity.AlertAPIEntity;
-import com.typesafe.config.ConfigObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class AlertEmailGenerator{
-	private String tplFile;
-	private String sender;
-	private String recipients;
-	private String subject;
-	private ConfigObject eagleProps;
-
-    private ThreadPoolExecutor executorPool;
-
-    private final static Logger LOG = LoggerFactory.getLogger(AlertEmailGenerator.class);
-
-    private final static long MAX_TIMEOUT_MS =60000;
-
-    public void sendAlertEmail(AlertAPIEntity entity) {
-		sendAlertEmail(entity, recipients, null);
-	}
-	
-	public void sendAlertEmail(AlertAPIEntity entity, String recipients) {
-		sendAlertEmail(entity, recipients, null);	
-	}
-	
-	public void sendAlertEmail(AlertAPIEntity entity, String recipients, String cc) {
-		AlertEmailContext email = new AlertEmailContext();
-		
-		AlertEmailComponent component = new AlertEmailComponent();
-		component.setAlertContext(entity.getAlertContext());
-		List<AlertEmailComponent> components = new ArrayList<AlertEmailComponent>();
-		components.add(component);		
-		email.setComponents(components);
-		if (entity.getAlertContext().getProperty(AlertConstants.SUBJECT) != null) {
-			email.setSubject(entity.getAlertContext().getProperty(AlertConstants.SUBJECT));
-		}
-		else email.setSubject(subject);
-		email.setVelocityTplFile(tplFile);
-		email.setRecipients(recipients);
-		email.setCc(cc);
-		email.setSender(sender);
-		
-		/** asynchronized email sending */
-		@SuppressWarnings("rawtypes")
-		AlertEmailSender thread = new AlertEmailSender(email, eagleProps);
-
-        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);
-        try {
-            future.get(MAX_TIMEOUT_MS, TimeUnit.MILLISECONDS);
-            LOG.info(String.format("Successfully send email to %s", recipients));
-        } catch (InterruptedException | ExecutionException  e) {
-            LOG.error(String.format("Failed to send email to %s, due to:%s",recipients,e),e);
-        } catch (TimeoutException e) {
-            LOG.error(String.format("Failed to send email to %s due to timeout exception, max timeout: %s ms ",recipients, MAX_TIMEOUT_MS),e);
-        }
-    }
-	
-	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 ConfigObject getEagleProps() {
-		return eagleProps;
-	}
-
-	public void setEagleProps(ConfigObject eagleProps) {
-		this.eagleProps = eagleProps;
-	}
-
-    public void setExecutorPool(ThreadPoolExecutor executorPool) {
-        this.executorPool = executorPool;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGeneratorBuilder.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGeneratorBuilder.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGeneratorBuilder.java
deleted file mode 100644
index 10be162..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertEmailGeneratorBuilder.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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 eagle.alert.notification;
-
-import java.util.concurrent.ThreadPoolExecutor;
-
-import com.typesafe.config.ConfigObject;
-
-public class AlertEmailGeneratorBuilder {
-	private AlertEmailGenerator generator;
-	private AlertEmailGeneratorBuilder(){
-		generator = new AlertEmailGenerator();
-	}
-	public static AlertEmailGeneratorBuilder newBuilder(){
-		return new AlertEmailGeneratorBuilder();
-	}
-	public AlertEmailGeneratorBuilder withSubject(String subject){
-		generator.setSubject(subject);
-		return this;
-	}
-	public AlertEmailGeneratorBuilder withSender(String sender){
-		generator.setSender(sender);
-		return this;
-	}
-	public AlertEmailGeneratorBuilder withRecipients(String recipients){
-		generator.setRecipients(recipients);
-		return this;
-	}
-	public AlertEmailGeneratorBuilder withTplFile(String tplFile){
-		generator.setTplFile(tplFile);
-		return this;
-	}
-	public AlertEmailGeneratorBuilder withEagleProps(ConfigObject eagleProps) {
-		generator.setEagleProps(eagleProps);
-		return this;
-	}
-    public AlertEmailGeneratorBuilder withExecutorPool(ThreadPoolExecutor threadPoolExecutor) {
-        generator.setExecutorPool(threadPoolExecutor);
-        return this;
-    }
-
-    public AlertEmailGenerator build(){
-		return this.generator;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertNotificationExecutor.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertNotificationExecutor.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertNotificationExecutor.java
deleted file mode 100644
index 6e5a3d7..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/AlertNotificationExecutor.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * 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 eagle.alert.notification;
-
-import com.fasterxml.jackson.databind.Module;
-import com.fasterxml.jackson.databind.jsontype.NamedType;
-import com.fasterxml.jackson.databind.module.SimpleModule;
-import com.sun.jersey.client.impl.CopyOnWriteHashMap;
-import com.typesafe.config.Config;
-import eagle.alert.config.EmailNotificationConfig;
-import eagle.alert.dao.AlertDefinitionDAO;
-import eagle.alert.entity.AlertAPIEntity;
-import eagle.alert.entity.AlertDefinitionAPIEntity;
-import eagle.alert.policy.DynamicPolicyLoader;
-import eagle.alert.policy.PolicyLifecycleMethods;
-import eagle.dataproc.core.JsonSerDeserUtils;
-import eagle.datastream.Collector;
-import eagle.datastream.JavaStormStreamExecutor1;
-import eagle.datastream.Tuple1;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.*;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-/**
- * notify alert by email, sms or other means
- * currently we only implements email notification
- */
-public class AlertNotificationExecutor extends JavaStormStreamExecutor1<String> implements PolicyLifecycleMethods {
-
-	private static final long serialVersionUID = 1690354365435407034L;
-	private static final Logger LOG = LoggerFactory.getLogger(AlertNotificationExecutor.class);
-	private Config config;
-
-	private List<String> alertExecutorIdList;
-	private volatile CopyOnWriteHashMap<String, List<AlertEmailGenerator>> alertEmailGeneratorsMap;
-	private AlertDefinitionDAO dao;
-
-    private final static int DEFAULT_THREAD_POOL_CORE_SIZE = 4;
-    private final static int DEFAULT_THREAD_POOL_MAX_SIZE = 8;
-    private final static long DEFAULT_THREAD_POOL_SHRINK_TIME = 60000L; // 1 minute
-
-    private transient ThreadPoolExecutor executorPool;
-
-    public AlertNotificationExecutor(List<String> alertExecutorIdList, AlertDefinitionDAO dao){
-		this.alertExecutorIdList = alertExecutorIdList;
-		this.dao = dao;
-	}
-	
-	public List<AlertEmailGenerator> createAlertEmailGenerator(AlertDefinitionAPIEntity alertDef) {		
-		Module module = new SimpleModule("notification").registerSubtypes(new NamedType(EmailNotificationConfig.class, "email"));
-		EmailNotificationConfig[] emailConfigs = new EmailNotificationConfig[0];
-		try {			
-			emailConfigs = JsonSerDeserUtils.deserialize(alertDef.getNotificationDef(), EmailNotificationConfig[].class, Arrays.asList(module));
-		}
-		catch (Exception ex) {
-			LOG.warn("Initial emailConfig error, wrong format or it's error " + ex.getMessage());
-		}
-		List<AlertEmailGenerator> gens = new ArrayList<AlertEmailGenerator>();
-		if (emailConfigs == null) {
-			return gens;		
-		}
-		for(EmailNotificationConfig emailConfig : emailConfigs) {
-			String tplFileName = emailConfig.getTplFileName();			
-			if (tplFileName == null || tplFileName.equals("")) { // empty tplFileName, use default tpl file name
-				tplFileName = "ALERT_DEFAULT.vm";
-			}
-			AlertEmailGenerator gen = AlertEmailGeneratorBuilder.newBuilder().
-																withEagleProps(config.getObject("eagleProps")).
-																withSubject(emailConfig.getSubject()).
-																withSender(emailConfig.getSender()).
-																withRecipients(emailConfig.getRecipients()).
-																withTplFile(tplFileName).
-                                                                withExecutorPool(executorPool).
-																build();
-			gens.add(gen);
-		}
-		return gens;
-	}
-	
-	/**
-	 * 1. register both file and database configuration
-	 * 2. create email generator from configuration
-	 */
-    @Override
-	public void init(){
-        executorPool = new ThreadPoolExecutor(DEFAULT_THREAD_POOL_CORE_SIZE, DEFAULT_THREAD_POOL_MAX_SIZE, DEFAULT_THREAD_POOL_SHRINK_TIME, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
-
-		Map<String, List<AlertEmailGenerator>> tmpEmailGenerators = new HashMap<String, List<AlertEmailGenerator>> ();
-		
-        String site = config.getString("eagleProps.site");
-        String dataSource = config.getString("eagleProps.dataSource");
-	    Map<String, Map<String, AlertDefinitionAPIEntity>> initialAlertDefs;
-	    try {
-	 		initialAlertDefs = dao.findActiveAlertDefsGroupbyAlertExecutorId(site, dataSource);
-	    }
-	    catch (Exception ex) {
- 			LOG.error("fail to initialize initialAlertDefs: ", ex);
-	        throw new IllegalStateException("fail to initialize initialAlertDefs: ", ex);
-        }
- 	   
-        if(initialAlertDefs == null || initialAlertDefs.isEmpty()){
-            LOG.warn("No alert definitions found for site: "+site+", dataSource: "+dataSource);
-        }
-        else {
-		    for (String alertExecutorId: alertExecutorIdList) {
-                if(initialAlertDefs.containsKey(alertExecutorId)) {
-                    for (AlertDefinitionAPIEntity alertDef : initialAlertDefs.get(alertExecutorId).values()) {
-                        List<AlertEmailGenerator> gens = createAlertEmailGenerator(alertDef);
-                        tmpEmailGenerators.put(alertDef.getTags().get("policyId"), gens);
-                    }
-                }else{
-                    LOG.info(String.format("No alert definitions found for site: %s, dataSource: %s, alertExecutorId: %s",site,dataSource,alertExecutorId));
-                }
-		    }
-        }
-		
-		alertEmailGeneratorsMap = new CopyOnWriteHashMap<String, List<AlertEmailGenerator>>();
-		alertEmailGeneratorsMap.putAll(tmpEmailGenerators);				
-		DynamicPolicyLoader policyLoader = DynamicPolicyLoader.getInstance();
-		policyLoader.init(initialAlertDefs, dao, config);
-		for (String alertExecutorId : alertExecutorIdList) {
-			policyLoader.addPolicyChangeListener(alertExecutorId, this);
-		}
-	}
-
-    @Override
-	public void prepareConfig(Config config) {
-		this.config = config;
-	}
-
-    @Override
-    public void flatMap(java.util.List<Object> input, Collector<Tuple1<String>> outputCollector){
-        String policyId = (String) input.get(0);
-        AlertAPIEntity alertEntity = (AlertAPIEntity) input.get(1);
-        processAlerts(policyId, Arrays.asList(alertEntity));
-    }
-	
-	//TODO: add a thread pool for email sender?
-	private void processAlerts(String policyId, List<AlertAPIEntity> list) {
-		List<AlertEmailGenerator> generators;
-		synchronized(alertEmailGeneratorsMap) {		
-			generators = alertEmailGeneratorsMap.get(policyId);
-		}
-		if (generators == null) {
-			LOG.warn("Notification config of policyId " + policyId + " has been deleted");
-			return;
-		}
-		for (AlertAPIEntity entity : list) {
-			for(AlertEmailGenerator generator : generators){
-				generator.sendAlertEmail(entity);
-			}
-		}
-	}
-
-	@Override
-	public void onPolicyCreated(Map<String, AlertDefinitionAPIEntity> added) {
-		if(LOG.isDebugEnabled()) LOG.debug(" alert notification config changed : " + added);
-		for(AlertDefinitionAPIEntity alertDef : added.values()){
-			LOG.info("alert notification config really changed " + alertDef);
-			List<AlertEmailGenerator> gens = createAlertEmailGenerator(alertDef);
-			synchronized(alertEmailGeneratorsMap) {		
-				alertEmailGeneratorsMap.put(alertDef.getTags().get("policyId"), gens);
-			}
-		}		
-	}
-
-	@Override
-	public void onPolicyChanged(Map<String, AlertDefinitionAPIEntity> changed) {
-		if(LOG.isDebugEnabled()) LOG.debug("alert notification config to be added : " + changed);
-		for(AlertDefinitionAPIEntity alertDef : changed.values()){
-			LOG.info("alert notification config really added " + alertDef);
-			List<AlertEmailGenerator> gens = createAlertEmailGenerator(alertDef);
-			synchronized(alertEmailGeneratorsMap) {					
-				alertEmailGeneratorsMap.put(alertDef.getTags().get("policyId"), gens);
-			}
-		}			
-	}
-
-	@Override
-	public void onPolicyDeleted(Map<String, AlertDefinitionAPIEntity> deleted) {
-		if(LOG.isDebugEnabled()) LOG.debug("alert notification config to be deleted : " + deleted);
-		for(AlertDefinitionAPIEntity alertDef : deleted.values()){
-			LOG.info("alert notification config really deleted " + alertDef);
-			synchronized(alertEmailGeneratorsMap) {		
-				alertEmailGeneratorsMap.remove(alertDef.getTags().get("policyId"));
-			}
-		}		
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/UrlBuilder.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/UrlBuilder.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/UrlBuilder.java
deleted file mode 100644
index fa5dc82..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/notification/UrlBuilder.java
+++ /dev/null
@@ -1,47 +0,0 @@
-package eagle.alert.notification;
-
-import eagle.alert.common.AlertConstants;
-import eagle.alert.entity.AlertAPIEntity;
-import eagle.common.EagleBase64Wrapper;
-import eagle.common.config.EagleConfigConstants;
-import eagle.log.entity.HBaseInternalLogHelper;
-import eagle.log.entity.InternalLog;
-import eagle.log.entity.RowkeyBuilder;
-import eagle.log.entity.meta.EntityDefinitionManager;
-import org.mortbay.util.UrlEncoded;
-import java.util.Map;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class UrlBuilder {
-
-    private static final Logger logger = LoggerFactory.getLogger(UrlBuilder.class);
-
-    public static String getEncodedRowkey(AlertAPIEntity entity) throws Exception {
-        InternalLog log = HBaseInternalLogHelper.convertToInternalLog(entity, EntityDefinitionManager.getEntityDefinitionByEntityClass(entity.getClass()));
-        return EagleBase64Wrapper.encodeByteArray2URLSafeString(RowkeyBuilder.buildRowkey(log));
-    }
-
-    public static String buildAlertDetailUrl(String host, int port, AlertAPIEntity entity) {
-        String baseUrl = "http://" + host + ":" + String.valueOf(port) + "/eagle-service/#/dam/alertDetail/";
-        try {
-            return baseUrl + UrlEncoded.encodeString(getEncodedRowkey(entity));
-        }
-        catch (Exception ex) {
-            logger.error("Fail to populate encodedRowkey for alert Entity" + entity.toString());
-            return "N/A";
-        }
-    }
-
-    public static String buiildPolicyDetailUrl(String host, int port, Map<String, String> tags) {
-        String baseUrl = "http://" + host + ":" + String.valueOf(port) + "/eagle-service/#/dam/policyDetail?";
-        String format = "policy=%s&site=%s&executor=%s";
-        String policy = tags.get(AlertConstants.POLICY_ID);
-        String site = tags.get(EagleConfigConstants.SITE);
-        String alertExecutorID = tags.get(AlertConstants.ALERT_EXECUTOR_ID);
-        if (policy != null && site != null && alertExecutorID != null) {
-            return baseUrl + String.format(format, policy, site, alertExecutorID);
-        }
-        return "N/A";
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/AlertPersistExecutor.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/AlertPersistExecutor.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/AlertPersistExecutor.java
deleted file mode 100644
index 543405d..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/AlertPersistExecutor.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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 eagle.alert.persist;
-
-import eagle.alert.entity.AlertAPIEntity;
-import com.typesafe.config.Config;
-import eagle.common.config.EagleConfigConstants;
-import eagle.datastream.Collector;
-import eagle.datastream.JavaStormStreamExecutor1;
-import eagle.datastream.Tuple1;
-
-import java.util.Arrays;
-
-public class AlertPersistExecutor extends JavaStormStreamExecutor1<String> {
-
-	private static final long serialVersionUID = 1L;
-	private Config config;
-	private EaglePersist persist;
-
-	public AlertPersistExecutor(){
-	}
-    @Override
-	public void prepareConfig(Config config) {
-		this.config = config;		
-	}
-
-    @Override
-	public void init() {
-		String host = config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.HOST);
-		int port = config.getInt(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.PORT);
-		String username = config.hasPath(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.USERNAME)
-				? config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.USERNAME) : null;
-		String password = config.hasPath(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.PASSWORD)
-				? config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.PASSWORD) : null;
-		this.persist = new EaglePersist(host, port, username, password);
-	}
-
-    @Override
-    public void flatMap(java.util.List<Object> input, Collector<Tuple1<String>> outputCollector){
-        persist.doPersist(Arrays.asList((AlertAPIEntity)(input.get(1))));
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/EaglePersist.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/EaglePersist.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/EaglePersist.java
deleted file mode 100644
index b5820c4..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/persist/EaglePersist.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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 eagle.alert.persist;
-
-import eagle.log.base.taggedlog.TaggedLogAPIEntity;
-import eagle.log.entity.GenericServiceAPIResponseEntity;
-import eagle.service.client.IEagleServiceClient;
-import eagle.service.client.impl.EagleServiceClientImpl;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-
-public class EaglePersist {
-		
-	private static Logger LOG = LoggerFactory.getLogger(EaglePersist.class);
-	private String eagleServiceHost;
-	private int eagleServicePort;
-	private String username;
-	private String password;
-
-	public EaglePersist(String eagleServiceHost, int eagleServicePort) {
-		this(eagleServiceHost, eagleServicePort, null, null);
-	}
-
-	public EaglePersist(String eagleServiceHost, int eagleServicePort, String username, String password) {
-		this.eagleServiceHost = eagleServiceHost;
-		this.eagleServicePort = eagleServicePort;
-		this.username = username;
-		this.password = password;
-	}
-	
-	public boolean doPersist(List<? extends TaggedLogAPIEntity> list) {
-		if (list.isEmpty()) return false;
-		LOG.info("Going to persist entities, type: " + " " + list.get(0).getClass().getSimpleName() + ", list size: " + list.size());
-		try {
-			IEagleServiceClient client = new EagleServiceClientImpl(eagleServiceHost, eagleServicePort, username, password);
-			GenericServiceAPIResponseEntity<String> response = client.create(list);
-			client.close();
-			if (response.isSuccess()) {
-				LOG.info("Successfully create entities " + list.toString());
-				return true;
-			}
-			else {
-				LOG.error("Fail to create entities");
-				return false;
-			}
-		}
-		catch (Exception ex) {
-			LOG.error("Got an exception in persisting entities" + ex.getMessage(), ex);
-			return false;
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DefaultPolicyPartitioner.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DefaultPolicyPartitioner.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DefaultPolicyPartitioner.java
deleted file mode 100644
index ef76c2f..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DefaultPolicyPartitioner.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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 eagle.alert.policy;
-
-
-public class DefaultPolicyPartitioner implements PolicyPartitioner{
-	@Override
-	public int partition(int numTotalPartitions, String policyType,
-			String policyId) {
-		final int prime = 31;
-		int result = 1;
-		result = result * prime + policyType.hashCode();
-		result = result < 0 ? result*-1 : result;
-		result = result * prime + policyId.hashCode();
-		result = result < 0 ? result*-1 : result;
-		return result % numTotalPartitions;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DynamicPolicyLoader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DynamicPolicyLoader.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DynamicPolicyLoader.java
deleted file mode 100644
index ba76a15..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/DynamicPolicyLoader.java
+++ /dev/null
@@ -1,246 +0,0 @@
-/*
- * 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 eagle.alert.policy;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import eagle.common.config.EagleConfigConstants;
-import org.apache.commons.collections.CollectionUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import eagle.alert.dao.AlertDefinitionDAO;
-import eagle.alert.entity.AlertDefinitionAPIEntity;
-import com.netflix.config.AbstractPollingScheduler;
-import com.netflix.config.ConcurrentCompositeConfiguration;
-import com.netflix.config.DynamicConfiguration;
-import com.netflix.config.FixedDelayPollingScheduler;
-import com.netflix.config.PollListener;
-import com.netflix.config.PollResult;
-import com.netflix.config.PolledConfigurationSource;
-import com.sun.jersey.client.impl.CopyOnWriteHashMap;
-import com.typesafe.config.Config;
-
-public class DynamicPolicyLoader {
-	private static final Logger LOG = LoggerFactory.getLogger(DynamicPolicyLoader.class);
-	
-	private final int defaultInitialDelayMillis = 30*1000;
-	private final int defaultDelayMillis = 60*1000;
-	private final boolean defaultIgnoreDeleteFromSource = true;
-	private volatile CopyOnWriteHashMap<String, List<PolicyLifecycleMethods>> policyChangeListeners = new CopyOnWriteHashMap<String, List<PolicyLifecycleMethods>>();
-	private static DynamicPolicyLoader instance = new DynamicPolicyLoader();
-	private volatile boolean initialized = false;
-	
-	public void addPolicyChangeListener(String alertExecutorId, PolicyLifecycleMethods alertExecutor){
-		synchronized(policyChangeListeners) {
-			if (policyChangeListeners.get(alertExecutorId) == null) {
-				policyChangeListeners.put(alertExecutorId, new ArrayList<PolicyLifecycleMethods>());
-			}
-			policyChangeListeners.get(alertExecutorId).add(alertExecutor);
-		}
-	}
-	
-	public static DynamicPolicyLoader getInstance(){
-		return instance;
-	}
-	
-	/**
-	 * singleton with init would be good for unit test as well, and it ensures that
-	 * initialization happens only once before you use it.  
-	 * @param config
-	 * @param dao
-	 */
-	public void init(Map<String, Map<String, AlertDefinitionAPIEntity>> initialAlertDefs, 
-			AlertDefinitionDAO dao, Config config){
-		if(!initialized){
-			synchronized(this){
-				if(!initialized){
-					internalInit(initialAlertDefs, dao, config);
-					initialized = true;
-				}
-			}
-		}
-	}
-	
-	/**
-	 * map from alertExecutorId+partitionId to AlertExecutor which implements PolicyLifecycleMethods
-	 * @param initialAlertDefs
-	 * @param dao
-	 * @param config
-	 */
-	private void internalInit(Map<String, Map<String, AlertDefinitionAPIEntity>> initialAlertDefs,
-			AlertDefinitionDAO dao, Config config){
-		if(!config.getBoolean("dynamicConfigSource.enabled")) {
-            return;
-        }
-		AbstractPollingScheduler scheduler = new FixedDelayPollingScheduler(
-                config.getInt("dynamicConfigSource.initDelayMillis"),
-                config.getInt("dynamicConfigSource.delayMillis"),
-                false
-        );
-
-		scheduler.addPollListener(new PollListener(){
-			@Override
-			public void handleEvent(EventType eventType, PollResult lastResult,
-					Throwable exception) {
-				if (lastResult == null) {
-					LOG.error("The lastResult is null, something must be wrong, probably the eagle service is dead!");
-					throw new RuntimeException("The lastResult is null, probably the eagle service is dead! ", exception);
-				}
-				Map<String, Object> added = lastResult.getAdded();
-				Map<String, Object> changed = lastResult.getChanged();
-				Map<String, Object> deleted = lastResult.getDeleted();
-				for(Map.Entry<String, List<PolicyLifecycleMethods>> entry : policyChangeListeners.entrySet()){
-					String alertExecutorId = entry.getKey();
-					for (PolicyLifecycleMethods policyLifecycleMethod : entry.getValue()) {
-						Map<String, AlertDefinitionAPIEntity> addedPolicies = (Map<String, AlertDefinitionAPIEntity>)added.get(trimPartitionNum(alertExecutorId));
-						if(addedPolicies != null && addedPolicies.size() > 0){
-							policyLifecycleMethod.onPolicyCreated(addedPolicies);
-						}
-						Map<String, AlertDefinitionAPIEntity> changedPolicies = (Map<String, AlertDefinitionAPIEntity>)changed.get(trimPartitionNum(alertExecutorId));
-						if(changedPolicies != null && changedPolicies.size() > 0){
-							policyLifecycleMethod.onPolicyChanged(changedPolicies);
-						}
-						Map<String, AlertDefinitionAPIEntity> deletedPolicies = (Map<String, AlertDefinitionAPIEntity>)deleted.get(trimPartitionNum(alertExecutorId));
-						if(deletedPolicies != null && deletedPolicies.size() > 0){
-							policyLifecycleMethod.onPolicyDeleted(deletedPolicies);
-						}
-					}
-				}
-			}
-			private String trimPartitionNum(String alertExecutorId){
-				int i = alertExecutorId.lastIndexOf('_');
-				if(i != -1){
-					return alertExecutorId.substring(0, i);
-				}
-				return alertExecutorId;
-			}
-		});
-		
-		ConcurrentCompositeConfiguration finalConfig = new ConcurrentCompositeConfiguration();
-		      
-		PolledConfigurationSource source = new DynamicPolicySource(initialAlertDefs, dao, config);
-
-		try{
-			DynamicConfiguration dbSourcedConfiguration = new DynamicConfiguration(source, scheduler);
-			finalConfig.addConfiguration(dbSourcedConfiguration);
-		}catch(Exception ex){
-			LOG.warn("Fail loading from DB, continue without DB sourced configuration", ex);
-		}
-	}
-	
-	public static class DynamicPolicySource implements PolledConfigurationSource{
-		private static Logger LOG = LoggerFactory.getLogger(DynamicPolicySource.class);
-		private Config config;
-		private AlertDefinitionDAO dao;
-		/**
-		 * mapping from alertExecutorId to list of policies 
-		 */
-		private Map<String, Map<String, AlertDefinitionAPIEntity>> cachedAlertDefs;
-		
-		public DynamicPolicySource(Map<String, Map<String, AlertDefinitionAPIEntity>> initialAlertDefs, AlertDefinitionDAO dao, Config config){
-			this.cachedAlertDefs = initialAlertDefs;
-			this.dao = dao;
-			this.config = config;
-		}
-
-		public PollResult poll(boolean initial, Object checkPoint) throws Exception {
-			LOG.info("Poll policy from eagle service " +  config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.HOST) +
-					":" + config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.EAGLE_SERVICE + "." + EagleConfigConstants.PORT) );
-			Map<String, Map<String, AlertDefinitionAPIEntity>> newAlertDefs = 
-					dao.findActiveAlertDefsGroupbyAlertExecutorId(config.getString("eagleProps.site"),
-                            config.getString("eagleProps.dataSource"));
-			
-			// compare runtime alertDefs with cachedAlertDefs and figure out what are added/deleted/updated
-			Map<String, Object> added = new HashMap<String, Object>();
-			Map<String, Object> changed = new HashMap<String, Object>();
-			Map<String, Object> deleted = new HashMap<String, Object>();
-			
-			Set<String> newAlertExecutorIds = newAlertDefs.keySet();
-			Set<String> cachedAlertExecutorIds = cachedAlertDefs.keySet();
-			
-			// dynamically adding new alert executor is not supported, because alert executor is pre-built while program starts up
-			Collection<String> addedAlertExecutorIds = CollectionUtils.subtract(newAlertExecutorIds, cachedAlertExecutorIds);
-			if(addedAlertExecutorIds != null && addedAlertExecutorIds.size() > 0){
-				LOG.warn("New alertExecutorIds are found : " + addedAlertExecutorIds);
-			}
-			
-			// if one alert executor is missing, it means all policy under that alert executor should be removed
-			Collection<String> deletedAlertExecutorIds = CollectionUtils.subtract(cachedAlertExecutorIds, newAlertExecutorIds);
-			if(deletedAlertExecutorIds != null && deletedAlertExecutorIds.size() > 0){
-				LOG.warn("Some alertExecutorIds are deleted : " + deletedAlertExecutorIds);
-				for(String deletedAlertExecutorId : deletedAlertExecutorIds){
-					deleted.put(deletedAlertExecutorId, cachedAlertDefs.get(deletedAlertExecutorId));
-				}
-			}
-			
-			// we need calculate added/updated/deleted policy for all executors which are not deleted
-//			Collection<String> updatedAlertExecutorIds = CollectionUtils.intersection(newAlertExecutorIds, cachedAlertExecutorIds);
-            Collection<String> updatedAlertExecutorIds = newAlertExecutorIds;
-			for(String updatedAlertExecutorId : updatedAlertExecutorIds){
-				Map<String, AlertDefinitionAPIEntity> newPolicies = newAlertDefs.get(updatedAlertExecutorId);
-				Map<String, AlertDefinitionAPIEntity> cachedPolicies = cachedAlertDefs.get(updatedAlertExecutorId);
-				PolicyComparator.compare(updatedAlertExecutorId, newPolicies, cachedPolicies, added, changed, deleted);
-			}
-			
-			cachedAlertDefs = newAlertDefs;
-			return PollResult.createIncremental(added, changed, deleted, new Date().getTime());
-		}
-	}
-	
-	public static class PolicyComparator{
-		public static void compare(String alertExecutorId, Map<String, AlertDefinitionAPIEntity> newPolicies, Map<String, AlertDefinitionAPIEntity> cachedPolicies, 
-				Map<String, Object> added, Map<String, Object> changed, Map<String, Object> deleted){
-			Set<String> newPolicyIds = newPolicies.keySet();
-            Set<String> cachedPolicyIds = cachedPolicies != null ? cachedPolicies.keySet() : new HashSet<String>();
-			Collection<String> addedPolicyIds = CollectionUtils.subtract(newPolicyIds, cachedPolicyIds);
-			Collection<String> deletedPolicyIds = CollectionUtils.subtract(cachedPolicyIds, newPolicyIds);
-			Collection<String> changedPolicyIds = CollectionUtils.intersection(cachedPolicyIds, newPolicyIds);
-			if(addedPolicyIds != null && addedPolicyIds.size() > 0){
-				Map<String, AlertDefinitionAPIEntity> tmp = new HashMap<String, AlertDefinitionAPIEntity>();
-				for(String addedPolicyId : addedPolicyIds){
-					tmp.put(addedPolicyId, newPolicies.get(addedPolicyId));
-				}
-				added.put(alertExecutorId, tmp);
-			}
-			if(deletedPolicyIds != null && deletedPolicyIds.size() > 0){
-				Map<String, AlertDefinitionAPIEntity> tmp = new HashMap<String, AlertDefinitionAPIEntity>();
-				for(String deletedPolicyId : deletedPolicyIds){
-					tmp.put(deletedPolicyId, cachedPolicies.get(deletedPolicyId));
-				}
-				deleted.put(alertExecutorId, tmp);
-			}
-			if(changedPolicyIds != null && changedPolicyIds.size() > 0){
-				Map<String, AlertDefinitionAPIEntity> tmp = new HashMap<String, AlertDefinitionAPIEntity>();
-				for(String changedPolicyId : changedPolicyIds){
-					// check if policy is really changed
-					if(!newPolicies.get(changedPolicyId).equals(cachedPolicies.get(changedPolicyId))){
-						tmp.put(changedPolicyId, newPolicies.get(changedPolicyId));
-					}
-				}
-				changed.put(alertExecutorId, tmp);
-			}
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PartitionUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PartitionUtils.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PartitionUtils.java
deleted file mode 100644
index c3b53c2..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PartitionUtils.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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 eagle.alert.policy;
-
-import eagle.alert.common.AlertConstants;
-import eagle.alert.entity.AlertDefinitionAPIEntity;
-
-public class PartitionUtils {
-	
-	public static boolean accept(AlertDefinitionAPIEntity alertDef, PolicyPartitioner partitioner, int numPartitions, int partitionSeq){
-		int targetPartitionSeq = partitioner.partition(numPartitions, alertDef.getTags().get(AlertConstants.POLICY_TYPE), alertDef.getTags().get(AlertConstants.POLICY_ID));
-		if(targetPartitionSeq == partitionSeq)
-			return true;
-		return false;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluator.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluator.java
deleted file mode 100644
index c209b97..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluator.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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 eagle.alert.policy;
-
-import java.util.List;
-import java.util.Map;
-
-import eagle.alert.entity.AlertDefinitionAPIEntity;
-import eagle.dataproc.core.ValuesArray;
-
-public interface PolicyEvaluator {
-	/**
-	 * take input and evaluate expression
-	 * input has 3 fields, first is siddhiAlertContext, second one is streamName, the third is map of attribute name/value
-	 * @param input
-	 * @throws Exception
-	 */
-	public void evaluate(ValuesArray input) throws Exception;
-	
-	/**
-	 * notify policy evaluator that policy is updated
-	 */
-	public void onPolicyUpdate(AlertDefinitionAPIEntity newAlertDef);
-	
-	/**
-	 * notify policy evaluator that policy is deleted, here is cleanup work for this policy evaluator
-	 */
-	public void onPolicyDelete();
-	
-	/**
-	 * get additional context
-	 */	
-	public Map<String, String> getAdditionalContext();
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluatorServiceProvider.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluatorServiceProvider.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluatorServiceProvider.java
deleted file mode 100644
index a022f98..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/eagle/alert/policy/PolicyEvaluatorServiceProvider.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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 eagle.alert.policy;
-
-import java.util.List;
-
-import com.fasterxml.jackson.databind.Module;
-
-/**
- * to provide extensibility, we need a clear differentiation between framework job and provider logic
- * policy evaluator framework:
- * - connect to eagle data source
- * - read all policy definitions
- * - compare with cached policy definitions
- * - figure out if policy is created, deleted or updated
- *   - if policy is created, then invoke onPolicyCreated
- *   - if policy is deleted, then invoke onPolicyDeleted
- *   - if policy is updated, then invoke onPolicyUpdated
- * - for policy update, replace old evaluator engine with new evaluator engine which is created by policy evaluator provider
- * - specify # of executors for this alert executor id
- * - dynamically balance # of policies evaluated by each alert executor
- *   - use zookeeper to balance. eaglePolicies/${alertExecutorId}/${alertExecutorInstanceId} => list of policies
- * 
- * policy evaluator business features:
- * - register mapping between policy type and PolicyEvaluator
- * - create evaluator engine runtime when configuration is changed
- *
- */
-public interface PolicyEvaluatorServiceProvider {
-	String getPolicyType();
-	Class<? extends PolicyEvaluator> getPolicyEvaluator();
-	List<Module> getBindingModules();
-}


Mime
View raw message