PipelineInterpreter implements MessageProcessor { @Override public Messages process(Messages messages) { try (Timer.Context ignored = executionTime.time()) { final State latestState = stateUpdater.getLatestState(); return process(messages, new NoopInterpreterListener(), latestState); } } @Inject  PipelineInterpreter(Journal journal,
                               MetricRegistry metricRegistry,
                               ConfigurationStateUpdater stateUpdater); @Override Messages process(Messages messages); Messages process(Messages messages, InterpreterListener interpreterListener, State state); List<Message> processForPipelines(Message message,
                                             Set<String> pipelineIds,
                                             InterpreterListener interpreterListener,
                                             State state); static final String GL2_PROCESSING_ERROR; }
@Test public void testCreateMessage() { final RuleService ruleService = mock(MongoDbRuleService.class); when(ruleService.loadAll()).thenReturn(Collections.singleton( RuleDao.create("abc", "title", "description", "rule \"creates message\"\n" + "when to_string($message.message) == \"original message\"\n" + "then\n" + " create_message(\"derived message\");\n" + "end", Tools.nowUTC(), null) )); final PipelineService pipelineService = mock(MongoDbPipelineService.class); when(pipelineService.loadAll()).thenReturn(Collections.singleton( PipelineDao.create("p1", "title", "description", "pipeline \"pipeline\"\n" + "stage 0 match all\n" + " rule \"creates message\";\n" + "end\n", Tools.nowUTC(), null) )); final Map<String, Function<?>> functions = ImmutableMap.of( CreateMessage.NAME, new CreateMessage(), StringConversion.NAME, new StringConversion()); final PipelineInterpreter interpreter = createPipelineInterpreter(ruleService, pipelineService, functions); Message msg = messageInDefaultStream("original message", "test"); final Messages processed = interpreter.process(msg); final Message[] messages = Iterables.toArray(processed, Message.class); assertEquals(2, messages.length); }
@Test public void testMatchAllContinuesIfAllRulesMatched() { final RuleService ruleService = mock(MongoDbRuleService.class); when(ruleService.loadAll()).thenReturn(ImmutableList.of(RULE_TRUE, RULE_FALSE, RULE_ADD_FOOBAR)); final PipelineService pipelineService = mock(MongoDbPipelineService.class); when(pipelineService.loadAll()).thenReturn(Collections.singleton( PipelineDao.create("p1", "title", "description", "pipeline \"pipeline\"\n" + "stage 0 match all\n" + " rule \"true\";\n" + "stage 1 match either\n" + " rule \"add_foobar\";\n" + "end\n", Tools.nowUTC(), null) )); final Map<String, Function<?>> functions = ImmutableMap.of(SetField.NAME, new SetField()); final PipelineInterpreter interpreter = createPipelineInterpreter(ruleService, pipelineService, functions); final Messages processed = interpreter.process(messageInDefaultStream("message", "test")); final List<Message> messages = ImmutableList.copyOf(processed); assertThat(messages).hasSize(1); final Message actualMessage = messages.get(0); assertThat(actualMessage.getFieldAs(String.class, "foobar")).isEqualTo("covfefe"); }
@Test public void testMatchAllDoesNotContinueIfNotAllRulesMatched() { final RuleService ruleService = mock(MongoDbRuleService.class); when(ruleService.loadAll()).thenReturn(ImmutableList.of(RULE_TRUE, RULE_FALSE, RULE_ADD_FOOBAR)); final PipelineService pipelineService = mock(MongoDbPipelineService.class); when(pipelineService.loadAll()).thenReturn(Collections.singleton( PipelineDao.create("p1", "title", "description", "pipeline \"pipeline\"\n" + "stage 0 match all\n" + " rule \"true\";\n" + " rule \"false\";\n" + "stage 1 match either\n" + " rule \"add_foobar\";\n" + "end\n", Tools.nowUTC(), null) )); final Map<String, Function<?>> functions = ImmutableMap.of(SetField.NAME, new SetField()); final PipelineInterpreter interpreter = createPipelineInterpreter(ruleService, pipelineService, functions); final Messages processed = interpreter.process(messageInDefaultStream("message", "test")); final List<Message> messages = ImmutableList.copyOf(processed); assertThat(messages).hasSize(1); final Message actualMessage = messages.get(0); assertThat(actualMessage.hasField("foobar")).isFalse(); }
@Test public void testMatchEitherContinuesIfOneRuleMatched() { final RuleService ruleService = mock(MongoDbRuleService.class); when(ruleService.loadAll()).thenReturn(ImmutableList.of(RULE_TRUE, RULE_FALSE, RULE_ADD_FOOBAR)); final PipelineService pipelineService = mock(MongoDbPipelineService.class); when(pipelineService.loadAll()).thenReturn(Collections.singleton( PipelineDao.create("p1", "title", "description", "pipeline \"pipeline\"\n" + "stage 0 match either\n" + " rule \"true\";\n" + " rule \"false\";\n" + "stage 1 match either\n" + " rule \"add_foobar\";\n" + "end\n", Tools.nowUTC(), null) )); final Map<String, Function<?>> functions = ImmutableMap.of(SetField.NAME, new SetField()); final PipelineInterpreter interpreter = createPipelineInterpreter(ruleService, pipelineService, functions); final Messages processed = interpreter.process(messageInDefaultStream("message", "test")); final List<Message> messages = ImmutableList.copyOf(processed); assertThat(messages).hasSize(1); final Message actualMessage = messages.get(0); assertThat(actualMessage.getFieldAs(String.class, "foobar")).isEqualTo("covfefe"); }
@Test public void testMatchEitherStopsIfNoRuleMatched() { final RuleService ruleService = mock(MongoDbRuleService.class); when(ruleService.loadAll()).thenReturn(ImmutableList.of(RULE_TRUE, RULE_FALSE, RULE_ADD_FOOBAR)); final PipelineService pipelineService = mock(MongoDbPipelineService.class); when(pipelineService.loadAll()).thenReturn(Collections.singleton( PipelineDao.create("p1", "title", "description", "pipeline \"pipeline\"\n" + "stage 0 match either\n" + " rule \"false\";\n" + "stage 1 match either\n" + " rule \"add_foobar\";\n" + "end\n", Tools.nowUTC(), null) )); final Map<String, Function<?>> functions = ImmutableMap.of(SetField.NAME, new SetField()); final PipelineInterpreter interpreter = createPipelineInterpreter(ruleService, pipelineService, functions); final Messages processed = interpreter.process(messageInDefaultStream("message", "test")); final List<Message> messages = ImmutableList.copyOf(processed); assertThat(messages).hasSize(1); final Message actualMessage = messages.get(0); assertThat(actualMessage.hasField("foobar")).isFalse(); }
@Test public void testMetrics() { final RuleService ruleService = new InMemoryRuleService(); ruleService.save(RuleDao.create("abc", "title", "description", "rule \"match_all\"\n" + "when true\n" + "then\n" + "end", Tools.nowUTC(), null) ); final PipelineService pipelineService = new InMemoryPipelineService(); pipelineService.save(PipelineDao.create("cde", "title", "description", "pipeline \"pipeline\"\n" + "stage 0 match all\n" + " rule \"match_all\";\n" + "stage 1 match all\n" + " rule \"match_all\";\n" + "end\n", Tools.nowUTC(), null) ); final PipelineStreamConnectionsService pipelineStreamConnectionsService = new InMemoryPipelineStreamConnectionsService(); pipelineStreamConnectionsService.save(PipelineConnections.create(null, DEFAULT_STREAM_ID, Collections.singleton("cde"))); final FunctionRegistry functionRegistry = new FunctionRegistry(Collections.emptyMap()); final PipelineRuleParser parser = new PipelineRuleParser(functionRegistry, new CodeGenerator(JavaCompiler::new)); final MetricRegistry metricRegistry = new MetricRegistry(); final ConfigurationStateUpdater stateUpdater = new ConfigurationStateUpdater(ruleService, pipelineService, pipelineStreamConnectionsService, parser, metricRegistry, functionRegistry, Executors.newScheduledThreadPool(1), mock(EventBus.class), (currentPipelines, streamPipelineConnections) -> new PipelineInterpreter.State(currentPipelines, streamPipelineConnections, new MetricRegistry(), 1, true), false); final PipelineInterpreter interpreter = new PipelineInterpreter( mock(Journal.class), metricRegistry, stateUpdater ); interpreter.process(messageInDefaultStream("", "")); final SortedMap<String, Meter> meters = metricRegistry.getMeters((name, metric) -> name.startsWith(name(Pipeline.class, "cde")) || name.startsWith(name(Rule.class, "abc"))); assertThat(meters.keySet()).containsExactlyInAnyOrder( name(Pipeline.class, "cde", "executed"), name(Pipeline.class, "cde", "stage", "0", "executed"), name(Pipeline.class, "cde", "stage", "1", "executed"), name(Rule.class, "abc", "executed"), name(Rule.class, "abc", "cde", "0", "executed"), name(Rule.class, "abc", "cde", "1", "executed"), name(Rule.class, "abc", "matched"), name(Rule.class, "abc", "cde", "0", "matched"), name(Rule.class, "abc", "cde", "1", "matched"), name(Rule.class, "abc", "not-matched"), name(Rule.class, "abc", "cde", "0", "not-matched"), name(Rule.class, "abc", "cde", "1", "not-matched"), name(Rule.class, "abc", "failed"), name(Rule.class, "abc", "cde", "0", "failed"), name(Rule.class, "abc", "cde", "1", "failed") ); assertThat(meters.get(name(Pipeline.class, "cde", "executed")).getCount()).isEqualTo(1L); assertThat(meters.get(name(Pipeline.class, "cde", "stage", "0", "executed")).getCount()).isEqualTo(1L); assertThat(meters.get(name(Pipeline.class, "cde", "stage", "1", "executed")).getCount()).isEqualTo(1L); assertThat(meters.get(name(Rule.class, "abc", "executed")).getCount()).isEqualTo(2L); assertThat(meters.get(name(Rule.class, "abc", "cde", "0", "executed")).getCount()).isEqualTo(1L); assertThat(meters.get(name(Rule.class, "abc", "cde", "1", "executed")).getCount()).isEqualTo(1L); assertThat(meters.get(name(Rule.class, "abc", "matched")).getCount()).isEqualTo(2L); assertThat(meters.get(name(Rule.class, "abc", "cde", "0", "matched")).getCount()).isEqualTo(1L); assertThat(meters.get(name(Rule.class, "abc", "cde", "1", "matched")).getCount()).isEqualTo(1L); assertThat(meters.get(name(Rule.class, "abc", "not-matched")).getCount()).isEqualTo(0L); assertThat(meters.get(name(Rule.class, "abc", "cde", "0", "not-matched")).getCount()).isEqualTo(0L); assertThat(meters.get(name(Rule.class, "abc", "cde", "1", "not-matched")).getCount()).isEqualTo(0L); assertThat(meters.get(name(Rule.class, "abc", "failed")).getCount()).isEqualTo(0L); assertThat(meters.get(name(Rule.class, "abc", "cde", "0", "failed")).getCount()).isEqualTo(0L); assertThat(meters.get(name(Rule.class, "abc", "cde", "1", "failed")).getCount()).isEqualTo(0L); }