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/06/11 15:45:09 UTC
[03/50] [abbrv] metron git commit: METRON-1584 Indexing Topology
Crashes with Invalid Message (nickwallen) closes apache/metron#1036
METRON-1584 Indexing Topology Crashes with Invalid Message (nickwallen) closes apache/metron#1036
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/b6808f76
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/b6808f76
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/b6808f76
Branch: refs/heads/feature/METRON-1554-pcap-query-panel
Commit: b6808f76134260e8a3c24c7641d6e4a235ef48e1
Parents: 45e3ed8
Author: nickwallen <ni...@nickallen.org>
Authored: Wed Jun 6 11:07:30 2018 -0400
Committer: nickallen <ni...@apache.org>
Committed: Wed Jun 6 11:07:30 2018 -0400
----------------------------------------------------------------------
.../bolt/BulkMessageWriterBoltTest.java | 52 +++++++++--
.../metron/writer/BulkWriterComponent.java | 36 +++++++-
.../writer/bolt/BulkMessageWriterBolt.java | 92 +++++++++++++++-----
3 files changed, 145 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/b6808f76/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
index dedf5e6..52516ac 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
@@ -118,28 +118,32 @@ public class BulkMessageWriterBoltTest extends BaseEnrichmentBoltTest {
private MessageGetStrategy messageGetStrategy;
@Test
- public void testSensorTypeMissing() throws Exception {
+ public void testSourceTypeMissing() throws Exception {
+
+ // setup the bolt
BulkMessageWriterBolt bulkMessageWriterBolt = new BulkMessageWriterBolt("zookeeperUrl")
- .withBulkMessageWriter(bulkMessageWriter).withMessageGetter(MessageGetters.JSON_FROM_FIELD.name())
+ .withBulkMessageWriter(bulkMessageWriter)
+ .withMessageGetter(MessageGetters.JSON_FROM_FIELD.name())
.withMessageGetterField("message");
bulkMessageWriterBolt.setCuratorFramework(client);
bulkMessageWriterBolt.setZKCache(cache);
bulkMessageWriterBolt.getConfigurations().updateSensorIndexingConfig(sensorType,
new FileInputStream(sampleSensorIndexingConfigPath));
+ // initialize the bolt
bulkMessageWriterBolt.declareOutputFields(declarer);
- verify(declarer, times(1)).declareStream(eq("error"), argThat(
- new FieldsMatcher("message")));
Map stormConf = new HashMap();
bulkMessageWriterBolt.prepare(stormConf, topologyContext, outputCollector);
- BulkWriterComponent<JSONObject> component = mock(BulkWriterComponent.class);
- bulkMessageWriterBolt.setWriterComponent(component);
- verify(bulkMessageWriter, times(1)).init(eq(stormConf),any(TopologyContext.class), any(WriterConfiguration.class));
+
+ // create a message with no source type
JSONObject message = (JSONObject) new JSONParser().parse(sampleMessageString);
message.remove("source.type");
when(tuple.getValueByField("message")).thenReturn(message);
+
+ // the tuple should be handled as an error and ack'd
bulkMessageWriterBolt.execute(tuple);
- verify(component, times(1)).error(eq("null"), any(), any(), any());
+ verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), any());
+ verify(outputCollector, times(1)).ack(tuple);
}
@Test
@@ -295,4 +299,36 @@ public class BulkMessageWriterBoltTest extends BaseEnrichmentBoltTest {
assertEquals(3, tupleList.size());
verify(outputCollector, times(5)).ack(tuple); // 3 messages + 2nd tick
}
+
+ /**
+ * If an invalid message is sent to indexing, the message should be handled as an error
+ * and the topology should continue processing.
+ */
+ @Test
+ public void testMessageInvalid() throws Exception {
+ FakeClock clock = new FakeClock();
+
+ // setup the bolt
+ BulkMessageWriterBolt bolt = new BulkMessageWriterBolt("zookeeperUrl")
+ .withBulkMessageWriter(bulkMessageWriter)
+ .withMessageGetter(MessageGetters.JSON_FROM_POSITION.name())
+ .withMessageGetterField("message");
+ bolt.setCuratorFramework(client);
+ bolt.setZKCache(cache);
+ bolt.getConfigurations().updateSensorIndexingConfig(sensorType, new FileInputStream(sampleSensorIndexingConfigPath));
+
+ // initialize the bolt
+ bolt.declareOutputFields(declarer);
+ Map stormConf = new HashMap();
+ bolt.prepare(stormConf, topologyContext, outputCollector, clock);
+
+ // execute a tuple that contains an invalid message
+ byte[] invalidJSON = "this is not valid JSON".getBytes();
+ when(tuple.getBinary(0)).thenReturn(invalidJSON);
+ bolt.execute(tuple);
+
+ // the tuple should be handled as an error and ack'd
+ verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), any());
+ verify(outputCollector, times(1)).ack(tuple);
+ }
}
http://git-wip-us.apache.org/repos/asf/metron/blob/b6808f76/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 37c624f..dfa265d 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
@@ -19,6 +19,7 @@
package org.apache.metron.writer;
import com.google.common.collect.Iterables;
+import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.writer.WriterConfiguration;
import org.apache.metron.common.error.MetronError;
@@ -41,6 +42,8 @@ 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.
@@ -115,15 +118,40 @@ public class BulkWriterComponent<MESSAGE_T> {
}
public void error(String sensorType, Throwable e, Iterable<Tuple> tuples, MessageGetStrategy messageGetStrategy) {
- tuples.forEach(t -> collector.ack(t));
+ LOG.error(format("Failing %d tuple(s); sensorType=%s", Iterables.size(tuples), sensorType), e);
MetronError error = new MetronError()
.withSensorType(sensorType)
.withErrorType(Constants.ErrorType.INDEXING_ERROR)
.withThrowable(e);
- if(!Iterables.isEmpty(tuples)) {
- LOG.error("Failing {} tuples", Iterables.size(tuples), e);
- }
tuples.forEach(t -> error.addRawMessage(messageGetStrategy.get(t)));
+ handleError(tuples, error);
+ }
+
+ /**
+ * Error a set of tuples that may not contain a valid message.
+ *
+ * <p>Without a valid message, the source type is unknown.
+ * <p>Without a valid message, the JSON message cannot be added to the error.
+ *
+ * @param e The exception that occurred.
+ * @param tuples The tuples to error that may not contain valid messages.
+ */
+ public void error(Throwable e, Iterable<Tuple> tuples) {
+ LOG.error(format("Failing %d tuple(s)", Iterables.size(tuples)), e);
+ MetronError error = new MetronError()
+ .withErrorType(Constants.ErrorType.INDEXING_ERROR)
+ .withThrowable(e);
+ handleError(tuples, error);
+ }
+
+ /**
+ * Errors a set of tuples.
+ *
+ * @param tuples The tuples to error.
+ * @param error
+ */
+ private void handleError(Iterable<Tuple> tuples, MetronError error) {
+ tuples.forEach(t -> collector.ack(t));
ErrorUtils.handleError(collector, error);
}
http://git-wip-us.apache.org/repos/asf/metron/blob/b6808f76/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
index b5b97d8..1d8f0c6 100644
--- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
+++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
@@ -234,41 +234,87 @@ public class BulkMessageWriterBolt extends ConfiguredIndexingBolt {
try
{
- JSONObject message = (JSONObject) messageGetStrategy.get(tuple);
+ JSONObject message = getMessage(tuple);
+ if(message == null) {
+ handleMissingMessage(tuple);
+ return;
+ }
+
String sensorType = MessageUtils.getSensorType(message);
+ if(sensorType == null) {
+ handleMissingSensorType(tuple, message);
+ return;
+ }
+
LOG.trace("Writing enrichment message: {}", message);
WriterConfiguration writerConfiguration = configurationTransformation.apply(
new IndexingWriterConfiguration(bulkMessageWriter.getName(), getConfigurations()));
- if(sensorType == null) {
- //sensor type somehow ended up being null. We want to error this message directly.
- getWriterComponent().error("null"
- , new Exception("Sensor type is not specified for message "
- + message.toJSONString()
- )
- , ImmutableList.of(tuple)
- , messageGetStrategy
- );
- }
- else {
- if (writerConfiguration.isDefault(sensorType)) {
- //want to warn, but not fail the tuple
- collector.reportError(new Exception("WARNING: Default and (likely) unoptimized writer config used for " + bulkMessageWriter.getName() + " writer and sensor " + sensorType));
- }
- getWriterComponent().write(sensorType
- , tuple
- , message
- , bulkMessageWriter
- , writerConfiguration
- , messageGetStrategy
- );
+ if (writerConfiguration.isDefault(sensorType)) {
+ //want to warn, but not fail the tuple
+ collector.reportError(new Exception("WARNING: Default and (likely) unoptimized writer config used for " + bulkMessageWriter.getName() + " writer and sensor " + sensorType));
}
+
+ getWriterComponent().write(sensorType
+ , tuple
+ , message
+ , bulkMessageWriter
+ , writerConfiguration
+ , messageGetStrategy
+ );
}
catch(Exception e) {
throw new RuntimeException("This should have been caught in the writerComponent. If you see this, file a JIRA", e);
}
}
+ /**
+ * Retrieves the JSON message contained in a tuple.
+ *
+ * @param tuple The tuple containing a JSON message.
+ * @return The JSON message contained in the tuple. If none, returns null.
+ */
+ private JSONObject getMessage(Tuple tuple) {
+ JSONObject message = null;
+ try {
+ message = (JSONObject) messageGetStrategy.get(tuple);
+
+ } catch(Throwable e) {
+ LOG.error("Unable to retrieve message from tuple", e);
+ }
+
+ return message;
+ }
+
+ /**
+ * Handles error processing when a message is missing a sensor type.
+ *
+ * @param tuple The tuple.
+ * @param message The message with no sensor type.
+ */
+ private void handleMissingSensorType(Tuple tuple, JSONObject message) {
+ // sensor type somehow ended up being null. We want to error this message directly.
+ LOG.debug("Message is missing sensor type");
+ getWriterComponent().error("null",
+ new Exception("Sensor type is not specified for message " + message.toJSONString()),
+ ImmutableList.of(tuple),
+ messageGetStrategy
+ );
+ }
+
+ /**
+ * Handles error processing when a tuple does not contain a valid message.
+ *
+ * @param tuple The tuple.
+ */
+ private void handleMissingMessage(Tuple tuple) {
+ LOG.debug("Unable to extract message from tuple; expected valid JSON");
+ getWriterComponent().error(
+ new Exception("Unable to extract message from tuple; expected valid JSON"),
+ ImmutableList.of(tuple)
+ );
+ }
+
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declareStream(Constants.ERROR_STREAM, new Fields("message"));