Repository: incubator-eagle Updated Branches: refs/heads/master a5537c0c2 -> cb3bf7b60
EAGLE-746 : PolicyValidation is insufficient, and will cause blocking for extended handler Author: ralphsu This closes #625 Project: http://git-wip-us.apache.org/repos/asf/incubator-eagle/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-eagle/commit/cb3bf7b6 Tree: http://git-wip-us.apache.org/repos/asf/incubator-eagle/tree/cb3bf7b6 Diff: http://git-wip-us.apache.org/repos/asf/incubator-eagle/diff/cb3bf7b6 Branch: refs/heads/master Commit: cb3bf7b6063adff6ef566650b7a438dfe4b52290 Parents: a5537c0 Author: Ralph, Su <suliang...@gmail.com> Authored: Tue Nov 8 19:45:49 2016 +0800 Committer: Ralph, Su <suliang...@gmail.com> Committed: Tue Nov 8 20:08:45 2016 +0800 ---------------------------------------------------------------------- .../engine/interpreter/PolicyInterpreter.java | 16 +- .../interpreter/PolicyInterpreterTest.java | 39 +- .../src/test/resources/interpreter/policy.json | 36 ++ .../src/test/resources/interpreter/streams.json | 524 +++++++++++++++++++ 4 files changed, 606 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cb3bf7b6/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java index 8aec294..4add3ff 100644 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java @@ -20,6 +20,7 @@ import com.google.common.base.Preconditions; import org.apache.eagle.alert.engine.coordinator.PolicyDefinition; import org.apache.eagle.alert.engine.coordinator.StreamDefinition; import org.apache.eagle.alert.engine.coordinator.StreamNotDefinedException; +import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandlers; import org.apache.eagle.alert.engine.evaluator.impl.SiddhiDefinitionAdapter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -88,12 +89,15 @@ public class PolicyInterpreter { } } - PolicyExecutionPlan policyExecutionPlan = parseExecutionPlan(policy.getDefinition().getValue(), inputDefinitions); - // Validate output - if (policy.getOutputStreams() != null) { - for (String outputStream : policy.getOutputStreams()) { - if (!policyExecutionPlan.getOutputStreams().containsKey(outputStream)) { - throw new StreamNotDefinedException("Output stream " + outputStream + " not defined"); + PolicyExecutionPlan policyExecutionPlan = null; + if (PolicyStreamHandlers.SIDDHI_ENGINE.equalsIgnoreCase(policy.getDefinition().getType())) { + policyExecutionPlan = parseExecutionPlan(policy.getDefinition().getValue(), inputDefinitions); + // Validate output + if (policy.getOutputStreams() != null) { + for (String outputStream : policy.getOutputStreams()) { + if (!policyExecutionPlan.getOutputStreams().containsKey(outputStream)) { + throw new StreamNotDefinedException("Output stream " + outputStream + " not defined"); + } } } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cb3bf7b6/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreterTest.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreterTest.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreterTest.java index cf1a7e9..a10538e 100644 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreterTest.java +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreterTest.java @@ -16,13 +16,14 @@ */ package org.apache.eagle.alert.engine.interpreter; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; import org.apache.eagle.alert.engine.coordinator.PolicyDefinition; import org.apache.eagle.alert.engine.coordinator.StreamColumn; import org.apache.eagle.alert.engine.coordinator.StreamDefinition; import org.apache.eagle.alert.engine.coordinator.StreamPartition; -import org.apache.eagle.alert.engine.interpreter.PolicyExecutionPlan; -import org.apache.eagle.alert.engine.interpreter.PolicyInterpreter; -import org.apache.eagle.alert.engine.interpreter.PolicyValidationResult; +import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandlers; import org.junit.Assert; import org.junit.Test; import org.wso2.siddhi.core.exception.DefinitionNotExistException; @@ -364,6 +365,38 @@ public class PolicyInterpreterTest { ); } + private static final ObjectMapper mapper = new ObjectMapper(); + + @Test + public void testLeftJoin() throws Exception { + PolicyDefinition def = mapper.readValue(PolicyInterpreterTest.class.getResourceAsStream("/interpreter/policy.json"), PolicyDefinition.class); + ArrayNode array = (ArrayNode)mapper.readTree(PolicyInterpreterTest.class.getResourceAsStream("/interpreter/streams.json")); + Map<String, StreamDefinition> allDefinitions = new HashMap<>(); + for(JsonNode node : array) { + StreamDefinition streamDef = mapper.readValue(node.toString(), StreamDefinition.class); + allDefinitions.put(streamDef.getStreamId(), streamDef); + } + PolicyValidationResult result = PolicyInterpreter.validate(def, allDefinitions); + Assert.assertTrue(result.isSuccess()); + } + + @Test + public void testExtendPolicy() throws Exception { + PolicyDefinition policyDefinition = new PolicyDefinition(); + policyDefinition.setName("test-extend-policy"); + policyDefinition.setInputStreams(Collections.singletonList("INPUT_STREAM_1")); + policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM_1")); + PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); + definition.setType(PolicyStreamHandlers.CUSTOMIZED_ENGINE); + policyDefinition.setDefinition(definition); + + Map<String, StreamDefinition> allDefinitions = new HashMap<>(); + allDefinitions.put("INPUT_STREAM_1", mockStreamDefinition("INPUT_STREAM_1")); + PolicyValidationResult result = PolicyInterpreter.validate(policyDefinition, allDefinitions); + Assert.assertTrue(result.isSuccess()); + } + + // -------------- // Helper Methods // -------------- http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cb3bf7b6/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/policy.json ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/policy.json b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/policy.json new file mode 100644 index 0000000..72eabad --- /dev/null +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/policy.json @@ -0,0 +1,36 @@ +{ + "name": "network_symptomatic_join", + "description": "Set parentKey of child alerts as docId of device down alert", + "inputStreams": [ + "networkCorrelateIntermediumStream", + "deviceDownAlertStream" + ], + "outputStreams": [ + "symptomaticAlertOutputStream" + ], + "definition": { + "type": "siddhi", + "value": "from networkCorrelateIntermediumStream#window.time(10 min) as left join deviceDownAlertStream[state=='open']#window.time(10 min) as right on left.linkedSwitch == right.entity select left.name, left.pod, left.docId, left.cause, left.firstOccurrenceTime, left.bubble, left.timestamp, left.severity, left.componentType, left.linkedSwitch, left.alertSource, left.entityType, left.ip, left.alertKey, left.message, 'symptomaticAlertOutputStream' as streamName, left.component, left.linkedPort, left.namespace, left.entity, left.dc, left.status, left.endTime, left.lastModifiedTime, right.docId as parentKey insert into symptomaticAlertOutputStream;", + "properties": {}, + "inputStreams": [], + "outputStreams": [] + }, + "policyStatus": "ENABLED", + "partitionSpec": [ + { + "streamId": "networkCorrelateIntermediumStream", + "type": "GROUPBY", + "columns": [ + "linkedSwitch" + ] + }, + { + "streamId": "deviceDownAlertStream", + "type": "GROUPBY", + "columns": [ + "entity" + ] + } + ], + "parallelismHint": 0 +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cb3bf7b6/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/streams.json ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/streams.json b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/streams.json new file mode 100644 index 0000000..ed7c78f --- /dev/null +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/test/resources/interpreter/streams.json @@ -0,0 +1,524 @@ +[ + { + "streamId": "deviceDownAlertStream", + "description": "Stream definition for symptomatic parent alerts", + "validate": false, + "timeseries": false, + "dataSource": "network_aggregate_alert_output_datasource", + "columns": [ + { + "name": "docId", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the docId for correlated alerts" + }, + { + "name": "parentKey", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the parent key for correlated alert" + }, + { + "name": "state", + "type": "STRING", + "defaultValue": "closed", + "required": true, + "description": "the alert state" + }, + { + "name": "name", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "namespace", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "timestamp", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "" + }, + { + "name": "lastModifiedTime", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "" + }, + { + "name": "severity", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "severity for symptomatic alert" + }, + { + "name": "severityCode", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "severity code for symptomatic alert" + }, + { + "name": "streamName", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "entity", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "pod", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "bubble", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "dc", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "cause", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "message", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "count", + "type": "LONG", + "defaultValue": 0, + "required": false + }, + { + "name": "firstOccurrenceTime", + "type": "LONG", + "defaultValue": 0, + "required": false + }, + { + "name": "alertKey", + "type": "STRING", + "defaultValue": "", + "required": true + } + ] + }, + { + "streamId": "networkCorrelateIntermediumStream", + "description": "The template aggregated *OUTPUT* data stream schema for network correlation intermedium step", + "validate": false, + "timeseries": false, + "dataSource": "network_aggregate_alert_output_datasource", + "columns": [ + { + "name": "docId", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the docId for aggregated alerts" + }, + { + "name": "firstOccurrenceTime", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "firstOccurTime for aggregated alert" + }, + { + "name": "endTime", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "the end time of aggregated alert, 0 as default" + }, + { + "name": "alertKey", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the alert key for aggregated alert" + }, + { + "name": "parentKey", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the parent key for aggregated alert" + }, + { + "name": "state", + "type": "STRING", + "defaultValue": "closed", + "required": true, + "description": "the alert state" + }, + { + "name": "name", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "namespace", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "timestamp", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "" + }, + { + "name": "lastModifiedTime", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "" + }, + { + "name": "severity", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "severity for aggregated alert" + }, + { + "name": "alertSource", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "streamName", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "entity", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "entityType", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "ip", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "pod", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "bubble", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "dc", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "component", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "componentType", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "status", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "cause", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "message", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "linkedPort", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "linkedSwitch", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "severityCode", + "type": "STRING", + "defaultValue": "", + "required": false, + "description": "" + }, + { + "name": "count", + "type": "LONG", + "defaultValue": 0, + "required": false, + "description": "" + } + ] + }, + { + "streamId": "symptomaticAlertOutputStream", + "description": "Stream definition for symptomatic child alerts", + "validate": false, + "timeseries": false, + "dataSource": "network_aggregate_alert_output_datasource", + "columns": [ + { + "name": "docId", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the docId for symptomatic alerts" + }, + { + "name": "firstOccurrenceTime", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "firstOccurTime for symptomatic alert" + }, + { + "name": "endTime", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "the end time of symptomatic alert, 0 as default" + }, + { + "name": "alertKey", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the alert key for symptomatic alert" + }, + { + "name": "parentKey", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "the parent key for symptomatic alert" + }, + { + "name": "state", + "type": "STRING", + "defaultValue": "closed", + "required": true, + "description": "the alert state" + }, + { + "name": "name", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "namespace", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "timestamp", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "" + }, + { + "name": "lastModifiedTime", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "" + }, + { + "name": "severity", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "severity for symptomatic alert" + }, + { + "name": "severityCode", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "severity code for symptomatic alert" + }, + { + "name": "streamName", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "entity", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "entityType", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "ip", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "pod", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "bubble", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "dc", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "component", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "componentType", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "status", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "alertSource", + "type": "STRING", + "defaultValue": "", + "required": false + }, + { + "name": "cause", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "message", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "linkedPort", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "linkedSwitch", + "type": "STRING", + "defaultValue": "", + "required": true, + "description": "" + }, + { + "name": "count", + "type": "LONG", + "defaultValue": 0, + "required": true, + "description": "" + } + ] + } +] \ No newline at end of file