PolicyInterpreter { public static PolicyValidationResult validate(PolicyDefinition policy, Map<String, StreamDefinition> allDefinitions) { Map<String, StreamDefinition> inputDefinitions = new HashMap<>(); PolicyValidationResult policyValidationResult = new PolicyValidationResult(); policyValidationResult.setPolicyDefinition(policy); try { if (policy.getInputStreams() != null) { for (String streamId : policy.getInputStreams()) { if (allDefinitions.containsKey(streamId)) { inputDefinitions.put(streamId, allDefinitions.get(streamId)); } else { throw new StreamNotDefinedException(streamId); } } } PolicyExecutionPlan policyExecutionPlan = null; if (PolicyStreamHandlers.SIDDHI_ENGINE.equalsIgnoreCase(policy.getDefinition().getType())) { policyExecutionPlan = parseExecutionPlan(policy.getDefinition().getValue(), inputDefinitions); if (policy.getOutputStreams() != null) { for (String outputStream : policy.getOutputStreams()) { if (!policyExecutionPlan.getOutputStreams().containsKey(outputStream)) { throw new StreamNotDefinedException("Output stream " + outputStream + " not defined"); } } } } policyValidationResult.setPolicyExecutionPlan(policyExecutionPlan); policyValidationResult.setSuccess(true); policyValidationResult.setMessage("Validated successfully"); } catch (Exception exception) { LOG.error("Got error to validate policy definition: {}", policy, exception); policyValidationResult.setSuccess(false); policyValidationResult.setMessage(exception.getMessage()); policyValidationResult.setStackTrace(exception); } return policyValidationResult; }  static PolicyParseResult parse(String executionPlanQuery); static PolicyExecutionPlan parseExecutionPlan(String policyDefinition, Map<String, StreamDefinition> inputStreamDefinitions); static PolicyExecutionPlan parseExecutionPlan(String executionPlanQuery); static PolicyValidationResult validate(PolicyDefinition policy, Map<String, StreamDefinition> allDefinitions);  }
@Test public void testValidPolicyWithExternalTimeWindow() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("INPUT_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM_1")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); definition.setValue("from INPUT_STREAM_1#window.externalTime(timestamp, 2 min) select name, sum(value) as total group by name insert into OUTPUT_STREAM_1 ;"); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("INPUT_STREAM_1", mockStreamDefinition("INPUT_STREAM_1")); put("INPUT_STREAM_2", mockStreamDefinition("INPUT_STREAM_2")); put("INPUT_STREAM_3", mockStreamDefinition("INPUT_STREAM_3")); put("INPUT_STREAM_4", mockStreamDefinition("INPUT_STREAM_4")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getOutputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getStreamPartitions().size()); Assert.assertNotNull(validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec()); }
@Test public void testValidPolicyWithTimeWindow() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("INPUT_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM_1")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); definition.setValue("from INPUT_STREAM_1#window.time(2 min) select name, sum(value) as total group by name insert into OUTPUT_STREAM_1 ;"); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("INPUT_STREAM_1", mockStreamDefinition("INPUT_STREAM_1")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getOutputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getStreamPartitions().size()); Assert.assertNull(validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec()); }
@Test public void testValidPolicyWithTooManyInputStreams() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Arrays.asList("INPUT_STREAM_1", "INPUT_STREAM_2")); policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM_1")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); definition.setValue("from INPUT_STREAM_1[value > 90.0] select * group by name insert into OUTPUT_STREAM_1;"); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("INPUT_STREAM_1", mockStreamDefinition("INPUT_STREAM_1")); put("INPUT_STREAM_2", mockStreamDefinition("INPUT_STREAM_2")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getOutputStreams().size()); }
@Test public void testValidPolicyWithTooFewOutputStreams() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Arrays.asList("INPUT_STREAM_1", "INPUT_STREAM_2")); policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM_1")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); definition.setValue( "from INPUT_STREAM_1[value > 90.0] select * group by name insert into OUTPUT_STREAM_1;" + "from INPUT_STREAM_1[value < 90.0] select * group by name insert into OUTPUT_STREAM_2;" ); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("INPUT_STREAM_1", mockStreamDefinition("INPUT_STREAM_1")); put("INPUT_STREAM_2", mockStreamDefinition("INPUT_STREAM_2")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(2, validation.getPolicyExecutionPlan().getOutputStreams().size()); }
@Test public void testInvalidPolicyForSyntaxError() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("INPUT_STREAM")); policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); definition.setValue("from INPUT_STREAM (value > 90.0) select * group by name insert into OUTPUT_STREAM;"); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("INPUT_STREAM", mockStreamDefinition("INPUT_STREAM")); } }); Assert.assertFalse(validation.isSuccess()); }
@Test public void testInvalidPolicyForNotDefinedInputStream() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("INPUT_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM_1")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); definition.setValue("from INPUT_STREAM_1[value > 90.0] select * group by name insert into OUTPUT_STREAM_1;"); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("INPUT_STREAM_2", mockStreamDefinition("INPUT_STREAM_2")); } }); Assert.assertFalse(validation.isSuccess()); }
@Test public void testInvalidPolicyForNotDefinedOutputStream() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("INPUT_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("OUTPUT_STREAM_2")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); definition.setValue("from INPUT_STREAM_1[value > 90.0] select * group by name insert into OUTPUT_STREAM_1;"); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("INPUT_STREAM_1", mockStreamDefinition("INPUT_STREAM_1")); } }); Assert.assertFalse(validation.isSuccess()); }
@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()); }
@Test public void testValidPolicyWithPattern() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); String policy = "from every a = HADOOP_JMX_METRIC_STREAM_1[component==\"namenode\" and metric == \"hadoop.namenode.dfs.missingblocks\"] " + "-> b = HADOOP_JMX_METRIC_STREAM_1[b.component==a.component and b.metric==a.metric and b.host==a.host and convert(b.value, \"long\") > convert(a.value, \"long\") ] " + "select b.metric, b.host as host, convert(b.value, \"long\") as newNumOfMissingBlocks, convert(a.value, \"long\") as oldNumOfMissingBlocks, b.timestamp as timestamp, b.component as component, b.site as site " + "group by b.metric insert into HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT;"; definition.setValue(policy); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("HADOOP_JMX_METRIC_STREAM_1", mockStreamDefinition("HADOOP_JMX_METRIC_STREAM_1")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getOutputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getStreamPartitions().size()); Assert.assertNull(validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec()); Assert.assertEquals(StreamPartition.Type.GROUPBY, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getType()); Assert.assertArrayEquals(new String[]{"metric"}, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getColumns().toArray()); Assert.assertEquals("HADOOP_JMX_METRIC_STREAM_1", validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getStreamId()); }
@Test public void testValidPolicyWithPatternSort() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); String policy = "from HADOOP_JMX_METRIC_STREAM_1[metric == \"hadoop.namenode.dfs.missingblocks\"]#window.externalTime(timestamp, 1 min) " + "select * group by site, host, component, metric insert into temp;\n" + "\n" + "from every a = HADOOP_JMX_METRIC_STREAM_1[metric == \"hadoop.namenode.dfs.missingblocks\"] -> b = HADOOP_JMX_METRIC_STREAM_1[b.component == a.component and b.metric == a.metric and b.host == a.host and convert(b.value, \"long\") > convert(a.value, \"long\") ] " + "select b.site, b.host, b.component, b.metric, convert(b.value, \"long\") as newNumOfMissingBlocks, convert(a.value, \"long\") as oldNumOfMissingBlocks, max(b.timestamp) as timestamp " + "group by b.site, b.host, b.component, b.metric insert into HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT;"; definition.setValue(policy); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("HADOOP_JMX_METRIC_STREAM_1", mockStreamDefinition("HADOOP_JMX_METRIC_STREAM_1")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(2, validation.getPolicyExecutionPlan().getOutputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getStreamPartitions().size()); Assert.assertNotNull(validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec()); Assert.assertEquals(60000, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); Assert.assertEquals(12000, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec().getWindowMargin()); Assert.assertEquals(StreamPartition.Type.GROUPBY, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getType()); Assert.assertArrayEquals(new String[]{"site", "host", "component", "metric"}, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getColumns().toArray()); Assert.assertEquals("HADOOP_JMX_METRIC_STREAM_1", validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getStreamId()); }
@Test public void testValidPolicyWithSequence() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); String policy = "from every a = HADOOP_JMX_METRIC_STREAM_1[component==\"namenode\" and metric == \"hadoop.namenode.dfs.missingblocks\"] " + ", b = HADOOP_JMX_METRIC_STREAM_1[b.component==a.component and b.metric==a.metric and b.host==a.host and convert(b.value, \"long\") > convert(a.value, \"long\") ] " + "select b.metric, b.host as host, convert(b.value, \"long\") as newNumOfMissingBlocks, convert(a.value, \"long\") as oldNumOfMissingBlocks, b.timestamp as timestamp, b.component as component, b.site as site " + "group by b.metric insert into HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT;"; definition.setValue(policy); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("HADOOP_JMX_METRIC_STREAM_1", mockStreamDefinition("HADOOP_JMX_METRIC_STREAM_1")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getOutputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getStreamPartitions().size()); Assert.assertNull(validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec()); Assert.assertEquals(StreamPartition.Type.GROUPBY, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getType()); Assert.assertArrayEquals(new String[]{"metric"}, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getColumns().toArray()); Assert.assertEquals("HADOOP_JMX_METRIC_STREAM_1", validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getStreamId()); }
@Test public void testValidPolicyWithSequenceSort() { PolicyDefinition policyDefinition = new PolicyDefinition(); policyDefinition.setName("test_policy"); policyDefinition.setInputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1")); policyDefinition.setOutputStreams(Collections.singletonList("HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT")); PolicyDefinition.Definition definition = new PolicyDefinition.Definition(); definition.setType("siddhi"); String policy = "from HADOOP_JMX_METRIC_STREAM_1[metric == \"hadoop.namenode.dfs.missingblocks\"]#window.externalTime(timestamp, 1 min) " + "select * group by site, host, component, metric insert into temp;\n" + "\n" + "from every a = HADOOP_JMX_METRIC_STREAM_1[metric == \"hadoop.namenode.dfs.missingblocks\"], b = HADOOP_JMX_METRIC_STREAM_1[b.component == a.component and b.metric == a.metric and b.host == a.host and convert(b.value, \"long\") > convert(a.value, \"long\") ] " + "select b.site, b.host, b.component, b.metric, convert(b.value, \"long\") as newNumOfMissingBlocks, convert(a.value, \"long\") as oldNumOfMissingBlocks, max(b.timestamp) as timestamp " + "group by b.site, b.host, b.component, b.metric insert into HADOOP_JMX_METRIC_STREAM_1_MISS_BLOCKS_OUT;"; definition.setValue(policy); definition.setInputStreams(policyDefinition.getInputStreams()); definition.setOutputStreams(policyDefinition.getOutputStreams()); policyDefinition.setDefinition(definition); PolicyValidationResult validation = PolicyInterpreter.validate(policyDefinition, new HashMap<String, StreamDefinition>() { { put("HADOOP_JMX_METRIC_STREAM_1", mockStreamDefinition("HADOOP_JMX_METRIC_STREAM_1")); } }); Assert.assertTrue(validation.isSuccess()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getInputStreams().size()); Assert.assertEquals(2, validation.getPolicyExecutionPlan().getOutputStreams().size()); Assert.assertEquals(1, validation.getPolicyExecutionPlan().getStreamPartitions().size()); Assert.assertNotNull(validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec()); Assert.assertEquals(60000, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); Assert.assertEquals(12000, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getSortSpec().getWindowMargin()); Assert.assertEquals(StreamPartition.Type.GROUPBY, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getType()); Assert.assertArrayEquals(new String[]{"site", "host", "component", "metric"}, validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getColumns().toArray()); Assert.assertEquals("HADOOP_JMX_METRIC_STREAM_1", validation.getPolicyExecutionPlan().getStreamPartitions().get(0).getStreamId()); }