You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by mm...@apache.org on 2018/07/20 15:40:29 UTC
[01/15] metron git commit: METRON-1656 Create KAKFA_SEEK function
(nickwallen) closes apache/metron#1097
Repository: metron
Updated Branches:
refs/heads/feature/METRON-1554-pcap-query-panel a5a51399d -> b580a69f1
METRON-1656 Create KAKFA_SEEK function (nickwallen) closes apache/metron#1097
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/7af11b62
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/7af11b62
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/7af11b62
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 7af11b626e2d30aaaf7c01c364295b0b407fae49
Parents: cbdaee1
Author: nickwallen <ni...@nickallen.org>
Authored: Wed Jul 11 11:10:06 2018 -0400
Committer: nickallen <ni...@apache.org>
Committed: Wed Jul 11 11:10:06 2018 -0400
----------------------------------------------------------------------
.../metron/management/KafkaFunctions.java | 109 +++++++++++++++++++
.../KafkaFunctionsIntegrationTest.java | 92 ++++++++++++++++
2 files changed, 201 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/7af11b62/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java b/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java
index 7c9c23f..76418b6 100644
--- a/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java
+++ b/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java
@@ -47,6 +47,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -763,6 +764,114 @@ public class KafkaFunctions {
}
/**
+ * KAFKA_SEEK
+ *
+ * <p>Seeks to a specific offset and returns the message.
+ *
+ * <p>Example: Find the message in 'topic', partition 2, offset 1001.
+ * <pre>
+ * {@code
+ * KAFKA_SEEK('topic', 1, 1001)
+ * }
+ * </pre>
+ *
+ * <p>By default, only the message value is returned. By setting the global property
+ * 'stellar.kafka.message.view' = 'rich' the function will return additional Kafka metadata
+ * including the topic, partition, offset, key, and timestamp contained in a map. Setting
+ * this property value to 'simple' or simply not setting the property value, will result
+ * in the default view behavior.
+ */
+ @Stellar(
+ namespace = "KAFKA",
+ name = "SEEK",
+ description = "Seeks to an offset within a topic and returns the message.",
+ params = {
+ "topic - The name of the Kafka topic",
+ "partition - The partition identifier; starts at 0.",
+ "offset - The offset within the partition; starts at 0.",
+ "config - Optional map of key/values that override any global properties."
+ },
+ returns = "The message at the given offset, if the offset exists. Otherwise, returns null."
+ )
+ public static class KafkaSeek implements StellarFunction {
+
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+ // required - the topic, partition, and offset are all required
+ String topic = getArg("topic", 0, String.class, args);
+ int partition = getArg("partition", 1, Integer.class, args);
+ int offset = getArg("offset", 2, Integer.class, args);
+
+ // optional - property overrides provided by the user
+ Map<String, String> overrides = new HashMap<>();
+ if(args.size() > 3) {
+ overrides = getArg("overrides", 3, Map.class, args);
+ }
+
+ Properties properties = buildKafkaProperties(overrides, context);
+ return seek(topic, partition, offset, properties);
+ }
+
+ /**
+ * Find messages in Kafka that satisfy a filter expression.
+ *
+ * @param topic The kafka topic.
+ * @param partition The partition identifier.
+ * @param offset The offset within the given partition.
+ * @param properties Function configuration values.
+ * @return A list of messages that satisfy the filter expression.
+ */
+ private Object seek(String topic, int partition, int offset, Properties properties) {
+ final int pollTimeout = getPollTimeout(properties);
+ final int maxWait = getMaxWait(properties);
+ Object message = null;
+ try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
+
+ // continue until we have the message or exceeded the max wait time
+ long wait = 0L;
+ final long start = clock.currentTimeMillis();
+ while(message == null && wait < maxWait) {
+
+ // seek to the offset
+ TopicPartition topar = new TopicPartition(topic, partition);
+ consumer.assign(Collections.singletonList(topar));
+ consumer.seek(topar, offset);
+
+ // poll kafka for messages
+ for(ConsumerRecord<String, String> record : consumer.poll(pollTimeout)) {
+
+ // kafka will attempt to be helpful and return a message, even if the actual offset does not exist
+ if(record.offset() == offset && record.partition() == partition) {
+ LOG.debug("KAFKA_SEEK found message; topic={}, partition={}, offset={}", topic, partition, offset);
+ message = render(record, properties);
+ }
+ }
+
+ // how long have we waited?
+ wait = clock.currentTimeMillis() - start;
+ if(LOG.isDebugEnabled() && message == null) {
+ LOG.debug("KAFKA_SEEK no message yet; topic={}, partition={}, offset={}, waitTime={} ms",
+ topic, partition, offset, wait);
+ }
+ }
+ }
+
+ return message;
+ }
+
+ @Override
+ public void initialize(Context context) {
+ // no initialization required
+ }
+
+ @Override
+ public boolean isInitialized() {
+ // no initialization required
+ return true;
+ }
+ }
+
+ /**
* Renders the Kafka record into a view.
*
* <p>A user can customize the way in which a Kafka record is rendered by altering
http://git-wip-us.apache.org/repos/asf/metron/blob/7af11b62/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java b/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java
index 5e045ad..09bce17 100644
--- a/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java
+++ b/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java
@@ -113,6 +113,7 @@ public class KafkaFunctionsIntegrationTest extends BaseIntegrationTest {
.withClass(KafkaFunctions.KafkaProps.class)
.withClass(KafkaFunctions.KafkaTail.class)
.withClass(KafkaFunctions.KafkaFind.class)
+ .withClass(KafkaFunctions.KafkaSeek.class)
.withClass(MapFunctions.MapGet.class);
}
@@ -593,6 +594,97 @@ public class KafkaFunctionsIntegrationTest extends BaseIntegrationTest {
}
/**
+ * KAFKA_SEEK should return the message at a given offset.
+ */
+ @Test
+ public void testKafkaSeek() throws Exception {
+
+ // use a unique topic name for this test
+ final String topicName = testName.getMethodName();
+ variables.put("topic", topicName);
+
+ // put 3 messages into the topic
+ run("KAFKA_PUT(topic, [ message1, message2, message3 ])");
+ {
+ // get the 3rd message from the topic
+ Object actual = run("KAFKA_SEEK(topic, 0, 2)");
+ assertEquals(message3, actual);
+ }
+ {
+ // get the 2nd message from the topic
+ Object actual = run("KAFKA_SEEK(topic, 0, 1)");
+ assertEquals(message2, actual);
+ }
+ {
+ // get the 1st message from the topic
+ Object actual = run("KAFKA_SEEK(topic, 0, 0)");
+ assertEquals(message1, actual);
+ }
+ }
+
+ /**
+ * KAFKA_SEEK should return null if the offset does not exist
+ */
+ @Test
+ public void testKafkaSeekToMissingOffset() throws Exception {
+
+ // use a unique topic name for this test
+ final String topicName = testName.getMethodName();
+ variables.put("topic", topicName);
+
+ // put 3 messages into the topic
+ run("KAFKA_PUT(topic, [ message1, message2, message3 ])");
+
+ // get the 3rd message from the topic
+ Object actual = run("KAFKA_SEEK(topic, 0, 9999)");
+ assertNull(actual);
+ }
+
+ /**
+ * KAFKA_SEEK should return null if the partition does not exist
+ */
+ @Test
+ public void testKafkaSeekToMissingPartition() throws Exception {
+
+ // use a unique topic name for this test
+ final String topicName = testName.getMethodName();
+ variables.put("topic", topicName);
+
+ // put 3 messages into the topic
+ run("KAFKA_PUT(topic, [ message1, message2, message3 ])");
+
+ // get the 3rd message from the topic
+ Object actual = run("KAFKA_SEEK(topic, 99999, 0)");
+ assertNull(actual);
+ }
+
+ /**
+ * KAFKA_SEEK should allow a user to see a detailed view of each Kafka record.
+ */
+ @Test
+ public void testKafkaSeekWithRichView() throws Exception {
+
+ // configure a detailed view of each message
+ global.put(KafkaFunctions.MESSAGE_VIEW_PROPERTY, KafkaFunctions.MESSAGE_VIEW_RICH);
+
+ // use a unique topic name for this test
+ final String topicName = testName.getMethodName();
+ variables.put("topic", topicName);
+
+ run("KAFKA_PUT(topic, [ message1, message2, message3 ])");
+ Object actual = run("KAFKA_SEEK(topic, 0, 0)");
+
+ // expect a 'rich' view of the record
+ assertTrue(actual instanceof Map);
+ Map<String, Object> view = (Map) actual;
+ assertNull(view.get("key"));
+ assertNotNull(view.get("offset"));
+ assertEquals(0, view.get("partition"));
+ assertEquals(topicName, view.get("topic"));
+ assertEquals(message1, view.get("value"));
+ }
+
+ /**
* Runs a Stellar expression.
* @param expression The expression to run.
*/
[11/15] metron git commit: METRON-1657 Parser aggregation in storm
(justinleet) closes apache/metron#1099
Posted by mm...@apache.org.
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
index 15ce735..06f4cec 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
@@ -58,6 +58,7 @@ import org.apache.metron.common.zookeeper.configurations.ConfigurationsUpdater;
import org.apache.metron.parsers.BasicParser;
import org.apache.metron.parsers.interfaces.MessageFilter;
import org.apache.metron.parsers.interfaces.MessageParser;
+import org.apache.metron.parsers.topology.ParserComponents;
import org.apache.metron.stellar.dsl.Context;
import org.apache.metron.test.bolt.BaseBoltTest;
import org.apache.metron.test.error.MetronErrorJSONMatcher;
@@ -185,7 +186,15 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testEmpty() throws Exception {
String sensorType = "yaf";
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(writer)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ null,
+ new WriterHandler(writer)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
return ParserBoltTest.createUpdater();
@@ -209,7 +218,7 @@ public class ParserBoltTest extends BaseBoltTest {
MetronError error = new MetronError()
.withErrorType(Constants.ErrorType.PARSER_ERROR)
.withThrowable(new NullPointerException())
- .withSensorType(sensorType)
+ .withSensorType(Collections.singleton(sensorType))
.addRawMessage(sampleBinary);
verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
}
@@ -217,7 +226,15 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testInvalid() throws Exception {
String sensorType = "yaf";
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(writer)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ null,
+ new WriterHandler(writer)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
return ParserBoltTest.createUpdater();
@@ -243,7 +260,7 @@ public class ParserBoltTest extends BaseBoltTest {
MetronError error = new MetronError()
.withErrorType(Constants.ErrorType.PARSER_INVALID)
- .withSensorType(sensorType)
+ .withSensorType(Collections.singleton(sensorType))
.withErrorFields(new HashSet<String>() {{ add("field"); }})
.addRawMessage(new JSONObject(){{
put("field", "invalidValue");
@@ -255,14 +272,20 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void test() throws Exception {
-
String sensorType = "yaf";
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(writer)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ null,
+ new WriterHandler(writer)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
return ParserBoltTest.createUpdater();
}
-
};
parserBolt.setCuratorFramework(client);
parserBolt.setZKCache(cache);
@@ -290,7 +313,6 @@ public class ParserBoltTest extends BaseBoltTest {
when(parser.validate(eq(messages.get(1)))).thenReturn(true);
when(filter.emitTuple(eq(messages.get(0)), any())).thenReturn(false);
when(filter.emitTuple(eq(messages.get(1)), any())).thenReturn(true);
- parserBolt.withMessageFilter(filter);
parserBolt.execute(tuple);
verify(writer, times(1)).write(eq(sensorType), any(ParserWriterConfiguration.class), eq(tuple), eq(finalMessage2));
verify(outputCollector, times(2)).ack(tuple);
@@ -317,21 +339,15 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testFilterSuccess() throws Exception {
String sensorType = "yaf";
-
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(batchWriter)) {
- @Override
- protected SensorParserConfig getSensorParserConfig() {
- try {
- return SensorParserConfig.fromBytes(Bytes.toBytes(sensorParserConfig));
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- @Override
- protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
- return ParserBoltTest.createUpdater(Optional.of(1));
- }
- };
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ null,
+ new WriterHandler(batchWriter)
+ )
+ );
+ ParserBolt parserBolt = buildParserBolt(parserMap, sensorParserConfig);
parserBolt.setCuratorFramework(client);
parserBolt.setZKCache(cache);
@@ -358,10 +374,17 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testFilterFailure() throws Exception {
String sensorType = "yaf";
-
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(batchWriter)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ null,
+ new WriterHandler(batchWriter)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
- protected SensorParserConfig getSensorParserConfig() {
+ protected SensorParserConfig getSensorParserConfig(String sensorType) {
try {
return SensorParserConfig.fromBytes(Bytes.toBytes(sensorParserConfig));
} catch (IOException e) {
@@ -433,21 +456,15 @@ public class ParserBoltTest extends BaseBoltTest {
}
};
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, dummyParser, new WriterHandler(recordingWriter)) {
- @Override
- protected SensorParserConfig getSensorParserConfig() {
- try {
- return SensorParserConfig.fromBytes(Bytes.toBytes(csvWithFieldTransformations));
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
- return ParserBoltTest.createUpdater(Optional.of(1));
- }
- };
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ dummyParser,
+ null,
+ new WriterHandler(recordingWriter)
+ )
+ );
+ ParserBolt parserBolt = buildParserBolt(parserMap, csvWithFieldTransformations);
parserBolt.setCuratorFramework(client);
parserBolt.setZKCache(cache);
@@ -461,10 +478,16 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testDefaultBatchSize() throws Exception {
-
String sensorType = "yaf";
-
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(batchWriter)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ filter,
+ new WriterHandler(batchWriter)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
// this uses default batch size
@@ -487,7 +510,6 @@ public class ParserBoltTest extends BaseBoltTest {
response.addSuccess(uniqueTuples[i]);
}
when(batchWriter.write(eq(sensorType), any(WriterConfiguration.class), eq(new HashSet<>(Arrays.asList(uniqueTuples))), any())).thenReturn(response);
- parserBolt.withMessageFilter(filter);
for (Tuple tuple : uniqueTuples) {
parserBolt.execute(tuple);
}
@@ -498,10 +520,16 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testLessRecordsThanDefaultBatchSize() throws Exception {
-
String sensorType = "yaf";
-
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(batchWriter)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ filter,
+ new WriterHandler(batchWriter)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
// this uses default batch size
@@ -524,7 +552,6 @@ public class ParserBoltTest extends BaseBoltTest {
uniqueTuples[i] = mock(Tuple.class);
response.addSuccess(uniqueTuples[i]);
}
- parserBolt.withMessageFilter(filter);
for (Tuple tuple : uniqueTuples) {
parserBolt.execute(tuple);
}
@@ -542,10 +569,16 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testBatchOfOne() throws Exception {
-
String sensorType = "yaf";
-
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(batchWriter)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ filter,
+ new WriterHandler(batchWriter)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
return ParserBoltTest.createUpdater(Optional.of(1));
@@ -563,17 +596,22 @@ public class ParserBoltTest extends BaseBoltTest {
BulkWriterResponse response = new BulkWriterResponse();
response.addSuccess(t1);
when(batchWriter.write(eq(sensorType), any(WriterConfiguration.class), eq(Collections.singleton(t1)), any())).thenReturn(response);
- parserBolt.withMessageFilter(filter);
parserBolt.execute(t1);
verify(outputCollector, times(1)).ack(t1);
}
@Test
public void testBatchOfFive() throws Exception {
-
String sensorType = "yaf";
-
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(batchWriter)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ filter,
+ new WriterHandler(batchWriter)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
return ParserBoltTest.createUpdater(Optional.of(5));
@@ -592,7 +630,6 @@ public class ParserBoltTest extends BaseBoltTest {
BulkWriterResponse response = new BulkWriterResponse();
response.addAllSuccesses(tuples);
when(batchWriter.write(eq(sensorType), any(WriterConfiguration.class), eq(tuples), any())).thenReturn(response);
- parserBolt.withMessageFilter(filter);
writeNonBatch(outputCollector, parserBolt, t1);
writeNonBatch(outputCollector, parserBolt, t2);
writeNonBatch(outputCollector, parserBolt, t3);
@@ -610,9 +647,16 @@ public class ParserBoltTest extends BaseBoltTest {
@Test
public void testBatchOfFiveWithError() throws Exception {
-
String sensorType = "yaf";
- ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(batchWriter)) {
+ Map<String, ParserComponents> parserMap = Collections.singletonMap(
+ sensorType,
+ new ParserComponents(
+ parser,
+ filter,
+ new WriterHandler(batchWriter)
+ )
+ );
+ ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserMap) {
@Override
protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
return ParserBoltTest.createUpdater(Optional.of(5));
@@ -629,7 +673,6 @@ public class ParserBoltTest extends BaseBoltTest {
when(parser.validate(any())).thenReturn(true);
when(parser.parseOptional(any())).thenReturn(Optional.of(ImmutableList.of(new JSONObject())));
when(filter.emitTuple(any(), any(Context.class))).thenReturn(true);
- parserBolt.withMessageFilter(filter);
parserBolt.execute(t1);
parserBolt.execute(t2);
parserBolt.execute(t3);
@@ -654,6 +697,25 @@ public class ParserBoltTest extends BaseBoltTest {
parserBolt.getConfigurations().updateGlobalConfig(globalConfig);
}
+ private ParserBolt buildParserBolt(Map<String, ParserComponents> parserMap,
+ String csvWithFieldTransformations) {
+ return new ParserBolt("zookeeperUrl", parserMap) {
+ @Override
+ protected SensorParserConfig getSensorParserConfig(String sensorType) {
+ try {
+ return SensorParserConfig.fromBytes(Bytes.toBytes(csvWithFieldTransformations));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected ConfigurationsUpdater<ParserConfigurations> createUpdater() {
+ return ParserBoltTest.createUpdater(Optional.of(1));
+ }
+ };
+ }
+
private static void writeNonBatch(OutputCollector collector, ParserBolt bolt, Tuple t) {
bolt.execute(t);
}
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
index a23c368..b04d8f7 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
@@ -18,6 +18,20 @@
package org.apache.metron.parsers.bolt;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.argThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import org.apache.log4j.Level;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.IndexingConfigurations;
@@ -38,20 +52,6 @@ import org.json.simple.JSONObject;
import org.junit.Test;
import org.mockito.Mock;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.argThat;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
public class WriterBoltTest extends BaseBoltTest{
@Mock
protected TopologyContext topologyContext;
@@ -164,7 +164,7 @@ public class WriterBoltTest extends BaseBoltTest{
MetronError error = new MetronError()
.withErrorType(Constants.ErrorType.DEFAULT_ERROR)
.withThrowable(new IllegalStateException("Unhandled bulk errors in response: {java.lang.Exception: write error=[tuple]}"))
- .withSensorType(sensorType)
+ .withSensorType(Collections.singleton(sensorType))
.addRawMessage(new JSONObject());
verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
}
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java
index ec7c3ab..2cba40a 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java
@@ -17,9 +17,18 @@
*/
package org.apache.metron.parsers.integration;
-import com.google.common.collect.ImmutableList;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
import org.apache.commons.lang.SerializationUtils;
-import org.apache.metron.common.configuration.ConfigurationsUtils;
import org.apache.metron.common.configuration.FieldValidator;
import org.apache.metron.common.configuration.ParserConfigurations;
import org.apache.metron.common.configuration.SensorParserConfig;
@@ -30,29 +39,10 @@ import org.apache.metron.common.writer.MessageWriter;
import org.apache.metron.integration.ProcessorResult;
import org.apache.metron.parsers.bolt.ParserBolt;
import org.apache.metron.parsers.bolt.WriterHandler;
-import org.apache.metron.parsers.interfaces.MessageParser;
-import org.apache.storm.generated.GlobalStreamId;
-import org.apache.storm.task.GeneralTopologyContext;
+import org.apache.metron.parsers.topology.ParserComponents;
import org.apache.storm.task.OutputCollector;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.MessageId;
import org.apache.storm.tuple.Tuple;
-import org.apache.storm.tuple.TupleImpl;
import org.json.simple.JSONObject;
-import org.mockito.Matchers;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import static org.mockito.Matchers.eq;
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,12 +84,20 @@ public class ParserDriver implements Serializable {
public ShimParserBolt(List<byte[]> output) {
super(null
- , sensorType == null?config.getSensorTopic():sensorType
- , ReflectionUtils.createInstance(config.getParserClassName())
- , new WriterHandler( new CollectingWriter(output))
+ , Collections.singletonMap(
+ sensorType == null ? config.getSensorTopic() : sensorType,
+ new ParserComponents(
+ ReflectionUtils.createInstance(config.getParserClassName()),
+ null,
+ new WriterHandler(new CollectingWriter(output))
+ )
+ )
);
this.output = output;
- getParser().configure(config.getParserConfig());
+ Map<String, ParserComponents> sensorToComponentMap = getSensorToComponentMap();
+ for(Entry<String, ParserComponents> sensorToComponents : sensorToComponentMap.entrySet()) {
+ sensorToComponents.getValue().getMessageParser().configure(config.getParserConfig());
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
index 7f40684..15b53b7 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
@@ -17,6 +17,17 @@
*/
package org.apache.metron.parsers.integration.components;
+import static org.apache.metron.integration.components.FluxTopologyComponent.assassinateSlots;
+import static org.apache.metron.integration.components.FluxTopologyComponent.cleanupWorkerDir;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import org.apache.commons.lang3.StringUtils;
import org.apache.metron.integration.InMemoryComponent;
import org.apache.metron.integration.UnableToStartException;
import org.apache.metron.integration.components.ZKServerComponent;
@@ -27,22 +38,13 @@ import org.apache.storm.generated.KillOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-
-import static org.apache.metron.integration.components.FluxTopologyComponent.assassinateSlots;
-import static org.apache.metron.integration.components.FluxTopologyComponent.cleanupWorkerDir;
-
public class ParserTopologyComponent implements InMemoryComponent {
protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private Properties topologyProperties;
private String brokerUrl;
- private String sensorType;
+ private List<String> sensorTypes;
private LocalCluster stormCluster;
private String outputTopic;
private String errorTopic;
@@ -51,7 +53,7 @@ public class ParserTopologyComponent implements InMemoryComponent {
Properties topologyProperties;
String brokerUrl;
- String sensorType;
+ List<String> sensorTypes;
String outputTopic;
String errorTopic;
@@ -63,8 +65,8 @@ public class ParserTopologyComponent implements InMemoryComponent {
this.brokerUrl = brokerUrl;
return this;
}
- public Builder withSensorType(String sensorType) {
- this.sensorType = sensorType;
+ public Builder withSensorTypes(List<String> sensorTypes) {
+ this.sensorTypes = sensorTypes;
return this;
}
@@ -80,7 +82,7 @@ public class ParserTopologyComponent implements InMemoryComponent {
public ParserTopologyComponent build() {
- if(sensorType == null) {
+ if(sensorTypes == null || sensorTypes.isEmpty()) {
throw new IllegalArgumentException("The sensor type must be defined.");
}
@@ -88,20 +90,20 @@ public class ParserTopologyComponent implements InMemoryComponent {
throw new IllegalArgumentException("The output topic must be defined.");
}
- return new ParserTopologyComponent(topologyProperties, brokerUrl, sensorType, outputTopic, errorTopic);
+ return new ParserTopologyComponent(topologyProperties, brokerUrl, sensorTypes, outputTopic, errorTopic);
}
}
- public ParserTopologyComponent(Properties topologyProperties, String brokerUrl, String sensorType, String outputTopic, String errorTopic) {
+ public ParserTopologyComponent(Properties topologyProperties, String brokerUrl, List<String> sensorTypes, String outputTopic, String errorTopic) {
this.topologyProperties = topologyProperties;
this.brokerUrl = brokerUrl;
- this.sensorType = sensorType;
+ this.sensorTypes = sensorTypes;
this.outputTopic = outputTopic;
this.errorTopic = errorTopic;
}
- public void updateSensorType(String sensorType) {
- this.sensorType = sensorType;
+ public void updateSensorTypes(List<String> sensorTypes) {
+ this.sensorTypes = sensorTypes;
}
@Override
@@ -112,14 +114,14 @@ public class ParserTopologyComponent implements InMemoryComponent {
ParserTopologyBuilder.ParserTopology topologyBuilder = ParserTopologyBuilder.build (
topologyProperties.getProperty(ZKServerComponent.ZOOKEEPER_PROPERTY),
Optional.ofNullable(brokerUrl),
- sensorType,
- (x,y) -> 1,
- (x,y) -> 1,
+ sensorTypes,
+ (x,y) -> Collections.nCopies(sensorTypes.size(), 1),
+ (x,y) -> Collections.nCopies(sensorTypes.size(), 1),
(x,y) -> 1,
(x,y) -> 1,
(x,y) -> 1,
(x,y) -> 1,
- (x,y) -> new HashMap<>(),
+ (x,y) -> Collections.nCopies(sensorTypes.size(), new HashMap<>()),
(x,y) -> null,
(x,y) -> outputTopic,
(x,y) -> errorTopic,
@@ -131,9 +133,9 @@ public class ParserTopologyComponent implements InMemoryComponent {
);
stormCluster = new LocalCluster();
- stormCluster.submitTopology(sensorType, stormConf, topologyBuilder.getBuilder().createTopology());
+ stormCluster.submitTopology(getTopologyName(), stormConf, topologyBuilder.getBuilder().createTopology());
} catch (Exception e) {
- throw new UnableToStartException("Unable to start parser topology for sensorType: " + sensorType, e);
+ throw new UnableToStartException("Unable to start parser topology for sensorTypes: " + sensorTypes, e);
}
}
@@ -177,7 +179,7 @@ public class ParserTopologyComponent implements InMemoryComponent {
protected void killTopology() {
KillOptions ko = new KillOptions();
ko.set_wait_secs(0);
- stormCluster.killTopologyWithOpts(sensorType, ko);
+ stormCluster.killTopologyWithOpts(getTopologyName(), ko);
try {
// Actually wait for it to die.
Thread.sleep(2000);
@@ -185,4 +187,8 @@ public class ParserTopologyComponent implements InMemoryComponent {
// Do nothing
}
}
+
+ protected String getTopologyName() {
+ return StringUtils.join(sensorTypes, "__");
+ }
}
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/topology/ParserTopologyCLITest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/topology/ParserTopologyCLITest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/topology/ParserTopologyCLITest.java
index fcfc93b..ae459f4 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/topology/ParserTopologyCLITest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/topology/ParserTopologyCLITest.java
@@ -18,31 +18,34 @@
package org.apache.metron.parsers.topology;
-import com.fasterxml.jackson.core.JsonProcessingException;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.Parser;
+import org.apache.commons.cli.PosixParser;
+import org.apache.commons.io.FileUtils;
import org.apache.log4j.Level;
import org.apache.metron.common.configuration.SensorParserConfig;
import org.apache.metron.common.utils.JSONUtils;
import org.apache.metron.parsers.topology.config.ValueSupplier;
import org.apache.metron.test.utils.UnitTestHelper;
import org.apache.storm.Config;
-import com.google.common.collect.ImmutableMap;
-import org.adrianwalker.multilinestring.Multiline;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.MissingOptionException;
-import org.apache.commons.cli.ParseException;
-import org.apache.commons.cli.PosixParser;
-import org.apache.commons.io.FileUtils;
import org.junit.Assert;
import org.junit.Test;
-import java.io.File;
-import java.io.IOException;
-import java.lang.ref.Reference;
-import java.util.*;
-import java.util.function.Predicate;
-import java.util.function.Supplier;
-
public class ParserTopologyCLITest {
@@ -103,11 +106,11 @@ public class ParserTopologyCLITest {
public void kafkaOffset(boolean longOpt) throws ParseException {
CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
.with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
- .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPE, "mysensor")
+ .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor")
.build(longOpt);
Assert.assertEquals("myzk", ParserTopologyCLI.ParserOptions.ZK_QUORUM.get(cli));
Assert.assertEquals("mybroker", ParserTopologyCLI.ParserOptions.BROKER_URL.get(cli));
- Assert.assertEquals("mysensor", ParserTopologyCLI.ParserOptions.SENSOR_TYPE.get(cli));
+ Assert.assertEquals("mysensor", ParserTopologyCLI.ParserOptions.SENSOR_TYPES.get(cli));
}
@Test
public void testCLI_happyPath() throws ParseException {
@@ -127,11 +130,11 @@ public class ParserTopologyCLITest {
public void happyPath(boolean longOpt) throws ParseException {
CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
.with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
- .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPE, "mysensor")
+ .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor")
.build(longOpt);
Assert.assertEquals("myzk", ParserTopologyCLI.ParserOptions.ZK_QUORUM.get(cli));
Assert.assertEquals("mybroker", ParserTopologyCLI.ParserOptions.BROKER_URL.get(cli));
- Assert.assertEquals("mysensor", ParserTopologyCLI.ParserOptions.SENSOR_TYPE.get(cli));
+ Assert.assertEquals("mysensor", ParserTopologyCLI.ParserOptions.SENSOR_TYPES.get(cli));
}
@Test
@@ -143,7 +146,7 @@ public class ParserTopologyCLITest {
public void testConfig_noExtra(boolean longOpt) throws ParseException {
CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
.with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
- .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPE, "mysensor")
+ .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor")
.with(ParserTopologyCLI.ParserOptions.NUM_WORKERS, "1")
.with(ParserTopologyCLI.ParserOptions.NUM_ACKERS, "2")
.with(ParserTopologyCLI.ParserOptions.NUM_MAX_TASK_PARALLELISM, "3")
@@ -166,7 +169,7 @@ public class ParserTopologyCLITest {
public void testOutputTopic(boolean longOpt) throws ParseException {
CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
.with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
- .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPE, "mysensor")
+ .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor")
.with(ParserTopologyCLI.ParserOptions.OUTPUT_TOPIC, "my_topic")
.build(longOpt);
Assert.assertEquals("my_topic", ParserTopologyCLI.ParserOptions.OUTPUT_TOPIC.get(cli));
@@ -193,7 +196,7 @@ public class ParserTopologyCLITest {
FileUtils.write(extraFile, extraConfig);
CommandLine cli = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
.with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
- .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPE, "mysensor")
+ .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor")
.with(ParserTopologyCLI.ParserOptions.MESSAGE_TIMEOUT, "4")
.with(ParserTopologyCLI.ParserOptions.EXTRA_OPTIONS, extraFile.getAbsolutePath())
.build(longOpt);
@@ -208,50 +211,50 @@ public class ParserTopologyCLITest {
}
private static class ParserInput {
- private Integer spoutParallelism;
- private Integer spoutNumTasks;
+ private List<Integer> spoutParallelism;
+ private List<Integer> spoutNumTasks;
private Integer parserParallelism;
private Integer parserNumTasks;
private Integer errorParallelism;
private Integer errorNumTasks;
- private Map<String, Object> spoutConfig;
+ private List<Map<String, Object>> spoutConfig;
private String securityProtocol;
private Config stormConf;
private String outputTopic;
private String errorTopic;
- public ParserInput(ValueSupplier<Integer> spoutParallelism,
- ValueSupplier<Integer> spoutNumTasks,
+ public ParserInput(ValueSupplier<List> spoutParallelism,
+ ValueSupplier<List> spoutNumTasks,
ValueSupplier<Integer> parserParallelism,
ValueSupplier<Integer> parserNumTasks,
ValueSupplier<Integer> errorParallelism,
ValueSupplier<Integer> errorNumTasks,
- ValueSupplier<Map> spoutConfig,
+ ValueSupplier<List> spoutConfig,
ValueSupplier<String> securityProtocol,
ValueSupplier<Config> stormConf,
ValueSupplier<String> outputTopic,
ValueSupplier<String> errorTopic,
- SensorParserConfig config
+ List<SensorParserConfig> configs
)
{
- this.spoutParallelism = spoutParallelism.get(config, Integer.class);
- this.spoutNumTasks = spoutNumTasks.get(config, Integer.class);
- this.parserParallelism = parserParallelism.get(config, Integer.class);
- this.parserNumTasks = parserNumTasks.get(config, Integer.class);
- this.errorParallelism = errorParallelism.get(config, Integer.class);
- this.errorNumTasks = errorNumTasks.get(config, Integer.class);
- this.spoutConfig = spoutConfig.get(config, Map.class);
- this.securityProtocol = securityProtocol.get(config, String.class);
- this.stormConf = stormConf.get(config, Config.class);
- this.outputTopic = outputTopic.get(config, String.class);
- this.errorTopic = outputTopic.get(config, String.class);
+ this.spoutParallelism = spoutParallelism.get(configs, List.class);
+ this.spoutNumTasks = spoutNumTasks.get(configs, List.class);
+ this.parserParallelism = parserParallelism.get(configs, Integer.class);
+ this.parserNumTasks = parserNumTasks.get(configs, Integer.class);
+ this.errorParallelism = errorParallelism.get(configs, Integer.class);
+ this.errorNumTasks = errorNumTasks.get(configs, Integer.class);
+ this.spoutConfig = spoutConfig.get(configs, List.class);
+ this.securityProtocol = securityProtocol.get(configs, String.class);
+ this.stormConf = stormConf.get(configs, Config.class);
+ this.outputTopic = outputTopic.get(configs, String.class);
+ this.errorTopic = errorTopic.get(configs, String.class);
}
- public Integer getSpoutParallelism() {
+ public List<Integer> getSpoutParallelism() {
return spoutParallelism;
}
- public Integer getSpoutNumTasks() {
+ public List<Integer> getSpoutNumTasks() {
return spoutNumTasks;
}
@@ -271,7 +274,7 @@ public class ParserTopologyCLITest {
return errorNumTasks;
}
- public Map<String, Object> getSpoutConfig() {
+ public List<Map<String, Object>> getSpoutConfig() {
return spoutConfig;
}
@@ -330,43 +333,116 @@ public class ParserTopologyCLITest {
@Test
public void testSpoutParallelism() throws Exception {
testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_PARALLELISM
- , "10"
- , input -> input.getSpoutParallelism().equals(10)
- , () -> {
- SensorParserConfig config = getBaseConfig();
- config.setSpoutParallelism(20);
- return config;
- }
- , input -> input.getSpoutParallelism().equals(20)
- );
+ , "10"
+ , input -> input.getSpoutParallelism().equals(Collections.singletonList(10))
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setSpoutParallelism(20);
+ return Collections.singletonList(config);
+ }
+ , input -> input.getSpoutParallelism().equals(Collections.singletonList(20))
+ );
+ }
+
+ @Test
+ public void testSpoutParallelismMultiple() throws Exception {
+ // Each spout uses it's own
+ // Return one per spout.
+ List<Integer> spoutParCli = new ArrayList<>();
+ spoutParCli.add(10);
+ spoutParCli.add(12);
+ List<Integer> spoutParConfig = new ArrayList<>();
+ spoutParConfig.add(20);
+ spoutParConfig.add(30);
+ testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_PARALLELISM
+ , "10,12"
+ , input -> input.getSpoutParallelism().equals(spoutParCli)
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setSpoutParallelism(20);
+ SensorParserConfig config2 = getBaseConfig();
+ config2.setSpoutParallelism(30);
+ List<SensorParserConfig> configs = new ArrayList<>();
+ configs.add(config);
+ configs.add(config2);
+ return configs;
+ }
+ , input -> input.getSpoutParallelism().equals(spoutParConfig)
+ );
}
@Test
public void testSpoutNumTasks() throws Exception {
testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_NUM_TASKS
, "10"
- , input -> input.getSpoutNumTasks().equals(10)
+ , input -> input.getSpoutNumTasks().equals(Collections.singletonList(10))
, () -> {
SensorParserConfig config = getBaseConfig();
config.setSpoutNumTasks(20);
- return config;
+ return Collections.singletonList(config);
}
- , input -> input.getSpoutNumTasks().equals(20)
+ , input -> input.getSpoutNumTasks().equals(Collections.singletonList(20))
);
}
@Test
+ public void testSpoutNumTasksMultiple() throws Exception {
+ // Return one per spout.
+ List<Integer> numTasksCli = new ArrayList<>();
+ numTasksCli.add(10);
+ numTasksCli.add(12);
+ List<Integer> numTasksConfig = new ArrayList<>();
+ numTasksConfig.add(20);
+ numTasksConfig.add(30);
+ testConfigOption(ParserTopologyCLI.ParserOptions.SPOUT_NUM_TASKS
+ , "10,12"
+ , input -> input.getSpoutNumTasks().equals(numTasksCli)
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setSpoutNumTasks(20);
+ SensorParserConfig config2 = getBaseConfig();
+ config2.setSpoutNumTasks(30);
+ List<SensorParserConfig> configs = new ArrayList<>();
+ configs.add(config);
+ configs.add(config2);
+ return configs;
+ }
+ , input -> input.getSpoutNumTasks().equals(numTasksConfig)
+ );
+ }
+
+ @Test
public void testParserParallelism() throws Exception {
testConfigOption(ParserTopologyCLI.ParserOptions.PARSER_PARALLELISM
- , "10"
- , input -> input.getParserParallelism().equals(10)
- , () -> {
- SensorParserConfig config = getBaseConfig();
- config.setParserParallelism(20);
- return config;
- }
- , input -> input.getParserParallelism().equals(20)
- );
+ , "10"
+ , input -> input.getParserParallelism().equals(10)
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setParserParallelism(20);
+ return Collections.singletonList(config);
+ }
+ , input -> input.getParserParallelism().equals(20)
+ );
+ }
+
+ @Test
+ public void testParserParallelismMultiple() throws Exception {
+ // Last one wins
+ testConfigOption(ParserTopologyCLI.ParserOptions.PARSER_PARALLELISM
+ , "10"
+ , input -> input.getParserParallelism().equals(10)
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setParserParallelism(20);
+ SensorParserConfig config2 = getBaseConfig();
+ config2.setParserParallelism(30);
+ List<SensorParserConfig> configs = new ArrayList<>();
+ configs.add(config);
+ configs.add(config2);
+ return configs;
+ }
+ , input -> input.getParserParallelism().equals(30)
+ );
}
@Test
@@ -377,13 +453,32 @@ public class ParserTopologyCLITest {
, () -> {
SensorParserConfig config = getBaseConfig();
config.setParserNumTasks(20);
- return config;
+ SensorParserConfig config2 = getBaseConfig();
+ config2.setParserNumTasks(30);
+ List<SensorParserConfig> configs = new ArrayList<>();
+ configs.add(config);
+ configs.add(config2);
+ return configs;
}
- , input -> input.getParserNumTasks().equals(20)
+ , input -> input.getParserNumTasks().equals(30)
);
}
@Test
+ public void testParserNumTasksMultiple() throws Exception {
+ testConfigOption(ParserTopologyCLI.ParserOptions.PARSER_NUM_TASKS
+ , "10"
+ , input -> input.getParserNumTasks().equals(10)
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setParserNumTasks(20);
+ return Collections.singletonList(config);
+ }
+ , input -> input.getParserNumTasks().equals(20)
+ );
+ }
+
+ @Test
public void testErrorParallelism() throws Exception {
testConfigOption(ParserTopologyCLI.ParserOptions.ERROR_WRITER_PARALLELISM
, "10"
@@ -391,7 +486,7 @@ public class ParserTopologyCLITest {
, () -> {
SensorParserConfig config = getBaseConfig();
config.setErrorWriterParallelism(20);
- return config;
+ return Collections.singletonList(config);
}
, input -> input.getErrorParallelism().equals(20)
);
@@ -405,7 +500,7 @@ public class ParserTopologyCLITest {
, () -> {
SensorParserConfig config = getBaseConfig();
config.setErrorWriterNumTasks(20);
- return config;
+ return Collections.singletonList(config);
}
, input -> input.getErrorNumTasks().equals(20)
);
@@ -419,13 +514,55 @@ public class ParserTopologyCLITest {
, () -> {
SensorParserConfig config = getBaseConfig();
config.setSecurityProtocol("KERBEROS");
- return config;
+ return Collections.singletonList(config);
}
, input -> input.getSecurityProtocol().equals("KERBEROS")
);
}
@Test
+ public void testSecurityProtocol_fromCLIMultipleUniform() throws Exception {
+ testConfigOption(ParserTopologyCLI.ParserOptions.SECURITY_PROTOCOL
+ , "PLAINTEXT"
+ , input -> input.getSecurityProtocol().equals("PLAINTEXT")
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setSecurityProtocol("PLAINTEXT");
+ SensorParserConfig config2 = getBaseConfig();
+ config2.setSecurityProtocol("PLAINTEXT");
+ List<SensorParserConfig> configs = new ArrayList<>();
+ configs.add(config);
+ configs.add(config2);
+ return configs;
+ }
+ , input -> input.getSecurityProtocol().equals("PLAINTEXT")
+ );
+ }
+
+ @Test
+ public void testSecurityProtocol_fromCLIMultipleMixed() throws Exception {
+ // Non plaintext wins
+ testConfigOption(ParserTopologyCLI.ParserOptions.SECURITY_PROTOCOL
+ , "PLAINTEXT"
+ , input -> input.getSecurityProtocol().equals("PLAINTEXT")
+ , () -> {
+ SensorParserConfig config = getBaseConfig();
+ config.setSecurityProtocol("PLAINTEXT");
+ SensorParserConfig config2 = getBaseConfig();
+ config2.setSecurityProtocol("KERBEROS");
+ SensorParserConfig config3 = getBaseConfig();
+ config3.setSecurityProtocol("PLAINTEXT");
+ List<SensorParserConfig> configs = new ArrayList<>();
+ configs.add(config);
+ configs.add(config2);
+ configs.add(config3);
+ return configs;
+ }
+ , input -> input.getSecurityProtocol().equals("KERBEROS")
+ );
+ }
+
+ @Test
public void testSecurityProtocol_fromSpout() throws Exception {
//Ultimately the order of precedence is CLI > spout config > parser config
File extraConfig = File.createTempFile("spoutConfig", "json");
@@ -444,7 +581,7 @@ public class ParserTopologyCLITest {
, () -> {
SensorParserConfig config = getBaseConfig();
config.setSecurityProtocol("PLAINTEXTSASL_FROM_ZK");
- return config;
+ return Collections.singletonList(config);
}
, input -> input.getSecurityProtocol().equals("PLAINTEXTSASL_FROM_ZK")
);
@@ -458,7 +595,7 @@ public class ParserTopologyCLITest {
, () -> {
SensorParserConfig config = getBaseConfig();
config.setSecurityProtocol("PLAINTEXTSASL_FROM_ZK");
- return config;
+ return Collections.singletonList(config);
}
, input -> input.getSecurityProtocol().equals("PLAINTEXTSASL_FROM_ZK")
);
@@ -481,7 +618,7 @@ public class ParserTopologyCLITest {
SensorParserConfig config = getBaseConfig();
config.setNumWorkers(100);
config.setNumAckers(200);
- return config;
+ return Collections.singletonList(config);
}
, input -> {
Config c = input.getStormConf();
@@ -519,7 +656,7 @@ public class ParserTopologyCLITest {
put(Config.TOPOLOGY_ACKER_EXECUTORS, 200);
}}
);
- return config;
+ return Collections.singletonList(config);
}
, input -> {
Config c = input.getStormConf();
@@ -542,22 +679,21 @@ public class ParserTopologyCLITest {
put(ParserTopologyCLI.ParserOptions.SPOUT_CONFIG, extraConfig.getAbsolutePath());
}};
Predicate<ParserInput> cliOverrideExpected = input -> {
- return input.getSpoutConfig().get("extra_config").equals("from_file");
+ return input.getSpoutConfig().get(0).get("extra_config").equals("from_file");
};
Predicate<ParserInput> configOverrideExpected = input -> {
- return input.getSpoutConfig().get("extra_config").equals("from_zk")
- ;
+ return input.getSpoutConfig().get(0).get("extra_config").equals("from_zk");
};
- Supplier<SensorParserConfig> configSupplier = () -> {
+ Supplier<List<SensorParserConfig>> configSupplier = () -> {
SensorParserConfig config = getBaseConfig();
config.setSpoutConfig(
new HashMap<String, Object>() {{
put("extra_config", "from_zk");
}}
);
- return config;
+ return Collections.singletonList(config);
};
testConfigOption( cliOptions
, cliOverrideExpected
@@ -573,7 +709,7 @@ public class ParserTopologyCLITest {
private void testConfigOption( ParserTopologyCLI.ParserOptions option
, String cliOverride
, Predicate<ParserInput> cliOverrideCondition
- , Supplier<SensorParserConfig> configSupplier
+ , Supplier<List<SensorParserConfig>> configSupplier
, Predicate<ParserInput> configOverrideCondition
) throws Exception {
testConfigOption(
@@ -588,48 +724,48 @@ public class ParserTopologyCLITest {
private void testConfigOption( EnumMap<ParserTopologyCLI.ParserOptions, String> options
, Predicate<ParserInput> cliOverrideCondition
- , Supplier<SensorParserConfig> configSupplier
+ , Supplier<List<SensorParserConfig>> configSupplier
, Predicate<ParserInput> configOverrideCondition
) throws Exception {
//CLI Override
- SensorParserConfig config = configSupplier.get();
+ List<SensorParserConfig> configs = configSupplier.get();
{
CLIBuilder builder = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
.with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
- .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPE, "mysensor");
+ .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor");
for(Map.Entry<ParserTopologyCLI.ParserOptions, String> entry : options.entrySet()) {
builder.with(entry.getKey(), entry.getValue());
}
CommandLine cmd = builder.build(true);
- ParserInput input = getInput(cmd, config);
+ ParserInput input = getInput(cmd, configs);
Assert.assertTrue(cliOverrideCondition.test(input));
}
// Config Override
{
CLIBuilder builder = new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
.with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
- .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPE, "mysensor");
+ .with(ParserTopologyCLI.ParserOptions.SENSOR_TYPES, "mysensor");
CommandLine cmd = builder.build(true);
- ParserInput input = getInput(cmd, config);
+ ParserInput input = getInput(cmd, configs);
Assert.assertTrue(configOverrideCondition.test(input));
}
}
- private static ParserInput getInput(CommandLine cmd, SensorParserConfig config ) throws Exception {
+ private static ParserInput getInput(CommandLine cmd, List<SensorParserConfig> configs ) throws Exception {
final ParserInput[] parserInput = new ParserInput[]{null};
new ParserTopologyCLI() {
@Override
protected ParserTopologyBuilder.ParserTopology getParserTopology(
String zookeeperUrl,
Optional<String> brokerUrl,
- String sensorType,
- ValueSupplier<Integer> spoutParallelism,
- ValueSupplier<Integer> spoutNumTasks,
+ List<String> sensorType,
+ ValueSupplier<List> spoutParallelism,
+ ValueSupplier<List> spoutNumTasks,
ValueSupplier<Integer> parserParallelism,
ValueSupplier<Integer> parserNumTasks,
ValueSupplier<Integer> errorParallelism,
ValueSupplier<Integer> errorNumTasks,
- ValueSupplier<Map> spoutConfig,
+ ValueSupplier<List> spoutConfig,
ValueSupplier<String> securityProtocol,
ValueSupplier<Config> stormConf,
ValueSupplier<String> outputTopic,
@@ -647,7 +783,7 @@ public class ParserTopologyCLITest {
stormConf,
outputTopic,
errorTopic,
- config
+ configs
);
return null;
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
index 49d7521..788df2d 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
@@ -19,6 +19,7 @@
package org.apache.metron.writers.integration;
import com.google.common.collect.ImmutableList;
+import java.util.Collections;
import org.adrianwalker.multilinestring.Multiline;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
@@ -107,7 +108,7 @@ public class SimpleHbaseEnrichmentWriterIntegrationTest extends BaseIntegrationT
.withParserSensorConfig(sensorType, parserConfig);
ParserTopologyComponent parserTopologyComponent = new ParserTopologyComponent.Builder()
- .withSensorType(sensorType)
+ .withSensorTypes(Collections.singletonList(sensorType))
.withTopologyProperties(topologyProperties)
.withBrokerUrl(kafkaComponent.getBrokerList())
.withOutputTopic(parserConfig.getOutputTopic())
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
index 99506de..cecba3d 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
@@ -27,6 +27,7 @@ import com.google.common.collect.Iterables;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -95,6 +96,7 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
* "sensorTopic": "dummy",
* "outputTopic": "output",
* "errorTopic": "parser_error",
+ * "readMetadata": true,
* "parserConfig": {
* "batchSize" : 1,
* "columns" : {
@@ -148,7 +150,12 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
}};
final Properties topologyProperties = new Properties();
- ComponentRunner runner = setupTopologyComponents(topologyProperties, sensorType, parserConfig, globalConfigWithValidation);
+ ComponentRunner runner = setupTopologyComponents(
+ topologyProperties,
+ Collections.singletonList(sensorType),
+ Collections.singletonList(parserConfig),
+ globalConfigWithValidation
+ );
try {
runner.start();
kafkaComponent.writeMessages(sensorType, inputMessages);
@@ -172,7 +179,7 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
@Test
public void parser_with_global_validations_writes_bad_records_to_error_topic() throws Exception {
- final String sensorType = "dummy";
+ final String sensorType = "dummy";
SensorParserConfig parserConfig = JSONUtils.INSTANCE.load(parserConfigJSON, SensorParserConfig.class);
final List<byte[]> inputMessages = new ArrayList<byte[]>() {{
add(Bytes.toBytes("valid,foo"));
@@ -181,7 +188,8 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
}};
final Properties topologyProperties = new Properties();
- ComponentRunner runner = setupTopologyComponents(topologyProperties, sensorType, parserConfig, globalConfigWithValidation);
+ ComponentRunner runner = setupTopologyComponents(topologyProperties, Collections.singletonList(sensorType),
+ Collections.singletonList(parserConfig), globalConfigWithValidation);
try {
runner.start();
kafkaComponent.writeMessages(sensorType, inputMessages);
@@ -223,27 +231,31 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
*
* @return runner
*/
- public ComponentRunner setupTopologyComponents(Properties topologyProperties, String sensorType,
- SensorParserConfig parserConfig, String globalConfig) {
+ public ComponentRunner setupTopologyComponents(Properties topologyProperties, List<String> sensorTypes,
+ List<SensorParserConfig> parserConfigs, String globalConfig) {
zkServerComponent = getZKServerComponent(topologyProperties);
- kafkaComponent = getKafkaComponent(topologyProperties, new ArrayList<KafkaComponent.Topic>() {{
- add(new KafkaComponent.Topic(sensorType, 1));
- add(new KafkaComponent.Topic(parserConfig.getErrorTopic(), 1));
- add(new KafkaComponent.Topic(Constants.ENRICHMENT_TOPIC, 1));
- }});
+ List<KafkaComponent.Topic> topics = new ArrayList<>();
+ for(String sensorType : sensorTypes) {
+ topics.add(new KafkaComponent.Topic(sensorType, 1));
+ }
+ topics.add(new KafkaComponent.Topic(Constants.ENRICHMENT_TOPIC, 1));
+ kafkaComponent = getKafkaComponent(topologyProperties, topics);
topologyProperties.setProperty("kafka.broker", kafkaComponent.getBrokerList());
configUploadComponent = new ConfigUploadComponent()
.withTopologyProperties(topologyProperties)
- .withGlobalConfig(globalConfig)
- .withParserSensorConfig(sensorType, parserConfig);
+ .withGlobalConfig(globalConfig);
+
+ for (int i = 0; i < sensorTypes.size(); ++i) {
+ configUploadComponent.withParserSensorConfig(sensorTypes.get(i), parserConfigs.get(i));
+ }
parserTopologyComponent = new ParserTopologyComponent.Builder()
- .withSensorType(sensorType)
+ .withSensorTypes(sensorTypes)
.withTopologyProperties(topologyProperties)
.withBrokerUrl(kafkaComponent.getBrokerList())
- .withErrorTopic(parserConfig.getErrorTopic())
- .withOutputTopic(parserConfig.getOutputTopic())
+ .withErrorTopic(parserConfigs.get(0).getErrorTopic())
+ .withOutputTopic(parserConfigs.get(0).getOutputTopic())
.build();
return new ComponentRunner.Builder()
@@ -325,8 +337,22 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
@Multiline
public static String offsetParserConfigJSON;
+ /**
+ * {
+ * "parserClassName":"org.apache.metron.writers.integration.WriterBoltIntegrationTest$DummyObjectParser",
+ * "sensorTopic":"dummyobjectparser",
+ * "outputTopic": "enrichments",
+ * "errorTopic": "parser_error",
+ * "parserConfig": {
+ * "batchSize" : 1
+ * }
+ * }
+ */
+ @Multiline
+ public static String dummyParserConfigJSON;
+
@Test
- public void commits_kafka_offsets_for_emtpy_objects() throws Exception {
+ public void commits_kafka_offsets_for_empty_objects() throws Exception {
final String sensorType = "emptyobjectparser";
SensorParserConfig parserConfig = JSONUtils.INSTANCE.load(offsetParserConfigJSON, SensorParserConfig.class);
final List<byte[]> inputMessages = new ArrayList<byte[]>() {{
@@ -335,7 +361,11 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
add(Bytes.toBytes("baz"));
}};
final Properties topologyProperties = new Properties();
- ComponentRunner runner = setupTopologyComponents(topologyProperties, sensorType, parserConfig, globalConfigEmpty);
+ ComponentRunner runner = setupTopologyComponents(
+ topologyProperties,
+ Collections.singletonList(sensorType),
+ Collections.singletonList(parserConfig),
+ globalConfigEmpty);
try {
runner.start();
kafkaComponent.writeMessages(sensorType, inputMessages);
@@ -356,6 +386,64 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
}
}
+ @Test
+ public void test_multiple_sensors() throws Exception {
+ // Setup first sensor
+ final String emptyObjectSensorType = "emptyobjectparser";
+ SensorParserConfig emptyObjectParserConfig = JSONUtils.INSTANCE.load(offsetParserConfigJSON, SensorParserConfig.class);
+ final List<byte[]> emptyObjectInputMessages = new ArrayList<byte[]>() {{
+ add(Bytes.toBytes("foo"));
+ add(Bytes.toBytes("bar"));
+ add(Bytes.toBytes("baz"));
+ }};
+
+ // Setup second sensor
+ final String dummySensorType = "dummyobjectparser";
+ SensorParserConfig dummyParserConfig = JSONUtils.INSTANCE.load(dummyParserConfigJSON, SensorParserConfig.class);
+ final List<byte[]> dummyInputMessages = new ArrayList<byte[]>() {{
+ add(Bytes.toBytes("dummy_foo"));
+ add(Bytes.toBytes("dummy_bar"));
+ add(Bytes.toBytes("dummy_baz"));
+ }};
+
+ final Properties topologyProperties = new Properties();
+
+ List<String> sensorTypes = new ArrayList<>();
+ sensorTypes.add(emptyObjectSensorType);
+ sensorTypes.add(dummySensorType);
+
+ List<SensorParserConfig> parserConfigs = new ArrayList<>();
+ parserConfigs.add(emptyObjectParserConfig);
+ parserConfigs.add(dummyParserConfig);
+
+ ComponentRunner runner = setupTopologyComponents(topologyProperties, sensorTypes, parserConfigs, globalConfigEmpty);
+ try {
+ runner.start();
+ kafkaComponent.writeMessages(emptyObjectSensorType, emptyObjectInputMessages);
+ kafkaComponent.writeMessages(dummySensorType, dummyInputMessages);
+
+ final List<byte[]> allInputMessages = new ArrayList<>();
+ allInputMessages.addAll(emptyObjectInputMessages);
+ allInputMessages.addAll(dummyInputMessages);
+ Processor allResultsProcessor = new AllResultsProcessor(allInputMessages, Constants.ENRICHMENT_TOPIC);
+ @SuppressWarnings("unchecked")
+ ProcessorResult<Set<JSONObject>> result = runner.process(allResultsProcessor);
+
+ // validate the output messages
+ assertThat(
+ "size should match",
+ result.getResult().size(),
+ equalTo(allInputMessages.size()));
+ for (JSONObject record : result.getResult()) {
+ assertThat("record should have a guid", record.containsKey("guid"), equalTo(true));
+ }
+ } finally {
+ if (runner != null) {
+ runner.stop();
+ }
+ }
+ }
+
/**
* Goal is to check returning an empty JSONObject in our List returned by parse.
*/
@@ -380,6 +468,34 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
}
}
+
+ /**
+ * Goal is to check returning an empty JSONObject in our List returned by parse.
+ */
+ public static class DummyObjectParser implements MessageParser<JSONObject>, Serializable {
+
+ @Override
+ public void init() {
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<JSONObject> parse(byte[] bytes) {
+ JSONObject dummy = new JSONObject();
+ dummy.put("dummy_key", "dummy_value");
+ return ImmutableList.of(dummy);
+ }
+
+ @Override
+ public boolean validate(JSONObject message) {
+ return true;
+ }
+
+ @Override
+ public void configure(Map<String, Object> map) {
+ }
+ }
+
/**
* Verifies all messages in the provided List of input messages appears in the specified
* Kafka output topic
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
index e35960f..7678584 100644
--- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
+++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
@@ -23,6 +23,7 @@ import static java.lang.String.format;
import com.google.common.collect.Iterables;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -118,7 +119,7 @@ public class BulkWriterComponent<MESSAGE_T> {
public void error(String sensorType, Throwable e, Iterable<Tuple> tuples, MessageGetStrategy messageGetStrategy) {
LOG.error(format("Failing %d tuple(s); sensorType=%s", Iterables.size(tuples), sensorType), e);
MetronError error = new MetronError()
- .withSensorType(sensorType)
+ .withSensorType(Collections.singleton(sensorType))
.withErrorType(Constants.ErrorType.INDEXING_ERROR)
.withThrowable(e);
tuples.forEach(t -> error.addRawMessage(messageGetStrategy.get(t)));
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java
index 0264b3d..c389854 100644
--- a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java
+++ b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java
@@ -17,11 +17,22 @@
*/
package org.apache.metron.writer;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+import static org.powermock.api.mockito.PowerMockito.verifyStatic;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.writer.WriterConfiguration;
import org.apache.metron.common.error.MetronError;
import org.apache.metron.common.message.MessageGetStrategy;
-import org.apache.metron.common.message.MessageGetters;
import org.apache.metron.common.utils.ErrorUtils;
import org.apache.metron.common.writer.BulkMessageWriter;
import org.apache.metron.common.writer.BulkWriterResponse;
@@ -38,19 +49,6 @@ import org.mockito.MockitoAnnotations;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static org.powermock.api.mockito.PowerMockito.mockStatic;
-import static org.powermock.api.mockito.PowerMockito.verifyStatic;
-
@RunWith(PowerMockRunner.class)
@PrepareForTest({BulkWriterComponent.class, ErrorUtils.class})
public class BulkWriterComponentTest {
@@ -130,7 +128,7 @@ public class BulkWriterComponentTest {
public void writeShouldProperlyHandleWriterErrors() throws Exception {
Throwable e = new Exception("test exception");
MetronError error = new MetronError()
- .withSensorType(sensorType)
+ .withSensorType(Collections.singleton(sensorType))
.withErrorType(Constants.ErrorType.INDEXING_ERROR).withThrowable(e).withRawMessages(Arrays.asList(message1, message2));
BulkWriterResponse response = new BulkWriterResponse();
response.addAllErrors(e, tupleList);
@@ -164,7 +162,7 @@ public class BulkWriterComponentTest {
public void writeShouldProperlyHandleWriterException() throws Exception {
Throwable e = new Exception("test exception");
MetronError error = new MetronError()
- .withSensorType(sensorType)
+ .withSensorType(Collections.singleton(sensorType))
.withErrorType(Constants.ErrorType.INDEXING_ERROR).withThrowable(e).withRawMessages(Arrays.asList(message1, message2));
BulkWriterResponse response = new BulkWriterResponse();
response.addAllErrors(e, tupleList);
@@ -183,10 +181,10 @@ public class BulkWriterComponentTest {
public void errorAllShouldClearMapsAndHandleErrors() throws Exception {
Throwable e = new Exception("test exception");
MetronError error1 = new MetronError()
- .withSensorType("sensor1")
+ .withSensorType(Collections.singleton("sensor1"))
.withErrorType(Constants.ErrorType.INDEXING_ERROR).withThrowable(e).withRawMessages(Collections.singletonList(message1));
MetronError error2 = new MetronError()
- .withSensorType("sensor2")
+ .withSensorType(Collections.singleton("sensor2"))
.withErrorType(Constants.ErrorType.INDEXING_ERROR).withThrowable(e).withRawMessages(Collections.singletonList(message2));
BulkWriterComponent<JSONObject> bulkWriterComponent = new BulkWriterComponent<>(collector);
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/use-cases/parser_chaining/README.md
----------------------------------------------------------------------
diff --git a/use-cases/parser_chaining/README.md b/use-cases/parser_chaining/README.md
index 26fd333..4055bcd 100644
--- a/use-cases/parser_chaining/README.md
+++ b/use-cases/parser_chaining/README.md
@@ -233,3 +233,17 @@ cat ~/data.log | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --b
```
You should see indices created for the `cisco-5-304` and `cisco-6-302` data with appropriate fields created for each type.
+
+# Aggregated Parsers with Parser Chaining
+Chained parsers can be run as aggregated parsers. These parsers continue to use the sensor specific Kafka topics, and do not do internal routing to the appropriate sensor.
+
+Instead of creating a topology per sensor, all 3 (`pix-syslog-parser`, `cisco-5-304`, and `cisco-6-302`) can be run in a single aggregated parser. It's also possible to aggregate a subset of these parsers (e.g. run `cisco-6-302` as it's own topology, and aggregate the other 2).
+
+The step to start parsers then becomes
+```
+$METRON_HOME/bin/start_parser_topology.sh -k $BROKERLIST -z $ZOOKEEPER -s cisco-6-302,cisco-5-304,pix_syslog_router
+```
+
+The flow through the Storm topology and Kafka topics:
+
+
\ No newline at end of file
[13/15] metron git commit: METRON-1684 Fix Markdown problems in
3rdPartyParser.md (justinleet) closes apache/metron#1110
Posted by mm...@apache.org.
METRON-1684 Fix Markdown problems in 3rdPartyParser.md (justinleet) closes apache/metron#1110
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/52bd310f
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/52bd310f
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/52bd310f
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 52bd310fcce68dad15eead57f1113092a30d9791
Parents: f434538
Author: justinleet <ju...@gmail.com>
Authored: Wed Jul 18 13:18:49 2018 -0400
Committer: leet <le...@apache.org>
Committed: Wed Jul 18 13:18:49 2018 -0400
----------------------------------------------------------------------
.../metron-parsers/3rdPartyParser.md | 408 +++++++++----------
1 file changed, 204 insertions(+), 204 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/52bd310f/metron-platform/metron-parsers/3rdPartyParser.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/3rdPartyParser.md b/metron-platform/metron-parsers/3rdPartyParser.md
index 739328f..0c2409e 100644
--- a/metron-platform/metron-parsers/3rdPartyParser.md
+++ b/metron-platform/metron-parsers/3rdPartyParser.md
@@ -54,180 +54,180 @@ project. We'll call it `extra_parsers`, so in your workspace, let's set
up the maven project:
* Create the maven infrastructure for `extra_parsers` via
- ```
- mkdir -p extra_parsers/src/{main,test}/java
- ```
+ ```
+ mkdir -p extra_parsers/src/{main,test}/java
+ ```
* Create a pom file indicating how we should build our parsers by
editing `extra_parsers/pom.xml` with the following content:
- ```
- <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <groupId>com.3rdparty</groupId>
- <artifactId>extra-parsers</artifactId>
- <packaging>jar</packaging>
- <version>1.0-SNAPSHOT</version>
- <name>extra-parsers</name>
- <url>http://thirdpartysoftware.org</url>
- <properties>
- <!-- The java version to conform to. Metron works all the way to 1.8 -->
- <java_version>1.8</java_version>
- <!-- The version of Metron that we'll be targetting. -->
- <metron_version>0.4.1</metron_version>
- <!-- To complete the simulation, we'll depend on a common dependency -->
- <guava_version>19.0</guava_version>
- <!-- We will shade our dependencies to create a single jar at the end -->
- <shade_version>2.4.3</shade_version>
- </properties>
- <dependencies>
- <!--
- We want to depend on Metron, but ensure that the scope is "provided"
- as we do not want to include it in our bundle.
- -->
- <dependency>
- <groupId>org.apache.metron</groupId>
- <artifactId>metron-parsers</artifactId>
- <version>${metron_version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava_version}</version>
- </dependency>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <version>3.8.1</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- <build>
- <plugins>
- <!-- We will set up the shade plugin to create a single jar at the
- end of the build lifecycle. We will exclude some things and
- relocate others to simulate a real situation.
-
- One thing to note is that it's a good practice to shade and
- relocate common libraries that may be dependencies in Metron.
- Your jar will be merged with the parsers jar, so the metron
- version will be included for all overlapping classes.
- So, shade and relocate to ensure that YOUR version of the library is used.
+ ```
+ <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>com.3rdparty</groupId>
+ <artifactId>extra-parsers</artifactId>
+ <packaging>jar</packaging>
+ <version>1.0-SNAPSHOT</version>
+ <name>extra-parsers</name>
+ <url>http://thirdpartysoftware.org</url>
+ <properties>
+ <!-- The java version to conform to. Metron works all the way to 1.8 -->
+ <java_version>1.8</java_version>
+ <!-- The version of Metron that we'll be targetting. -->
+ <metron_version>0.4.1</metron_version>
+ <!-- To complete the simulation, we'll depend on a common dependency -->
+ <guava_version>19.0</guava_version>
+ <!-- We will shade our dependencies to create a single jar at the end -->
+ <shade_version>2.4.3</shade_version>
+ </properties>
+ <dependencies>
+ <!--
+ We want to depend on Metron, but ensure that the scope is "provided"
+ as we do not want to include it in our bundle.
-->
-
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <version>${shade_version}</version>
- <configuration>
- <createDependencyReducedPom>true</createDependencyReducedPom>
- <artifactSet>
- <excludes>
- <!-- Exclude slf4j for no reason other than to illustrate how to exclude dependencies.
- The metron team has nothing against slf4j. :-)
- -->
- <exclude>*slf4j*</exclude>
- </excludes>
- </artifactSet>
- </configuration>
- <executions>
- <execution>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- <configuration>
- <shadedArtifactAttached>true</shadedArtifactAttached>
- <shadedClassifierName>uber</shadedClassifierName>
- <filters>
- <filter>
- <!-- Sometimes these get added and confuse the uber jar out of shade -->
- <artifact>*:*</artifact>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-parsers</artifactId>
+ <version>${metron_version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${guava_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>3.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <!-- We will set up the shade plugin to create a single jar at the
+ end of the build lifecycle. We will exclude some things and
+ relocate others to simulate a real situation.
+
+ One thing to note is that it's a good practice to shade and
+ relocate common libraries that may be dependencies in Metron.
+ Your jar will be merged with the parsers jar, so the metron
+ version will be included for all overlapping classes.
+ So, shade and relocate to ensure that YOUR version of the library is used.
+ -->
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${shade_version}</version>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <artifactSet>
+ <excludes>
+ <!-- Exclude slf4j for no reason other than to illustrate how to exclude dependencies.
+ The metron team has nothing against slf4j. :-)
+ -->
+ <exclude>*slf4j*</exclude>
+ </excludes>
+ </artifactSet>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadedArtifactAttached>true</shadedArtifactAttached>
+ <shadedClassifierName>uber</shadedClassifierName>
+ <filters>
+ <filter>
+ <!-- Sometimes these get added and confuse the uber jar out of shade -->
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <relocations>
+ <!-- Relocate guava as it's used in Metron and I really want 0.19 -->
+ <relocation>
+ <pattern>com.google</pattern>
+ <shadedPattern>com.thirdparty.guava</shadedPattern>
+ </relocation>
+ </relocations>
+ <artifactSet>
<excludes>
- <exclude>META-INF/*.SF</exclude>
- <exclude>META-INF/*.DSA</exclude>
- <exclude>META-INF/*.RSA</exclude>
+ <!-- We can also exclude by artifactId and groupId -->
+ <exclude>storm:storm-core:*</exclude>
+ <exclude>storm:storm-lib:*</exclude>
+ <exclude>org.slf4j.impl*</exclude>
+ <exclude>org.slf4j:slf4j-log4j*</exclude>
</excludes>
- </filter>
- </filters>
- <relocations>
- <!-- Relocate guava as it's used in Metron and I really want 0.19 -->
- <relocation>
- <pattern>com.google</pattern>
- <shadedPattern>com.thirdparty.guava</shadedPattern>
- </relocation>
- </relocations>
- <artifactSet>
- <excludes>
- <!-- We can also exclude by artifactId and groupId -->
- <exclude>storm:storm-core:*</exclude>
- <exclude>storm:storm-lib:*</exclude>
- <exclude>org.slf4j.impl*</exclude>
- <exclude>org.slf4j:slf4j-log4j*</exclude>
- </excludes>
- </artifactSet>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <!--
- We want to make sure we compile using java 1.8.
- -->
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <version>3.5.1</version>
- <configuration>
- <forceJavacCompilerUse>true</forceJavacCompilerUse>
- <source>${java_version}</source>
- <compilerArgument>-Xlint:unchecked</compilerArgument>
- <target>${java_version}</target>
- <showWarnings>true</showWarnings>
- </configuration>
- </plugin>
- </plugins>
- </build>
- </project>
- ```
+ </artifactSet>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!--
+ We want to make sure we compile using java 1.8.
+ -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.5.1</version>
+ <configuration>
+ <forceJavacCompilerUse>true</forceJavacCompilerUse>
+ <source>${java_version}</source>
+ <compilerArgument>-Xlint:unchecked</compilerArgument>
+ <target>${java_version}</target>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </project>
+ ```
* Now let's create our parser `com.thirdparty.SimpleParser` by creating the file `extra-parsers/src/main/java/com/thirdparty/SimpleParser.java` with the following content:
- ```
- package com.thirdparty;
-
- import com.google.common.base.Splitter;
- import com.google.common.collect.ImmutableList;
- import com.google.common.collect.Iterables;
- import org.apache.metron.parsers.BasicParser;
- import org.json.simple.JSONObject;
-
- import java.util.List;
- import java.util.Map;
-
- public class SimpleParser extends BasicParser {
- @Override
- public void init() {
-
- }
-
- @Override
- public List<JSONObject> parse(byte[] bytes) {
- String input = new String(bytes);
- Iterable<String> it = Splitter.on(",").split(input);
- JSONObject ret = new JSONObject();
- ret.put("original_string", input);
- ret.put("timestamp", System.currentTimeMillis());
- ret.put("first", Iterables.getFirst(it, "missing"));
- ret.put("last", Iterables.getLast(it, "missing"));
- return ImmutableList.of(ret);
+ ```
+ package com.thirdparty;
+
+ import com.google.common.base.Splitter;
+ import com.google.common.collect.ImmutableList;
+ import com.google.common.collect.Iterables;
+ import org.apache.metron.parsers.BasicParser;
+ import org.json.simple.JSONObject;
+
+ import java.util.List;
+ import java.util.Map;
+
+ public class SimpleParser extends BasicParser {
+ @Override
+ public void init() {
+
+ }
+
+ @Override
+ public List<JSONObject> parse(byte[] bytes) {
+ String input = new String(bytes);
+ Iterable<String> it = Splitter.on(",").split(input);
+ JSONObject ret = new JSONObject();
+ ret.put("original_string", input);
+ ret.put("timestamp", System.currentTimeMillis());
+ ret.put("first", Iterables.getFirst(it, "missing"));
+ ret.put("last", Iterables.getLast(it, "missing"));
+ return ImmutableList.of(ret);
+ }
+
+ @Override
+ public void configure(Map<String, Object> map) {
+
+ }
}
-
- @Override
- public void configure(Map<String, Object> map) {
-
- }
- }
- ```
+ ```
* Compile the parser via `mvn clean package` in `extra_parsers`
* This will create a jar containing your parser and its dependencies (sans Metron dependencies) in `extra-parsers/target/extra-parsers-1.0-SNAPSHOT-uber.jar`
@@ -289,52 +289,52 @@ $METRON_HOME/bin/zk_load_configs.sh -m PUSH -i $METRON_HOME/config/zookeeper/ -z
Now we can start the parser and send some data through:
* Start the parser
- ```
- $METRON_HOME/bin/start_parser_topology.sh -k $BROKERLIST -z $ZOOKEEPER -s test
- ```
+ ```
+ $METRON_HOME/bin/start_parser_topology.sh -k $BROKERLIST -z $ZOOKEEPER -s test
+ ```
* Send example data through:
- ```
- echo "apache,metron" | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list $BROKERLIST --topic test
- ```
+ ```
+ echo "apache,metron" | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list $BROKERLIST --topic test
+ ```
* Validate data was written in ES:
- ```
- curl -XPOST "http://$ES_HOST/test*/_search?pretty" -d '
- {
- "_source" : [ "original_string", "timestamp", "first", "last"]
- }
- '
- ```
+ ```
+ curl -XPOST "http://$ES_HOST/test*/_search?pretty" -d '
+ {
+ "_source" : [ "original_string", "timestamp", "first", "last"]
+ }
+ '
+ ```
* This should yield something like:
- ```
- {
- "took" : 23,
- "timed_out" : false,
- "_shards" : {
- "total" : 1,
- "successful" : 1,
- "failed" : 0
- },
- "hits" : {
- "total" : 1,
- "max_score" : 1.0,
- "hits" : [ {
- "_index" : "test_index_2017.10.04.17",
- "_type" : "test_doc",
- "_id" : "3ae4dd4d-8c09-4f2a-93c0-26ec5508baaa",
- "_score" : 1.0,
- "_source" : {
- "original_string" : "apache,metron",
- "last" : "metron",
- "first" : "apache",
- "timestamp" : 1507138373223
- }
- } ]
+ ```
+ {
+ "took" : 23,
+ "timed_out" : false,
+ "_shards" : {
+ "total" : 1,
+ "successful" : 1,
+ "failed" : 0
+ },
+ "hits" : {
+ "total" : 1,
+ "max_score" : 1.0,
+ "hits" : [ {
+ "_index" : "test_index_2017.10.04.17",
+ "_type" : "test_doc",
+ "_id" : "3ae4dd4d-8c09-4f2a-93c0-26ec5508baaa",
+ "_score" : 1.0,
+ "_source" : {
+ "original_string" : "apache,metron",
+ "last" : "metron",
+ "first" : "apache",
+ "timestamp" : 1507138373223
+ }
+ } ]
+ }
}
- }
- ```
+ ```
### Via the Management UI
[05/15] metron git commit: METRON-1658: Upgrade bro to 2.5.4 closes
apache/incubator-metron#1101
Posted by mm...@apache.org.
METRON-1658: Upgrade bro to 2.5.4 closes apache/incubator-metron#1101
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/ae915a02
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/ae915a02
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/ae915a02
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: ae915a02ce90fb9b4c03166e7b57f8aa2a5d7890
Parents: 75d6a72
Author: JonZeolla <ze...@gmail.com>
Authored: Mon Jul 16 12:01:49 2018 -0400
Committer: cstella <ce...@gmail.com>
Committed: Mon Jul 16 12:01:49 2018 -0400
----------------------------------------------------------------------
metron-deployment/ansible/playbooks/docker_probe_install.yml | 2 +-
metron-deployment/ansible/roles/bro/vars/main.yml | 2 +-
.../development/centos6/ansible/inventory/group_vars/all | 2 +-
.../development/ubuntu14/ansible/inventory/group_vars/all | 2 +-
4 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/ae915a02/metron-deployment/ansible/playbooks/docker_probe_install.yml
----------------------------------------------------------------------
diff --git a/metron-deployment/ansible/playbooks/docker_probe_install.yml b/metron-deployment/ansible/playbooks/docker_probe_install.yml
index 3de212e..5e5165e 100644
--- a/metron-deployment/ansible/playbooks/docker_probe_install.yml
+++ b/metron-deployment/ansible/playbooks/docker_probe_install.yml
@@ -32,7 +32,7 @@
vars:
metron_version: 0.5.1
metron_directory: /usr/metron/{{ metron_version }}
- bro_version: "2.5.2"
+ bro_version: "2.5.4"
fixbuf_version: "1.7.1"
yaf_version: "2.8.0"
daq_version: "2.0.6-1"
http://git-wip-us.apache.org/repos/asf/metron/blob/ae915a02/metron-deployment/ansible/roles/bro/vars/main.yml
----------------------------------------------------------------------
diff --git a/metron-deployment/ansible/roles/bro/vars/main.yml b/metron-deployment/ansible/roles/bro/vars/main.yml
index 11d6c31..9e6f221 100644
--- a/metron-deployment/ansible/roles/bro/vars/main.yml
+++ b/metron-deployment/ansible/roles/bro/vars/main.yml
@@ -17,7 +17,7 @@
---
bro_home: /usr/local/bro
bro_bin: "{{ bro_home }}/bin"
-bro_version: 2.5.2
+bro_version: 2.5.4
bro_daemon_log: /var/log/bro.log
bro_topic: bro
http://git-wip-us.apache.org/repos/asf/metron/blob/ae915a02/metron-deployment/development/centos6/ansible/inventory/group_vars/all
----------------------------------------------------------------------
diff --git a/metron-deployment/development/centos6/ansible/inventory/group_vars/all b/metron-deployment/development/centos6/ansible/inventory/group_vars/all
index 95792c3..192ed16 100644
--- a/metron-deployment/development/centos6/ansible/inventory/group_vars/all
+++ b/metron-deployment/development/centos6/ansible/inventory/group_vars/all
@@ -37,7 +37,7 @@ enrichment_hbase_table: enrichment
# metron
metron_version: 0.5.1
metron_directory: /usr/metron/{{ metron_version }}
-bro_version: "2.5.2"
+bro_version: "2.5.4"
fixbuf_version: "1.7.1"
yaf_version: "2.8.0"
daq_version: "2.0.6-1"
http://git-wip-us.apache.org/repos/asf/metron/blob/ae915a02/metron-deployment/development/ubuntu14/ansible/inventory/group_vars/all
----------------------------------------------------------------------
diff --git a/metron-deployment/development/ubuntu14/ansible/inventory/group_vars/all b/metron-deployment/development/ubuntu14/ansible/inventory/group_vars/all
index 6276fc0..195972a 100644
--- a/metron-deployment/development/ubuntu14/ansible/inventory/group_vars/all
+++ b/metron-deployment/development/ubuntu14/ansible/inventory/group_vars/all
@@ -37,7 +37,7 @@ enrichment_hbase_table: enrichment
# metron
metron_version: 0.5.1
metron_directory: /usr/metron/{{ metron_version }}
-bro_version: "2.5.2"
+bro_version: "2.5.4"
fixbuf_version: "1.7.1"
yaf_version: "2.8.0"
daq_version: "2.0.6-1"
[08/15] metron git commit: METRON-1673 Fix Javadoc errors
(justinleet) closes apache/metron#1107
Posted by mm...@apache.org.
METRON-1673 Fix Javadoc errors (justinleet) closes apache/metron#1107
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/2a1cbbbf
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/2a1cbbbf
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/2a1cbbbf
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 2a1cbbbf2029faec0b265e83f270a31e75ef9d9a
Parents: 0c20fd1
Author: justinleet <ju...@gmail.com>
Authored: Tue Jul 17 09:18:08 2018 -0400
Committer: leet <le...@apache.org>
Committed: Tue Jul 17 09:18:08 2018 -0400
----------------------------------------------------------------------
.../metron/profiler/client/window/Window.java | 8 ++---
.../profiler/client/window/WindowProcessor.java | 35 ++++++++++++--------
.../client/window/predicates/DayPredicates.java | 2 +-
.../profiler/DefaultMessageDistributor.java | 18 +++++-----
.../apache/metron/rest/config/KafkaConfig.java | 4 +--
.../configuration/IndexingConfigurations.java | 2 +-
.../configuration/SensorParserConfig.java | 6 ++--
.../writer/ConfigurationsStrategies.java | 2 +-
.../common/performance/PerformanceLogger.java | 10 +++---
.../apache/metron/common/system/FakeClock.java | 18 +++++-----
.../metron/common/utils/ReflectionUtils.java | 2 +-
.../configurations/ConfigurationsUpdater.java | 1 -
.../indexing/dao/metaalert/MetaAlertDao.java | 6 ++--
.../metron/writer/BulkWriterComponent.java | 24 ++++++--------
.../stellar/common/CachingStellarProcessor.java | 2 +-
.../org/apache/metron/stellar/dsl/Context.java | 2 +-
16 files changed, 72 insertions(+), 70 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/Window.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/Window.java b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/Window.java
index 964d594..9fd6339 100644
--- a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/Window.java
+++ b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/Window.java
@@ -20,14 +20,12 @@
package org.apache.metron.profiler.client.window;
import com.google.common.collect.Iterables;
-import org.apache.commons.lang3.Range;
-
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.function.Function;
import java.util.function.Predicate;
-import java.util.stream.Stream;
+import org.apache.commons.lang3.Range;
/**
* A window is intended to compute the set of window intervals across time based on a reference time.
@@ -72,7 +70,7 @@ public class Window {
* Get the set of inclusion predicates. If any of these are true as applied to the window interval start time,
* then a field is included unless it's explicitly excluded.
* @param now
- * @return Iterable<Predicate<Long>> returns a set of inclusion predicates. If any of these are true as
+ * @return {@code Iterable<Predicate<Long>>} returns a set of inclusion predicates. If any of these are true as
* applied to the window interval start time, then a field is included unless it's explicitly excluded
*/
public Iterable<Predicate<Long>> getIncludes(long now) {
@@ -87,7 +85,7 @@ public class Window {
* Get the set of exclusion predicates. If any of these exclusion predicates are true as applied to the window
* interval start time, then the interval is excluded. NOTE: Exclusions trump inclusions.
* @param now
- * @return Iterable<Predicate<Long>> returns the set of exclusion predicates. If any of these exclusion
+ * @return {@code Iterable<Predicate<Long>>} returns the set of exclusion predicates. If any of these exclusion
* predicates are true as applied to the window interval start time, then the interval is excluded.
* Exclusions trump inclusions.
*/
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/WindowProcessor.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/WindowProcessor.java b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/WindowProcessor.java
index da4ce3c..66a4f7d 100644
--- a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/WindowProcessor.java
+++ b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/WindowProcessor.java
@@ -19,24 +19,31 @@
*/
package org.apache.metron.profiler.client.window;
-import org.antlr.v4.runtime.*;
+import static org.apache.commons.lang3.StringUtils.isEmpty;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import org.antlr.v4.runtime.ANTLRInputStream;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.ParserRuleContext;
+import org.antlr.v4.runtime.TokenStream;
import org.antlr.v4.runtime.tree.ParseTree;
-import org.apache.metron.stellar.dsl.ErrorListener;
-import org.apache.metron.stellar.dsl.GrammarUtils;
-import org.apache.metron.stellar.dsl.ParseException;
-import org.apache.metron.stellar.dsl.Token;
-import org.apache.metron.stellar.common.utils.ConversionUtils;
import org.apache.metron.profiler.client.window.generated.WindowBaseListener;
import org.apache.metron.profiler.client.window.generated.WindowLexer;
import org.apache.metron.profiler.client.window.generated.WindowParser;
import org.apache.metron.profiler.client.window.predicates.DayPredicates;
-
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
-import java.util.function.Predicate;
-
-import static org.apache.commons.lang3.StringUtils.isEmpty;
+import org.apache.metron.stellar.common.utils.ConversionUtils;
+import org.apache.metron.stellar.dsl.ErrorListener;
+import org.apache.metron.stellar.dsl.GrammarUtils;
+import org.apache.metron.stellar.dsl.ParseException;
+import org.apache.metron.stellar.dsl.Token;
/**
* The WindowProcessor instance provides the parser callbacks for the Window selector language. This constructs
@@ -142,7 +149,7 @@ public class WindowProcessor extends WindowBaseListener {
}
/**
- * Normalize the day specifier e.g. tuesdays -> tuesday and push onto the stack.
+ * Normalize the day specifier e.g. tuesdays -{@literal >} tuesday and push onto the stack.
* @param ctx
*/
@Override
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/predicates/DayPredicates.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/predicates/DayPredicates.java b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/predicates/DayPredicates.java
index e8b985c..2c5adb4 100644
--- a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/predicates/DayPredicates.java
+++ b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/window/predicates/DayPredicates.java
@@ -104,7 +104,7 @@ public enum DayPredicates {
* Create a Predicate given a set of arguments.
* @param name
* @param arg
- * @return Predicate<Long> return a Predicate given a set of arguments
+ * @return {@code Predicate<Long>} return a Predicate given a set of arguments
*/
public static Predicate<Long> create(String name, List<String> arg) {
return DayPredicates.valueOf(name).predicateCreator.apply(arg);
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/DefaultMessageDistributor.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/DefaultMessageDistributor.java b/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/DefaultMessageDistributor.java
index 70f4228..82f7174 100644
--- a/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/DefaultMessageDistributor.java
+++ b/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/DefaultMessageDistributor.java
@@ -20,26 +20,25 @@
package org.apache.metron.profiler;
+import static java.lang.String.format;
+
import com.google.common.base.Ticker;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
-import org.apache.commons.lang.builder.HashCodeBuilder;
-import org.apache.metron.common.configuration.profiler.ProfileConfig;
-import org.apache.metron.stellar.dsl.Context;
-import org.json.simple.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-
-import static java.lang.String.format;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.metron.common.configuration.profiler.ProfileConfig;
+import org.apache.metron.stellar.dsl.Context;
+import org.json.simple.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* The default implementation of a {@link MessageDistributor}.
@@ -151,7 +150,6 @@ public class DefaultMessageDistributor implements MessageDistributor {
* @param timestamp The timestamp of the message.
* @param route The message route.
* @param context The Stellar execution context.
- * @throws ExecutionException
*/
@Override
public void distribute(JSONObject message, long timestamp, MessageRoute route, Context context) {
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java
index 7e9b468..03db036 100644
--- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java
+++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java
@@ -73,9 +73,9 @@ public class KafkaConfig {
}
/**
- * Create properties that will be used by {@link this#createConsumerFactory()}
+ * Create properties that will be used by {@link org.apache.metron.rest.config.KafkaConfig#createConsumerFactory()}
*
- * @return Configurations used by {@link this#createConsumerFactory()}.
+ * @return Configurations used by {@link org.apache.metron.rest.config.KafkaConfig#createConsumerFactory()}.
*/
@Bean
public Map<String, Object> consumerProperties() {
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java
index 7ecb606..2f01884 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java
@@ -112,7 +112,7 @@ public class IndexingConfigurations extends Configurations {
/**
* Returns all configured values of batchTimeout, for all configured sensors,
- * but only for the specific writer identified by {@param writerName}. So, if it is
+ * but only for the specific writer identified by {@code writerName}. So, if it is
* an hdfs writer, it will return the batchTimeouts for hdfs writers for all the sensors.
* The goal is to return to a {@link org.apache.metron.common.bolt.ConfiguredBolt}
* the set of all and only batchTimeouts relevant to that ConfiguredBolt.
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
index 01630c1..4cc4e61 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
@@ -200,8 +200,10 @@ public class SensorParserConfig implements Serializable {
/**
* Configures the cache that backs stellar field transformations.
*
- * <li>stellar.cache.maxSize - The maximum number of elements in the cache.
- * <li>stellar.cache.maxTimeRetain - The maximum amount of time an element is kept in the cache (in minutes).
+ * <ul>
+ * <li>stellar.cache.maxSize - The maximum number of elements in the cache.
+ * <li>stellar.cache.maxTimeRetain - The maximum amount of time an element is kept in the cache (in minutes).
+ * </ul>
*/
private Map<String, Object> cacheConfig = new HashMap<>();
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/writer/ConfigurationsStrategies.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/writer/ConfigurationsStrategies.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/writer/ConfigurationsStrategies.java
index f22dd01..bafbbe1 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/writer/ConfigurationsStrategies.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/writer/ConfigurationsStrategies.java
@@ -134,7 +134,7 @@ public enum ConfigurationsStrategies implements ConfigurationStrategy {
/**
* Config updater
* @param reloadable callback
- * @param configSupplier Supplier provides config of type <? extends Configurations>
+ * @param configSupplier Supplier provides config of type {@code <? extends Configurations>}
* @return Config updater
*/
@Override
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-common/src/main/java/org/apache/metron/common/performance/PerformanceLogger.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/performance/PerformanceLogger.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/performance/PerformanceLogger.java
index 61015ef..78aacd3 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/performance/PerformanceLogger.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/performance/PerformanceLogger.java
@@ -77,7 +77,7 @@ public class PerformanceLogger {
/**
* Log a message at DEBUG level for the given markName.
* Warns when logging for a markName that hasn't been set.
- * <p/>
+ * <p>
* <p>This form avoids superfluous string concatenation when the logger
* is disabled for the DEBUG level.</p>
*
@@ -92,7 +92,7 @@ public class PerformanceLogger {
/**
* Log a message at DEBUG level for the given markName according to the specified message.
* Warns when logging for a markName that hasn't been set.
- * <p/>
+ * <p>
* <p>This form avoids superfluous string concatenation when the logger
* is disabled for the DEBUG level.</p>
*
@@ -127,7 +127,7 @@ public class PerformanceLogger {
/**
* Log a message at DEBUG level for the given markName according to the specified format
* and argument. Warns when logging for a markName that hasn't been set.
- * <p/>
+ * <p>
* <p>This form avoids superfluous string concatenation when the logger
* is disabled for the DEBUG level.</p>
*
@@ -144,7 +144,7 @@ public class PerformanceLogger {
/**
* Log a message at DEBUG level according to the specified format and argument.
- * <p/>
+ * <p>
* <p>This form avoids superfluous string concatenation when the logger
* is disabled for the DEBUG level.</p>
*
@@ -162,7 +162,7 @@ public class PerformanceLogger {
/**
* Log a message at DEBUG level according to the specified format and arguments.
- * <p/>
+ * <p>
* <p>This form avoids superfluous string concatenation when the logger
* is disabled for the DEBUG level. However, this variant incurs the hidden
* (and relatively small) cost of creating an <code>Object[]</code> before invoking the method,
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-common/src/main/java/org/apache/metron/common/system/FakeClock.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/system/FakeClock.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/system/FakeClock.java
index e34a26d..d0f9b5e 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/system/FakeClock.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/system/FakeClock.java
@@ -24,11 +24,11 @@ import java.util.concurrent.TimeUnit;
* never advances itself, but allows you to increment it by any desired amount.
*
* Note that the base class is not the Java 8 Clock, but rather the Clock we
- * defined in {@org.apache.metron.common.system.Clock}. Fundamental units of time
+ * defined in {@link org.apache.metron.common.system.Clock}. Fundamental units of time
* are milliseconds.
*
- * Three exceptions are also defined: {@IllegalArgumentClockNegative},
- * {@IllegalArgumentClockZero}, and {@IllegalArgumentClockOverflow}.
+ * Three exceptions are also defined: {@link IllegalArgumentClockNegative},
+ * {@link IllegalArgumentClockZero}, and {@link IllegalArgumentClockOverflow}.
* These are thrown in various circumstances that imply the FakeClock is
* being used outside of its design intent. They are subclasses of IllegalArgumentException,
* hence unchecked.
@@ -52,7 +52,7 @@ public class FakeClock extends Clock {
* If we really get a demand for this capability, we'll add methods that don't
* check for this.
* @throws IllegalArgumentClockOverflow (unchecked) if you try to add a duration
- * that would overflow the Long value of {@currentTimeMillis}
+ * that would overflow the Long value of {@code currentTimeMillis}
*/
public void elapseMillis(long duration_ms) {
long instant_ms = now_ms + duration_ms;
@@ -71,7 +71,7 @@ public class FakeClock extends Clock {
/**
* Advance the fake clock by a number of seconds.
- * See {@elapseMillis} for details.
+ * See {@code elapseMillis} for details.
*
* @param duration_secs
*/
@@ -93,7 +93,7 @@ public class FakeClock extends Clock {
* Why? Because it implies your test code has lost track of previous increments,
* which might be problematic, so we do this in the spirit of "fail fast".
* If you *meant* to lose track, for instance if you were using random numbers of events,
- * or whatever, you can always orient yourself in time by reading {@currentTimeMillis}.
+ * or whatever, you can always orient yourself in time by reading {@code currentTimeMillis}.
*/
public void advanceToMillis(long instant_ms) {
if (instant_ms < now_ms) {
@@ -111,7 +111,7 @@ public class FakeClock extends Clock {
/**
* Advance the fake clock to a point in time specified as seconds after 0.
- * See {@advanceToMillis} for details.
+ * See {@code advanceToMillis} for details.
*
* @param instant_secs - epoch time in seconds
*/
@@ -138,7 +138,7 @@ public class FakeClock extends Clock {
* Why? Because it implies your test code has lost track of previous increments,
* which might be problematic, so we do this in the spirit of "fail fast".
* If you *meant* to lose track, for instance if you were using random numbers of events,
- * or whatever, you can always orient yourself in time by reading {@currentTimeMillis}.
+ * or whatever, you can always orient yourself in time by reading {@code currentTimeMillis}.
*
* Note that argument does not apply to ellapseMillis(0), so it does not throw
* this exception.
@@ -151,7 +151,7 @@ public class FakeClock extends Clock {
/**
* IllegalArgumentClockOverflow (unchecked) is thrown if you try to add a duration
- * that would overflow the Long value of {@currentTimeMillis}
+ * that would overflow the Long value of {@code currentTimeMillis}
*/
public static class IllegalArgumentClockOverflow extends IllegalArgumentException {
public IllegalArgumentClockOverflow(String s) {
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ReflectionUtils.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ReflectionUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ReflectionUtils.java
index ee6b041..1f216a8 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ReflectionUtils.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ReflectionUtils.java
@@ -95,7 +95,7 @@ public class ReflectionUtils {
* Create instance from passed class name with specified parameter types and arguments. If parameter
* types is null, defaults to attempting to instantiate the no-arg constructor.
*
- * @param clazz Class to create instance from
+ * @param className Class to create instance from
* @param parameterTypes parameter types to use for looking up the desired constructor
* @param parameters arguments to pass into the constructor when instantiating the object.
* @param <T> Instance created from passed class
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-common/src/main/java/org/apache/metron/common/zookeeper/configurations/ConfigurationsUpdater.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/zookeeper/configurations/ConfigurationsUpdater.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/zookeeper/configurations/ConfigurationsUpdater.java
index 9ff2bee..f616ece 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/zookeeper/configurations/ConfigurationsUpdater.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/zookeeper/configurations/ConfigurationsUpdater.java
@@ -118,7 +118,6 @@ public abstract class ConfigurationsUpdater<T extends Configurations> implements
* this is ONLY called on deletes to path to the zookeeper nodes which correspond
* to your configurations type (rather than all configurations type).
* @param name the path
- * @throws IOException when update is unable to happen
*/
public abstract void delete(String name);
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/metaalert/MetaAlertDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/metaalert/MetaAlertDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/metaalert/MetaAlertDao.java
index c9e6711..66d1d93 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/metaalert/MetaAlertDao.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/metaalert/MetaAlertDao.java
@@ -29,7 +29,7 @@ import org.apache.metron.indexing.dao.IndexDao;
* status field that controls it's inclusion in search results and a groups field that can be used to track
* the groups a meta alert was created from.
*
- * </p>
+ * <p>
* The structure of a meta alert is as follows:
* {
* "guid": "meta alert guid",
@@ -47,13 +47,13 @@ import org.apache.metron.indexing.dao.IndexDao;
* "median": 10
* }
*
- * </p>
+ * <p>
* A child alert that has been added to a meta alert will store the meta alert GUID in a "metaalerts" field.
* This field is an array of meta alert GUIDs, meaning a child alert can be contained in multiple meta alerts.
* Any update to a child alert will trigger an update to the meta alert so that the alert inside a meta alert
* and the original alert will be kept in sync.
*
- * </p>
+ * <p>
* Other fields can be added to a meta alert through the patch method on the IndexDao interface. However, attempts
* to directly change the "alerts" or "status" field will result in an exception.
*/
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
index dfa265d..e35960f 100644
--- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
+++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
@@ -18,8 +18,17 @@
package org.apache.metron.writer;
+import static java.lang.String.format;
+
import com.google.common.collect.Iterables;
-import org.apache.commons.lang.exception.ExceptionUtils;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.writer.WriterConfiguration;
import org.apache.metron.common.error.MetronError;
@@ -33,17 +42,6 @@ import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.TimeUnit;
-
-import static java.lang.String.format;
-
/**
* This component implements message batching, with both flush on queue size, and flush on queue timeout.
* There is a queue for each sensorType.
@@ -54,7 +52,7 @@ import static java.lang.String.format;
* modified by batchTimeoutDivisor, in case multiple batching writers are daisy-chained in one topology.
* - If some sensors configure their own batchTimeouts, they are compared with the default. Batch
* timeouts greater than the default will be ignored, because they can cause message recycling in Storm.
- * Batch timeouts configured to <= zero, or undefined, mean use the default.
+ * Batch timeouts configured to {@literal <}= zero, or undefined, mean use the default.
* - The *smallest* configured batchTimeout among all sensor types, greater than zero and less than
* the default, will be used to configure the 'topology.tick.tuple.freq.secs' for the Bolt. If there are no
* valid configured batchTimeouts, the defaultBatchTimeout will be used.
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/CachingStellarProcessor.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/CachingStellarProcessor.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/CachingStellarProcessor.java
index 19de14e..28d69b3 100644
--- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/CachingStellarProcessor.java
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/CachingStellarProcessor.java
@@ -184,7 +184,7 @@ public class CachingStellarProcessor extends StellarProcessor {
}
/**
- * Create a cache given a config. Note that if the cache size is <= 0, then no cache will be returned.
+ * Create a cache given a config. Note that if the cache size is {@literal <}= 0, then no cache will be returned.
* @param config
* @return A cache.
*/
http://git-wip-us.apache.org/repos/asf/metron/blob/2a1cbbbf/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/Context.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/Context.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/Context.java
index 8a477c4..0f04151 100644
--- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/Context.java
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/Context.java
@@ -63,7 +63,7 @@ public class Context implements Serializable {
SHELL_VARIABLES
,
/**
- * This capability indicates that the StellarProcessor should use a Caffeine cache to cache expression -> results. If an expression
+ * This capability indicates that the StellarProcessor should use a Caffeine cache to cache expression -{@literal >} results. If an expression
* is in the cache, then the cached result will be returned instead of recomputing.
*/
CACHE
[10/15] metron git commit: METRON-1657 Parser aggregation in storm
(justinleet) closes apache/metron#1099
Posted by mm...@apache.org.
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/use-cases/parser_chaining/aggregated_parser_chaining_flow.svg
----------------------------------------------------------------------
diff --git a/use-cases/parser_chaining/aggregated_parser_chaining_flow.svg b/use-cases/parser_chaining/aggregated_parser_chaining_flow.svg
new file mode 100644
index 0000000..28529c8
--- /dev/null
+++ b/use-cases/parser_chaining/aggregated_parser_chaining_flow.svg
@@ -0,0 +1,14 @@
+<!--
+ 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.
+ -->
+<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="787px" height="967px" version="1.1" content="<mxfile userAgent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36" version="8.9.5" editor="www.draw.io" type="device"><diagram name="Page-1" id="c7558073-3199-34d8-9f00-42111426c3f3">7Vtbb6M6EP41eewKMBDy2Nt2pXOr1COd3acVAYdYJRiB06b7648dbMAXEppCk27al+KxMXi+b8YzYzIB16vNXRHmy79wDNOJY8WbCbiZOI5tzyz6j0leKokf2JUgKVDMBzWCB/QLciG/L1mjGJbSQIJxSlAuCyOcZTAikiwsCvwsD1vgVH5qHiZQEzxEYapL/0MxWVbSwJk28m8QJUvxZMviLz4Po8ekwOuMP2/igMX2r+pehWIuPr5chjF+bonA7QRcFxiT6mq1uYYp061QW3Xf147e+r0LmJE+N4DqhqcwXfOl/42+Iv5u5EXo43mJCHzIw4i1nynmE3AVpijJaDOij4IFFTzBgiCqw0vesUJxzCa4WuCMPPDJLNouSYEfa8UCNgKl6TVOMZ3mJsMZu0lfCV8cewzctER8ZXcQryApXugQ3gu86g5OQsB1/twgKni6bIHpc1nIOZTU8zZ6pBdclR04uJpec7T5Wb6UKU5+Un4wjTk+ZYfNFPJAcLFiGs9pT0t+CjiU9GkoS6jAH
ggV1/2yFxgbTEdCRliqGZoSZiVm0KT0mVdzdpWwqz/CxWO49UI5ijRYqBXn7DJ6SVEWb0EwITVnfgHGf85Trs3GW/yzJvRWyOVpOIfpVd2pIGIAaUlWYs4KVNFPmWRt/4xwd0BnALgTTdv3JSxtR8fSNljZzB8CS93MNGxgTD07b3JtZfi2JXyt0uAGke/8Pnb9g1kTZTRtZXQB37lxbRtNn6DIIoWbS7ZD1epnL7hb+XQ9eF1E9R7J1UnCIoFinGtGqQ2DZYChFhYwDQl6kt/EBA5/xj1G9B0boxaelNNgqphqtQB+U3s/Uudx5Xl8IM9TrVmbZ8uUetk9yeMfgTzMBRtHDkYPYCCHZU2Oyg7bkR2+6sj7sqMOm5SNYwR2CD+2ix0mH9/yxGGZV7HpAm1gvNet99qNu/gzgCevbU14cuHZWywBJpbUwjf58jqQ2KFwk5XALOaXN1EaliXdniUU2kpsovFdVtul48a9N97+h/D2fFPgXXxbsI2wHuj2XYNhVyzdadieATFvIKevWKNqjn3NOpDnqdsjmLUh4RHBHg10MxHuRaiM8IV3AViEUXXTqdsjPkPE/iGiZ0v4eo4h3h8tRnRMMWIX4j5F3PlE/K2IewFQUrx3xtzTMJct+qMm3m+zQzCTo7DgPfNuRw+3ZZs7S0y8qXVMTKa/Q8ylJ+KGpL2Vs+GCLHGCszBtZXKHRmVeYAjLnP3Z+Ihhmafk0K4an/eNy7SJxgzMgpOnogjxB6HbLmIPSkVwTCrW84gUYabUc3pn/kqOUMcTY1BxdipU7F8/Oiw3PZBoxhpTFXOdSCoa2MOkolN1mx2QZqBHhem8aWaqeFRh5NH82UwpZDpKutLbn6kTzUYkmh7lwaxA0XJF1V9+prW9g3UgdljhZUzB+mhJLdC3JUPRooHxjmqgWl7JDpstmk3l24RKGachTBVEZD0XsES/wnmdK+WMltu1eFcT74blXWuCy+ozDruVhqVwQQxJGG
VWk0z9yxo3F+4OtzMAdq6SaE0DT8fOMWCnFjUP+xxAP3SWChLWCpZlmECDNZ4HPJ7yRYBnsCzXgI4aBh6Gjn70IFnW2aPjK2HRO8OjV/Muk6SASUhgTOWiYHQvPN0cp6QHMPLu0b3h9K8pmTbCotoIOfh7drvxIFT3LjvQT/eMe9cwEHYV/6Qjls69a6/V/WYnsVMZK+AGGlaBASq1PHMYVHq0+Jow48ygcv1jQnUyNbNXZJA9vp6Sc0t7zPKtOCmSv6U6ao4J7JlyeOb7B2aZwNemspWpBswz3Y9YN+ss6h6Hjr6JjtNj0tENbJVDakjQ+zwBqFN56neCA9LR6/Fx6MnRsSGdLRfYvlh1+x4WiCqChaJ7zsFG46lr4ik4Kk+VCrBteQeyNFAnUo/PhuSontp8JI4Gr+aowc+OxtHpJ0cH4ejbcrczrzuqJ6DvW3f09AzB9Dukz5SuXSnxHQkxxzFUSixHR6wWvgIy2mx+7FdZZ/OLSnD7Pw==</diagram></mxfile>" style="background-color: rgb(255, 255, 255);"><defs/><g transform="translate(0.5,0.5)"><rect x="1" y="19" width="90" height="60" fill="none" stroke="#000000" stroke-width="3" pointer-events="none"/><g transform="translate(34.5,42.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="22" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inlin
e-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 22px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">NiFi</div></div></foreignObject><text x="11" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">NiFi</text></switch></g><rect x="310.5" y="19" width="137" height="60" fill="none" stroke="#000000" stroke-width="3" pointer-events="none"/><g transform="translate(331.5,28.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="95" height="40" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 95px; white-space: nowrap; word-wrap: normal; text-
align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">pix_syslog_router<br />Storm spout<div><br /></div></div></div></foreignObject><text x="48" y="26" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">[Not supported by viewer]</text></switch></g><path d="M 132 20 C 132 39 242 39 242 20" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><path d="M 132 20 C 132 -5.33 242 -5.33 242 20 L 242 78 C 242 103.33 132 103.33 132 78 Z" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><g transform="translate(137.5,49.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="98" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0
); line-height: 1.2; vertical-align: top; width: 100px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">pix_syslog_sensor<br />Kafka topic</div></div></foreignObject><text x="49" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">pix_syslog_sensor<br>Kafka topic</text></switch></g><path d="M 243.5 55 L 243.5 43 L 288.5 43 L 288.5 31.5 L 309.5 49 L 288.5 66.5 L 288.5 55 Z" fill="none" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 92.5 55 L 92.5 43 L 109.5 43 L 109.5 31.5 L 130.5 49 L 109.5 66.5 L 109.5 55 Z" fill="none" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><rect x="229" y="155" width="300" height="300" fill="none" stroke="#000000" stroke-width="3" pointer-events="none"/><path d="M 373 8
0.5 L 385 80.5 L 385 132.5 L 396.5 132.5 L 379 153.5 L 361.5 132.5 L 373 132.5 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 117 528 C 117 547 227 547 227 528" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><path d="M 117 528 C 117 502.67 227 502.67 227 528 L 227 586 C 227 611.33 117 611.33 117 586 Z" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><g transform="translate(140.5,557.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="62" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 64px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w
3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;"><span>cisco-5-304</span><br />Kafka topic</div></div></foreignObject><text x="31" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">[Not supported by viewer]</text></switch></g><path d="M 549.5 528 C 549.5 547 659.5 547 659.5 528" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><path d="M 549.5 528 C 549.5 502.67 659.5 502.67 659.5 528 L 659.5 586 C 659.5 611.33 549.5 611.33 549.5 586 Z" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><g transform="translate(573.5,557.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="62" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height:
1.2; vertical-align: top; width: 64px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;"><span>cisco-6-302</span><br />Kafka topic</div></div></foreignObject><text x="31" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">[Not supported by viewer]</text></switch></g><rect x="105" y="669" width="137" height="60" fill="none" stroke="#000000" stroke-width="3" pointer-events="none"/><g transform="translate(140.5,678.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="65" height="40" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 65px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmln
s="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">cisco-5-304<br />Storm spout<div><br /></div></div></div></foreignObject><text x="33" y="26" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">[Not supported by viewer]</text></switch></g><rect x="536" y="669" width="137" height="60" fill="none" stroke="#000000" stroke-width="3" pointer-events="none"/><g transform="translate(571.5,678.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="65" height="40" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 65px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">
cisco-6-302<br />Storm spout<div><br /></div></div></div></foreignObject><text x="33" y="26" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">[Not supported by viewer]</text></switch></g><path d="M 431.5 272 L 431.5 260 L 599 260 Q 611 260 611 272 L 611 486.5 L 622.5 486.5 L 605 507.5 L 587.5 486.5 L 599 486.5 L 599 272 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 611 486.5 L 622.5 486.5 L 605 507.5 L 587.5 486.5 L 599 486.5" fill="none" stroke="#000000" stroke-width="3" stroke-linejoin="flat" stroke-miterlimit="4" pointer-events="none"/><path d="M 326.5 260 L 326.5 272 L 178 272 L 178 486.5 L 189.5 486.5 L 172 507.5 L 154.5 486.5 L 166 486.5 L 166 272 Q 166 260 178 260 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 178 486.5 L 189.5 486.5 L 172 507.5 L 154.5 486.5 L 166 486.5" fill="
none" stroke="#000000" stroke-width="3" stroke-linejoin="flat" stroke-miterlimit="4" pointer-events="none"/><path d="M 166.05 606.69 L 178.04 606.31 L 179.29 646.32 L 190.79 645.96 L 173.95 667.5 L 155.81 647.06 L 167.3 646.7 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 599 606.5 L 611 606.5 L 611 646.5 L 622.5 646.5 L 605 667.5 L 587.5 646.5 L 599 646.5 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 324 888 C 324 907 434 907 434 888" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><path d="M 324 888 C 324 862.67 434 862.67 434 888 L 434 946 C 434 971.33 324 971.33 324 946 Z" fill="none" stroke="#000000" stroke-width="3" stroke-miterlimit="10" pointer-events="none"/><g transform="translate(345.5,917.5)"><switch><foreignObject style="overflow:visible;" pointer-events="a
ll" width="66" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 66px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">enrichments<br />Kafka topic</div></div></foreignObject><text x="33" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">enrichments<br>Kafka topic</text></switch></g><g transform="translate(437.5,776.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="100" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); l
ine-height: 1.2; vertical-align: top; white-space: nowrap;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">cisco-6-302<br />Grok parser output<br /></div></div></foreignObject><text x="50" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">cisco-6-302<br>Grok parser output<br></text></switch></g><g transform="translate(22.5,341.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="120" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; white-space: nowrap;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">cisco-5-304 messages<br /></div></div></foreignObject><text x="60" y="12" fill
="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">cisco-5-304 messages<br></text></switch></g><g transform="translate(647.5,341.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="120" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; white-space: nowrap;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">cisco-6-302 messages<br /></div></div></foreignObject><text x="60" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">cisco-6-302 messages<br></text></switch></g><g transform="translate(324.5,171.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="108" height="26" requiredFeatures="http://www.w3
.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 108px; white-space: normal; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">Aggregated Storm Parser bolt</div></div></foreignObject><text x="54" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">Aggregated Storm Parser bolt</text></switch></g><rect x="244" y="337" width="80" height="80" fill="none" stroke="#000000" stroke-width="3" pointer-events="none"/><g transform="translate(251.5,363.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="64" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size:
12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 64px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">cisco-5-304<br />Grok parser<br /></div></div></foreignObject><text x="32" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">[Not supported by viewer]</text></switch></g><rect x="434" y="337" width="80" height="80" fill="none" stroke="#000000" stroke-width="3" pointer-events="none"/><g transform="translate(441.5,363.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="64" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 64px; white-space: no
wrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">cisco-6-302<br />Grok parser<br /></div></div></foreignObject><text x="32" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">[Not supported by viewer]</text></switch></g><path d="M 243.5 705 L 243.5 693 L 278 693 L 278 439.5 L 266.5 439.5 L 284 418.5 L 301.5 439.5 L 290 439.5 L 290 693 Q 290 705 278 705 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 278 439.5 L 266.5 439.5 L 284 418.5 L 301.5 439.5 L 290 439.5" fill="none" stroke="#000000" stroke-width="3" stroke-linejoin="flat" stroke-miterlimit="4" pointer-events="none"/><path d="M 534.5 693 L 534.5 705 L 480 705 Q 468 705 468 693 L 468 439.5 L 456.5 439.5 L 474 418.5 L 491.5 439.5 L 480 439.5 L 480 693 Z" fill="#ffffff" stroke="#000000" stroke-width="3"
stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 468 439.5 L 456.5 439.5 L 474 418.5 L 491.5 439.5 L 480 439.5" fill="none" stroke="#000000" stroke-width="3" stroke-linejoin="flat" stroke-miterlimit="4" pointer-events="none"/><path d="M 325.5 383 L 325.5 371 L 335 371 Q 347 371 347 383 L 347 851.5 L 358.5 851.5 L 341 872.5 L 323.5 851.5 L 335 851.5 L 335 383 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 347 851.5 L 358.5 851.5 L 341 872.5 L 323.5 851.5 L 335 851.5" fill="none" stroke="#000000" stroke-width="3" stroke-linejoin="flat" stroke-miterlimit="4" pointer-events="none"/><path d="M 432.5 371 L 432.5 383 L 424 383 L 424 851.5 L 435.5 851.5 L 418 872.5 L 400.5 851.5 L 412 851.5 L 412 383 Q 412 371 424 371 Z" fill="#ffffff" stroke="#000000" stroke-width="3" stroke-linejoin="round" stroke-miterlimit="10" pointer-events="none"/><path d="M 424 851.5 L 435.5 851.5 L
418 872.5 L 400.5 851.5 L 412 851.5" fill="none" stroke="#000000" stroke-width="3" stroke-linejoin="flat" stroke-miterlimit="4" pointer-events="none"/><g transform="translate(217.5,776.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="100" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; white-space: nowrap;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">cisco-5-304<br />Grok parser output<br /></div></div></foreignObject><text x="50" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">cisco-5-304<br>Grok parser output<br></text></switch></g><rect x="328" y="215" width="102" height="102" fill="none" stroke="#000000" stroke-width="3" pointer-
events="none"/><g transform="translate(331.5,252.5)"><switch><foreignObject style="overflow:visible;" pointer-events="all" width="94" height="26" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility"><div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; width: 96px; white-space: nowrap; word-wrap: normal; text-align: center;"><div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">pix_syslog_router<br />Grok parser<br /></div></div></foreignObject><text x="47" y="19" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">pix_syslog_router<br>Grok parser<br></text></switch></g></g></svg>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/use-cases/parser_chaining/aggregated_parser_chaining_flow.xml
----------------------------------------------------------------------
diff --git a/use-cases/parser_chaining/aggregated_parser_chaining_flow.xml b/use-cases/parser_chaining/aggregated_parser_chaining_flow.xml
new file mode 100644
index 0000000..22cda9f
--- /dev/null
+++ b/use-cases/parser_chaining/aggregated_parser_chaining_flow.xml
@@ -0,0 +1,14 @@
+<!--
+ 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.
+ -->
+<!-- This is a draw.io diagram. You can load it from http://www.draw.io -->
+<mxfile userAgent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36" version="8.9.5" editor="www.draw.io" type="device"><diagram name="Page-1" id="c7558073-3199-34d8-9f00-42111426c3f3">7Vtbb6M6EP41eewKMBDy2Nt2pXOr1COd3acVAYdYJRiB06b7648dbMAXEppCk27al+KxMXi+b8YzYzIB16vNXRHmy79wDNOJY8WbCbiZOI5tzyz6j0leKokf2JUgKVDMBzWCB/QLciG/L1mjGJbSQIJxSlAuCyOcZTAikiwsCvwsD1vgVH5qHiZQEzxEYapL/0MxWVbSwJk28m8QJUvxZMviLz4Po8ekwOuMP2/igMX2r+pehWIuPr5chjF+bonA7QRcFxiT6mq1uYYp061QW3Xf147e+r0LmJE+N4DqhqcwXfOl/42+Iv5u5EXo43mJCHzIw4i1nynmE3AVpijJaDOij4IFFTzBgiCqw0vesUJxzCa4WuCMPPDJLNouSYEfa8UCNgKl6TVOMZ3mJsMZu0lfCV8cewzctER8ZXcQryApXugQ3gu86g5OQsB1/twgKni6bIHpc1nIOZTU8zZ6pBdclR04uJpec7T5Wb6UKU5+Un4wjTk+ZYfNFPJAcLFiGs9pT0t+CjiU9GkoS6jAHggV1/2yFxgbTEdCRliqGZoSZiVm0KT0mVdzdpWwqz/CxWO49UI5ijRYqBXn7DJ6SVEWb0EwITVnfgHGf85Trs3GW/yzJvRWyOVpOIfpVd2pIGIAaUlWYs4KVNFPmWRt/4xwd0BnALgTTdv3JSxtR8fSNljZzB8CS93MNGxgTD07b3JtZfi2JXyt0uAGke/8Pnb9g1kTZTRtZXQB
37lxbRtNn6DIIoWbS7ZD1epnL7hb+XQ9eF1E9R7J1UnCIoFinGtGqQ2DZYChFhYwDQl6kt/EBA5/xj1G9B0boxaelNNgqphqtQB+U3s/Uudx5Xl8IM9TrVmbZ8uUetk9yeMfgTzMBRtHDkYPYCCHZU2Oyg7bkR2+6sj7sqMOm5SNYwR2CD+2ix0mH9/yxGGZV7HpAm1gvNet99qNu/gzgCevbU14cuHZWywBJpbUwjf58jqQ2KFwk5XALOaXN1EaliXdniUU2kpsovFdVtul48a9N97+h/D2fFPgXXxbsI2wHuj2XYNhVyzdadieATFvIKevWKNqjn3NOpDnqdsjmLUh4RHBHg10MxHuRaiM8IV3AViEUXXTqdsjPkPE/iGiZ0v4eo4h3h8tRnRMMWIX4j5F3PlE/K2IewFQUrx3xtzTMJct+qMm3m+zQzCTo7DgPfNuRw+3ZZs7S0y8qXVMTKa/Q8ylJ+KGpL2Vs+GCLHGCszBtZXKHRmVeYAjLnP3Z+Ihhmafk0K4an/eNy7SJxgzMgpOnogjxB6HbLmIPSkVwTCrW84gUYabUc3pn/kqOUMcTY1BxdipU7F8/Oiw3PZBoxhpTFXOdSCoa2MOkolN1mx2QZqBHhem8aWaqeFRh5NH82UwpZDpKutLbn6kTzUYkmh7lwaxA0XJF1V9+prW9g3UgdljhZUzB+mhJLdC3JUPRooHxjmqgWl7JDpstmk3l24RKGachTBVEZD0XsES/wnmdK+WMltu1eFcT74blXWuCy+ozDruVhqVwQQxJGGVWk0z9yxo3F+4OtzMAdq6SaE0DT8fOMWCnFjUP+xxAP3SWChLWCpZlmECDNZ4HPJ7yRYBnsCzXgI4aBh6Gjn70IFnW2aPjK2HRO8OjV/Muk6SASUhgTOWiYHQvPN0cp6QHMPLu0b3h9K8pmTbCotoIOfh7drvxIFT3LjvQT/eMe9cwEHYV/6Qjls69a6/V/WYnsVMZK+AGGlaBA
Sq1PHMYVHq0+Jow48ygcv1jQnUyNbNXZJA9vp6Sc0t7zPKtOCmSv6U6ao4J7JlyeOb7B2aZwNemspWpBswz3Y9YN+ss6h6Hjr6JjtNj0tENbJVDakjQ+zwBqFN56neCA9LR6/Fx6MnRsSGdLRfYvlh1+x4WiCqChaJ7zsFG46lr4ik4Kk+VCrBteQeyNFAnUo/PhuSontp8JI4Gr+aowc+OxtHpJ0cH4ejbcrczrzuqJ6DvW3f09AzB9Dukz5SuXSnxHQkxxzFUSixHR6wWvgIy2mx+7FdZZ/OLSnD7Pw==</diagram></mxfile>
\ No newline at end of file
[12/15] metron git commit: METRON-1657 Parser aggregation in storm
(justinleet) closes apache/metron#1099
Posted by mm...@apache.org.
METRON-1657 Parser aggregation in storm (justinleet) closes apache/metron#1099
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/f4345383
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/f4345383
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/f4345383
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: f43453830469b7fad14b59c95ad8122f320a4b05
Parents: 3a2675e
Author: justinleet <ju...@gmail.com>
Authored: Wed Jul 18 08:54:43 2018 -0400
Committer: leet <le...@apache.org>
Committed: Wed Jul 18 08:54:43 2018 -0400
----------------------------------------------------------------------
.../service/impl/StormAdminServiceImpl.java | 23 +-
.../rest/service/impl/StormCLIWrapper.java | 25 +-
.../common/bolt/ConfiguredParserBolt.java | 10 +-
.../configuration/SensorParserConfig.java | 6 +
.../apache/metron/common/error/MetronError.java | 64 ++--
.../common/bolt/ConfiguredParserBoltTest.java | 2 +-
.../metron/common/error/MetronErrorTest.java | 16 +-
metron-platform/metron-parsers/README.md | 25 +-
.../apache/metron/parsers/bolt/ParserBolt.java | 241 ++++++++------
.../apache/metron/parsers/bolt/WriterBolt.java | 3 +-
.../parsers/topology/ParserComponents.java | 67 ++++
.../parsers/topology/ParserTopologyBuilder.java | 200 ++++++++----
.../parsers/topology/ParserTopologyCLI.java | 264 ++++++++++-----
.../parsers/topology/config/ValueSupplier.java | 3 +-
.../metron/parsers/bolt/ParserBoltTest.java | 182 +++++++----
.../metron/parsers/bolt/WriterBoltTest.java | 30 +-
.../parsers/integration/ParserDriver.java | 50 ++-
.../components/ParserTopologyComponent.java | 58 ++--
.../parsers/topology/ParserTopologyCLITest.java | 318 +++++++++++++------
...pleHbaseEnrichmentWriterIntegrationTest.java | 3 +-
.../integration/WriterBoltIntegrationTest.java | 150 ++++++++-
.../metron/writer/BulkWriterComponent.java | 3 +-
.../metron/writer/BulkWriterComponentTest.java | 34 +-
use-cases/parser_chaining/README.md | 14 +
.../aggregated_parser_chaining_flow.svg | 14 +
.../aggregated_parser_chaining_flow.xml | 14 +
26 files changed, 1259 insertions(+), 560 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormAdminServiceImpl.java
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormAdminServiceImpl.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormAdminServiceImpl.java
index 40b01f1..4569a23 100644
--- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormAdminServiceImpl.java
+++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormAdminServiceImpl.java
@@ -50,17 +50,28 @@ public class StormAdminServiceImpl implements StormAdminService {
TopologyResponse topologyResponse = new TopologyResponse();
if (globalConfigService.get() == null) {
topologyResponse.setErrorMessage(TopologyStatusCode.GLOBAL_CONFIG_MISSING.toString());
- } else if (sensorParserConfigService.findOne(name) == null) {
- topologyResponse.setErrorMessage(TopologyStatusCode.SENSOR_PARSER_CONFIG_MISSING.toString());
- } else {
- topologyResponse = createResponse(stormCLIClientWrapper.startParserTopology(name), TopologyStatusCode.STARTED, TopologyStatusCode.START_ERROR);
+ return topologyResponse;
}
- return topologyResponse;
+
+ String[] sensorTypes = name.split(",");
+ for (String sensorType : sensorTypes) {
+ if (sensorParserConfigService.findOne(sensorType.trim()) == null) {
+ topologyResponse
+ .setErrorMessage(TopologyStatusCode.SENSOR_PARSER_CONFIG_MISSING.toString());
+ return topologyResponse;
+ }
+ }
+
+ return createResponse(
+ stormCLIClientWrapper.startParserTopology(name),
+ TopologyStatusCode.STARTED,
+ TopologyStatusCode.START_ERROR
+ );
}
@Override
public TopologyResponse stopParserTopology(String name, boolean stopNow) throws RestException {
- return createResponse(stormCLIClientWrapper.stopParserTopology(name, stopNow), TopologyStatusCode.STOPPED, TopologyStatusCode.STOP_ERROR);
+ return createResponse(stormCLIClientWrapper.stopParserTopology(name.replaceAll(",", "__"), stopNow), TopologyStatusCode.STOPPED, TopologyStatusCode.STOP_ERROR);
}
@Override
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java
index 26049dd..86e3512 100644
--- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java
+++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java
@@ -17,14 +17,8 @@
*/
package org.apache.metron.rest.service.impl;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.metron.common.utils.KafkaUtils;
-import org.apache.metron.rest.MetronRestConstants;
-import org.apache.metron.rest.RestException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.core.env.Environment;
+import static java.util.stream.Collectors.toList;
+import static org.apache.metron.rest.MetronRestConstants.ENRICHMENT_TOPOLOGY_NAME;
import java.io.BufferedReader;
import java.io.IOException;
@@ -34,9 +28,14 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-
-import static java.util.stream.Collectors.toList;
-import static org.apache.metron.rest.MetronRestConstants.ENRICHMENT_TOPOLOGY_NAME;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.metron.common.utils.KafkaUtils;
+import org.apache.metron.rest.MetronRestConstants;
+import org.apache.metron.rest.RestException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.env.Environment;
public class StormCLIWrapper {
@@ -99,13 +98,13 @@ public class StormCLIWrapper {
return exitValue;
}
- protected String[] getParserStartCommand(String name) {
+ protected String[] getParserStartCommand(String names) {
List<String> command = new ArrayList<>();
command.add( environment.getProperty(MetronRestConstants.PARSER_SCRIPT_PATH_SPRING_PROPERTY));
// sensor type
command.add( "-s");
- command.add( name);
+ command.add( names);
// zookeeper
command.add( "-z");
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java
index 1cb4e2e..14ce50b 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java
@@ -28,18 +28,12 @@ public abstract class ConfiguredParserBolt extends ConfiguredBolt<ParserConfigur
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
protected final ParserConfigurations configurations = new ParserConfigurations();
- private String sensorType;
- public ConfiguredParserBolt(String zookeeperUrl, String sensorType) {
+ public ConfiguredParserBolt(String zookeeperUrl) {
super(zookeeperUrl, "PARSERS");
- this.sensorType = sensorType;
}
- protected SensorParserConfig getSensorParserConfig() {
+ protected SensorParserConfig getSensorParserConfig(String sensorType) {
return getConfigurations().getSensorParserConfig(sensorType);
}
- public String getSensorType() {
- return sensorType;
- }
-
}
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
index 4cc4e61..d3eca42 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/SensorParserConfig.java
@@ -125,6 +125,7 @@ public class SensorParserConfig implements Serializable {
/**
* The parallelism of the Kafka spout.
+ * If multiple sensors are specified, each sensor will use it's own configured value.
*
* <p>This property can be overridden on the CLI.
*/
@@ -132,6 +133,7 @@ public class SensorParserConfig implements Serializable {
/**
* The number of tasks for the Kafka spout.
+ * If multiple sensors are specified, each sensor will use it's own configured value.
*
* <p>This property can be overridden on the CLI.
*/
@@ -139,6 +141,7 @@ public class SensorParserConfig implements Serializable {
/**
* The parallelism of the parser bolt.
+ * If multiple sensors are defined, the last one's config will win.
*
* <p>This property can be overridden on the CLI.
*/
@@ -146,6 +149,7 @@ public class SensorParserConfig implements Serializable {
/**
* The number of tasks for the parser bolt.
+ * If multiple sensors are defined, the last one's config will win.
*
* <p>This property can be overridden on the CLI.
*/
@@ -174,6 +178,7 @@ public class SensorParserConfig implements Serializable {
/**
* The Kafka security protocol.
+ * If multiple sensors are defined, any non PLAINTEXT configuration will be used.
*
* <p>This property can be overridden on the CLI. This property can also be overridden by the spout config.
*/
@@ -199,6 +204,7 @@ public class SensorParserConfig implements Serializable {
/**
* Configures the cache that backs stellar field transformations.
+ * If there are multiple sensors, the configs are merged, and the last non-empty config wins.
*
* <ul>
* <li>stellar.cache.maxSize - The maximum number of elements in the cache.
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java
index bc02c5c..0493be6 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java
@@ -17,26 +17,31 @@
*/
package org.apache.metron.common.error;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.metron.common.Constants.ERROR_TYPE;
+import static org.apache.metron.common.Constants.ErrorFields;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.metron.common.Constants;
import org.apache.metron.common.Constants.ErrorType;
import org.apache.metron.common.utils.HashUtils;
+import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.*;
-
-import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.apache.metron.common.Constants.ERROR_TYPE;
-import static org.apache.metron.common.Constants.ErrorFields;
-
public class MetronError {
private String message;
private Throwable throwable;
- private String sensorType = ERROR_TYPE;
+ private Set<String> sensorTypes = Collections.singleton(ERROR_TYPE);
private ErrorType errorType = ErrorType.DEFAULT_ERROR;
private Set<String> errorFields;
private List<Object> rawMessages;
@@ -51,8 +56,8 @@ public class MetronError {
return this;
}
- public MetronError withSensorType(String sensorType) {
- this.sensorType = sensorType;
+ public MetronError withSensorType(Set<String> sensorTypes) {
+ this.sensorTypes = sensorTypes;
return this;
}
@@ -91,7 +96,12 @@ public class MetronError {
JSONObject errorMessage = new JSONObject();
errorMessage.put(Constants.GUID, UUID.randomUUID().toString());
errorMessage.put(Constants.SENSOR_TYPE, "error");
- errorMessage.put(ErrorFields.FAILED_SENSOR_TYPE.getName(), sensorType);
+ if (sensorTypes.size() == 1) {
+ errorMessage.put(ErrorFields.FAILED_SENSOR_TYPE.getName(), sensorTypes.iterator().next());
+ } else {
+ errorMessage
+ .put(ErrorFields.FAILED_SENSOR_TYPE.getName(), new JSONArray().addAll(sensorTypes));
+ }
errorMessage.put(ErrorFields.ERROR_TYPE.getName(), errorType.getType());
addMessageString(errorMessage);
@@ -184,34 +194,42 @@ public class MetronError {
@Override
public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
MetronError that = (MetronError) o;
- if (message != null ? !message.equals(that.message) : that.message != null)
+ if (message != null ? !message.equals(that.message) : that.message != null) {
return false;
- if (throwable != null ? !throwable.equals(that.throwable) : that.throwable != null)
+ }
+ if (getThrowable() != null ? !getThrowable().equals(that.getThrowable())
+ : that.getThrowable() != null) {
return false;
- if (sensorType != null ? !sensorType.equals(that.sensorType) : that.sensorType != null)
+ }
+ if (sensorTypes != null ? !sensorTypes.equals(that.sensorTypes) : that.sensorTypes != null) {
return false;
- if (errorType != null ? !errorType.equals(that.errorType) : that.errorType != null)
+ }
+ if (errorType != that.errorType) {
return false;
- if (errorFields != null ? !errorFields.equals(that.errorFields) : that.errorFields != null)
+ }
+ if (errorFields != null ? !errorFields.equals(that.errorFields) : that.errorFields != null) {
return false;
+ }
return rawMessages != null ? rawMessages.equals(that.rawMessages) : that.rawMessages == null;
-
}
@Override
public int hashCode() {
int result = message != null ? message.hashCode() : 0;
- result = 31 * result + (throwable != null ? throwable.hashCode() : 0);
- result = 31 * result + (sensorType != null ? sensorType.hashCode() : 0);
+ result = 31 * result + (getThrowable() != null ? getThrowable().hashCode() : 0);
+ result = 31 * result + (sensorTypes != null ? sensorTypes.hashCode() : 0);
result = 31 * result + (errorType != null ? errorType.hashCode() : 0);
result = 31 * result + (errorFields != null ? errorFields.hashCode() : 0);
result = 31 * result + (rawMessages != null ? rawMessages.hashCode() : 0);
return result;
}
-
}
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java
index 27b0469..3deba78 100644
--- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java
+++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java
@@ -46,7 +46,7 @@ public class ConfiguredParserBoltTest extends BaseConfiguredBoltTest {
public static class StandAloneConfiguredParserBolt extends ConfiguredParserBolt {
public StandAloneConfiguredParserBolt(String zookeeperUrl) {
- super(zookeeperUrl, null);
+ super(zookeeperUrl);
}
@Override
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-common/src/test/java/org/apache/metron/common/error/MetronErrorTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/error/MetronErrorTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/error/MetronErrorTest.java
index e7390de..177a232 100644
--- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/error/MetronErrorTest.java
+++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/error/MetronErrorTest.java
@@ -17,20 +17,20 @@
*/
package org.apache.metron.common.error;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
import com.google.common.collect.Sets;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Collections;
import org.apache.commons.lang3.StringUtils;
import org.apache.metron.common.Constants;
import org.json.simple.JSONObject;
import org.junit.Before;
import org.junit.Test;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.Arrays;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
public class MetronErrorTest {
private JSONObject message1 = new JSONObject();
@@ -47,7 +47,7 @@ public class MetronErrorTest {
MetronError error = new MetronError()
.withMessage("test message")
.withErrorType(Constants.ErrorType.PARSER_ERROR)
- .withSensorType("sensorType");
+ .withSensorType(Collections.singleton("sensorType"));
JSONObject errorJSON = error.getJSONObject();
assertEquals("test message", errorJSON.get(Constants.ErrorFields.MESSAGE.getName()));
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/README.md b/metron-platform/metron-parsers/README.md
index 43bcc4a..0e428e3 100644
--- a/metron-platform/metron-parsers/README.md
+++ b/metron-platform/metron-parsers/README.md
@@ -82,6 +82,13 @@ topology in kafka. Errors are collected with the context of the error
(e.g. stacktrace) and original message causing the error and sent to an
`error` queue. Invalid messages as determined by global validation
functions are also treated as errors and sent to an `error` queue.
+
+Multiple sensors can be aggregated into a single Storm topology. When this is done, there will be
+multiple Kafka spouts, but only a single parser bolt which will handle delegating to the correct
+parser as needed. There are some constraints around this, in particular regarding some configuration.
+Additionally, all sensors must flow to the same error topic. The Kafka topic is retrieved from the input Tuple itself.
+
+A worked example of this can be found in the [Parser Chaining use case](../../use-cases/parser_chaining/README.md#aggregated-parsers-with-parser-chaining).
## Message Format
@@ -101,7 +108,7 @@ Where appropriate there is also a standardization around the 5-tuple JSON fields
* timestamp (epoch)
* original_string: A human friendly string representation of the message
-The timestamp and original_string fields are madatory. The remaining standard fields are optional. If any of the optional fields are not applicable then the field should be left out of the JSON.
+The timestamp and original_string fields are mandatory. The remaining standard fields are optional. If any of the optional fields are not applicable then the field should be left out of the JSON.
So putting it all together a typical Metron message with all 5-tuple fields present would look like the following:
@@ -138,6 +145,8 @@ the following fields:
* `raw_message_bytes` : The raw message bytes
* `error_hash` : A hash of the error message
+When aggregating multiple sensors, all sensors must be using the same error topic.
+
## Parser Configuration
The configuration for the various parser topologies is defined by JSON
@@ -170,16 +179,16 @@ then it is assumed to be a regex and will match any topic matching the pattern (
parameter `topology.message.timeout.secs`. Ignored if batchSize is `1`, since this disables batching.
* The kafka writer can be configured within the parser config as well. (This is all configured a priori, but this is convenient for overriding the settings). See [here](../metron-writer/README.md#kafka-writer)
* `fieldTransformations` : An array of complex objects representing the transformations to be done on the message generated from the parser before writing out to the kafka topic.
-* `spoutParallelism` : The kafka spout parallelism (default to `1`). This can be overridden on the command line.
-* `spoutNumTasks` : The number of tasks for the spout (default to `1`). This can be overridden on the command line.
-* `parserParallelism` : The parser bolt parallelism (default to `1`). This can be overridden on the command line.
-* `parserNumTasks` : The number of tasks for the parser bolt (default to `1`). This can be overridden on the command line.
+* `spoutParallelism` : The kafka spout parallelism (default to `1`). This can be overridden on the command line, and if there are multiple sensors should be in a comma separated list in the same order as the sensors.
+* `spoutNumTasks` : The number of tasks for the spout (default to `1`). This can be overridden on the command line, and if there are multiple sensors should be in a comma separated list in the same order as the sensors.
+* `parserParallelism` : The parser bolt parallelism (default to `1`). If there are multiple sensors, the last one's configuration will be used. This can be overridden on the command line.
+* `parserNumTasks` : The number of tasks for the parser bolt (default to `1`). If there are multiple sensors, the last one's configuration will be used. This can be overridden on the command line.
* `errorWriterParallelism` : The error writer bolt parallelism (default to `1`). This can be overridden on the command line.
* `errorWriterNumTasks` : The number of tasks for the error writer bolt (default to `1`). This can be overridden on the command line.
* `numWorkers` : The number of workers to use in the topology (default is the storm default of `1`).
* `numAckers` : The number of acker executors to use in the topology (default is the storm default of `1`).
-* `spoutConfig` : A map representing a custom spout config (this is a map). This can be overridden on the command line.
-* `securityProtocol` : The security protocol to use for reading from kafka (this is a string). This can be overridden on the command line and also specified in the spout config via the `security.protocol` key. If both are specified, then they are merged and the CLI will take precedence.
+* `spoutConfig` : A map representing a custom spout config (this is a map). If there are multiple sensors, the configs will be merged with the last specified taking precedence. This can be overridden on the command line.
+* `securityProtocol` : The security protocol to use for reading from kafka (this is a string). This can be overridden on the command line and also specified in the spout config via the `security.protocol` key. If both are specified, then they are merged and the CLI will take precedence. If multiple sensors are used, any non "PLAINTEXT" value will be used.
* `stormConfig` : The storm config to use (this is a map). This can be overridden on the command line. If both are specified, they are merged with CLI properties taking precedence.
* `cacheConfig` : Cache config for stellar field transformations. This configures a least frequently used cache. This is a map with the following keys. If not explicitly configured (the default), then no cache will be used.
* `stellar.cache.maxSize` - The maximum number of elements in the cache. Default is to not use a cache.
@@ -598,6 +607,8 @@ and pass `--extra_topology_options custom_config.json` to `start_parser_topology
Default installed Metron is untuned for production deployment. There
are a few knobs to tune to get the most out of your system.
+When using aggregated parsers, it's highly recommended to aggregate parsers with similar velocity and parser complexity together.
+
# Notes on Adding a New Sensor
In order to allow for meta alerts to be queries alongside regular alerts in Elasticsearch 2.x,
it is necessary to add an additional field to the templates and mapping for existing sensors.
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
index f68c670..213d02c 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
@@ -19,7 +19,6 @@ package org.apache.metron.parsers.bolt;
import com.github.benmanes.caffeine.cache.Cache;
-import java.io.IOException;
import java.io.Serializable;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
@@ -27,6 +26,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
@@ -42,15 +42,17 @@ import org.apache.metron.common.configuration.writer.WriterConfiguration;
import org.apache.metron.common.error.MetronError;
import org.apache.metron.common.message.MessageGetStrategy;
import org.apache.metron.common.message.MessageGetters;
+import org.apache.metron.common.message.metadata.RawMessage;
import org.apache.metron.common.message.metadata.RawMessageUtil;
import org.apache.metron.common.utils.ErrorUtils;
import org.apache.metron.parsers.filters.Filters;
import org.apache.metron.parsers.interfaces.MessageFilter;
import org.apache.metron.parsers.interfaces.MessageParser;
-import org.apache.metron.common.message.metadata.RawMessage;
+import org.apache.metron.parsers.topology.ParserComponents;
import org.apache.metron.stellar.common.CachingStellarProcessor;
import org.apache.metron.stellar.dsl.Context;
import org.apache.metron.stellar.dsl.StellarFunctions;
+import org.apache.metron.storm.kafka.flux.SimpleStormKafkaBuilder.FieldsConfiguration;
import org.apache.metron.writer.WriterToBulkWriter;
import org.apache.metron.writer.bolt.BatchTimeoutHelper;
import org.apache.storm.Config;
@@ -69,10 +71,9 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private OutputCollector collector;
- private MessageParser<JSONObject> parser;
- //default filter is noop, so pass everything through.
- private MessageFilter<JSONObject> filter;
- private WriterHandler writer;
+ private Map<String, ParserComponents> sensorToComponentMap;
+ private Map<String, String> topicToSensorMap = new HashMap<>();
+
private Context stellarContext;
private transient MessageGetStrategy messageGetStrategy;
private transient Cache<CachingStellarProcessor.Key, Object> cache;
@@ -81,20 +82,21 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
private int batchTimeoutDivisor = 1;
public ParserBolt( String zookeeperUrl
- , String sensorType
- , MessageParser<JSONObject> parser
- , WriterHandler writer
- )
- {
- super(zookeeperUrl, sensorType);
- this.writer = writer;
- this.parser = parser;
- }
+ , Map<String, ParserComponents> sensorToComponentMap
+ ) {
+ super(zookeeperUrl);
+ this.sensorToComponentMap = sensorToComponentMap;
-
- public ParserBolt withMessageFilter(MessageFilter<JSONObject> filter) {
- this.filter = filter;
- return this;
+ // Ensure that all sensors are either bulk sensors or not bulk sensors. Can't mix and match.
+ Boolean handleAcks = null;
+ for (Map.Entry<String, ParserComponents> entry : sensorToComponentMap.entrySet()) {
+ boolean writerHandleAck = entry.getValue().getWriter().handleAck();
+ if (handleAcks == null) {
+ handleAcks = writerHandleAck;
+ } else if (!handleAcks.equals(writerHandleAck)) {
+ throw new IllegalArgumentException("All writers must match when calling handleAck()");
+ }
+ }
}
/**
@@ -137,8 +139,8 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
return defaultBatchTimeout;
}
- public MessageParser<JSONObject> getParser() {
- return parser;
+ public Map<String, ParserComponents> getSensorToComponentMap() {
+ return sensorToComponentMap;
}
/**
@@ -153,14 +155,15 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
// to get the valid WriterConfiguration. But don't store any non-serializable objects,
// else Storm will throw a runtime error.
Function<WriterConfiguration, WriterConfiguration> configurationXform;
- if(writer.isWriterToBulkWriter()) {
+ WriterHandler writer = sensorToComponentMap.entrySet().iterator().next().getValue().getWriter();
+ if (writer.isWriterToBulkWriter()) {
configurationXform = WriterToBulkWriter.TRANSFORMATION;
- }
- else {
+ } else {
configurationXform = x -> x;
}
WriterConfiguration writerconf = configurationXform
- .apply(getConfigurationStrategy().createWriterConfig(writer.getBulkMessageWriter(), getConfigurations()));
+ .apply(getConfigurationStrategy()
+ .createWriterConfig(writer.getBulkMessageWriter(), getConfigurations()));
BatchTimeoutHelper timeoutHelper = new BatchTimeoutHelper(writerconf::getAllConfiguredTimeouts, batchTimeoutDivisor);
this.requestedTickFreqSecs = timeoutHelper.getRecommendedTickInterval();
@@ -182,40 +185,61 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
super.prepare(stormConf, context, collector);
messageGetStrategy = MessageGetters.DEFAULT_BYTES_FROM_POSITION.get();
this.collector = collector;
- if(getSensorParserConfig() != null) {
- cache = CachingStellarProcessor.createCache(getSensorParserConfig().getCacheConfig());
- }
- initializeStellar();
- if(getSensorParserConfig() != null && filter == null) {
- getSensorParserConfig().getParserConfig().putIfAbsent("stellarContext", stellarContext);
- if (!StringUtils.isEmpty(getSensorParserConfig().getFilterClassName())) {
- filter = Filters.get(getSensorParserConfig().getFilterClassName()
- , getSensorParserConfig().getParserConfig()
- );
+
+ // Build the Stellar cache
+ Map<String, Object> cacheConfig = new HashMap<>();
+ for (Map.Entry<String, ParserComponents> entry: sensorToComponentMap.entrySet()) {
+ String sensor = entry.getKey();
+ SensorParserConfig config = getSensorParserConfig(sensor);
+
+ if (config != null) {
+ cacheConfig.putAll(config.getCacheConfig());
}
}
+ cache = CachingStellarProcessor.createCache(cacheConfig);
- parser.init();
+ // Need to prep all sensors
+ for (Map.Entry<String, ParserComponents> entry: sensorToComponentMap.entrySet()) {
+ String sensor = entry.getKey();
+ MessageParser<JSONObject> parser = entry.getValue().getMessageParser();
- writer.init(stormConf, context, collector, getConfigurations());
- if (defaultBatchTimeout == 0) {
- //This means getComponentConfiguration was never called to initialize defaultBatchTimeout,
- //probably because we are in a unit test scenario. So calculate it here.
- WriterConfiguration writerConfig = getConfigurationStrategy()
- .createWriterConfig(writer.getBulkMessageWriter(), getConfigurations());
- BatchTimeoutHelper timeoutHelper = new BatchTimeoutHelper(writerConfig::getAllConfiguredTimeouts, batchTimeoutDivisor);
- defaultBatchTimeout = timeoutHelper.getDefaultBatchTimeout();
- }
- writer.setDefaultBatchTimeout(defaultBatchTimeout);
+ initializeStellar();
+ if (getSensorParserConfig(sensor) != null && sensorToComponentMap.get(sensor).getFilter() == null) {
+ getSensorParserConfig(sensor).getParserConfig().putIfAbsent("stellarContext", stellarContext);
+ if (!StringUtils.isEmpty(getSensorParserConfig(sensor).getFilterClassName())) {
+ MessageFilter<JSONObject> filter = Filters.get(
+ getSensorParserConfig(sensor).getFilterClassName(),
+ getSensorParserConfig(sensor).getParserConfig()
+ );
+ getSensorToComponentMap().get(sensor).setFilter(filter);
+ }
+ }
- SensorParserConfig config = getSensorParserConfig();
- if(config != null) {
- config.init();
- }
- else {
- throw new IllegalStateException("Unable to retrieve a parser config for " + getSensorType());
+ parser.init();
+
+ SensorParserConfig config = getSensorParserConfig(sensor);
+ if (config != null) {
+ config.init();
+ topicToSensorMap.put(config.getSensorTopic(), sensor);
+ } else {
+ throw new IllegalStateException(
+ "Unable to retrieve a parser config for " + sensor);
+ }
+ parser.configure(config.getParserConfig());
+
+ WriterHandler writer = sensorToComponentMap.get(sensor).getWriter();
+ writer.init(stormConf, context, collector, getConfigurations());
+ if (defaultBatchTimeout == 0) {
+ //This means getComponentConfiguration was never called to initialize defaultBatchTimeout,
+ //probably because we are in a unit test scenario. So calculate it here.
+ WriterConfiguration writerConfig = getConfigurationStrategy()
+ .createWriterConfig(writer.getBulkMessageWriter(), getConfigurations());
+ BatchTimeoutHelper timeoutHelper = new BatchTimeoutHelper(
+ writerConfig::getAllConfiguredTimeouts, batchTimeoutDivisor);
+ defaultBatchTimeout = timeoutHelper.getDefaultBatchTimeout();
+ }
+ writer.setDefaultBatchTimeout(defaultBatchTimeout);
}
- parser.configure(config.getParserConfig());
}
protected void initializeStellar() {
@@ -237,7 +261,9 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
public void execute(Tuple tuple) {
if (TupleUtils.isTick(tuple)) {
try {
- writer.flush(getConfigurations(), messageGetStrategy);
+ for (Entry<String, ParserComponents> entry : sensorToComponentMap.entrySet()) {
+ entry.getValue().getWriter().flush(getConfigurations(), messageGetStrategy);
+ }
} catch (Exception e) {
throw new RuntimeException(
"This should have been caught in the writerHandler. If you see this, file a JIRA", e);
@@ -246,69 +272,101 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
}
return;
}
- SensorParserConfig sensorParserConfig = getSensorParserConfig();
+
byte[] originalMessage = (byte[]) messageGetStrategy.get(tuple);
try {
- //we want to ack the tuple in the situation where we have are not doing a bulk write
- //otherwise we want to defer to the writerComponent who will ack on bulk commit.
- boolean ackTuple = !writer.handleAck();
+ SensorParserConfig sensorParserConfig;
+ MessageParser<JSONObject> parser;
+ String sensor;
+ Map<String, Object> metadata;
+ if (sensorToComponentMap.size() == 1) {
+ // There's only one parser, so grab info directly
+ Entry<String, ParserComponents> sensorParser = sensorToComponentMap.entrySet().iterator()
+ .next();
+ sensor = sensorParser.getKey();
+ parser = sensorParser.getValue().getMessageParser();
+ sensorParserConfig = getSensorParserConfig(sensor);
+ } else {
+ // There's multiple parsers, so pull the topic from the Tuple and look up the sensor
+ String topic = tuple.getStringByField(FieldsConfiguration.TOPIC.getFieldName());
+ sensor = topicToSensorMap.get(topic);
+ parser = sensorToComponentMap.get(sensor).getMessageParser();
+ sensorParserConfig = getSensorParserConfig(sensor);
+ }
+
+ List<FieldValidator> fieldValidations = getConfigurations().getFieldValidations();
+ boolean ackTuple = false;
int numWritten = 0;
- if(sensorParserConfig != null) {
+ if (sensorParserConfig != null) {
RawMessage rawMessage = RawMessageUtil.INSTANCE.getRawMessage( sensorParserConfig.getRawMessageStrategy()
- , tuple
- , originalMessage
- , sensorParserConfig.getReadMetadata()
- , sensorParserConfig.getRawMessageStrategyConfig()
- );
- Map<String, Object> metadata = rawMessage.getMetadata();
- List<FieldValidator> fieldValidations = getConfigurations().getFieldValidations();
+ , tuple
+ , originalMessage
+ , sensorParserConfig.getReadMetadata()
+ , sensorParserConfig.getRawMessageStrategyConfig()
+ );
+ metadata = rawMessage.getMetadata();
Optional<List<JSONObject>> messages = parser.parseOptional(rawMessage.getMessage());
for (JSONObject message : messages.orElse(Collections.emptyList())) {
- sensorParserConfig.getRawMessageStrategy().mergeMetadata( message
- , metadata
- , sensorParserConfig.getMergeMetadata()
- , sensorParserConfig.getRawMessageStrategyConfig()
- );
- message.put(Constants.SENSOR_TYPE, getSensorType());
+ //we want to ack the tuple in the situation where we have are not doing a bulk write
+ //otherwise we want to defer to the writerComponent who will ack on bulk commit.
+ WriterHandler writer = sensorToComponentMap.get(sensor).getWriter();
+ ackTuple = !writer.handleAck();
+
+ sensorParserConfig.getRawMessageStrategy().mergeMetadata(
+ message,
+ metadata,
+ sensorParserConfig.getMergeMetadata(),
+ sensorParserConfig.getRawMessageStrategyConfig()
+ );
+ message.put(Constants.SENSOR_TYPE, sensor);
+
for (FieldTransformer handler : sensorParserConfig.getFieldTransformations()) {
if (handler != null) {
- if(!sensorParserConfig.getMergeMetadata()) {
+ if (!sensorParserConfig.getMergeMetadata()) {
//if we haven't merged metadata, then we need to pass them along as configuration params.
- handler.transformAndUpdate(message, stellarContext, sensorParserConfig.getParserConfig(), metadata);
- }
- else {
- handler.transformAndUpdate(message, stellarContext, sensorParserConfig.getParserConfig());
+ handler.transformAndUpdate(
+ message,
+ stellarContext,
+ sensorParserConfig.getParserConfig(),
+ metadata
+ );
+ } else {
+ handler.transformAndUpdate(
+ message,
+ stellarContext,
+ sensorParserConfig.getParserConfig()
+ );
}
}
}
- if(!message.containsKey(Constants.GUID)) {
+ if (!message.containsKey(Constants.GUID)) {
message.put(Constants.GUID, UUID.randomUUID().toString());
}
+ MessageFilter<JSONObject> filter = sensorToComponentMap.get(sensor).getFilter();
if (filter == null || filter.emitTuple(message, stellarContext)) {
boolean isInvalid = !parser.validate(message);
List<FieldValidator> failedValidators = null;
- if(!isInvalid) {
+ if (!isInvalid) {
failedValidators = getFailedValidators(message, fieldValidations);
isInvalid = !failedValidators.isEmpty();
}
- if( isInvalid) {
+ if (isInvalid) {
MetronError error = new MetronError()
- .withErrorType(Constants.ErrorType.PARSER_INVALID)
- .withSensorType(getSensorType())
- .addRawMessage(message);
- Set<String> errorFields = failedValidators == null?null:failedValidators.stream()
- .flatMap(fieldValidator -> fieldValidator.getInput().stream())
- .collect(Collectors.toSet());
+ .withErrorType(Constants.ErrorType.PARSER_INVALID)
+ .withSensorType(Collections.singleton(sensor))
+ .addRawMessage(message);
+ Set<String> errorFields = failedValidators == null ? null : failedValidators.stream()
+ .flatMap(fieldValidator -> fieldValidator.getInput().stream())
+ .collect(Collectors.toSet());
if (errorFields != null && !errorFields.isEmpty()) {
error.withErrorFields(errorFields);
}
ErrorUtils.handleError(collector, error);
- }
- else {
+ } else {
numWritten++;
- writer.write(getSensorType(), tuple, message, getConfigurations(), messageGetStrategy);
+ writer.write(sensor, tuple, message, getConfigurations(), messageGetStrategy);
}
}
}
@@ -316,9 +374,10 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
//if we are supposed to ack the tuple OR if we've never passed this tuple to the bulk writer
//(meaning that none of the messages are valid either globally or locally)
//then we want to handle the ack ourselves.
- if(ackTuple || numWritten == 0) {
+ if (ackTuple || numWritten == 0) {
collector.ack(tuple);
}
+
} catch (Throwable ex) {
handleError(originalMessage, tuple, ex, collector);
}
@@ -328,7 +387,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
MetronError error = new MetronError()
.withErrorType(Constants.ErrorType.PARSER_ERROR)
.withThrowable(ex)
- .withSensorType(getSensorType())
+ .withSensorType(sensorToComponentMap.keySet())
.addRawMessage(originalMessage);
ErrorUtils.handleError(collector, error);
collector.ack(tuple);
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
index ef93ba2..fdfceda 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
@@ -18,6 +18,7 @@
package org.apache.metron.parsers.bolt;
+import java.util.Collections;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.ParserConfigurations;
import org.apache.metron.common.error.MetronError;
@@ -84,7 +85,7 @@ public class WriterBolt extends BaseRichBolt {
MetronError error = new MetronError()
.withErrorType(errorType)
.withThrowable(e)
- .withSensorType(sensorType)
+ .withSensorType(Collections.singleton(sensorType))
.addRawMessage(message);
ErrorUtils.handleError(collector, error);
collector.ack(tuple);
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserComponents.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserComponents.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserComponents.java
new file mode 100644
index 0000000..32d56b9
--- /dev/null
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserComponents.java
@@ -0,0 +1,67 @@
+/*
+ * 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.parsers.topology;
+
+import java.io.Serializable;
+import org.apache.metron.parsers.bolt.WriterHandler;
+import org.apache.metron.parsers.interfaces.MessageFilter;
+import org.apache.metron.parsers.interfaces.MessageParser;
+import org.json.simple.JSONObject;
+
+public class ParserComponents implements Serializable {
+ private static final long serialVersionUID = 7880346740026374665L;
+
+ private MessageParser<JSONObject> messageParser;
+ private MessageFilter<JSONObject> filter;
+ private WriterHandler writer;
+
+ public ParserComponents(
+ MessageParser<JSONObject> messageParser,
+ MessageFilter<JSONObject> filter,
+ WriterHandler writer) {
+ this.messageParser = messageParser;
+ this.filter = filter;
+ this.writer = writer;
+ }
+
+ public MessageParser<JSONObject> getMessageParser() {
+ return messageParser;
+ }
+
+ public MessageFilter<JSONObject> getFilter() {
+ return filter;
+ }
+
+ public WriterHandler getWriter() {
+ return writer;
+ }
+
+ public void setMessageParser(
+ MessageParser<JSONObject> messageParser) {
+ this.messageParser = messageParser;
+ }
+
+ public void setFilter(
+ MessageFilter<JSONObject> filter) {
+ this.filter = filter;
+ }
+
+ public void setWriter(WriterHandler writer) {
+ this.writer = writer;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
index 5b3e0d5..d20e1a5 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
@@ -17,35 +17,45 @@
*/
package org.apache.metron.parsers.topology;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.metron.common.utils.KafkaUtils;
-import org.apache.metron.parsers.topology.config.ValueSupplier;
-import org.apache.metron.storm.kafka.flux.SimpleStormKafkaBuilder;
-import org.apache.metron.storm.kafka.flux.SpoutConfiguration;
-import org.apache.metron.storm.kafka.flux.StormKafkaSpout;
-import org.apache.storm.Config;
-import org.apache.storm.kafka.spout.KafkaSpout;
-import org.apache.storm.kafka.spout.KafkaSpoutConfig;
-import org.apache.storm.topology.TopologyBuilder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.ConfigurationsUtils;
import org.apache.metron.common.configuration.ParserConfigurations;
import org.apache.metron.common.configuration.SensorParserConfig;
import org.apache.metron.common.configuration.writer.ParserWriterConfiguration;
+import org.apache.metron.common.utils.KafkaUtils;
+import org.apache.metron.common.utils.ReflectionUtils;
import org.apache.metron.common.writer.BulkMessageWriter;
import org.apache.metron.common.writer.MessageWriter;
-import org.apache.metron.common.utils.ReflectionUtils;
import org.apache.metron.parsers.bolt.ParserBolt;
import org.apache.metron.parsers.bolt.WriterBolt;
import org.apache.metron.parsers.bolt.WriterHandler;
+import org.apache.metron.parsers.filters.Filters;
+import org.apache.metron.parsers.interfaces.MessageFilter;
import org.apache.metron.parsers.interfaces.MessageParser;
+import org.apache.metron.parsers.topology.config.ValueSupplier;
+import org.apache.metron.storm.kafka.flux.SimpleStormKafkaBuilder;
+import org.apache.metron.storm.kafka.flux.SpoutConfiguration;
+import org.apache.metron.storm.kafka.flux.StormKafkaSpout;
import org.apache.metron.writer.AbstractWriter;
import org.apache.metron.writer.kafka.KafkaWriter;
+import org.apache.storm.Config;
+import org.apache.storm.kafka.spout.KafkaSpout;
+import org.apache.storm.kafka.spout.KafkaSpoutConfig;
+import org.apache.storm.topology.BoltDeclarer;
+import org.apache.storm.topology.TopologyBuilder;
import org.json.simple.JSONObject;
-import java.util.*;
-
/**
* Builds a Storm topology that parses telemetry data received from a sensor.
*/
@@ -75,7 +85,7 @@ public class ParserTopologyBuilder {
*
* @param zookeeperUrl Zookeeper URL
* @param brokerUrl Kafka Broker URL
- * @param sensorType Type of sensor
+ * @param sensorTypes Type of sensor
* @param spoutParallelismSupplier Supplier for the parallelism hint for the spout
* @param spoutNumTasksSupplier Supplier for the number of tasks for the spout
* @param parserParallelismSupplier Supplier for the parallelism hint for the parser bolt
@@ -91,14 +101,14 @@ public class ParserTopologyBuilder {
*/
public static ParserTopology build(String zookeeperUrl,
Optional<String> brokerUrl,
- String sensorType,
- ValueSupplier<Integer> spoutParallelismSupplier,
- ValueSupplier<Integer> spoutNumTasksSupplier,
+ List<String> sensorTypes,
+ ValueSupplier<List> spoutParallelismSupplier,
+ ValueSupplier<List> spoutNumTasksSupplier,
ValueSupplier<Integer> parserParallelismSupplier,
ValueSupplier<Integer> parserNumTasksSupplier,
ValueSupplier<Integer> errorWriterParallelismSupplier,
ValueSupplier<Integer> errorWriterNumTasksSupplier,
- ValueSupplier<Map> kafkaSpoutConfigSupplier,
+ ValueSupplier<List> kafkaSpoutConfigSupplier,
ValueSupplier<String> securityProtocolSupplier,
ValueSupplier<String> outputTopicSupplier,
ValueSupplier<String> errorTopicSupplier,
@@ -107,40 +117,72 @@ public class ParserTopologyBuilder {
// fetch configuration from zookeeper
ParserConfigurations configs = new ParserConfigurations();
- SensorParserConfig parserConfig = getSensorParserConfig(zookeeperUrl, sensorType, configs);
- int spoutParallelism = spoutParallelismSupplier.get(parserConfig, Integer.class);
- int spoutNumTasks = spoutNumTasksSupplier.get(parserConfig, Integer.class);
- int parserParallelism = parserParallelismSupplier.get(parserConfig, Integer.class);
- int parserNumTasks = parserNumTasksSupplier.get(parserConfig, Integer.class);
- int errorWriterParallelism = errorWriterParallelismSupplier.get(parserConfig, Integer.class);
- int errorWriterNumTasks = errorWriterNumTasksSupplier.get(parserConfig, Integer.class);
- String outputTopic = outputTopicSupplier.get(parserConfig, String.class);
-
- Map<String, Object> kafkaSpoutConfig = kafkaSpoutConfigSupplier.get(parserConfig, Map.class);
- Optional<String> securityProtocol = Optional.ofNullable(securityProtocolSupplier.get(parserConfig, String.class));
+ Map<String, SensorParserConfig> sensorToParserConfigs = getSensorParserConfig(zookeeperUrl, sensorTypes, configs);
+ Collection<SensorParserConfig> parserConfigs = sensorToParserConfigs.values();
+
+ @SuppressWarnings("unchecked")
+ List<Integer> spoutParallelism = (List<Integer>) spoutParallelismSupplier.get(parserConfigs, List.class);
+ @SuppressWarnings("unchecked")
+ List<Integer> spoutNumTasks = (List<Integer>) spoutNumTasksSupplier.get(parserConfigs, List.class);
+ int parserParallelism = parserParallelismSupplier.get(parserConfigs, Integer.class);
+ int parserNumTasks = parserNumTasksSupplier.get(parserConfigs, Integer.class);
+ int errorWriterParallelism = errorWriterParallelismSupplier.get(parserConfigs, Integer.class);
+ int errorWriterNumTasks = errorWriterNumTasksSupplier.get(parserConfigs, Integer.class);
+ String outputTopic = outputTopicSupplier.get(parserConfigs, String.class);
+
+ List<Map<String, Object>> kafkaSpoutConfig = kafkaSpoutConfigSupplier.get(parserConfigs, List.class);
+ Optional<String> securityProtocol = Optional.ofNullable(securityProtocolSupplier.get(parserConfigs, String.class));
// create the spout
TopologyBuilder builder = new TopologyBuilder();
- KafkaSpout kafkaSpout = createKafkaSpout(zookeeperUrl, sensorType, securityProtocol, Optional.ofNullable(kafkaSpoutConfig), parserConfig);
- builder.setSpout("kafkaSpout", kafkaSpout, spoutParallelism)
- .setNumTasks(spoutNumTasks);
+ int i = 0;
+ List<String> spoutIds = new ArrayList<>();
+ for (Entry<String, SensorParserConfig> entry: sensorToParserConfigs.entrySet()) {
+ KafkaSpout kafkaSpout = createKafkaSpout(zookeeperUrl, entry.getKey(), securityProtocol,
+ Optional.ofNullable(kafkaSpoutConfig.get(i)), entry.getValue());
+ String spoutId = sensorToParserConfigs.size() > 1 ? "kafkaSpout-" + entry.getKey() : "kafkaSpout";
+ builder.setSpout(spoutId, kafkaSpout, spoutParallelism.get(i))
+ .setNumTasks(spoutNumTasks.get(i));
+ spoutIds.add(spoutId);
+ ++i;
+ }
// create the parser bolt
- ParserBolt parserBolt = createParserBolt(zookeeperUrl, brokerUrl, sensorType, securityProtocol, configs, parserConfig, Optional.of(outputTopic));
- builder.setBolt("parserBolt", parserBolt, parserParallelism)
- .setNumTasks(parserNumTasks)
- .localOrShuffleGrouping("kafkaSpout");
+ ParserBolt parserBolt = createParserBolt(
+ zookeeperUrl,
+ brokerUrl,
+ sensorToParserConfigs,
+ securityProtocol,
+ configs,
+ Optional.ofNullable(outputTopic)
+ );
+
+ BoltDeclarer boltDeclarer = builder
+ .setBolt("parserBolt", parserBolt, parserParallelism)
+ .setNumTasks(parserNumTasks);
+
+ for (String spoutId : spoutIds) {
+ boltDeclarer.localOrShuffleGrouping(spoutId);
+ }
// create the error bolt, if needed
if (errorWriterNumTasks > 0) {
- String errorTopic = errorTopicSupplier.get(parserConfig, String.class);
- WriterBolt errorBolt = createErrorBolt(zookeeperUrl, brokerUrl, sensorType, securityProtocol, configs, parserConfig, errorTopic);
+ String errorTopic = errorTopicSupplier.get(parserConfigs, String.class);
+ WriterBolt errorBolt = createErrorBolt(
+ zookeeperUrl,
+ brokerUrl,
+ sensorTypes.get(0),
+ securityProtocol,
+ configs,
+ parserConfigs.iterator().next(),
+ errorTopic
+ );
builder.setBolt("errorMessageWriter", errorBolt, errorWriterParallelism)
.setNumTasks(errorWriterNumTasks)
.localOrShuffleGrouping("parserBolt", Constants.ERROR_STREAM);
}
- return new ParserTopology(builder, stormConfigSupplier.get(parserConfig, Config.class));
+ return new ParserTopology(builder, stormConfigSupplier.get(parserConfigs, Config.class));
}
/**
@@ -216,42 +258,62 @@ public class ParserTopologyBuilder {
*
* @param zookeeperUrl Zookeeper URL
* @param brokerUrl Kafka Broker URL
- * @param sensorType Type of sensor that is being consumed.
+ * @param sensorTypeToParserConfig
* @param configs
- * @param parserConfig
* @return A Storm bolt that parses input from a sensor
*/
private static ParserBolt createParserBolt( String zookeeperUrl,
Optional<String> brokerUrl,
- String sensorType,
+ Map<String, SensorParserConfig> sensorTypeToParserConfig,
Optional<String> securityProtocol,
ParserConfigurations configs,
- SensorParserConfig parserConfig,
Optional<String> outputTopic) {
- // create message parser
- MessageParser<JSONObject> parser = ReflectionUtils.createInstance(parserConfig.getParserClassName());
- parser.configure(parserConfig.getParserConfig());
+ Map<String, ParserComponents> parserBoltConfigs = new HashMap<>();
+ for( Entry<String, SensorParserConfig> entry : sensorTypeToParserConfig.entrySet()) {
+ String sensorType = entry.getKey();
+ SensorParserConfig parserConfig = entry.getValue();
+ // create message parser
+ MessageParser<JSONObject> parser = ReflectionUtils
+ .createInstance(parserConfig.getParserClassName());
+ parser.configure(parserConfig.getParserConfig());
+
+ // create message filter
+ MessageFilter<JSONObject> filter = null;
+ if (!StringUtils.isEmpty(parserConfig.getFilterClassName())) {
+ filter = Filters.get(
+ parserConfig.getFilterClassName(),
+ parserConfig.getParserConfig()
+ );
+ }
- // create a writer
- AbstractWriter writer;
- if(parserConfig.getWriterClassName() == null) {
+ // create a writer
+ AbstractWriter writer;
+ if (parserConfig.getWriterClassName() == null) {
- // if not configured, use a sensible default
- writer = createKafkaWriter(brokerUrl, zookeeperUrl, securityProtocol)
- .withTopic(outputTopic.orElse(Constants.ENRICHMENT_TOPIC));
+ // if not configured, use a sensible default
+ writer = createKafkaWriter(brokerUrl, zookeeperUrl, securityProtocol)
+ .withTopic(outputTopic.orElse(Constants.ENRICHMENT_TOPIC));
- } else {
- writer = ReflectionUtils.createInstance(parserConfig.getWriterClassName());
- }
+ } else {
+ writer = ReflectionUtils.createInstance(parserConfig.getWriterClassName());
+ }
- // configure it
- writer.configure(sensorType, new ParserWriterConfiguration(configs));
+ // configure it
+ writer.configure(sensorType, new ParserWriterConfiguration(configs));
- // create a writer handler
- WriterHandler writerHandler = createWriterHandler(writer);
+ // create a writer handler
+ WriterHandler writerHandler = createWriterHandler(writer);
+
+ ParserComponents components = new ParserComponents(
+ parser,
+ filter,
+ writerHandler
+ );
+ parserBoltConfigs.put(sensorType, components);
+ }
- return new ParserBolt(zookeeperUrl, sensorType, parser, writerHandler);
+ return new ParserBolt(zookeeperUrl, parserBoltConfigs);
}
/**
@@ -304,22 +366,26 @@ public class ParserTopologyBuilder {
* Fetch the parser configuration from Zookeeper.
*
* @param zookeeperUrl Zookeeper URL
- * @param sensorType Type of sensor
+ * @param sensorTypes Types of sensor
* @param configs
* @return
* @throws Exception
*/
- private static SensorParserConfig getSensorParserConfig(String zookeeperUrl, String sensorType, ParserConfigurations configs) throws Exception {
+ private static Map<String, SensorParserConfig> getSensorParserConfig(String zookeeperUrl, List<String> sensorTypes, ParserConfigurations configs) throws Exception {
+ Map<String, SensorParserConfig> parserConfigs = new HashMap<>();
try(CuratorFramework client = ConfigurationsUtils.getClient(zookeeperUrl)) {
client.start();
ConfigurationsUtils.updateParserConfigsFromZookeeper(configs, client);
- SensorParserConfig parserConfig = configs.getSensorParserConfig(sensorType);
- if (parserConfig == null) {
- throw new IllegalStateException("Cannot find the parser configuration in zookeeper for " + sensorType + "." +
- " Please check that it exists in zookeeper by using the 'zk_load_configs.sh -m DUMP' command.");
+ for (String sensorType : sensorTypes) {
+ SensorParserConfig parserConfig = configs.getSensorParserConfig(sensorType);
+ if (parserConfig == null) {
+ throw new IllegalStateException("Cannot find the parser configuration in zookeeper for " + sensorType + "." +
+ " Please check that it exists in zookeeper by using the 'zk_load_configs.sh -m DUMP' command.");
+ }
+ parserConfigs.put(sensorType, parserConfig);
}
- return parserConfig;
}
+ return parserConfigs;
}
/**
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
index f60ff44..eb39f89 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
@@ -17,27 +17,44 @@
*/
package org.apache.metron.parsers.topology;
+import com.google.common.base.Joiner;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.PosixParser;
+import org.apache.commons.io.FileUtils;
+import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.metron.common.Constants;
+import org.apache.metron.common.configuration.SensorParserConfig;
+import org.apache.metron.common.utils.JSONUtils;
+import org.apache.metron.common.utils.KafkaUtils;
+import org.apache.metron.parsers.topology.config.Arg;
+import org.apache.metron.parsers.topology.config.ConfigHandlers;
import org.apache.metron.parsers.topology.config.ValueSupplier;
import org.apache.metron.storm.kafka.flux.SpoutConfiguration;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.utils.Utils;
-import com.google.common.base.Joiner;
-import org.apache.commons.cli.*;
-import org.apache.commons.io.FileUtils;
-import org.apache.metron.common.utils.JSONUtils;
-import org.apache.metron.parsers.topology.config.Arg;
-import org.apache.metron.parsers.topology.config.ConfigHandlers;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.*;
-import java.util.function.Function;
public class ParserTopologyCLI {
+ private static final String STORM_JOB_SEPARATOR = "__";
+
public enum ParserOptions {
HELP("h", code -> {
Option o = new Option(code, "help", false, "This screen");
@@ -45,7 +62,7 @@ public class ParserTopologyCLI {
return o;
}),
ZK_QUORUM("z", code -> {
- Option o = new Option(code, "zk", true, "Zookeeper Quroum URL (zk1:2181,zk2:2181,...");
+ Option o = new Option(code, "zk", true, "Zookeeper Quorum URL (zk1:2181,zk2:2181,...");
o.setArgName("ZK_QUORUM");
o.setRequired(true);
return o;
@@ -56,14 +73,14 @@ public class ParserTopologyCLI {
o.setRequired(false);
return o;
}),
- SENSOR_TYPE("s", code -> {
- Option o = new Option(code, "sensor", true, "Sensor Type");
- o.setArgName("SENSOR_TYPE");
+ SENSOR_TYPES("s", code -> {
+ Option o = new Option(code, "sensor", true, "Sensor Types as comma-separated list");
+ o.setArgName("SENSOR_TYPES");
o.setRequired(true);
return o;
}),
SPOUT_PARALLELISM("sp", code -> {
- Option o = new Option(code, "spout_p", true, "Spout Parallelism Hint");
+ Option o = new Option(code, "spout_p", true, "Spout Parallelism Hint. If multiple sensors are specified, this should be a comma separated list in the same order.");
o.setArgName("SPOUT_PARALLELISM_HINT");
o.setRequired(false);
o.setType(Number.class);
@@ -91,7 +108,7 @@ public class ParserTopologyCLI {
return o;
}),
SPOUT_NUM_TASKS("snt", code -> {
- Option o = new Option(code, "spout_num_tasks", true, "Spout Num Tasks");
+ Option o = new Option(code, "spout_num_tasks", true, "Spout Num Tasks. If multiple sensors are specified, this should be a comma separated list in the same order.");
o.setArgName("NUM_TASKS");
o.setRequired(false);
o.setType(Number.class);
@@ -307,7 +324,9 @@ public class ParserTopologyCLI {
public ParserTopologyBuilder.ParserTopology createParserTopology(final CommandLine cmd) throws Exception {
String zookeeperUrl = ParserOptions.ZK_QUORUM.get(cmd);
Optional<String> brokerUrl = ParserOptions.BROKER_URL.has(cmd)?Optional.of(ParserOptions.BROKER_URL.get(cmd)):Optional.empty();
- String sensorType= ParserOptions.SENSOR_TYPE.get(cmd);
+ String sensorTypeRaw= ParserOptions.SENSOR_TYPES.get(cmd);
+ List<String> sensorTypes = Arrays.stream(sensorTypeRaw.split(",")).map(String::trim).collect(
+ Collectors.toList());
/*
* It bears mentioning why we're creating this ValueSupplier indirection here.
@@ -328,116 +347,191 @@ public class ParserTopologyCLI {
*/
// kafka spout parallelism
- ValueSupplier<Integer> spoutParallelism = (parserConfig, clazz) -> {
+ ValueSupplier<List> spoutParallelism = (parserConfigs, clazz) -> {
if(ParserOptions.SPOUT_PARALLELISM.has(cmd)) {
- return Integer.parseInt(ParserOptions.SPOUT_PARALLELISM.get(cmd, "1"));
+ // Handle the case where there's only one and we can default reasonably
+ if( parserConfigs.size() == 1) {
+ return Collections.singletonList(Integer.parseInt(ParserOptions.SPOUT_PARALLELISM.get(cmd, "1")));
+ }
+
+ // Handle the multiple explicitly passed spout parallelism's case.
+ String parallelismRaw = ParserOptions.SPOUT_PARALLELISM.get(cmd, "1");
+ List<String> parallelisms = Arrays.stream(parallelismRaw.split(",")).map(String::trim).collect(
+ Collectors.toList());
+ if (parallelisms.size() != parserConfigs.size()) {
+ throw new IllegalArgumentException("Spout parallelism should match number of sensors 1:1");
+ }
+ List<Integer> spoutParallelisms = new ArrayList<>();
+ for (String s : parallelisms) {
+ spoutParallelisms.add(Integer.parseInt(s));
+ }
+ return spoutParallelisms;
+ }
+
+ List<Integer> spoutParallelisms = new ArrayList<>();
+ for (SensorParserConfig parserConfig : parserConfigs) {
+ spoutParallelisms.add(parserConfig.getSpoutParallelism());
}
- return Optional.ofNullable(parserConfig.getSpoutParallelism()).orElse(1);
+ return spoutParallelisms;
};
// kafka spout number of tasks
- ValueSupplier<Integer> spoutNumTasks = (parserConfig, clazz) -> {
+ ValueSupplier<List> spoutNumTasks = (parserConfigs, clazz) -> {
if(ParserOptions.SPOUT_NUM_TASKS.has(cmd)) {
- return Integer.parseInt(ParserOptions.SPOUT_NUM_TASKS.get(cmd, "1"));
+ // Handle the case where there's only one and we can default reasonably
+ if( parserConfigs.size() == 1) {
+ return Collections.singletonList(Integer.parseInt(ParserOptions.SPOUT_NUM_TASKS.get(cmd, "1")));
+ }
+
+ // Handle the multiple explicitly passed spout parallelism's case.
+ String numTasksRaw = ParserOptions.SPOUT_NUM_TASKS.get(cmd, "1");
+ List<String> numTasks = Arrays.stream(numTasksRaw.split(",")).map(String::trim).collect(
+ Collectors.toList());
+ if (numTasks.size() != parserConfigs.size()) {
+ throw new IllegalArgumentException("Spout num tasks should match number of sensors 1:1");
+ }
+ List<Integer> spoutTasksList = new ArrayList<>();
+ for (String s : numTasks) {
+ spoutTasksList.add(Integer.parseInt(s));
+ }
+ return spoutTasksList;
+ }
+
+ List<Integer> numTasks = new ArrayList<>();
+ for (SensorParserConfig parserConfig : parserConfigs) {
+ numTasks.add(parserConfig.getSpoutNumTasks());
}
- return Optional.ofNullable(parserConfig.getSpoutNumTasks()).orElse(1);
+ return numTasks;
};
// parser bolt parallelism
- ValueSupplier<Integer> parserParallelism = (parserConfig, clazz) -> {
+ ValueSupplier<Integer> parserParallelism = (parserConfigs, clazz) -> {
if(ParserOptions.PARSER_PARALLELISM.has(cmd)) {
return Integer.parseInt(ParserOptions.PARSER_PARALLELISM.get(cmd, "1"));
}
- return Optional.ofNullable(parserConfig.getParserParallelism()).orElse(1);
+ int retValue = 1;
+ for (SensorParserConfig config : parserConfigs) {
+ Integer configValue = config.getParserParallelism();
+ retValue = configValue == null ? retValue : configValue;
+ }
+ return retValue;
};
// parser bolt number of tasks
- ValueSupplier<Integer> parserNumTasks = (parserConfig, clazz) -> {
+ ValueSupplier<Integer> parserNumTasks = (parserConfigs, clazz) -> {
if(ParserOptions.PARSER_NUM_TASKS.has(cmd)) {
return Integer.parseInt(ParserOptions.PARSER_NUM_TASKS.get(cmd, "1"));
}
- return Optional.ofNullable(parserConfig.getParserNumTasks()).orElse(1);
+ int retValue = 1;
+ for (SensorParserConfig config : parserConfigs) {
+ Integer configValue = config.getParserNumTasks();
+ retValue = configValue == null ? retValue : configValue;
+ }
+ return retValue;
};
// error bolt parallelism
- ValueSupplier<Integer> errorParallelism = (parserConfig, clazz) -> {
+ ValueSupplier<Integer> errorParallelism = (parserConfigs, clazz) -> {
if(ParserOptions.ERROR_WRITER_PARALLELISM.has(cmd)) {
return Integer.parseInt(ParserOptions.ERROR_WRITER_PARALLELISM.get(cmd, "1"));
}
- return Optional.ofNullable(parserConfig.getErrorWriterParallelism()).orElse(1);
+ int retValue = 1;
+ for (SensorParserConfig config : parserConfigs) {
+ Integer configValue = config.getErrorWriterParallelism();
+ retValue = configValue == null ? retValue : configValue;
+ }
+ return retValue;
};
// error bolt number of tasks
- ValueSupplier<Integer> errorNumTasks = (parserConfig, clazz) -> {
+ ValueSupplier<Integer> errorNumTasks = (parserConfigs, clazz) -> {
if(ParserOptions.ERROR_WRITER_NUM_TASKS.has(cmd)) {
return Integer.parseInt(ParserOptions.ERROR_WRITER_NUM_TASKS.get(cmd, "1"));
}
- return Optional.ofNullable(parserConfig.getErrorWriterNumTasks()).orElse(1);
+ int retValue = 1;
+ for (SensorParserConfig config : parserConfigs) {
+ Integer configValue = config.getErrorWriterNumTasks();
+ retValue = configValue == null ? retValue : configValue;
+ }
+ return retValue;
};
// kafka spout config
- ValueSupplier<Map> spoutConfig = (parserConfig, clazz) -> {
+ ValueSupplier<List> spoutConfig = (parserConfigs, clazz) -> {
if(ParserOptions.SPOUT_CONFIG.has(cmd)) {
- return readJSONMapFromFile(new File(ParserOptions.SPOUT_CONFIG.get(cmd)));
+ return Collections.singletonList(readJSONMapFromFile(new File(ParserOptions.SPOUT_CONFIG.get(cmd))));
+ }
+ List<Map<String, Object>> retValue = new ArrayList<>();
+ for (SensorParserConfig config : parserConfigs) {
+ retValue.add(config.getSpoutConfig());
}
- return Optional.ofNullable(parserConfig.getSpoutConfig()).orElse(new HashMap<>());
+ return retValue;
};
// security protocol
- ValueSupplier<String> securityProtocol = (parserConfig, clazz) -> {
+ ValueSupplier<String> securityProtocol = (parserConfigs, clazz) -> {
Optional<String> sp = Optional.empty();
if (ParserOptions.SECURITY_PROTOCOL.has(cmd)) {
sp = Optional.of(ParserOptions.SECURITY_PROTOCOL.get(cmd));
}
+ // Need to adjust to handle list of spoutConfigs. Any non-plaintext wins
if (!sp.isPresent()) {
- sp = getSecurityProtocol(sp, spoutConfig.get(parserConfig, Map.class));
+ sp = getSecurityProtocol(sp, spoutConfig.get(parserConfigs, List.class));
+ }
+ // Need to look through parserConfigs for any non-plaintext
+ String parserConfigSp = SecurityProtocol.PLAINTEXT.name;
+ for (SensorParserConfig config : parserConfigs) {
+ String configSp = config.getSecurityProtocol();
+ if (!SecurityProtocol.PLAINTEXT.name.equals(configSp)) {
+ // We have a winner
+ parserConfigSp = configSp;
+ }
}
- return sp.orElse(Optional.ofNullable(parserConfig.getSecurityProtocol()).orElse(null));
+
+ return sp.orElse(Optional.ofNullable(parserConfigSp).orElse(null));
};
// storm configuration
- ValueSupplier<Config> stormConf = (parserConfig, clazz) -> {
- Map<String, Object> c = parserConfig.getStormConfig();
+ ValueSupplier<Config> stormConf = (parserConfigs, clazz) -> {
+ // Last one wins
Config finalConfig = new Config();
- if(c != null && !c.isEmpty()) {
- finalConfig.putAll(c);
- }
- if(parserConfig.getNumAckers() != null) {
- Config.setNumAckers(finalConfig, parserConfig.getNumAckers());
- }
- if(parserConfig.getNumWorkers() != null) {
- Config.setNumWorkers(finalConfig, parserConfig.getNumWorkers());
+ for (SensorParserConfig parserConfig : parserConfigs) {
+ Map<String, Object> c = parserConfig.getStormConfig();
+ if (c != null && !c.isEmpty()) {
+ finalConfig.putAll(c);
+ }
+ if (parserConfig.getNumAckers() != null) {
+ Config.setNumAckers(finalConfig, parserConfig.getNumAckers());
+ }
+ if (parserConfig.getNumWorkers() != null) {
+ Config.setNumWorkers(finalConfig, parserConfig.getNumWorkers());
+ }
}
return ParserOptions.getConfig(cmd, finalConfig).orElse(finalConfig);
};
// output topic
- ValueSupplier<String> outputTopic = (parserConfig, clazz) -> {
- String topic;
+ ValueSupplier<String> outputTopic = (parserConfigs, clazz) -> {
+ String topic = null;
if(ParserOptions.OUTPUT_TOPIC.has(cmd)) {
topic = ParserOptions.OUTPUT_TOPIC.get(cmd);
-
- } else if(parserConfig.getOutputTopic() != null) {
- topic = parserConfig.getOutputTopic();
-
- } else {
- topic = Constants.ENRICHMENT_TOPIC;
}
return topic;
};
- // error topic
- ValueSupplier<String> errorTopic = (parserConfig, clazz) -> {
- String topic;
-
- if(parserConfig.getErrorTopic() != null) {
- topic = parserConfig.getErrorTopic();
-
- } else {
- // topic will to set to the 'parser.error.topic' setting in globals when the error bolt is created
- topic = null;
+ // Error topic will throw an exception if the topics aren't all the same.
+ ValueSupplier<String> errorTopic = (parserConfigs, clazz) -> {
+ // topic will to set to the 'parser.error.topic' setting in globals when the error bolt is created
+ String topic = null;
+ for (SensorParserConfig parserConfig : parserConfigs) {
+ String currentTopic = parserConfig.getErrorTopic();
+ if(topic != null && !topic.equals(currentTopic)) {
+ throw new IllegalArgumentException(
+ "Parser Aggregation specified with differing error topics");
+ }
+ topic = currentTopic;
}
return topic;
@@ -446,7 +540,7 @@ public class ParserTopologyCLI {
return getParserTopology(
zookeeperUrl,
brokerUrl,
- sensorType,
+ sensorTypes,
spoutParallelism,
spoutNumTasks,
parserParallelism,
@@ -462,14 +556,14 @@ public class ParserTopologyCLI {
protected ParserTopologyBuilder.ParserTopology getParserTopology( String zookeeperUrl,
Optional<String> brokerUrl,
- String sensorType,
- ValueSupplier<Integer> spoutParallelism,
- ValueSupplier<Integer> spoutNumTasks,
+ List<String> sensorTypes,
+ ValueSupplier<List> spoutParallelism,
+ ValueSupplier<List> spoutNumTasks,
ValueSupplier<Integer> parserParallelism,
ValueSupplier<Integer> parserNumTasks,
ValueSupplier<Integer> errorParallelism,
ValueSupplier<Integer> errorNumTasks,
- ValueSupplier<Map> spoutConfig,
+ ValueSupplier<List> spoutConfig,
ValueSupplier<String> securityProtocol,
ValueSupplier<Config> stormConf,
ValueSupplier<String> outputTopic,
@@ -477,7 +571,7 @@ public class ParserTopologyCLI {
return ParserTopologyBuilder.build(
zookeeperUrl,
brokerUrl,
- sensorType,
+ sensorTypes,
spoutParallelism,
spoutNumTasks,
parserParallelism,
@@ -505,15 +599,15 @@ public class ParserTopologyCLI {
}
ParserTopologyCLI cli = new ParserTopologyCLI();
ParserTopologyBuilder.ParserTopology topology = cli.createParserTopology(cmd);
- String sensorType= ParserOptions.SENSOR_TYPE.get(cmd);
+ String sensorTypes = ParserOptions.SENSOR_TYPES.get(cmd);
if (ParserOptions.TEST.has(cmd)) {
topology.getTopologyConfig().put(Config.TOPOLOGY_DEBUG, true);
LocalCluster cluster = new LocalCluster();
- cluster.submitTopology(sensorType, topology.getTopologyConfig(), topology.getBuilder().createTopology());
+ cluster.submitTopology(sensorTypes.replaceAll(",", STORM_JOB_SEPARATOR), topology.getTopologyConfig(), topology.getBuilder().createTopology());
Utils.sleep(300000);
cluster.shutdown();
} else {
- StormSubmitter.submitTopology(sensorType, topology.getTopologyConfig(), topology.getBuilder().createTopology());
+ StormSubmitter.submitTopology(sensorTypes.replaceAll(",", STORM_JOB_SEPARATOR), topology.getTopologyConfig(), topology.getBuilder().createTopology());
}
} catch (Exception e) {
e.printStackTrace();
@@ -521,15 +615,27 @@ public class ParserTopologyCLI {
}
}
- private static Optional<String> getSecurityProtocol(Optional<String> protocol, Map<String, Object> spoutConfig) {
+ private static Optional<String> getSecurityProtocol(Optional<String> protocol, List<Map<String, Object>> spoutConfig) {
Optional<String> ret = protocol;
- if(ret.isPresent() && protocol.get().equalsIgnoreCase("PLAINTEXT")) {
+ if(ret.isPresent() && protocol.get().equalsIgnoreCase(SecurityProtocol.PLAINTEXT.name)) {
ret = Optional.empty();
}
if(!ret.isPresent()) {
- ret = Optional.ofNullable((String) spoutConfig.get("security.protocol"));
+ // Need to look through spoutConfig for any non-plaintext
+ String spoutConfigSp = null;
+ for (Map<String, Object> config: spoutConfig) {
+ String configSp = (String) config.get(KafkaUtils.SECURITY_PROTOCOL);
+ if (configSp != null && !SecurityProtocol.PLAINTEXT.name.equals(configSp)) {
+ // We have a winner
+ spoutConfigSp = configSp;
+ } else if (configSp != null) {
+ // Use something explicitly defined.
+ spoutConfigSp = configSp;
+ }
+ }
+ ret = Optional.ofNullable(spoutConfigSp);
}
- if(ret.isPresent() && ret.get().equalsIgnoreCase("PLAINTEXT")) {
+ if(ret.isPresent() && ret.get().equalsIgnoreCase(SecurityProtocol.PLAINTEXT.name)) {
ret = Optional.empty();
}
return ret;
http://git-wip-us.apache.org/repos/asf/metron/blob/f4345383/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ValueSupplier.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ValueSupplier.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ValueSupplier.java
index 0ede0f8..98aca7b 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ValueSupplier.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ValueSupplier.java
@@ -18,6 +18,7 @@
package org.apache.metron.parsers.topology.config;
+import java.util.Collection;
import org.apache.metron.common.configuration.SensorParserConfig;
@@ -26,5 +27,5 @@ import org.apache.metron.common.configuration.SensorParserConfig;
* @param <T>
*/
public interface ValueSupplier<T> {
- T get(SensorParserConfig config, Class<T> clazz);
+ T get(Collection<SensorParserConfig> config, Class<T> clazz);
}
[15/15] metron git commit: Merge branch 'master' into
feature/METRON-1554-pcap-query-panel
Posted by mm...@apache.org.
Merge branch 'master' into feature/METRON-1554-pcap-query-panel
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/b580a69f
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/b580a69f
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/b580a69f
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: b580a69f1d9d3346171d749dcd6f791a9bc6d894
Parents: a5a5139 aceca46
Author: Michael Miklavcic <mi...@gmail.com>
Authored: Fri Jul 20 09:40:09 2018 -0600
Committer: Michael Miklavcic <mi...@gmail.com>
Committed: Fri Jul 20 09:40:09 2018 -0600
----------------------------------------------------------------------
.travis.yml | 14 +-
.../metron/profiler/client/window/Window.java | 8 +-
.../profiler/client/window/WindowProcessor.java | 35 +-
.../client/window/predicates/DayPredicates.java | 2 +-
.../profiler/DefaultMessageDistributor.java | 18 +-
.../ansible/playbooks/docker_probe_install.yml | 2 +-
.../ansible/roles/bro/vars/main.yml | 2 +-
.../centos6/ansible/inventory/group_vars/all | 2 +-
.../ubuntu14/ansible/inventory/group_vars/all | 2 +-
.../package/scripts/alerts_ui_commands.py | 9 +-
.../package/scripts/management_ui_commands.py | 9 +-
metron-deployment/scripts/platform-info.sh | 4 +
.../alert-details-status.e2e-spec.ts | 129 +++---
.../meta-alerts/meta-alert.e2e-spec.ts | 30 +-
metron-interface/metron-alerts/karma.conf.js | 15 +-
metron-interface/metron-alerts/package.json | 3 +-
metron-interface/metron-alerts/pom.xml | 14 +-
.../apache/metron/rest/config/KafkaConfig.java | 4 +-
.../service/impl/StormAdminServiceImpl.java | 23 +-
.../rest/service/impl/StormCLIWrapper.java | 25 +-
.../common/bolt/ConfiguredParserBolt.java | 10 +-
.../configuration/IndexingConfigurations.java | 2 +-
.../configuration/SensorParserConfig.java | 12 +-
.../writer/ConfigurationsStrategies.java | 2 +-
.../apache/metron/common/error/MetronError.java | 64 +--
.../common/performance/PerformanceLogger.java | 10 +-
.../apache/metron/common/system/FakeClock.java | 18 +-
.../metron/common/utils/ReflectionUtils.java | 2 +-
.../configurations/ConfigurationsUpdater.java | 1 -
.../common/bolt/ConfiguredParserBoltTest.java | 2 +-
.../metron/common/error/MetronErrorTest.java | 16 +-
.../indexing/dao/metaalert/MetaAlertDao.java | 6 +-
.../dao/metaalert/MetaAlertIntegrationTest.java | 56 +++
.../metron/management/KafkaFunctions.java | 109 +++++
.../KafkaFunctionsIntegrationTest.java | 92 +++++
.../metron-parsers/3rdPartyParser.md | 408 +++++++++----------
metron-platform/metron-parsers/README.md | 25 +-
.../apache/metron/parsers/bolt/ParserBolt.java | 241 ++++++-----
.../apache/metron/parsers/bolt/WriterBolt.java | 3 +-
.../parsers/topology/ParserComponents.java | 67 +++
.../parsers/topology/ParserTopologyBuilder.java | 200 ++++++---
.../parsers/topology/ParserTopologyCLI.java | 264 ++++++++----
.../parsers/topology/config/ValueSupplier.java | 3 +-
.../metron/parsers/bolt/ParserBoltTest.java | 182 ++++++---
.../metron/parsers/bolt/WriterBoltTest.java | 30 +-
.../parsers/integration/ParserDriver.java | 50 ++-
.../components/ParserTopologyComponent.java | 58 +--
.../parsers/topology/ParserTopologyCLITest.java | 318 ++++++++++-----
...pleHbaseEnrichmentWriterIntegrationTest.java | 3 +-
.../integration/WriterBoltIntegrationTest.java | 150 ++++++-
.../src/main/config/schema/metaalert/schema.xml | 6 +-
.../metron/writer/BulkWriterComponent.java | 27 +-
.../metron/writer/BulkWriterComponentTest.java | 34 +-
.../stellar/common/CachingStellarProcessor.java | 2 +-
.../org/apache/metron/stellar/dsl/Context.java | 2 +-
.../dsl/functions/DateFunctionsTest.java | 3 +-
use-cases/forensic_clustering/README.md | 228 +++++++----
use-cases/parser_chaining/README.md | 14 +
.../aggregated_parser_chaining_flow.svg | 14 +
.../aggregated_parser_chaining_flow.xml | 14 +
use-cases/typosquat_detection/README.md | 9 +-
61 files changed, 2109 insertions(+), 998 deletions(-)
----------------------------------------------------------------------
[04/15] metron git commit: METRON-1236 Add start/stop/restart
commands that execute successfully,
when ambari agents run as non-root user closes apache/incubator-metron#1105
Posted by mm...@apache.org.
METRON-1236 Add start/stop/restart commands that execute successfully, when ambari agents run as non-root user closes apache/incubator-metron#1105
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/75d6a72a
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/75d6a72a
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/75d6a72a
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 75d6a72a3867d1f89dbec8dc059870482ac81e94
Parents: 4544da7
Author: Stefan Kupstaitis-Dunkler <st...@gmail.com>
Authored: Mon Jul 16 11:58:24 2018 -0400
Committer: cstella <ce...@gmail.com>
Committed: Mon Jul 16 11:58:24 2018 -0400
----------------------------------------------------------------------
.../METRON/CURRENT/package/scripts/alerts_ui_commands.py | 9 ++++++---
.../CURRENT/package/scripts/management_ui_commands.py | 9 ++++++---
2 files changed, 12 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/75d6a72a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/alerts_ui_commands.py
----------------------------------------------------------------------
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/alerts_ui_commands.py b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/alerts_ui_commands.py
index 41cd375..f837043 100644
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/alerts_ui_commands.py
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/alerts_ui_commands.py
@@ -38,7 +38,8 @@ class AlertsUICommands:
:param env: Environment
"""
Logger.info('Starting Alerts UI')
- Execute("service metron-alerts-ui start")
+ start_cmd = ('service', 'metron-alerts-ui', 'start')
+ Execute(start_cmd, sudo=True)
Logger.info('Done starting Alerts UI')
def stop_alerts_ui(self):
@@ -47,7 +48,8 @@ class AlertsUICommands:
:param env: Environment
"""
Logger.info('Stopping Alerts UI')
- Execute("service metron-alerts-ui stop")
+ stop_cmd = ('service', 'metron-alerts-ui', 'stop')
+ Execute(stop_cmd, sudo=True)
Logger.info('Done stopping Alerts UI')
def restart_alerts_ui(self, env):
@@ -56,7 +58,8 @@ class AlertsUICommands:
:param env: Environment
"""
Logger.info('Restarting the Alerts UI')
- Execute('service metron-alerts-ui restart')
+ restart_cmd = ('service', 'metron-alerts-ui', 'restart')
+ Execute(restart_cmd, sudo=True)
Logger.info('Done restarting the Alerts UI')
def status_alerts_ui(self, env):
http://git-wip-us.apache.org/repos/asf/metron/blob/75d6a72a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/management_ui_commands.py
----------------------------------------------------------------------
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/management_ui_commands.py b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/management_ui_commands.py
index 1f16fa4..4e81df6 100644
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/management_ui_commands.py
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/management_ui_commands.py
@@ -40,7 +40,8 @@ class ManagementUICommands:
:param env: Environment
"""
Logger.info('Starting Management UI')
- Execute("service metron-management-ui start")
+ start_cmd = ('service', 'metron-management-ui', 'start')
+ Execute(start_cmd, sudo=True)
Logger.info('Done starting Management UI')
def stop_management_ui(self):
@@ -49,7 +50,8 @@ class ManagementUICommands:
:param env: Environment
"""
Logger.info('Stopping Management UI')
- Execute("service metron-management-ui stop")
+ stop_cmd = ('service', 'metron-management-ui', 'stop')
+ Execute(stop_cmd, sudo=True)
Logger.info('Done stopping Management UI')
def restart_management_ui(self, env):
@@ -58,7 +60,8 @@ class ManagementUICommands:
:param env: Environment
"""
Logger.info('Restarting the Management UI')
- Execute('service metron-management-ui restart')
+ restart_cmd = ('service', 'metron-management-ui', 'restart')
+ Execute(restart_cmd, sudo=True)
Logger.info('Done restarting the Management UI')
def status_management_ui(self, env):
[02/15] metron git commit: METRON-1660 On Solr,
sorting by threat score fails (justinleet) closes apache/metron#1102
Posted by mm...@apache.org.
METRON-1660 On Solr, sorting by threat score fails (justinleet) closes apache/metron#1102
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/28f4b570
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/28f4b570
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/28f4b570
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 28f4b570493eda0a23317f520b89cb370e606ca0
Parents: 7af11b6
Author: justinleet <ju...@gmail.com>
Authored: Wed Jul 11 15:48:08 2018 -0400
Committer: leet <le...@apache.org>
Committed: Wed Jul 11 15:48:08 2018 -0400
----------------------------------------------------------------------
.../dao/metaalert/MetaAlertIntegrationTest.java | 56 ++++++++++++++++++++
.../src/main/config/schema/metaalert/schema.xml | 6 ++-
2 files changed, 61 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/28f4b570/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/metaalert/MetaAlertIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/metaalert/MetaAlertIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/metaalert/MetaAlertIntegrationTest.java
index 6f96fb5..f754b81 100644
--- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/metaalert/MetaAlertIntegrationTest.java
+++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/metaalert/MetaAlertIntegrationTest.java
@@ -51,6 +51,7 @@ import org.apache.metron.indexing.dao.search.SearchRequest;
import org.apache.metron.indexing.dao.search.SearchResponse;
import org.apache.metron.indexing.dao.search.SearchResult;
import org.apache.metron.indexing.dao.search.SortField;
+import org.apache.metron.indexing.dao.search.SortOrder;
import org.apache.metron.indexing.dao.update.Document;
import org.apache.metron.indexing.dao.update.OriginalNotFoundException;
import org.apache.metron.indexing.dao.update.PatchRequest;
@@ -194,6 +195,60 @@ public abstract class MetaAlertIntegrationTest {
}
@Test
+ public void shouldSortByThreatTriageScore() throws Exception {
+ // Load alerts
+ List<Map<String, Object>> alerts = buildAlerts(2);
+ alerts.get(0).put(METAALERT_FIELD, "meta_active_0");
+ addRecords(alerts, getTestIndexFullName(), SENSOR_NAME);
+
+ // Load metaAlerts
+ List<Map<String, Object>> metaAlerts = buildMetaAlerts(1, MetaAlertStatus.ACTIVE,
+ Optional.of(Collections.singletonList(alerts.get(0))));
+ // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
+ addRecords(metaAlerts, getMetaAlertIndex(), METAALERT_TYPE);
+
+ // Verify load was successful
+ List<GetRequest> createdDocs = metaAlerts.stream().map(metaAlert ->
+ new GetRequest((String) metaAlert.get(Constants.GUID), METAALERT_TYPE))
+ .collect(Collectors.toList());
+ createdDocs.addAll(alerts.stream().map(alert ->
+ new GetRequest((String) alert.get(Constants.GUID), SENSOR_NAME))
+ .collect(Collectors.toList()));
+ findCreatedDocs(createdDocs);
+
+ // Test descending
+ SortField sf = new SortField();
+ sf.setField(getThreatTriageField());
+ sf.setSortOrder(SortOrder.DESC.getSortOrder());
+ SearchRequest sr = new SearchRequest();
+ sr.setQuery("*:*");
+ sr.setSize(5);
+ sr.setIndices(Arrays.asList(getTestIndexName(), METAALERT_TYPE));
+ sr.setSort(Collections.singletonList(sf));
+
+ SearchResponse result = metaDao.search(sr);
+ List<SearchResult> results = result.getResults();
+ Assert.assertEquals(2, results.size());
+ Assert.assertEquals("meta_active_0", results.get((0)).getId());
+ Assert.assertEquals("message_1", results.get((1)).getId());
+
+ // Test ascending
+ SortField sfAsc = new SortField();
+ sfAsc.setField(getThreatTriageField());
+ sfAsc.setSortOrder(SortOrder.ASC.getSortOrder());
+ SearchRequest srAsc = new SearchRequest();
+ srAsc.setQuery("*:*");
+ srAsc.setSize(2);
+ srAsc.setIndices(Arrays.asList(getTestIndexName(), METAALERT_TYPE));
+ srAsc.setSort(Collections.singletonList(sfAsc));
+ result = metaDao.search(srAsc);
+ results = result.getResults();
+ Assert.assertEquals("message_1", results.get((0)).getId());
+ Assert.assertEquals("meta_active_0", results.get((1)).getId());
+ Assert.assertEquals(2, results.size());
+ }
+
+ @Test
public void getAllMetaAlertsForAlertShouldThrowExceptionForEmptyGuid() throws Exception {
try {
metaDao.getAllMetaAlertsForAlert("");
@@ -960,6 +1015,7 @@ public abstract class MetaAlertIntegrationTest {
metaAlert.put(Constants.GUID, guid);
metaAlert.put(getSourceTypeField(), METAALERT_TYPE);
metaAlert.put(STATUS_FIELD, status.getStatusString());
+ metaAlert.put(getThreatTriageField(), 100.0d);
if (alerts.isPresent()) {
List<Map<String, Object>> alertsList = alerts.get();
metaAlert.put(ALERT_FIELD, alertsList);
http://git-wip-us.apache.org/repos/asf/metron/blob/28f4b570/metron-platform/metron-solr/src/main/config/schema/metaalert/schema.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-solr/src/main/config/schema/metaalert/schema.xml b/metron-platform/metron-solr/src/main/config/schema/metaalert/schema.xml
index 63e729b..6555bf6 100644
--- a/metron-platform/metron-solr/src/main/config/schema/metaalert/schema.xml
+++ b/metron-platform/metron-solr/src/main/config/schema/metaalert/schema.xml
@@ -28,7 +28,6 @@
<field name="timestamp" type="plong" indexed="true" stored="true"/>
<field name="score" type="pdouble" indexed="true" stored="true"/>
<field name="status" type="string" indexed="true" stored="true"/>
- <field name="threat:triage:score" type="pdouble" indexed="true" stored="true"/>
<field name="average" type="pdouble" indexed="true" stored="true"/>
<field name="min" type="pdouble" indexed="true" stored="true"/>
<field name="median" type="pdouble" indexed="true" stored="true"/>
@@ -40,6 +39,11 @@
<!-- Ensure that metaalerts child field is multivalued -->
<field name="metaalerts" type="string" multiValued="true" indexed="true" stored="true"/>
+ <!-- Threat Intel Scoring Field -->
+ <!-- This is a double from the method of calculation. It'll still sort alongside pfloat -->
+ <dynamicField name="*score" type="pdouble" multiValued="false" docValues="true"/>
+
+ <!-- Catch all, if we don't know about it, it gets dropped. -->
<dynamicField name="*" type="ignored" indexed="true" stored="true" multiValued="false" docValues="true"/>
<uniqueKey>guid</uniqueKey>
[14/15] metron git commit: METRON-1672 Add metron-alerts'
s UI unit tests to travis build process (justinleet) closes
apache/metron#1106
Posted by mm...@apache.org.
METRON-1672 Add metron-alerts's UI unit tests to travis build process (justinleet) closes apache/metron#1106
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/aceca469
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/aceca469
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/aceca469
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: aceca469a83d17c033ea7a47ff82c61962319b3a
Parents: 52bd310
Author: justinleet <ju...@gmail.com>
Authored: Thu Jul 19 09:30:59 2018 -0400
Committer: leet <le...@apache.org>
Committed: Thu Jul 19 09:30:59 2018 -0400
----------------------------------------------------------------------
.travis.yml | 14 +++++++++++---
metron-interface/metron-alerts/karma.conf.js | 15 ++++++++++++++-
metron-interface/metron-alerts/package.json | 3 ++-
metron-interface/metron-alerts/pom.xml | 14 ++++++++++++--
4 files changed, 39 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/aceca469/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index 71d7165..3a73d7e 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -16,9 +16,14 @@
dist: trusty
sudo: required
+addons:
+ chrome: stable
env:
- - TEST_TYPE=default
+ - SCRIPT="mvn surefire:test@unit-tests -T 2C"
+ - SCRIPT="mvn surefire:test@integration-tests"
+ - SCRIPT="mvn test --projects metron-interface/metron-config,metron-interface/metron-alerts"
+ - SCRIPT="./dev-utilities/build-utils/verify_licenses.sh"
install: true
language: java
@@ -33,10 +38,13 @@ before_install:
- npm config set prefix $HOME/.npm-prefix --global
install:
- - time mvn -q -T 2C -DskipTests clean install
+ - time mvn install -T 2C -q -DskipTests=true -Dmaven.javadoc.skip=true -B -V
script:
- - time mvn -q -T 2C surefire:test@unit-tests && time mvn -q surefire:test@integration-tests && time mvn -q test --projects metron-interface/metron-config && time dev-utilities/build-utils/verify_licenses.sh
+ - time $SCRIPT
+
+matrix:
+ fast_finish: true
before_cache:
- rm -rf $HOME/.m2/repository/org/apache/metron
http://git-wip-us.apache.org/repos/asf/metron/blob/aceca469/metron-interface/metron-alerts/karma.conf.js
----------------------------------------------------------------------
diff --git a/metron-interface/metron-alerts/karma.conf.js b/metron-interface/metron-alerts/karma.conf.js
index f245e1d..7f29c92 100644
--- a/metron-interface/metron-alerts/karma.conf.js
+++ b/metron-interface/metron-alerts/karma.conf.js
@@ -18,6 +18,8 @@
// Karma configuration file, see link for more information
// https://karma-runner.github.io/0.13/config/configuration-file.html
+process.env.CHROME_BIN = require('puppeteer').executablePath()
+
module.exports = function (config) {
config.set({
basePath: '',
@@ -55,7 +57,18 @@ module.exports = function (config) {
colors: true,
logLevel: config.LOG_INFO,
autoWatch: true,
- browsers: ['Chrome'],
+ browsers: ['Chrome','ChromeHeadless'],
+ customLaunchers: {
+ ChromeHeadless: {
+ base: 'Chrome',
+ flags: [
+ '--no-sandbox',
+ '--headless',
+ '--disable-gpu',
+ '--remote-debugging-port=9222'
+ ]
+ }
+ },
singleRun: false
});
};
http://git-wip-us.apache.org/repos/asf/metron/blob/aceca469/metron-interface/metron-alerts/package.json
----------------------------------------------------------------------
diff --git a/metron-interface/metron-alerts/package.json b/metron-interface/metron-alerts/package.json
index 0a40f0d..9cfc09f 100644
--- a/metron-interface/metron-alerts/package.json
+++ b/metron-interface/metron-alerts/package.json
@@ -7,7 +7,7 @@
"build": "./node_modules/@angular/cli/bin/ng build -prod",
"start": "ng serve",
"lint": "tslint \"src/**/*.ts\"",
- "test": "./node_modules/@angular/cli/bin/ng test --watch=false",
+ "test": "karma start --single-run --browsers ChromeHeadless karma.conf.js",
"pree2e": "webdriver-manager update",
"e2e": "protractor-flake --protractor-path=./node_modules/.bin/protractor --max-attempts=3 -- ./protractor.conf.js"
},
@@ -58,6 +58,7 @@
"optimist": "0.6.1",
"protractor": "5.3.0",
"protractor-flake": "^3.3.0",
+ "puppeteer": "^1.6.0",
"serve-favicon": "2.4.2",
"serve-static": "1.12.1",
"ssh2": "^0.5.5",
http://git-wip-us.apache.org/repos/asf/metron/blob/aceca469/metron-interface/metron-alerts/pom.xml
----------------------------------------------------------------------
diff --git a/metron-interface/metron-alerts/pom.xml b/metron-interface/metron-alerts/pom.xml
index e7706b3..996f473 100644
--- a/metron-interface/metron-alerts/pom.xml
+++ b/metron-interface/metron-alerts/pom.xml
@@ -25,8 +25,8 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
- <node.version>v7.10.0</node.version>
- <npm.version>4.2.0</npm.version>
+ <node.version>v9.11.1</node.version>
+ <npm.version>5.8.0</npm.version>
</properties>
<dependencies>
</dependencies>
@@ -70,6 +70,16 @@
<arguments>run build</arguments>
</configuration>
</execution>
+ <execution>
+ <phase>test</phase>
+ <id>npm test</id>
+ <goals>
+ <goal>npm</goal>
+ </goals>
+ <configuration>
+ <arguments>test</arguments>
+ </configuration>
+ </execution>
</executions>
</plugin>
<plugin>
[07/15] metron git commit: METRON-1620: Fixes for forensic clustering
use case example (mmiklavc via mmiklavc) closes apache/metron#1065
Posted by mm...@apache.org.
METRON-1620: Fixes for forensic clustering use case example (mmiklavc via mmiklavc) closes apache/metron#1065
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/0c20fd1a
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/0c20fd1a
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/0c20fd1a
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 0c20fd1a3a809a8ad2e8cc280a2a5382f3d5a7e8
Parents: a4bec95
Author: mmiklavc <mi...@gmail.com>
Authored: Mon Jul 16 12:54:53 2018 -0600
Committer: Michael Miklavcic <mi...@gmail.com>
Committed: Mon Jul 16 12:54:53 2018 -0600
----------------------------------------------------------------------
use-cases/forensic_clustering/README.md | 228 +++++++++++++++++++--------
use-cases/typosquat_detection/README.md | 9 +-
2 files changed, 166 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/0c20fd1a/use-cases/forensic_clustering/README.md
----------------------------------------------------------------------
diff --git a/use-cases/forensic_clustering/README.md b/use-cases/forensic_clustering/README.md
index dac116c..fd631c1 100644
--- a/use-cases/forensic_clustering/README.md
+++ b/use-cases/forensic_clustering/README.md
@@ -157,28 +157,138 @@ we have valid data.
```
-Before we start, we will want to install ES mappings so ES knows how to interpret our fields:
+Before we start, we will want to install ES template mappings so ES knows how to interpret our fields:
```
-curl -XPUT 'http://$ES_HOST/cowrie*/_mapping/cowrie_doc' -d '
+curl -XPUT $ES_HOST'/_template/cowrie_index' -d '
{
+ "template": "cowrie_index*",
+ "mappings": {
+ "cowrie_doc": {
+ "dynamic_templates": [
+ {
+ "geo_location_point": {
+ "match": "enrichments:geo:*:location_point",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "geo_point"
+ }
+ }
+ },
+ {
+ "geo_country": {
+ "match": "enrichments:geo:*:country",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "keyword"
+ }
+ }
+ },
+ {
+ "geo_city": {
+ "match": "enrichments:geo:*:city",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "keyword"
+ }
+ }
+ },
+ {
+ "geo_location_id": {
+ "match": "enrichments:geo:*:locID",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "keyword"
+ }
+ }
+ },
+ {
+ "geo_dma_code": {
+ "match": "enrichments:geo:*:dmaCode",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "keyword"
+ }
+ }
+ },
+ {
+ "geo_postal_code": {
+ "match": "enrichments:geo:*:postalCode",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "keyword"
+ }
+ }
+ },
+ {
+ "geo_latitude": {
+ "match": "enrichments:geo:*:latitude",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "float"
+ }
+ }
+ },
+ {
+ "geo_longitude": {
+ "match": "enrichments:geo:*:longitude",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "float"
+ }
+ }
+ },
+ {
+ "timestamps": {
+ "match": "*:ts",
+ "match_mapping_type": "*",
+ "mapping": {
+ "type": "date",
+ "format": "epoch_millis"
+ }
+ }
+ },
+ {
+ "threat_triage_score": {
+ "mapping": {
+ "type": "float"
+ },
+ "match": "threat:triage:*score",
+ "match_mapping_type": "*"
+ }
+ },
+ {
+ "threat_triage_reason": {
+ "mapping": {
+ "type": "text",
+ "fielddata": "true"
+ },
+ "match": "threat:triage:rules:*:reason",
+ "match_mapping_type": "*"
+ }
+ },
+ {
+ "threat_triage_name": {
+ "mapping": {
+ "type": "text",
+ "fielddata": "true"
+ },
+ "match": "threat:triage:rules:*:name",
+ "match_mapping_type": "*"
+ }
+ }
+ ],
"properties" : {
- "adapter:stellaradapter:begin:ts" : {
- "type" : "string"
- },
- "adapter:stellaradapter:end:ts" : {
- "type" : "string"
- },
"blacklisted" : {
"type" : "boolean"
},
"compCS" : {
- "type" : "string"
+ "type" : "keyword"
},
"data" : {
- "type" : "string"
+ "type" : "keyword"
},
"dst_ip" : {
- "type" : "string"
+ "type" : "keyword"
},
"dst_port" : {
"type" : "long"
@@ -187,117 +297,87 @@ curl -XPUT 'http://$ES_HOST/cowrie*/_mapping/cowrie_doc' -d '
"type" : "double"
},
"encCS" : {
- "type" : "string"
- },
- "enrichmentjoinbolt:joiner:ts" : {
- "type" : "string"
- },
- "enrichmentsplitterbolt:splitter:begin:ts" : {
- "type" : "string"
- },
- "enrichmentsplitterbolt:splitter:end:ts" : {
- "type" : "string"
+ "type" : "keyword"
},
"eventid" : {
- "type" : "string"
+ "type" : "keyword"
},
"guid" : {
- "type" : "string"
+ "type" : "keyword"
},
"input" : {
- "type" : "string"
+ "type" : "keyword"
},
"isError" : {
"type" : "long"
},
"is_alert" : {
- "type" : "string"
+ "type" : "keyword"
},
"kexAlgs" : {
- "type" : "string"
+ "type" : "keyword"
},
"keyAlgs" : {
- "type" : "string"
+ "type" : "keyword"
},
"macCS" : {
- "type" : "string"
+ "type" : "keyword"
},
"message" : {
- "type" : "string"
+ "type" : "keyword"
},
- "original_string" : {
- "type" : "string"
+ "original_keyword" : {
+ "type" : "keyword"
},
"password" : {
- "type" : "string"
+ "type" : "keyword"
},
"sensor" : {
- "type" : "string"
+ "type" : "keyword"
},
"session" : {
- "type" : "string"
+ "type" : "keyword"
},
"similarity_bin" : {
- "type" : "string"
+ "type" : "keyword"
},
"size" : {
"type" : "long"
},
"source:type" : {
- "type" : "string"
+ "type" : "keyword"
},
"src_ip" : {
- "type" : "string"
+ "type" : "keyword"
},
"src_port" : {
"type" : "long"
},
"system" : {
- "type" : "string"
- },
- "threat:triage:rules:0:comment" : {
- "type" : "string"
- },
- "threat:triage:rules:0:name" : {
- "type" : "string"
+ "type" : "keyword"
},
- "threat:triage:rules:0:reason" : {
- "type" : "string"
- },
- "threat:triage:rules:0:score" : {
- "type" : "long"
- },
- "threat:triage:score" : {
- "type" : "double"
- },
- "threatinteljoinbolt:joiner:ts" : {
- "type" : "string"
- },
- "threatintelsplitterbolt:splitter:begin:ts" : {
- "type" : "string"
- },
- "threatintelsplitterbolt:splitter:end:ts" : {
- "type" : "string"
- },
- "timestamp" : {
- "type" : "long"
+ "timestamp": {
+ "type": "date",
+ "format": "epoch_millis"
},
"tlsh" : {
- "type" : "string"
+ "type" : "keyword"
},
"ttylog" : {
- "type" : "string"
+ "type" : "keyword"
},
"username" : {
- "type" : "string"
+ "type" : "keyword"
},
"version" : {
- "type" : "string"
+ "type" : "keyword"
},
- "alert" : {
+ "metron_alert" : {
"type" : "nested"
}
}
+ }
+ }
}
'
```
@@ -408,7 +488,7 @@ We want to pull a snapshot of the cowrie logs, so create `~/load_data.sh` with t
COWRIE_HOME=~/cowrie
for i in cowrie.1626302-1636522.json cowrie.16879981-16892488.json cowrie.21312194-21331475.json cowrie.698260-710913.json cowrie.762933-772239.json cowrie.929866-939552.json cowrie.1246880-1248235.json cowrie.19285959-19295444.json cowrie.16542668-16581213.json cowrie.5849832-5871517.json cowrie.6607473-6609163.json;do
echo $i
- cat $COWRIE_HOME/$i | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list node1:6667 --topic cowrie
+ cat $COWRIE_HOME/$i | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list $BROKERLIST --topic cowrie
sleep 2
done
```
@@ -448,3 +528,11 @@ As you can see, we have found a few more malicious actors:
* 94.78.80.45
Now we can look at *other* things that they're doing to build and refine our definition of what an alert is without resorting to hard-coding of rules. Note that nothing in our enrichments actually used the string `busybox`, so this is a more general purpose way of navigating similar things.
+
+### Version Info
+
+Verified against:
+
+- METRON_VERSION=0.5.0
+- ELASTIC_VERSION=5.6.2
+
http://git-wip-us.apache.org/repos/asf/metron/blob/0c20fd1a/use-cases/typosquat_detection/README.md
----------------------------------------------------------------------
diff --git a/use-cases/typosquat_detection/README.md b/use-cases/typosquat_detection/README.md
index 63624c7..12a770d 100644
--- a/use-cases/typosquat_detection/README.md
+++ b/use-cases/typosquat_detection/README.md
@@ -417,7 +417,7 @@ curl -XPOST "http://$ES_HOST/_template/squid_index" -d '{
"url" : {
"type" : "text","fielddata" : true
},
- "alert" : {
+ "metron_alert" : {
"type" : "nested"
}
}
@@ -448,3 +448,10 @@ From there you should see the following data from squid with one as an alert and
Now, if you drill down into the alert, you can see our fields and the reasons specified
+
+### Version Info
+
+Verified against:
+
+- METRON_VERSION=0.5.0
+- ELASTIC_VERSION=5.6.2
[09/15] metron git commit: METRON-1651 Fixing failing protractor e2e
test (tiborm via merrimanr) closes apache/metron#1095
Posted by mm...@apache.org.
METRON-1651 Fixing failing protractor e2e test (tiborm via merrimanr) closes apache/metron#1095
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/3a2675e5
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/3a2675e5
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/3a2675e5
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 3a2675e5f5fee0d461d75fde128b482e7e2a920a
Parents: 2a1cbbb
Author: tiborm <ti...@gmail.com>
Authored: Tue Jul 17 11:43:35 2018 -0500
Committer: rmerriman <me...@gmail.com>
Committed: Tue Jul 17 11:43:35 2018 -0500
----------------------------------------------------------------------
.../alert-details-status.e2e-spec.ts | 129 ++++++++++---------
.../meta-alerts/meta-alert.e2e-spec.ts | 30 +++--
2 files changed, 85 insertions(+), 74 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/3a2675e5/metron-interface/metron-alerts/e2e/alert-details/alert-status/alert-details-status.e2e-spec.ts
----------------------------------------------------------------------
diff --git a/metron-interface/metron-alerts/e2e/alert-details/alert-status/alert-details-status.e2e-spec.ts b/metron-interface/metron-alerts/e2e/alert-details/alert-status/alert-details-status.e2e-spec.ts
index 28d9ed0..ae58ea1 100644
--- a/metron-interface/metron-alerts/e2e/alert-details/alert-status/alert-details-status.e2e-spec.ts
+++ b/metron-interface/metron-alerts/e2e/alert-details/alert-status/alert-details-status.e2e-spec.ts
@@ -49,7 +49,13 @@ describe('Test spec for metron details page', function() {
jasmine.addMatchers(customMatchers);
});
- it('should change alert statuses', async function() : Promise<any> {
+ /**
+ * Test is failing because of a known issue of Metron Alert UI.
+ * https://issues.apache.org/jira/browse/METRON-1654
+ *
+ * Till the fix become available the test should remain ignored.
+ * */
+ xit('should change alert statuses', async function() : Promise<any> {
let alertId = '2cc174d7-c049-aaf4-d0d6-138073777309';
await page.navigateTo(alertId);
@@ -69,84 +75,85 @@ describe('Test spec for metron details page', function() {
await page.clickNew();
});
- // The below code will fail until this issue is resolved in Protractor: https://github.com/angular/protractor/issues/4693
- // This is because the connection resets before deleting the test comment, which causes the assertion to be false
-
- // it('should add comments for table view', async function() : Promise<any> {
- // let comment1 = 'This is a sample comment';
- // let comment2 = 'This is a sample comment again';
- // let userNameAndTimestamp = '- admin - a few seconds ago';
- // let alertId = '2cc174d7-c049-aaf4-d0d6-138073777309';
+ /**
+ * The below code will fail until this issue is resolved in Protractor: https://github.com/angular/protractor/issues/4693
+ * This is because the connection resets before deleting the test comment, which causes the assertion to be false
+ */
+ xit('should add comments for table view', async function() : Promise<any> {
+ let comment1 = 'This is a sample comment';
+ let comment2 = 'This is a sample comment again';
+ let userNameAndTimestamp = '- admin - a few seconds ago';
+ let alertId = '2cc174d7-c049-aaf4-d0d6-138073777309';
- // page.navigateTo(alertId);
+ page.navigateTo(alertId);
- // await page.clickCommentsInSideNav();
- // await page.addCommentAndSave(comment1, 0);
+ await page.clickCommentsInSideNav();
+ await page.addCommentAndSave(comment1, 0);
- // expect(await page.getCommentsText()).toEqual([comment1]);
- // expect(await page.getCommentsUserNameAndTimeStamp()).toEqual([userNameAndTimestamp]);
+ expect(await page.getCommentsText()).toEqual([comment1]);
+ expect(await page.getCommentsUserNameAndTimeStamp()).toEqual([userNameAndTimestamp]);
- // await page.addCommentAndSave(comment2, 0);
- // expect(await page.getCommentsText()).toEqual([comment2, comment1]);
- // expect(await page.getCommentsUserNameAndTimeStamp()).toEqual([userNameAndTimestamp, userNameAndTimestamp]);
+ await page.addCommentAndSave(comment2, 0);
+ expect(await page.getCommentsText()).toEqual([comment2, comment1]);
+ expect(await page.getCommentsUserNameAndTimeStamp()).toEqual([userNameAndTimestamp, userNameAndTimestamp]);
- // await page.deleteComment();
- // await page.clickNoForConfirmation();
- // expect(await page.getCommentsText()).toEqual([comment2, comment1]);
+ await page.deleteComment();
+ await page.clickNoForConfirmation();
+ expect(await page.getCommentsText()).toEqual([comment2, comment1]);
- // await page.deleteComment();
- // await page.clickYesForConfirmation(comment2);
- // expect(await page.getCommentsText()).toEqual([comment1]);
+ await page.deleteComment();
+ await page.clickYesForConfirmation(comment2);
+ expect(await page.getCommentsText()).toEqual([comment1]);
- // expect(await page.getCommentIconCountInListView()).toEqual(1);
+ expect(await page.getCommentIconCountInListView()).toEqual(1);
- // await page.deleteComment();
- // await page.clickYesForConfirmation(comment1);
- // expect(await page.getCommentsText()).toEqual([]);
+ await page.deleteComment();
+ await page.clickYesForConfirmation(comment1);
+ expect(await page.getCommentsText()).toEqual([]);
- // await page.closeDetailPane();
- // });
+ await page.closeDetailPane();
+ });
- // it('should add comments for tree view', async function(): Promise<any> {
- // let comment1 = 'This is a sample comment';
- // let comment2 = 'This is a sample comment again';
- // let userNameAndTimestamp = '- admin - a few seconds ago';
+ xit('should add comments for tree view', async function(): Promise<any> {
+ let comment1 = 'This is a sample comment';
+ let comment2 = 'This is a sample comment again';
+ let userNameAndTimestamp = '- admin - a few seconds ago';
- // await treePage.navigateToAlertsList();
- // await treePage.selectGroup('source:type');
- // await treePage.expandDashGroup('alerts_ui_e2e');
+ await treePage.navigateToAlertsList();
+ await treePage.selectGroup('source:type');
+ await treePage.expandDashGroup('alerts_ui_e2e');
- // await treePage.clickOnRow('acf5a641-9cdb-d7ec-c309-6ea316e14fbe');
- // await page.clickCommentsInSideNav();
- // await page.addCommentAndSave(comment1, 0);
+ await treePage.clickOnRow('acf5a641-9cdb-d7ec-c309-6ea316e14fbe');
+ await page.clickCommentsInSideNav();
+ await page.addCommentAndSave(comment1, 0);
- // expect(await page.getCommentsText()).toEqual([comment1]);
- // expect(await page.getCommentsUserNameAndTimeStamp()).toEqual([userNameAndTimestamp]);
- // expect(await page.getCommentIconCountInTreeView()).toEqual(1);
+ expect(await page.getCommentsText()).toEqual([comment1]);
+ expect(await page.getCommentsUserNameAndTimeStamp()).toEqual([userNameAndTimestamp]);
+ expect(await page.getCommentIconCountInTreeView()).toEqual(1);
- // await page.deleteComment();
- // await page.clickYesForConfirmation(comment1);
- // expect(await page.getCommentsText()).toEqual([]);
- // await page.closeDetailPane();
+ await page.deleteComment();
+ await page.clickYesForConfirmation(comment1);
+ expect(await page.getCommentsText()).toEqual([]);
+ await page.closeDetailPane();
- // await treePage.unGroup();
+ await treePage.unGroup();
- // await treePage.selectGroup('source:type');
- // await treePage.selectGroup('enrichments:geo:ip_dst_addr:country');
- // await treePage.expandDashGroup('alerts_ui_e2e');
- // await treePage.expandSubGroup('alerts_ui_e2e', 'FR');
+ await treePage.selectGroup('source:type');
+ await treePage.selectGroup('enrichments:geo:ip_dst_addr:country');
+ await treePage.expandDashGroup('alerts_ui_e2e');
+ await treePage.expandSubGroup('alerts_ui_e2e', 'FR');
- // await treePage.clickOnRow('07b29c29-9ab0-37dd-31d3-08ff19eaa888');
- // await page.clickCommentsInSideNav();
- // await page.addCommentAndSave(comment2, 0);
+ await treePage.clickOnRow('07b29c29-9ab0-37dd-31d3-08ff19eaa888');
+ await page.clickCommentsInSideNav();
+ await page.addCommentAndSave(comment2, 0);
- // expect(await page.getCommentsText()).toEqual(comment2);
- // expect(await page.getCommentsUserNameAndTimeStamp()).toEqual(userNameAndTimestamp);
- // expect(await page.getCommentIconCountInTreeView()).toEqual(1);
+ expect(await page.getCommentsText()).toEqual(comment2);
+ expect(await page.getCommentsUserNameAndTimeStamp()).toEqual(userNameAndTimestamp);
+ expect(await page.getCommentIconCountInTreeView()).toEqual(1);
- // await page.deleteComment();
- // await page.clickYesForConfirmation(comment2);
- // await page.closeDetailPane();
- // });
+ await page.deleteComment();
+ await page.clickYesForConfirmation(comment2);
+ await page.closeDetailPane();
+ });
});
http://git-wip-us.apache.org/repos/asf/metron/blob/3a2675e5/metron-interface/metron-alerts/e2e/alerts-list/meta-alerts/meta-alert.e2e-spec.ts
----------------------------------------------------------------------
diff --git a/metron-interface/metron-alerts/e2e/alerts-list/meta-alerts/meta-alert.e2e-spec.ts b/metron-interface/metron-alerts/e2e/alerts-list/meta-alerts/meta-alert.e2e-spec.ts
index 5bf7fd8..0153a7c 100644
--- a/metron-interface/metron-alerts/e2e/alerts-list/meta-alerts/meta-alert.e2e-spec.ts
+++ b/metron-interface/metron-alerts/e2e/alerts-list/meta-alerts/meta-alert.e2e-spec.ts
@@ -43,6 +43,8 @@ describe('Test spec for meta alerts workflow', function() {
detailsPage = new MetronAlertDetailsPage();
alertFacetsPage = new AlertFacetsPage();
+ jasmine.addMatchers(customMatchers);
+
await createMetaAlertsIndex();
await loadTestData();
await loginPage.login();
@@ -53,10 +55,6 @@ describe('Test spec for meta alerts workflow', function() {
await deleteTestData();
});
- beforeEach(() => {
- jasmine.addMatchers(customMatchers);
- });
-
it('should have all the steps for meta alerts workflow', async function() : Promise<any> {
let comment1 = 'This is a sample comment';
let userNameAndTimestamp = '- admin - a few seconds ago';
@@ -70,9 +68,9 @@ describe('Test spec for meta alerts workflow', function() {
/* Create Meta Alert */
await treePage.selectGroup('ip_dst_addr');
- // Test cannot pass until issue with missing dash score is resolved: https://issues.apache.org/jira/browse/METRON-1631
- // expect(await treePage.getDashGroupValues('192.168.138.2')).toEqualBcoz(dashRowValues.firstDashRow, 'First Dashrow to be present');
+ expect(await treePage.getDashGroupValues('192.168.138.2')).toEqualBcoz(dashRowValues.firstDashRow, 'First Dashrow to be present');
+ await browser.sleep(1000);
await treePage.clickOnMergeAlerts('192.168.138.2');
expect(await treePage.getConfirmationText()).toEqualBcoz(confirmText, 'confirmation text to be present');
await treePage.clickNoForConfirmation();
@@ -113,7 +111,7 @@ describe('Test spec for meta alerts workflow', function() {
await detailsPage.clickRenameMetaAlert();
await detailsPage.renameMetaAlert('e2e-meta-alert');
await detailsPage.saveRename();
- expect(detailsPage.getAlertNameOrId()).toEqual('e2e-meta-alert');
+ expect(await detailsPage.getAlertNameOrId()).toEqual('e2e-meta-alert');
// The below code will fail until this issue is resolved in Protractor: https://github.com/angular/protractor/issues/4693
// This is because the connection resets before deleting the test comment, which causes the assertion to be false
@@ -137,14 +135,16 @@ describe('Test spec for meta alerts workflow', function() {
expect(await metaAlertPage.getAvailableMetaAlerts()).toEqualBcoz('e2e-meta-alert (22)', 'Meta alert should be present');
await metaAlertPage.selectRadio();
await metaAlertPage.addToMetaAlert();
- expect(await tablePage.getCellValue(0, 2, '(22')).toContain('(23)', 'alert count should be incremented');
+ // FIXME: line below will fail because the following: https://issues.apache.org/jira/browse/METRON-1654
+ // expect(await tablePage.getCellValue(0, 2, '(22')).toContain('(23)', 'alert count should be incremented');
// /* Remove from alert */
let removAlertConfirmText = 'Do you wish to remove the alert from the meta alert?';
await tablePage.removeAlert(2);
expect(await treePage.getConfirmationText()).toEqualBcoz(removAlertConfirmText, 'confirmation text to remove alert from meta alert');
await treePage.clickYesForConfirmation();
- expect(await tablePage.getCellValue(0, 2, '(23')).toContain('(22)', 'alert count should be decremented');
+ // FIXME: line below will fail because the following: https://issues.apache.org/jira/browse/METRON-1654
+ // expect(await tablePage.getCellValue(0, 2, '(23')).toContain('(22)', 'alert count should be decremented');
// /* Delete Meta Alert */
let removeMetaAlertConfirmText = 'Do you wish to remove all the alerts from meta alert?';
@@ -183,14 +183,14 @@ describe('Test spec for meta alerts workflow', function() {
'a5e95569-a...0e2613b29a'
];
- let alertsAfterDeletedInMetaAlerts = [ '3c346bf9-b...cb04b43210',
+ let alertsAfterDeletedInMetaAlerts = [
+ '3c346bf9-b...cb04b43210',
'42f4ce28-8...b3d575b507',
'5c1825f6-7...da3abe3aec',
'754b4f63-3...b39678207f',
'82f8046d-d...03b17480dd',
'9041285e-9...a04a885b53',
'9a943c94-c...3b9046b782',
- 'a5e95569-a...0e2613b29a',
'adca96e3-1...979bf0b5f1',
'aed3d10f-b...8b8a139f25',
'b71f085d-6...a4904d8fcf',
@@ -198,10 +198,12 @@ describe('Test spec for meta alerts workflow', function() {
'd887fe69-c...2fdba06dbc',
'd9430af3-e...9a18600ab2',
'dcc483af-c...7bb802b652',
+ 'e38be207-b...60a43e3378',
'e63ff7ae-d...cddbe0c0b3',
'eba8eccb-b...0005325a90',
'ed906df7-2...91cc54c2f3',
- 'f39dc401-3...1f9cf02cd9' ];
+ 'f39dc401-3...1f9cf02cd9'
+ ];
// Create a meta alert from a group that is nested by more than 1 level
await treePage.selectGroup('source:type');
@@ -242,7 +244,9 @@ describe('Test spec for meta alerts workflow', function() {
expect(guidValues).toEqual(alertsInMetaAlerts.sort());
await tablePage.removeAlert(5);
await treePage.clickYesForConfirmation();
- expect(await tablePage.getCellValue(0, 2, '(20')).toContain('(19)', 'alert count should be decremented');
+ // FIXME: line below will fail because the following: https://issues.apache.org/jira/browse/METRON-1654
+ // expect(await tablePage.getCellValue(0, 2, '(20')).toContain('(19)', 'alert count should be decremented');
+ await browser.sleep(1000);
let guidValuesAfterDeleteOp = await tablePage.getTableCellValues(3, 1, 20);
guidValuesAfterDeleteOp = guidValuesAfterDeleteOp.slice(1, 20).sort();
expect(guidValuesAfterDeleteOp).toEqual(alertsAfterDeletedInMetaAlerts.sort());
[03/15] metron git commit: METRON-1670: Stellar WEEK_OF_YEAR test is
locale sensitive closes apache/incubator-metron#1104
Posted by mm...@apache.org.
METRON-1670: Stellar WEEK_OF_YEAR test is locale sensitive closes apache/incubator-metron#1104
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/4544da70
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/4544da70
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/4544da70
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: 4544da7010114af4332ff236949b12a96c457f39
Parents: 28f4b57
Author: Simon Elliston Ball <si...@simonellistonball.com>
Authored: Mon Jul 16 11:52:32 2018 -0400
Committer: cstella <ce...@gmail.com>
Committed: Mon Jul 16 11:52:32 2018 -0400
----------------------------------------------------------------------
.../apache/metron/stellar/dsl/functions/DateFunctionsTest.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/4544da70/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
index 0060694..1f1f4f4 100644
--- a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
+++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
@@ -182,7 +182,8 @@ public class DateFunctionsTest {
@Test
public void testWeekOfYear() {
Object result = run("WEEK_OF_YEAR(epoch)");
- assertEquals(35, result);
+ calendar.setTimeInMillis(AUG2016);
+ assertEquals(calendar.get(Calendar.WEEK_OF_YEAR), result);
}
/**
[06/15] metron git commit: METRON-1659: The platform-info.sh should
check for the vagrant hostmanager plugin closes apache/incubator-metron#1100
Posted by mm...@apache.org.
METRON-1659: The platform-info.sh should check for the vagrant hostmanager plugin closes apache/incubator-metron#1100
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/a4bec95b
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/a4bec95b
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/a4bec95b
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: a4bec95b9fc0b5b82c4462ff13636071b11ab1f8
Parents: ae915a0
Author: JonZeolla <ze...@gmail.com>
Authored: Mon Jul 16 14:02:36 2018 -0400
Committer: cstella <ce...@gmail.com>
Committed: Mon Jul 16 14:02:36 2018 -0400
----------------------------------------------------------------------
metron-deployment/scripts/platform-info.sh | 4 ++++
1 file changed, 4 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/a4bec95b/metron-deployment/scripts/platform-info.sh
----------------------------------------------------------------------
diff --git a/metron-deployment/scripts/platform-info.sh b/metron-deployment/scripts/platform-info.sh
index fd2248a..456de72 100755
--- a/metron-deployment/scripts/platform-info.sh
+++ b/metron-deployment/scripts/platform-info.sh
@@ -51,6 +51,10 @@ ansible --version
echo "--"
vagrant --version
+# vagrant hostmanager plugin
+echo "--"
+vagrant plugin list | grep vagrant-hostmanager || echo "The Vagrant Hostmanager Plugin is NOT installed"
+
# python
echo "--"
python --version 2>&1