You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ot...@apache.org on 2018/01/25 14:05:36 UTC
[13/21] metron git commit: METRON-1230: As a stopgap prior to
METRON-777, add more simplistic sideloading of custom Parsers closes
apache/incubator-metron#785
METRON-1230: As a stopgap prior to METRON-777, add more simplistic sideloading of custom Parsers closes apache/incubator-metron#785
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/a262f87f
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/a262f87f
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/a262f87f
Branch: refs/heads/feature/METRON-1211-extensions-parsers-gradual
Commit: a262f87f0c31637d76a77a538976673c87188eeb
Parents: 5d3e73a
Author: cstella <ce...@gmail.com>
Authored: Tue Jan 16 12:37:59 2018 -0500
Committer: cstella <ce...@gmail.com>
Committed: Tue Jan 16 12:37:59 2018 -0500
----------------------------------------------------------------------
metron-interface/metron-rest/pom.xml | 3 +
.../metron/rest/MetronRestApplication.java | 2 +
.../impl/SensorParserConfigServiceImpl.java | 20 +-
.../apache/metron/rest/util/ParserIndex.java | 92 ++++++
.../metron-rest/src/main/scripts/metron-rest.sh | 17 +
.../metron-parsers/3rdPartyParser.md | 330 +++++++++++++++++++
metron-platform/metron-parsers/pom.xml | 6 +
.../topology/MergeAndShadeTransformer.java | 101 ++++++
.../parsers/topology/ParserTopologyCLI.java | 18 +-
.../src/main/scripts/start_parser_topology.sh | 9 +-
10 files changed, 576 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-interface/metron-rest/pom.xml
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/pom.xml b/metron-interface/metron-rest/pom.xml
index 302b784..44bad97 100644
--- a/metron-interface/metron-rest/pom.xml
+++ b/metron-interface/metron-rest/pom.xml
@@ -189,6 +189,7 @@
<groupId>org.apache.metron</groupId>
<artifactId>metron-parsers</artifactId>
<version>${project.parent.version}</version>
+ <scope>provided</scope>
<exclusions>
<exclusion>
<groupId>org.apache.hbase</groupId>
@@ -317,6 +318,7 @@
<groupId>org.apache.metron</groupId>
<artifactId>metron-indexing</artifactId>
<version>${project.parent.version}</version>
+ <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.metron</groupId>
@@ -417,6 +419,7 @@
<goal>shade</goal>
</goals>
<configuration>
+ <createDependencyReducedPom>false</createDependencyReducedPom>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.handlers</resource>
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/MetronRestApplication.java
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/MetronRestApplication.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/MetronRestApplication.java
index ac89b1d..5135849 100644
--- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/MetronRestApplication.java
+++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/MetronRestApplication.java
@@ -17,6 +17,7 @@
*/
package org.apache.metron.rest;
+import org.apache.metron.rest.util.ParserIndex;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -24,6 +25,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
public class MetronRestApplication {
public static void main(String[] args) {
+ ParserIndex.reload();
SpringApplication.run(MetronRestApplication.class, args);
}
}
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/SensorParserConfigServiceImpl.java
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/SensorParserConfigServiceImpl.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/SensorParserConfigServiceImpl.java
index c460e3c..85b84b8 100644
--- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/SensorParserConfigServiceImpl.java
+++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/SensorParserConfigServiceImpl.java
@@ -38,10 +38,12 @@ import org.apache.metron.rest.RestException;
import org.apache.metron.rest.model.ParseMessageRequest;
import org.apache.metron.rest.service.GrokService;
import org.apache.metron.rest.service.SensorParserConfigService;
+import org.apache.metron.rest.util.ParserIndex;
import org.apache.metron.common.zookeeper.ZKConfigurationsCache;
import org.apache.zookeeper.KeeperException;
import org.json.simple.JSONObject;
import org.reflections.Reflections;
+import org.reflections.util.ConfigurationBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@@ -118,29 +120,15 @@ public class SensorParserConfigServiceImpl implements SensorParserConfigService
@Override
public Map<String, String> getAvailableParsers() {
- if (availableParsers == null) {
- availableParsers = new HashMap<>();
- Set<Class<? extends MessageParser>> parserClasses = getParserClasses();
- parserClasses.forEach(parserClass -> {
- if (!"BasicParser".equals(parserClass.getSimpleName())) {
- availableParsers.put(parserClass.getSimpleName().replaceAll("Basic|Parser", ""),
- parserClass.getName());
- }
- });
- }
- return availableParsers;
+ return ParserIndex.INSTANCE.getIndex();
}
@Override
public Map<String, String> reloadAvailableParsers() {
- availableParsers = null;
+ ParserIndex.INSTANCE.reload();
return getAvailableParsers();
}
- private Set<Class<? extends MessageParser>> getParserClasses() {
- Reflections reflections = new Reflections("org.apache.metron.parsers");
- return reflections.getSubTypesOf(MessageParser.class);
- }
@Override
public JSONObject parseMessage(ParseMessageRequest parseMessageRequest) throws RestException {
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/util/ParserIndex.java
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/util/ParserIndex.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/util/ParserIndex.java
new file mode 100644
index 0000000..ef9c83c
--- /dev/null
+++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/util/ParserIndex.java
@@ -0,0 +1,92 @@
+/**
+ * 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.rest.util;
+
+import org.apache.metron.parsers.interfaces.MessageParser;
+import org.reflections.Reflections;
+import org.reflections.util.ClasspathHelper;
+import org.reflections.util.ConfigurationBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+import java.net.URL;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Index the parsers. Analyzing the classpath is a costly operation, so caching it makes sense.
+ * Eventually, we will probably want to have a timer that periodically reindexes so that new parsers show up.
+ */
+public enum ParserIndex {
+ INSTANCE;
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private static Set<Class<? extends MessageParser>> index;
+ private static Map<String, String> availableParsers ;
+
+ static {
+ load();
+ }
+
+ public synchronized Map<String, String> getIndex() {
+ if(availableParsers == null) {
+ load();
+ }
+ return availableParsers;
+ }
+
+ public synchronized Set<Class<? extends MessageParser>> getClasses() {
+ if(index == null) {
+ load();
+ }
+ return index;
+ }
+
+ public static void reload() {
+ load();
+ }
+
+ /**
+ * To handle the situation where classpath is specified in the manifest of the jar, we have to augment the URLs.
+ * This happens as part of the surefire plugin as well as elsewhere in the wild (especially in maven when running tests. ;).
+ * @param classLoaders
+ * @return A collection of URLs representing the effective classpath URLs
+ */
+ private static Collection<URL> effectiveClassPathUrls(ClassLoader... classLoaders) {
+ return ClasspathHelper.forManifest(ClasspathHelper.forClassLoader(classLoaders));
+ }
+
+ private static synchronized void load() {
+ LOG.debug("Starting Parser Index Load");
+ ClassLoader classLoader = ParserIndex.class.getClassLoader();
+ Reflections reflections = new Reflections(new ConfigurationBuilder().setUrls(effectiveClassPathUrls(classLoader)));
+ Set<Class<? extends MessageParser>> indexLoc = reflections.getSubTypesOf(MessageParser.class);
+ Map<String, String> availableParsersLoc = new HashMap<>();
+ indexLoc.forEach(parserClass -> {
+ if (!"BasicParser".equals(parserClass.getSimpleName())) {
+ availableParsersLoc.put(parserClass.getSimpleName().replaceAll("Basic|Parser", ""),
+ parserClass.getName());
+ }
+ });
+ LOG.debug("Finished Parser Index Load; found {} parsers, indexed {} parsers", indexLoc.size(), availableParsersLoc.size());
+ index = indexLoc;
+ availableParsers = availableParsersLoc;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-interface/metron-rest/src/main/scripts/metron-rest.sh
----------------------------------------------------------------------
diff --git a/metron-interface/metron-rest/src/main/scripts/metron-rest.sh b/metron-interface/metron-rest/src/main/scripts/metron-rest.sh
index 638589a..f9a2b69 100644
--- a/metron-interface/metron-rest/src/main/scripts/metron-rest.sh
+++ b/metron-interface/metron-rest/src/main/scripts/metron-rest.sh
@@ -21,6 +21,12 @@ if [ -z "${METRON_JDBC_PASSWORD}" ]; then
echo "METRON_JDBC_PASSWORD unset. Exiting."
exit 1
fi
+## Join a list by a character
+function join_by {
+ local IFS="$1"
+ shift
+ echo "$*"
+}
METRON_VERSION=${project.version}
METRON_HOME="${METRON_HOME:-/usr/metron/${METRON_VERSION}}"
@@ -29,6 +35,8 @@ METRON_REST_PORT=8082
METRON_SYSCONFIG="${METRON_SYSCONFIG:-/etc/default/metron}"
METRON_LOG_DIR="${METRON_LOG_DIR:-/var/log/metron}"
METRON_PID_FILE="${METRON_PID_FILE:-/var/run/metron/metron-rest.pid}"
+PARSER_CONTRIB=${PARSER_CONTRIB:-$METRON_HOME/parser_contrib}
+PARSER_LIB=$(find $METRON_HOME/lib/ -name metron-parsers*.jar)
echo "METRON_VERSION=${METRON_VERSION}"
echo "METRON_HOME=${METRON_HOME}"
@@ -47,6 +55,15 @@ rest_jar_pattern="${METRON_HOME}/lib/metron-rest*.jar"
rest_files=( ${rest_jar_pattern} )
echo "Default metron-rest jar is: ${rest_files[0]}"
METRON_REST_CLASSPATH+=":${rest_files[0]}"
+METRON_REST_CLASSPATH+=":$PARSER_LIB"
+
+if [ -d "$PARSER_CONTRIB" ]; then
+ contrib_jar_pattern="${PARSER_CONTRIB}/*.jar"
+ contrib_list=( $contrib_jar_pattern ) # expand the glob to a list
+ contrib_classpath=$(join_by : "${contrib_list[@]}") #join the list by a colon
+ echo "Parser Contrib jars are: $contrib_classpath"
+ METRON_REST_CLASSPATH+=":${contrib_classpath}"
+fi
echo "METRON_SPRING_PROFILES_ACTIVE=${METRON_SPRING_PROFILES_ACTIVE}"
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-platform/metron-parsers/3rdPartyParser.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/3rdPartyParser.md b/metron-platform/metron-parsers/3rdPartyParser.md
new file mode 100644
index 0000000..61095ea
--- /dev/null
+++ b/metron-platform/metron-parsers/3rdPartyParser.md
@@ -0,0 +1,330 @@
+<!--
+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.
+-->
+
+# Custom Metron Parsers
+
+We have many stock parsers for normal operations. Some of these are
+networking and cybersecurity focused (e.g. the ASA Parser), some of
+these are general purpose (e.g. the CSVParser), but inevitably users
+will want to extend the system to process their own data formats. To
+enable this, this is a walkthrough of how to create and use a custom
+parser within Metron.
+
+# Writing A Custom Parser
+Before we can use a parser, we will need to create a custom parser. The
+parser is the workhorse of Metron ingest. It provides the mapping
+between the raw data coming in via the Kafka value and a `JSONObject`,
+the internal data structure provided.
+
+## Implementation
+
+In order to do create a custom parser, we need to do one of the following:
+* Write a class which conforms to the `org.apache.metron.parsers.interfaces.MessageParser<JSONObject>` and `java.util.Serializable` interfaces
+ * Implement `init()`, `validate(JSONObject message)`, and `List<JSONObject> parse(byte[] rawMessage)`
+* Write a class which extends `org.apache.metron.parsers.BasicParser`
+ * Provides convenience implementations to `validate` which ensures `timestamp` and `original_string` fields exist.
+
+## Example
+
+In order to illustrate how this might be done, let's create a very
+simple parser that takes a comma separated pair and creates a couple of
+fields:
+* `original_string` -- the raw data
+* `timestamp` -- the current time
+* `first` -- the first field of the comma separated pair
+* `last` -- the last field of the comma separated pair
+
+For this demonstration, let's create a maven project to compile our
+project. We'll call it `extra_parsers`, so in your workspace, let's set
+up the maven project:
+* Create the maven infrastructure for `extra_parsers` via
+```
+mkdir -p extra_parsers/src/{main,test}/java
+```
+* Create a pom file indicating how we should build our parsers by
+ editing `extra_parsers/pom.xml` with the following content:
+```
+<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/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>com.3rdparty</groupId>
+ <artifactId>extra-parsers</artifactId>
+ <packaging>jar</packaging>
+ <version>1.0-SNAPSHOT</version>
+ <name>extra-parsers</name>
+ <url>http://thirdpartysoftware.org</url>
+ <properties>
+ <!-- The java version to conform to. Metron works all the way to 1.8 -->
+ <java_version>1.8</java_version>
+ <!-- The version of Metron that we'll be targetting. -->
+ <metron_version>0.4.1</metron_version>
+ <!-- To complete the simulation, we'll depend on a common dependency -->
+ <guava_version>19.0</guava_version>
+ <!-- We will shade our dependencies to create a single jar at the end -->
+ <shade_version>2.4.3</shade_version>
+ </properties>
+ <dependencies>
+ <!--
+ We want to depend on Metron, but ensure that the scope is "provided"
+ as we do not want to include it in our bundle.
+ -->
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-parsers</artifactId>
+ <version>${metron_version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${guava_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>3.8.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <!-- We will set up the shade plugin to create a single jar at the
+ end of the build lifecycle. We will exclude some things and
+ relocate others to simulate a real situation.
+
+ One thing to note is that it's a good practice to shade and
+ relocate common libraries that may be dependencies in Metron.
+ Your jar will be merged with the parsers jar, so the metron
+ version will be included for all overlapping classes.
+ So, shade and relocate to ensure that YOUR version of the library is used.
+ -->
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${shade_version}</version>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <artifactSet>
+ <excludes>
+ <!-- Exclude slf4j for no reason other than to illustrate how to exclude dependencies.
+ The metron team has nothing against slf4j. :-)
+ -->
+ <exclude>*slf4j*</exclude>
+ </excludes>
+ </artifactSet>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadedArtifactAttached>true</shadedArtifactAttached>
+ <shadedClassifierName>uber</shadedClassifierName>
+ <filters>
+ <filter>
+ <!-- Sometimes these get added and confuse the uber jar out of shade -->
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <relocations>
+ <!-- Relocate guava as it's used in Metron and I really want 0.19 -->
+ <relocation>
+ <pattern>com.google</pattern>
+ <shadedPattern>com.thirdparty.guava</shadedPattern>
+ </relocation>
+ </relocations>
+ <artifactSet>
+ <excludes>
+ <!-- We can also exclude by artifactId and groupId -->
+ <exclude>storm:storm-core:*</exclude>
+ <exclude>storm:storm-lib:*</exclude>
+ <exclude>org.slf4j.impl*</exclude>
+ <exclude>org.slf4j:slf4j-log4j*</exclude>
+ </excludes>
+ </artifactSet>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!--
+ We want to make sure we compile using java 1.8.
+ -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.5.1</version>
+ <configuration>
+ <forceJavacCompilerUse>true</forceJavacCompilerUse>
+ <source>${java_version}</source>
+ <compilerArgument>-Xlint:unchecked</compilerArgument>
+ <target>${java_version}</target>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
+```
+* Now let's create our parser `com.thirdparty.SimpleParser` by creating the file `extra-parsers/src/main/java/com/thirdparty/SimpleParser.java` with the following content:
+```
+package com.thirdparty;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.metron.parsers.BasicParser;
+import org.json.simple.JSONObject;
+
+import java.util.List;
+import java.util.Map;
+
+public class SimpleParser extends BasicParser {
+ @Override
+ public void init() {
+
+ }
+
+ @Override
+ public List<JSONObject> parse(byte[] bytes) {
+ String input = new String(bytes);
+ Iterable<String> it = Splitter.on(",").split(input);
+ JSONObject ret = new JSONObject();
+ ret.put("original_string", input);
+ ret.put("timestamp", System.currentTimeMillis());
+ ret.put("first", Iterables.getFirst(it, "missing"));
+ ret.put("last", Iterables.getLast(it, "missing"));
+ return ImmutableList.of(ret);
+ }
+
+ @Override
+ public void configure(Map<String, Object> map) {
+
+ }
+}
+```
+* Compile the parser via `mvn clean package` in `extra_parsers`
+
+This will create a jar containing your parser and its dependencies (sans Metron dependencies) in `extra-parsers/target/extra-parsers-1.0-SNAPSHOT-uber.jar`
+
+# Deploying Your Custom Parser
+
+In order to deploy your newly built custom parser, you would place the jar file above in the `$METRON_HOME/parser_contrib` directory on the Metron host (i.e. any host you would start parsers from or, alternatively, where the Metron REST is hosted).
+
+## Example
+
+Let's work through deploying the example above.
+
+### Preliminaries
+
+We assume that the following environment variables are set:
+* `METRON_HOME` - the home directory for metron
+* `ZOOKEEPER` - The zookeeper quorum (comma separated with port specified: e.g. `node1:2181` for full-dev)
+* `BROKERLIST` - The Kafka broker list (comma separated with port specified: e.g. `node1:6667` for full-dev)
+* `ES_HOST` - The elasticsearch master (and port) e.g. `node1:9200` for full-dev.
+
+Also, this does not assume that you are using a kerberized cluster. If you are, then the parser start command will adjust slightly to include the security protocol.
+
+### Copy the jar file up
+
+Copy the jar file located in `extra-parsers/target/extra-parsers-1.0-SNAPSHOT-uber.jar` to `$METRON_HOME/parser_contrib` and ensure the permissions are such that the `metron` user can read and execute.
+
+### Restart the REST service in Ambari
+
+In order for new parsers to be picked up, the REST service must be restarted. You can do that from within Ambari by restarting the `Metron REST` service.
+
+### Push the Zookeeper Configs
+
+Now push the config to Zookeeper with the following command:
+`$METRON_HOME/bin/zk_load_configs.sh -m PUSH -i $METRON_HOME/config/zookeeper/ -z $ZOOKEEPER`
+
+
+### Create a Kafka Topic
+
+Create a kafka topic, let's call it `test` via:
+`/usr/hdp/current/kafka-broker/bin/kafka-topics.sh --zookeeper $ZOOKEEPER --create --topic test --partitions 1 --replication-factor 1`
+
+Note, in a real deployment, that topic would be named something more descriptive and would have replication factor and partitions set to something less trivial.
+
+### Configure Test Parser
+
+Create the a file called `$METRON_HOME/config/zookeeper/parsers/test.json` with the following content:
+```
+{
+ "parserClassName":"com.thirdparty.SimpleParser",
+ "sensorTopic":"test"
+}
+```
+
+### Start Parser
+Now we can start the parser and send some data through:
+* Start the parser
+```
+$METRON_HOME/bin/start_parser_topology.sh -k $BROKERLIST -z $ZOOKEEPER -s test
+```
+* Send example data through:
+```
+echo "apache,metron" | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list $BROKERLIST --topic test
+```
+* Validate data was written in ES:
+```
+curl -XPOST "http://$ES_HOST/test*/_search?pretty" -d '
+{
+ "_source" : [ "original_string", "timestamp", "first", "last"]
+}
+'
+```
+This should yield something like:
+```
+{
+ "took" : 23,
+ "timed_out" : false,
+ "_shards" : {
+ "total" : 1,
+ "successful" : 1,
+ "failed" : 0
+ },
+ "hits" : {
+ "total" : 1,
+ "max_score" : 1.0,
+ "hits" : [ {
+ "_index" : "test_index_2017.10.04.17",
+ "_type" : "test_doc",
+ "_id" : "3ae4dd4d-8c09-4f2a-93c0-26ec5508baaa",
+ "_score" : 1.0,
+ "_source" : {
+ "original_string" : "apache,metron",
+ "last" : "metron",
+ "first" : "apache",
+ "timestamp" : 1507138373223
+ }
+ } ]
+ }
+}
+```
+
+### Via the Management UI
+
+As long as the REST service is restarted after new parsers are added to `$METRON_HOME/parser_contrib`, they are available in the UI to creating and deploying parsers.
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-platform/metron-parsers/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/pom.xml b/metron-platform/metron-parsers/pom.xml
index 18377d3..f856654 100644
--- a/metron-platform/metron-parsers/pom.xml
+++ b/metron-platform/metron-parsers/pom.xml
@@ -125,6 +125,12 @@
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
+ <artifactId>storm-rename-hack</artifactId>
+ <version>${global_storm_version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>${global_storm_version}</version>
<scope>provided</scope>
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/MergeAndShadeTransformer.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/MergeAndShadeTransformer.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/MergeAndShadeTransformer.java
new file mode 100644
index 0000000..eaadf71
--- /dev/null
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/MergeAndShadeTransformer.java
@@ -0,0 +1,101 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.parsers.topology;
+
+import com.google.common.base.Splitter;
+import org.apache.storm.daemon.JarTransformer;
+import org.apache.storm.hack.StormShadeTransformer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.lang.invoke.MethodHandles;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.jar.JarEntry;
+import java.util.jar.JarInputStream;
+import java.util.jar.JarOutputStream;
+
+/**
+ * This is a storm jar transformer that will add in additional jars pulled from an
+ * environment variable. The jars will be merged with the main uber jar and then
+ * the resulting jar will be shaded and relocated according to the StormShadeTransformer.
+ *
+ */
+public class MergeAndShadeTransformer implements JarTransformer {
+ public static final String EXTRA_JARS_ENV = "EXTRA_JARS";
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private StormShadeTransformer underlyingTransformer = new StormShadeTransformer();
+ @Override
+ public void transform(InputStream input, OutputStream output) throws IOException {
+ String extraJars = System.getenv().get(EXTRA_JARS_ENV);
+ if(extraJars == null || extraJars.length() == 0) {
+ underlyingTransformer.transform(input, output);
+ return;
+ }
+ File tmpFile = File.createTempFile("metron", "jar");
+ tmpFile.deleteOnExit();
+ Set<String> entries = new HashSet<>();
+ try (JarOutputStream jout = new JarOutputStream(new BufferedOutputStream(new FileOutputStream(tmpFile)))) {
+ try (JarInputStream jin = new JarInputStream(new BufferedInputStream(input))){
+ copy(jin, jout, entries);
+ }
+ for (String fileStr : Splitter.on(",").split(extraJars)) {
+ File f = new File(fileStr);
+ if (!f.exists()) {
+ continue;
+ }
+ LOG.info("Merging jar {} from {}", f.getName(), f.getAbsolutePath());
+ try (JarInputStream jin = new JarInputStream(new BufferedInputStream(new FileInputStream(f)))) {
+ copy(jin, jout, entries);
+ }
+ }
+ }
+ underlyingTransformer.transform(new BufferedInputStream(new FileInputStream(tmpFile)), output);
+ }
+
+ /**
+ * Merges two jars. The first jar will get merged into the output jar.
+ * A running set of jar entries is kept so that duplicates are skipped.
+ * This has the side-effect that the first instance of a given entry will be added
+ * and all subsequent entries are skipped.
+ *
+ * @param jin The input jar
+ * @param jout The output jar
+ * @param entries The set of existing entries. Note that this set will be mutated as part of this call.
+ * @return The set of entries.
+ * @throws IOException
+ */
+ private Set<String> copy(JarInputStream jin, JarOutputStream jout, Set<String> entries) throws IOException {
+ byte[] buffer = new byte[1024];
+ for(JarEntry entry = jin.getNextJarEntry(); entry != null; entry = jin.getNextJarEntry()) {
+ if(entries.contains(entry.getName())) {
+ continue;
+ }
+ LOG.debug("Merging jar entry {}", entry.getName());
+ entries.add(entry.getName());
+ jout.putNextEntry(entry);
+ int len = 0;
+ while( (len = jin.read(buffer)) > 0 ) {
+ jout.write(buffer, 0, len);
+ }
+ }
+ return entries;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
index b5ee628..4ce0508 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
@@ -23,10 +23,6 @@ import org.apache.metron.storm.kafka.flux.SpoutConfiguration;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.AlreadyAliveException;
-import org.apache.storm.generated.AuthorizationException;
-import org.apache.storm.generated.InvalidTopologyException;
-import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Joiner;
@@ -285,7 +281,19 @@ public class ParserTopologyCLI {
}
private static CommandLine parse(Options options, String[] args) {
- CommandLineParser parser = new PosixParser();
+ /*
+ * The general gist is that in order to pass args to storm jar,
+ * we have to disregard options that we don't know about in the CLI.
+ * Storm will ignore our args, we have to do the same.
+ */
+ CommandLineParser parser = new PosixParser() {
+ @Override
+ protected void processOption(String arg, ListIterator iter) throws ParseException {
+ if(getOptions().hasOption(arg)) {
+ super.processOption(arg, iter);
+ }
+ }
+ };
try {
return ParserOptions.parse(parser, args);
} catch (ParseException pe) {
http://git-wip-us.apache.org/repos/asf/metron/blob/a262f87f/metron-platform/metron-parsers/src/main/scripts/start_parser_topology.sh
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/scripts/start_parser_topology.sh b/metron-platform/metron-parsers/src/main/scripts/start_parser_topology.sh
index 8faf89e..00ac809 100755
--- a/metron-platform/metron-parsers/src/main/scripts/start_parser_topology.sh
+++ b/metron-platform/metron-parsers/src/main/scripts/start_parser_topology.sh
@@ -19,4 +19,11 @@
METRON_VERSION=${project.version}
METRON_HOME=/usr/metron/$METRON_VERSION
TOPOLOGY_JAR=metron-parsers-$METRON_VERSION-uber.jar
-storm jar $METRON_HOME/lib/$TOPOLOGY_JAR org.apache.metron.parsers.topology.ParserTopologyCLI "$@"
+PARSER_CONTRIB=${PARSER_CONTRIB:-$METRON_HOME/parser_contrib}
+if [ -d "$PARSER_CONTRIB" ]; then
+ export STORM_EXT_CLASSPATH=$METRON_HOME/lib/$TOPOLOGY_JAR
+ export EXTRA_JARS=$(ls -m $PARSER_CONTRIB/*.jar | tr -d ' ' | tr -d '\n' | sed 's/\/\//\//g')
+ storm jar $METRON_HOME/lib/$TOPOLOGY_JAR org.apache.metron.parsers.topology.ParserTopologyCLI "$@" -c client.jartransformer.class=org.apache.metron.parsers.topology.MergeAndShadeTransformer
+else
+ storm jar $METRON_HOME/lib/$TOPOLOGY_JAR org.apache.metron.parsers.topology.ParserTopologyCLI "$@"
+fi