You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by rm...@apache.org on 2016/04/26 16:46:17 UTC
[29/51] [partial] incubator-metron git commit: METRON-113 Project
Reorganization (merrimanr) closes apache/incubator-metron#88
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/pom.xml b/metron-platform/metron-elasticsearch/pom.xml
new file mode 100644
index 0000000..974d396
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/pom.xml
@@ -0,0 +1,250 @@
+<?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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-platform</artifactId>
+ <version>0.1BETA</version>
+ </parent>
+ <artifactId>metron-elasticsearch</artifactId>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${global_hbase_guava_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-enrichment</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <version>${global_elasticsearch_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ <version>${global_storm_version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <groupId>org.slf4j</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${global_hbase_version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <version>${global_hadoop_version}</version>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${global_hadoop_version}</version>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_2.9.2</artifactId>
+ <version>${global_kafka_version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-integration-test</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>${global_mockito_version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+ <reporting>
+ <plugins>
+ <!-- Normally, dependency report takes time, skip it -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-project-info-reports-plugin</artifactId>
+ <version>2.7</version>
+
+ <configuration>
+ <dependencyLocationsEnabled>false</dependencyLocationsEnabled>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>emma-maven-plugin</artifactId>
+ <version>1.0-alpha-3</version>
+ <inherited>true</inherited>
+ </plugin>
+ </plugins>
+ </reporting>
+
+ <build>
+ <plugins>
+ <plugin>
+ <!-- Separates the unit tests from the integration tests. -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.12.4</version>
+ <configuration>
+ <!-- Skip the default running of this plug-in (or everything is run twice...see below) -->
+ <argLine>-Xmx2048m -XX:MaxPermSize=256m</argLine>
+ <skip>true</skip>
+ <!-- Show 100% of the lines from the stack trace (doesn't work) -->
+ <trimStackTrace>false</trimStackTrace>
+ </configuration>
+ <executions>
+ <execution>
+ <id>unit-tests</id>
+ <phase>test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <!-- Never skip running the tests when the test phase is invoked -->
+ <skip>false</skip>
+ <includes>
+ <!-- Include unit tests within integration-test phase. -->
+ <include>**/*Test.java</include>
+ </includes>
+ <excludes>
+ <!-- Exclude integration tests within (unit) test phase. -->
+ <exclude>**/*IntegrationTest.java</exclude>
+ </excludes>
+ </configuration>
+ </execution>
+ <execution>
+ <id>integration-tests</id>
+ <phase>integration-test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <!-- Never skip running the tests when the integration-test phase is invoked -->
+ <skip>false</skip>
+ <includes>
+ <!-- Include integration tests within integration-test phase. -->
+ <include>**/*IntegrationTest.java</include>
+ </includes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>2.3</version>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <excludes>
+ <exclude>storm:storm-core:*</exclude>
+ <exclude>storm:storm-lib:*</exclude>
+ <exclude>org.slf4j.impl*</exclude>
+ <exclude>org.slf4j:slf4j-log4j*</exclude>
+ </excludes>
+ </artifactSet>
+ <transformers>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.DontIncludeResourceTransformer">
+ <resource>.yaml</resource>
+ </transformer>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass></mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <configuration>
+ <descriptor>src/main/assembly/assembly.xml</descriptor>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id> <!-- this is used for inheritance merges -->
+ <phase>package</phase> <!-- bind to the packaging phase -->
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/main/assembly/assembly.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/main/assembly/assembly.xml b/metron-platform/metron-elasticsearch/src/main/assembly/assembly.xml
new file mode 100644
index 0000000..bacaae3
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/main/assembly/assembly.xml
@@ -0,0 +1,52 @@
+<!--
+ 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.
+ -->
+
+<assembly>
+ <id>archive</id>
+ <formats>
+ <format>tar.gz</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${project.basedir}/src/main/config</directory>
+ <outputDirectory>/config</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ <excludes>
+ <exclude>**/*.formatted</exclude>
+ <exclude>**/*.filtered</exclude>
+ </excludes>
+ <fileMode>0644</fileMode>
+ <lineEnding>unix</lineEnding>
+ </fileSet>
+ <fileSet>
+ <directory>${project.basedir}/src/main/scripts</directory>
+ <outputDirectory>/scripts</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ <excludes>
+ <exclude>**/*.formatted</exclude>
+ <exclude>**/*.filtered</exclude>
+ </excludes>
+ <fileMode>0644</fileMode>
+ <lineEnding>unix</lineEnding>
+ </fileSet>
+ <fileSet>
+ <directory>${project.basedir}/target</directory>
+ <includes>
+ <include>${project.artifactId}-${project.version}.jar</include>
+ </includes>
+ <outputDirectory>/lib</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ </fileSet>
+ </fileSets>
+</assembly>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/main/config/elasticsearch.properties
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/main/config/elasticsearch.properties b/metron-platform/metron-elasticsearch/src/main/config/elasticsearch.properties
new file mode 100644
index 0000000..e2370ab
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/main/config/elasticsearch.properties
@@ -0,0 +1,109 @@
+# 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.
+
+
+##### Kafka #####
+
+kafka.zk=node1:2181
+kafka.broker=node1:6667
+spout.kafka.topic.asa=asa
+spout.kafka.topic.bro=bro
+spout.kafka.topic.fireeye=fireeye
+spout.kafka.topic.ise=ise
+spout.kafka.topic.lancope=lancope
+spout.kafka.topic.paloalto=paloalto
+spout.kafka.topic.pcap=pcap
+spout.kafka.topic.snort=snort
+spout.kafka.topic.yaf=yaf
+
+##### Indexing #####
+writer.class.name=org.apache.metron.elasticsearch.writer.ElasticsearchWriter
+
+##### ElasticSearch #####
+
+es.ip=10.22.0.214
+es.port=9300
+es.clustername=elasticsearch
+
+##### MySQL #####
+
+mysql.ip=10.22.0.214
+mysql.port=3306
+mysql.username=root
+mysql.password=hadoop123
+
+##### Metrics #####
+
+#reporters
+org.apache.metron.metrics.reporter.graphite=true
+org.apache.metron.metrics.reporter.console=false
+org.apache.metron.metrics.reporter.jmx=false
+
+#Graphite Addresses
+
+org.apache.metron.metrics.graphite.address=localhost
+org.apache.metron.metrics.graphite.port=2023
+
+#TelemetryParserBolt
+org.apache.metron.metrics.TelemetryParserBolt.acks=true
+org.apache.metron.metrics.TelemetryParserBolt.emits=true
+org.apache.metron.metrics.TelemetryParserBolt.fails=true
+
+
+#GenericEnrichmentBolt
+org.apache.metron.metrics.GenericEnrichmentBolt.acks=true
+org.apache.metron.metrics.GenericEnrichmentBolt.emits=true
+org.apache.metron.metrics.GenericEnrichmentBolt.fails=true
+
+
+#TelemetryIndexingBolt
+org.apache.metron.metrics.TelemetryIndexingBolt.acks=true
+org.apache.metron.metrics.TelemetryIndexingBolt.emits=true
+org.apache.metron.metrics.TelemetryIndexingBolt.fails=true
+
+##### Host Enrichment #####
+
+org.apache.metron.enrichment.host.known_hosts=[{"ip":"10.1.128.236", "local":"YES", "type":"webserver", "asset_value" : "important"},\
+{"ip":"10.1.128.237", "local":"UNKNOWN", "type":"unknown", "asset_value" : "important"},\
+{"ip":"10.60.10.254", "local":"YES", "type":"printer", "asset_value" : "important"}]
+
+##### HDFS #####
+
+bolt.hdfs.batch.size=5000
+bolt.hdfs.field.delimiter=|
+bolt.hdfs.file.rotation.size.in.mb=5
+bolt.hdfs.file.system.url=hdfs://iot01.cloud.hortonworks.com:8020
+bolt.hdfs.wip.file.path=/paloalto/wip
+bolt.hdfs.finished.file.path=/paloalto/rotated
+bolt.hdfs.compression.codec.class=org.apache.hadoop.io.compress.SnappyCodec
+index.hdfs.output=/tmp/metron/enriched
+
+##### HBase #####
+bolt.hbase.table.name=pcap
+bolt.hbase.table.fields=t:value
+bolt.hbase.table.key.tuple.field.name=key
+bolt.hbase.table.timestamp.tuple.field.name=timestamp
+bolt.hbase.enable.batching=false
+bolt.hbase.write.buffer.size.in.bytes=2000000
+bolt.hbase.durability=SKIP_WAL
+bolt.hbase.partitioner.region.info.refresh.interval.mins=60
+
+##### Threat Intel #####
+
+threat.intel.tracker.table=
+threat.intel.tracker.cf=
+threat.intel.ip.table=
+threat.intel.ip.cf=
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java
new file mode 100644
index 0000000..6b54fec
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.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.elasticsearch.writer;
+
+import backtype.storm.tuple.Tuple;
+import org.apache.metron.common.configuration.Configurations;
+import org.apache.metron.common.configuration.SensorEnrichmentConfig;
+import org.apache.metron.common.interfaces.BulkMessageWriter;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.json.simple.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+public class ElasticsearchWriter implements BulkMessageWriter<JSONObject>, Serializable {
+
+ private Map<String, String> optionalSettings;
+ private transient TransportClient client;
+ private SimpleDateFormat dateFormat;
+ private static final Logger LOG = LoggerFactory
+ .getLogger(ElasticsearchWriter.class);
+
+ public ElasticsearchWriter withOptionalSettings(Map<String, String> optionalSettings) {
+ this.optionalSettings = optionalSettings;
+ return this;
+ }
+
+ @Override
+ public void init(Map stormConf, Configurations configurations) {
+ Map<String, Object> globalConfiguration = configurations.getGlobalConfig();
+ ImmutableSettings.Builder builder = ImmutableSettings.settingsBuilder();
+ builder.put("cluster.name", globalConfiguration.get("es.clustername"));
+ builder.put("client.transport.ping_timeout","500s");
+ if (optionalSettings != null) {
+ builder.put(optionalSettings);
+ }
+ client = new TransportClient(builder.build())
+ .addTransportAddress(new InetSocketTransportAddress(globalConfiguration.get("es.ip").toString(), Integer.parseInt(globalConfiguration.get("es.port").toString())));
+ dateFormat = new SimpleDateFormat((String) globalConfiguration.get("es.date.format"));
+
+ }
+
+ @Override
+ public void write(String sensorType, Configurations configurations, List<Tuple> tuples, List<JSONObject> messages) throws Exception {
+ SensorEnrichmentConfig sensorEnrichmentConfig = configurations.getSensorEnrichmentConfig(sensorType);
+ String indexPostfix = dateFormat.format(new Date());
+ BulkRequestBuilder bulkRequest = client.prepareBulk();
+ for(JSONObject message: messages) {
+ String indexName = sensorType;
+ if (sensorEnrichmentConfig != null) {
+ indexName = sensorEnrichmentConfig.getIndex();
+ }
+ IndexRequestBuilder indexRequestBuilder = client.prepareIndex(indexName + "_index_" + indexPostfix,
+ sensorType + "_doc");
+
+ indexRequestBuilder.setSource(message.toJSONString());
+ bulkRequest.add(indexRequestBuilder);
+ }
+ BulkResponse resp = bulkRequest.execute().actionGet();
+ if (resp.hasFailures()) {
+ throw new Exception(resp.buildFailureMessage());
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ client.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/main/scripts/start_elasticsearch_topology.sh
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/main/scripts/start_elasticsearch_topology.sh b/metron-platform/metron-elasticsearch/src/main/scripts/start_elasticsearch_topology.sh
new file mode 100755
index 0000000..ed80d82
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/main/scripts/start_elasticsearch_topology.sh
@@ -0,0 +1,22 @@
+#!/bin/bash
+#
+# 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.
+#
+METRON_VERSION=0.1BETA
+METRON_HOME=/usr/metron/$METRON_VERSION
+TOPOLOGY_JAR=metron-elasticsearch-$METRON_VERSION.jar
+storm jar $METRON_HOME/lib/$TOPOLOGY_JAR org.apache.storm.flux.Flux --remote $METRON_HOME/config/enrichment/remote.yaml --filter $METRON_HOME/config/elasticsearch.properties
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchEnrichmentIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchEnrichmentIntegrationTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchEnrichmentIntegrationTest.java
new file mode 100644
index 0000000..6931aff
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchEnrichmentIntegrationTest.java
@@ -0,0 +1,89 @@
+/**
+ * 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.elasticsearch.integration;
+
+import org.apache.metron.integration.EnrichmentIntegrationTest;
+import org.apache.metron.integration.ComponentRunner;
+import org.apache.metron.integration.InMemoryComponent;
+import org.apache.metron.integration.Processor;
+import org.apache.metron.integration.ReadinessState;
+import org.apache.metron.elasticsearch.integration.components.ElasticSearchComponent;
+
+import java.io.File;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+public class ElasticsearchEnrichmentIntegrationTest extends EnrichmentIntegrationTest {
+
+ private String indexDir = "target/elasticsearch";
+ private String dateFormat = "yyyy.MM.dd.HH";
+ private String index = "yaf_index_" + new SimpleDateFormat(dateFormat).format(new Date());
+
+ @Override
+ public InMemoryComponent getSearchComponent(final Properties topologyProperties) {
+ return new ElasticSearchComponent.Builder()
+ .withHttpPort(9211)
+ .withIndexDir(new File(indexDir))
+ .build();
+ }
+
+ @Override
+ public Processor<List<Map<String, Object>>> getProcessor(final List<byte[]> inputMessages) {
+ return new Processor<List<Map<String, Object>>>() {
+ List<Map<String, Object>> docs = null;
+ public ReadinessState process(ComponentRunner runner) {
+ ElasticSearchComponent elasticSearchComponent = runner.getComponent("search", ElasticSearchComponent.class);
+ if (elasticSearchComponent.hasIndex(index)) {
+ List<Map<String, Object>> docsFromDisk;
+ try {
+ docs = elasticSearchComponent.getAllIndexedDocs(index, "yaf_doc");
+ docsFromDisk = readDocsFromDisk(hdfsDir);
+ System.out.println(docs.size() + " vs " + inputMessages.size() + " vs " + docsFromDisk.size());
+ } catch (IOException e) {
+ throw new IllegalStateException("Unable to retrieve indexed documents.", e);
+ }
+ if (docs.size() < inputMessages.size() || docs.size() != docsFromDisk.size()) {
+ return ReadinessState.NOT_READY;
+ } else {
+ return ReadinessState.READY;
+ }
+ } else {
+ return ReadinessState.NOT_READY;
+ }
+ }
+
+ public List<Map<String, Object>> getResult() {
+ return docs;
+ }
+ };
+ }
+
+ @Override
+ public void setAdditionalProperties(Properties topologyProperties) {
+ topologyProperties.setProperty("writer.class.name", "org.apache.metron.elasticsearch.writer.ElasticsearchWriter");
+ }
+
+ @Override
+ public String cleanField(String field) {
+ return field;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java
new file mode 100644
index 0000000..61d50f1
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java
@@ -0,0 +1,186 @@
+/**
+ * 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.elasticsearch.integration.components;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.metron.integration.InMemoryComponent;
+import org.apache.metron.integration.UnableToStartException;
+import org.elasticsearch.ElasticsearchTimeoutException;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.ElasticsearchClient;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.node.NodeBuilder;
+import org.elasticsearch.search.SearchHit;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class ElasticSearchComponent implements InMemoryComponent {
+
+ public static class Builder{
+ private int httpPort;
+ private File indexDir;
+ private Map<String, String> extraElasticSearchSettings = null;
+ public Builder withHttpPort(int httpPort) {
+ this.httpPort = httpPort;
+ return this;
+ }
+ public Builder withIndexDir(File indexDir) {
+ this.indexDir = indexDir;
+ return this;
+ }
+ public Builder withExtraElasticSearchSettings(Map<String, String> extraElasticSearchSettings) {
+ this.extraElasticSearchSettings = extraElasticSearchSettings;
+ return this;
+ }
+ public ElasticSearchComponent build() {
+ return new ElasticSearchComponent(httpPort, indexDir, extraElasticSearchSettings);
+ }
+ }
+
+ private Client client;
+ private Node node;
+ private int httpPort;
+ private File indexDir;
+ private Map<String, String> extraElasticSearchSettings;
+
+ public ElasticSearchComponent(int httpPort, File indexDir) {
+ this(httpPort, indexDir, null);
+ }
+ public ElasticSearchComponent(int httpPort, File indexDir, Map<String, String> extraElasticSearchSettings) {
+ this.httpPort = httpPort;
+ this.indexDir = indexDir;
+ this.extraElasticSearchSettings = extraElasticSearchSettings;
+ }
+ public Client getClient() {
+ return client;
+ }
+
+ private void cleanDir(File dir) throws IOException {
+ if(dir.exists()) {
+ FileUtils.deleteDirectory(dir);
+ }
+ dir.mkdirs();
+ }
+ public void start() throws UnableToStartException {
+ File logDir= new File(indexDir, "/logs");
+ File dataDir= new File(indexDir, "/data");
+ try {
+ cleanDir(logDir);
+ cleanDir(dataDir);
+
+ } catch (IOException e) {
+ throw new UnableToStartException("Unable to clean log or data directories", e);
+ }
+ ImmutableSettings.Builder immutableSettings = ImmutableSettings.settingsBuilder()
+ .put("node.http.enabled", true)
+ .put("http.port", httpPort)
+ .put("cluster.name", "metron")
+ .put("path.logs",logDir.getAbsolutePath())
+ .put("path.data",dataDir.getAbsolutePath())
+ .put("gateway.type", "none")
+ .put("index.store.type", "memory")
+ .put("index.number_of_shards", 1)
+ .put("node.mode", "network")
+ .put("index.number_of_replicas", 1);
+ if(extraElasticSearchSettings != null) {
+ immutableSettings = immutableSettings.put(extraElasticSearchSettings);
+ }
+ Settings settings = immutableSettings.build();
+ node = NodeBuilder.nodeBuilder().settings(settings).node();
+ node.start();
+ settings = ImmutableSettings.settingsBuilder()
+ .put("cluster.name", "metron").build();
+ client = new TransportClient(settings)
+ .addTransportAddress(new InetSocketTransportAddress("localhost",
+ 9300));
+
+ waitForCluster(client, ClusterHealthStatus.YELLOW, new TimeValue(60000));
+ }
+
+ public static void waitForCluster(ElasticsearchClient client, ClusterHealthStatus status, TimeValue timeout) throws UnableToStartException {
+ try {
+ ClusterHealthResponse healthResponse =
+ (ClusterHealthResponse)client.execute(ClusterHealthAction.INSTANCE, new ClusterHealthRequest().waitForStatus(status).timeout(timeout)).actionGet();
+ if (healthResponse != null && healthResponse.isTimedOut()) {
+ throw new UnableToStartException("cluster state is " + healthResponse.getStatus().name()
+ + " and not " + status.name()
+ + ", from here on, everything will fail!");
+ }
+ } catch (ElasticsearchTimeoutException e) {
+ throw new UnableToStartException("timeout, cluster does not respond to health request, cowardly refusing to continue with operations");
+ }
+ }
+
+ public List<Map<String, Object>> getAllIndexedDocs(String index, String sourceType) throws IOException {
+ return getAllIndexedDocs(index, sourceType, null);
+ }
+ public List<Map<String, Object>> getAllIndexedDocs(String index, String sourceType, String subMessage) throws IOException {
+ getClient().admin().indices().refresh(new RefreshRequest());
+ SearchResponse response = getClient().prepareSearch(index)
+ .setTypes(sourceType)
+ .setSource("message")
+ .setFrom(0)
+ .setSize(1000)
+ .execute().actionGet();
+ List<Map<String, Object>> ret = new ArrayList<Map<String, Object>>();
+ for (SearchHit hit : response.getHits()) {
+ Object o = null;
+ if(subMessage == null) {
+ o = hit.getSource();
+ }
+ else {
+ o = hit.getSource().get(subMessage);
+ }
+ ret.add((Map<String, Object>)(o));
+ }
+ return ret;
+ }
+ public boolean hasIndex(String indexName) {
+ Set<String> indices = getClient().admin()
+ .indices()
+ .stats(new IndicesStatsRequest())
+ .actionGet()
+ .getIndices()
+ .keySet();
+ return indices.contains(indexName);
+
+ }
+
+ public void stop() {
+ node.stop();
+ node = null;
+ client = null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/resources/log4j.properties b/metron-platform/metron-elasticsearch/src/test/resources/log4j.properties
new file mode 100644
index 0000000..0d50388
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/test/resources/log4j.properties
@@ -0,0 +1,24 @@
+# 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.
+
+# Root logger option
+log4j.rootLogger=ERROR, stdout
+
+# Direct log messages to stdout
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.Target=System.out
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-elasticsearch/src/test/resources/log4j2.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/resources/log4j2.xml b/metron-platform/metron-elasticsearch/src/test/resources/log4j2.xml
new file mode 100755
index 0000000..68d5eac
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/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/0117987e/metron-platform/metron-enrichment/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/README.md b/metron-platform/metron-enrichment/README.md
new file mode 100644
index 0000000..83d2926
--- /dev/null
+++ b/metron-platform/metron-enrichment/README.md
@@ -0,0 +1,125 @@
+#metron-enrichment
+
+##Module Description
+
+This module enables enrichment of message metafields with additional information from various enrichment sources. Currently there is only a limited number of enrichments available, but this is an extensible framework that can be extended with additional enrichments. Enrichments currently available are geo, whois, hosts, and CIF.
+
+##Message Format
+
+Enrichment bolts are designed to go after the parser bolts. Parser bolts will parse the telemetry, taking it from its native format and producing a standard JSON that would look like so:
+
+```json
+{
+"message":
+{"ip_src_addr": xxxx,
+"ip_dst_addr": xxxx,
+"ip_src_port": xxxx,
+"ip_dst_port": xxxx,
+"protocol": xxxx,
+"additional-field 1": xxx,
+}
+
+}
+```
+
+A single enrichment bolt would enrich the message and produce a JSON enrichment and attach it to the message. Enrichments are stackable so multiple enrichments can be attached sequentially after a single parser bolt. Stacked enrichments would produce messages under the "enrichment" tag and attach it to the message like so:
+
+```json
+{
+"message":
+{"ip_src_addr": xxxx,
+"ip_dst_addr": xxxx,
+"ip_src_port": xxxx,
+"ip_dst_port": xxxx,
+"protocol": xxxx,
+"additional-field 1": xxxx,
+},
+"enrichment" : {"geo": xxxx, "whois": xxxx, "hosts": xxxxx, "CIF": "xxxxx"}
+
+}
+```
+
+##Enrichment Sources
+
+Each enrichment has to have an anrichment source which can serve as a lookup table for enriching relevant message fields. In order to minimize the use of additional platforms and tools we primarily try to rely on HBase as much as possible to store the enrichment information for lookup by key. In order to use Hbase we have to pre-process the enrichment feeds for bulk-loading into HBase with specific key format optimized for retrieval as well as utilize caches within the enrichment bolts to be able to provide enrichments real-time. Our wiki contains information on how to setup the environment, pre-process feeds, and plug in the enrichment sources.
+
+##Enrichment Bolt
+
+The enrichment bolt is designed to be extensible to be re-used for all kinds of enrichment processes. The bolt signature for declaration in a storm topology is as follows:
+
+
+
+```
+GenericEnrichmentBolt geo_enrichment = new GenericEnrichmentBolt()
+.withEnrichmentTag(
+config.getString("bolt.enrichment.geo.enrichment_tag"))
+.withAdapter(geo_adapter)
+.withMaxTimeRetain(
+config.getInt("bolt.enrichment.geo.MAX_TIME_RETAIN_MINUTES"))
+.withMaxCacheSize(
+config.getInt("bolt.enrichment.geo.MAX_CACHE_SIZE_OBJECTS_NUM"))
+.withKeys(geo_keys).withMetricConfiguration(config);
+
+```
+
+EnrichmentTag - Name of the enrichment (geo, whois, hosts, etc)
+Keys - Keys which this enrichment is able to enrich (hosts field for hosts enrichment, source_ip, dest_ip, for geo enrichment, etc)
+MaxTimeToRetain & MaxCacheSize - define the caching policy of the enrichment bolt
+Adapter - which adapter to use with the enrichment bolt instance
+
+###Geo Adapter
+Geo adapter is able to do geo enrichment on hosts and destination IPs. The open source verison of the geo adapter uses the free Geo feeds from MaxMind. The format of these feeds does not easily lend itself to a no-sql DB so this adapter is designed to work with mySql. But it is extensible enough to be made work with a variety of other back ends.
+
+The signature of a geo adapter is as follows;
+
+```
+GeoMysqlAdapter geo_adapter = new GeoMysqlAdapter(
+config.getString("mysql.ip"), config.getInt("mysql.port"),
+config.getString("mysql.username"),
+config.getString("mysql.password"),
+config.getString("bolt.enrichment.geo.adapter.table"));
+
+```
+
+###Hosts Adapter
+The hosts adapter is designed to enrich message format with the static host information that can be read from a standard text file. This adapter is intended for use with a network crawling script that can identify all customer assets and place them in a text file. For example, this script would identify all workstations, printers, appliantces, etc. Then if any of these assets are seen in the telemetry messages flowing through the adapter this enrichment would fire and the relevant known information about a host would be attached. We are currently working on porting this adapter to work with HBase, but this work is not ready yet. The known hosts file is located under the /etc/whitelists config directory of Metron.
+
+The signature of the hosts adapter is as follows:
+
+```
+Map<String, JSONObject> known_hosts = SettingsLoader
+.loadKnownHosts(hosts_path);
+
+HostFromPropertiesFileAdapter host_adapter = new HostFromPropertiesFileAdapter(
+known_hosts);
+
+```
+* The source and dest ips refer to the name of the message JSON key where the host information is located
+
+###Whois Adapter
+Whois adapter enriches the host name with additional whois information obtained from our proprietary Cisco feed. The enricher itself is provided in this open source distribution, but the feed is not. You have to have your own feed in order to use it. Alternatively, you can contact us for providing you with this feed, but we would have to charge you a fee (we can't distribute it for free). The implemetation of the whois enrichment we provide works with HBase
+
+The signature of the whois adapter is as follows:
+
+```
+
+EnrichmentAdapter whois_adapter = new WhoisHBaseAdapter(
+config.getString("bolt.enrichment.whois.hbase.table.name"),
+config.getString("kafka.zk.list"),
+config.getString("kafka.zk.port"));
+```
+
+###CIF Adapter
+CIF adapter is designed to take in CIF feeds and cross-reference them against every message processed by Storm. If there is a hit then the relevant information is attached to the message.
+
+The signature of the CIF adapter is as follows:
+
+```
+CIFHbaseAdapter = new CIFHbaseAdapter(config
+.getString("kafka.zk.list"), config
+.getString("kafka.zk.port"), config
+.getString("bolt.enrichment.cif.tablename")))
+```
+
+##Stacking Enrichments
+Enrichments can be stacked. By default each enrichment bolt listens on the "message" stream. In order to create and stack enrichment bolts create a new bolt and instantiate the appropariate adapter. You can look at our sample topologies to see how enrichments can be stacked
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-enrichment/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/pom.xml b/metron-platform/metron-enrichment/pom.xml
new file mode 100644
index 0000000..f6244a0
--- /dev/null
+++ b/metron-platform/metron-enrichment/pom.xml
@@ -0,0 +1,255 @@
+<?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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-platform</artifactId>
+ <version>0.1BETA</version>
+ </parent>
+ <artifactId>metron-enrichment</artifactId>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+ <mysql.version>5.1.31</mysql.version>
+ <slf4j.version>1.7.7</slf4j.version>
+ <storm.hdfs.version>0.1.2</storm.hdfs.version>
+ <guava.version>${global_hbase_guava_version}</guava.version>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-common</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-hbase</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>${slf4j.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>${mysql.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${global_hbase_version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <version>${global_hadoop_version}</version>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ </exclusions>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ <version>${global_storm_version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <groupId>org.slf4j</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-hdfs</artifactId>
+ <version>${global_storm_version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${global_guava_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${global_hadoop_version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>${global_mockito_version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>commons-validator</groupId>
+ <artifactId>commons-validator</artifactId>
+ <version>1.4.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-test-utilities</artifactId>
+ <version>0.1BETA</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <reporting>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemProperties>
+ <property>
+ <name>mode</name>
+ <value>global</value>
+ </property>
+ </systemProperties>
+ </configuration>
+ </plugin>
+ <!-- Normally, dependency report takes time, skip it -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-project-info-reports-plugin</artifactId>
+ <version>2.7</version>
+
+ <configuration>
+ <dependencyLocationsEnabled>false</dependencyLocationsEnabled>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>emma-maven-plugin</artifactId>
+ <version>1.0-alpha-3</version>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-pmd-plugin</artifactId>
+ <configuration>
+ <targetJdk>1.7</targetJdk>
+ </configuration>
+ </plugin>
+ </plugins>
+ </reporting>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version>
+ <configuration>
+ <source>1.7</source>
+ <target>1.7</target>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>1.4</version>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <artifactSet>
+ <excludes>
+ <exclude>*slf4j*</exclude>
+ </excludes>
+ </artifactSet>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <relocations>
+ <relocation>
+ <pattern>com.google.common</pattern>
+ <shadedPattern>org.apache.metron.guava</shadedPattern>
+ </relocation>
+ </relocations>
+ <transformers>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.DontIncludeResourceTransformer">
+ <resource>.yaml</resource>
+ </transformer>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass></mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <configuration>
+ <descriptor>src/main/assembly/assembly.xml</descriptor>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id> <!-- this is used for inheritance merges -->
+ <phase>package</phase> <!-- bind to the packaging phase -->
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-enrichment/src/main/assembly/assembly.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/assembly/assembly.xml b/metron-platform/metron-enrichment/src/main/assembly/assembly.xml
new file mode 100644
index 0000000..796dcc4
--- /dev/null
+++ b/metron-platform/metron-enrichment/src/main/assembly/assembly.xml
@@ -0,0 +1,44 @@
+<!--
+ 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.
+ -->
+
+<assembly>
+ <id>archive</id>
+ <formats>
+ <format>tar.gz</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${project.basedir}/src/main/scripts</directory>
+ <outputDirectory>/scripts</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ <excludes>
+ <exclude>**/*.formatted</exclude>
+ <exclude>**/*.filtered</exclude>
+ </excludes>
+ <fileMode>0755</fileMode>
+ <lineEnding>unix</lineEnding>
+ </fileSet>
+ <fileSet>
+ <directory>${project.basedir}/src/main/flux</directory>
+ <outputDirectory>/flux</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ <excludes>
+ <exclude>**/*.formatted</exclude>
+ <exclude>**/*.filtered</exclude>
+ </excludes>
+ <fileMode>0644</fileMode>
+ <lineEnding>unix</lineEnding>
+ </fileSet>
+ </fileSets>
+</assembly>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/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
new file mode 100644
index 0000000..b499b24
--- /dev/null
+++ b/metron-platform/metron-enrichment/src/main/flux/enrichment/remote.yaml
@@ -0,0 +1,413 @@
+# 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.
+
+name: "enrichment"
+config:
+ topology.workers: 1
+ topology.acker.executors: 0
+
+components:
+# Enrichment
+ - id: "jdbcConfig"
+ className: "org.apache.metron.enrichment.adapters.jdbc.MySqlConfig"
+ properties:
+ - name: "host"
+ value: "${mysql.ip}"
+ - name: "port"
+ value: ${mysql.port}
+ - name: "username"
+ value: "${mysql.username}"
+ - name: "password"
+ value: "${mysql.password}"
+ - name: "table"
+ value: "GEO"
+ - id: "geoEnrichmentAdapter"
+ className: "org.apache.metron.enrichment.adapters.geo.GeoAdapter"
+ configMethods:
+ - name: "withJdbcConfig"
+ args:
+ - ref: "jdbcConfig"
+ - id: "geoEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "geo"
+ - ref: "geoEnrichmentAdapter"
+ - id: "hostEnrichmentAdapter"
+ className: "org.apache.metron.enrichment.adapters.host.HostFromJSONListAdapter"
+ constructorArgs:
+ - '${org.apache.metron.enrichment.host.known_hosts}'
+ - id: "hostEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "host"
+ - ref: "hostEnrichmentAdapter"
+
+ - id: "simpleHBaseEnrichmentConfig"
+ className: "org.apache.metron.enrichment.adapters.simplehbase.SimpleHBaseConfig"
+ configMethods:
+ - name: "withProviderImpl"
+ args:
+ - "${hbase.provider.impl}"
+ - name: "withHBaseTable"
+ args:
+ - "${enrichment.simple.hbase.table}"
+ - name: "withHBaseCF"
+ args:
+ - "${enrichment.simple.hbase.cf}"
+ - id: "simpleHBaseEnrichmentAdapter"
+ className: "org.apache.metron.enrichment.adapters.simplehbase.SimpleHBaseAdapter"
+ configMethods:
+ - name: "withConfig"
+ args:
+ - ref: "simpleHBaseEnrichmentConfig"
+ - id: "simpleHBaseEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "hbaseEnrichment"
+ - ref: "simpleHBaseEnrichmentAdapter"
+ - id: "enrichments"
+ className: "java.util.ArrayList"
+ configMethods:
+ - name: "add"
+ args:
+ - ref: "geoEnrichment"
+ - name: "add"
+ args:
+ - ref: "hostEnrichment"
+ - name: "add"
+ args:
+ - ref: "simpleHBaseEnrichment"
+# Threat Intel
+
+ - id: "simpleHBaseThreatIntelConfig"
+ className: "org.apache.metron.enrichment.adapters.threatintel.ThreatIntelConfig"
+ configMethods:
+ - name: "withProviderImpl"
+ args:
+ - "${hbase.provider.impl}"
+ - name: "withTrackerHBaseTable"
+ args:
+ - "${threat.intel.tracker.table}"
+ - name: "withTrackerHBaseCF"
+ args:
+ - "${threat.intel.tracker.cf}"
+ - name: "withHBaseTable"
+ args:
+ - "${threat.intel.simple.hbase.table}"
+ - name: "withHBaseCF"
+ args:
+ - "${threat.intel.simple.hbase.cf}"
+ - id: "simpleHBaseThreatIntelAdapter"
+ className: "org.apache.metron.enrichment.adapters.threatintel.ThreatIntelAdapter"
+ configMethods:
+ - name: "withConfig"
+ args:
+ - ref: "simpleHBaseThreatIntelConfig"
+ - id: "simpleHBaseThreatIntelEnrichment"
+ className: "org.apache.metron.enrichment.configuration.Enrichment"
+ constructorArgs:
+ - "hbaseThreatIntel"
+ - ref: "simpleHBaseThreatIntelAdapter"
+
+ - id: "threatIntels"
+ className: "java.util.ArrayList"
+ configMethods:
+ - name: "add"
+ args:
+ - ref: "simpleHBaseThreatIntelEnrichment"
+
+ - id: "fileNameFormat"
+ className: "org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat"
+ configMethods:
+ - name: "withPrefix"
+ args:
+ - "enrichment-"
+ - name: "withExtension"
+ args:
+ - ".json"
+ - name: "withPath"
+ args:
+ - "${index.hdfs.output}"
+
+ - id: "hdfsRotationPolicy"
+ className: "${bolt.hdfs.rotation.policy}"
+ constructorArgs:
+ - ${bolt.hdfs.rotation.policy.count}
+ - "${bolt.hdfs.rotation.policy.units}"
+#indexing
+ - id: "hdfsWriter"
+ className: "org.apache.metron.writer.hdfs.HdfsWriter"
+ configMethods:
+ - name: "withFileNameFormat"
+ args:
+ - ref: "fileNameFormat"
+ - name: "withRotationPolicy"
+ args:
+ - ref: "hdfsRotationPolicy"
+
+ - id: "indexWriter"
+ className: "${writer.class.name}"
+
+#kafka/zookeeper
+ - id: "zkHosts"
+ className: "storm.kafka.ZkHosts"
+ constructorArgs:
+ - "${kafka.zk}"
+ - id: "kafkaConfig"
+ className: "storm.kafka.SpoutConfig"
+ constructorArgs:
+ # zookeeper hosts
+ - ref: "zkHosts"
+ # topic name
+ - "enrichments"
+ # zk root
+ - ""
+ # id
+ - "enrichments"
+ properties:
+ - name: "ignoreZkOffsets"
+ value: true
+ - name: "startOffsetTime"
+ value: -1
+
+spouts:
+ - id: "kafkaSpout"
+ className: "storm.kafka.KafkaSpout"
+ constructorArgs:
+ - ref: "kafkaConfig"
+bolts:
+# Enrichment Bolts
+ - id: "enrichmentSplitBolt"
+ className: "org.apache.metron.enrichment.bolt.EnrichmentSplitterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichments"
+ args:
+ - ref: "enrichments"
+ - id: "geoEnrichmentBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "geoEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
+ - id: "hostEnrichmentBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "hostEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
+ - id: "simpleHBaseEnrichmentBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "simpleHBaseEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
+ - id: "enrichmentJoinBolt"
+ className: "org.apache.metron.enrichment.bolt.EnrichmentJoinBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
+
+# Threat Intel Bolts
+ - id: "threatIntelSplitBolt"
+ className: "org.apache.metron.enrichment.bolt.ThreatIntelSplitterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichments"
+ args:
+ - ref: "threatIntels"
+ - name: "withMessageFieldName"
+ args: ["message"]
+ - id: "simpleHBaseThreatIntelBolt"
+ className: "org.apache.metron.enrichment.bolt.GenericEnrichmentBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withEnrichment"
+ args:
+ - ref: "simpleHBaseThreatIntelEnrichment"
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
+ - id: "threatIntelJoinBolt"
+ className: "org.apache.metron.enrichment.bolt.ThreatIntelJoinBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withMaxCacheSize"
+ args: [10000]
+ - name: "withMaxTimeRetain"
+ args: [10]
+# Indexing Bolts
+ - id: "indexingBolt"
+ className: "org.apache.metron.enrichment.bolt.BulkMessageWriterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withBulkMessageWriter"
+ args:
+ - ref: "indexWriter"
+ - id: "hdfsIndexingBolt"
+ className: "org.apache.metron.enrichment.bolt.BulkMessageWriterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ configMethods:
+ - name: "withBulkMessageWriter"
+ args:
+ - ref: "hdfsWriter"
+
+
+streams:
+#parser
+ - name: "spout -> enrichmentSplit"
+ from: "kafkaSpout"
+ to: "enrichmentSplitBolt"
+ grouping:
+ type: SHUFFLE
+
+#enrichment
+ - name: "enrichmentSplit -> host"
+ from: "enrichmentSplitBolt"
+ to: "hostEnrichmentBolt"
+ grouping:
+ streamId: "host"
+ type: FIELDS
+ args: ["key"]
+ - name: "enrichmentSplit -> geo"
+ from: "enrichmentSplitBolt"
+ to: "geoEnrichmentBolt"
+ grouping:
+ streamId: "geo"
+ type: FIELDS
+ args: ["key"]
+
+ - name: "enrichmentSplit -> simpleHBaseEnrichmentBolt"
+ from: "enrichmentSplitBolt"
+ to: "simpleHBaseEnrichmentBolt"
+ grouping:
+ streamId: "hbaseEnrichment"
+ type: FIELDS
+ args: ["key"]
+
+ - name: "splitter -> join"
+ from: "enrichmentSplitBolt"
+ to: "enrichmentJoinBolt"
+ grouping:
+ streamId: "message"
+ type: FIELDS
+ args: ["key"]
+ - name: "geo -> join"
+ from: "geoEnrichmentBolt"
+ to: "enrichmentJoinBolt"
+ grouping:
+ streamId: "geo"
+ type: FIELDS
+ args: ["key"]
+
+
+ - name: "simpleHBaseEnrichmentBolt -> join"
+ from: "simpleHBaseEnrichmentBolt"
+ to: "enrichmentJoinBolt"
+ grouping:
+ streamId: "hbaseEnrichment"
+ type: FIELDS
+ args: ["key"]
+ - name: "host -> join"
+ from: "hostEnrichmentBolt"
+ to: "enrichmentJoinBolt"
+ grouping:
+ streamId: "host"
+ type: FIELDS
+ args: ["key"]
+
+#threat intel
+ - name: "enrichmentJoin -> threatSplit"
+ from: "enrichmentJoinBolt"
+ to: "threatIntelSplitBolt"
+ grouping:
+ streamId: "message"
+ type: FIELDS
+ args: ["key"]
+
+ - name: "threatSplit -> simpleHBaseThreatIntel"
+ from: "threatIntelSplitBolt"
+ to: "simpleHBaseThreatIntelBolt"
+ grouping:
+ streamId: "hbaseThreatIntel"
+ type: FIELDS
+ args: ["key"]
+
+ - name: "simpleHBaseThreatIntel -> join"
+ from: "simpleHBaseThreatIntelBolt"
+ to: "threatIntelJoinBolt"
+ grouping:
+ streamId: "hbaseThreatIntel"
+ type: FIELDS
+ args: ["key"]
+ - name: "threatIntelSplit -> threatIntelJoin"
+ from: "threatIntelSplitBolt"
+ to: "threatIntelJoinBolt"
+ grouping:
+ streamId: "message"
+ type: FIELDS
+ args: ["key"]
+#indexing
+ - name: "threatIntelJoin -> indexing"
+ from: "threatIntelJoinBolt"
+ to: "indexingBolt"
+ grouping:
+ streamId: "message"
+ type: FIELDS
+ args: ["key"]
+
+ - name: "threatIntelJoin -> hdfs"
+ from: "threatIntelJoinBolt"
+ to: "hdfsIndexingBolt"
+ grouping:
+ streamId: "message"
+ type: SHUFFLE
+
+ - name: "indexingBolt -> errorIndexingBolt"
+ from: "indexingBolt"
+ to: "indexingBolt"
+ grouping:
+ streamId: "error"
+ type: SHUFFLE