You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ce...@apache.org on 2016/08/23 15:41:27 UTC
[1/2] incubator-metron git commit: METRON-341: Create stellar
enrichment adapter closes apache/incubator-metron#216
Repository: incubator-metron
Updated Branches:
refs/heads/master f5c6d0093 -> 58294b665
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/stellar/StellarAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/stellar/StellarAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/stellar/StellarAdapter.java
new file mode 100644
index 0000000..3ba7755
--- /dev/null
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/stellar/StellarAdapter.java
@@ -0,0 +1,110 @@
+/**
+ * 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.adapters.stellar;
+
+import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler;
+import org.apache.metron.common.dsl.Context;
+import org.apache.metron.common.dsl.MapVariableResolver;
+import org.apache.metron.common.dsl.StellarFunctions;
+import org.apache.metron.common.dsl.VariableResolver;
+import org.apache.metron.common.stellar.StellarProcessor;
+import org.apache.metron.enrichment.bolt.CacheKey;
+import org.apache.metron.enrichment.interfaces.EnrichmentAdapter;
+import org.json.simple.JSONObject;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.function.Function;
+
+import static org.apache.metron.enrichment.bolt.GenericEnrichmentBolt.STELLAR_CONTEXT_CONF;
+
+public class StellarAdapter implements EnrichmentAdapter<CacheKey>,Serializable {
+
+ private enum EnrichmentType implements Function<SensorEnrichmentConfig, ConfigHandler>{
+ ENRICHMENT(config -> config.getEnrichment().getEnrichmentConfigs().get("stellar"))
+ ,THREAT_INTEL(config -> config.getThreatIntel().getEnrichmentConfigs().get("stellar"))
+ ;
+ Function<SensorEnrichmentConfig, ConfigHandler> func;
+ EnrichmentType(Function<SensorEnrichmentConfig, ConfigHandler> func) {
+ this.func = func;
+ }
+
+ @Override
+ public ConfigHandler apply(SensorEnrichmentConfig cacheKey) {
+ return func.apply(cacheKey);
+ }
+ }
+ transient Function<SensorEnrichmentConfig, ConfigHandler> getHandler;
+ private String enrichmentType;
+ public StellarAdapter ofType(String enrichmentType) {
+ this.enrichmentType = enrichmentType;
+ return this;
+ }
+ @Override
+ public String getOutputPrefix(CacheKey value) {
+ return "";
+ }
+
+ @Override
+ public void logAccess(CacheKey value) {
+
+ }
+
+ @Override
+ public String getStreamSubGroup(String enrichmentType, String field) {
+ return field;
+ }
+
+ @Override
+ public JSONObject enrich(CacheKey value) {
+ Context stellarContext = (Context) value.getConfig().getConfiguration().get(STELLAR_CONTEXT_CONF);
+ ConfigHandler handler = getHandler.apply(value.getConfig());
+ Map<String, Object> globalConfig = value.getConfig().getConfiguration();
+ Map<String, Object> sensorConfig = value.getConfig().getEnrichment().getConfig();
+ if(handler == null) {
+ return new JSONObject();
+ }
+ Map<String, Object> message = value.getValue(Map.class);
+ VariableResolver resolver = new MapVariableResolver(message, sensorConfig, globalConfig);
+ StellarProcessor processor = new StellarProcessor();
+ Map<String, Object> stellarStatements = value.getField().length() == 0? handler.getConfig()
+ : (Map)handler.getConfig().get(value.getField());
+ if(stellarStatements != null) {
+ for (Map.Entry<String, Object> kv : stellarStatements.entrySet()) {
+ if(kv.getValue() instanceof String) {
+ String stellarStatement = (String) kv.getValue();
+ Object o = processor.parse(stellarStatement, resolver, StellarFunctions.FUNCTION_RESOLVER(), stellarContext);
+ message.put(kv.getKey(), o);
+ }
+ }
+ }
+ return new JSONObject(message);
+ }
+
+ @Override
+ public boolean initializeAdapter() {
+ getHandler = EnrichmentType.valueOf(enrichmentType);
+ return true;
+ }
+
+ @Override
+ public void cleanup() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java
index 09bb0c1..cb7ebe0 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java
@@ -57,7 +57,7 @@ public class ThreatIntelAdapter implements EnrichmentAdapter<CacheKey>,Serializa
List<String> enrichmentTypes = value.getConfig().getThreatIntel().getFieldToTypeMap().get(value.getField());
if(enrichmentTypes != null) {
for(String enrichmentType : enrichmentTypes) {
- lookup.getAccessTracker().logAccess(new EnrichmentKey(enrichmentType, value.getValue()));
+ lookup.getAccessTracker().logAccess(new EnrichmentKey(enrichmentType, value.getValue(String.class)));
}
}
}
@@ -77,7 +77,7 @@ public class ThreatIntelAdapter implements EnrichmentAdapter<CacheKey>,Serializa
try {
for (Boolean isThreat :
lookup.exists(Iterables.transform(enrichmentTypes
- , new EnrichmentUtils.TypeToKey(value.getValue()
+ , new EnrichmentUtils.TypeToKey(value.getValue(String.class)
, lookup.getTable()
, value.getConfig().getThreatIntel()
)
@@ -142,4 +142,9 @@ public class ThreatIntelAdapter implements EnrichmentAdapter<CacheKey>,Serializa
throw new RuntimeException("Unable to cleanup access tracker", e);
}
}
+
+ @Override
+ public String getOutputPrefix(CacheKey value) {
+ return value.getField();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/CacheKey.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/CacheKey.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/CacheKey.java
index 4dbd33e..a3a92ac 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/CacheKey.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/CacheKey.java
@@ -18,13 +18,14 @@
package org.apache.metron.enrichment.bolt;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.utils.ConversionUtils;
public class CacheKey {
private String field;
- private String value;
+ private Object value;
private SensorEnrichmentConfig config;
- public CacheKey(String field, String value, SensorEnrichmentConfig config) {
+ public CacheKey(String field, Object value, SensorEnrichmentConfig config) {
this.field = field;
this.value = value;
this.config = config;
@@ -34,10 +35,17 @@ public class CacheKey {
return field;
}
- public String getValue() {
+ public Object getValue() {
return value;
}
+ public <T> T getValue(Class<T> clazz) {
+ return clazz.cast(getValue());
+ }
+
+ public <T> T coerceValue(Class<T> clazz) {
+ return ConversionUtils.convert(getValue(), clazz);
+ }
public SensorEnrichmentConfig getConfig() {
return config;
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java
index 6340c95..dd936f6 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java
@@ -18,17 +18,15 @@
package org.apache.metron.enrichment.bolt;
import backtype.storm.task.TopologyContext;
+import com.google.common.base.Joiner;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler;
import org.apache.metron.common.utils.MessageUtils;
import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
public class EnrichmentJoinBolt extends JoinBolt<JSONObject> {
@@ -52,13 +50,17 @@ public class EnrichmentJoinBolt extends JoinBolt<JSONObject> {
String errorMessage = "Unable to find source type for message: " + message;
throw new IllegalStateException(errorMessage);
}
- Map<String, List<String>> fieldMap = getFieldMap(sourceType);
+ Map<String, Object> fieldMap = getFieldMap(sourceType);
+ Map<String, ConfigHandler> handlerMap = getFieldToHandlerMap(sourceType);
if(fieldMap != null) {
- for (String enrichmentType : getFieldMap(sourceType).keySet()) {
- streamIds.add(enrichmentType);
+ for (String enrichmentType : fieldMap.keySet()) {
+ ConfigHandler handler = handlerMap.get(enrichmentType);
+ for(String subgroup : handler.getType().getSubgroups(handler.getConfig())) {
+ streamIds.add(Joiner.on(":").join(enrichmentType, subgroup));
+ }
}
}
- streamIds.add("message");
+ streamIds.add("message:");
return streamIds;
}
@@ -84,7 +86,21 @@ public class EnrichmentJoinBolt extends JoinBolt<JSONObject> {
return message;
}
- public Map<String, List<String>> getFieldMap(String sourceType) {
+ protected Map<String, ConfigHandler> getFieldToHandlerMap(String sensorType) {
+ if(sensorType != null) {
+ SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sensorType);
+ if (config != null) {
+ return config.getEnrichment().getEnrichmentConfigs();
+ } else {
+ LOG.error("Unable to retrieve a sensor enrichment config of " + sensorType);
+ }
+ } else {
+ LOG.error("Trying to retrieve a field map with sensor type of null");
+ }
+ return new HashMap<>();
+ }
+
+ public Map<String, Object> getFieldMap(String sourceType) {
if(sourceType != null) {
SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sourceType);
if (config != null && config.getEnrichment() != null) {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBolt.java
index 4b5c7bb..e67db7e 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBolt.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBolt.java
@@ -22,6 +22,7 @@ import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler;
import org.apache.metron.enrichment.configuration.Enrichment;
import org.apache.metron.enrichment.utils.EnrichmentUtils;
import org.apache.metron.common.utils.MessageUtils;
@@ -35,119 +36,139 @@ import java.io.UnsupportedEncodingException;
import java.util.*;
public class EnrichmentSplitterBolt extends SplitBolt<JSONObject> {
- protected static final Logger LOG = LoggerFactory.getLogger(EnrichmentSplitterBolt.class);
- private List<Enrichment> enrichments;
- protected String messageFieldName;
- private transient JSONParser parser;
-
-
- public EnrichmentSplitterBolt(String zookeeperUrl) {
- super(zookeeperUrl);
+ protected static final Logger LOG = LoggerFactory.getLogger(EnrichmentSplitterBolt.class);
+ private List<Enrichment> enrichments;
+ protected String messageFieldName;
+ private transient JSONParser parser;
+
+
+ public EnrichmentSplitterBolt(String zookeeperUrl) {
+ super(zookeeperUrl);
+ }
+
+ public EnrichmentSplitterBolt withEnrichments(List<Enrichment> enrichments) {
+ this.enrichments = enrichments;
+ return this;
+ }
+
+ public EnrichmentSplitterBolt withMessageFieldName(String messageFieldName) {
+ this.messageFieldName = messageFieldName;
+ return this;
+ }
+ @Override
+ public void prepare(Map map, TopologyContext topologyContext) {
+ parser = new JSONParser();
+ }
+ @Override
+ public String getKey(Tuple tuple, JSONObject message) {
+ String key = null;
+ try {
+ key = tuple.getStringByField("key");
}
-
- public EnrichmentSplitterBolt withEnrichments(List<Enrichment> enrichments) {
- this.enrichments = enrichments;
- return this;
+ catch(Throwable t) {
+ //swallowing this just in case.
}
-
- public EnrichmentSplitterBolt withMessageFieldName(String messageFieldName) {
- this.messageFieldName = messageFieldName;
- return this;
+ if(key != null) {
+ return key;
}
- @Override
- public void prepare(Map map, TopologyContext topologyContext) {
- parser = new JSONParser();
+ else {
+ return UUID.randomUUID().toString();
}
- @Override
- public String getKey(Tuple tuple, JSONObject message) {
- String key = null;
- try {
- key = tuple.getStringByField("key");
- }
- catch(Throwable t) {
- //swallowing this just in case.
- }
- if(key != null) {
- return key;
- }
- else {
- return UUID.randomUUID().toString();
- }
+ }
+
+ @Override
+ public JSONObject generateMessage(Tuple tuple) {
+ JSONObject message = null;
+ if (messageFieldName == null) {
+ byte[] data = tuple.getBinary(0);
+ try {
+ message = (JSONObject) parser.parse(new String(data, "UTF8"));
+ message.put(getClass().getSimpleName().toLowerCase() + ".splitter.begin.ts", "" + System.currentTimeMillis());
+ } catch (ParseException | UnsupportedEncodingException e) {
+ e.printStackTrace();
+ }
+ } else {
+ message = (JSONObject) tuple.getValueByField(messageFieldName);
+ message.put(getClass().getSimpleName().toLowerCase() + ".splitter.begin.ts", "" + System.currentTimeMillis());
}
-
- @Override
- public JSONObject generateMessage(Tuple tuple) {
- JSONObject message = null;
- if (messageFieldName == null) {
- byte[] data = tuple.getBinary(0);
- try {
- message = (JSONObject) parser.parse(new String(data, "UTF8"));
- message.put(getClass().getSimpleName().toLowerCase() + ".splitter.begin.ts", "" + System.currentTimeMillis());
- } catch (ParseException | UnsupportedEncodingException e) {
- e.printStackTrace();
- }
- } else {
- message = (JSONObject) tuple.getValueByField(messageFieldName);
- message.put(getClass().getSimpleName().toLowerCase() + ".splitter.begin.ts", "" + System.currentTimeMillis());
- }
- return message;
+ return message;
+ }
+
+ @Override
+ public Set<String> getStreamIds() {
+ Set<String> streamIds = new HashSet<>();
+ for(Enrichment enrichment: enrichments) {
+ streamIds.add(enrichment.getType());
}
-
- @Override
- public Set<String> getStreamIds() {
- Set<String> streamIds = new HashSet<>();
- for(Enrichment enrichment: enrichments) {
- streamIds.add(enrichment.getType());
- }
- return streamIds;
+ return streamIds;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Map<String, List<JSONObject>> splitMessage(JSONObject message) {
+ Map<String, List<JSONObject>> streamMessageMap = new HashMap<>();
+ String sensorType = MessageUtils.getSensorType(message);
+ Map<String, Object> enrichmentFieldMap = getFieldMap(sensorType);
+ Map<String, ConfigHandler> fieldToHandler = getFieldToHandlerMap(sensorType);
+ Set<String> enrichmentTypes = new HashSet<>(enrichmentFieldMap.keySet());
+ enrichmentTypes.addAll(fieldToHandler.keySet());
+ for (String enrichmentType : enrichmentTypes) {
+ Object fields = enrichmentFieldMap.get(enrichmentType);
+ ConfigHandler retriever = fieldToHandler.get(enrichmentType);
+
+ List<JSONObject> enrichmentObject = retriever.getType()
+ .splitByFields( message
+ , fields
+ , field -> getKeyName(enrichmentType, field)
+ , retriever.getConfig()
+ );
+ for(JSONObject eo : enrichmentObject) {
+ eo.put(Constants.SENSOR_TYPE, sensorType);
+ }
+ streamMessageMap.put(enrichmentType, enrichmentObject);
}
-
- @SuppressWarnings("unchecked")
- @Override
- public Map<String, JSONObject> splitMessage(JSONObject message) {
- Map<String, JSONObject> streamMessageMap = new HashMap<>();
- String sensorType = MessageUtils.getSensorType(message);
- Map<String, List<String>> enrichmentFieldMap = getFieldMap(sensorType);
- for (String enrichmentType : enrichmentFieldMap.keySet()) {
- List<String> fields = enrichmentFieldMap.get(enrichmentType);
- JSONObject enrichmentObject = new JSONObject();
- if (fields != null && fields.size() > 0) {
- for (String field : fields) {
- enrichmentObject.put(getKeyName(enrichmentType, field), message.get(field));
- }
- enrichmentObject.put(Constants.SENSOR_TYPE, sensorType);
- streamMessageMap.put(enrichmentType, enrichmentObject);
- }
- }
- message.put(getClass().getSimpleName().toLowerCase() + ".splitter.end.ts", "" + System.currentTimeMillis());
- return streamMessageMap;
+ message.put(getClass().getSimpleName().toLowerCase() + ".splitter.end.ts", "" + System.currentTimeMillis());
+ return streamMessageMap;
+ }
+
+ protected Map<String, ConfigHandler> getFieldToHandlerMap(String sensorType) {
+ if(sensorType != null) {
+ SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sensorType);
+ if (config != null) {
+ return config.getEnrichment().getEnrichmentConfigs();
+ } else {
+ LOG.error("Unable to retrieve a sensor enrichment config of " + sensorType);
+ }
+ } else {
+ LOG.error("Trying to retrieve a field map with sensor type of null");
}
-
- protected Map<String, List<String>> getFieldMap(String sensorType) {
- if(sensorType != null) {
- SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sensorType);
- if (config != null) {
- return config.getEnrichment().getFieldMap();
- } else {
- LOG.error("Unable to retrieve a sensor enrichment config of " + sensorType);
- }
- } else {
- LOG.error("Trying to retrieve a field map with sensor type of null");
- }
- return new HashMap<>();
+ return new HashMap<>();
+ }
+ protected Map<String, Object > getFieldMap(String sensorType) {
+ if(sensorType != null) {
+ SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sensorType);
+ if (config != null) {
+ return config.getEnrichment().getFieldMap();
+ } else {
+ LOG.error("Unable to retrieve a sensor enrichment config of " + sensorType);
+ }
+ } else {
+ LOG.error("Trying to retrieve a field map with sensor type of null");
}
+ return new HashMap<>();
+ }
- protected String getKeyName(String type, String field) {
- return EnrichmentUtils.getEnrichmentKey(type, field);
- }
+ protected String getKeyName(String type, String field) {
+ return EnrichmentUtils.getEnrichmentKey(type, field);
+ }
- @Override
- public void declareOther(OutputFieldsDeclarer declarer) {
+ @Override
+ public void declareOther(OutputFieldsDeclarer declarer) {
- }
+ }
- @Override
- public void emitOther(Tuple tuple, JSONObject message) {
+ @Override
+ public void emitOther(Tuple tuple, JSONObject message) {
- }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/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 d4acd08..5039f4b 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
@@ -27,10 +27,13 @@ import backtype.storm.tuple.Values;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
+import org.apache.commons.lang3.StringUtils;
import org.apache.metron.common.Constants;
import org.apache.metron.common.bolt.ConfiguredEnrichmentBolt;
import org.apache.metron.common.configuration.ConfigurationType;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.dsl.Context;
+import org.apache.metron.common.dsl.StellarFunctions;
import org.apache.metron.common.utils.ErrorUtils;
import org.apache.metron.enrichment.configuration.Enrichment;
import org.apache.metron.enrichment.interfaces.EnrichmentAdapter;
@@ -65,8 +68,9 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
private static final Logger LOG = LoggerFactory
.getLogger(GenericEnrichmentBolt.class);
+ public static final String STELLAR_CONTEXT_CONF = "stellarContext";
private OutputCollector collector;
-
+ private Context stellarContext;
protected String enrichmentType;
protected EnrichmentAdapter<CacheKey> adapter;
protected transient CacheLoader<CacheKey, JSONObject> loader;
@@ -148,11 +152,19 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
LOG.error("[Metron] EnrichmentSplitterBolt could not initialize adapter");
throw new IllegalStateException("Could not initialize adapter...");
}
+ initializeStellar();
+ }
+
+ protected void initializeStellar() {
+ stellarContext = new Context.Builder()
+ .with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
+ .build();
+ StellarFunctions.FUNCTION_RESOLVER().initializeFunctions(stellarContext);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declareStream(enrichmentType, new Fields("key", "message"));
+ declarer.declareStream(enrichmentType, new Fields("key", "message", "subgroup"));
declarer.declareStream("error", new Fields("message"));
}
@@ -161,6 +173,7 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
public void execute(Tuple tuple) {
String key = tuple.getStringByField("key");
JSONObject rawMessage = (JSONObject) tuple.getValueByField("message");
+ String subGroup = "";
JSONObject enrichedMessage = new JSONObject();
enrichedMessage.put("adapter." + adapter.getClass().getSimpleName().toLowerCase() + ".begin.ts", "" + System.currentTimeMillis());
@@ -177,23 +190,27 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
throw new RuntimeException("Source type is missing from enrichment fragment: " + rawMessage.toJSONString());
}
boolean error = false;
+ String prefix = null;
for (Object o : rawMessage.keySet()) {
String field = (String) o;
- String value = (String) rawMessage.get(field);
+ Object value = rawMessage.get(field);
if (field.equals(Constants.SENSOR_TYPE)) {
enrichedMessage.put(Constants.SENSOR_TYPE, value);
} else {
JSONObject enrichedField = new JSONObject();
- if (value != null && value.length() != 0) {
+ if (value != null) {
SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sourceType);
if(config == null) {
LOG.error("Unable to find " + config);
error = true;
continue;
}
+ config.getConfiguration().putIfAbsent(STELLAR_CONTEXT_CONF, stellarContext);
CacheKey cacheKey= new CacheKey(field, value, config);
try {
adapter.logAccess(cacheKey);
+ prefix = adapter.getOutputPrefix(cacheKey);
+ subGroup = adapter.getStreamSubGroup(enrichmentType, field);
enrichedField = cache.getUnchecked(cacheKey);
if (enrichedField == null)
throw new Exception("[Metron] Could not enrich string: "
@@ -205,12 +222,15 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
continue;
}
}
- if (!enrichedField.isEmpty()) {
+ if ( !enrichedField.isEmpty()) {
for (Object enrichedKey : enrichedField.keySet()) {
- enrichedMessage.put(field + "." + enrichedKey, enrichedField.get(enrichedKey));
+ if(!StringUtils.isEmpty(prefix)) {
+ enrichedMessage.put(field + "." + enrichedKey, enrichedField.get(enrichedKey));
+ }
+ else {
+ enrichedMessage.put(enrichedKey, enrichedField.get(enrichedKey));
+ }
}
- } else {
- enrichedMessage.put(field, "");
}
}
}
@@ -220,13 +240,13 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
throw new Exception("Unable to enrich " + enrichedMessage + " check logs for specifics.");
}
if (enrichedMessage != null && !enrichedMessage.isEmpty()) {
- collector.emit(enrichmentType, new Values(key, enrichedMessage));
+ 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));
+ collector.emit(enrichmentType, new Values(key, enrichedMessage, subGroup));
}
collector.emit("error", new Values(error));
}
@@ -236,4 +256,8 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt {
public void cleanup() {
adapter.cleanup();
}
+
+ public Context getStellarContext() {
+ return stellarContext;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java
index 771097f..2eb4b0a 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java
@@ -91,6 +91,8 @@ public abstract class JoinBolt<V> extends ConfiguredEnrichmentBolt {
public void execute(Tuple tuple) {
String streamId = tuple.getSourceStreamId();
String key = (String) tuple.getValueByField("key");
+ String subgroup = (String) tuple.getValueByField("subgroup");
+ streamId = Joiner.on(":").join("" + streamId, subgroup == null?"":subgroup);
V message = (V) tuple.getValueByField("message");
try {
Map<String, V> streamMessageMap = cache.get(key);
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java
index 0515d29..e4b34a0 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java
@@ -26,6 +26,7 @@ import backtype.storm.tuple.Values;
import org.apache.metron.common.bolt.ConfiguredBolt;
import org.apache.metron.common.bolt.ConfiguredEnrichmentBolt;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -53,7 +54,7 @@ public abstract class SplitBolt<T extends Cloneable> extends
@Override
public final void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declareStream("message", new Fields("key", "message"));
+ declarer.declareStream("message", new Fields("key", "message", "subgroup"));
for (String streamId : getStreamIds()) {
declarer.declareStream(streamId, new Fields("key", "message"));
}
@@ -64,15 +65,22 @@ public abstract class SplitBolt<T extends Cloneable> extends
public void emit(Tuple tuple, T message) {
if (message == null) return;
String key = getKey(tuple, message);
- Map<String, T> streamMessageMap = splitMessage(message);
+ Map<String, List<T>> streamMessageMap = splitMessage(message);
for (String streamId : streamMessageMap.keySet()) {
- T streamMessage = streamMessageMap.get(streamId);
- if (streamMessage == null) {
- streamMessage = getDefaultMessage(streamId);
+ List<T> streamMessages = streamMessageMap.get(streamId);
+ if(streamMessages != null) {
+ for(T streamMessage : streamMessages) {
+ if (streamMessage == null) {
+ streamMessage = getDefaultMessage(streamId);
+ }
+ collector.emit(streamId, new Values(key, streamMessage));
+ }
+ }
+ else {
+ throw new IllegalArgumentException("Enrichment must send some list of messages, not null.");
}
- collector.emit(streamId, new Values(key, streamMessage));
}
- collector.emit("message", tuple, new Values(key, message));
+ collector.emit("message", tuple, new Values(key, message, ""));
collector.ack(tuple);
emitOther(tuple, message);
}
@@ -90,7 +98,7 @@ public abstract class SplitBolt<T extends Cloneable> extends
public abstract T generateMessage(Tuple tuple);
- public abstract Map<String, T> splitMessage(T message);
+ public abstract Map<String, List<T>> splitMessage(T message);
public abstract void declareOther(OutputFieldsDeclarer declarer);
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java
index 2ea05e7..0029fa2 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java
@@ -21,7 +21,9 @@ import backtype.storm.task.TopologyContext;
import com.google.common.base.Joiner;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler;
import org.apache.metron.common.configuration.enrichment.threatintel.ThreatTriageConfig;
+import org.apache.metron.common.dsl.Context;
import org.apache.metron.common.dsl.FunctionResolver;
import org.apache.metron.common.dsl.StellarFunctions;
import org.apache.metron.common.dsl.functions.ConversionFunctions;
@@ -32,6 +34,7 @@ import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -39,20 +42,42 @@ public class ThreatIntelJoinBolt extends EnrichmentJoinBolt {
protected static final Logger LOG = LoggerFactory
.getLogger(ThreatIntelJoinBolt.class);
- FunctionResolver functionResolver;
+ private FunctionResolver functionResolver;
+ private org.apache.metron.common.dsl.Context stellarContext;
public ThreatIntelJoinBolt(String zookeeperUrl) {
super(zookeeperUrl);
}
@Override
+ protected Map<String, ConfigHandler> getFieldToHandlerMap(String sensorType) {
+ if(sensorType != null) {
+ SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sensorType);
+ if (config != null) {
+ return config.getThreatIntel().getEnrichmentConfigs();
+ } else {
+ LOG.error("Unable to retrieve a sensor enrichment config of " + sensorType);
+ }
+ } else {
+ LOG.error("Trying to retrieve a field map with sensor type of null");
+ }
+ return new HashMap<>();
+ }
+
+ @Override
public void prepare(Map map, TopologyContext topologyContext) {
super.prepare(map, topologyContext);
- functionResolver = StellarFunctions.FUNCTION_RESOLVER();
- functionResolver.initializeFunctions(context);
+ initializeStellar();
+ }
+
+ protected void initializeStellar() {
+ this.stellarContext = new Context.Builder()
+ .with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
+ .build();
+ StellarFunctions.FUNCTION_RESOLVER().initializeFunctions(stellarContext);
}
@Override
- public Map<String, List<String>> getFieldMap(String sourceType) {
+ public Map<String, Object> getFieldMap(String sourceType) {
SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sourceType);
if(config != null) {
return config.getThreatIntel().getFieldMap();
@@ -105,7 +130,7 @@ public class ThreatIntelJoinBolt extends EnrichmentJoinBolt {
LOG.debug(sourceType + ": Empty rules!");
}
- ThreatTriageProcessor threatTriageProcessor = new ThreatTriageProcessor(config, functionResolver, context);
+ ThreatTriageProcessor threatTriageProcessor = new ThreatTriageProcessor(config, functionResolver, stellarContext);
Double triageLevel = threatTriageProcessor.apply(ret);
if(LOG.isDebugEnabled()) {
String rules = Joiner.on('\n').join(triageConfig.getRiskLevelRules().entrySet());
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBolt.java
index f5b6399..b39434f 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBolt.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBolt.java
@@ -18,10 +18,10 @@
package org.apache.metron.enrichment.bolt;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler;
import org.apache.metron.enrichment.utils.ThreatIntelUtils;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
public class ThreatIntelSplitterBolt extends EnrichmentSplitterBolt {
@@ -31,7 +31,22 @@ public class ThreatIntelSplitterBolt extends EnrichmentSplitterBolt {
}
@Override
- protected Map<String, List<String>> getFieldMap(String sensorType) {
+ protected Map<String, ConfigHandler> getFieldToHandlerMap(String sensorType) {
+ if(sensorType != null) {
+ SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sensorType);
+ if (config != null) {
+ return config.getThreatIntel().getEnrichmentConfigs();
+ } else {
+ LOG.error("Unable to retrieve a sensor config of " + sensorType);
+ }
+ } else {
+ LOG.error("Trying to retrieve a field map with sensor type of null");
+ }
+ return new HashMap<>();
+ }
+
+ @Override
+ protected Map<String, Object> getFieldMap(String sensorType) {
if (sensorType != null) {
SensorEnrichmentConfig config = getConfigurations().getSensorEnrichmentConfig(sensorType);
if (config != null) {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/interfaces/EnrichmentAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/interfaces/EnrichmentAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/interfaces/EnrichmentAdapter.java
index 28f9956..c2682bc 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/interfaces/EnrichmentAdapter.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/interfaces/EnrichmentAdapter.java
@@ -18,6 +18,7 @@
package org.apache.metron.enrichment.interfaces;
+import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
import org.json.simple.JSONObject;
public interface EnrichmentAdapter<T>
@@ -26,4 +27,8 @@ public interface EnrichmentAdapter<T>
JSONObject enrich(T value);
boolean initializeAdapter();
void cleanup();
+ String getOutputPrefix(T value);
+ default String getStreamSubGroup(String enrichmentType, String field) {
+ return "";
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java
index d7b3629..56ddf08 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java
@@ -74,8 +74,7 @@ public class EnrichmentJoinBoltTest extends BaseEnrichmentBoltTest {
enrichmentJoinBolt.withMaxTimeRetain(10000);
enrichmentJoinBolt.prepare(new HashMap<>(), topologyContext, outputCollector);
Set<String> actualStreamIds = enrichmentJoinBolt.getStreamIds(sampleMessage);
- streamIds.add("message");
- Assert.assertEquals(streamIds, actualStreamIds);
+ Assert.assertEquals(joinStreamIds, actualStreamIds);
Map<String, JSONObject> streamMessageMap = new HashMap<>();
streamMessageMap.put("message", sampleMessage);
streamMessageMap.put("enriched", enrichedMessage);
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBoltTest.java
index 67b12b9..d5b6e6a 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentSplitterBoltTest.java
@@ -17,6 +17,7 @@
*/
package org.apache.metron.enrichment.bolt;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.apache.metron.test.bolt.BaseEnrichmentBoltTest;
import org.apache.metron.enrichment.configuration.Enrichment;
@@ -46,10 +47,13 @@ public class EnrichmentSplitterBoltTest extends BaseEnrichmentBoltTest {
host.setType("host");
final Enrichment hbaseEnrichment = new Enrichment();
hbaseEnrichment.setType("hbaseEnrichment");
+ final Enrichment stellarEnrichment = new Enrichment();
+ stellarEnrichment.setType("stellar");
List<Enrichment> enrichments = new ArrayList<Enrichment>() {{
add(geo);
add(host);
add(hbaseEnrichment);
+ add(stellarEnrichment);
}};
EnrichmentSplitterBolt enrichmentSplitterBolt = new EnrichmentSplitterBolt("zookeeperUrl").withEnrichments(enrichments);
@@ -76,11 +80,11 @@ public class EnrichmentSplitterBoltTest extends BaseEnrichmentBoltTest {
Set<String> actualStreamIds = enrichmentSplitterBolt.getStreamIds();
Assert.assertEquals(streamIds, actualStreamIds);
- Map<String, JSONObject> actualSplitMessages = enrichmentSplitterBolt.splitMessage(sampleMessage);
- Assert.assertEquals(3, actualSplitMessages.size());
- Assert.assertEquals(geoMessage, actualSplitMessages.get("geo"));
- Assert.assertEquals(hostMessage, actualSplitMessages.get("host"));
- Assert.assertEquals(hbaseEnrichmentMessage, actualSplitMessages.get("hbaseEnrichment"));
+ Map<String, List<JSONObject> > actualSplitMessages = enrichmentSplitterBolt.splitMessage(sampleMessage);
+ Assert.assertEquals(enrichments.size(), actualSplitMessages.size());
+ Assert.assertEquals(ImmutableList.of(geoMessage), actualSplitMessages.get("geo"));
+ Assert.assertEquals(ImmutableList.of(hostMessage), actualSplitMessages.get("host"));
+ Assert.assertEquals(ImmutableList.of(hbaseEnrichmentMessage), actualSplitMessages.get("hbaseEnrichment"));
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
index 31168ff..3b4fed0 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
@@ -21,6 +21,7 @@ import backtype.storm.tuple.Values;
import com.google.common.collect.ImmutableMap;
import org.adrianwalker.multilinestring.Multiline;
import org.apache.metron.TestConstants;
+import org.apache.metron.common.dsl.StellarFunctions;
import org.apache.metron.test.bolt.BaseEnrichmentBoltTest;
import org.apache.metron.enrichment.configuration.Enrichment;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
@@ -44,10 +45,7 @@ import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.argThat;
import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
public class GenericEnrichmentBoltTest extends BaseEnrichmentBoltTest {
@@ -142,7 +140,12 @@ public class GenericEnrichmentBoltTest extends BaseEnrichmentBoltTest {
Enrichment<EnrichmentAdapter<CacheKey>> testEnrichment = new Enrichment<>();
testEnrichment.setType(enrichmentType);
testEnrichment.setAdapter(enrichmentAdapter);
- GenericEnrichmentBolt genericEnrichmentBolt = new GenericEnrichmentBolt("zookeeperUrl");
+ GenericEnrichmentBolt genericEnrichmentBolt = new GenericEnrichmentBolt("zookeeperUrl") {
+ @Override
+ protected void initializeStellar() {
+ //do not initialize stellar here.
+ }
+ };
genericEnrichmentBolt.setCuratorFramework(client);
genericEnrichmentBolt.setTreeCache(cache);
genericEnrichmentBolt.getConfigurations().updateSensorEnrichmentConfig(sensorType, new FileInputStream(sampleSensorEnrichmentConfigPath));
@@ -170,21 +173,26 @@ public class GenericEnrichmentBoltTest extends BaseEnrichmentBoltTest {
fail("An exception should be thrown if enrichment adapter initialization fails");
} catch(IllegalStateException e) {}
genericEnrichmentBolt.declareOutputFields(declarer);
- verify(declarer, times(1)).declareStream(eq(enrichmentType), argThat(new FieldsMatcher("key", "message")));
+ verify(declarer, times(1)).declareStream(eq(enrichmentType), argThat(new FieldsMatcher("key", "message", "subgroup")));
verify(declarer, times(1)).declareStream(eq("error"), argThat(new FieldsMatcher("message")));
when(tuple.getStringByField("key")).thenReturn(null);
genericEnrichmentBolt.execute(tuple);
verify(outputCollector, times(1)).emit(eq("error"), any(Values.class));
when(tuple.getStringByField("key")).thenReturn(key);
when(tuple.getValueByField("message")).thenReturn(originalMessage);
+ when(enrichmentAdapter.enrich(any())).thenReturn(new JSONObject());
genericEnrichmentBolt.execute(tuple);
verify(outputCollector, times(1)).emit(eq(enrichmentType), argThat(new EnrichedMessageMatcher(key, new JSONObject(ImmutableMap.of("source.type", "test")))));
reset(enrichmentAdapter);
SensorEnrichmentConfig sensorEnrichmentConfig = SensorEnrichmentConfig.
fromBytes(ConfigurationsUtils.readSensorEnrichmentConfigsFromFile(TestConstants.SAMPLE_CONFIG_PATH).get(sensorType));
+ sensorEnrichmentConfig.getConfiguration().put(GenericEnrichmentBolt.STELLAR_CONTEXT_CONF, genericEnrichmentBolt.getStellarContext());
CacheKey cacheKey1 = new CacheKey("field1", "value1", sensorEnrichmentConfig);
CacheKey cacheKey2 = new CacheKey("field2", "value2", sensorEnrichmentConfig);
+ genericEnrichmentBolt.cache.invalidateAll();
+ when(enrichmentAdapter.getOutputPrefix(cacheKey1)).thenReturn("field1");
+ when(enrichmentAdapter.getOutputPrefix(cacheKey2)).thenReturn("field2");
when(enrichmentAdapter.enrich(cacheKey1)).thenReturn(enrichedField1);
when(enrichmentAdapter.enrich(cacheKey2)).thenReturn(enrichedField2);
genericEnrichmentBolt.execute(tuple);
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
index 90ec6ad..931297d 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
@@ -28,6 +28,7 @@ import org.junit.Before;
import org.junit.Test;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -54,7 +55,11 @@ public class JoinBoltTest extends BaseEnrichmentBoltTest {
@Override
public Set<String> getStreamIds(JSONObject value) {
- return streamIds;
+ HashSet<String> ret = new HashSet<>();
+ for(String s : streamIds) {
+ ret.add(s + ":");
+ }
+ return ret;
}
@Override
@@ -115,6 +120,10 @@ public class JoinBoltTest extends BaseEnrichmentBoltTest {
when(tuple.getSourceStreamId()).thenReturn("hbaseEnrichment");
when(tuple.getValueByField("message")).thenReturn(hbaseEnrichmentMessage);
joinBolt.execute(tuple);
+ when(tuple.getSourceStreamId()).thenReturn("stellar");
+ when(tuple.getValueByField("message")).thenReturn(new JSONObject());
+ verify(outputCollector, times(0)).emit(eq("message"), any(tuple.getClass()), eq(new Values(key, joinedMessage)));
+ joinBolt.execute(tuple);
verify(outputCollector, times(1)).emit(eq("message"), any(tuple.getClass()), eq(new Values(key, joinedMessage)));
verify(outputCollector, times(1)).ack(tuple);
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/SplitBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/SplitBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/SplitBoltTest.java
index fc18ad0..6653b6d 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/SplitBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/SplitBoltTest.java
@@ -21,6 +21,7 @@ import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
+import com.google.common.collect.ImmutableList;
import junit.framework.Assert;
import org.apache.metron.common.configuration.ConfigurationType;
import org.apache.metron.test.bolt.BaseEnrichmentBoltTest;
@@ -28,6 +29,7 @@ import org.json.simple.JSONObject;
import org.junit.Test;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -67,7 +69,7 @@ public class SplitBoltTest extends BaseEnrichmentBoltTest {
}
@Override
- public Map<String, JSONObject> splitMessage(JSONObject message) {
+ public Map<String, List<JSONObject>> splitMessage(JSONObject message) {
return null;
}
@@ -90,20 +92,20 @@ public class SplitBoltTest extends BaseEnrichmentBoltTest {
doCallRealMethod().when(splitBolt).reloadCallback(anyString(), any(ConfigurationType.class));
splitBolt.prepare(new HashMap(), topologyContext, outputCollector);
splitBolt.declareOutputFields(declarer);
- verify(declarer, times(1)).declareStream(eq("message"), argThat(new FieldsMatcher("key", "message")));
+ verify(declarer, times(1)).declareStream(eq("message"), argThat(new FieldsMatcher("key", "message", "subgroup")));
for(String streamId: streamIds) {
verify(declarer, times(1)).declareStream(eq(streamId), argThat(new FieldsMatcher("key", "message")));
}
verify(declarer, times(1)).declareStream(eq("error"), argThat(new FieldsMatcher("message")));
JSONObject sampleMessage = splitBolt.generateMessage(tuple);
- Map<String, JSONObject> streamMessageMap = new HashMap<>();
- streamMessageMap.put("geo", geoMessage);
- streamMessageMap.put("host", hostMessage);
- streamMessageMap.put("hbaseEnrichment", hbaseEnrichmentMessage);
+ Map<String, List<JSONObject>> streamMessageMap = new HashMap<>();
+ streamMessageMap.put("geo", ImmutableList.of(geoMessage));
+ streamMessageMap.put("host", ImmutableList.of(hostMessage));
+ streamMessageMap.put("hbaseEnrichment", ImmutableList.of(hbaseEnrichmentMessage));
doReturn(streamMessageMap).when(splitBolt).splitMessage(sampleMessage);
splitBolt.execute(tuple);
- verify(outputCollector, times(1)).emit(eq("message"), any(tuple.getClass()), eq(new Values(key, sampleMessage)));
+ verify(outputCollector, times(1)).emit(eq("message"), any(tuple.getClass()), eq(new Values(key, sampleMessage, "")));
verify(outputCollector, times(1)).emit(eq("geo"), eq(new Values(key, geoMessage)));
verify(outputCollector, times(1)).emit(eq("host"), eq(new Values(key, hostMessage)));
verify(outputCollector, times(1)).emit(eq("hbaseEnrichment"), eq(new Values(key, hbaseEnrichmentMessage)));
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java
index b109960..b3004bb 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java
@@ -116,7 +116,7 @@ public class ThreatIntelJoinBoltTest extends BaseEnrichmentBoltTest {
threatIntelJoinBolt.withMaxCacheSize(100);
threatIntelJoinBolt.withMaxTimeRetain(10000);
threatIntelJoinBolt.prepare(new HashMap<>(), topologyContext, outputCollector);
- Map<String, List<String>> fieldMap = threatIntelJoinBolt.getFieldMap("incorrectSourceType");
+ Map<String, Object> fieldMap = threatIntelJoinBolt.getFieldMap("incorrectSourceType");
Assert.assertNull(fieldMap);
fieldMap = threatIntelJoinBolt.getFieldMap(sensorType);
Assert.assertTrue(fieldMap.containsKey("hbaseThreatIntel"));
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBoltTest.java
index 77ed32f..6898c89 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelSplitterBoltTest.java
@@ -37,7 +37,7 @@ public class ThreatIntelSplitterBoltTest extends BaseEnrichmentBoltTest {
threatIntelSplitterBolt.setTreeCache(cache);
threatIntelSplitterBolt.getConfigurations().updateSensorEnrichmentConfig(sensorType, new FileInputStream(sampleSensorEnrichmentConfigPath));
threatIntelSplitterBolt.prepare(new HashMap<>(), topologyContext, outputCollector);
- Map<String, List<String>> fieldMap = threatIntelSplitterBolt.getFieldMap(sensorType);
+ Map<String, Object> fieldMap = threatIntelSplitterBolt.getFieldMap(sensorType);
Assert.assertTrue(fieldMap.containsKey(threatIntelType));
String fieldName = threatIntelSplitterBolt.getKeyName(threatIntelType, "field");
Assert.assertEquals("threatintels.hbaseThreatIntel.field", fieldName);
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/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 6b12264..35fe7ce 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
@@ -184,6 +184,11 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
//ensure we always have a source ip and destination ip
Assert.assertNotNull(jsonDoc.get(SRC_IP));
Assert.assertNotNull(jsonDoc.get(DST_IP));
+ Assert.assertNotNull(jsonDoc.get("ALL_CAPS"));
+ Assert.assertNotNull(jsonDoc.get("foo"));
+ Assert.assertEquals("TEST", jsonDoc.get("ALL_CAPS"));
+ Assert.assertNotNull(jsonDoc.get("bar"));
+ Assert.assertEquals("TEST", jsonDoc.get("bar"));
}
private static class EvaluationPayload {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/mock/MockGeoAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/mock/MockGeoAdapter.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/mock/MockGeoAdapter.java
index 14a4493..a724bcd 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/mock/MockGeoAdapter.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/mock/MockGeoAdapter.java
@@ -61,4 +61,9 @@ public class MockGeoAdapter implements EnrichmentAdapter<CacheKey>,
public void cleanup() {
}
+
+ @Override
+ public String getOutputPrefix(CacheKey value) {
+ return value.getField();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/resources/log4j.properties b/metron-platform/metron-enrichment/src/test/resources/log4j.properties
new file mode 100644
index 0000000..ae15e0a
--- /dev/null
+++ b/metron-platform/metron-enrichment/src/test/resources/log4j.properties
@@ -0,0 +1,19 @@
+# Licensed 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.
+
+# log4j configuration used during build and unit tests
+
+log4j.rootLogger=error,stdout
+log4j.threshhold=ERROR
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} %-5p [%t] %c{2} (%F:%M(%L)) - %m%n
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/test/resources/log4j2.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/resources/log4j2.xml b/metron-platform/metron-enrichment/src/test/resources/log4j2.xml
new file mode 100755
index 0000000..68d5eac
--- /dev/null
+++ b/metron-platform/metron-enrichment/src/test/resources/log4j2.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+
+<configuration monitorInterval="60">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%-4r [%t] %-5p %c{1.} - %msg%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="error">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</configuration>
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java
index 2545266..d19633e 100644
--- a/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java
+++ b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java
@@ -70,7 +70,6 @@ public class HBaseClientTest {
Configuration config = HBaseConfiguration.create();
config.set("hbase.master.hostname", "localhost");
config.set("hbase.regionserver.hostname", "localhost");
-
util = new HBaseTestingUtility(config);
util.startMiniCluster();
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-integration-test/src/main/config/zookeeper/enrichments/test.json
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/config/zookeeper/enrichments/test.json b/metron-platform/metron-integration-test/src/main/config/zookeeper/enrichments/test.json
index d72851e..5c7adf6 100644
--- a/metron-platform/metron-integration-test/src/main/config/zookeeper/enrichments/test.json
+++ b/metron-platform/metron-integration-test/src/main/config/zookeeper/enrichments/test.json
@@ -14,7 +14,15 @@
"hbaseEnrichment": [
"ip_src_addr",
"ip_dst_addr"
- ]
+ ],
+ "stellar" : {
+ "config" : {
+ "numeric" : {
+ "foo": "1 + 1"
+ }
+ ,"ALL_CAPS" : "TO_UPPER(source.type)"
+ }
+ }
}
,"fieldToTypeMap": {
"ip_src_addr": [
@@ -30,7 +38,12 @@
"hbaseThreatIntel": [
"ip_src_addr",
"ip_dst_addr"
- ]
+ ],
+ "stellar" : {
+ "config" : {
+ "bar" : "TO_UPPER(source.type)"
+ }
+ }
},
"fieldToTypeMap": {
"ip_src_addr": [
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/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 746ec18..2b457a7 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
@@ -27,6 +27,7 @@ import org.apache.metron.common.Constants;
import org.apache.metron.common.bolt.ConfiguredParserBolt;
import org.apache.metron.common.configuration.FieldValidator;
import org.apache.metron.common.configuration.SensorParserConfig;
+import org.apache.metron.common.dsl.Context;
import org.apache.metron.common.dsl.FunctionResolver;
import org.apache.metron.common.dsl.StellarFunctions;
import org.apache.metron.parsers.filters.Filters;
@@ -49,6 +50,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
private MessageParser<JSONObject> parser;
private MessageFilter<JSONObject> filter = new GenericMessageFilter();
private WriterHandler writer;
+ private org.apache.metron.common.dsl.Context stellarContext;
public ParserBolt( String zookeeperUrl
, String sensorType
, MessageParser<JSONObject> parser
@@ -91,9 +93,15 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
throw new IllegalStateException("Unable to retrieve a parser config for " + getSensorType());
}
parser.configure(config.getParserConfig());
- StellarFunctions.FUNCTION_RESOLVER().initializeFunctions(super.context);
+ initializeStellar();
}
+ protected void initializeStellar() {
+ this.stellarContext = new Context.Builder()
+ .with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
+ .build();
+ StellarFunctions.FUNCTION_RESOLVER().initializeFunctions(stellarContext);
+ }
@SuppressWarnings("unchecked")
@Override
@@ -109,11 +117,11 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
List<FieldValidator> fieldValidations = getConfigurations().getFieldValidations();
Optional<List<JSONObject>> messages = parser.parseOptional(originalMessage);
for (JSONObject message : messages.orElse(Collections.emptyList())) {
- if (parser.validate(message) && filter != null && filter.emitTuple(message, super.context)) {
+ if (parser.validate(message) && filter != null && filter.emitTuple(message, stellarContext)) {
message.put(Constants.SENSOR_TYPE, getSensorType());
for (FieldTransformer handler : sensorParserConfig.getFieldTransformations()) {
if (handler != null) {
- handler.transformAndUpdate(message, sensorParserConfig.getParserConfig(), super.context);
+ handler.transformAndUpdate(message, sensorParserConfig.getParserConfig(), stellarContext);
}
}
numWritten++;
@@ -146,7 +154,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
private boolean isGloballyValid(JSONObject input, List<FieldValidator> validators) {
for(FieldValidator validator : validators) {
- if(!validator.isValid(input, getConfigurations().getGlobalConfig(), super.context)) {
+ if(!validator.isValid(input, getConfigurations().getGlobalConfig(), stellarContext)) {
return false;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/bolt/BaseEnrichmentBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/bolt/BaseEnrichmentBoltTest.java b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/bolt/BaseEnrichmentBoltTest.java
index 69c2174..de08cf7 100644
--- a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/bolt/BaseEnrichmentBoltTest.java
+++ b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/bolt/BaseEnrichmentBoltTest.java
@@ -31,6 +31,7 @@ public class BaseEnrichmentBoltTest extends BaseBoltTest {
public String sampleSensorEnrichmentConfigPath = TestConstants.SAMPLE_CONFIG_PATH + "enrichments/test.json";
protected Set<String> streamIds = new HashSet<>();
+ protected Set<String> joinStreamIds = new HashSet<>();
protected String key = "someKey";
protected String sensorType = "test";
@@ -87,7 +88,14 @@ public class BaseEnrichmentBoltTest extends BaseBoltTest {
hostMessage = (JSONObject) parser.parse(hostMessageString);
hbaseEnrichmentMessage = (JSONObject) parser.parse(hbaseEnrichmentMessageString);
streamIds.add("geo");
+ streamIds.add("stellar");
streamIds.add("host");
streamIds.add("hbaseEnrichment");
+ joinStreamIds.add("geo:");
+ joinStreamIds.add("stellar:");
+ joinStreamIds.add("stellar:numeric");
+ joinStreamIds.add("host:");
+ joinStreamIds.add("hbaseEnrichment:");
+ joinStreamIds.add("message:");
}
}
[2/2] incubator-metron git commit: METRON-341: Create stellar
enrichment adapter closes apache/incubator-metron#216
Posted by ce...@apache.org.
METRON-341: Create stellar enrichment adapter closes apache/incubator-metron#216
Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/58294b66
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/58294b66
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/58294b66
Branch: refs/heads/master
Commit: 58294b6659856b4e5d1feecc2b670251a15bc09e
Parents: f5c6d00
Author: cstella <ce...@gmail.com>
Authored: Tue Aug 23 11:41:07 2016 -0400
Committer: cstella <ce...@gmail.com>
Committed: Tue Aug 23 11:41:07 2016 -0400
----------------------------------------------------------------------
.../src/test/resources/log4j2.xml | 31 +++
.../profiler/bolt/ProfileBuilderBolt.java | 23 +-
.../profiler/bolt/ProfileSplitterBolt.java | 21 +-
.../stellar/DefaultStellarExecutor.java | 15 +-
.../profiler/stellar/StellarExecutor.java | 7 +-
.../util/DefaultStellarExecutorTest.java | 27 +--
metron-platform/metron-common/README.md | 56 ++++-
.../metron/common/bolt/ConfiguredBolt.java | 7 +-
.../enrichment/EnrichmentConfig.java | 24 +-
.../SensorEnrichmentUpdateConfig.java | 10 +-
.../enrichment/handler/Config.java | 36 +++
.../enrichment/handler/ConfigHandler.java | 76 +++++++
.../enrichment/handler/Configs.java | 51 +++++
.../enrichment/handler/ListConfig.java | 51 +++++
.../enrichment/handler/StellarConfig.java | 94 ++++++++
.../metron/common/dsl/MapVariableResolver.java | 4 +-
.../common/stellar/BaseStellarProcessor.java | 35 +++
.../bolt/ConfiguredEnrichmentBoltTest.java | 4 +-
.../configuration/EnrichmentConfigTest.java | 59 +++++
.../SensorEnrichmentUpdateConfigTest.java | 41 ++--
.../metron/common/stellar/StellarTest.java | 18 ++
.../src/main/flux/enrichment/remote.yaml | 93 +++++++-
.../src/main/flux/enrichment/test.yaml | 86 ++++++-
.../adapters/cif/CIFHbaseAdapter.java | 7 +-
.../enrichment/adapters/geo/GeoAdapter.java | 8 +-
.../adapters/host/HostFromJSONListAdapter.java | 7 +-
.../host/HostFromPropertiesFileAdapter.java | 5 +
.../simplehbase/SimpleHBaseAdapter.java | 7 +-
.../adapters/stellar/StellarAdapter.java | 110 +++++++++
.../threatintel/ThreatIntelAdapter.java | 9 +-
.../apache/metron/enrichment/bolt/CacheKey.java | 14 +-
.../enrichment/bolt/EnrichmentJoinBolt.java | 36 ++-
.../enrichment/bolt/EnrichmentSplitterBolt.java | 223 ++++++++++---------
.../enrichment/bolt/GenericEnrichmentBolt.java | 44 +++-
.../apache/metron/enrichment/bolt/JoinBolt.java | 2 +
.../metron/enrichment/bolt/SplitBolt.java | 24 +-
.../enrichment/bolt/ThreatIntelJoinBolt.java | 35 ++-
.../bolt/ThreatIntelSplitterBolt.java | 19 +-
.../interfaces/EnrichmentAdapter.java | 5 +
.../enrichment/bolt/EnrichmentJoinBoltTest.java | 3 +-
.../bolt/EnrichmentSplitterBoltTest.java | 14 +-
.../bolt/GenericEnrichmentBoltTest.java | 20 +-
.../metron/enrichment/bolt/JoinBoltTest.java | 11 +-
.../metron/enrichment/bolt/SplitBoltTest.java | 16 +-
.../bolt/ThreatIntelJoinBoltTest.java | 2 +-
.../bolt/ThreatIntelSplitterBoltTest.java | 2 +-
.../integration/EnrichmentIntegrationTest.java | 5 +
.../integration/mock/MockGeoAdapter.java | 5 +
.../src/test/resources/log4j.properties | 19 ++
.../src/test/resources/log4j2.xml | 31 +++
.../metron/hbase/client/HBaseClientTest.java | 1 -
.../main/config/zookeeper/enrichments/test.json | 17 +-
.../apache/metron/parsers/bolt/ParserBolt.java | 16 +-
.../test/bolt/BaseEnrichmentBoltTest.java | 8 +
54 files changed, 1350 insertions(+), 244 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-analytics/metron-maas-service/src/test/resources/log4j2.xml
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/src/test/resources/log4j2.xml b/metron-analytics/metron-maas-service/src/test/resources/log4j2.xml
new file mode 100755
index 0000000..68d5eac
--- /dev/null
+++ b/metron-analytics/metron-maas-service/src/test/resources/log4j2.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+
+<configuration monitorInterval="60">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%-4r [%t] %-5p %c{1.} - %msg%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="error">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</configuration>
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java
index fa25149..f5cfce7 100644
--- a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java
@@ -30,7 +30,9 @@ import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import org.apache.metron.common.bolt.ConfiguredProfilerBolt;
import org.apache.metron.common.configuration.profiler.ProfileConfig;
+import org.apache.metron.common.dsl.Context;
import org.apache.metron.common.dsl.ParseException;
+import org.apache.metron.common.dsl.StellarFunctions;
import org.apache.metron.profiler.ProfileMeasurement;
import org.apache.metron.profiler.stellar.StellarExecutor;
import org.json.simple.JSONObject;
@@ -83,6 +85,11 @@ public class ProfileBuilderBolt extends ConfiguredProfilerBolt {
*/
private transient JSONParser parser;
+ /**
+ * Stellar context
+ */
+ private Context stellarContext;
+
private OutputCollector collector;
/**
@@ -103,11 +110,19 @@ public class ProfileBuilderBolt extends ConfiguredProfilerBolt {
return conf;
}
+ protected void initializeStellar() {
+ stellarContext = new Context.Builder()
+ .with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
+ .build();
+ StellarFunctions.FUNCTION_RESOLVER().initializeFunctions(stellarContext);
+ }
+
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
super.prepare(stormConf, context, collector);
this.collector = collector;
this.parser = new JSONParser();
+ initializeStellar();
}
/**
@@ -181,7 +196,7 @@ public class ProfileBuilderBolt extends ConfiguredProfilerBolt {
try {
JSONObject message = (JSONObject) input.getValueByField("message");
Map<String, String> expressions = profileConfig.getInit();
- expressions.forEach((var, expr) -> executor.assign(var, expr, message));
+ expressions.forEach((var, expr) -> executor.assign(var, expr, message, stellarContext));
} catch(ParseException e) {
String msg = format("Bad 'init' expression: %s, profile=%s, entity=%s, start=%d",
@@ -200,7 +215,7 @@ public class ProfileBuilderBolt extends ConfiguredProfilerBolt {
// execute each of the 'update' expressions
try {
Map<String, String> expressions = profileConfig.getUpdate();
- expressions.forEach((var, expr) -> executor.assign(var, expr, message));
+ expressions.forEach((var, expr) -> executor.assign(var, expr, message, stellarContext));
} catch(ParseException e) {
String msg = format("Bad 'update' expression: %s, profile=%s, entity=%s, start=%d",
@@ -224,8 +239,8 @@ public class ProfileBuilderBolt extends ConfiguredProfilerBolt {
Object result;
try {
String resultExpr = profileConfig.getResult();
- result = executor.execute(resultExpr, new JSONObject(), Object.class);
-
+ result = executor.execute(resultExpr, new JSONObject(), Object.class, stellarContext);
+
} catch(ParseException e) {
throw new ParseException("Bad 'result' expression", e);
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java
index 5804068..3e4c4ec 100644
--- a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java
@@ -30,12 +30,15 @@ import org.apache.commons.lang.StringUtils;
import org.apache.metron.common.bolt.ConfiguredProfilerBolt;
import org.apache.metron.common.configuration.profiler.ProfileConfig;
import org.apache.metron.common.configuration.profiler.ProfilerConfig;
+import org.apache.metron.common.dsl.Context;
+import org.apache.metron.common.dsl.StellarFunctions;
import org.apache.metron.profiler.stellar.StellarExecutor;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import scala.reflect.generic.Types;
import java.io.UnsupportedEncodingException;
import java.util.Map;
@@ -64,6 +67,12 @@ public class ProfileSplitterBolt extends ConfiguredProfilerBolt {
private StellarExecutor executor;
/**
+ * Stellar context
+ */
+ private Context stellarContext;
+
+
+ /**
* @param zookeeperUrl The Zookeeper URL that contains the configuration for this bolt.
*/
public ProfileSplitterBolt(String zookeeperUrl) {
@@ -75,6 +84,14 @@ public class ProfileSplitterBolt extends ConfiguredProfilerBolt {
super.prepare(stormConf, context, collector);
this.collector = collector;
this.parser = new JSONParser();
+ initializeStellar();
+ }
+
+ protected void initializeStellar() {
+ stellarContext = new Context.Builder()
+ .with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
+ .build();
+ StellarFunctions.FUNCTION_RESOLVER().initializeFunctions(stellarContext);
}
@Override
@@ -119,10 +136,10 @@ public class ProfileSplitterBolt extends ConfiguredProfilerBolt {
// is this message needed by this profile?
String onlyIf = profile.getOnlyif();
- if (StringUtils.isBlank(onlyIf) || executor.execute(onlyIf, message, Boolean.class)) {
+ if (StringUtils.isBlank(onlyIf) || executor.execute(onlyIf, message, Boolean.class, stellarContext)) {
// what is the name of the entity in this message?
- String entity = executor.execute(profile.getForeach(), message, String.class);
+ String entity = executor.execute(profile.getForeach(), message, String.class, stellarContext);
// emit a message for the bolt responsible for building this profile
collector.emit(input, new Values(entity, profile, message));
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java
index a45e4a7..7eaf38a 100644
--- a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java
@@ -65,10 +65,11 @@ public class DefaultStellarExecutor implements StellarExecutor, Serializable {
* @param variable The variable name to assign to.
* @param expression The expression to execute.
* @param message The message that provides additional context for the expression.
+ * @param stellarContext The context which holds global state for Stellar functions
*/
@Override
- public void assign(String variable, String expression, JSONObject message) {
- Object result = execute(expression, message);
+ public void assign(String variable, String expression, JSONObject message, Context stellarContext) {
+ Object result = execute(expression, message, stellarContext);
state.put(variable, result);
}
@@ -79,10 +80,11 @@ public class DefaultStellarExecutor implements StellarExecutor, Serializable {
* @param message The message that is accessible when Stellar is executed.
* @param clazz The expected class of the expression's result.
* @param <T> The expected class of the expression's result.
+ * @param stellarContext The context which holds global state for Stellar functions
*/
@Override
- public <T> T execute(String expr, JSONObject message, Class<T> clazz) {
- Object resultObject = execute(expr, message);
+ public <T> T execute(String expr, JSONObject message, Class<T> clazz, Context stellarContext) {
+ Object resultObject = execute(expr, message, stellarContext);
// perform type conversion, if necessary
T result = ConversionUtils.convert(resultObject, clazz);
@@ -104,12 +106,13 @@ public class DefaultStellarExecutor implements StellarExecutor, Serializable {
*
* @param expr The expression to execute.
* @param msg The message that is accessible when Stellar is executed.
+ * @param stellarContext The context which holds global state for Stellar functions
*/
- private Object execute(String expr, JSONObject msg) {
+ private Object execute(String expr, JSONObject msg, Context stellarContext) {
try {
VariableResolver resolver = new MapVariableResolver(state, msg);
StellarProcessor processor = new StellarProcessor();
- return processor.parse(expr, resolver, StellarFunctions.FUNCTION_RESOLVER(), Context.EMPTY_CONTEXT());
+ return processor.parse(expr, resolver, StellarFunctions.FUNCTION_RESOLVER(), stellarContext);
} catch (ParseException e) {
throw new ParseException(String.format("Bad expression: expr=%s, msg=%s, state=%s", expr, msg, state));
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java
index 9005e00..8d6660c 100644
--- a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java
@@ -20,6 +20,7 @@
package org.apache.metron.profiler.stellar;
+import org.apache.metron.common.dsl.Context;
import org.json.simple.JSONObject;
import java.util.Map;
@@ -39,8 +40,9 @@ public interface StellarExecutor {
* @param variable The name of the variable to assign to.
* @param expression The expression to execute.
* @param message The message that provides additional context for the expression.
+ * @param stellarContext The context which holds global state for Stellar functions
*/
- void assign(String variable, String expression, JSONObject message);
+ void assign(String variable, String expression, JSONObject message, Context stellarContext);
/**
* Execute a Stellar expression and return the result.
@@ -49,8 +51,9 @@ public interface StellarExecutor {
* @param message The message that is accessible when Stellar is executed.
* @param clazz The expected class of the expression's result.
* @param <T> The expected class of the expression's result.
+ * @param stellarContext The context which holds global state for Stellar functions
*/
- <T> T execute(String expression, JSONObject message, Class<T> clazz);
+ <T> T execute(String expression, JSONObject message, Class<T> clazz, Context stellarContext);
/**
* The current state of the Stellar execution environment.
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java
index 534c724..81df399 100644
--- a/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java
+++ b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java
@@ -21,6 +21,7 @@
package org.apache.metron.profiler.util;
import org.adrianwalker.multilinestring.Multiline;
+import org.apache.metron.common.dsl.Context;
import org.apache.metron.profiler.stellar.DefaultStellarExecutor;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
@@ -66,7 +67,7 @@ public class DefaultStellarExecutorTest {
*/
@Test
public void testAssign() {
- executor.assign("foo", "2", message);
+ executor.assign("foo", "2", message, Context.EMPTY_CONTEXT());
// verify
Object var = executor.getState().get("foo");
@@ -79,7 +80,7 @@ public class DefaultStellarExecutorTest {
*/
@Test
public void testAssignWithVariableResolution() {
- executor.assign("foo", "ip_src_addr", message);
+ executor.assign("foo", "ip_src_addr", message, Context.EMPTY_CONTEXT());
// verify
Object var = executor.getState().get("foo");
@@ -92,9 +93,9 @@ public class DefaultStellarExecutorTest {
*/
@Test
public void testState() {
- executor.assign("two", "2", message);
- executor.assign("four", "4", message);
- executor.assign("sum", "two + four", message);
+ executor.assign("two", "2", message, Context.EMPTY_CONTEXT());
+ executor.assign("four", "4", message, Context.EMPTY_CONTEXT());
+ executor.assign("sum", "two + four", message, Context.EMPTY_CONTEXT());
// verify
Object var = executor.getState().get("sum");
@@ -106,7 +107,7 @@ public class DefaultStellarExecutorTest {
*/
@Test
public void testClearState() {
- executor.assign("two", "2", message);
+ executor.assign("two", "2", message, Context.EMPTY_CONTEXT());
executor.clearState();
// verify
@@ -122,7 +123,7 @@ public class DefaultStellarExecutorTest {
*/
@Test
public void testExecuteTransformation() {
- String actual = executor.execute("TO_UPPER('lowercase')", message, String.class);
+ String actual = executor.execute("TO_UPPER('lowercase')", message, String.class, Context.EMPTY_CONTEXT());
assertThat(actual, equalTo("LOWERCASE"));
}
@@ -135,7 +136,7 @@ public class DefaultStellarExecutorTest {
*/
@Test
public void testExecutePredicate() {
- boolean actual = executor.execute("IS_INTEGER(2)", message, Boolean.class);
+ boolean actual = executor.execute("IS_INTEGER(2)", message, Boolean.class, Context.EMPTY_CONTEXT());
assertThat(actual, equalTo(true));
}
@@ -144,7 +145,7 @@ public class DefaultStellarExecutorTest {
*/
@Test(expected = RuntimeException.class)
public void testExecuteWithWrongType() {
- executor.execute("2 + 2", message, Boolean.class);
+ executor.execute("2 + 2", message, Boolean.class, Context.EMPTY_CONTEXT());
}
/**
@@ -152,9 +153,9 @@ public class DefaultStellarExecutorTest {
*/
@Test
public void testExecuteWithTypeConversion() {
- executor.execute("2", message, Double.class);
- executor.execute("2", message, Float.class);
- executor.execute("2", message, Short.class);
- executor.execute("2", message, Long.class);
+ executor.execute("2", message, Double.class, Context.EMPTY_CONTEXT());
+ executor.execute("2", message, Float.class, Context.EMPTY_CONTEXT());
+ executor.execute("2", message, Short.class, Context.EMPTY_CONTEXT());
+ executor.execute("2", message, Long.class, Context.EMPTY_CONTEXT());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/README.md b/metron-platform/metron-common/README.md
index 2137469..2d3350e 100644
--- a/metron-platform/metron-common/README.md
+++ b/metron-platform/metron-common/README.md
@@ -151,13 +151,65 @@ The configuration is a complex JSON object with the following top level fields:
| Field | Description | Example |
|------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------|
| `fieldToTypeMap` | In the case of a simple HBase enrichment (i.e. a key/value lookup), the mapping between fields and the enrichment types associated with those fields must be known. This enrichment type is used as part of the HBase key. | `"fieldToTypeMap" : { "ip_src_addr" : [ "asset_enrichment" ] }` |
-| `fieldMap` | The map of enrichment bolts names to fields in the JSON messages.,Each field is sent to the enrichment referenced in the key. | `"fieldMap": {"hbaseEnrichment": ["ip_src_addr","ip_dst_addr"]}` |
-| `config` | The general configuration for the enrichment | `"config": {"typeToColumnFamily": { "asset_enrichment","cf" } }` |
+| `fieldMap` | The map of enrichment bolts names to configuration handlers which know how to split the message up. The simplest of which is just a list of fields. More complex examples would be the stellar enrichment which provides stellar statements. Each field is sent to the enrichment referenced in the key. | `"fieldMap": {"hbaseEnrichment": ["ip_src_addr","ip_dst_addr"]}` |
+| `config` | The general configuration for the enrichment | `"config": {"typeToColumnFamily": { "asset_enrichment" : "cf" } }` |
The `config` map is intended to house enrichment specific configuration.
For instance, for the `hbaseEnrichment`, the mappings between the
enrichment types to the column families is specified.
+The `fieldMap`contents are of interest because they contain the routing and configuration information for the enrichments. When we say 'routing', we mean how the messages get split up and sent to the enrichment adapter bolts. The simplest, by far, is just providing a simple list as in
+```
+ "fieldMap": {
+ "geo": [
+ "ip_src_addr",
+ "ip_dst_addr"
+ ],
+ "host": [
+ "ip_src_addr",
+ "ip_dst_addr"
+ ],
+ "hbaseEnrichment": [
+ "ip_src_addr",
+ "ip_dst_addr"
+ ]
+ }
+```
+For the `geo`, `host` and `hbaseEnrichment`, this is sufficient. However, more complex enrichments may contain their own configuration. Currently, the `stellar` enrichment requires a more complex configuration, such as:
+```
+ "fieldMap": {
+ ...
+ "stellar" : {
+ "config" : {
+ "numeric" : {
+ "foo": "1 + 1"
+ }
+ ,"ALL_CAPS" : "TO_UPPER(source.type)"
+ }
+ }
+ }
+```
+
+Whereas the simpler enrichments just need a set of fields explicitly stated so they can be separated from the message and sent to the enrichment adapter bolt for enrichment and ultimately joined back in the join bolt, the stellar enrichment has its set of required fields implicitly stated through usage. For instance, if your stellar statement references a field, it should be included and if not, then it should not be included. We did not want to require users to make explicit the implicit.
+
+The other way in which the stellar enrichment is somewhat more complex is in how the statements are executed. In the general purpose case for a list of fields, those fields are used to create a message to send to the enrichment adapter bolt and that bolt's worker will handle the fields one by one in serial for a given message. For stellar enrichment, we wanted to have a more complex design so that users could specify the groups of stellar statements sent to the same worker in the same message (and thus executed sequentially). Consider the following configuration:
+```
+ "fieldMap": {
+ "stellar" : {
+ "config" : {
+ "numeric" : {
+ "foo": "1 + 1"
+ "bar" : TO_LOWER(source.type)"
+ }
+ ,"text" : {
+ "ALL_CAPS" : "TO_UPPER(source.type)"
+ }
+ }
+ }
+ }
+```
+We have a group called `numeric` whose stellar statements will be executed sequentially. In parallel to that, we have the group of stellar statements under the group `text` executing. The intent here is to allow you to not force higher latency operations to be done sequentially.
+
###The `threatIntel` Configuration
| Field | Description | Example |
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java
index fd51505..9318bf0 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java
@@ -44,7 +44,6 @@ public abstract class ConfiguredBolt<CONFIG_T extends Configurations> extends Ba
protected CuratorFramework client;
protected TreeCache cache;
- protected org.apache.metron.common.dsl.Context context;
private final CONFIG_T configurations = defaultConfigurations();
public ConfiguredBolt(String zookeeperUrl) {
this.zookeeperUrl = zookeeperUrl;
@@ -65,6 +64,8 @@ public abstract class ConfiguredBolt<CONFIG_T extends Configurations> extends Ba
}
protected abstract CONFIG_T defaultConfigurations();
+
+
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
try {
@@ -73,9 +74,7 @@ public abstract class ConfiguredBolt<CONFIG_T extends Configurations> extends Ba
client = CuratorFrameworkFactory.newClient(zookeeperUrl, retryPolicy);
}
client.start();
- this.context = new Context.Builder()
- .with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
- .build();
+
if (cache == null) {
cache = new TreeCache(client, Constants.ZOOKEEPER_TOPOLOGY_ROOT);
TreeCacheListener listener = new TreeCacheListener() {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/EnrichmentConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/EnrichmentConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/EnrichmentConfig.java
index 932b070..3261c9f 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/EnrichmentConfig.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/EnrichmentConfig.java
@@ -18,12 +18,16 @@
package org.apache.metron.common.configuration.enrichment;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler;
+
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class EnrichmentConfig {
- private Map<String, List<String>> fieldMap = new HashMap<>();
+ private Map<String, Object> fieldMap = new HashMap<>();
+ private Map<String, ConfigHandler> enrichmentConfigs = new HashMap<>();
private Map<String, List<String>> fieldToTypeMap = new HashMap<>();
private Map<String, Object> config = new HashMap<>();
@@ -35,12 +39,26 @@ public class EnrichmentConfig {
this.config = config;
}
- public Map<String, List<String>> getFieldMap() {
+ public Map<String, Object> getFieldMap() {
return fieldMap;
}
- public void setFieldMap(Map<String, List<String>> fieldMap) {
+ @JsonIgnore
+ public Map<String, ConfigHandler> getEnrichmentConfigs() {
+ return enrichmentConfigs;
+ }
+
+
+ public void setFieldMap(Map<String, Object> fieldMap) {
this.fieldMap = fieldMap;
+ for(Map.Entry<String, Object> kv : fieldMap.entrySet()) {
+ if(kv.getValue() instanceof List) {
+ enrichmentConfigs.put(kv.getKey(), new ConfigHandler((List<String>)kv.getValue()));
+ }
+ else {
+ enrichmentConfigs.put(kv.getKey(), new ConfigHandler(kv.getKey(), (Map<String, Object>)kv.getValue()));
+ }
+ }
}
public Map<String, List<String>> getFieldToTypeMap() {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/SensorEnrichmentUpdateConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/SensorEnrichmentUpdateConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/SensorEnrichmentUpdateConfig.java
index 259fcf0..cff7d03 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/SensorEnrichmentUpdateConfig.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/SensorEnrichmentUpdateConfig.java
@@ -116,13 +116,13 @@ public class SensorEnrichmentUpdateConfig {
Map<String, SensorEnrichmentConfig> sourceConfigsChanged = new HashMap<>();
for (Map.Entry<String, FieldList> kv : sensorToFieldList.entrySet()) {
SensorEnrichmentConfig config = findConfigBySensorType(scHandler, sourceConfigsChanged, kv.getKey());
- Map<String, List<String> > fieldMap = null;
- Map<String, List<String> > fieldToTypeMap = null;
+ Map<String, Object > fieldMap = null;
+ Map<String, List<String>> fieldToTypeMap = null;
List<String> fieldList = null;
if(kv.getValue().type == Type.THREAT_INTEL) {
fieldMap = config.getThreatIntel().getFieldMap();
if(fieldMap!= null) {
- fieldList = fieldMap.get(Constants.SIMPLE_HBASE_THREAT_INTEL);
+ fieldList = (List<String>)fieldMap.get(Constants.SIMPLE_HBASE_THREAT_INTEL);
} else {
fieldMap = new HashMap<>();
}
@@ -139,7 +139,7 @@ public class SensorEnrichmentUpdateConfig {
else if(kv.getValue().type == Type.ENRICHMENT) {
fieldMap = config.getEnrichment().getFieldMap();
if(fieldMap!= null) {
- fieldList = fieldMap.get(Constants.SIMPLE_HBASE_ENRICHMENT);
+ fieldList = (List<String>)fieldMap.get(Constants.SIMPLE_HBASE_ENRICHMENT);
} else {
fieldMap = new HashMap<>();
}
@@ -180,7 +180,7 @@ public class SensorEnrichmentUpdateConfig {
final HashSet<String> types = new HashSet<>(fieldToType.getValue());
int sizeBefore = 0;
if(fieldToTypeMap.containsKey(field)) {
- List<String> typeList = fieldToTypeMap.get(field);
+ List<String> typeList = (List<String>)fieldToTypeMap.get(field);
sizeBefore = new HashSet<>(typeList).size();
types.addAll(typeList);
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Config.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Config.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Config.java
new file mode 100644
index 0000000..f447326
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Config.java
@@ -0,0 +1,36 @@
+/**
+ * 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.common.configuration.enrichment.handler;
+
+
+import com.google.common.collect.ImmutableList;
+import org.json.simple.JSONObject;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+public interface Config {
+ List<JSONObject> splitByFields( JSONObject message
+ , Object fields
+ , Function<String, String> fieldToEnrichmentKey
+ , Map<String, Object> config
+ );
+
+ List<String> getSubgroups(Map<String, Object> config);
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ConfigHandler.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ConfigHandler.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ConfigHandler.java
new file mode 100644
index 0000000..996b33f
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ConfigHandler.java
@@ -0,0 +1,76 @@
+/**
+ * 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.common.configuration.enrichment.handler;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ConfigHandler {
+ private Map<String, Object> config;
+ private Configs type = Configs.LIST;
+ public ConfigHandler(String enrichment, Map<String, Object> obj) {
+ config = (Map<String, Object>) obj.get("config");
+ if(obj.containsKey("type")) {
+ type = Configs.valueOf((String) obj.get("type"));
+ }
+ else {
+ //TODO: make this more adaptable
+ type = Configs.valueOf(enrichment.toUpperCase());
+ }
+ }
+
+ public ConfigHandler(List<String> obj) {
+ config = new HashMap<>();
+ type = Configs.LIST;
+ }
+ public Map<String, Object> getConfig() {
+ return config;
+ }
+
+ public void setConfig(Map<String, Object> config) {
+ this.config = config;
+ }
+
+ public Configs getType() {
+ return type;
+ }
+
+ public void setType(Configs retriever) {
+ this.type = retriever;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ConfigHandler that = (ConfigHandler) o;
+
+ if (getConfig() != null ? !getConfig().equals(that.getConfig()) : that.getConfig() != null) return false;
+ return getType() != null ? getType().equals(that.getType()) : that.getType() == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = getConfig() != null ? getConfig().hashCode() : 0;
+ result = 31 * result + (getType() != null ? getType().hashCode() : 0);
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Configs.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Configs.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Configs.java
new file mode 100644
index 0000000..210caf4
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/Configs.java
@@ -0,0 +1,51 @@
+/**
+ * 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.common.configuration.enrichment.handler;
+
+import org.json.simple.JSONObject;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+public enum Configs implements Config {
+ STELLAR(new StellarConfig())
+ ,LIST(new ListConfig())
+ ;
+ Config configCreator;
+ Configs(Config retriever) {
+ this.configCreator = retriever;
+ }
+
+ @Override
+ public List<JSONObject> splitByFields(JSONObject message
+ , Object fields
+ , Function<String, String> fieldToEnrichmentKey
+ , Map<String, Object> config
+ )
+ {
+ return configCreator.splitByFields(message, fields, fieldToEnrichmentKey, config);
+ }
+
+ @Override
+ public List<String> getSubgroups(Map<String, Object> config) {
+ return configCreator.getSubgroups(config);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ListConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ListConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ListConfig.java
new file mode 100644
index 0000000..7803947
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/ListConfig.java
@@ -0,0 +1,51 @@
+/**
+ * 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.common.configuration.enrichment.handler;
+
+import com.google.common.collect.ImmutableList;
+import org.json.simple.JSONObject;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+public class ListConfig implements Config {
+
+ @Override
+ public List<JSONObject> splitByFields( JSONObject message
+ , Object fieldsObj
+ , Function<String, String> fieldToEnrichmentKey
+ , Map<String, Object> config
+ )
+ {
+ List<String> fields = (List<String>)fieldsObj;
+ JSONObject enrichmentObject = new JSONObject();
+ if (fields != null && fields.size() > 0) {
+ for (String field : fields) {
+ enrichmentObject.put(fieldToEnrichmentKey.apply(field), message.get(field));
+ }
+
+ }
+ return ImmutableList.of(enrichmentObject);
+ }
+
+ @Override
+ public List<String> getSubgroups(Map<String, Object> config) {
+ return ImmutableList.of("");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/StellarConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/StellarConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/StellarConfig.java
new file mode 100644
index 0000000..9a26873
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/enrichment/handler/StellarConfig.java
@@ -0,0 +1,94 @@
+/**
+ * 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.common.configuration.enrichment.handler;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.metron.common.stellar.StellarProcessor;
+import org.json.simple.JSONObject;
+
+import java.util.*;
+import java.util.function.Function;
+
+public class StellarConfig implements Config {
+
+ @Override
+ public List<String> getSubgroups(Map<String, Object> config) {
+ boolean includeEmpty = false;
+ List<String> ret = new ArrayList<>();
+ for(Map.Entry<String, Object> kv : config.entrySet()) {
+ if(kv.getValue() instanceof String) {
+ includeEmpty = true;
+ }
+ else if(kv.getValue() instanceof Map) {
+ ret.add(kv.getKey());
+ }
+ }
+ if(includeEmpty) {
+ ret.add("");
+ }
+ return ret;
+ }
+
+ @Override
+ public List<JSONObject> splitByFields( JSONObject message
+ , Object fields
+ , Function<String, String> fieldToEnrichmentKey
+ , Map<String, Object> config
+ )
+ {
+ StellarProcessor processor = new StellarProcessor();
+ List<JSONObject> messages = new ArrayList<>();
+ Map<String, String> defaultStellarStatementGroup = new HashMap<>();
+ for(Map.Entry<String, Object> kv : config.entrySet()) {
+ if(kv.getValue() instanceof String) {
+ defaultStellarStatementGroup.put(kv.getKey(), (String)kv.getValue());
+ }
+ else if(kv.getValue() instanceof Map) {
+ JSONObject ret = new JSONObject();
+ ret.put(kv.getKey(), getMessage(processor, (Map<String, String>) kv.getValue(), message));
+ messages.add(ret);
+ }
+ }
+ if(defaultStellarStatementGroup.size() > 0)
+ {
+ JSONObject ret = new JSONObject();
+ ret.put("", getMessage(processor, defaultStellarStatementGroup, message));
+ messages.add(ret);
+ }
+ return messages;
+ }
+
+ private Map<String, Object> getMessage( StellarProcessor processor
+ , Map<String, String> stellarStatementGroup
+ , JSONObject message
+ )
+ {
+ Set<String> stellarFields = new HashSet<>();
+ for(String stellarStatement: stellarStatementGroup.values()) {
+ Set<String> variables = processor.variablesUsed(stellarStatement);
+ if(variables != null) {
+ stellarFields.addAll(variables);
+ }
+ }
+ Map<String, Object> messageSegment = new HashMap<>();
+ for(String variable : stellarFields) {
+ messageSegment.put(variable, message.get(variable));
+ }
+ return messageSegment;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
index a178851..8d56af2 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
@@ -29,7 +29,9 @@ public class MapVariableResolver implements VariableResolver {
public MapVariableResolver(Map variableMappingOne, Map... variableMapping) {
variableMappings.add(variableMappingOne);
for(Map m : variableMapping) {
- this.variableMappings.add(m);
+ if(m != null && m.size() > 0) {
+ this.variableMappings.add(m);
+ }
}
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
index 78ba2de..e0b92ec 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
@@ -21,7 +21,13 @@ package org.apache.metron.common.stellar;
import org.antlr.v4.runtime.ANTLRInputStream;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.TokenStream;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
import org.apache.metron.common.dsl.*;
+import org.apache.metron.common.stellar.generated.StellarBaseListener;
import org.apache.metron.common.stellar.generated.StellarLexer;
import org.apache.metron.common.stellar.generated.StellarParser;
@@ -34,6 +40,35 @@ public class BaseStellarProcessor<T> {
public BaseStellarProcessor(Class<T> clazz) {
this.clazz = clazz;
}
+
+ public Set<String> variablesUsed(String rule) {
+ if (rule == null || isEmpty(rule.trim())) {
+ return null;
+ }
+ ANTLRInputStream input = new ANTLRInputStream(rule);
+ StellarLexer lexer = new StellarLexer(input);
+ lexer.removeErrorListeners();
+ lexer.addErrorListener(new ErrorListener());
+ TokenStream tokens = new CommonTokenStream(lexer);
+ StellarParser parser = new StellarParser(tokens);
+ final Set<String> ret = new HashSet<>();
+ parser.addParseListener(new StellarBaseListener() {
+ @Override
+ public void exitVariable(StellarParser.VariableContext ctx) {
+ ret.add(ctx.getText());
+ }
+ @Override
+ public void exitExistsFunc(StellarParser.ExistsFuncContext ctx) {
+ String variable = ctx.getChild(2).getText();
+ ret.add(variable);
+ }
+ });
+ parser.removeErrorListeners();
+ parser.addErrorListener(new ErrorListener());
+ parser.transformation();
+ return ret;
+ }
+
public T parse( String rule
, VariableResolver variableResolver
, Function<String, StellarFunction> functionResolver
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java
index 520b430..64fdace 100644
--- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java
+++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java
@@ -118,12 +118,12 @@ public class ConfiguredEnrichmentBoltTest extends BaseConfiguredBoltTest {
SensorEnrichmentConfig testSensorConfig = new SensorEnrichmentConfig();
testSensorConfig.setBatchSize(50);
testSensorConfig.setIndex("test");
- Map<String, List<String>> enrichmentFieldMap = new HashMap<>();
+ Map<String, Object> enrichmentFieldMap = new HashMap<>();
enrichmentFieldMap.put("enrichmentTest", new ArrayList<String>() {{
add("enrichmentField");
}});
testSensorConfig.getEnrichment().setFieldMap(enrichmentFieldMap);
- Map<String, List<String>> threatIntelFieldMap = new HashMap<>();
+ Map<String, Object> threatIntelFieldMap = new HashMap<>();
threatIntelFieldMap.put("threatIntelTest", new ArrayList<String>() {{
add("threatIntelField");
}});
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/EnrichmentConfigTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/EnrichmentConfigTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/EnrichmentConfigTest.java
new file mode 100644
index 0000000..21937d0
--- /dev/null
+++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/EnrichmentConfigTest.java
@@ -0,0 +1,59 @@
+/**
+ * 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.common.configuration;
+
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.metron.common.configuration.enrichment.EnrichmentConfig;
+import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler;
+import org.apache.metron.common.configuration.enrichment.handler.Configs;
+import org.apache.metron.common.utils.JSONUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Map;
+
+public class EnrichmentConfigTest {
+ /**
+ {
+ "fieldMap": {
+ "geo": ["ip_dst_addr", "ip_src_addr"],
+ "host": ["host"],
+ "stellar" : {
+ "type" : "STELLAR"
+ ,"config" : {
+ "foo" : "1 + 1"
+ ,"ALL_CAPS" : "TO_UPPER(source.type)"
+ }
+ }
+ }
+ }
+ */
+ @Multiline
+ public static String sourceConfigStr;
+
+ @Test
+ public void testSerialization() throws Exception
+ {
+ EnrichmentConfig config = JSONUtils.INSTANCE.load(sourceConfigStr, EnrichmentConfig.class);
+ Assert.assertTrue(config.getFieldMap().get("stellar") instanceof Map);
+ Assert.assertTrue(config.getEnrichmentConfigs().get("stellar") instanceof ConfigHandler);
+ Assert.assertEquals(Configs.STELLAR, ((ConfigHandler)config.getEnrichmentConfigs().get("stellar")).getType());
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/SensorEnrichmentUpdateConfigTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/SensorEnrichmentUpdateConfigTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/SensorEnrichmentUpdateConfigTest.java
index 681c4c1..f907144 100644
--- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/SensorEnrichmentUpdateConfigTest.java
+++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/SensorEnrichmentUpdateConfigTest.java
@@ -26,6 +26,7 @@ import org.junit.Assert;
import org.junit.Test;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
public class SensorEnrichmentUpdateConfigTest {
@@ -101,18 +102,18 @@ public class SensorEnrichmentUpdateConfigTest {
Assert.assertNotNull(finalEnrichmentConfig.get("bro"));
Assert.assertNotSame(finalEnrichmentConfig.get("bro"), broSc);
Assert.assertEquals( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldMap().get(Constants.SIMPLE_HBASE_THREAT_INTEL).size()
+ , ((List<String>)finalEnrichmentConfig.get("bro").getThreatIntel().getFieldMap().get(Constants.SIMPLE_HBASE_THREAT_INTEL)).size()
, 2
);
Assert.assertEquals(1, finalEnrichmentConfig.get("bro").getThreatIntel().getTriageConfig().getRiskLevelRules().size());
Assert.assertTrue( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldMap()
- .get(Constants.SIMPLE_HBASE_THREAT_INTEL)
+ , ((List<String>)finalEnrichmentConfig.get("bro").getThreatIntel().getFieldMap()
+ .get(Constants.SIMPLE_HBASE_THREAT_INTEL))
.contains("ip_src_addr")
);
Assert.assertTrue( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldMap()
- .get(Constants.SIMPLE_HBASE_THREAT_INTEL)
+ , ((List<String>)finalEnrichmentConfig.get("bro").getThreatIntel().getFieldMap()
+ .get(Constants.SIMPLE_HBASE_THREAT_INTEL))
.contains("ip_dst_addr")
);
Assert.assertEquals( finalEnrichmentConfig.get("bro").toJSON()
@@ -120,24 +121,24 @@ public class SensorEnrichmentUpdateConfigTest {
, 2
);
Assert.assertEquals( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_src_addr").size()
+ , ((List<String>)(finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_src_addr"))).size()
, 2
);
Assert.assertTrue( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_src_addr").contains("playful")
+ , ((List<String>)(finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_src_addr"))).contains("playful")
);
Assert.assertTrue( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_src_addr").contains("malicious_ip")
+ , ((List<String>)(finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_src_addr"))).contains("malicious_ip")
);
Assert.assertEquals( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_dst_addr").size()
+ , ((List<String>)(finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_dst_addr"))).size()
, 2
);
Assert.assertTrue( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_dst_addr").contains("playful")
+ , ((List<String>)(finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_dst_addr"))).contains("playful")
);
Assert.assertTrue( finalEnrichmentConfig.get("bro").toJSON()
- , finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_dst_addr").contains("malicious_ip")
+ , ((List<String>)(finalEnrichmentConfig.get("bro").getThreatIntel().getFieldToTypeMap().get("ip_dst_addr"))).contains("malicious_ip")
);
}
@@ -184,17 +185,17 @@ public class SensorEnrichmentUpdateConfigTest {
Assert.assertNotNull(outputScs.get("bro"));
Assert.assertNotSame(outputScs.get("bro"), broSc);
Assert.assertEquals( outputScs.get("bro").toJSON()
- , outputScs.get("bro").getEnrichment().getFieldMap().get(Constants.SIMPLE_HBASE_ENRICHMENT).size()
+ , ((List<String>)outputScs.get("bro").getEnrichment().getFieldMap().get(Constants.SIMPLE_HBASE_ENRICHMENT)).size()
, 2
);
Assert.assertTrue( outputScs.get("bro").toJSON()
- , outputScs.get("bro").getEnrichment().getFieldMap()
- .get(Constants.SIMPLE_HBASE_ENRICHMENT)
+ , ((List<String>)outputScs.get("bro").getEnrichment().getFieldMap()
+ .get(Constants.SIMPLE_HBASE_ENRICHMENT))
.contains("ip_src_addr")
);
Assert.assertTrue( outputScs.get("bro").toJSON()
- , outputScs.get("bro").getEnrichment().getFieldMap()
- .get(Constants.SIMPLE_HBASE_ENRICHMENT)
+ , ((List<String>)outputScs.get("bro").getEnrichment().getFieldMap()
+ .get(Constants.SIMPLE_HBASE_ENRICHMENT))
.contains("ip_dst_addr")
);
Assert.assertEquals( outputScs.get("bro").toJSON()
@@ -202,19 +203,19 @@ public class SensorEnrichmentUpdateConfigTest {
, 2
);
Assert.assertEquals( outputScs.get("bro").toJSON()
- , outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_src_addr").size()
+ , ((List<String>)(outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_src_addr"))).size()
, 1
);
Assert.assertEquals( outputScs.get("bro").toJSON()
- , outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_src_addr").get(0)
+ , ((List<String>)(outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_src_addr"))).get(0)
, "playful"
);
Assert.assertEquals( outputScs.get("bro").toJSON()
- , outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_dst_addr").size()
+ , ((List<String>)(outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_dst_addr"))).size()
, 1
);
Assert.assertEquals( outputScs.get("bro").toJSON()
- , outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_dst_addr").get(0)
+ , ((List<String>)(outputScs.get("bro").getEnrichment().getFieldToTypeMap().get("ip_dst_addr"))).get(0)
, "playful"
);
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-common/src/test/java/org/apache/metron/common/stellar/StellarTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/stellar/StellarTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/stellar/StellarTest.java
index abcf204..033f719 100644
--- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/stellar/StellarTest.java
+++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/stellar/StellarTest.java
@@ -20,16 +20,34 @@ package org.apache.metron.common.stellar;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import org.apache.metron.common.dsl.*;
import org.junit.Assert;
import org.junit.Test;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
public class StellarTest {
@Test
+ public void testVariablesUsed() {
+ StellarProcessor processor = new StellarProcessor();
+ {
+ Assert.assertEquals(new HashSet<>(), processor.variablesUsed("if 1 < 2 then 'one' else 'two'"));
+ }
+ {
+ Assert.assertEquals(ImmutableSet.of("one")
+ , processor.variablesUsed("if 1 < 2 then one else 'two'"));
+ }
+ {
+ Assert.assertEquals(ImmutableSet.of("one", "two")
+ , processor.variablesUsed("if 1 < 2 then one else two"));
+ }
+ }
+
+ @Test
public void testFunctionEmptyArgs() {
{
String query = "STARTS_WITH(casey, 'case') or MAP_EXISTS()";
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/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 69d5f66..e182ade 100644
--- a/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml
+++ b/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml
@@ -21,6 +21,19 @@ config:
components:
# Enrichment
+ - id: "stellarEnrichmentAdapter"
+ className: "org.apache.metron.enrichment.adapters.stellar.StellarAdapter"
+ configMethods:
+ - name: "ofType"
+ args:
+ - "ENRICHMENT"
+
+ - id: "stellarEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "stellar"
+ - ref: "stellarEnrichmentAdapter"
+
- id: "jdbcConfig"
className: "org.apache.metron.enrichment.adapters.jdbc.MySqlConfig"
properties:
@@ -90,8 +103,22 @@ components:
- name: "add"
args:
- ref: "simpleHBaseEnrichment"
-# Threat Intel
+ - name: "add"
+ args:
+ - ref: "stellarEnrichment"
+# Threat Intel
+ - id: "stellarThreatIntelAdapter"
+ className: "org.apache.metron.enrichment.adapters.stellar.StellarAdapter"
+ configMethods:
+ - name: "ofType"
+ args:
+ - "THREAT_INTEL"
+ - id: "stellarThreatIntelEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "stellar"
+ - ref: "stellarThreatIntelAdapter"
- id: "simpleHBaseThreatIntelConfig"
className: "org.apache.metron.enrichment.adapters.threatintel.ThreatIntelConfig"
configMethods:
@@ -128,6 +155,10 @@ components:
- name: "add"
args:
- ref: "simpleHBaseThreatIntelEnrichment"
+ - name: "add"
+ args:
+ - ref: "stellarThreatIntelEnrichment"
+
#indexing
- id: "kafkaWriter"
@@ -189,6 +220,19 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "stellarEnrichmentBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "stellarEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
+
- id: "hostEnrichmentBolt"
className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
constructorArgs:
@@ -246,6 +290,18 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "stellarThreatIntelBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "stellarThreatIntelEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
- id: "threatIntelJoinBolt"
className: "org.apache.metron.enrichment.bolt.ThreatIntelJoinBolt"
constructorArgs:
@@ -256,7 +312,6 @@ bolts:
- name: "withMaxTimeRetain"
args: [10]
# Indexing Bolts
-
- id: "outputBolt"
className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
constructorArgs:
@@ -291,6 +346,15 @@ streams:
type: FIELDS
args: ["key"]
+ - name: "enrichmentSplit -> stellar"
+ from: "enrichmentSplitBolt"
+ to: "stellarEnrichmentBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
+
+
- name: "enrichmentSplit -> simpleHBaseEnrichmentBolt"
from: "enrichmentSplitBolt"
to: "simpleHBaseEnrichmentBolt"
@@ -313,6 +377,14 @@ streams:
streamId: "geo"
type: FIELDS
args: ["key"]
+ - name: "stellar -> join"
+ from: "stellarEnrichmentBolt"
+ to: "enrichmentJoinBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
+
- name: "simpleHBaseEnrichmentBolt -> join"
@@ -346,6 +418,14 @@ streams:
streamId: "hbaseThreatIntel"
type: FIELDS
args: ["key"]
+ - name: "threatSplit -> stellarThreatIntel"
+ from: "threatIntelSplitBolt"
+ to: "stellarThreatIntelBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
+
- name: "simpleHBaseThreatIntel -> join"
from: "simpleHBaseThreatIntelBolt"
@@ -354,6 +434,15 @@ streams:
streamId: "hbaseThreatIntel"
type: FIELDS
args: ["key"]
+
+ - name: "stellarThreatIntel -> join"
+ from: "stellarThreatIntelBolt"
+ to: "threatIntelJoinBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
+
- name: "threatIntelSplit -> threatIntelJoin"
from: "threatIntelSplitBolt"
to: "threatIntelJoinBolt"
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/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 0f391c4..f8d7b83 100644
--- a/metron-platform/metron-enrichment/src/main/flux/enrichment/test.yaml
+++ b/metron-platform/metron-enrichment/src/main/flux/enrichment/test.yaml
@@ -20,6 +20,19 @@ config:
components:
# Enrichment
+ - id: "stellarEnrichmentAdapter"
+ className: "org.apache.metron.enrichment.adapters.stellar.StellarAdapter"
+ configMethods:
+ - name: "ofType"
+ args:
+ - "ENRICHMENT"
+
+ - id: "stellarEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "stellar"
+ - ref: "stellarEnrichmentAdapter"
+
- id: "geoEnrichmentAdapter"
className: "org.apache.metron.enrichment.integration.mock.MockGeoAdapter"
- id: "geoEnrichment"
@@ -72,8 +85,21 @@ components:
- name: "add"
args:
- ref: "simpleHBaseEnrichment"
+ - name: "add"
+ args:
+ - ref: "stellarEnrichment"
# Threat Intel
-
+ - id: "stellarThreatIntelAdapter"
+ className: "org.apache.metron.enrichment.adapters.stellar.StellarAdapter"
+ configMethods:
+ - name: "ofType"
+ args:
+ - "THREAT_INTEL"
+ - id: "stellarThreatIntelEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "stellar"
+ - ref: "stellarThreatIntelAdapter"
- id: "simpleHBaseThreatIntelConfig"
className: "org.apache.metron.enrichment.adapters.threatintel.ThreatIntelConfig"
configMethods:
@@ -110,7 +136,9 @@ components:
- name: "add"
args:
- ref: "simpleHBaseThreatIntelEnrichment"
-
+ - name: "add"
+ args:
+ - ref: "stellarThreatIntelEnrichment"
#indexing
- id: "kafkaWriter"
@@ -172,6 +200,18 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "stellarEnrichmentBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "stellarEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
- id: "hostEnrichmentBolt"
className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
constructorArgs:
@@ -229,6 +269,18 @@ bolts:
args: [10000]
- name: "withMaxTimeRetain"
args: [10]
+ - id: "stellarThreatIntelBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "stellarThreatIntelEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
- id: "threatIntelJoinBolt"
className: "org.apache.metron.enrichment.bolt.ThreatIntelJoinBolt"
constructorArgs:
@@ -273,6 +325,14 @@ streams:
type: FIELDS
args: ["key"]
+ - name: "enrichmentSplit -> stellar"
+ from: "enrichmentSplitBolt"
+ to: "stellarEnrichmentBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
+
- name: "enrichmentSplit -> simpleHBaseEnrichmentBolt"
from: "enrichmentSplitBolt"
to: "simpleHBaseEnrichmentBolt"
@@ -295,6 +355,13 @@ streams:
streamId: "geo"
type: FIELDS
args: ["key"]
+ - name: "stellar -> join"
+ from: "stellarEnrichmentBolt"
+ to: "enrichmentJoinBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
- name: "simpleHBaseEnrichmentBolt -> join"
@@ -328,6 +395,13 @@ streams:
streamId: "hbaseThreatIntel"
type: FIELDS
args: ["key"]
+ - name: "threatSplit -> stellarThreatIntel"
+ from: "threatIntelSplitBolt"
+ to: "stellarThreatIntelBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
- name: "simpleHBaseThreatIntel -> join"
from: "simpleHBaseThreatIntelBolt"
@@ -336,6 +410,14 @@ streams:
streamId: "hbaseThreatIntel"
type: FIELDS
args: ["key"]
+
+ - name: "stellarThreatIntel -> join"
+ from: "stellarThreatIntelBolt"
+ to: "threatIntelJoinBolt"
+ grouping:
+ streamId: "stellar"
+ type: FIELDS
+ args: ["key"]
- name: "threatIntelSplit -> threatIntelJoin"
from: "threatIntelSplitBolt"
to: "threatIntelJoinBolt"
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java
index 63d6c0b..c173444 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java
@@ -61,7 +61,7 @@ public class CIFHbaseAdapter implements EnrichmentAdapter<CacheKey>,Serializable
}
public JSONObject enrich(CacheKey k) {
- String metadata = k.getValue();
+ String metadata = k.getValue(String.class);
JSONObject output = new JSONObject();
LOGGER.debug("=======Looking Up For:" + metadata);
output.putAll(getCIFObject(metadata));
@@ -135,4 +135,9 @@ public class CIFHbaseAdapter implements EnrichmentAdapter<CacheKey>,Serializable
public void cleanup() {
}
+
+ @Override
+ public String getOutputPrefix(CacheKey value) {
+ return value.getField();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/geo/GeoAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/geo/GeoAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/geo/GeoAdapter.java
index fb3febe..5b1bf9b 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/geo/GeoAdapter.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/geo/GeoAdapter.java
@@ -35,6 +35,10 @@ public class GeoAdapter extends JdbcAdapter {
}
+ @Override
+ public String getOutputPrefix(CacheKey value) {
+ return value.getField();
+ }
@SuppressWarnings("unchecked")
@Override
@@ -45,10 +49,10 @@ public class GeoAdapter extends JdbcAdapter {
return enriched;
}
try {
- InetAddress addr = InetAddress.getByName(value.getValue());
+ InetAddress addr = InetAddress.getByName(value.getValue(String.class));
if (addr.isAnyLocalAddress() || addr.isLoopbackAddress()
|| addr.isSiteLocalAddress() || addr.isMulticastAddress()
- || !ipvalidator.isValidInet4Address(value.getValue())) {
+ || !ipvalidator.isValidInet4Address(value.getValue(String.class))) {
return new JSONObject();
}
String locidQuery = "select IPTOLOCID(\"" + value.getValue()
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromJSONListAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromJSONListAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromJSONListAdapter.java
index 347cc03..881f9db 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromJSONListAdapter.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromJSONListAdapter.java
@@ -41,6 +41,11 @@ public class HostFromJSONListAdapter extends AbstractHostAdapter {
}
@Override
+ public String getOutputPrefix(CacheKey value) {
+ return value.getField();
+ }
+
+ @Override
public boolean initializeAdapter()
{
@@ -58,7 +63,7 @@ public class HostFromJSONListAdapter extends AbstractHostAdapter {
@SuppressWarnings("unchecked")
@Override
public JSONObject enrich(CacheKey k) {
- String metadata = k.getValue();
+ String metadata = k.getValue(String.class);
if(!_known_hosts.containsKey(metadata))
return new JSONObject();
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromPropertiesFileAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromPropertiesFileAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromPropertiesFileAdapter.java
index f92bd3f..0555c1a 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromPropertiesFileAdapter.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/host/HostFromPropertiesFileAdapter.java
@@ -44,6 +44,11 @@ public class HostFromPropertiesFileAdapter extends AbstractHostAdapter {
}
@Override
+ public String getOutputPrefix(CacheKey value) {
+ return value.getField();
+ }
+
+ @Override
public void logAccess(CacheKey value) {
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/58294b66/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java
index a152d26..d2af3cc 100644
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java
+++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java
@@ -76,7 +76,7 @@ public class SimpleHBaseAdapter implements EnrichmentAdapter<CacheKey>,Serializa
try {
for (LookupKV<EnrichmentKey, EnrichmentValue> kv :
lookup.get(Iterables.transform(enrichmentTypes
- , new EnrichmentUtils.TypeToKey( value.getValue()
+ , new EnrichmentUtils.TypeToKey( value.getValue(String.class)
, lookup.getTable()
, value.getConfig().getEnrichment()
)
@@ -126,4 +126,9 @@ public class SimpleHBaseAdapter implements EnrichmentAdapter<CacheKey>,Serializa
_LOG.error("Unable to cleanup access tracker", e);
}
}
+
+ @Override
+ public String getOutputPrefix(CacheKey value) {
+ return value.getField();
+ }
}