eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [6/8] incubator-eagle git commit: EAGLE-79 Provide aggregation and persistence DSL support
Date Tue, 12 Jan 2016 07:47:59 GMT
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyDefinition.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyDefinition.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyDefinition.java
deleted file mode 100644
index 39eb0c9..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyDefinition.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 org.apache.eagle.alert.siddhi;
-
-import org.apache.eagle.alert.config.AbstractPolicyDefinition;
-
-/**
- * siddhi policy definition has the following format
- * {
-        "type":"SiddhiCEPEngine",
-		"expression" : "from every b1=HeapUsage[metric == 'eagle.metric.gc'] -> a1=FullGCEvent[eventName == 'full gc'] -> b2=HeapUsage[metric == b1.metric and host == b1.host and value >= b1.value * 0.8] within 100 sec select a1.eventName, b1.metric, b2.timestamp, 60 as timerange insert into GCMonitor; "
-	}
- */
-public class SiddhiPolicyDefinition extends AbstractPolicyDefinition {
-	private String expression;
-
-	public String getExpression() {
-		return expression;
-	}
-	public void setExpression(String expression) {
-		this.expression = expression;
-	}
-	
-	@Override
-	public String toString(){
-		return expression;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluator.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluator.java
deleted file mode 100644
index 87eaf48..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluator.java
+++ /dev/null
@@ -1,269 +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 org.apache.eagle.alert.siddhi;
-
-import java.lang.reflect.Field;
-import java.util.*;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-
-import org.apache.eagle.alert.config.AbstractPolicyDefinition;
-import org.apache.eagle.alert.entity.AlertStreamSchemaEntity;
-import org.apache.eagle.alert.common.AlertConstants;
-import org.apache.eagle.alert.policy.PolicyManager;
-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.query.output.callback.QueryCallback;
-import org.wso2.siddhi.core.stream.input.InputHandler;
-import org.wso2.siddhi.query.api.execution.query.Query;
-import org.wso2.siddhi.query.api.execution.query.selection.OutputAttribute;
-
-import org.apache.eagle.alert.entity.AlertAPIEntity;
-import org.apache.eagle.alert.entity.AlertDefinitionAPIEntity;
-import org.apache.eagle.alert.policy.PolicyEvaluator;
-import org.apache.eagle.dataproc.core.JsonSerDeserUtils;
-import org.apache.eagle.dataproc.core.ValuesArray;
-import com.typesafe.config.Config;
-
-/**
- * when policy is updated or deleted, SiddhiManager.shutdown should be invoked to release resources.
- * during this time, synchronization is important
- */
-public class SiddhiPolicyEvaluator implements PolicyEvaluator{
-	private final static Logger LOG = LoggerFactory.getLogger(SiddhiPolicyEvaluator.class);	
-	public static final int DEFAULT_QUEUE_SIZE = 1000;
-	private final BlockingQueue<AlertAPIEntity> queue = new ArrayBlockingQueue<AlertAPIEntity>(DEFAULT_QUEUE_SIZE);
-	private volatile SiddhiRuntime siddhiRuntime;
-	private String[] sourceStreams;
-	private boolean needValidation;
-	private String policyId;
-	private Config config;
-	private final static String EXECUTION_PLAN_NAME = "query";
-	
-	/**
-	 * everything dependent on policyDef should be together and switched in runtime
-	 */
-	public static class SiddhiRuntime{
-		QueryCallback callback;
-		Map<String, InputHandler> siddhiInputHandlers;
-		SiddhiManager siddhiManager;
-		SiddhiPolicyDefinition policyDef;
-		List<String> outputFields;
-		String executionPlanName;
-	}
-	
-	public SiddhiPolicyEvaluator(Config config, String policyName, AbstractPolicyDefinition policyDef, String[] sourceStreams){
-		this(config, policyName, policyDef, sourceStreams, false);
-	}
-	
-	public SiddhiPolicyEvaluator(Config config, String policyId, AbstractPolicyDefinition policyDef, String[] sourceStreams, boolean needValidation){
-		this.config = config;
-		this.policyId = policyId;
-		this.needValidation = needValidation;
-		this.sourceStreams = sourceStreams; 
-		init(policyDef);
-	}
-	
-	public void init(AbstractPolicyDefinition policyDef){			
-		siddhiRuntime = createSiddhiRuntime((SiddhiPolicyDefinition)policyDef);
-	}
-
-	public static String addContextFieldIfNotExist(String expression) {		
-		// select fieldA, fieldB --> select eagleAlertContext, fieldA, fieldB
-		int pos = expression.indexOf("select ") + 7;
-		int index = pos;
-		boolean isSelectStarPattern = true;
-		while(index < expression.length()) {
-			if (expression.charAt(index) == ' ') index++;
-			else if (expression.charAt(index) == '*') break;
-			else {
-				isSelectStarPattern = false;
-				break;
-			}
-		}
-		if (isSelectStarPattern) return expression;
-		StringBuilder sb = new StringBuilder();
-		sb.append(expression.substring(0, pos));
-		sb.append(SiddhiStreamMetadataUtils.EAGLE_ALERT_CONTEXT_FIELD + ",");
-		sb.append(expression.substring(pos, expression.length()));
-		return sb.toString();
-	}
-
-	private SiddhiRuntime createSiddhiRuntime(SiddhiPolicyDefinition policyDef){
-		SiddhiManager siddhiManager = new SiddhiManager();
-		Map<String, InputHandler> siddhiInputHandlers = new HashMap<String, InputHandler>();
-
-		StringBuilder sb = new StringBuilder();		
-		for(String sourceStream : sourceStreams){
-			String streamDef = SiddhiStreamMetadataUtils.convertToStreamDef(sourceStream);
-			LOG.info("Siddhi stream definition : " + streamDef);
-			sb.append(streamDef);
-		}
-		
-		String expression = addContextFieldIfNotExist(policyDef.getExpression());
-		String executionPlan = sb.toString() + " @info(name = '" + EXECUTION_PLAN_NAME + "') " +  expression;
-		ExecutionPlanRuntime executionPlanRuntime = siddhiManager.createExecutionPlanRuntime(executionPlan);
-		
-		for(String sourceStream : sourceStreams){			
-			siddhiInputHandlers.put(sourceStream, executionPlanRuntime.getInputHandler(sourceStream));
-		}
-		executionPlanRuntime.start();
-
-		QueryCallback callback = new SiddhiQueryCallbackImpl(config, this);		
-
-		LOG.info("Siddhi query: " + expression);
-		executionPlanRuntime.addCallback(EXECUTION_PLAN_NAME, callback);
-
-		List<String> outputFields = new ArrayList<String>();
-		try {
-	        Field field = QueryCallback.class.getDeclaredField(EXECUTION_PLAN_NAME);
-	        field.setAccessible(true);
-	        Query query = (Query)field.get(callback);
-	        List<OutputAttribute> list = query.getSelector().getSelectionList();
-	        for (OutputAttribute output : list) {	        	
-	        	outputFields.add(output.getRename());
-	        }
-		}
-		catch (Exception ex) {
-			LOG.error("Got an Exception when initial outputFields ", ex);
-		}
-		SiddhiRuntime runtime = new SiddhiRuntime();
-		runtime.siddhiInputHandlers = siddhiInputHandlers;
-		runtime.siddhiManager = siddhiManager;
-		runtime.callback = callback;
-		runtime.policyDef = policyDef;
-		runtime.outputFields = outputFields;
-		runtime.executionPlanName = executionPlanRuntime.getName();
-		return runtime;
-	}
-	
-	/**
-	 * 1. input has 3 fields, first is siddhi context, second is streamName, the last one is map of attribute name/value
-	 * 2. runtime check for input data (This is very expensive, so we ignore for now)
-	 *     the size of input map should be equal to size of attributes which stream metadata defines
-	 *     the attribute names should be equal to attribute names which stream metadata defines
-	 *     the input field cannot be null
-	 */
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	@Override
-	public void evaluate(ValuesArray data) throws Exception {
-		if(LOG.isDebugEnabled()) LOG.debug("Siddhi policy evaluator consumers data :" + data);
-        Object siddhiAlertContext = data.get(0);
-		String streamName = (String)data.get(1);
-		SortedMap map = (SortedMap)data.get(2);
-		validateEventInRuntime(streamName, map);
-		synchronized(siddhiRuntime){
-			//insert siddhiAlertContext into the first field
-			List<Object> input = new ArrayList<>();
-			input.add(siddhiAlertContext);
-			putAttrsIntoInputStream(input, streamName, map);
-			siddhiRuntime.siddhiInputHandlers.get(streamName).send(input.toArray(new Object[0]));
-		}
-	}
-
-	/**
-	 * This is a heavy operation, we should avoid to use.
-     *
-     * This validation method will skip invalid fields in event which are not declared in stream schema otherwise it will cause exception for siddhi engine.
-     *
-     * @see <a href="https://issues.apache.org/jira/browse/EAGLE-49">https://issues.apache.org/jira/browse/EAGLE-49</a>
-     *
-	 * @param sourceStream source steam id
-	 * @param data input event
-	 */
-	private void validateEventInRuntime(String sourceStream, SortedMap data){
-		if(!needValidation)
-			return;
-		SortedMap<String, AlertStreamSchemaEntity> map = StreamMetadataManager.getInstance().getMetadataEntityMapForStream(sourceStream);
-        if(!map.keySet().equals(data.keySet())){
-            Set<Object> badKeys = new TreeSet<>();
-            for(Object key:data.keySet()) if(!map.containsKey(key)) badKeys.add(key);
-            LOG.warn(String.format("Ignore invalid fields %s in event: %s from stream: %s, valid fields are: %s", badKeys.toString(),data.toString(), sourceStream,map.keySet().toString()));
-            for(Object key:badKeys) data.remove(key);
-        }
-	}
-
-	private void putAttrsIntoInputStream(List<Object> input, String streamName, SortedMap map) {
-		if(!needValidation) {
-			input.addAll(map.values());
-			return;
-		}
-		for (Object key : map.keySet()) {
-			Object value = map.get(key);
-			if (value == null) {
-				input.add(SiddhiStreamMetadataUtils.getAttrDefaultValue(streamName, (String)key));
-			}
-			else input.add(value);
-		}
-	}
-
-	@Override
-	public void onPolicyUpdate(AlertDefinitionAPIEntity newAlertDef) {
-		AbstractPolicyDefinition policyDef = null;
-		try {
-			policyDef = JsonSerDeserUtils.deserialize(newAlertDef.getPolicyDef(), 
-					AbstractPolicyDefinition.class, PolicyManager.getInstance().getPolicyModules(newAlertDef.getTags().get(AlertConstants.POLICY_TYPE)));
-		}
-		catch (Exception ex) {
-			LOG.error("Initial policy def error, ", ex);
-		}
-		SiddhiRuntime previous = siddhiRuntime;
-		siddhiRuntime = createSiddhiRuntime((SiddhiPolicyDefinition)policyDef);
-		synchronized(previous){
-			previous.siddhiManager.getExecutionPlanRuntime(previous.executionPlanName).shutdown();
-		}
-	}
-	
-	@Override
-	public void onPolicyDelete(){
-		synchronized(siddhiRuntime){
-			LOG.info("Going to shutdown siddhi execution plan, planName: " + siddhiRuntime.executionPlanName);
-			siddhiRuntime.siddhiManager.getExecutionPlanRuntime(siddhiRuntime.executionPlanName).shutdown();
-			LOG.info("Siddhi execution plan " + siddhiRuntime.executionPlanName + " is successfully shutdown ");
-		}
-	}
-	
-	@Override
-	public String toString(){
-		// show the policyDef
-		return siddhiRuntime.policyDef.toString();
-	}
-	
-	public String[] getStreamNames() {
-		return sourceStreams;
-	}
-
-	public Map<String, String> getAdditionalContext() {
-		Map<String, String> context = new HashMap<String, String>();
-		StringBuilder sourceStreams = new StringBuilder();
-		for (String streamName : getStreamNames()) {
-			sourceStreams.append(streamName + ",");
-		}
-		if (sourceStreams.length() > 0) {
-			sourceStreams.deleteCharAt(sourceStreams.length() - 1);
-		}
-		context.put(AlertConstants.SOURCE_STREAMS, sourceStreams.toString());
-		context.put(AlertConstants.POLICY_ID, policyId);
-		return context;
-	}
-
-	public List<String> getOutputStreamAttrNameList() {
-		return siddhiRuntime.outputFields;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluatorServiceProviderImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluatorServiceProviderImpl.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluatorServiceProviderImpl.java
deleted file mode 100644
index 168b04f..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiPolicyEvaluatorServiceProviderImpl.java
+++ /dev/null
@@ -1,45 +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 org.apache.eagle.alert.siddhi;
-
-import java.util.Arrays;
-import java.util.List;
-
-import org.apache.eagle.alert.common.AlertConstants;
-import org.apache.eagle.alert.policy.PolicyEvaluator;
-import org.apache.eagle.alert.policy.PolicyEvaluatorServiceProvider;
-import com.fasterxml.jackson.databind.Module;
-import com.fasterxml.jackson.databind.jsontype.NamedType;
-import com.fasterxml.jackson.databind.module.SimpleModule;
-
-public class SiddhiPolicyEvaluatorServiceProviderImpl implements PolicyEvaluatorServiceProvider {
-	@Override
-	public String getPolicyType() {
-		return AlertConstants.policyType.siddhiCEPEngine.name();
-	}
-
-	@Override
-	public Class<? extends PolicyEvaluator> getPolicyEvaluator() {
-		return SiddhiPolicyEvaluator.class;
-	}
-
-	@Override
-	public List<Module> getBindingModules() {
-		Module module1 = new SimpleModule(AlertConstants.POLICY_DEFINITION).registerSubtypes(new NamedType(SiddhiPolicyDefinition.class, getPolicyType()));
-		return Arrays.asList(module1);
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiQueryCallbackImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiQueryCallbackImpl.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiQueryCallbackImpl.java
deleted file mode 100644
index 44b9c77..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiQueryCallbackImpl.java
+++ /dev/null
@@ -1,85 +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 org.apache.eagle.alert.siddhi;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import org.apache.eagle.alert.entity.AlertAPIEntity;
-import org.apache.eagle.executor.AlertExecutor;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.wso2.siddhi.core.event.Event;
-import org.wso2.siddhi.core.query.output.callback.QueryCallback;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.typesafe.config.Config;
-
-public class SiddhiQueryCallbackImpl extends QueryCallback{
-
-	private SiddhiPolicyEvaluator evaluator;
-	public static final Logger LOG = LoggerFactory.getLogger(SiddhiQueryCallbackImpl.class);
-	public static final ObjectMapper mapper = new ObjectMapper();	
-	public Config config;
-	
-	public SiddhiQueryCallbackImpl(Config config, SiddhiPolicyEvaluator evaluator) {
-		this.config = config;		
-		this.evaluator = evaluator;
-	}
-	
-	public List<String> getOutputMessage(Event event) {
-		Object[] data = event.getData();
-		List<String> rets = new ArrayList<String>();
-		boolean isFirst = true;
-		for (Object object : data) {
-			// The first field is siddhiAlertContext, skip it
-			if (isFirst) {
-				isFirst = false;
-				continue;
-			}
-			String value = null;
-			if (object instanceof Double) {
-				value = String.valueOf((Double)object);
-			}
-			else if (object instanceof Integer) {
-				value = String.valueOf((Integer)object);
-			}
-			else if (object instanceof Long) {
-				value = String.valueOf((Long)object);
-			}
-			else if (object instanceof String) {
-				value = (String)object;
-			}
-			else if (object instanceof Boolean) {
-				value = String.valueOf((Boolean)object);
-			}
-			rets.add(value);
-		}
-		return rets;
-	}
-	
-	@Override
-	public void receive(long timeStamp, Event[] inEvents, Event[] removeEvents) {
-		Object[] data = inEvents[0].getData();
-		EagleAlertContext siddhiAlertContext = (EagleAlertContext)data[0];
-		List<String> rets = getOutputMessage(inEvents[0]);
-		AlertAPIEntity alert = SiddhiAlertAPIEntityRendner.render(config, rets, siddhiAlertContext, timeStamp);
-		AlertExecutor alertExecutor = siddhiAlertContext.alertExecutor;
-		alertExecutor.onAlerts(siddhiAlertContext, Arrays.asList(alert));
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiStreamMetadataUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiStreamMetadataUtils.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiStreamMetadataUtils.java
deleted file mode 100644
index 92394aa..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/SiddhiStreamMetadataUtils.java
+++ /dev/null
@@ -1,119 +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 org.apache.eagle.alert.siddhi;
-
-import java.util.Map;
-import java.util.SortedMap;
-
-import org.apache.eagle.alert.entity.AlertStreamSchemaEntity;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * convert metadata entities for a stream to stream definition for siddhi cep engine
- * define stream HeapUsage (metric string, host string, value double, timestamp long)
- */
-public class SiddhiStreamMetadataUtils {
-	private final static Logger LOG = LoggerFactory.getLogger(SiddhiStreamMetadataUtils.class);
-	
-	public final static String EAGLE_ALERT_CONTEXT_FIELD = "eagleAlertContext";
-
-	public static SortedMap<String, AlertStreamSchemaEntity> getAttrMap(String streamName) {
-		SortedMap<String, AlertStreamSchemaEntity> map = StreamMetadataManager.getInstance().getMetadataEntityMapForStream(streamName);
-		if(map == null || map.size() == 0){
-			throw new IllegalStateException("alert stream schema should never be empty");
-		}
-		return map;
-	}
-
-	/**
-     * @see org.wso2.siddhi.query.api.definition.Attribute.Type
-     * make sure StreamMetadataManager.init is invoked before this method
-	 * @param streamName
-	 * @return
-	 */
-	public static String convertToStreamDef(String streamName){
-		SortedMap<String, AlertStreamSchemaEntity> map = getAttrMap(streamName);
-		StringBuilder sb = new StringBuilder();		
-		sb.append(EAGLE_ALERT_CONTEXT_FIELD + " object,");
-		for(Map.Entry<String, AlertStreamSchemaEntity> entry : map.entrySet()){
-            appendAttributeNameType(sb, entry.getKey(), entry.getValue().getAttrType());
-		}
-		if(sb.length() > 0){
-			sb.deleteCharAt(sb.length()-1);
-		}
-		
-		String siddhiStreamDefFormat = "define stream " + streamName + "(" + "%s" + ");";
-		return String.format(siddhiStreamDefFormat, sb.toString());
-	}
-
-    public static String convertToStreamDef(String streamName, Map<String, String> eventSchema){
-        StringBuilder sb = new StringBuilder();
-        sb.append("context" + " object,");
-        for(Map.Entry<String, String> entry : eventSchema.entrySet()){
-            appendAttributeNameType(sb, entry.getKey(), entry.getValue());
-        }
-        if(sb.length() > 0){
-            sb.deleteCharAt(sb.length()-1);
-        }
-
-        String siddhiStreamDefFormat = "define stream " + streamName + "(" + "%s" + ");";
-        return String.format(siddhiStreamDefFormat, sb.toString());
-    }
-
-    private static void appendAttributeNameType(StringBuilder sb, String attrName, String attrType){
-        sb.append(attrName);
-        sb.append(" ");
-        if(attrType.equalsIgnoreCase(AttributeType.STRING.name())){
-            sb.append("string");
-        }else if(attrType.equalsIgnoreCase(AttributeType.INTEGER.name())){
-            sb.append("int");
-        }else if(attrType.equalsIgnoreCase(AttributeType.LONG.name())){
-            sb.append("long");
-        }else if(attrType.equalsIgnoreCase(AttributeType.BOOL.name())){
-            sb.append("bool");
-        }else if(attrType.equalsIgnoreCase(AttributeType.FLOAT.name())){
-            sb.append("float");
-        }else if(attrType.equalsIgnoreCase(AttributeType.DOUBLE.name())){
-            sb.append("double");
-        }else{
-            LOG.warn("AttrType is not recognized, ignore : " + attrType);
-        }
-        sb.append(",");
-    }
-
-	public static Object getAttrDefaultValue(String streamName, String attrName){
-		SortedMap<String, AlertStreamSchemaEntity> map = getAttrMap(streamName);
-		AlertStreamSchemaEntity entity = map.get(attrName);
-		if (entity.getDefaultValue() != null) {
-			return entity.getDefaultValue();
-		}
-		else {
-			String attrType = entity.getAttrType();
-			if (attrType.equalsIgnoreCase(AttributeType.STRING.name())) {
-				return "NA";
-			} else if (attrType.equalsIgnoreCase(AttributeType.INTEGER.name()) || attrType.equalsIgnoreCase(AttributeType.LONG.name())) {
-				return -1;
-			} else if (attrType.equalsIgnoreCase(AttributeType.BOOL.name())) {
-				return true;
-			} else {
-				LOG.warn("AttrType is not recognized: " + attrType + ", treat it as string");
-				return "N/A";
-			}
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/StreamMetadataManager.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/StreamMetadataManager.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/StreamMetadataManager.java
deleted file mode 100644
index 618d245..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/StreamMetadataManager.java
+++ /dev/null
@@ -1,128 +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 org.apache.eagle.alert.siddhi;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.SortedMap;
-import java.util.TreeMap;
-
-import org.apache.eagle.alert.common.AlertConstants;
-import org.apache.eagle.alert.entity.AlertStreamSchemaEntity;
-import com.typesafe.config.Config;
-import org.apache.eagle.alert.dao.AlertStreamSchemaDAO;
-import org.apache.eagle.common.config.EagleConfigConstants;
-import org.apache.commons.collections.map.UnmodifiableMap;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * centralized memory where all stream metadata sit on, it is not mutable data
- */
-public class StreamMetadataManager {
-	private static final Logger LOG = LoggerFactory.getLogger(StreamMetadataManager.class);
-	
-	private static StreamMetadataManager instance = new StreamMetadataManager();
-	private Map<String, List<AlertStreamSchemaEntity>> map = new HashMap<String, List<AlertStreamSchemaEntity>>();
-	private Map<String, SortedMap<String, AlertStreamSchemaEntity>> map2 = new HashMap<String, SortedMap<String, AlertStreamSchemaEntity>>();
-	private volatile boolean initialized = false;
-	
-	private StreamMetadataManager(){
-	}
-	
-	public static StreamMetadataManager getInstance(){
-		return instance;
-	}
-	
-	private void internalInit(Config config, AlertStreamSchemaDAO dao){
-		try{
-			String dataSource = config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.DATA_SOURCE);
-			List<AlertStreamSchemaEntity> list = dao.findAlertStreamSchemaByDataSource(dataSource);
-			if(list == null)
-				return;
-			for (AlertStreamSchemaEntity entity : list) {
-				String streamName = entity.getTags().get(AlertConstants.STREAM_NAME);
-				if (map.get(streamName) == null) {
-					map.put(streamName, new ArrayList<AlertStreamSchemaEntity>());
-					map2.put(streamName, new TreeMap<String, AlertStreamSchemaEntity>());
-				}
-				map.get(streamName).add(entity);
-				map2.get(streamName).put(entity.getTags().get(AlertConstants.ATTR_NAME), entity);
-			}
-		}catch(Exception ex){
-			LOG.error("Fail building metadata manger", ex);
-			throw new IllegalStateException(ex);
-		}
-	}
-	
-	/**
-	 * 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(Config config, AlertStreamSchemaDAO dao){
-		if(!initialized){
-			synchronized(this){
-				if(!initialized){
-                    if(LOG.isDebugEnabled()) LOG.debug("Initializing ...");
-					internalInit(config, dao);
-					initialized = true;
-                    LOG.info("Successfully initialized");
-				}
-			}
-		}else{
-            LOG.info("Already initialized, skip");
-        }
-	}
-
-	// Only for unit test purpose
-	public void reset() {
-		synchronized (this) {
-			initialized = false;
-			map.clear();
-			map2.clear();
-		}
-	}
-
-	private void ensureInitialized(){
-		if(!initialized)
-			throw new IllegalStateException("StreamMetadataManager should be initialized before using it");
-	}
-	
-	public List<AlertStreamSchemaEntity> getMetadataEntitiesForStream(String streamName){
-		ensureInitialized();
-		return getMetadataEntitiesForAllStreams().get(streamName);
-	}
-	
-	public Map<String, List<AlertStreamSchemaEntity>> getMetadataEntitiesForAllStreams(){
-		ensureInitialized();
-		return UnmodifiableMap.decorate(map);
-	}
-	
-	public SortedMap<String, AlertStreamSchemaEntity> getMetadataEntityMapForStream(String streamName){
-		ensureInitialized();
-		return getMetadataEntityMapForAllStreams().get(streamName);
-	}
-	
-	public Map<String, SortedMap<String, AlertStreamSchemaEntity>> getMetadataEntityMapForAllStreams(){
-		ensureInitialized();
-		return UnmodifiableMap.decorate(map2);
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/ContainsIgnoreCaseExtension.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/ContainsIgnoreCaseExtension.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/ContainsIgnoreCaseExtension.java
deleted file mode 100644
index cf76134..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/ContainsIgnoreCaseExtension.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
- *
- * Licensed 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.siddhi.extension;
-
-import org.wso2.siddhi.core.config.ExecutionPlanContext;
-import org.wso2.siddhi.core.exception.ExecutionPlanRuntimeException;
-import org.wso2.siddhi.core.executor.ExpressionExecutor;
-import org.wso2.siddhi.core.executor.function.FunctionExecutor;
-import org.wso2.siddhi.query.api.definition.Attribute;
-import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
-
-public class ContainsIgnoreCaseExtension extends FunctionExecutor {
-
-    Attribute.Type returnType = Attribute.Type.BOOL;
-
-    @Override
-    protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
-        if (attributeExpressionExecutors.length != 2) {
-            throw new ExecutionPlanValidationException("Invalid no of arguments passed to str:containsIgnoreCase() function, required 2, " +
-                    "but found " + attributeExpressionExecutors.length);
-        }
-        if (attributeExpressionExecutors[0].getReturnType() != Attribute.Type.STRING) {
-            throw new ExecutionPlanValidationException("Invalid parameter type found for the first argument of str:containsIgnoreCase() function, " +
-                    "required "+ Attribute.Type.STRING+", but found "+attributeExpressionExecutors[0].getReturnType().toString());
-        }
-        if (attributeExpressionExecutors[1].getReturnType() != Attribute.Type.STRING) {
-            throw new ExecutionPlanValidationException("Invalid parameter type found for the second argument of str:containsIgnoreCase() function, " +
-                    "required "+ Attribute.Type.STRING+", but found "+attributeExpressionExecutors[1].getReturnType().toString());
-        }
-    }
-
-    @Override
-    protected Object execute(Object[] data) {
-        if (data[0] == null) {
-            throw new ExecutionPlanRuntimeException("Invalid input given to str:containsIgnoreCase() function. First argument cannot be null");
-        }
-        if (data[1] == null) {
-            throw new ExecutionPlanRuntimeException("Invalid input given to str:containsIgnoreCase() function. Second argument cannot be null");
-        }
-        String str1 = (String)data[0];
-        String str2 = (String)data[1];
-        return str1.toUpperCase().contains(str2.toUpperCase());
-    }
-
-    @Override
-    protected Object execute(Object data) {
-        return null; //Since the containsIgnoreCase function takes in 2 parameters, this method does not get called. Hence, not implemented.
-    }
-
-    @Override
-    public void start() {
-        //Nothing to start
-    }
-
-    @Override
-    public void stop() {
-        //Nothing to stop
-    }
-
-    @Override
-    public Attribute.Type getReturnType() {
-        return returnType;
-    }
-
-    @Override
-    public Object[] currentState() {
-        return new Object[]{};
-    }
-
-    @Override
-    public void restoreState(Object[] state) {
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/EqualsIgnoreCaseExtension.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/EqualsIgnoreCaseExtension.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/EqualsIgnoreCaseExtension.java
deleted file mode 100644
index 0b6e7ec..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/EqualsIgnoreCaseExtension.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
- *
- * Licensed 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.siddhi.extension;
-
-import org.wso2.siddhi.core.config.ExecutionPlanContext;
-import org.wso2.siddhi.core.exception.ExecutionPlanRuntimeException;
-import org.wso2.siddhi.core.executor.ConstantExpressionExecutor;
-import org.wso2.siddhi.core.executor.ExpressionExecutor;
-import org.wso2.siddhi.core.executor.function.FunctionExecutor;
-import org.wso2.siddhi.query.api.definition.Attribute;
-import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
-
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-public class EqualsIgnoreCaseExtension extends FunctionExecutor {
-
-    Attribute.Type returnType = Attribute.Type.BOOL;
-
-    @Override
-    protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
-        if (attributeExpressionExecutors.length != 2) {
-            throw new ExecutionPlanValidationException("Invalid no of arguments passed to str:equalsIgnoreCase() function, required 2, " +
-                    "but found " + attributeExpressionExecutors.length);
-        }
-        if (attributeExpressionExecutors[0].getReturnType() != Attribute.Type.STRING) {
-            throw new ExecutionPlanValidationException("Invalid parameter type found for the first argument of str:equalsIgnoreCase() function, " +
-                    "required "+ Attribute.Type.STRING+", but found "+attributeExpressionExecutors[0].getReturnType().toString());
-        }
-        if (attributeExpressionExecutors[1].getReturnType() != Attribute.Type.STRING) {
-            throw new ExecutionPlanValidationException("Invalid parameter type found for the second argument of str:equalsIgnoreCase() function, " +
-                    "required "+ Attribute.Type.STRING+", but found "+attributeExpressionExecutors[1].getReturnType().toString());
-        }
-    }
-
-    @Override
-    protected Object execute(Object[] data) {
-        if (data[0] == null) {
-            throw new ExecutionPlanRuntimeException("Invalid input given to str:equalsIgnoreCase() function. First argument cannot be null");
-        }
-        if (data[1] == null) {
-            throw new ExecutionPlanRuntimeException("Invalid input given to str:equalsIgnoreCase() function. Second argument cannot be null");
-        }
-        String str1 = (String)data[0];
-        String str2 = (String)data[1];
-        return str1.equalsIgnoreCase(str2);
-    }
-
-    @Override
-    protected Object execute(Object data) {
-        return null; //Since the equalsIgnoreCase function takes in 2 parameters, this method does not get called. Hence, not implemented.
-    }
-
-    @Override
-    public void start() {
-        //Nothing to start
-    }
-
-    @Override
-    public void stop() {
-        //Nothing to stop
-    }
-
-    @Override
-    public Attribute.Type getReturnType() {
-        return returnType;
-    }
-
-    @Override
-    public Object[] currentState() {
-        return new Object[]{};
-    }
-
-    @Override
-    public void restoreState(Object[] state) {
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/RegexpIgnoreCaseFunctionExtension.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/RegexpIgnoreCaseFunctionExtension.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/RegexpIgnoreCaseFunctionExtension.java
deleted file mode 100644
index 0bf80de..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/alert/siddhi/extension/RegexpIgnoreCaseFunctionExtension.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
- *
- * Licensed 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.siddhi.extension;
-
-import org.wso2.siddhi.core.config.ExecutionPlanContext;
-import org.wso2.siddhi.core.exception.ExecutionPlanRuntimeException;
-import org.wso2.siddhi.core.executor.ConstantExpressionExecutor;
-import org.wso2.siddhi.core.executor.ExpressionExecutor;
-import org.wso2.siddhi.extension.string.RegexpFunctionExtension;
-import org.wso2.siddhi.query.api.definition.Attribute;
-import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
-
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-/**
- * regexpIgnoreCase(string, regex)
- * Tells whether or not this 'string' matches the given regular expression 'regex'.
- * Accept Type(s): (STRING,STRING)
- * Return Type(s): BOOLEAN
- */
-public class RegexpIgnoreCaseFunctionExtension extends RegexpFunctionExtension {
-
-    //state-variables
-    boolean isRegexConstant = false;
-    String regexConstant;
-    Pattern patternConstant;
-
-    @Override
-    protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
-        if (attributeExpressionExecutors.length != 2) {
-            throw new ExecutionPlanValidationException("Invalid no of arguments passed to str:regexpIgnoreCase() function, required 2, " +
-                    "but found " + attributeExpressionExecutors.length);
-        }
-        if (attributeExpressionExecutors[0].getReturnType() != Attribute.Type.STRING) {
-            throw new ExecutionPlanValidationException("Invalid parameter type found for the first argument of str:regexpIgnoreCase() function, " +
-                    "required "+ Attribute.Type.STRING+", but found "+attributeExpressionExecutors[0].getReturnType().toString());
-        }
-        if (attributeExpressionExecutors[1].getReturnType() != Attribute.Type.STRING) {
-            throw new ExecutionPlanValidationException("Invalid parameter type found for the second argument of str:regexpIgnoreCase() function, " +
-                    "required "+ Attribute.Type.STRING+", but found "+attributeExpressionExecutors[1].getReturnType().toString());
-        }
-        if(attributeExpressionExecutors[1] instanceof ConstantExpressionExecutor){
-            isRegexConstant = true;
-            regexConstant = (String) ((ConstantExpressionExecutor) attributeExpressionExecutors[1]).getValue();
-            patternConstant = Pattern.compile(regexConstant, Pattern.CASE_INSENSITIVE);
-        }
-    }
-
-    @Override
-    protected Object execute(Object[] data) {
-        String regex;
-        Pattern pattern;
-        Matcher matcher;
-
-        if (data[0] == null) {
-            throw new ExecutionPlanRuntimeException("Invalid input given to str:regexpIgnoreCase() function. First argument cannot be null");
-        }
-        if (data[1] == null) {
-            throw new ExecutionPlanRuntimeException("Invalid input given to str:regexpIgnoreCase() function. Second argument cannot be null");
-        }
-        String source = (String) data[0];
-
-        if(!isRegexConstant){
-            regex = (String) data[1];
-            pattern = Pattern.compile(regex, Pattern.CASE_INSENSITIVE);
-            matcher = pattern.matcher(source);
-            return matcher.matches();
-
-        } else {
-            matcher = patternConstant.matcher(source);
-            return matcher.matches();
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutor.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutor.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutor.java
deleted file mode 100644
index 44014da..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutor.java
+++ /dev/null
@@ -1,395 +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 org.apache.eagle.executor;
-
-import com.codahale.metrics.MetricRegistry;
-import com.sun.jersey.client.impl.CopyOnWriteHashMap;
-import com.typesafe.config.Config;
-import org.apache.commons.lang3.time.DateUtils;
-import org.apache.eagle.alert.common.AlertConstants;
-import org.apache.eagle.alert.config.AbstractPolicyDefinition;
-import org.apache.eagle.alert.dao.AlertDefinitionDAO;
-import org.apache.eagle.alert.dao.AlertStreamSchemaDAO;
-import org.apache.eagle.alert.dao.AlertStreamSchemaDAOImpl;
-import org.apache.eagle.alert.entity.AlertAPIEntity;
-import org.apache.eagle.alert.entity.AlertDefinitionAPIEntity;
-import org.apache.eagle.alert.policy.*;
-import org.apache.eagle.alert.siddhi.EagleAlertContext;
-import org.apache.eagle.alert.siddhi.SiddhiAlertHandler;
-import org.apache.eagle.alert.siddhi.StreamMetadataManager;
-import org.apache.eagle.common.config.EagleConfigConstants;
-import org.apache.eagle.dataproc.core.JsonSerDeserUtils;
-import org.apache.eagle.dataproc.core.ValuesArray;
-import org.apache.eagle.datastream.Collector;
-import org.apache.eagle.datastream.JavaStormStreamExecutor2;
-import org.apache.eagle.datastream.Tuple2;
-import org.apache.eagle.metric.reportor.EagleCounterMetric;
-import org.apache.eagle.metric.reportor.EagleMetricListener;
-import org.apache.eagle.metric.reportor.EagleServiceReporterMetricListener;
-import org.apache.eagle.metric.reportor.MetricKeyCodeDecoder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.lang.management.ManagementFactory;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-public class AlertExecutor extends JavaStormStreamExecutor2<String, AlertAPIEntity> implements PolicyLifecycleMethods, SiddhiAlertHandler, PolicyDistributionReportMethods {
-	private static final long serialVersionUID = 1L;
-
-	private static final Logger LOG = LoggerFactory.getLogger(AlertExecutor.class);
-
-	private String alertExecutorId;
-	private volatile CopyOnWriteHashMap<String, PolicyEvaluator> policyEvaluators;
-	private PolicyPartitioner partitioner;
-	private int numPartitions;
-	private int partitionSeq;
-	private Config config;
-	private Map<String, Map<String, AlertDefinitionAPIEntity>> initialAlertDefs;
-	private AlertDefinitionDAO alertDefinitionDao;
-	private String[] sourceStreams;
-	private static String EAGLE_EVENT_COUNT = "eagle.event.count";
-	private static String EAGLE_POLICY_EVAL_COUNT = "eagle.policy.eval.count";
-	private static String EAGLE_POLICY_EVAL_FAIL_COUNT = "eagle.policy.eval.fail.count";
-	private static String EAGLE_ALERT_COUNT = "eagle.alert.count";
-	private static String EAGLE_ALERT_FAIL_COUNT = "eagle.alert.fail.count";
-	private	static long MERITE_GRANULARITY = DateUtils.MILLIS_PER_MINUTE;
-	private Map<String, Map<String, String>> dimensionsMap; // cache it for performance
-	private Map<String, String> baseDimensions;
-	private MetricRegistry registry;
-	private EagleMetricListener listener;
-
-	public AlertExecutor(String alertExecutorId, PolicyPartitioner partitioner, int numPartitions, int partitionSeq,
-                         AlertDefinitionDAO alertDefinitionDao, String[] sourceStreams){
-		this.alertExecutorId = alertExecutorId;
-		this.partitioner = partitioner;
-		this.numPartitions = numPartitions;
-		this.partitionSeq = partitionSeq;
-		this.alertDefinitionDao = alertDefinitionDao;
-		this.sourceStreams = sourceStreams;
-	}
-	
-	public String getAlertExecutorId(){
-		return this.alertExecutorId;
-	}
-
-	public int getNumPartitions() {
-		return this.numPartitions;
-	}
-	
-	public int getPartitionSeq(){
-		return this.partitionSeq;
-	}
-	
-	public PolicyPartitioner getPolicyPartitioner() {
-		return this.partitioner;
-	}
-	
-	public Map<String, Map<String, AlertDefinitionAPIEntity>> getInitialAlertDefs() {
-		return this.initialAlertDefs;
-	}
-		
-	public AlertDefinitionDAO getAlertDefinitionDao() {
-		return alertDefinitionDao;
-	}
-
-    public Map<String, PolicyEvaluator> getPolicyEvaluators(){
-        return policyEvaluators;
-    }
-	
-	@Override
-	public void prepareConfig(Config config) {
-		this.config = config;
-	}
-	
-	private void initMetricReportor() {
-		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;
-
-		//TODO: need to it replace it with batch flush listener
-		registry = new MetricRegistry();
-		listener = new EagleServiceReporterMetricListener(host, port, username, password);
-
-		baseDimensions = new HashMap<>();
-		baseDimensions.put(AlertConstants.ALERT_EXECUTOR_ID, alertExecutorId);
-		baseDimensions.put(AlertConstants.PARTITIONSEQ, String.valueOf(partitionSeq));
-		baseDimensions.put(AlertConstants.SOURCE, ManagementFactory.getRuntimeMXBean().getName());
-		baseDimensions.put(EagleConfigConstants.DATA_SOURCE, config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.DATA_SOURCE));
-		baseDimensions.put(EagleConfigConstants.SITE, config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.SITE));
-		dimensionsMap = new HashMap<>();
-	}
-
-    /**
-     * for unit test purpose only
-     * @param config
-     * @return
-     */
-    public AlertStreamSchemaDAO getAlertStreamSchemaDAO(Config config){
-        return new AlertStreamSchemaDAOImpl(config);
-    }
-
-	@Override
-	public void init() {
-		// initialize StreamMetadataManager before it is used
-		StreamMetadataManager.getInstance().init(config, getAlertStreamSchemaDAO(config));
-		// for each AlertDefinition, to create a PolicyEvaluator
-		Map<String, PolicyEvaluator> tmpPolicyEvaluators = new HashMap<String, PolicyEvaluator>();
-		
-        String site = config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.SITE);
-		String dataSource = config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.DATA_SOURCE);
-		try {
-			initialAlertDefs = alertDefinitionDao.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 was found for site: " + site + ", dataSource: " + dataSource);
-        }
-        else if (initialAlertDefs.get(alertExecutorId) != null) { 
-			for(AlertDefinitionAPIEntity alertDef : initialAlertDefs.get(alertExecutorId).values()){
-				int part = partitioner.partition(numPartitions, alertDef.getTags().get(AlertConstants.POLICY_TYPE), alertDef.getTags().get(AlertConstants.POLICY_ID));
-				if (part == partitionSeq) {
-					tmpPolicyEvaluators.put(alertDef.getTags().get(AlertConstants.POLICY_ID), createPolicyEvaluator(alertDef));
-				}
-			}
-		}
-		
-		policyEvaluators = new CopyOnWriteHashMap<>();
-		// for efficiency, we don't put single policy evaluator
-		policyEvaluators.putAll(tmpPolicyEvaluators);
-		DynamicPolicyLoader policyLoader = DynamicPolicyLoader.getInstance();
-		
-		policyLoader.init(initialAlertDefs, alertDefinitionDao, config);
-        String fullQualifiedAlertExecutorId = alertExecutorId + "_" + partitionSeq;
-		policyLoader.addPolicyChangeListener(fullQualifiedAlertExecutorId, this);
-        policyLoader.addPolicyDistributionReporter(fullQualifiedAlertExecutorId, this);
-		LOG.info("Alert Executor created, partitionSeq: " + partitionSeq + " , numPartitions: " + numPartitions);
-        LOG.info("All policy evaluators: " + policyEvaluators);
-		
-		initMetricReportor();
-	}
-
-    /**
-     * Create PolicyEvaluator instance according to policyType-mapped policy evaluator class
-     *
-     * @param alertDef alert definition
-     * @return PolicyEvaluator instance
-     */
-	private PolicyEvaluator createPolicyEvaluator(AlertDefinitionAPIEntity alertDef){
-		String policyType = alertDef.getTags().get(AlertConstants.POLICY_TYPE);
-		Class<? extends PolicyEvaluator> evalCls = PolicyManager.getInstance().getPolicyEvaluator(policyType);
-		if(evalCls == null){
-			String msg = "No policy evaluator defined for policy type : " + policyType;
-			LOG.error(msg);
-			throw new IllegalStateException(msg);
-		}
-		
-		// check out whether strong incoming data validation is necessary
-        String needValidationConfigKey= AlertConstants.ALERT_EXECUTOR_CONFIGS + "." + alertExecutorId + ".needValidation";
-
-        // Default: true
-        boolean needValidation = !config.hasPath(needValidationConfigKey) || config.getBoolean(needValidationConfigKey);
-
-		AbstractPolicyDefinition policyDef = null;
-		try {
-			policyDef = JsonSerDeserUtils.deserialize(alertDef.getPolicyDef(), AbstractPolicyDefinition.class, PolicyManager.getInstance().getPolicyModules(policyType));
-		} catch (Exception ex) {
-			LOG.error("Fail initial alert policy def: "+alertDef.getPolicyDef(), ex);
-		}
-		PolicyEvaluator pe;
-		try{
-            // Create evaluator instances
-			pe = evalCls.getConstructor(Config.class, String.class, AbstractPolicyDefinition.class, String[].class, boolean.class).newInstance(config, alertDef.getTags().get("policyId"), policyDef, sourceStreams, needValidation);
-		}catch(Exception ex){
-			LOG.error("Fail creating new policyEvaluator", ex);
-			LOG.warn("Broken policy definition and stop running : " + alertDef.getPolicyDef());
-			throw new IllegalStateException(ex);
-		}
-		return pe;
-	}
-
-    /**
-     * verify both alertExecutor logic name and partition id
-     * @param alertDef alert definition
-     *
-     * @return whether accept the alert definition
-     */
-	private boolean accept(AlertDefinitionAPIEntity alertDef){
-        if(!alertDef.getTags().get("alertExecutorId").equals(alertExecutorId)) {
-            if(LOG.isDebugEnabled()){
-                LOG.debug("alertDef does not belong to this alertExecutorId : " + alertExecutorId + ", alertDef : " + alertDef);
-            }
-            return false;
-        }
-		int targetPartitionSeq = partitioner.partition(numPartitions, alertDef.getTags().get(AlertConstants.POLICY_TYPE), alertDef.getTags().get(AlertConstants.POLICY_ID));
-		if(targetPartitionSeq == partitionSeq)
-			return true;
-		return false;
-	}
-	
-	private long trim(long value, long granularity) {
-		return value / granularity * granularity;
-	}
-
-	private void updateCounter(String name, Map<String, String> dimensions, double value) {
-		long current = System.currentTimeMillis();
-		String metricName = MetricKeyCodeDecoder.codeMetricKey(name, dimensions);
-		if (registry.getMetrics().get(metricName) == null) {
-			EagleCounterMetric metric = new EagleCounterMetric(current, metricName, value, MERITE_GRANULARITY);
-			metric.registerListener(listener);
-			registry.register(metricName, metric);
-		} else {
-			EagleCounterMetric metric = (EagleCounterMetric) registry.getMetrics().get(metricName);
-			metric.update(value, current);
-			//TODO: need remove unused metric from registry
-		}
-	}
-	
-	private void updateCounter(String name, Map<String, String> dimensions) {
-		updateCounter(name, dimensions, 1.0);
-	}
-	
-	protected Map<String, String> getDimensions(String policyId) {
-		if (dimensionsMap.get(policyId) == null) {
-			Map<String, String> newDimensions = new HashMap<String, String>(baseDimensions);
-			newDimensions.put(AlertConstants.POLICY_ID, policyId);
-			dimensionsMap.put(policyId, newDimensions);
-		}
-		return dimensionsMap.get(policyId);
-	}
-
-    /**
-     * within this single executor, execute all PolicyEvaluator sequentially
-     * the contract for input:
-     * 1. total # of fields for input is 3, which is fixed
-     * 2. the first field is key
-     * 3. the second field is stream name
-     * 4. the third field is value which is java SortedMap
-     */
-    @Override
-    public void flatMap(java.util.List<Object> input, Collector<Tuple2<String, AlertAPIEntity>> outputCollector){
-        if(input.size() != 3)
-            throw new IllegalStateException("AlertExecutor always consumes exactly 3 fields: key, stream name and value(SortedMap)");
-        if(LOG.isDebugEnabled()) LOG.debug("Msg is coming " + input.get(2));
-        if(LOG.isDebugEnabled()) LOG.debug("Current policyEvaluators: " + policyEvaluators.keySet().toString());
-
-        updateCounter(EAGLE_EVENT_COUNT, baseDimensions);
-        try{
-            synchronized(this.policyEvaluators) {
-                for(Entry<String, PolicyEvaluator> entry : policyEvaluators.entrySet()){
-                    String policyId = entry.getKey();
-                    PolicyEvaluator evaluator = entry.getValue();
-                    updateCounter(EAGLE_POLICY_EVAL_COUNT, getDimensions(policyId));
-                    try {
-                        EagleAlertContext siddhiAlertContext = new EagleAlertContext();
-                        siddhiAlertContext.alertExecutor = this;
-                        siddhiAlertContext.policyId = policyId;
-                        siddhiAlertContext.evaluator = evaluator;
-                        siddhiAlertContext.outputCollector = outputCollector;
-                        evaluator.evaluate(new ValuesArray(siddhiAlertContext, input.get(1), input.get(2)));
-                    }
-                    catch (Exception ex) {
-                        LOG.error("Got an exception, but continue to run " + input.get(2).toString(), ex);
-                        updateCounter(EAGLE_POLICY_EVAL_FAIL_COUNT, getDimensions(policyId));
-                    }
-                }
-            }
-        } catch(Exception ex){
-            LOG.error(alertExecutorId + ", partition " + partitionSeq + ", error fetching alerts, but continue to run", ex);
-            updateCounter(EAGLE_ALERT_FAIL_COUNT, baseDimensions);
-        }
-    }
-
-	@Override
-	public void onPolicyCreated(Map<String, AlertDefinitionAPIEntity> added) {
-		if(LOG.isDebugEnabled()) LOG.debug(alertExecutorId + ", partition " + partitionSeq + " policy added : " + added + " policyEvaluators " + policyEvaluators);
-		for(AlertDefinitionAPIEntity alertDef : added.values()){
-			if(!accept(alertDef))
-				continue;
-			LOG.info(alertExecutorId + ", partition " + partitionSeq + " policy really added " + alertDef);
-			PolicyEvaluator newEvaluator = createPolicyEvaluator(alertDef);
-			if(newEvaluator != null){
-				synchronized(this.policyEvaluators) {
-					policyEvaluators.put(alertDef.getTags().get(AlertConstants.POLICY_ID), newEvaluator);
-				}
-			}
-		}
-	}
-
-	@Override
-	public void onPolicyChanged(Map<String, AlertDefinitionAPIEntity> changed) {
-		if(LOG.isDebugEnabled()) LOG.debug(alertExecutorId + ", partition " + partitionSeq + " policy changed : " + changed);
-		for(AlertDefinitionAPIEntity alertDef : changed.values()){
-			if(!accept(alertDef))
-				continue;
-			LOG.info(alertExecutorId + ", partition " + partitionSeq + " policy really changed " + alertDef);
-			synchronized(this.policyEvaluators) {
-				PolicyEvaluator pe = policyEvaluators.get(alertDef.getTags().get(AlertConstants.POLICY_ID));
-				pe.onPolicyUpdate(alertDef);
-			}
-		}
-	}
-
-	@Override
-	public void onPolicyDeleted(Map<String, AlertDefinitionAPIEntity> deleted) {
-		if(LOG.isDebugEnabled()) LOG.debug(alertExecutorId + ", partition " + partitionSeq + " policy deleted : " + deleted);
-		for(AlertDefinitionAPIEntity alertDef : deleted.values()){
-			if(!accept(alertDef))
-				continue;
-			LOG.info(alertExecutorId + ", partition " + partitionSeq + " policy really deleted " + alertDef);
-			String policyId = alertDef.getTags().get(AlertConstants.POLICY_ID);
-			synchronized(this.policyEvaluators) {			
-				if (policyEvaluators.containsKey(policyId)) {
-					PolicyEvaluator pe = policyEvaluators.remove(alertDef.getTags().get(AlertConstants.POLICY_ID));
-					pe.onPolicyDelete();
-				}
-			}
-		}
-	}
-
-	@Override
-	public void onAlerts(EagleAlertContext context, List<AlertAPIEntity> alerts) {
-		if(alerts != null && !alerts.isEmpty()){
-			String policyId = context.policyId;
-            LOG.info(String.format("Detected %s alerts for policy %s",alerts.size(),policyId));
-			Collector outputCollector = context.outputCollector;
-			PolicyEvaluator evaluator = context.evaluator;
-			updateCounter(EAGLE_ALERT_COUNT, getDimensions(policyId), alerts.size());
-			for (AlertAPIEntity entity : alerts) {
-				synchronized(this) {
-					outputCollector.collect(new Tuple2(policyId, entity));
-				}
-				if(LOG.isDebugEnabled()) LOG.debug("A new alert is triggered: "+alertExecutorId + ", partition " + partitionSeq + ", Got an alert with output context: " + entity.getAlertContext() + ", for policy " + evaluator);
-			}
-		}
-	}
-
-    @Override
-    public void report() {
-        PolicyDistroStatsLogReporter appender = new PolicyDistroStatsLogReporter();
-        appender.reportPolicyMembership(alertExecutorId + "_" + partitionSeq, policyEvaluators.keySet());
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutorCreationUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutorCreationUtils.java b/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutorCreationUtils.java
deleted file mode 100644
index 547e39d..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/java/org/apache/eagle/executor/AlertExecutorCreationUtils.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 org.apache.eagle.executor;
-
-import org.apache.eagle.alert.common.AlertConstants;
-import org.apache.eagle.alert.dao.AlertDefinitionDAO;
-import org.apache.eagle.alert.dao.AlertDefinitionDAOImpl;
-import org.apache.eagle.alert.dao.AlertExecutorDAOImpl;
-import org.apache.eagle.alert.entity.AlertExecutorEntity;
-import org.apache.eagle.alert.policy.DefaultPolicyPartitioner;
-import org.apache.eagle.alert.policy.PolicyPartitioner;
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigValue;
-import org.apache.eagle.common.config.EagleConfigConstants;
-import org.apache.eagle.service.client.EagleServiceConnector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Create alert executors and provide callback for programmer to link alert executor to immediate parent executors
- *
- * <br/><br/>
- * Explanations for programId, alertExecutorId and policy<br/><br/>
- * - programId - distributed or single-process program for example one storm topology<br/>
- * - alertExecutorId - one process/thread which executes multiple policies<br/>
- * - policy - some rules to be evaluated<br/>
- *
- * <br/>
- *
- * Normally the mapping is like following:
- * <pre>
- * programId (1:N) alertExecutorId
- * alertExecutorId (1:N) policy
- * </pre>
- */
-public class AlertExecutorCreationUtils {
-	private final static Logger LOG = LoggerFactory.getLogger(AlertExecutorCreationUtils.class);
-
-    public static AlertExecutor[] createAlertExecutors(Config config, String alertExecutorId) throws Exception{
-        // Read site and dataSource from configuration.
-        String dataSource = config.getString(EagleConfigConstants.EAGLE_PROPS + "." + EagleConfigConstants.DATA_SOURCE);
-        LOG.info("Loading alerting definitions for dataSource: " + dataSource);
-
-        // Get map from alertExecutorId to alert stream
-        // (dataSource) => Map[alertExecutorId:String,streamName:List[String]]
-        List<String> streamNames = new ArrayList<String>();
-        AlertExecutorDAOImpl alertExecutorDAO = new AlertExecutorDAOImpl(new EagleServiceConnector(config));
-        List<AlertExecutorEntity> alertExecutorEntities = alertExecutorDAO.findAlertExecutor(dataSource, alertExecutorId);
-        for(AlertExecutorEntity entity : alertExecutorEntities){
-            streamNames.add(entity.getTags().get(AlertConstants.STREAM_NAME));
-        }
-
-        if(streamNames.isEmpty()){
-            throw new IllegalStateException("upstream names should not be empty for alert " + alertExecutorId);
-        }
-        return createAlertExecutors(config, new AlertDefinitionDAOImpl(new EagleServiceConnector(config)),
-                streamNames, alertExecutorId);
-    }
-
-    /**
-     * Build DAG Tasks based on persisted alert definition and schemas from eagle store.
-     *
-     * <h3>Require configuration:</h3>
-     *
-     * <ul>
-     * <li>eagleProps.site: program site id.</li>
-     * <li>eagleProps.dataSource: program data source.</li>
-     * <li>alertExecutorConfigs: only configured executor will be built into execution tasks.</li>
-     * </ul>
-     *
-     * <h3>Steps:</h3>
-     *
-     * <ol>
-     * <li>(upstreamTasks) => Map[streamName:String,upstreamTask:Task]</li>
-     * <li>(dataSource) => Map[alertExecutorId:String,streamName:List[String]]</li>
-     * <li>(site,dataSource) => Map[alertExecutorId,Map[policyId,alertDefinition]]</li>
-     * <li>(config["alertExecutorConfigs"]) => AlertExecutor(alertExecutorID, partitioner, numPartitions, partitionSeq, alertDefs, alertDefDAO, sourceStreams)[]</li>
-     * </ol>
-     */
-	public static AlertExecutor[] createAlertExecutors(Config config, AlertDefinitionDAO alertDefDAO,
-			List<String> streamNames, String alertExecutorId) throws Exception{
-		// Read `alertExecutorConfigs` from configuration and get config for this alertExecutorId
-        int numPartitions =1;
-        String partitionerCls = DefaultPolicyPartitioner.class.getCanonicalName();
-        String alertExecutorConfigsKey = "alertExecutorConfigs";
-        if(config.hasPath(alertExecutorConfigsKey)) {
-            Map<String, ConfigValue> alertExecutorConfigs = config.getObject(alertExecutorConfigsKey);
-            if(alertExecutorConfigs !=null && alertExecutorConfigs.containsKey(alertExecutorId)) {
-                Map<String, Object> alertExecutorConfig = (Map<String, Object>) alertExecutorConfigs.get(alertExecutorId).unwrapped();
-                int parts = 0;
-                if(alertExecutorConfig.containsKey("parallelism")) parts = (int) (alertExecutorConfig.get("parallelism"));
-                numPartitions = parts == 0 ? 1 : parts;
-                if(alertExecutorConfig.containsKey("partitioner")) partitionerCls = (String) alertExecutorConfig.get("partitioner");
-            }
-        }
-
-        return createAlertExecutors(alertDefDAO, streamNames, alertExecutorId, numPartitions, partitionerCls);
-	}
-
-    /**
-     * Build alert executors and assign alert definitions between these executors by partitioner (alertExecutorConfigs["${alertExecutorId}"]["partitioner"])
-     */
-	public static AlertExecutor[] createAlertExecutors(AlertDefinitionDAO alertDefDAO, List<String> sourceStreams,
-                                                          String alertExecutorID, int numPartitions, String partitionerCls) throws Exception{
-		LOG.info("Creating alert executors with alertExecutorID: " + alertExecutorID + ", numPartitions: " + numPartitions + ", Partition class is: "+ partitionerCls);
-
-		PolicyPartitioner partitioner = (PolicyPartitioner)Class.forName(partitionerCls).newInstance();
-		AlertExecutor[] alertExecutors = new AlertExecutor[numPartitions];
-        String[] _sourceStreams = sourceStreams.toArray(new String[0]);
-
-		for(int i = 0; i < numPartitions; i++){
-			alertExecutors[i] = new AlertExecutor(alertExecutorID, partitioner, numPartitions, i, alertDefDAO,_sourceStreams);
-		}	
-		return alertExecutors;
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.alert.policy.PolicyEvaluatorServiceProvider
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.alert.policy.PolicyEvaluatorServiceProvider b/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.alert.policy.PolicyEvaluatorServiceProvider
deleted file mode 100644
index dc4e0a1..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.alert.policy.PolicyEvaluatorServiceProvider
+++ /dev/null
@@ -1,16 +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.
-
-org.apache.eagle.alert.siddhi.SiddhiPolicyEvaluatorServiceProviderImpl
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.policy.PolicyEvaluatorServiceProvider
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.policy.PolicyEvaluatorServiceProvider b/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.policy.PolicyEvaluatorServiceProvider
new file mode 100644
index 0000000..eac2bfd
--- /dev/null
+++ b/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/META-INF/services/org.apache.eagle.policy.PolicyEvaluatorServiceProvider
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.eagle.policy.siddhi.SiddhiPolicyEvaluatorServiceProviderImpl
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/str.siddhiext
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/str.siddhiext b/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/str.siddhiext
deleted file mode 100644
index 4d5e237..0000000
--- a/eagle-core/eagle-alert/eagle-alert-process/src/main/resources/str.siddhiext
+++ /dev/null
@@ -1,35 +0,0 @@
-#
-# Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
-#
-# Licensed 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.
-#
-
-charAt=org.wso2.siddhi.extension.string.CharAtFunctionExtension
-coalesce=org.wso2.siddhi.extension.string.CoalesceFunctionExtension
-concat=org.wso2.siddhi.extension.string.ConcatFunctionExtension
-length=org.wso2.siddhi.extension.string.LengthFunctionExtension
-lower=org.wso2.siddhi.extension.string.LowerFunctionExtension
-regexp=org.wso2.siddhi.extension.string.RegexpFunctionExtension
-repeat=org.wso2.siddhi.extension.string.RepeatFunctionExtension
-replaceAll=org.wso2.siddhi.extension.string.ReplaceAllFunctionExtension
-replaceFirst=org.wso2.siddhi.extension.string.ReplaceFirstFunctionExtension
-reverse=org.wso2.siddhi.extension.string.ReverseFunctionExtension
-strcmp=org.wso2.siddhi.extension.string.StrcmpFunctionExtension
-substr=org.wso2.siddhi.extension.string.SubstrFunctionExtension
-trim=org.wso2.siddhi.extension.string.TrimFunctionExtension
-upper=org.wso2.siddhi.extension.string.UpperFunctionExtension
-hex=org.wso2.siddhi.extension.string.HexFunctionExtension
-unhex=org.wso2.siddhi.extension.string.UnhexFunctionExtension
-equalsIgnoreCase=org.apache.eagle.alert.siddhi.extension.EqualsIgnoreCaseExtension
-containsIgnoreCase=org.apache.eagle.alert.siddhi.extension.ContainsIgnoreCaseExtension
-regexpIgnoreCase=org.apache.eagle.alert.siddhi.extension.RegexpIgnoreCaseFunctionExtension

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/cep/TestSiddhiEvaluator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/cep/TestSiddhiEvaluator.java b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/cep/TestSiddhiEvaluator.java
index 521790a..3d69887 100644
--- a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/cep/TestSiddhiEvaluator.java
+++ b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/cep/TestSiddhiEvaluator.java
@@ -18,22 +18,23 @@ package org.apache.eagle.alert.cep;
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
-import org.apache.eagle.alert.dao.AlertDefinitionDAO;
-import org.apache.eagle.alert.dao.AlertDefinitionDAOImpl;
-import org.apache.eagle.alert.dao.AlertStreamSchemaDAO;
-import org.apache.eagle.alert.dao.AlertStreamSchemaDAOImpl;
+import junit.framework.Assert;
 import org.apache.eagle.alert.entity.AlertAPIEntity;
 import org.apache.eagle.alert.entity.AlertDefinitionAPIEntity;
 import org.apache.eagle.alert.entity.AlertStreamSchemaEntity;
-import org.apache.eagle.alert.siddhi.EagleAlertContext;
-import org.apache.eagle.alert.siddhi.SiddhiPolicyDefinition;
-import org.apache.eagle.alert.siddhi.SiddhiPolicyEvaluator;
-import org.apache.eagle.alert.siddhi.StreamMetadataManager;
+import org.apache.eagle.alert.executor.AlertExecutor;
+import org.apache.eagle.alert.siddhi.SiddhiAlertAPIEntityRender;
 import org.apache.eagle.dataproc.core.ValuesArray;
 import org.apache.eagle.datastream.Collector;
 import org.apache.eagle.datastream.Tuple2;
-import org.apache.eagle.executor.AlertExecutor;
-import junit.framework.Assert;
+import org.apache.eagle.policy.PolicyEvaluationContext;
+import org.apache.eagle.policy.dao.AlertDefinitionDAOImpl;
+import org.apache.eagle.policy.dao.AlertStreamSchemaDAO;
+import org.apache.eagle.policy.dao.AlertStreamSchemaDAOImpl;
+import org.apache.eagle.policy.dao.PolicyDefinitionDAO;
+import org.apache.eagle.policy.siddhi.SiddhiPolicyDefinition;
+import org.apache.eagle.policy.siddhi.SiddhiPolicyEvaluator;
+import org.apache.eagle.policy.siddhi.StreamMetadataManager;
 import org.apache.eagle.service.client.EagleServiceConnector;
 import org.junit.Test;
 
@@ -92,26 +93,28 @@ public class TestSiddhiEvaluator {
 							"select * " +
 							"insert into outputStream ;";
         policyDef.setExpression(expression);
-        SiddhiPolicyEvaluator evaluator = new SiddhiPolicyEvaluator(config, "testPolicy", policyDef, new String[]{"hdfsAuditLogEventStream"});
-		EagleAlertContext context = new EagleAlertContext();
+        SiddhiPolicyEvaluator<AlertDefinitionAPIEntity, AlertAPIEntity> evaluator = new SiddhiPolicyEvaluator<AlertDefinitionAPIEntity, AlertAPIEntity>(config, "testPolicy", policyDef, new String[]{"hdfsAuditLogEventStream"});
+		PolicyEvaluationContext context = new PolicyEvaluationContext();
 
-		AlertDefinitionDAO alertDao = new AlertDefinitionDAOImpl(new EagleServiceConnector(null, null)) {
+		PolicyDefinitionDAO alertDao = new AlertDefinitionDAOImpl(new EagleServiceConnector(null, null)) {
 			@Override
-			public Map<String, Map<String, AlertDefinitionAPIEntity>> findActiveAlertDefsGroupbyAlertExecutorId(String site, String dataSource) throws Exception {
+			public Map<String, Map<String, AlertDefinitionAPIEntity>> findActivePoliciesGroupbyExecutorId(String site, String dataSource) throws Exception {
 				return null;
 			}
 		};
 
-		context.alertExecutor = new AlertExecutor("alertExecutorId", null, 3, 1, alertDao, new String[]{"hdfsAuditLogEventStream"}) {
+		AlertExecutor alertExecutor = new AlertExecutor("alertExecutorId", null, 3, 1, alertDao, new String[]{"hdfsAuditLogEventStream"}) {
 			@Override
 			protected Map<String, String> getDimensions(String policyId) {
 				return new HashMap<String, String>();
 			}
 		};
-		context.alertExecutor.prepareConfig(config);
-		context.alertExecutor.init();
+		alertExecutor.prepareConfig(config);
+		alertExecutor.init();
+		context.alertExecutor = alertExecutor;
 		context.evaluator = evaluator;
 		context.policyId = "testPolicy";
+		context.resultRender = new SiddhiAlertAPIEntityRender();
 		context.outputCollector = new Collector<Tuple2<String, AlertAPIEntity>> () {
 			@Override
 			public void collect(Tuple2<String, AlertAPIEntity> stringAlertAPIEntityTuple2) {

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestAlertDefinitionDAOImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestAlertDefinitionDAOImpl.java b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestAlertDefinitionDAOImpl.java
index 70853b4..783abea 100644
--- a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestAlertDefinitionDAOImpl.java
+++ b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestAlertDefinitionDAOImpl.java
@@ -18,8 +18,10 @@ package org.apache.eagle.alert.dao;
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
+import org.apache.eagle.policy.dao.AlertDefinitionDAOImpl;
 import org.apache.eagle.alert.entity.AlertDefinitionAPIEntity;
 import org.apache.eagle.common.config.EagleConfigConstants;
+import org.apache.eagle.policy.dao.PolicyDefinitionDAO;
 import org.apache.eagle.service.client.EagleServiceConnector;
 import org.junit.Assert;
 import org.junit.Test;
@@ -52,9 +54,9 @@ public class TestAlertDefinitionDAOImpl {
 
 		String site = "sandbox";
 		String dataSource = "UnitTest";
-		AlertDefinitionDAO dao = new AlertDefinitionDAOImpl(new EagleServiceConnector(eagleServiceHost, eagleServicePort)) {
+		PolicyDefinitionDAO dao = new AlertDefinitionDAOImpl(new EagleServiceConnector(eagleServiceHost, eagleServicePort)) {
 			@Override
-			public List<AlertDefinitionAPIEntity> findActiveAlertDefs(String site, String dataSource) throws Exception {
+			public List<AlertDefinitionAPIEntity> findActivePolicies(String site, String dataSource) throws Exception {
 				List<AlertDefinitionAPIEntity> list = new ArrayList<AlertDefinitionAPIEntity>();
 				list.add(buildTestAlertDefEntity(site, dataSource, "TestExecutor1", "TestPolicyIDA", "TestPolicyTypeA"));
 				list.add(buildTestAlertDefEntity(site, dataSource, "TestExecutor1", "TestPolicyIDB", "TestPolicyTypeB"));
@@ -64,7 +66,7 @@ public class TestAlertDefinitionDAOImpl {
 			}
 		};
 
-		Map<String, Map<String, AlertDefinitionAPIEntity>> retMap = dao.findActiveAlertDefsGroupbyAlertExecutorId(site, dataSource);
+		Map<String, Map<String, AlertDefinitionAPIEntity>> retMap = dao.findActivePoliciesGroupbyExecutorId(site, dataSource);
 		
 		Assert.assertEquals(2, retMap.size());
 		Assert.assertEquals(2, retMap.get("TestExecutor1").size());

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestSiddhiStreamMetadataUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestSiddhiStreamMetadataUtils.java b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestSiddhiStreamMetadataUtils.java
index b7f40fa..82aa89f 100644
--- a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestSiddhiStreamMetadataUtils.java
+++ b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestSiddhiStreamMetadataUtils.java
@@ -24,9 +24,10 @@ import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 import junit.framework.Assert;
 
+import org.apache.eagle.policy.dao.AlertStreamSchemaDAO;
 import org.apache.eagle.alert.entity.AlertStreamSchemaEntity;
-import org.apache.eagle.alert.siddhi.SiddhiStreamMetadataUtils;
-import org.apache.eagle.alert.siddhi.StreamMetadataManager;
+import org.apache.eagle.policy.siddhi.SiddhiStreamMetadataUtils;
+import org.apache.eagle.policy.siddhi.StreamMetadataManager;
 import org.junit.Test;
 
 public class TestSiddhiStreamMetadataUtils {
@@ -43,7 +44,7 @@ public class TestSiddhiStreamMetadataUtils {
 			}
 		});
 		String siddhiStreamDef = SiddhiStreamMetadataUtils.convertToStreamDef("testStreamName");
-		Assert.assertEquals("define stream " + "testStreamName" + "(eagleAlertContext object,attrName1 string,attrName2 long);", siddhiStreamDef);
+		Assert.assertEquals("define stream " + "testStreamName" + "(eagleAlertContext object, attrName1 string,attrName2 long);", siddhiStreamDef);
 		StreamMetadataManager.getInstance().reset();
 	}
 	

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/da8f419c/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestStreamDefinitionDAOImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestStreamDefinitionDAOImpl.java b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestStreamDefinitionDAOImpl.java
index 80aacd6..91c5332 100644
--- a/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestStreamDefinitionDAOImpl.java
+++ b/eagle-core/eagle-alert/eagle-alert-process/src/test/java/org/apache/eagle/alert/dao/TestStreamDefinitionDAOImpl.java
@@ -18,8 +18,10 @@ package org.apache.eagle.alert.dao;
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
+import org.apache.eagle.policy.dao.AlertStreamSchemaDAO;
+import org.apache.eagle.policy.dao.AlertStreamSchemaDAOImpl;
 import org.apache.eagle.alert.entity.AlertStreamSchemaEntity;
-import org.apache.eagle.alert.siddhi.StreamMetadataManager;
+import org.apache.eagle.policy.siddhi.StreamMetadataManager;
 import org.junit.Assert;
 import org.junit.Test;
 


Mime
View raw message