You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by le...@apache.org on 2016/11/23 17:42:26 UTC
incubator-metron git commit: METRON-578 Missing error handling bolts
for enrichment and threat intel closes apache/incubator-metron#363
Repository: incubator-metron
Updated Branches:
refs/heads/master 8ec1b5d79 -> 054a4eaee
METRON-578 Missing error handling bolts for enrichment and threat intel closes apache/incubator-metron#363
Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/054a4eae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/054a4eae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/054a4eae
Branch: refs/heads/master
Commit: 054a4eaeea1d402d76ab4eb0836951654b784696
Parents: 8ec1b5d
Author: justinleet <ju...@gmail.com>
Authored: Wed Nov 23 12:36:22 2016 -0500
Committer: justinjleet <ju...@gmail.com>
Committed: Wed Nov 23 12:36:22 2016 -0500
----------------------------------------------------------------------
.../package/scripts/enrichment_commands.py | 32 +++--
.../package/scripts/params/params_linux.py | 2 +
.../package/scripts/params/status_params.py | 2 +
.../package/templates/enrichment.properties.j2 | 2 +
.../roles/metron_kafka_topics/defaults/main.yml | 2 +
.../org/apache/metron/common/Constants.java | 2 +
.../src/main/config/enrichment.properties | 2 +
.../src/main/flux/enrichment/remote.yaml | 88 ++++++++++++++
.../src/main/flux/enrichment/test.yaml | 119 +++++++++++++++++++
.../enrichment/bolt/GenericEnrichmentBolt.java | 29 +++--
.../enrichment/bolt/ErrorEnrichmentBolt.java | 46 +++++++
.../integration/EnrichmentIntegrationTest.java | 35 +++---
12 files changed, 322 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/enrichment_commands.py
----------------------------------------------------------------------
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/enrichment_commands.py b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/enrichment_commands.py
index f43073d..084c3e2 100755
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/enrichment_commands.py
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/enrichment_commands.py
@@ -20,7 +20,6 @@ import time
from resource_management.core.logger import Logger
from resource_management.core.resources.system import Execute, File
-from resource_management.libraries.resources.hdfs_resource import HdfsResource
import metron_service
@@ -30,6 +29,8 @@ class EnrichmentCommands:
__params = None
__enrichment_topology = None
__enrichment_topic = None
+ __enrichment_error_topic = None
+ __threat_intel_error_topic = None
__configured = False
def __init__(self, params):
@@ -38,6 +39,8 @@ class EnrichmentCommands:
self.__params = params
self.__enrichment_topology = params.metron_enrichment_topology
self.__enrichment_topic = params.metron_enrichment_topic
+ self.__enrichment_error_topic = params.metron_enrichment_error_topic
+ self.__threat_intel_error_topic = params.metron_threat_intel_error_topic
self.__configured = os.path.isfile(self.__params.enrichment_configured_flag_file)
def is_configured(self):
@@ -71,11 +74,11 @@ class EnrichmentCommands:
yum_repo_types[repo_type]()
Logger.info("Writing out repo file")
repo_template = ("echo \"[METRON-0.3.0]\n"
- "name=Metron 0.3.0 packages\n"
- "baseurl={0}\n"
- "gpgcheck=0\n"
- "enabled=1\n\""
- " > /etc/yum.repos.d/metron.repo")
+ "name=Metron 0.3.0 packages\n"
+ "baseurl={0}\n"
+ "gpgcheck=0\n"
+ "enabled=1\n\""
+ " > /etc/yum.repos.d/metron.repo")
Execute(repo_template.format(self.__params.repo_url))
else:
raise ValueError("Unsupported repo type '{0}'".format(repo_type))
@@ -93,15 +96,18 @@ class EnrichmentCommands:
replication_factor = 1
retention_gigabytes = int(self.__params.metron_topic_retention)
retention_bytes = retention_gigabytes * 1024 * 1024 * 1024
+
Logger.info("Creating topics for enrichment")
+ topics = [self.__enrichment_topic, self.__enrichment_error_topic, self.__threat_intel_error_topic]
+ for topic in topics:
+ Logger.info("Creating topic'{0}'".format(topic))
+ Execute(command_template.format(self.__params.kafka_bin_dir,
+ self.__params.zookeeper_quorum,
+ topic,
+ num_partitions,
+ replication_factor,
+ retention_bytes))
- Logger.info("Creating topic'{0}'".format(self.__enrichment_topic))
- Execute(command_template.format(self.__params.kafka_bin_dir,
- self.__params.zookeeper_quorum,
- self.__enrichment_topic,
- num_partitions,
- replication_factor,
- retention_bytes))
Logger.info("Done creating Kafka topics")
def init_hdfs_dir(self):
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/params_linux.py
----------------------------------------------------------------------
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/params_linux.py b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/params_linux.py
index c418ad9..030f8d1 100755
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/params_linux.py
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/params_linux.py
@@ -179,6 +179,8 @@ threatintel_cf = status_params.threatintel_cf
metron_enrichment_topology = status_params.metron_enrichment_topology
metron_enrichment_topic = status_params.metron_enrichment_topic
+metron_enrichment_error_topic = status_params.metron_enrichment_error_topic
+metron_threat_intel_error_topic = status_params.metron_threat_intel_error_topic
# ES Templates
bro_index_path = tmp_dir + "/bro_index.template"
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/status_params.py
----------------------------------------------------------------------
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/status_params.py b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/status_params.py
index 196defb..92c9d06 100644
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/status_params.py
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/scripts/params/status_params.py
@@ -34,6 +34,8 @@ parsers_configured_flag_file = metron_zookeeper_config_path + '/../metron_parser
# Enrichment
metron_enrichment_topology = 'enrichment'
metron_enrichment_topic = 'enrichments'
+metron_enrichment_error_topic = 'enrichments_error'
+metron_threat_intel_error_topic = 'threatintel_error'
enrichment_table = 'enrichment'
enrichment_cf = 't'
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/templates/enrichment.properties.j2
----------------------------------------------------------------------
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/templates/enrichment.properties.j2 b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/templates/enrichment.properties.j2
index bab2f52..14403de 100755
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/templates/enrichment.properties.j2
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/0.3.0/package/templates/enrichment.properties.j2
@@ -20,6 +20,8 @@
kafka.zk={{zookeeper_quorum}}
kafka.broker={{kafka_brokers}}
enrichment.output.topic=indexing
+enrichment.error.topic=enrichments_error
+threat.intel.error.topic=threatintel_error
##### MySQL #####
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-deployment/roles/metron_kafka_topics/defaults/main.yml
----------------------------------------------------------------------
diff --git a/metron-deployment/roles/metron_kafka_topics/defaults/main.yml b/metron-deployment/roles/metron_kafka_topics/defaults/main.yml
index fa0ee02..4a97a8e 100644
--- a/metron-deployment/roles/metron_kafka_topics/defaults/main.yml
+++ b/metron-deployment/roles/metron_kafka_topics/defaults/main.yml
@@ -22,6 +22,8 @@ topics_to_create:
- { topic: "yaf", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
- { topic: "snort", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
- { topic: "enrichments", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
+ - { topic: "enrichments_error", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
+ - { topic: "threatintel_error", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
- { topic: "parser_invalid", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
- { topic: "parser_error", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
- { topic: "indexing", num_partitions: 1, replication_factor: 1, retention_gb: 10 }
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-platform/metron-common/src/main/java/org/apache/metron/common/Constants.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/Constants.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/Constants.java
index 8b05a0b..4230678 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/Constants.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/Constants.java
@@ -27,6 +27,8 @@ public class Constants {
public static final long DEFAULT_CONFIGURED_BOLT_TIMEOUT = 5000;
public static final String SENSOR_TYPE = "source.type";
public static final String ENRICHMENT_TOPIC = "enrichments";
+ public static final String ENRICHMENT_ERROR_TOPIC = "enrichments_error";
+ public static final String THREAT_INTEL_ERROR_TOPIC = "threatintel_error";
public static final String INDEXING_TOPIC = "indexing";
public static final String INDEXING_ERROR_TOPIC = "indexing_error";
public static final String DEFAULT_PARSER_ERROR_TOPIC = "parser_error";
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-platform/metron-enrichment/src/main/config/enrichment.properties
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/config/enrichment.properties b/metron-platform/metron-enrichment/src/main/config/enrichment.properties
index e32915c..10206ea 100644
--- a/metron-platform/metron-enrichment/src/main/config/enrichment.properties
+++ b/metron-platform/metron-enrichment/src/main/config/enrichment.properties
@@ -20,6 +20,8 @@
kafka.zk=node1:2181
kafka.broker=node1:6667
enrichment.output.topic=indexing
+enrichment.error.topic=enrichments_error
+threat.intel.error.topic=threatintel_error
##### MySQL #####
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml b/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml
index 03f48a0..510bf18 100644
--- a/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml
+++ b/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml
@@ -107,6 +107,17 @@ components:
args:
- ref: "stellarEnrichment"
+ #enrichment error
+ - id: "enrichmentErrorKafkaWriter"
+ className: "org.apache.metron.writer.kafka.KafkaWriter"
+ configMethods:
+ - name: "withTopic"
+ args:
+ - "${enrichment.error.topic}"
+ - name: "withZkQuorum"
+ args:
+ - "${kafka.zk}"
+
# Threat Intel
- id: "stellarThreatIntelAdapter"
className: "org.apache.metron.enrichment.adapters.stellar.StellarAdapter"
@@ -159,6 +170,16 @@ components:
args:
- ref: "stellarThreatIntelEnrichment"
+ #threatintel error
+ - id: "threatIntelErrorKafkaWriter"
+ className: "org.apache.metron.writer.kafka.KafkaWriter"
+ configMethods:
+ - name: "withTopic"
+ args:
+ - "${threat.intel.error.topic}"
+ - name: "withZkQuorum"
+ args:
+ - "${kafka.zk}"
#indexing
- id: "kafkaWriter"
@@ -266,6 +287,14 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "enrichmentErrorOutputBolt"
+ className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withMessageWriter"
+ args:
+ - ref: "enrichmentErrorKafkaWriter"
# Threat Intel Bolts
- id: "threatIntelSplitBolt"
@@ -311,6 +340,15 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "threatIntelErrorOutputBolt"
+ className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withMessageWriter"
+ args:
+ - ref: "threatIntelErrorKafkaWriter"
+
# Indexing Bolts
- id: "outputBolt"
className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
@@ -402,6 +440,39 @@ streams:
type: FIELDS
args: ["key"]
+ # Error output
+ - name: "geoEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "geoEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "stellarEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "stellarEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "hostEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "hostEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "simpleHBaseEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "simpleHBaseEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
#threat intel
- name: "enrichmentJoin -> threatSplit"
from: "enrichmentJoinBolt"
@@ -459,3 +530,20 @@ streams:
type: FIELDS
args: ["key"]
+ # Error output
+ - name: "simpleHBaseThreatIntelBolt -> threatIntelErrorOutputBolt"
+ from: "simpleHBaseThreatIntelBolt"
+ to: "threatIntelErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "stellarThreatIntelBolt -> threatIntelErrorOutputBolt"
+ from: "stellarThreatIntelBolt"
+ to: "threatIntelErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-platform/metron-enrichment/src/main/flux/enrichment/test.yaml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/flux/enrichment/test.yaml b/metron-platform/metron-enrichment/src/main/flux/enrichment/test.yaml
index ca5868f..f3ce2bc 100644
--- a/metron-platform/metron-enrichment/src/main/flux/enrichment/test.yaml
+++ b/metron-platform/metron-enrichment/src/main/flux/enrichment/test.yaml
@@ -88,6 +88,18 @@ components:
- name: "add"
args:
- ref: "stellarEnrichment"
+
+ #enrichment error
+ - id: "enrichmentErrorKafkaWriter"
+ className: "org.apache.metron.writer.kafka.KafkaWriter"
+ configMethods:
+ - name: "withTopic"
+ args:
+ - "${enrichment.error.topic}"
+ - name: "withZkQuorum"
+ args:
+ - "${kafka.zk}"
+
# Threat Intel
- id: "stellarThreatIntelAdapter"
className: "org.apache.metron.enrichment.adapters.stellar.StellarAdapter"
@@ -140,6 +152,17 @@ components:
args:
- ref: "stellarThreatIntelEnrichment"
+ #threatintel error
+ - id: "threatIntelErrorKafkaWriter"
+ className: "org.apache.metron.writer.kafka.KafkaWriter"
+ configMethods:
+ - name: "withTopic"
+ args:
+ - "${threat.intel.error.topic}"
+ - name: "withZkQuorum"
+ args:
+ - "${kafka.zk}"
+
#indexing
- id: "kafkaWriter"
className: "org.apache.metron.writer.kafka.KafkaWriter"
@@ -236,6 +259,18 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "ErrorEnrichmentBolt"
+ className: "org.apache.metron.enrichment.bolt.ErrorEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "geoEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
- id: "enrichmentJoinBolt"
className: "org.apache.metron.enrichment.bolt.EnrichmentJoinBolt"
constructorArgs:
@@ -245,6 +280,14 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "enrichmentErrorOutputBolt"
+ className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withMessageWriter"
+ args:
+ - ref: "enrichmentErrorKafkaWriter"
# Threat Intel Bolts
- id: "threatIntelSplitBolt"
@@ -290,6 +333,15 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "threatIntelErrorOutputBolt"
+ className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withMessageWriter"
+ args:
+ - ref: "threatIntelErrorKafkaWriter"
+
# Indexing Bolts
- id: "outputBolt"
className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
@@ -341,6 +393,14 @@ streams:
type: FIELDS
args: ["key"]
+ - name: "enrichmentSplit -> ErrorEnrichmentBolt"
+ from: "enrichmentSplitBolt"
+ to: "ErrorEnrichmentBolt"
+ grouping:
+ streamId: "hbaseEnrichment"
+ type: FIELDS
+ args: ["key"]
+
- name: "splitter -> join"
from: "enrichmentSplitBolt"
to: "enrichmentJoinBolt"
@@ -379,6 +439,47 @@ streams:
type: FIELDS
args: ["key"]
+ # Error output
+ - name: "geoEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "geoEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "stellarEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "stellarEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "hostEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "hostEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "simpleHBaseEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "simpleHBaseEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "ErrorEnrichmentBolt -> enrichmentErrorOutputBolt"
+ from: "ErrorEnrichmentBolt"
+ to: "enrichmentErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
#threat intel
- name: "enrichmentJoin -> threatSplit"
from: "enrichmentJoinBolt"
@@ -434,3 +535,21 @@ streams:
streamId: "message"
type: FIELDS
args: ["key"]
+
+ # Error output
+ - name: "simpleHBaseThreatIntelBolt -> threatIntelErrorOutputBolt"
+ from: "simpleHBaseThreatIntelBolt"
+ to: "threatIntelErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
+ - name: "stellarThreatIntelBolt -> threatIntelErrorOutputBolt"
+ from: "stellarThreatIntelBolt"
+ to: "threatIntelErrorOutputBolt"
+ grouping:
+ streamId: "error"
+ type: FIELDS
+ args: ["message"]
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java
index 51d98b9..dc2e4f3 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java
@@ -69,6 +69,7 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
private static final Logger LOG = LoggerFactory
.getLogger(GenericEnrichmentBolt.class);
public static final String STELLAR_CONTEXT_CONF = "stellarContext";
+ private static final String ERROR_STREAM = "error";
private OutputCollector collector;
private Context stellarContext;
protected String enrichmentType;
@@ -166,7 +167,7 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declareStream(enrichmentType, new Fields("key", "message", "subgroup"));
- declarer.declareStream("error", new Fields("message"));
+ declarer.declareStream(ERROR_STREAM, new Fields("message"));
}
@SuppressWarnings("unchecked")
@@ -202,7 +203,7 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
if (value != null) {
SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sourceType);
if(config == null) {
- LOG.error("Unable to find " + config);
+ LOG.error("Unable to find SensorEnrichmentConfig for sourceType: " + sourceType);
error = true;
continue;
}
@@ -240,17 +241,27 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
if(error) {
throw new Exception("Unable to enrich " + enrichedMessage + " check logs for specifics.");
}
- if (enrichedMessage != null && !enrichedMessage.isEmpty()) {
+ if (!enrichedMessage.isEmpty()) {
collector.emit(enrichmentType, new Values(key, enrichedMessage, subGroup));
}
} catch (Exception e) {
- LOG.error("[Metron] Unable to enrich message: " + rawMessage, e);
- JSONObject error = ErrorUtils.generateErrorMessage("Enrichment problem: " + rawMessage, e);
- if (key != null) {
- collector.emit(enrichmentType, new Values(key, enrichedMessage, subGroup));
- }
- collector.emit("error", new Values(error));
+ handleError(key, rawMessage, subGroup, enrichedMessage, e);
+ }
+ }
+
+ // Made protected to allow for error testing in integration test. Directly flaws inputs while everything is functioning hits other
+ // errors, so this is made available in order to ensure ERROR_STREAM is output properly.
+ protected void handleError(String key, JSONObject rawMessage, String subGroup, JSONObject enrichedMessage, Exception e) {
+ LOG.error("[Metron] Unable to enrich message: " + rawMessage, e);
+ JSONObject error = ErrorUtils.generateErrorMessage("Enrichment problem: " + rawMessage, e);
+ if (key != null) {
+ collector.emit(enrichmentType, new Values(key, enrichedMessage, subGroup));
}
+ collector.emit(ERROR_STREAM, new Values(error));
+ }
+
+ protected void handleError() {
+
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ErrorEnrichmentBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ErrorEnrichmentBolt.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ErrorEnrichmentBolt.java
new file mode 100644
index 0000000..571079c
--- /dev/null
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ErrorEnrichmentBolt.java
@@ -0,0 +1,46 @@
+/*
+ * 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.enrichment.bolt;
+
+import org.json.simple.JSONObject;
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * Exists in order to provide a bolt that tests that when the GenericEnrichmentBolt writes to error, that it actually carries
+ * through the queue
+ */
+public class ErrorEnrichmentBolt extends GenericEnrichmentBolt {
+
+ public static final String TEST_ERROR_MESSAGE = "Test throwing error from ErrorEnrichmentBolt";
+
+ public ErrorEnrichmentBolt(String zookeeperUrl) {
+ super(zookeeperUrl);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void execute(Tuple tuple) {
+ JSONObject rawMessage = new JSONObject();
+ rawMessage.put("rawMessage", "Error Test Raw Message String");
+
+ JSONObject enrichedMessage= new JSONObject();
+ enrichedMessage.put("enrichedMessage", "Error Test Enriched Message String");
+ handleError("key", rawMessage, "subgroup", enrichedMessage, new IllegalStateException(TEST_ERROR_MESSAGE));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/054a4eae/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
index f6c3a14..a0199a1 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
@@ -26,6 +26,7 @@ import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.metron.TestConstants;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.EnrichmentConfigurations;
+import org.apache.metron.enrichment.bolt.ErrorEnrichmentBolt;
import org.apache.metron.enrichment.lookup.accesstracker.PersistentBloomTrackerCreator;
import org.apache.metron.enrichment.stellar.SimpleHBaseEnrichmentFunctions;
import org.apache.metron.hbase.TableProvider;
@@ -103,11 +104,13 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
setProperty("enrichment.simple.hbase.table", enrichmentsTableName);
setProperty("enrichment.simple.hbase.cf", cf);
setProperty("enrichment.output.topic", Constants.INDEXING_TOPIC);
+ setProperty("enrichment.error.topic", Constants.ENRICHMENT_ERROR_TOPIC);
}};
final ZKServerComponent zkServerComponent = getZKServerComponent(topologyProperties);
final KafkaComponent kafkaComponent = getKafkaComponent(topologyProperties, new ArrayList<KafkaComponent.Topic>() {{
add(new KafkaComponent.Topic(Constants.ENRICHMENT_TOPIC, 1));
add(new KafkaComponent.Topic(Constants.INDEXING_TOPIC, 1));
+ add(new KafkaComponent.Topic(Constants.ENRICHMENT_ERROR_TOPIC, 1));
}});
String globalConfigStr = null;
{
@@ -163,18 +166,14 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
kafkaComponent.writeMessages(Constants.ENRICHMENT_TOPIC, inputMessages);
ProcessorResult<List<Map<String, Object>>> result = runner.process(getProcessor(inputMessages));
+ // We expect failures, so we don't care if result returned failure or not
List<Map<String, Object>> docs = result.getResult();
- if (result.failed()){
- StringBuffer buffer = new StringBuffer();
- result.getBadResults(buffer);
- buffer.append(String.format("%d Valid Messages Processed", docs.size())).append("\n");
- dumpParsedMessages(docs,buffer);
- Assert.fail(buffer.toString());
- } else {
- Assert.assertEquals(inputMessages.size(), docs.size());
- List<Map<String, Object>> cleanedDocs = docs;
- validateAll(cleanedDocs);
- }
+ Assert.assertEquals(inputMessages.size(), docs.size());
+ validateAll(docs);
+
+ List<byte[]> errors = result.getProcessErrors();
+ Assert.assertEquals(inputMessages.size(), errors.size());
+ validateErrors(result.getProcessErrors());
} finally {
runner.stop();
}
@@ -199,6 +198,13 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
}
}
+ protected void validateErrors(List<byte[]> errors) {
+ for(byte[] error : errors) {
+ // Don't reconstruct the entire message, just ensure it contains the known error message inside.
+ Assert.assertTrue(new String(error).contains(ErrorEnrichmentBolt.TEST_ERROR_MESSAGE));
+ }
+ }
+
public static void baseValidation(Map<String, Object> jsonDoc) {
assertEnrichmentsExists("threatintels.", setOf("hbaseThreatIntel"), jsonDoc.keySet());
assertEnrichmentsExists("enrichments.", setOf("geo", "host", "hbaseEnrichment" ), jsonDoc.keySet());
@@ -448,6 +454,7 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
public ReadinessState process(ComponentRunner runner) {
KafkaComponent kafkaComponent = runner.getComponent("kafka", KafkaComponent.class);
List<byte[]> messages = kafkaComponent.readMessages(Constants.INDEXING_TOPIC);
+ errors = kafkaComponent.readMessages(Constants.ENRICHMENT_ERROR_TOPIC);
if (messages.size() == inputMessages.size()) {
docs = new ArrayList<>();
for(byte[] message : messages) {
@@ -459,12 +466,6 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
}
return ReadinessState.READY;
} else {
- errors = kafkaComponent.readMessages(Constants.ERROR_STREAM);
- invalids = kafkaComponent.readMessages(Constants.INVALID_STREAM);
- if(errors.size() > 0 || invalids.size() > 0) {
- messages = messages;
- return ReadinessState.READY;
- }
return ReadinessState.NOT_READY;
}
}