| /* |
| * |
| * 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.metron.profiler.storm; |
| |
| import org.adrianwalker.multilinestring.Multiline; |
| import org.apache.metron.common.configuration.profiler.ProfileConfig; |
| import org.apache.metron.common.configuration.profiler.ProfilerConfig; |
| import org.apache.metron.common.utils.JSONUtils; |
| import org.apache.metron.profiler.DefaultMessageRouter; |
| import org.apache.metron.profiler.clock.FixedClockFactory; |
| import org.apache.metron.test.bolt.BaseBoltTest; |
| import org.apache.storm.tuple.Tuple; |
| import org.apache.storm.tuple.Values; |
| import org.json.simple.JSONObject; |
| import org.json.simple.parser.JSONParser; |
| import org.json.simple.parser.ParseException; |
| import org.junit.jupiter.api.BeforeEach; |
| import org.junit.jupiter.api.Test; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.InputStream; |
| import java.nio.charset.StandardCharsets; |
| import java.util.HashMap; |
| |
| import static org.apache.metron.storm.kafka.flux.SimpleStormKafkaBuilder.FieldsConfiguration.VALUE; |
| import static org.mockito.Mockito.*; |
| |
| /** |
| * Tests the ProfileSplitterBolt. |
| */ |
| public class ProfileSplitterBoltTest extends BaseBoltTest { |
| |
| /** |
| * { |
| * "ip_src_addr": "10.0.0.1", |
| * "ip_dst_addr": "10.0.0.20", |
| * "protocol": "HTTP", |
| * "timestamp.custom": 2222222222222, |
| * "timestamp.string": "3333333333333" |
| * } |
| */ |
| @Multiline |
| private String input; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "test", |
| * "foreach": "ip_src_addr", |
| * "onlyif": "protocol == 'HTTP'", |
| * "init": {}, |
| * "update": {}, |
| * "result": "2" |
| * } |
| * ] |
| * } |
| */ |
| @Multiline |
| private String profileWithOnlyIfTrue; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "test", |
| * "foreach": "ip_src_addr", |
| * "onlyif": "false", |
| * "init": {}, |
| * "update": {}, |
| * "result": "2" |
| * } |
| * ] |
| * } |
| */ |
| @Multiline |
| private String profileWithOnlyIfFalse; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "test", |
| * "foreach": "ip_src_addr", |
| * "init": {}, |
| * "update": {}, |
| * "result": "2" |
| * } |
| * ] |
| * } |
| */ |
| @Multiline |
| private String profileWithOnlyIfMissing; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "test", |
| * "foreach": "ip_src_addr", |
| * "onlyif": "NOT-VALID", |
| * "init": {}, |
| * "update": {}, |
| * "result": "2" |
| * } |
| * ] |
| * } |
| */ |
| @Multiline |
| private String profileWithOnlyIfInvalid; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "test", |
| * "foreach": "ip_src_addr", |
| * "init": {}, |
| * "update": {}, |
| * "result": "2" |
| * } |
| * ], |
| * "timestampField": "timestamp.custom" |
| * } |
| */ |
| @Multiline |
| private String profileUsingCustomTimestampField; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "test", |
| * "foreach": "ip_src_addr", |
| * "init": {}, |
| * "update": {}, |
| * "result": "2" |
| * } |
| * ], |
| * "timestampField": "timestamp.missing" |
| * } |
| */ |
| @Multiline |
| private String profileUsingMissingTimestampField; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "test", |
| * "foreach": "ip_src_addr", |
| * "init": {}, |
| * "update": {}, |
| * "result": "2" |
| * } |
| * ], |
| * "timestampField": "timestamp.string" |
| * } |
| */ |
| @Multiline |
| private String profileUsingStringTimestampField; |
| |
| /** |
| * { |
| * "profiles": [ |
| * ] |
| * } |
| */ |
| @Multiline |
| private String noProfilesDefined; |
| |
| /** |
| * { |
| * "profiles": [ |
| * { |
| * "profile": "profile1", |
| * "foreach": "'global'", |
| * "result": "1" |
| * }, |
| * { |
| * "profile": "profile2", |
| * "foreach": "'global'", |
| * "result": "2" |
| * } |
| * ] |
| * } |
| */ |
| @Multiline |
| private String twoProfilesDefined; |
| |
| private JSONObject message; |
| private long timestamp = 3333333; |
| |
| @BeforeEach |
| public void setup() throws ParseException { |
| // parse the input message |
| JSONParser parser = new JSONParser(); |
| message = (JSONObject) parser.parse(input); |
| |
| // ensure the tuple returns the expected json message |
| when(tuple.getBinaryByField(VALUE.getFieldName())).thenReturn(input.getBytes(StandardCharsets.UTF_8)); |
| } |
| |
| /** |
| * Ensure that a tuple with the correct fields is emitted to downstream bolts |
| * when a profile is defined. |
| */ |
| @Test |
| public void testEmitTupleWithOneProfile() throws Exception { |
| |
| // setup the bolt and execute a tuple |
| ProfilerConfig config = toProfilerConfig(profileWithOnlyIfTrue); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // the expected tuple fields |
| String expectedEntity = "10.0.0.1"; |
| ProfileConfig expectedConfig = config.getProfiles().get(0); |
| Values expected = new Values(message, timestamp, expectedEntity, expectedConfig); |
| |
| // a tuple should be emitted for the downstream profile builder |
| verify(outputCollector, times(1)) |
| .emit(eq(tuple), eq(expected)); |
| |
| // the original tuple should be ack'd |
| verify(outputCollector, times(1)) |
| .ack(eq(tuple)); |
| } |
| |
| /** |
| * If there are two profiles that need the same message, then two tuples should |
| * be emitted. One tuple for each profile. |
| */ |
| @Test |
| public void testEmitTupleWithTwoProfiles() throws Exception { |
| |
| // setup the bolt and execute a tuple |
| ProfilerConfig config = toProfilerConfig(twoProfilesDefined); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // the expected tuple fields |
| final String expectedEntity = "global"; |
| { |
| // a tuple should be emitted for the first profile |
| ProfileConfig profile1 = config.getProfiles().get(0); |
| Values expected = new Values(message, timestamp, expectedEntity, profile1); |
| verify(outputCollector, times(1)) |
| .emit(eq(tuple), eq(expected)); |
| } |
| { |
| // a tuple should be emitted for the second profile |
| ProfileConfig profile2 = config.getProfiles().get(1); |
| Values expected = new Values(message, timestamp, expectedEntity, profile2); |
| verify(outputCollector, times(1)) |
| .emit(eq(tuple), eq(expected)); |
| } |
| |
| // the original tuple should be ack'd |
| verify(outputCollector, times(1)) |
| .ack(eq(tuple)); |
| } |
| |
| /** |
| * No tuples should be emitted, if no profiles are defined. |
| */ |
| @Test |
| public void testNoProfilesDefined() throws Exception { |
| |
| // setup the bolt and execute a tuple |
| ProfilerConfig config = toProfilerConfig(noProfilesDefined); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // no tuple should be emitted |
| verify(outputCollector, times(0)) |
| .emit(any(Tuple.class), any()); |
| |
| // the original tuple should be ack'd |
| verify(outputCollector, times(1)) |
| .ack(eq(tuple)); |
| } |
| |
| /** |
| * What happens when a profile's 'onlyif' expression is true? The message |
| * should be applied to the profile. |
| */ |
| @Test |
| public void testOnlyIfTrue() throws Exception { |
| |
| ProfilerConfig config = toProfilerConfig(profileWithOnlyIfTrue); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // a tuple should be emitted for the downstream profile builder |
| verify(outputCollector, times(1)) |
| .emit(eq(tuple), any(Values.class)); |
| |
| // the original tuple should be ack'd |
| verify(outputCollector, times(1)) |
| .ack(eq(tuple)); |
| } |
| |
| /** |
| * All messages are applied to a profile where 'onlyif' is missing. A profile with no |
| * 'onlyif' is treated the same as if 'onlyif=true'. |
| */ |
| @Test |
| public void testOnlyIfMissing() throws Exception { |
| |
| ProfilerConfig config = toProfilerConfig(profileWithOnlyIfMissing); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // a tuple should be emitted for the downstream profile builder |
| verify(outputCollector, times(1)) |
| .emit(eq(tuple), any(Values.class)); |
| |
| // the original tuple should be ack'd |
| verify(outputCollector, times(1)) |
| .ack(eq(tuple)); |
| } |
| |
| /** |
| * What happens when a profile's 'onlyif' expression is false? The message |
| * should NOT be applied to the profile. |
| */ |
| @Test |
| public void testOnlyIfFalse() throws Exception { |
| |
| ProfilerConfig config = toProfilerConfig(profileWithOnlyIfFalse); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // a tuple should NOT be emitted for the downstream profile builder |
| verify(outputCollector, times(0)) |
| .emit(any()); |
| |
| // the original tuple should be ack'd |
| verify(outputCollector, times(1)) |
| .ack(eq(tuple)); |
| } |
| |
| /** |
| * The entity associated with a profile is defined with a Stellar expression. That expression |
| * can refer to any field within the message. |
| * |
| * In this case the entity is defined as 'ip_src_addr' which is resolved to '10.0.0.1' based on |
| * the data contained within the message. |
| */ |
| @Test |
| public void testResolveEntityName() throws Exception { |
| |
| ProfilerConfig config = toProfilerConfig(profileWithOnlyIfTrue); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // expected values |
| String expectedEntity = "10.0.0.1"; |
| ProfileConfig expectedConfig = config.getProfiles().get(0); |
| Values expected = new Values(message, timestamp, expectedEntity, expectedConfig); |
| |
| // a tuple should be emitted for the downstream profile builder |
| verify(outputCollector, times(1)) |
| .emit(eq(tuple), eq(expected)); |
| |
| // the original tuple should be ack'd |
| verify(outputCollector, times(1)) |
| .ack(eq(tuple)); |
| } |
| |
| /** |
| * What happens when invalid Stella code is used for 'onlyif'? The invalid profile should be ignored. |
| */ |
| @Test |
| public void testOnlyIfInvalid() throws Exception { |
| |
| ProfilerConfig config = toProfilerConfig(profileWithOnlyIfInvalid); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // a tuple should NOT be emitted for the downstream profile builder |
| verify(outputCollector, times(0)) |
| .emit(any(Values.class)); |
| } |
| |
| @Test |
| public void testWithNullMessage() throws Exception { |
| |
| // ensure the tuple returns null to mimic a null message in kafka |
| when(tuple.getBinary(0)).thenReturn(null); |
| |
| ProfilerConfig config = toProfilerConfig(profileWithOnlyIfInvalid); |
| ProfileSplitterBolt bolt = createBolt(config); |
| bolt.execute(tuple); |
| |
| // a tuple should NOT be emitted for the downstream profile builder |
| verify(outputCollector, times(0)) |
| .emit(any(Values.class)); |
| |
| } |
| |
| /** |
| * Creates a ProfilerConfig based on a string containing JSON. |
| * |
| * @param configAsJSON The config as JSON. |
| * @return The ProfilerConfig. |
| * @throws Exception |
| */ |
| private ProfilerConfig toProfilerConfig(String configAsJSON) throws Exception { |
| InputStream in = new ByteArrayInputStream(configAsJSON.getBytes(StandardCharsets.UTF_8)); |
| return JSONUtils.INSTANCE.load(in, ProfilerConfig.class); |
| } |
| |
| /** |
| * Create a ProfileSplitterBolt to test |
| */ |
| private ProfileSplitterBolt createBolt(ProfilerConfig config) { |
| |
| ProfileSplitterBolt bolt = new ProfileSplitterBolt("zookeeperURL"); |
| bolt.setCuratorFramework(client); |
| bolt.setZKCache(cache); |
| bolt.getConfigurations().updateProfilerConfig(config); |
| bolt.prepare(new HashMap<>(), topologyContext, outputCollector); |
| |
| // set the clock factory AFTER calling prepare to use the fixed clock factory |
| DefaultMessageRouter router = new DefaultMessageRouter(bolt.getStellarContext()); |
| router.setClockFactory(new FixedClockFactory(timestamp)); |
| bolt.setRouter(router); |
| |
| return bolt; |
| } |
| |
| } |
