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:45:56 UTC
[08/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-test-utilities/src/main/java/org/apache/metron/test/spouts/GenericInternalTestSpout.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/spouts/GenericInternalTestSpout.java b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/spouts/GenericInternalTestSpout.java
new file mode 100644
index 0000000..a36d99d
--- /dev/null
+++ b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/spouts/GenericInternalTestSpout.java
@@ -0,0 +1,139 @@
+/**
+ * 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.test.spouts;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
+import org.apache.metron.test.converters.BinaryConverters;
+import org.apache.metron.test.converters.IConverter;
+import org.apache.metron.test.filereaders.FileReader;
+
+import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseRichSpout;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Values;
+import backtype.storm.utils.Utils;
+
+
+public class GenericInternalTestSpout extends BaseRichSpout {
+
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = -2379344923143372543L;
+
+ List<String> jsons;
+
+ private String _filename;
+ private int _delay = 100;
+ private boolean _repeating = true;
+
+ private SpoutOutputCollector _collector;
+ private IConverter _converter;
+ private FileReader Reader;
+ private int cnt = 0;
+
+ public GenericInternalTestSpout withFilename(String filename)
+ {
+ if(filename != null && filename.length() > 0 && filename.charAt(0) == '$') {
+ filename = Iterables.getLast(Splitter.on("}").split(filename));
+ }
+ _filename = filename;
+ return this;
+ }
+ public GenericInternalTestSpout withMillisecondDelay(Integer delay)
+ {
+ _delay = delay;
+ return this;
+ }
+
+ public GenericInternalTestSpout withRepeating(Boolean repeating)
+ {
+ _repeating = repeating;
+ return this;
+ }
+
+ public GenericInternalTestSpout withBinaryConverter(String converter) {
+ if(converter == null) {
+ _converter = BinaryConverters.DEFAULT;
+ }
+ else {
+ _converter = BinaryConverters.valueOf(converter);
+ }
+ return this;
+ }
+
+
+ @SuppressWarnings("rawtypes")
+ public void open(Map conf, TopologyContext context,
+ SpoutOutputCollector collector) {
+
+ _collector = collector;
+ try {
+ Reader = new FileReader();
+ jsons = Reader.readFromFile(_filename);
+
+ } catch (Throwable e)
+ {
+ System.out.println("Could not read sample JSONs");
+ e.printStackTrace();
+ }
+
+ }
+
+ public void nextTuple() {
+ Utils.sleep(_delay);
+
+ if(cnt < jsons.size())
+ {
+ byte[] value;
+ if (_converter != null) {
+ value = _converter.convert(jsons.get(cnt));
+ } else {
+ value = jsons.get(cnt).getBytes();
+ }
+ _collector.emit(new Values(value));
+ }
+ cnt ++;
+
+ if(_repeating && cnt == jsons.size() -1 )
+ cnt = 0;
+ }
+
+ @Override
+ public void ack(Object id) {
+ }
+
+ @Override
+ public void fail(Object id) {
+ }
+
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("message"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/KafkaLoader.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/KafkaLoader.java b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/KafkaLoader.java
new file mode 100644
index 0000000..294f8fd
--- /dev/null
+++ b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/KafkaLoader.java
@@ -0,0 +1,86 @@
+/**
+ * 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.test.utils;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.util.HashMap;
+import java.util.Map;
+
+public class KafkaLoader {
+
+ private String brokerUrl;
+ private String topic;
+ private String samplePath;
+ private int delay = 1000;
+ private int iterations = -1;
+ private KafkaProducer kafkaProducer;
+
+ public KafkaLoader(String brokerUrl, String topic, String samplePath) {
+ this.brokerUrl = brokerUrl;
+ this.topic = topic;
+ this.samplePath = samplePath;
+ }
+
+ public KafkaLoader withDelay(int delay) {
+ this.delay = delay;
+ return this;
+ }
+
+ public KafkaLoader withIterations(int iterations) {
+ this.iterations = iterations;
+ return this;
+ }
+
+ public void start() {
+ Map<String, Object> producerConfig = new HashMap<>();
+ producerConfig.put("bootstrap.servers", brokerUrl);
+ producerConfig.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ producerConfig.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ kafkaProducer = new KafkaProducer<>(producerConfig);
+ try {
+ while (iterations == -1 || iterations-- > 0) {
+ BufferedReader reader = new BufferedReader(new FileReader(samplePath));
+ String line;
+ while((line = reader.readLine()) != null) {
+ kafkaProducer.send(new ProducerRecord<String, String>(topic, line));
+ Thread.sleep(delay);
+ }
+ reader.close();
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ public void stop() {
+ kafkaProducer.close();
+ }
+
+
+ public static void main(String[] args) {
+ KafkaLoader kafkaLoader = new KafkaLoader(args[0], args[1], args[2]);
+ if (args.length > 3) kafkaLoader.withDelay(Integer.parseInt(args[3]));
+ if (args.length > 4) kafkaLoader.withIterations(Integer.parseInt(args[4]));
+ kafkaLoader.start();
+ kafkaLoader.stop();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/UnitTestHelper.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/UnitTestHelper.java b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/UnitTestHelper.java
new file mode 100644
index 0000000..aff48aa
--- /dev/null
+++ b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/utils/UnitTestHelper.java
@@ -0,0 +1,84 @@
+/**
+ * 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.test.utils;
+
+import org.apache.log4j.ConsoleAppender;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.log4j.PatternLayout;
+import org.junit.Assert;
+
+import java.io.File;
+import java.util.Set;
+import java.util.Stack;
+
+public class UnitTestHelper {
+ public static String findDir(String name) {
+ return findDir(new File("."), name);
+ }
+
+ public static String findDir(File startDir, String name) {
+ Stack<File> s = new Stack<File>();
+ s.push(startDir);
+ while(!s.empty()) {
+ File parent = s.pop();
+ if(parent.getName().equalsIgnoreCase(name)) {
+ return parent.getAbsolutePath();
+ }
+ else {
+ File[] children = parent.listFiles();
+ if(children != null) {
+ for (File child : children) {
+ s.push(child);
+ }
+ }
+ }
+ }
+ return null;
+ }
+
+ public static <T> void assertSetEqual(String type, Set<T> expectedPcapIds, Set<T> found) {
+ boolean mismatch = false;
+ for(T f : found) {
+ if(!expectedPcapIds.contains(f)) {
+ mismatch = true;
+ System.out.println("Found " + type + " that I did not expect: " + f);
+ }
+ }
+ for(T expectedId : expectedPcapIds) {
+ if(!found.contains(expectedId)) {
+ mismatch = true;
+ System.out.println("Expected " + type + " that I did not index: " + expectedId);
+ }
+ }
+ Assert.assertFalse(mismatch);
+ }
+
+ public static void verboseLogging() {
+ verboseLogging("%d [%p|%c|%C{1}] %m%n", Level.ALL);
+ }
+ public static void verboseLogging(String pattern, Level level) {
+ ConsoleAppender console = new ConsoleAppender(); //create appender
+ //configure the appender
+ console.setLayout(new PatternLayout(pattern));
+ console.setThreshold(level);
+ console.activateOptions();
+ //add appender to any Logger (here is root)
+ Logger.getRootLogger().addAppender(console);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/pom.xml b/metron-platform/pom.xml
new file mode 100644
index 0000000..12ef866
--- /dev/null
+++ b/metron-platform/pom.xml
@@ -0,0 +1,265 @@
+<?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>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-platform</artifactId>
+ <version>0.1BETA</version>
+ <packaging>pom</packaging>
+ <name>metron-platform</name>
+ <description>Stream analytics for Metron</description>
+ <url>https://metron.incubator.apache.org/</url>
+ <scm>
+ <connection>scm:git:https://git-wip-us.apache.org/repos/asf/incubator-metron.git</connection>
+ <developerConnection>scm:git:https://git-wip-us.apache.org/repos/asf/incubator-metron.git</developerConnection>
+ <tag>HEAD</tag>
+ <url>https://git-wip-us.apache.org/repos/asf/incubator-metron</url>
+ </scm>
+ <properties>
+ <twitter>@ApacheMetron</twitter>
+ <global_opencsv_version>3.7</global_opencsv_version>
+ <global_storm_version>0.10.0</global_storm_version>
+ <global_flux_version>0.10.0</global_flux_version>
+ <global_pcap_version>1.7.1</global_pcap_version>
+ <global_kafka_version>0.8.2.2</global_kafka_version>
+ <global_hadoop_version>2.7.1</global_hadoop_version>
+ <global_hbase_version>1.1.1</global_hbase_version>
+ <global_flume_version>1.5.2</global_flume_version>
+ <global_elasticsearch_version>1.7.4</global_elasticsearch_version>
+ <global_json_simple_version>1.1.1</global_json_simple_version>
+ <global_metrics_version>3.0.2</global_metrics_version>
+ <global_junit_version>4.4</global_junit_version>
+ <global_guava_version>17.0</global_guava_version>
+ <global_hbase_guava_version>12.0</global_hbase_guava_version>
+ <global_json_schema_validator_version>2.2.5</global_json_schema_validator_version>
+ <global_slf4j_version>1.7.7</global_slf4j_version>
+ <global_opencsv_version>3.7</global_opencsv_version>
+ <global_solr_version>5.2.1</global_solr_version>
+ <global_mockito_version>1.10.19</global_mockito_version>
+ </properties>
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ <distribution>repo</distribution>
+ </license>
+ </licenses>
+ <modules>
+ <module>metron-common</module>
+ <module>metron-enrichment</module>
+ <module>metron-elasticsearch</module>
+ <module>metron-solr</module>
+ <module>metron-parsers</module>
+ <module>metron-data-management</module>
+ <module>metron-pcap</module>
+ <module>metron-integration-test</module>
+ <module>metron-test-utilities</module>
+ <module>metron-api</module>
+ <module>metron-hbase</module>
+ </modules>
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${global_mockito_version}</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.12</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.adrianwalker</groupId>
+ <artifactId>multiline-string</artifactId>
+ <version>0.1.2</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <pluginManagement>
+ <plugins>
+ <plugin>
+ <!-- Separates the unit tests from the integration tests. -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.18</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>
+ </plugins>
+ </pluginManagement>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <executions>
+ <execution>
+ <id>checkstyle</id>
+ <phase>package</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <configLocation>style/checkstyle.xml</configLocation>
+ <headerLocation>style/LICENSE</headerLocation>
+ <failOnViolation>true</failOnViolation>
+ <includeTestSourceDirectory>true</includeTestSourceDirectory>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.5.1</version>
+ <configuration>
+ <source>1.7</source>
+ <target>1.7</target>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>cobertura-maven-plugin</artifactId>
+ <version>2.7</version>
+ <configuration>
+ <check />
+ <formats>
+ <format>html</format>
+ </formats>
+ <aggregate>true</aggregate>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <version>0.11</version>
+ <configuration>
+ <excludes>
+ <exclude>**/README.md</exclude>
+ <exclude>**/*.json</exclude>
+ <exclude>**/*.log</exclude>
+ <exclude>**/src/main/resources/patterns/**</exclude>
+ <exclude>**/src/main/resources/sample/data/SampleIndexed/**</exclude>
+ <exclude>**/src/main/resources/sample/data/SampleInput/**</exclude>
+ <exclude>**/src/main/resources/sample/data/SampleParsed/**</exclude>
+ <exclude>**/dependency-reduced-pom.xml</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ <reporting>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.18</version>
+ <configuration>
+ <argLine>-Xmx2048m -XX:MaxPermSize=256m</argLine>
+ <systemProperties>
+ <property>
+ <name>mode</name>
+ <value>local</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.apache.maven.plugins</groupId>
+ <artifactId>maven-pmd-plugin</artifactId>
+ <version>3.3</version>
+ <configuration>
+ <targetJdk>1.7</targetJdk>
+ </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>
+ <repositories>
+ <repository>
+ <id>clojars.org</id>
+ <url>http://clojars.org/repo</url>
+ </repository>
+ <repository>
+ <id>multiline-release-repo</id>
+ <url>https://raw.github.com/benelog/multiline/master/maven-repository</url>
+ <snapshots>
+ <enabled>false</enabled>
+ </snapshots>
+ </repository>
+ </repositories>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/style/LICENSE.config
----------------------------------------------------------------------
diff --git a/metron-platform/style/LICENSE.config b/metron-platform/style/LICENSE.config
new file mode 100644
index 0000000..826d578
--- /dev/null
+++ b/metron-platform/style/LICENSE.config
@@ -0,0 +1,16 @@
+# 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.
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/style/LICENSE.java
----------------------------------------------------------------------
diff --git a/metron-platform/style/LICENSE.java b/metron-platform/style/LICENSE.java
new file mode 100644
index 0000000..5d5f1e3
--- /dev/null
+++ b/metron-platform/style/LICENSE.java
@@ -0,0 +1,17 @@
+/**
+ * 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.
+ */
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/style/LICENSE.xml
----------------------------------------------------------------------
diff --git a/metron-platform/style/LICENSE.xml b/metron-platform/style/LICENSE.xml
new file mode 100644
index 0000000..05af984
--- /dev/null
+++ b/metron-platform/style/LICENSE.xml
@@ -0,0 +1,16 @@
+<!--
+ 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.
+-->
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/style/checkstyle.xml
----------------------------------------------------------------------
diff --git a/metron-platform/style/checkstyle.xml b/metron-platform/style/checkstyle.xml
new file mode 100644
index 0000000..5f8b5c5
--- /dev/null
+++ b/metron-platform/style/checkstyle.xml
@@ -0,0 +1,33 @@
+<?xml version="1.0"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ -->
+<!DOCTYPE module PUBLIC "-//Puppy Crawl//DTD Check Configuration 1.1//EN" "http://www.puppycrawl.com/dtds/configuration_1_1.dtd">
+
+<module name="Checker">
+ <property name="localeLanguage" value="en"/>
+
+ <!-- Verify that EVERY source file has the appropriate license -->
+ <module name="Header">
+ <property name="headerFile" value="style/LICENSE.java"/>
+ <property name="fileExtensions" value="java"/>
+ </module>
+ <module name="Header">
+ <property name="headerFile" value="style/LICENSE.config"/>
+ <property name="fileExtensions" value="properties, yaml"/>
+ </module>
+
+</module>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/README.md
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/README.md b/metron-streaming/Metron-Alerts/README.md
deleted file mode 100644
index 2b87d50..0000000
--- a/metron-streaming/Metron-Alerts/README.md
+++ /dev/null
@@ -1,104 +0,0 @@
-#Metron-Alerts
-
-##Module Description
-
-This module enables telemetry alerts. It splits the mssage stream into two streams. The original message is emitted on the "message" stream. The corresponding alert is emitted on the "alerts" stream. The two are tied together through the alerts UUID.
-
-##Message Format
-
-Assuming the original message (with enrichments enabled) has the following format:
-
-```json
-{
-"message":
-{"ip_src_addr": xxxx,
-"ip_dst_addr": xxxx,
-"ip_src_port": xxxx,
-"ip_dst_port": xxxx,
-"protocol": xxxx,
-"timestamp": xxxx.
-"original_string": xxxx,
-"additional-field 1": xxxx,
-},
-"enrichment" : {"geo": xxxx, "whois": xxxx, "hosts": xxxxx, "CIF": "xxxxx"}
-
-}
-```
-
-The telemetry message will be tagged with a UUID alert tag like so:
-
-```json
-{
-"message":
-{"ip_src_addr": xxxx,
-"ip_dst_addr": xxxx,
-"ip_src_port": xxxx,
-"ip_dst_port": xxxx,
-"protocol": xxxx,
-"timestamp": xxxx,
-"original_string": xxxx,
-"additional-field 1": xxxx,
-},
-"enrichment" : {"geo": xxxx, "whois": xxxx, "hosts": xxxxx, "CIF": "xxxxx"},
-"alerts": [UUID1, UUID2, UUID3, etc]
-
-}
-```
-
-The alert will be fired on the "alerts" stream and can be customized to have any format as long as it includes the required mandatory fields. The mandatory fields are:
-
-* timestamp (epoch): The time from the message that triggered the alert
-* description: A human friendly string representation of the alert
-* alert_id: The UUID generated for the alert. This uniquely identifies an alert
-
-There are other standard but not mandatory fields that can be leveraged by metron-ui and other alert consumers:
-
-* designated_host: The IP address that corresponds to an asset. Ex. The IP address of the company device associated with the alert.
-* enrichment: A copy of the enrichment data from the message that triggered the alert
-* priority: The priority of the alert. Mustb e set to one of HIGH, MED or LOW
-
-An example of an alert with all mandatory and standard fields would look like so:
-
-```json
-{
-"timestamp": xxxx,
-"alert_id": UUID,
-"description": xxxx,
-"designated_host": xxxx,
-"enrichment": { "geo": xxxx, "whois": xxxx, "cif": xxxx },
-"priority": "MED"
-}
-```
-
-##Alerts Bolt
-
-The bolt can be extended with a variety of alerts adapters. The ability to stack alerts is currently in beta, but is not currently advisable. We advice to only have one alerts bolt per topology. The adapters are rules-based adapters which fire alerts when rules are a match. Currently only Java adapters are provided, but there are future plans to provide Grok-Based adapters as well.
-
-The signature of the Alerts bolt is as follows:
-
-```
-TelemetryAlertsBolt alerts_bolt = new TelemetryAlertsBolt()
-.withIdentifier(alerts_identifier).withMaxCacheSize(1000)
-.withMaxTimeRetain(3600).withAlertsAdapter(alerts_adapter)
-.withMetricConfiguration(config);
-```
-Identifier - JSON key where the alert is attached
-TimeRetain & MaxCacheSize - Caching parameters for the bolt
-MetricConfiguration - export custom bolt metrics to graphite (if not null)
-AlertsAdapter - pick the appropriate adapter for generating the alerts
-
-### Java Adapters
-
-Java adapters are designed for high volume topologies, but are not easily extensible. The adapters provided are:
-
-* org.apache.metron.alerts.adapters.AllAlertsAdapter - will tag every single message with the static alert (appropriate for topologies like Sourcefire, etc, where every single message is an alert)
-* org.apache.metron.alerts.adapters.HbaseWhiteAndBlacklistAdapter - will read white and blacklists from HBase and fire alerts if source or dest IP are not on the whitelist or if any IP is on the blacklist
-* org.apache.metron.alerts.adapters.CIFAlertsAdapter - will alert on messages that have results in enrichment.cif.
-* org.apache.metron.alerts.adpaters.KeywordsAlertAdapter - will alert on messages that contain any of a list of keywords
-###Grok Adapters
-
-Grok alerts adapters for Metron are still under devleopment
-
-###Stacking Alert Adapters
-
-The functionality to stack alerts adapters is still under development
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/pom.xml
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/pom.xml b/metron-streaming/Metron-Alerts/pom.xml
deleted file mode 100644
index ec921ee..0000000
--- a/metron-streaming/Metron-Alerts/pom.xml
+++ /dev/null
@@ -1,169 +0,0 @@
-<?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-Streaming</artifactId>
- <version>0.1BETA</version>
- </parent>
- <artifactId>Metron-Alerts</artifactId>
- <name>Metron-Alerts</name>
- <description>Taggers for alerts</description>
- <properties>
- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
- <commons.validator.version>1.4.0</commons.validator.version>
- </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-Common</artifactId>
- <version>${project.parent.version}</version>
- </dependency>
- <dependency>
- <groupId>com.googlecode.json-simple</groupId>
- <artifactId>json-simple</artifactId>
- <version>${global_json_simple_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.kafka</groupId>
- <artifactId>kafka_2.9.2</artifactId>
- <version>${global_kafka_version}</version>
- <scope>provided</scope>
- <exclusions>
- <exclusion>
- <groupId>com.sun.jmx</groupId>
- <artifactId>jmxri</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.sun.jdmk</groupId>
- <artifactId>jmxtools</artifactId>
- </exclusion>
- <exclusion>
- <groupId>javax.jms</groupId>
- <artifactId>jms</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase-client</artifactId>
- <version>${global_hbase_version}</version>
- <scope>provided</scope>
- <exclusions>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>com.codahale.metrics</groupId>
- <artifactId>metrics-core</artifactId>
- <version>${global_metrics_version}</version>
- </dependency>
- <dependency>
- <groupId>commons-validator</groupId>
- <artifactId>commons-validator</artifactId>
- <version>${commons.validator.version}</version>
- <exclusions>
- <exclusion>
-
- <groupId>commons-beanutils</groupId>
-
- <artifactId>commons-beanutils</artifactId>
-
- </exclusion>
- </exclusions>
- </dependency>
- </dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <version>2.18</version>
- <configuration>
- <systemProperties>
- <property>
- <name>mode</name>
- <value>local</value>
- </property>
- </systemProperties>
- </configuration>
- </plugin>
- <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-pmd-plugin</artifactId>
- <version>3.3</version>
- <configuration>
- <targetJdk>1.7</targetJdk>
- </configuration>
- </plugin>
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>emma-maven-plugin</artifactId>
- <version>1.0-alpha-3</version>
- <inherited>true</inherited>
- </plugin>
- </plugins>
- <resources>
- <resource>
- <directory>src/main/resources</directory>
- </resource>
- </resources>
- </build>
-</project>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/AbstractAlertBolt.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/AbstractAlertBolt.java b/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/AbstractAlertBolt.java
deleted file mode 100644
index f482c2a..0000000
--- a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/AbstractAlertBolt.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/**
- * 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.alerts;
-
-import java.io.IOException;
-import java.util.Map;
-
-import org.json.simple.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import backtype.storm.task.OutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.topology.base.BaseRichBolt;
-import backtype.storm.tuple.Fields;
-
-import com.codahale.metrics.Counter;
-import com.google.common.cache.Cache;
-import org.apache.metron.alerts.interfaces.AlertsAdapter;
-import org.apache.metron.metrics.MetricReporter;
-
-@SuppressWarnings("rawtypes")
-public abstract class AbstractAlertBolt extends BaseRichBolt {
- /**
- *
- */
- private static final long serialVersionUID = -6710596708304282838L;
-
- transient Cache<String, String> cache;
-
- protected static final Logger LOG = LoggerFactory
- .getLogger(AbstractAlertBolt.class);
-
- protected OutputCollector _collector;
- protected AlertsAdapter _adapter;
-
- protected String OutputFieldName;
- protected JSONObject _identifier;
- protected MetricReporter _reporter;
-
- protected int _MAX_CACHE_SIZE_OBJECTS_NUM = -1;
- protected int _MAX_TIME_RETAIN_MINUTES = -1;
-
- protected Counter ackCounter, emitCounter, failCounter;
-
- protected void registerCounters() {
-
- String ackString = _adapter.getClass().getSimpleName() + ".ack";
-
- String emitString = _adapter.getClass().getSimpleName() + ".emit";
-
- String failString = _adapter.getClass().getSimpleName() + ".fail";
-
- ackCounter = _reporter.registerCounter(ackString);
- emitCounter = _reporter.registerCounter(emitString);
- failCounter = _reporter.registerCounter(failString);
-
- }
-
- public final void prepare(Map conf, TopologyContext topologyContext,
- OutputCollector collector) {
- _collector = collector;
-
- if (this._adapter == null)
- throw new IllegalStateException("Alerts adapter must be specified");
- if (this._identifier == null)
- throw new IllegalStateException("Identifier must be specified");
-
- if (this._MAX_CACHE_SIZE_OBJECTS_NUM == -1)
- throw new IllegalStateException("MAX_CACHE_SIZE_OBJECTS_NUM must be specified");
- if (this._MAX_TIME_RETAIN_MINUTES == -1)
- throw new IllegalStateException("MAX_TIME_RETAIN_MINUTES must be specified");
-
- try {
- doPrepare(conf, topologyContext, collector);
- } catch (IOException e) {
- LOG.error("Counld not initialize...");
- e.printStackTrace();
- }
-
- boolean success = _adapter.initialize();
-
- try {
- if (!success)
-
- throw new Exception("Could not initialize adapter");
- } catch (Exception e) {
-
- e.printStackTrace();
- }
- }
-
- public void declareOutputFields(OutputFieldsDeclarer declearer) {
- declearer.declareStream("message", new Fields("key", "message"));
- declearer.declareStream("alert", new Fields( "message"));
- declearer.declareStream("error", new Fields("message"));
- }
-
- abstract void doPrepare(Map conf, TopologyContext topologyContext,
- OutputCollector collector) throws IOException;
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/TelemetryAlertsBolt.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/TelemetryAlertsBolt.java b/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/TelemetryAlertsBolt.java
deleted file mode 100644
index 663ae40..0000000
--- a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/TelemetryAlertsBolt.java
+++ /dev/null
@@ -1,254 +0,0 @@
-/**
- * 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.alerts;
-
-import java.io.IOException;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.commons.configuration.Configuration;
-import org.json.simple.JSONArray;
-import org.json.simple.JSONObject;
-
-import backtype.storm.task.OutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.tuple.Tuple;
-import backtype.storm.tuple.Values;
-
-import com.google.common.cache.CacheBuilder;
-import org.apache.metron.alerts.interfaces.AlertsAdapter;
-import org.apache.metron.helpers.topology.ErrorUtils;
-import org.apache.metron.json.serialization.JSONEncoderHelper;
-import org.apache.metron.metrics.MetricReporter;
-
-@SuppressWarnings("rawtypes")
-public class TelemetryAlertsBolt extends AbstractAlertBolt {
-
- /**
- * Use an adapter to tag existing telemetry messages with alerts. The list
- * of available tagger adapters is located under
- * org.apache.metron.tagging.adapters. At the time of the release the following
- * adapters are available:
- *
- * <p>
- * <ul>
- * <li>RegexTagger = read a list or regular expressions and tag a message if
- * they exist in a message
- * <li>StaticAllTagger = tag each message with a static alert
- * <ul>
- * <p>
- */
- private static final long serialVersionUID = -2647123143398352020L;
- private Properties metricProperties;
- private JSONObject metricConfiguration;
-
- // private AlertsCache suppressed_alerts;
-
- /**
- *
- * @param tagger
- * - tagger adapter for generating alert tags
- * @return instance of bolt
- */
- public TelemetryAlertsBolt withAlertsAdapter(AlertsAdapter tagger) {
- _adapter = tagger;
- return this;
- }
-
- /**
- *
- * @param OutputFieldName
- * - output name of the tuple coming out of this bolt
- * @return - instance of this bolt
- */
- public TelemetryAlertsBolt withOutputFieldName(String OutputFieldName) {
- this.OutputFieldName = OutputFieldName;
- return this;
- }
-
- /**
- *
- * @param metricProperties
- * - metric output to graphite
- * @return - instance of this bolt
- */
- public TelemetryAlertsBolt withMetricProperties(Properties metricProperties) {
- this.metricProperties = metricProperties;
- return this;
- }
-
- /**
- *
- * @param identifier
- * - the identifier tag for tagging telemetry messages with
- * alerts out of this bolt
- * @return - instance of this bolt
- */
-
- public TelemetryAlertsBolt withIdentifier(JSONObject identifier) {
- this._identifier = identifier;
- return this;
- }
-
- /**
- * @param config
- * A class for generating custom metrics into graphite
- * @return Instance of this class
- */
-
- public TelemetryAlertsBolt withMetricConfiguration(Configuration config) {
- this.metricConfiguration = JSONEncoderHelper.getJSON(config
- .subset("org.apache.metron.metrics"));
- return this;
- }
-
- /**
- * @param MAX_CACHE_SIZE_OBJECTS_NUM
- * Maximum size of cache before flushing
- * @return Instance of this class
- */
-
- public TelemetryAlertsBolt withMaxCacheSize(int MAX_CACHE_SIZE_OBJECTS_NUM) {
- _MAX_CACHE_SIZE_OBJECTS_NUM = MAX_CACHE_SIZE_OBJECTS_NUM;
- return this;
- }
-
- /**
- * @param MAX_TIME_RETAIN_MINUTES
- * Maximum time to retain cached entry before expiring
- * @return Instance of this class
- */
-
- public TelemetryAlertsBolt withMaxTimeRetain(int MAX_TIME_RETAIN_MINUTES) {
- _MAX_TIME_RETAIN_MINUTES = MAX_TIME_RETAIN_MINUTES;
- return this;
- }
-
- @Override
- void doPrepare(Map conf, TopologyContext topologyContext,
- OutputCollector collector) throws IOException {
-
- cache = CacheBuilder.newBuilder().maximumSize(_MAX_CACHE_SIZE_OBJECTS_NUM)
- .expireAfterWrite(_MAX_TIME_RETAIN_MINUTES, TimeUnit.MINUTES).build();
-
- LOG.info("[Metron] Preparing TelemetryAlert Bolt...");
-
- try {
- _reporter = new MetricReporter();
- _reporter.initialize(metricProperties, TelemetryAlertsBolt.class);
- LOG.info("[Metron] Initialized metrics");
- } catch (Exception e) {
- LOG.info("[Metron] Could not initialize metrics");
- }
- }
-
- @SuppressWarnings("unchecked")
- public void execute(Tuple tuple) {
-
- LOG.trace("[Metron] Starting to process message for alerts");
- JSONObject original_message = null;
- String key = null;
-
- try {
-
- key = tuple.getStringByField("key");
- original_message = (JSONObject) tuple.getValueByField("message");
-
- if (original_message == null || original_message.isEmpty())
- throw new Exception("Could not parse message from byte stream");
-
- if(key == null)
- throw new Exception("Key is not valid");
-
- LOG.trace("[Metron] Received tuple: " + original_message);
-
- JSONObject alerts_tag = new JSONObject();
- Map<String, JSONObject> alerts_list = _adapter
- .alert(original_message);
- JSONArray uuid_list = new JSONArray();
-
- if (alerts_list == null || alerts_list.isEmpty()) {
- System.out.println("[Metron] No alerts detected in: "
- + original_message);
- _collector.ack(tuple);
- _collector.emit("message", new Values(key, original_message));
- } else {
- for (String alert : alerts_list.keySet()) {
- uuid_list.add(alert);
-
- LOG.trace("[Metron] Checking alerts cache: " + alert);
-
- if (cache.getIfPresent(alert) == null) {
- System.out.println("[Metron]: Alert not found in cache: " + alert);
-
- JSONObject global_alert = new JSONObject();
- global_alert.putAll(_identifier);
- global_alert.putAll(alerts_list.get(alert));
- global_alert.put("timestamp", System.currentTimeMillis());
- _collector.emit("alert", new Values(global_alert));
-
- cache.put(alert, "");
-
- } else
- LOG.trace("Alert located in cache: " + alert);
-
- LOG.debug("[Metron] Alerts are: " + alerts_list);
-
- if (original_message.containsKey("alerts")) {
- JSONArray already_triggered = (JSONArray) original_message
- .get("alerts");
-
- uuid_list.addAll(already_triggered);
- LOG.trace("[Metron] Messages already had alerts...tagging more");
- }
-
- original_message.put("alerts", uuid_list);
-
- LOG.debug("[Metron] Detected alerts: " + alerts_tag);
-
- _collector.ack(tuple);
- _collector.emit("message", new Values(key, original_message));
-
- }
-
- /*
- * if (metricConfiguration != null) { emitCounter.inc();
- * ackCounter.inc(); }
- */
- }
-
- } catch (Exception e) {
- e.printStackTrace();
- LOG.error("Failed to tag message :" + original_message);
- e.printStackTrace();
- _collector.fail(tuple);
-
- /*
- * if (metricConfiguration != null) { failCounter.inc(); }
- */
-
-
- JSONObject error = ErrorUtils.generateErrorMessage(
- "Alerts problem: " + original_message, e);
- _collector.emit("error", new Values(error));
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AbstractAlertAdapter.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AbstractAlertAdapter.java b/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AbstractAlertAdapter.java
deleted file mode 100644
index 6c1b8d1..0000000
--- a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AbstractAlertAdapter.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/**
- * 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.alerts.adapters;
-
-import java.io.Serializable;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import org.apache.metron.alerts.interfaces.AlertsAdapter;
-
-@SuppressWarnings("serial")
-public abstract class AbstractAlertAdapter implements AlertsAdapter, Serializable{
-
- protected static final Logger _LOG = LoggerFactory
- .getLogger(AbstractAlertAdapter.class);
-
-
- protected Cache<String, String> cache;
-
- protected String generateAlertId(String source_ip, String dst_ip,
- int alert_type) {
-
- String key = makeKey(source_ip, dst_ip, alert_type);
-
- if (cache.getIfPresent(key) != null)
- return cache.getIfPresent(key);
-
- String new_UUID = System.currentTimeMillis() + "-" + UUID.randomUUID();
-
- cache.put(key, new_UUID);
- key = makeKey(dst_ip, source_ip, alert_type);
- cache.put(key, new_UUID);
-
- return new_UUID;
-
- }
-
- private String makeKey(String ip1, String ip2, int alert_type) {
- return (ip1 + "-" + ip2 + "-" + alert_type);
- }
-
- protected void generateCache(int _MAX_CACHE_SIZE_OBJECTS_NUM, int _MAX_TIME_RETAIN_MINUTES)
- {
- cache = CacheBuilder.newBuilder().maximumSize(_MAX_CACHE_SIZE_OBJECTS_NUM)
- .expireAfterWrite(_MAX_TIME_RETAIN_MINUTES, TimeUnit.MINUTES).build();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AllAlertAdapter.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AllAlertAdapter.java b/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AllAlertAdapter.java
deleted file mode 100644
index b527991..0000000
--- a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/AllAlertAdapter.java
+++ /dev/null
@@ -1,292 +0,0 @@
-/**
- * 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.alerts.adapters;
-
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.commons.validator.routines.InetAddressValidator;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.HBaseAdmin;
-import org.apache.hadoop.hbase.client.HConnection;
-import org.apache.hadoop.hbase.client.HConnectionManager;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.json.simple.JSONObject;
-import org.apache.log4j.Logger;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import org.apache.metron.alerts.interfaces.AlertsAdapter;
-
-@SuppressWarnings("serial")
-public class AllAlertAdapter implements AlertsAdapter, Serializable {
-
- HTableInterface blacklist_table;
- HTableInterface whitelist_table;
- InetAddressValidator ipvalidator = new InetAddressValidator();
- String _whitelist_table_name;
- String _blacklist_table_name;
- String _quorum;
- String _port;
- String _topologyname;
- Configuration conf = null;
-
- Cache<String, String> cache;
- String _topology_name;
-
- Set<String> loaded_whitelist = new HashSet<String>();
- Set<String> loaded_blacklist = new HashSet<String>();
-
- protected static final Logger LOG = Logger
- .getLogger(AllAlertAdapter.class);
-
- public AllAlertAdapter(Map<String, String> config) {
- try {
- if(!config.containsKey("whitelist_table_name"))
- throw new Exception("Whitelist table name is missing");
-
- _whitelist_table_name = config.get("whitelist_table_name");
-
- if(!config.containsKey("blacklist_table_name"))
- throw new Exception("Blacklist table name is missing");
-
- _blacklist_table_name = config.get("blacklist_table_name");
-
- if(!config.containsKey("quorum"))
- throw new Exception("Quorum name is missing");
-
- _quorum = config.get("quorum");
-
- if(!config.containsKey("port"))
- throw new Exception("port name is missing");
-
- _port = config.get("port");
-
- if(!config.containsKey("_MAX_CACHE_SIZE_OBJECTS_NUM"))
- throw new Exception("_MAX_CACHE_SIZE_OBJECTS_NUM name is missing");
-
- int _MAX_CACHE_SIZE_OBJECTS_NUM = Integer.parseInt(config
- .get("_MAX_CACHE_SIZE_OBJECTS_NUM"));
-
- if(!config.containsKey("_MAX_TIME_RETAIN_MINUTES"))
- throw new Exception("_MAX_TIME_RETAIN_MINUTES name is missing");
-
- int _MAX_TIME_RETAIN_MINUTES = Integer.parseInt(config
- .get("_MAX_TIME_RETAIN_MINUTES"));
-
- cache = CacheBuilder.newBuilder().maximumSize(_MAX_CACHE_SIZE_OBJECTS_NUM)
- .expireAfterWrite(_MAX_TIME_RETAIN_MINUTES, TimeUnit.MINUTES)
- .build();
- } catch (Exception e) {
- System.out.println("Could not initialize Alerts Adapter");
- e.printStackTrace();
- System.exit(0);
- }
- }
-
- @SuppressWarnings("resource")
- @Override
- public boolean initialize() {
-
- conf = HBaseConfiguration.create();
- //conf.set("hbase.zookeeper.quorum", _quorum);
- //conf.set("hbase.zookeeper.property.clientPort", _port);
-
- LOG.trace("[Metron] Connecting to hbase with conf:" + conf);
- LOG.trace("[Metron] Whitelist table name: " + _whitelist_table_name);
- LOG.trace("[Metron] Whitelist table name: " + _blacklist_table_name);
- LOG.trace("[Metron] ZK Client/port: "
- + conf.get("hbase.zookeeper.quorum") + " -> "
- + conf.get("hbase.zookeeper.property.clientPort"));
-
- try {
-
- LOG.trace("[Metron] Attempting to connect to hbase");
-
- HConnection connection = HConnectionManager.createConnection(conf);
-
- LOG.trace("[Metron] CONNECTED TO HBASE");
-
- HBaseAdmin hba = new HBaseAdmin(conf);
-
- if (!hba.tableExists(_whitelist_table_name))
- throw new Exception("Whitelist table doesn't exist");
-
- if (!hba.tableExists(_blacklist_table_name))
- throw new Exception("Blacklist table doesn't exist");
-
- whitelist_table = new HTable(conf, _whitelist_table_name);
-
- LOG.trace("[Metron] CONNECTED TO TABLE: " + _whitelist_table_name);
- blacklist_table = new HTable(conf, _blacklist_table_name);
- LOG.trace("[Metron] CONNECTED TO TABLE: " + _blacklist_table_name);
-
- if (connection == null || whitelist_table == null
- || blacklist_table == null)
- throw new Exception("Unable to initialize hbase connection");
-
- Scan scan = new Scan();
-
- ResultScanner rs = whitelist_table.getScanner(scan);
- try {
- for (Result r = rs.next(); r != null; r = rs.next()) {
- loaded_whitelist.add(Bytes.toString(r.getRow()));
- }
- } catch (Exception e) {
- LOG.trace("[Metron] COULD NOT READ FROM HBASE");
- e.printStackTrace();
- } finally {
- rs.close(); // always close the ResultScanner!
- hba.close();
- }
- whitelist_table.close();
-
- LOG.trace("[Metron] READ IN WHITELIST: " + loaded_whitelist.size());
-
- System.out.println("LOADED WHITELIST IS: ");
-
- for(String str: loaded_whitelist)
- System.out.println("WHITELIST: " + str);
-
- scan = new Scan();
-
- rs = blacklist_table.getScanner(scan);
- try {
- for (Result r = rs.next(); r != null; r = rs.next()) {
- loaded_blacklist.add(Bytes.toString(r.getRow()));
- }
- } catch (Exception e) {
- LOG.trace("[Metron] COULD NOT READ FROM HBASE");
- e.printStackTrace();
- } finally {
- rs.close(); // always close the ResultScanner!
- hba.close();
- }
- blacklist_table.close();
-
- LOG.trace("[Metron] READ IN WHITELIST: " + loaded_whitelist.size());
-
- rs.close(); // always close the ResultScanner!
- hba.close();
-
- return true;
- } catch (Exception e) {
-
- e.printStackTrace();
- }
-
- return false;
-
- }
-
- @Override
- public boolean refresh() throws Exception {
- // TODO Auto-generated method stub
- return false;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public Map<String, JSONObject> alert(JSONObject raw_message) {
-
- Map<String, JSONObject> alerts = new HashMap<String, JSONObject>();
- JSONObject content = (JSONObject) raw_message.get("message");
-
- JSONObject enrichment = null;
-
- if (raw_message.containsKey("enrichment"))
- enrichment = (JSONObject) raw_message.get("enrichment");
-
- JSONObject alert = new JSONObject();
-
-
-
- String source = "unknown";
- String dest = "unknown";
- String host = "unknown";
-
- if (content.containsKey("ip_src_addr"))
- {
- source = content.get("ip_src_addr").toString();
-
- if(RangeChecker.checkRange(loaded_whitelist, source))
- host = source;
- }
-
- if (content.containsKey("ip_dst_addr"))
- {
- dest = content.get("ip_dst_addr").toString();
-
- if(RangeChecker.checkRange(loaded_whitelist, dest))
- host = dest;
- }
-
- alert.put("designated_host", host);
- alert.put("description", content.get("original_string").toString());
- alert.put("priority", "MED");
-
- String alert_id = generateAlertId(source, dest, 0);
-
- alert.put("alert_id", alert_id);
- alerts.put(alert_id, alert);
-
- alert.put("enrichment", enrichment);
-
- return alerts;
-
- }
-
- @Override
- public boolean containsAlertId(String alert) {
- // TODO Auto-generated method stub
- return false;
- }
-
- protected String generateAlertId(String source_ip, String dst_ip,
- int alert_type) {
-
- String key = makeKey(source_ip, dst_ip, alert_type);
-
- if (cache.getIfPresent(key) != null)
- return cache.getIfPresent(key);
-
- String new_UUID = System.currentTimeMillis() + "-" + UUID.randomUUID();
-
- cache.put(key, new_UUID);
- key = makeKey(dst_ip, source_ip, alert_type);
- cache.put(key, new_UUID);
-
- return new_UUID;
-
- }
-
- private String makeKey(String ip1, String ip2, int alert_type) {
- return (ip1 + "-" + ip2 + "-" + alert_type);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/CIFAlertsAdapter.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/CIFAlertsAdapter.java b/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/CIFAlertsAdapter.java
deleted file mode 100644
index 4e8e025..0000000
--- a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/CIFAlertsAdapter.java
+++ /dev/null
@@ -1,328 +0,0 @@
-/**
- * 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.alerts.adapters;
-
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.commons.validator.routines.InetAddressValidator;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.HBaseAdmin;
-import org.apache.hadoop.hbase.client.HConnection;
-import org.apache.hadoop.hbase.client.HConnectionManager;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.json.simple.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import org.apache.metron.alerts.interfaces.AlertsAdapter;
-
-@SuppressWarnings("serial")
-public class CIFAlertsAdapter implements AlertsAdapter, Serializable {
-
- String enrichment_tag;
-
- HTableInterface blacklist_table;
- HTableInterface whitelist_table;
- InetAddressValidator ipvalidator = new InetAddressValidator();
- String _whitelist_table_name;
- String _blacklist_table_name;
- String _quorum;
- String _port;
- String _topologyname;
- Configuration conf = null;
-
- Cache<String, String> cache;
- String _topology_name;
-
- Set<String> loaded_whitelist = new HashSet<String>();
- Set<String> loaded_blacklist = new HashSet<String>();
-
- protected static final Logger LOG = LoggerFactory
- .getLogger(CIFAlertsAdapter.class);
-
- public CIFAlertsAdapter(Map<String, String> config) {
- try {
-
- if (!config.containsKey("whitelist_table_name"))
- throw new Exception("Whitelist table name is missing");
-
- _whitelist_table_name = config.get("whitelist_table_name");
-
- if (!config.containsKey("blacklist_table_name"))
- throw new Exception("Blacklist table name is missing");
-
- _blacklist_table_name = config.get("blacklist_table_name");
-
- if (!config.containsKey("quorum"))
- throw new Exception("Quorum name is missing");
-
- _quorum = config.get("quorum");
-
- if (!config.containsKey("port"))
- throw new Exception("port name is missing");
-
- _port = config.get("port");
-
- if (!config.containsKey("_MAX_CACHE_SIZE_OBJECTS_NUM"))
- throw new Exception("_MAX_CACHE_SIZE_OBJECTS_NUM name is missing");
-
- int _MAX_CACHE_SIZE_OBJECTS_NUM = Integer.parseInt(config
- .get("_MAX_CACHE_SIZE_OBJECTS_NUM"));
-
- if (!config.containsKey("_MAX_TIME_RETAIN_MINUTES"))
- throw new Exception("_MAX_TIME_RETAIN_MINUTES name is missing");
-
- int _MAX_TIME_RETAIN_MINUTES = Integer.parseInt(config
- .get("_MAX_TIME_RETAIN_MINUTES"));
-
- cache = CacheBuilder.newBuilder().maximumSize(_MAX_CACHE_SIZE_OBJECTS_NUM)
- .expireAfterWrite(_MAX_TIME_RETAIN_MINUTES, TimeUnit.MINUTES)
- .build();
-
- enrichment_tag = config.get("enrichment_tag");
-
- } catch (Exception e) {
- System.out.println("Could not initialize alerts adapter");
- e.printStackTrace();
- System.exit(0);
- }
- }
-
- @SuppressWarnings("resource")
- @Override
- public boolean initialize() {
-
- conf = HBaseConfiguration.create();
- // conf.set("hbase.zookeeper.quorum", _quorum);
- // conf.set("hbase.zookeeper.property.clientPort", _port);
-
- LOG.trace("[Metron] Connecting to hbase with conf:" + conf);
- LOG.trace("[Metron] Whitelist table name: " + _whitelist_table_name);
- LOG.trace("[Metron] Whitelist table name: " + _blacklist_table_name);
- LOG.trace("[Metron] ZK Client/port: "
- + conf.get("hbase.zookeeper.quorum") + " -> "
- + conf.get("hbase.zookeeper.property.clientPort"));
-
- try {
-
- LOG.trace("[Metron] Attempting to connect to hbase");
-
- HConnection connection = HConnectionManager.createConnection(conf);
-
- LOG.trace("[Metron] CONNECTED TO HBASE");
-
- HBaseAdmin hba = new HBaseAdmin(conf);
-
- if (!hba.tableExists(_whitelist_table_name))
- throw new Exception("Whitelist table doesn't exist");
-
- if (!hba.tableExists(_blacklist_table_name))
- throw new Exception("Blacklist table doesn't exist");
-
- whitelist_table = new HTable(conf, _whitelist_table_name);
-
- LOG.trace("[Metron] CONNECTED TO TABLE: " + _whitelist_table_name);
- blacklist_table = new HTable(conf, _blacklist_table_name);
- LOG.trace("[Metron] CONNECTED TO TABLE: " + _blacklist_table_name);
-
- if (connection == null || whitelist_table == null
- || blacklist_table == null)
- throw new Exception("Unable to initialize hbase connection");
-
- Scan scan = new Scan();
-
- ResultScanner rs = whitelist_table.getScanner(scan);
- try {
- for (Result r = rs.next(); r != null; r = rs.next()) {
- loaded_whitelist.add(Bytes.toString(r.getRow()));
- }
- } catch (Exception e) {
- LOG.trace("[Metron] COULD NOT READ FROM HBASE");
- e.printStackTrace();
- } finally {
- rs.close(); // always close the ResultScanner!
- hba.close();
- }
- whitelist_table.close();
-
- LOG.trace("[Metron] READ IN WHITELIST: " + loaded_whitelist.size());
-
- scan = new Scan();
-
- rs = blacklist_table.getScanner(scan);
- try {
- for (Result r = rs.next(); r != null; r = rs.next()) {
- loaded_blacklist.add(Bytes.toString(r.getRow()));
- }
- } catch (Exception e) {
- LOG.trace("[Metron] COULD NOT READ FROM HBASE");
- e.printStackTrace();
- } finally {
- rs.close(); // always close the ResultScanner!
- hba.close();
- }
- blacklist_table.close();
-
- LOG.trace("[Metron] READ IN WHITELIST: " + loaded_whitelist.size());
-
- rs.close(); // always close the ResultScanner!
- hba.close();
-
- return true;
- } catch (Exception e) {
-
- e.printStackTrace();
- }
-
- return false;
-
- }
-
- @Override
- public boolean refresh() throws Exception {
- return true;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public Map<String, JSONObject> alert(JSONObject raw_message) {
-
- System.out.println("LOOKING FOR ENRICHMENT TAG: " + enrichment_tag);
-
- Map<String, JSONObject> alerts = new HashMap<String, JSONObject>();
- JSONObject content = (JSONObject) raw_message.get("message");
-
- JSONObject enrichment = null;
-
- if (raw_message.containsKey("enrichment"))
- enrichment = (JSONObject) raw_message.get("enrichment");
- else
- return null;
-
- if (enrichment.containsKey(enrichment_tag)) {
-
- System.out.println("FOUND TAG: " + enrichment_tag);
-
- JSONObject cif = (JSONObject) enrichment.get(enrichment_tag);
-
- int cnt = 0;
- Object enriched_key = null;
-
- for (Object key : cif.keySet()) {
- JSONObject tmp = (JSONObject) cif.get(key);
- cnt = cnt + tmp.size();
- if (tmp.size() > 0)
- enriched_key = key;
- }
-
- if (cnt == 0) {
- System.out.println("TAG HAS NO ELEMENTS");
- return null;
- }
-
- JSONObject alert = new JSONObject();
-
- String source = "unknown";
- String dest = "unknown";
- String host = "unknown";
-
- if (content.containsKey("ip_src_addr")) {
- source = content.get("ip_src_addr").toString();
-
- if (RangeChecker.checkRange(loaded_whitelist, source))
- host = source;
- }
-
- if (content.containsKey("ip_dst_addr")) {
- dest = content.get("ip_dst_addr").toString();
-
- if (RangeChecker.checkRange(loaded_whitelist, dest))
- host = dest;
- }
-
- JSONObject cifQualifier = (JSONObject) cif.get(enriched_key);
-
- alert.put("designated_host", host);
- String description = new StringBuilder()
- .append(host)
- .append(" communicated with a host (")
- .append(content.get(enriched_key).toString())
- .append(") identified as ")
- .append(cifQualifier.keySet().iterator().next().toString())
- .append(" by CIF")
- .toString();
- alert.put("description", description);
- alert.put("priority", "MED");
-
- String alert_id = generateAlertId(source, dest, 0);
-
- alert.put("alert_id", alert_id);
- alerts.put(alert_id, alert);
-
- alert.put("enrichment", enrichment);
-
- return alerts;
- } else {
- System.out.println("DID NOT FIND TAG: " + enrichment_tag);
- return null;
- }
-
- }
-
- @Override
- public boolean containsAlertId(String alert) {
- // TODO Auto-generated method stub
- return false;
- }
-
- protected String generateAlertId(String source_ip, String dst_ip,
- int alert_type) {
-
- String key = makeKey(source_ip, dst_ip, alert_type);
-
- if (cache.getIfPresent(key) != null)
- return cache.getIfPresent(key);
-
- String new_UUID = System.currentTimeMillis() + "-" + UUID.randomUUID();
-
- cache.put(key, new_UUID);
- key = makeKey(dst_ip, source_ip, alert_type);
- cache.put(key, new_UUID);
-
- return new_UUID;
-
- }
-
- private String makeKey(String ip1, String ip2, int alert_type) {
- return (ip1 + "-" + ip2 + "-" + alert_type);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/HbaseWhiteAndBlacklistAdapter.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/HbaseWhiteAndBlacklistAdapter.java b/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/HbaseWhiteAndBlacklistAdapter.java
deleted file mode 100644
index 3673138..0000000
--- a/metron-streaming/Metron-Alerts/src/main/java/org/apache/metron/alerts/adapters/HbaseWhiteAndBlacklistAdapter.java
+++ /dev/null
@@ -1,483 +0,0 @@
-/**
- * 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.alerts.adapters;
-
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.commons.validator.routines.InetAddressValidator;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.HBaseAdmin;
-import org.apache.hadoop.hbase.client.HConnection;
-import org.apache.hadoop.hbase.client.HConnectionManager;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.json.simple.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import org.apache.metron.alerts.interfaces.AlertsAdapter;
-
-public class HbaseWhiteAndBlacklistAdapter implements AlertsAdapter,
- Serializable {
-
- HTableInterface blacklist_table;
- HTableInterface whitelist_table;
- InetAddressValidator ipvalidator = new InetAddressValidator();
- String _whitelist_table_name;
- String _blacklist_table_name;
- String _quorum;
- String _port;
- String _topologyname;
- Configuration conf = null;
-
- Cache<String, String> cache;
- String _topology_name;
-
- Set<String> loaded_whitelist = new HashSet<String>();
- Set<String> loaded_blacklist = new HashSet<String>();
-
- protected static final Logger LOG = LoggerFactory
- .getLogger(HbaseWhiteAndBlacklistAdapter.class);
-
- public HbaseWhiteAndBlacklistAdapter(Map<String, String> config) {
-
- try {
- if(!config.containsKey("whitelist_table_name"))
- throw new Exception("Whitelist table name is missing");
-
- _whitelist_table_name = config.get("whitelist_table_name");
-
- if(!config.containsKey("blacklist_table_name"))
- throw new Exception("Blacklist table name is missing");
-
- _blacklist_table_name = config.get("blacklist_table_name");
-
- if(!config.containsKey("quorum"))
- throw new Exception("Quorum name is missing");
-
- _quorum = config.get("quorum");
-
- if(!config.containsKey("port"))
- throw new Exception("port name is missing");
-
- _port = config.get("port");
-
- if(!config.containsKey("_MAX_CACHE_SIZE_OBJECTS_NUM"))
- throw new Exception("_MAX_CACHE_SIZE_OBJECTS_NUM name is missing");
-
- int _MAX_CACHE_SIZE_OBJECTS_NUM = Integer.parseInt(config
- .get("_MAX_CACHE_SIZE_OBJECTS_NUM"));
-
- if(!config.containsKey("_MAX_TIME_RETAIN_MINUTES"))
- throw new Exception("_MAX_TIME_RETAIN_MINUTES name is missing");
-
- int _MAX_TIME_RETAIN_MINUTES = Integer.parseInt(config
- .get("_MAX_TIME_RETAIN_MINUTES"));
-
- cache = CacheBuilder.newBuilder().maximumSize(_MAX_CACHE_SIZE_OBJECTS_NUM)
- .expireAfterWrite(_MAX_TIME_RETAIN_MINUTES, TimeUnit.MINUTES)
- .build();
- } catch (Exception e) {
- System.out.println("Could not initialize Alerts Adapter");
- e.printStackTrace();
- System.exit(0);
- }
-
- }
-
- public boolean initialize() {
-
- conf = HBaseConfiguration.create();
- //conf.set("hbase.zookeeper.quorum", _quorum);
- //conf.set("hbase.zookeeper.property.clientPort", _port);
-
- LOG.trace("[Metron] Connecting to hbase with conf:" + conf);
- LOG.trace("[Metron] Whitelist table name: " + _whitelist_table_name);
- LOG.trace("[Metron] Whitelist table name: " + _blacklist_table_name);
- LOG.trace("[Metron] ZK Client/port: "
- + conf.get("hbase.zookeeper.quorum") + " -> "
- + conf.get("hbase.zookeeper.property.clientPort"));
-
- try {
-
- LOG.trace("[Metron] Attempting to connect to hbase");
-
- HConnection connection = HConnectionManager.createConnection(conf);
-
- LOG.trace("[Metron] CONNECTED TO HBASE");
-
- HBaseAdmin hba = new HBaseAdmin(conf);
-
- if (!hba.tableExists(_whitelist_table_name))
- throw new Exception("Whitelist table doesn't exist");
-
- if (!hba.tableExists(_blacklist_table_name))
- throw new Exception("Blacklist table doesn't exist");
-
- whitelist_table = new HTable(conf, _whitelist_table_name);
-
- LOG.trace("[Metron] CONNECTED TO TABLE: " + _whitelist_table_name);
- blacklist_table = new HTable(conf, _blacklist_table_name);
- LOG.trace("[Metron] CONNECTED TO TABLE: " + _blacklist_table_name);
-
- if (connection == null || whitelist_table == null
- || blacklist_table == null)
- throw new Exception("Unable to initialize hbase connection");
-
- Scan scan = new Scan();
-
- ResultScanner rs = whitelist_table.getScanner(scan);
- try {
- for (Result r = rs.next(); r != null; r = rs.next()) {
- loaded_whitelist.add(Bytes.toString(r.getRow()));
- }
- } catch (Exception e) {
- LOG.trace("[Metron] COULD NOT READ FROM HBASE");
- e.printStackTrace();
- } finally {
- rs.close(); // always close the ResultScanner!
- hba.close();
- }
- whitelist_table.close();
-
- LOG.trace("[Metron] READ IN WHITELIST: " + loaded_whitelist.size());
-
- scan = new Scan();
-
- rs = blacklist_table.getScanner(scan);
- try {
- for (Result r = rs.next(); r != null; r = rs.next()) {
- loaded_blacklist.add(Bytes.toString(r.getRow()));
- }
- } catch (Exception e) {
- LOG.trace("[Metron] COULD NOT READ FROM HBASE");
- e.printStackTrace();
- } finally {
- rs.close(); // always close the ResultScanner!
- hba.close();
- }
- blacklist_table.close();
-
- LOG.trace("[Metron] READ IN WHITELIST: " + loaded_whitelist.size());
-
- rs.close(); // always close the ResultScanner!
- hba.close();
-
- return true;
- } catch (Exception e) {
-
- e.printStackTrace();
- }
-
- return false;
-
- }
-
- protected String generateAlertId(String source_ip, String dst_ip,
- int alert_type) {
-
- String key = makeKey(source_ip, dst_ip, alert_type);
-
- if (cache.getIfPresent(key) != null)
- return cache.getIfPresent(key);
-
- String new_UUID = System.currentTimeMillis() + "-" + UUID.randomUUID();
-
- cache.put(key, new_UUID);
- key = makeKey(dst_ip, source_ip, alert_type);
- cache.put(key, new_UUID);
-
- return new_UUID;
-
- }
-
- public boolean refresh() throws Exception {
- // TODO Auto-generated method stub
- return false;
- }
-
- private String makeKey(String ip1, String ip2, int alert_type) {
- return (ip1 + "-" + ip2 + "-" + alert_type);
- }
-
- @SuppressWarnings("unchecked")
- public Map<String, JSONObject> alert(JSONObject raw_message) {
-
- Map<String, JSONObject> alerts = new HashMap<String, JSONObject>();
-
- JSONObject content = (JSONObject) raw_message.get("message");
- JSONObject enrichment = null;
-
- if (raw_message.containsKey("enrichment"))
- enrichment = (JSONObject) raw_message.get("enrichment");
-
- if (!content.containsKey("ip_src_addr")
- || !content.containsKey("ip_dst_addr")) {
-
- int alert_type = 0;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", "Uknown");
- alert.put("source", "NA");
- alert.put("dest", "NA");
- alert.put("body", "Source or destination IP is missing");
-
- String alert_id = UUID.randomUUID().toString();
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
-
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- LOG.trace("[Metron] Returning alert: " + alerts);
-
- return alerts;
-
- }
-
- String source_ip = content.get("ip_src_addr").toString();
- String dst_ip = content.get("ip_dst_addr").toString();
-
- if (source_ip == null && dst_ip == null) {
-
- int alert_type = 1;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", "Uknown");
- alert.put("source", source_ip);
- alert.put("dest", dst_ip);
- alert.put(
- "body",
- "This communication does not contain a source or destination IP string. Communication between two IPs: "
- + source_ip + " -> " + dst_ip);
-
- String alert_id = generateAlertId(source_ip, dst_ip, alert_type);
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- LOG.trace("[Metron] Returning alert: " + alerts);
-
- return alerts;
-
- }
-
- if (!ipvalidator.isValidInet4Address(source_ip)
- && !ipvalidator.isValidInet4Address(dst_ip)) {
- int alert_type = 2;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", "Uknown");
- alert.put("source", source_ip);
- alert.put("dest", dst_ip);
- alert.put(
- "content",
- "This communication contains souce and destination IP strings, but these strings are not valid. Communication between two IPs: "
- + source_ip + " -> " + dst_ip);
-
- String alert_id = generateAlertId(source_ip, dst_ip, alert_type);
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- LOG.trace("[Metron] Returning alert: " + alerts);
-
- return alerts;
-
- }
-
- String designated_host = null;
-
- if (loaded_whitelist.contains(source_ip))
- designated_host = source_ip;
- else if (loaded_whitelist.contains(dst_ip))
- designated_host = dst_ip;
-
- if (designated_host == null) {
- int alert_type = 3;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", "Uknown");
- alert.put("source", source_ip);
- alert.put("dest", dst_ip);
- alert.put(
- "content",
- "This communication does not contain a source or a destination IP that is in the white list. Communication between two IPs: "
- + source_ip + " -> " + dst_ip);
-
- String alert_id = generateAlertId(source_ip, dst_ip, alert_type);
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- LOG.trace("[Metron] Returning alert: " + alerts);
-
- return alerts;
-
- }
-
- if (source_ip.equals(designated_host)
- && !ipvalidator.isValidInet4Address(dst_ip)) {
- int alert_type = 4;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", designated_host);
- alert.put("source", source_ip);
- alert.put("dest", dst_ip);
- alert.put(
- "content",
- "This communication contains an IP that is not valid. Communication between two IPs: "
- + source_ip + " -> " + dst_ip);
-
- String alert_id = generateAlertId(source_ip, dst_ip, alert_type);
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- }
-
- if (dst_ip.equals(designated_host)
- && !ipvalidator.isValidInet4Address(source_ip)) {
- int alert_type = 5;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", designated_host);
- alert.put("source", source_ip);
- alert.put("dest", dst_ip);
- alert.put(
- "content",
- "This communication contains IP that is not valid. Communication between two IPs: "
- + source_ip + " -> " + dst_ip);
-
- String alert_id = generateAlertId(source_ip, dst_ip, alert_type);
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- }
-
- if (loaded_blacklist.contains(source_ip)) {
- int alert_type = 6;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", designated_host);
- alert.put("source", source_ip);
- alert.put("dest", dst_ip);
- alert.put(
- "content",
- "This communication contains IP that is black listed. Communication between two IPs: "
- + source_ip + " -> " + dst_ip);
-
- String alert_id = generateAlertId(source_ip, dst_ip, alert_type);
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- }
-
- if (loaded_blacklist.contains(dst_ip)) {
- int alert_type = 7;
-
- JSONObject alert = new JSONObject();
-
- alert.put("title", "IP Check Error Type: : " + alert_type);
- alert.put("priority", "1");
- alert.put("type", "error");
- alert.put("designated_host", designated_host);
- alert.put("source", source_ip);
- alert.put("dest", dst_ip);
- alert.put(
- "content",
- "This communication contains IP that is black listed. Communication between two IPs: "
- + source_ip + " -> " + dst_ip);
-
- String alert_id = generateAlertId(source_ip, dst_ip, alert_type);
-
- alert.put("reference_id", alert_id);
- alerts.put(alert_id, alert);
- if (enrichment != null)
- alert.put("enrichment", enrichment);
-
- }
-
- if (alerts.isEmpty())
- return null;
- else
- return alerts;
- }
-
- public boolean containsAlertId(String alert) {
- // TODO Auto-generated method stub
- return false;
- }
-
-}