You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ni...@apache.org on 2016/08/22 14:48:27 UTC
[1/3] incubator-metron git commit: METRON-309 Created a Normalcy
Profiler (nickwallen) closes apache/incubator-metron#208
Repository: incubator-metron
Updated Branches:
refs/heads/master 1ea8d9ee4 -> 817548647
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/HBaseBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/HBaseBolt.java b/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/HBaseBolt.java
new file mode 100644
index 0000000..4c70ae4
--- /dev/null
+++ b/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/HBaseBolt.java
@@ -0,0 +1,202 @@
+/*
+ *
+ * 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.hbase.bolt;
+
+import backtype.storm.Config;
+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.Tuple;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.metron.hbase.HTableProvider;
+import org.apache.metron.hbase.TableProvider;
+import org.apache.metron.hbase.client.HBaseClient;
+import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
+import org.apache.storm.hbase.common.ColumnList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A bolt that writes to HBase.
+ *
+ * Each bolt defined within a topology can interact with only a single HBase table.
+ */
+public class HBaseBolt extends BaseRichBolt {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseBolt.class);
+
+ /**
+ * Should the write-ahead-log be used.
+ */
+ private boolean writeToWAL = true;
+
+ /**
+ * The interval in seconds at which the writes are flushed to Hbase.
+ */
+ private int flushIntervalSecs = 1;
+
+ /**
+ * The batch size.
+ */
+ private int batchSize = 1000;
+
+ /**
+ * The name of the HBase table. Each bolt communicates with a single HBase table.
+ */
+ protected String tableName;
+
+ /**
+ * The mapper which defines how tuple fields are mapped to HBase.
+ */
+ protected HBaseMapper mapper;
+
+ /**
+ * The name of the class that should be used as a table provider.
+ */
+ protected String tableProvider = "org.apache.metron.hbase.HTableProvider";
+
+ private List<Mutation> batchMutations;
+ private BatchHelper batchHelper;
+ protected OutputCollector collector;
+ protected transient HBaseClient hbaseClient;
+
+ public HBaseBolt(String tableName, HBaseMapper mapper) {
+ this.tableName = tableName;
+ this.mapper = mapper;
+ this.batchMutations = new LinkedList<>();
+ }
+
+ public HBaseBolt writeToWAL(boolean writeToWAL) {
+ this.writeToWAL = writeToWAL;
+ return this;
+ }
+
+ public HBaseBolt withTableProvider(String tableProvider) {
+ this.tableProvider = tableProvider;
+ return this;
+ }
+
+ public HBaseBolt withBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ return this;
+ }
+
+ public HBaseBolt withFlushIntervalSecs(int flushIntervalSecs) {
+ this.flushIntervalSecs = flushIntervalSecs;
+ return this;
+ }
+
+ public void setClient(HBaseClient hbaseClient) {
+ this.hbaseClient = hbaseClient;
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ Config conf = new Config();
+ conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, flushIntervalSecs);
+ return conf;
+ }
+
+ @Override
+ public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
+ this.collector = collector;
+ this.batchHelper = new BatchHelper(batchSize, collector);
+
+ TableProvider provider = getTableProvider(tableProvider);
+ hbaseClient = new HBaseClient(provider, HBaseConfiguration.create(), tableName);
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+ // nothing emitted
+ }
+
+ @Override
+ public void execute(Tuple tuple) {
+ try {
+ if (batchHelper.shouldHandle(tuple)) {
+ save(tuple);
+ }
+
+ if (batchHelper.shouldFlush()) {
+ flush();
+ }
+
+ } catch (Exception e) {
+ batchHelper.fail(e);
+ batchMutations.clear();
+ }
+ }
+
+ /**
+ * Saves an operation for later.
+ * @param tuple Contains the data elements that need written to HBase.
+ */
+ private void save(Tuple tuple) {
+ byte[] rowKey = this.mapper.rowKey(tuple);
+ ColumnList cols = this.mapper.columns(tuple);
+ Durability durability = writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL;
+ List<Mutation> mutations = hbaseClient.constructMutationReq(rowKey, cols, durability);
+ batchMutations.addAll(mutations);
+ batchHelper.addBatch(tuple);
+ }
+
+ /**
+ * Flush all saved operations.
+ */
+ private void flush() {
+ this.hbaseClient.batchMutate(batchMutations);
+ batchHelper.ack();
+ batchMutations.clear();
+ }
+
+ /**
+ *
+ * @param connectorImpl
+ * @return
+ */
+ private static TableProvider getTableProvider(String connectorImpl) {
+
+ // if class name not defined, use a reasonable default
+ if(StringUtils.isEmpty(connectorImpl) || connectorImpl.charAt(0) == '$') {
+ return new HTableProvider();
+ }
+
+ // instantiate the table provider
+ try {
+ Class<? extends TableProvider> clazz = (Class<? extends TableProvider>) Class.forName(connectorImpl);
+ return clazz.getConstructor().newInstance();
+
+ } catch (InstantiationException | IllegalAccessException | IllegalStateException |
+ InvocationTargetException | NoSuchMethodException | ClassNotFoundException e) {
+ throw new IllegalStateException("Unable to instantiate connector", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/client/HBaseClient.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/client/HBaseClient.java b/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/client/HBaseClient.java
new file mode 100644
index 0000000..888e485
--- /dev/null
+++ b/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/client/HBaseClient.java
@@ -0,0 +1,165 @@
+/*
+ *
+ * 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.hbase.client;
+
+import com.google.common.collect.Lists;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.security.UserProvider;
+import org.apache.metron.hbase.TableProvider;
+import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
+import org.apache.storm.hbase.common.ColumnList;
+import org.apache.storm.hbase.security.HBaseSecurityUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A client that interacts with HBase.
+ */
+public class HBaseClient implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseClient.class);
+ private HTableInterface table;
+
+ public HBaseClient(TableProvider provider, final Configuration configuration, final String tableName) {
+ try {
+ this.table = provider.getTable(configuration, tableName);
+
+ } catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public List<Mutation> constructMutationReq(byte[] rowKey, ColumnList cols, Durability durability) {
+ List<Mutation> mutations = Lists.newArrayList();
+
+ if (cols.hasColumns()) {
+ Put put = createPut(rowKey, cols, durability);
+ mutations.add(put);
+ }
+
+ if (cols.hasCounters()) {
+ Increment inc = createIncrement(rowKey, cols, durability);
+ mutations.add(inc);
+ }
+
+ if (mutations.isEmpty()) {
+ mutations.add(new Put(rowKey));
+ }
+
+ return mutations;
+ }
+
+ public void batchMutate(List<Mutation> mutations) {
+ Object[] result = new Object[mutations.size()];
+ try {
+ table.batch(mutations, result);
+
+ } catch (InterruptedException | IOException e) {
+ LOG.warn("Error performing a mutation to HBase.", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public Get constructGetRequests(byte[] rowKey, HBaseProjectionCriteria projectionCriteria) {
+ Get get = new Get(rowKey);
+
+ if (projectionCriteria != null) {
+ for (byte[] columnFamily : projectionCriteria.getColumnFamilies()) {
+ get.addFamily(columnFamily);
+ }
+
+ for (HBaseProjectionCriteria.ColumnMetaData columnMetaData : projectionCriteria.getColumns()) {
+ get.addColumn(columnMetaData.getColumnFamily(), columnMetaData.getQualifier());
+ }
+ }
+
+ return get;
+ }
+
+ public Result[] batchGet(List<Get> gets) {
+ try {
+ return table.get(gets);
+
+ } catch (Exception e) {
+ LOG.warn("Could not perform HBase lookup.", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ table.close();
+ }
+
+ /**
+ * Creates an HBase Put.
+ * @param rowKey The row key.
+ * @param cols The columns to put.
+ * @param durability The durability of the put.
+ */
+ private Put createPut(byte[] rowKey, ColumnList cols, Durability durability) {
+ Put put = new Put(rowKey);
+ put.setDurability(durability);
+
+ for (ColumnList.Column col : cols.getColumns()) {
+ if (col.getTs() > 0) {
+ put.add(col.getFamily(),
+ col.getQualifier(),
+ col.getTs(),
+ col.getValue());
+
+ } else {
+ put.add(col.getFamily(),
+ col.getQualifier(),
+ col.getValue());
+ }
+ }
+
+ return put;
+ }
+
+ /**
+ * Creates an HBase Increment for a counter.
+ * @param rowKey The row key.
+ * @param cols The columns to include.
+ * @param durability The durability of the increment.
+ */
+ private Increment createIncrement(byte[] rowKey, ColumnList cols, Durability durability) {
+ Increment inc = new Increment(rowKey);
+ inc.setDurability(durability);
+
+ for (ColumnList.Counter cnt : cols.getCounters()) {
+ inc.addColumn(
+ cnt.getFamily(),
+ cnt.getQualifier(),
+ cnt.getIncrement());
+ }
+
+ return inc;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/Widget.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/Widget.java b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/Widget.java
new file mode 100644
index 0000000..d47e332
--- /dev/null
+++ b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/Widget.java
@@ -0,0 +1,77 @@
+/*
+ *
+ * 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.hbase;
+
+/**
+ * A simple POJO used for testing.
+ */
+public class Widget {
+
+ /**
+ * The name of the widget.
+ */
+ private String name;
+
+ /**
+ * The cost of the widget.
+ */
+ private int cost;
+
+ public Widget(String name, int cost) {
+ this.name = name;
+ this.cost = cost;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public int getCost() {
+ return cost;
+ }
+
+ public void setCost(int cost) {
+ this.cost = cost;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ Widget widget = (Widget) o;
+
+ if (cost != widget.cost) return false;
+ return name != null ? name.equals(widget.name) : widget.name == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = name != null ? name.hashCode() : 0;
+ result = 31 * result + cost;
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/WidgetMapper.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/WidgetMapper.java b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/WidgetMapper.java
new file mode 100644
index 0000000..c7d3fae
--- /dev/null
+++ b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/WidgetMapper.java
@@ -0,0 +1,55 @@
+/*
+ *
+ * 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.hbase;
+
+import backtype.storm.tuple.Tuple;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
+import org.apache.storm.hbase.common.ColumnList;
+
+import java.util.Calendar;
+
+/**
+ * Maps a Widget to HBase. Used only for testing.
+ */
+public class WidgetMapper implements HBaseMapper {
+
+ @Override
+ public byte[] rowKey(Tuple tuple) {
+ Widget w = (Widget) tuple.getValueByField("widget");
+ return Bytes.toBytes(w.getName());
+ }
+
+ @Override
+ public ColumnList columns(Tuple tuple) {
+ Widget w = (Widget) tuple.getValueByField("widget");
+
+ ColumnList cols = new ColumnList();
+ cols.addColumn(CF, QNAME, Bytes.toBytes(w.getName()));
+ cols.addColumn(CF, QCOST, Bytes.toBytes(w.getCost()));
+ return cols;
+ }
+
+ public static final String CF_STRING = "cfWidget";
+ public static final byte[] CF = Bytes.toBytes(CF_STRING);
+ public static final byte[] QNAME = Bytes.toBytes("qName");
+ public static final byte[] QCOST = Bytes.toBytes("qCost");
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/bolt/HBaseBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/bolt/HBaseBoltTest.java b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/bolt/HBaseBoltTest.java
new file mode 100644
index 0000000..d790d51
--- /dev/null
+++ b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/bolt/HBaseBoltTest.java
@@ -0,0 +1,156 @@
+/*
+ *
+ * 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.hbase.bolt;
+
+import backtype.storm.Constants;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.hbase.Widget;
+import org.apache.metron.hbase.WidgetMapper;
+import org.apache.metron.hbase.client.HBaseClient;
+import org.apache.metron.test.bolt.BaseBoltTest;
+import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mock;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.*;
+
+/**
+ * Tests the HBaseBolt.
+ */
+public class HBaseBoltTest extends BaseBoltTest {
+
+ private static final String tableName = "widgets";
+
+ @Mock
+ private HBaseClient client;
+
+ @Mock
+ private Tuple tuple1;
+
+ @Mock
+ private Tuple tuple2;
+
+ private Widget widget1;
+
+ private Widget widget2;
+
+ private HBaseMapper mapper;
+
+
+ @Before
+ public void setupTuples() throws Exception {
+
+ // setup the first tuple
+ widget1 = new Widget("widget1", 100);
+ when(tuple1.getValueByField(eq("widget"))).thenReturn(widget1);
+
+ // setup the second tuple
+ widget2 = new Widget("widget2", 200);
+ when(tuple2.getValueByField(eq("widget"))).thenReturn(widget2);
+ }
+
+ @Before
+ public void setup() throws Exception {
+
+ // create a mapper
+ mapper = new WidgetMapper();
+
+ }
+
+ /**
+ * Create a ProfileBuilderBolt to test
+ */
+ private HBaseBolt createBolt() throws IOException {
+ HBaseBolt bolt = new HBaseBolt(tableName, mapper)
+ .withBatchSize(2);
+
+ bolt.prepare(Collections.emptyMap(), topologyContext, outputCollector);
+ bolt.setClient(client);
+ return bolt;
+ }
+
+ /**
+ * What happens if the batch is full?
+ *
+ * If the batch size is 2 and we have received 2 tuples the batch should be flushed.
+ */
+ @Test
+ public void testBatchReady() throws Exception {
+ HBaseBolt bolt = createBolt();
+ bolt.execute(tuple1);
+ bolt.execute(tuple2);
+ verify(client, times(1)).batchMutate(any(List.class));
+ }
+
+ /**
+ * What happens if the batch is not full?
+ *
+ * If the batch size is 2 and we have only received 2 tuple, the batch should not be flushed.
+ */
+ @Test
+ public void testBatchNotReady() throws Exception {
+ HBaseBolt bolt = createBolt();
+ bolt.execute(tuple1);
+ verify(client, times(0)).batchMutate(any(List.class));
+ }
+
+ /**
+ * What happens if the batch timeout is reached?
+ */
+ @Test
+ public void testTimeFlush() throws Exception {
+ HBaseBolt bolt = createBolt();
+
+ // the batch is not ready to write
+ bolt.execute(tuple1);
+ verify(client, times(0)).batchMutate(any(List.class));
+
+ // the batch should be written after the tick tuple
+ bolt.execute(mockTickTuple());
+ verify(client, times(1)).batchMutate(any(List.class));
+ }
+
+ private static Tuple mockTuple(String componentId, String streamId) {
+ Tuple tuple = mock(Tuple.class);
+ when(tuple.getSourceComponent()).thenReturn(componentId);
+ when(tuple.getSourceStreamId()).thenReturn(streamId);
+ return tuple;
+ }
+
+ private static Tuple mockTickTuple() {
+ return mockTuple(Constants.SYSTEM_COMPONENT_ID, Constants.SYSTEM_TICK_STREAM_ID);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java
new file mode 100644
index 0000000..f024420
--- /dev/null
+++ b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java
@@ -0,0 +1,149 @@
+/*
+ *
+ * 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.hbase.client;
+
+import backtype.storm.tuple.Tuple;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.hbase.TableProvider;
+import org.apache.metron.hbase.Widget;
+import org.apache.metron.hbase.WidgetMapper;
+import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
+import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
+import org.apache.storm.hbase.common.ColumnList;
+import org.junit.*;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.*;
+
+/**
+ * Tests the HBaseClient
+ */
+public class HBaseClientTest {
+
+ private static final String tableName = "widgets";
+
+ private static HBaseTestingUtility util;
+ private HBaseClient client;
+ private HTableInterface table;
+
+ private Tuple tuple1;
+ private Tuple tuple2;
+
+ private Widget widget1;
+ private Widget widget2;
+
+ private HBaseMapper mapper;
+
+ @BeforeClass
+ public static void startHBase() throws Exception {
+ util = new HBaseTestingUtility();
+ util.startMiniCluster();
+ }
+
+ @AfterClass
+ public static void stopHBase() throws Exception {
+ util.shutdownMiniCluster();
+ util.cleanupTestDir();
+ }
+
+ @Before
+ public void setupTuples() throws Exception {
+
+ // setup the first tuple
+ widget1 = new Widget("widget1", 100);
+ tuple1 = mock(Tuple.class);
+ when(tuple1.getValueByField(eq("widget"))).thenReturn(widget1);
+
+ // setup the second tuple
+ widget2 = new Widget("widget2", 200);
+ tuple2 = mock(Tuple.class);
+ when(tuple2.getValueByField(eq("widget"))).thenReturn(widget2);
+ }
+
+ @Before
+ public void setup() throws Exception {
+
+ // create a mapper
+ mapper = new WidgetMapper();
+
+ // create the table
+ table = util.createTable(Bytes.toBytes(tableName), WidgetMapper.CF);
+
+ // setup the client
+ client = new HBaseClient((c,t) -> table, table.getConfiguration(), tableName);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ util.deleteTable(tableName);
+ }
+
+ @Test
+ public void testWrite() throws Exception {
+
+ // add a tuple to the batch
+ byte[] rowKey1 = mapper.rowKey(tuple1);
+ ColumnList cols1 = mapper.columns(tuple1);
+ List<Mutation> mutations1 = client.constructMutationReq(rowKey1, cols1, Durability.SYNC_WAL);
+ client.batchMutate(mutations1);
+
+ HBaseProjectionCriteria criteria = new HBaseProjectionCriteria();
+ criteria.addColumnFamily(WidgetMapper.CF_STRING);
+
+ // read back the tuple
+ Get get1 = client.constructGetRequests(rowKey1, criteria);
+ Result[] results = client.batchGet(Arrays.asList(get1));
+ Assert.assertEquals(1, results.length);
+ }
+
+ @Test
+ public void testBatchWrite() throws Exception {
+
+ // add a tuple to the batch
+ byte[] rowKey1 = mapper.rowKey(tuple1);
+ ColumnList cols1 = mapper.columns(tuple1);
+ List<Mutation> mutations1 = client.constructMutationReq(rowKey1, cols1, Durability.SYNC_WAL);
+ client.batchMutate(mutations1);
+
+ // add another tuple to the batch
+ byte[] rowKey2 = mapper.rowKey(tuple1);
+ ColumnList cols2 = mapper.columns(tuple1);
+ List<Mutation> mutations2 = client.constructMutationReq(rowKey2, cols2, Durability.SYNC_WAL);
+ client.batchMutate(mutations2);
+
+ HBaseProjectionCriteria criteria = new HBaseProjectionCriteria();
+ criteria.addColumnFamily(WidgetMapper.CF_STRING);
+
+ // read back both tuples
+ Get get1 = client.constructGetRequests(rowKey1, criteria);
+ Get get2 = client.constructGetRequests(rowKey2, criteria);
+ Result[] results = client.batchGet(Arrays.asList(get1, get2));
+ Assert.assertEquals(2, results.length);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/components/FluxTopologyComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/components/FluxTopologyComponent.java b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/components/FluxTopologyComponent.java
index 39eca58..75be2d7 100644
--- a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/components/FluxTopologyComponent.java
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/components/FluxTopologyComponent.java
@@ -6,9 +6,9 @@
* 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
- *
+ * <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.
@@ -20,6 +20,7 @@ package org.apache.metron.integration.components;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
+import backtype.storm.generated.TopologyInfo;
import org.apache.metron.integration.InMemoryComponent;
import org.apache.metron.integration.UnableToStartException;
import org.apache.storm.flux.FluxBuilder;
@@ -28,110 +29,122 @@ import org.apache.storm.flux.model.TopologyDef;
import org.apache.storm.flux.parser.FluxParser;
import org.apache.thrift7.TException;
import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
public class FluxTopologyComponent implements InMemoryComponent {
- LocalCluster stormCluster;
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FluxTopologyComponent.class);
+
+ LocalCluster stormCluster;
+ String topologyName;
+ File topologyLocation;
+ Properties topologyProperties;
+
+ public static class Builder {
+
String topologyName;
File topologyLocation;
Properties topologyProperties;
- public static class Builder {
- String topologyName;
- File topologyLocation;
- Properties topologyProperties;
- public Builder withTopologyName(String name) {
- this.topologyName = name;
- return this;
- }
- public Builder withTopologyLocation(File location) {
- this.topologyLocation = location;
- return this;
- }
- public Builder withTopologyProperties(Properties properties) {
- this.topologyProperties = properties;
- return this;
- }
-
- public FluxTopologyComponent build() {
- return new FluxTopologyComponent(topologyName, topologyLocation, topologyProperties);
- }
+ public Builder withTopologyName(String name) {
+ this.topologyName = name;
+ return this;
}
- public FluxTopologyComponent(String topologyName, File topologyLocation, Properties topologyProperties) {
- this.topologyName = topologyName;
- this.topologyLocation = topologyLocation;
- this.topologyProperties = topologyProperties;
+ public Builder withTopologyLocation(File location) {
+ this.topologyLocation = location;
+ return this;
}
- public LocalCluster getStormCluster() {
- return stormCluster;
+ public Builder withTopologyProperties(Properties properties) {
+ this.topologyProperties = properties;
+ return this;
}
- public String getTopologyName() {
- return topologyName;
+ public FluxTopologyComponent build() {
+ return new FluxTopologyComponent(topologyName, topologyLocation, topologyProperties);
}
+ }
- public File getTopologyLocation() {
- return topologyLocation;
- }
+ public FluxTopologyComponent(String topologyName, File topologyLocation, Properties topologyProperties) {
+ this.topologyName = topologyName;
+ this.topologyLocation = topologyLocation;
+ this.topologyProperties = topologyProperties;
+ }
- public Properties getTopologyProperties() {
- return topologyProperties;
- }
+ public LocalCluster getStormCluster() {
+ return stormCluster;
+ }
- public String getZookeeperConnectString() {
- return "localhost:2000";
- }
- public void start() throws UnableToStartException{
- try {
- stormCluster = new LocalCluster();
- } catch (Exception e) {
- throw new UnableToStartException("Unable to start flux topology: " + getTopologyLocation(), e);
- }
- }
+ public String getTopologyName() {
+ return topologyName;
+ }
- public void stop() {
- if(stormCluster != null) {
- stormCluster.shutdown();
- }
- }
+ public File getTopologyLocation() {
+ return topologyLocation;
+ }
- public void submitTopology() throws NoSuchMethodException, IOException, InstantiationException, TException, IllegalAccessException, InvocationTargetException, ClassNotFoundException {
- startTopology(getTopologyName(), getTopologyLocation(), getTopologyProperties());
- }
- private void startTopology(String topologyName, File topologyLoc, Properties properties) throws IOException, ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, TException {
- TopologyDef topologyDef = loadYaml(topologyName, topologyLoc, properties);
- Config conf = FluxBuilder.buildConfig(topologyDef);
- ExecutionContext context = new ExecutionContext(topologyDef, conf);
- StormTopology topology = FluxBuilder.buildTopology(context);
- Assert.assertNotNull(topology);
- topology.validate();
- stormCluster.submitTopology(topologyName, conf, topology);
+ public Properties getTopologyProperties() {
+ return topologyProperties;
+ }
+
+ public String getZookeeperConnectString() {
+ return "localhost:2000";
+ }
+
+ public void start() throws UnableToStartException {
+ try {
+ stormCluster = new LocalCluster();
+
+ } catch (Exception e) {
+ throw new UnableToStartException("Unable to start flux topology: " + getTopologyLocation(), e);
}
+ }
- private static TopologyDef loadYaml(String topologyName, File yamlFile, Properties properties) throws IOException {
- File tmpFile = File.createTempFile(topologyName, "props");
- tmpFile.deleteOnExit();
- FileWriter propWriter = null;
- try {
- propWriter = new FileWriter(tmpFile);
- properties.store(propWriter, topologyName + " properties");
- }
- finally {
- if(propWriter != null) {
- propWriter.close();
- return FluxParser.parseFile(yamlFile.getAbsolutePath(), false, true, tmpFile.getAbsolutePath(), false);
- }
-
- return null;
- }
+ public void stop() {
+ if (stormCluster != null) {
+ stormCluster.shutdown();
+ }
+ }
+
+ public void submitTopology() throws NoSuchMethodException, IOException, InstantiationException, TException, IllegalAccessException, InvocationTargetException, ClassNotFoundException {
+ startTopology(getTopologyName(), getTopologyLocation(), getTopologyProperties());
+ }
+
+ private void startTopology(String topologyName, File topologyLoc, Properties properties) throws IOException, ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, TException {
+ TopologyDef topologyDef = loadYaml(topologyName, topologyLoc, properties);
+ Config conf = FluxBuilder.buildConfig(topologyDef);
+ ExecutionContext context = new ExecutionContext(topologyDef, conf);
+ StormTopology topology = FluxBuilder.buildTopology(context);
+ Assert.assertNotNull(topology);
+ topology.validate();
+ stormCluster.submitTopology(topologyName, conf, topology);
+ }
+
+ private static TopologyDef loadYaml(String topologyName, File yamlFile, Properties properties) throws IOException {
+ File tmpFile = File.createTempFile(topologyName, "props");
+ tmpFile.deleteOnExit();
+ FileWriter propWriter = null;
+ try {
+ propWriter = new FileWriter(tmpFile);
+ properties.store(propWriter, topologyName + " properties");
+ } finally {
+ if (propWriter != null) {
+ propWriter.close();
+ return FluxParser.parseFile(yamlFile.getAbsolutePath(), false, true, tmpFile.getAbsolutePath(), false);
+ }
+
+ return null;
}
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/mock/MockHTable.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/mock/MockHTable.java b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/mock/MockHTable.java
index 6606fdc..75d8104 100644
--- a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/mock/MockHTable.java
+++ b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/mock/MockHTable.java
@@ -118,7 +118,6 @@ public class MockHTable implements HTableInterface {
}
}
-
@Override
public byte[] getTableName() {
return Bytes.toBytes(tableName);
@@ -200,8 +199,8 @@ public class MockHTable implements HTableInterface {
@Override
public void batch(List<? extends Row> list, Object[] objects) throws IOException, InterruptedException {
- throw new UnsupportedOperationException();
-
+ Object[] results = batch(list);
+ System.arraycopy(results, 0, objects, 0, objects.length);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/pom.xml b/metron-platform/pom.xml
index 1f8d5f1..018e728 100644
--- a/metron-platform/pom.xml
+++ b/metron-platform/pom.xml
@@ -73,7 +73,6 @@
</dependencies>
<build>
<plugins>
-
</plugins>
</build>
<reporting>
[2/3] incubator-metron git commit: METRON-309 Created a Normalcy
Profiler (nickwallen) closes apache/incubator-metron#208
Posted by ni...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileSplitterBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileSplitterBoltTest.java b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileSplitterBoltTest.java
new file mode 100644
index 0000000..13f8171
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileSplitterBoltTest.java
@@ -0,0 +1,212 @@
+/*
+ *
+ * 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.profiler.bolt;
+
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.metron.profiler.stellar.DefaultStellarExecutor;
+import org.apache.metron.test.bolt.BaseBoltTest;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.refEq;
+import static org.mockito.Mockito.*;
+
+/**
+ * Tests the ProfileSplitterBolt.
+ */
+public class ProfileSplitterBoltTest extends BaseBoltTest {
+
+ /**
+ * {
+ * "ip_src_addr": "10.0.0.1",
+ * "ip_dst_addr": "10.0.0.20",
+ * "protocol": "HTTP"
+ * }
+ */
+ @Multiline
+ private String input;
+
+ /**
+ * {
+ * "inputTopic": "enrichment",
+ * "profiles": [
+ * {
+ * "profile": "test",
+ * "foreach": "ip_src_addr",
+ * "onlyif": "protocol == 'HTTP'",
+ * "init": {},
+ * "update": {},
+ * "result": 2
+ * }
+ * ]
+ * }
+ */
+ @Multiline
+ private String onlyIfTrue;
+
+ /**
+ * {
+ * "inputTopic": "enrichment",
+ * "profiles": [
+ * {
+ * "profile": "test",
+ * "foreach": "ip_src_addr",
+ * "onlyif": "false",
+ * "init": {},
+ * "update": {},
+ * "result": 2
+ * }
+ * ]
+ * }
+ */
+ @Multiline
+ private String onlyIfFalse;
+
+ /**
+ * {
+ * "inputTopic": "enrichment",
+ * "profiles": [
+ * {
+ * "profile": "test",
+ * "foreach": "ip_src_addr",
+ * "onlyif": "NOT-VALID",
+ * "init": {},
+ * "update": {},
+ * "result": 2
+ * }
+ * ]
+ * }
+ */
+ @Multiline
+ private String invalidOnlyIf;
+
+ private JSONObject message;
+
+ @Before
+ public void setup() throws ParseException {
+
+ // parse the input message
+ JSONParser parser = new JSONParser();
+ message = (JSONObject) parser.parse(input);
+
+ // ensure the tuple returns the expected json message
+ when(tuple.getBinary(0)).thenReturn(input.getBytes());
+ }
+
+ /**
+ * Create a ProfileSplitterBolt to test
+ */
+ private ProfileSplitterBolt createBolt(String profilerConfig) throws IOException {
+
+ ProfileSplitterBolt bolt = new ProfileSplitterBolt("zookeeperURL");
+ bolt.setCuratorFramework(client);
+ bolt.setTreeCache(cache);
+ bolt.getConfigurations().updateProfilerConfig(profilerConfig.getBytes("UTF-8"));
+ bolt.setExecutor(new DefaultStellarExecutor());
+
+ bolt.prepare(new HashMap<>(), topologyContext, outputCollector);
+ return bolt;
+ }
+
+ /**
+ * What happens when a message is received that is needed by a profile?
+ *
+ * This occurs when a profile's 'onlyif' expression is true.
+ */
+ @Test
+ public void testMessageNeededByProfile() throws Exception {
+
+ // setup
+ ProfileSplitterBolt bolt = createBolt(onlyIfTrue);
+
+ // execute
+ bolt.execute(tuple);
+
+ // a tuple should be emitted for the downstream profile builder
+ verify(outputCollector, times(1)).emit(refEq(tuple), any(Values.class));
+
+ // the original tuple should be ack'd
+ verify(outputCollector, times(1)).ack(tuple);
+ }
+
+ /**
+ * What happens when a message is received that is NOT needed by a profile?
+ *
+ * This occurs when a profile's 'onlyif' expression is false
+ */
+ @Test
+ public void testMessageNotNeededByProfile() throws Exception {
+
+ // setup
+ ProfileSplitterBolt bolt = createBolt(onlyIfFalse);
+
+ // execute
+ bolt.execute(tuple);
+
+ // a tuple should NOT be emitted for the downstream profile builder
+ verify(outputCollector, times(0)).emit(any(Values.class));
+
+ // the original tuple should be ack'd
+ verify(outputCollector, times(1)).ack(tuple);
+ }
+
+ /**
+ * The entity associated with a ProfileMeasurement can be defined using a variable that is resolved
+ * via Stella. In this case the entity is defined as 'ip_src_addr' which is resolved to
+ * '10.0.0.1' based on the data contained within the message.
+ */
+ @Test
+ public void testResolveEntityName() throws Exception {
+
+ // setup
+ ProfileSplitterBolt bolt = createBolt(onlyIfTrue);
+
+ // execute
+ bolt.execute(tuple);
+
+ // verify - the entity name comes from variable resolution in stella
+ String expectedEntity = "10.0.0.1";
+ verify(outputCollector, times(1)).emit(any(Tuple.class), refEq(new Values(expectedEntity, onlyIfTrue, message)));
+ }
+
+ /**
+ * What happens when invalid Stella code is used for 'onlyif'?
+ */
+ @Test(expected = org.apache.metron.common.dsl.ParseException.class)
+ public void testOnlyIfInvalid() throws Exception {
+
+ // setup
+ ProfileSplitterBolt bolt = createBolt(invalidOnlyIf);
+
+ // execute
+ bolt.execute(tuple);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ConfigUploadComponent.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ConfigUploadComponent.java b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ConfigUploadComponent.java
new file mode 100644
index 0000000..f096d51
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ConfigUploadComponent.java
@@ -0,0 +1,119 @@
+/*
+ *
+ * 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.profiler.integration;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.metron.common.configuration.SensorParserConfig;
+import org.apache.metron.integration.InMemoryComponent;
+import org.apache.metron.integration.UnableToStartException;
+import org.apache.metron.integration.components.KafkaWithZKComponent;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import static org.apache.metron.common.configuration.ConfigurationsUtils.*;
+
+public class ConfigUploadComponent implements InMemoryComponent {
+
+ private Properties topologyProperties;
+ private String globalConfigPath;
+ private String parserConfigsPath;
+ private String enrichmentConfigsPath;
+ private String profilerConfigPath;
+ private Optional<String> globalConfig = Optional.empty();
+ private Map<String, SensorParserConfig> parserSensorConfigs = new HashMap<>();
+
+ public ConfigUploadComponent withTopologyProperties(Properties topologyProperties) {
+ this.topologyProperties = topologyProperties;
+ return this;
+ }
+
+ public ConfigUploadComponent withGlobalConfigsPath(String globalConfigPath) {
+ this.globalConfigPath = globalConfigPath;
+ return this;
+ }
+
+ public ConfigUploadComponent withParserConfigsPath(String parserConfigsPath) {
+ this.parserConfigsPath = parserConfigsPath;
+ return this;
+ }
+ public ConfigUploadComponent withEnrichmentConfigsPath(String enrichmentConfigsPath) {
+ this.enrichmentConfigsPath = enrichmentConfigsPath;
+ return this;
+ }
+
+ public ConfigUploadComponent withProfilerConfigsPath(String profilerConfigsPath) {
+ this.profilerConfigPath = profilerConfigsPath;
+ return this;
+ }
+
+ public ConfigUploadComponent withParserSensorConfig(String name, SensorParserConfig config) {
+ parserSensorConfigs.put(name, config);
+ return this;
+ }
+
+ public ConfigUploadComponent withGlobalConfig(String globalConfig) {
+ this.globalConfig = Optional.ofNullable(globalConfig);
+ return this;
+ }
+
+ @Override
+ public void start() throws UnableToStartException {
+ try {
+ final String zookeeperUrl = topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY);
+
+ if(globalConfigPath != null) {
+ uploadConfigsToZookeeper(globalConfigPath, parserConfigsPath, enrichmentConfigsPath, profilerConfigPath, zookeeperUrl);
+ }
+
+ for(Map.Entry<String, SensorParserConfig> kv : parserSensorConfigs.entrySet()) {
+ writeSensorParserConfigToZookeeper(kv.getKey(), kv.getValue(), zookeeperUrl);
+ }
+
+ if(globalConfig.isPresent()) {
+ writeGlobalConfigToZookeeper(globalConfig.get().getBytes(), zookeeperUrl);
+ }
+
+ } catch (Exception e) {
+ throw new UnableToStartException(e.getMessage(), e);
+ }
+ }
+
+ public SensorParserConfig getSensorParserConfig(String sensorType) {
+ SensorParserConfig sensorParserConfig = new SensorParserConfig();
+ CuratorFramework client = getClient(topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY));
+ client.start();
+ try {
+ sensorParserConfig = readSensorParserConfigFromZookeeper(sensorType, client);
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ client.close();
+ }
+ return sensorParserConfig;
+ }
+
+ @Override
+ public void stop() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ProfilerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ProfilerIntegrationTest.java b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ProfilerIntegrationTest.java
new file mode 100644
index 0000000..d0d22b8
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/integration/ProfilerIntegrationTest.java
@@ -0,0 +1,226 @@
+/*
+ *
+ * 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.profiler.integration;
+
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.common.Constants;
+import org.apache.metron.common.spout.kafka.SpoutConfig;
+import org.apache.metron.hbase.TableProvider;
+import org.apache.metron.integration.BaseIntegrationTest;
+import org.apache.metron.integration.ComponentRunner;
+import org.apache.metron.integration.components.FluxTopologyComponent;
+import org.apache.metron.integration.components.KafkaWithZKComponent;
+import org.apache.metron.profiler.bolt.ProfileHBaseMapper;
+import org.apache.metron.test.mock.MockHTable;
+import org.junit.After;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.google.code.tempusfugit.temporal.Duration.seconds;
+import static com.google.code.tempusfugit.temporal.Timeout.timeout;
+import static com.google.code.tempusfugit.temporal.WaitFor.waitOrTimeout;
+
+/**
+ * An integration test of the Profiler topology.
+ */
+public class ProfilerIntegrationTest extends BaseIntegrationTest {
+
+ private static final String TEST_RESOURCES = "../../metron-analytics/metron-profiler/src/test";
+ private static final String FLUX_PATH = "../metron-profiler/src/main/flux/profiler/remote.yaml";
+
+ /**
+ * {
+ * "ip_src_addr": "10.0.0.1",
+ * "protocol": "HTTPS",
+ * "length": 10,
+ * "bytes_in": 234
+ * }
+ */
+ @Multiline
+ private String message1;
+
+ /**
+ * {
+ * "ip_src_addr": "10.0.0.2",
+ * "protocol": "HTTP",
+ * "length": 20,
+ * "bytes_in": 390
+ * }
+ */
+ @Multiline
+ private String message2;
+
+ /**
+ * {
+ * "ip_src_addr": "10.0.0.3",
+ * "protocol": "DNS",
+ * "length": 30,
+ * "bytes_in": 560
+ * }
+ */
+ @Multiline
+ private String message3;
+
+ private FluxTopologyComponent fluxComponent;
+ private KafkaWithZKComponent kafkaComponent;
+ private List<byte[]> input;
+ private ComponentRunner runner;
+ private MockHTable profilerTable;
+
+ private static final String tableName = "profiler";
+ private static final String columnFamily = "P";
+
+ /**
+ * A TableProvider that allows us to mock HBase.
+ */
+ public static class MockTableProvider implements TableProvider, Serializable {
+
+ MockHTable.Provider provider = new MockHTable.Provider();
+
+ @Override
+ public HTableInterface getTable(Configuration config, String tableName) throws IOException {
+ return provider.getTable(config, tableName);
+ }
+ }
+
+ public void setup(String pathToConfig) throws Exception {
+
+ // create input messages for the profiler to consume
+ input = Stream.of(message1, message2, message3)
+ .map(Bytes::toBytes)
+ .map(m -> Collections.nCopies(5, m))
+ .flatMap(l -> l.stream())
+ .collect(Collectors.toList());
+
+ // storm topology properties
+ final Properties topologyProperties = new Properties() {{
+ setProperty("kafka.start", SpoutConfig.Offset.BEGINNING.name());
+ setProperty("profiler.workers", "1");
+ setProperty("profiler.executors", "0");
+ setProperty("profiler.input.topic", Constants.INDEXING_TOPIC);
+ setProperty("profiler.flush.interval.seconds", "15");
+ setProperty("profiler.hbase.salt.divisor", "10");
+ setProperty("profiler.hbase.table", tableName);
+ setProperty("profiler.hbase.batch", "10");
+ setProperty("profiler.hbase.flush.interval.seconds", "2");
+ setProperty("hbase.provider.impl", "" + MockTableProvider.class.getName());
+ }};
+
+ // create the mock table
+ profilerTable = (MockHTable) MockHTable.Provider.addToCache(tableName, columnFamily);
+
+ // create the input topic
+ kafkaComponent = getKafkaComponent(topologyProperties,
+ Arrays.asList(new KafkaWithZKComponent.Topic(Constants.INDEXING_TOPIC, 1)));
+
+ // upload profiler configuration to zookeeper
+ ConfigUploadComponent configUploadComponent = new ConfigUploadComponent()
+ .withTopologyProperties(topologyProperties)
+ .withGlobalConfigsPath(pathToConfig)
+ .withProfilerConfigsPath(pathToConfig);
+
+ // load flux definition for the profiler topology
+ fluxComponent = new FluxTopologyComponent.Builder()
+ .withTopologyLocation(new File(FLUX_PATH))
+ .withTopologyName("profiler")
+ .withTopologyProperties(topologyProperties)
+ .build();
+
+ // start all components
+ runner = new ComponentRunner.Builder()
+ .withComponent("kafka", kafkaComponent)
+ .withComponent("config", configUploadComponent)
+ .withComponent("storm", fluxComponent)
+ .withMillisecondsBetweenAttempts(15000)
+ .withNumRetries(10)
+ .build();
+ runner.start();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (runner != null) {
+ runner.stop();
+ }
+ }
+
+ /**
+ * Tests the first example contained within the README.
+ */
+ @Test
+ public void testExample1() throws Exception {
+
+ setup(TEST_RESOURCES + "/config/zookeeper/readme-example-1");
+
+ // start the topology and write test messages to kafka
+ fluxComponent.submitTopology();
+ kafkaComponent.writeMessages(Constants.INDEXING_TOPIC, input);
+
+ // verify - ensure the profile is being persisted
+ waitOrTimeout(() -> profilerTable.getPutLog().size() > 0,
+ timeout(seconds(90)));
+ }
+
+ /**
+ * Tests the second example contained within the README.
+ */
+ @Test
+ public void testExample2() throws Exception {
+
+ setup(TEST_RESOURCES + "/config/zookeeper/readme-example-2");
+
+ // start the topology and write test messages to kafka
+ fluxComponent.submitTopology();
+ kafkaComponent.writeMessages(Constants.INDEXING_TOPIC, input);
+
+ // verify - ensure the profile is being persisted
+ waitOrTimeout(() -> profilerTable.getPutLog().size() > 0,
+ timeout(seconds(90)));
+ }
+
+ /**
+ * Tests the third example contained within the README.
+ */
+ @Test
+ public void testExample3() throws Exception {
+
+ setup(TEST_RESOURCES + "/config/zookeeper/readme-example-3");
+
+ // start the topology and write test messages to kafka
+ fluxComponent.submitTopology();
+ kafkaComponent.writeMessages(Constants.INDEXING_TOPIC, input);
+
+ // verify - ensure the profile is being persisted
+ waitOrTimeout(() -> profilerTable.getPutLog().size() > 0,
+ timeout(seconds(90)));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java
new file mode 100644
index 0000000..68085d5
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/util/DefaultStellarExecutorTest.java
@@ -0,0 +1,161 @@
+/*
+ *
+ * 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.profiler.util;
+
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.metron.profiler.stellar.DefaultStellarExecutor;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests the DefaultStellarExecutor.
+ */
+public class DefaultStellarExecutorTest {
+
+ /**
+ * {
+ * "ip_src_addr": "10.0.0.1",
+ * "ip_dst_addr": "10.0.0.20"
+ * }
+ */
+ @Multiline
+ private String input;
+
+ private JSONObject message;
+ private DefaultStellarExecutor executor;
+
+ @Before
+ public void setup() throws ParseException {
+
+ // parse the input message
+ JSONParser parser = new JSONParser();
+ message = (JSONObject) parser.parse(input);
+
+ // create the executor to test
+ executor = new DefaultStellarExecutor();
+ }
+
+ /**
+ * Ensure that a value can be assigned to a variable.
+ */
+ @Test
+ public void testAssign() {
+ executor.assign("foo", "2", message);
+
+ // verify
+ Object var = executor.getState().get("foo");
+ assertThat(var, instanceOf(Integer.class));
+ assertThat(var, equalTo(2));
+ }
+
+ /**
+ * Ensure that a variable can be resolved from a message field.
+ */
+ @Test
+ public void testAssignWithVariableResolution() {
+ executor.assign("foo", "ip_src_addr", message);
+
+ // verify
+ Object var = executor.getState().get("foo");
+ assertThat(var, instanceOf(String.class));
+ assertThat(var, equalTo("10.0.0.1"));
+ }
+
+ /**
+ * Ensure that state is maintained correctly in the execution environment.
+ */
+ @Test
+ public void testState() {
+ executor.assign("two", "2", message);
+ executor.assign("four", "4", message);
+ executor.assign("sum", "two + four", message);
+
+ // verify
+ Object var = executor.getState().get("sum");
+ assertEquals(6.0, var);
+ }
+
+ /**
+ * Ensure that state is maintained correctly in the execution environment.
+ */
+ @Test
+ public void testClearState() {
+ executor.assign("two", "2", message);
+ executor.clearState();
+
+ // verify
+ assertThat(executor.getState().containsKey("two"), equalTo(false));
+ }
+
+ /**
+ * Ensure that a Transformation function can be executed.
+ *
+ * There are two sets of functions in Stellar currently. One can be executed with
+ * a PredicateProcessor and the other a TransformationProcessor. The StellarExecutor
+ * abstracts away that complication.
+ */
+ @Test
+ public void testExecuteTransformation() {
+ String actual = executor.execute("TO_UPPER('lowercase')", message, String.class);
+ assertThat(actual, equalTo("LOWERCASE"));
+ }
+
+ /**
+ * Ensure that a Predicate function can be executed.
+ *
+ * There are two sets of functions in Stellar currently. One can be executed with
+ * a PredicateProcessor and the other a TransformationProcessor. The StellarExecutor
+ * abstracts away that complication.
+ */
+ @Test
+ public void testExecutePredicate() {
+ boolean actual = executor.execute("IS_INTEGER(2)", message, Boolean.class);
+ assertThat(actual, equalTo(true));
+ }
+
+ /**
+ * An exception is expected if an expression results in an unexpected type.
+ */
+ @Test(expected = RuntimeException.class)
+ public void testExecuteWithWrongType() {
+ executor.execute("2 + 2", message, Boolean.class);
+ }
+
+ /**
+ * A best effort should be made to do sensible type conversions.
+ */
+ @Test
+ public void testExecuteWithTypeConversion() {
+ executor.execute("2", message, Double.class);
+ executor.execute("2", message, Float.class);
+ executor.execute("2", message, Short.class);
+ executor.execute("2", message, Long.class);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/pom.xml
----------------------------------------------------------------------
diff --git a/metron-analytics/pom.xml b/metron-analytics/pom.xml
index 8930f5f..3b4f5bd 100644
--- a/metron-analytics/pom.xml
+++ b/metron-analytics/pom.xml
@@ -32,7 +32,6 @@
<tag>HEAD</tag>
<url>https://git-wip-us.apache.org/repos/asf/incubator-metron</url>
</scm>
-
<licenses>
<license>
<name>The Apache Software License, Version 2.0</name>
@@ -43,6 +42,7 @@
<modules>
<module>metron-maas-service</module>
<module>metron-maas-common</module>
+ <module>metron-profiler</module>
</modules>
<dependencies>
<dependency>
@@ -60,7 +60,6 @@
</dependencies>
<build>
<plugins>
-
</plugins>
</build>
<reporting>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-deployment/roles/metron_streaming/defaults/main.yml
----------------------------------------------------------------------
diff --git a/metron-deployment/roles/metron_streaming/defaults/main.yml b/metron-deployment/roles/metron_streaming/defaults/main.yml
index 593c0a8..b048bf0 100644
--- a/metron-deployment/roles/metron_streaming/defaults/main.yml
+++ b/metron-deployment/roles/metron_streaming/defaults/main.yml
@@ -21,7 +21,9 @@ metron_elasticsearch_jar_name: metron-elasticsearch-{{ metron_version }}.jar
metron_pcap_jar_name: metron-pcap-backend-{{ metron_version }}.jar
metron_parsers_jar_name: metron-parsers-{{ metron_version }}.jar
metron_enrichment_jar_name: metron-enrichment-{{ metron_version }}-uber.jar
+metron_profiler_jar_name: metron-profiler-{{ metron_version }}.jar
+# bundle names
metron_common_bundle_name: metron-common-{{ metron_version }}-archive.tar.gz
metron_pcap_bundle_name: metron-pcap-backend-{{ metron_version }}-archive.tar.gz
metron_data_management_bundle_name: metron-data-management-{{ metron_version }}-archive.tar.gz
@@ -31,6 +33,9 @@ metron_solr_bundle_name: metron-solr-{{ metron_version }}-archive.tar.gz
metron_elasticsearch_bundle_name: metron-elasticsearch-{{ metron_version }}-archive.tar.gz
metron_parsers_bundle_name: metron-parsers-{{ metron_version }}-archive.tar.gz
metron_maas_bundle_name: metron-maas-service-{{ metron_version }}-archive.tar.gz
+metron_profiler_bundle_name: metron-profiler-{{ metron_version }}-archive.tar.gz
+
+# bundle paths
metron_pcap_bundle_path: "{{ playbook_dir }}/../../metron-platform/metron-pcap-backend/target/{{ metron_pcap_bundle_name }}"
metron_common_bundle_path: "{{ playbook_dir }}/../../metron-platform/metron-common/target/{{ metron_common_bundle_name }}"
metron_data_management_bundle_path: "{{ playbook_dir }}/../../metron-platform/metron-data-management/target/{{ metron_data_management_bundle_name }}"
@@ -40,7 +45,9 @@ metron_solr_bundle_path: "{{ playbook_dir }}/../../metron-platform/metron-solr/t
metron_elasticsearch_bundle_path: "{{ playbook_dir }}/../../metron-platform/metron-elasticsearch/target/{{ metron_elasticsearch_bundle_name }}"
metron_parsers_bundle_path: "{{ playbook_dir }}/../../metron-platform/metron-parsers/target/{{ metron_parsers_bundle_name }}"
metron_maas_bundle_path: "{{ playbook_dir }}/../../metron-analytics/metron-maas-service/target/{{ metron_maas_bundle_name }}"
+metron_profiler_bundle_path: "{{ playbook_dir }}/../../metron-analytics/metron-profiler/target/{{ metron_profiler_bundle_name }}"
+# configuration paths
config_path: "{{ metron_directory }}/config"
zookeeper_config_path: "{{ config_path }}/zookeeper"
zookeeper_global_config_path: "{{ zookeeper_config_path }}/global.json"
@@ -49,6 +56,8 @@ metron_solr_properties_config_path: "{{ metron_directory }}/config/solr.properti
metron_elasticsearch_properties_config_path: "{{ metron_directory }}/config/elasticsearch.properties"
metron_enrichment_properties_config_path: "{{ metron_directory }}/config/enrichment.properties"
metron_parsers_properties_config_path: "{{ metron_directory }}/config/parsers.properties"
+metron_profiler_properties_config_path: "{{ metron_directory }}/config/profiler.properties"
+
hbase_config_path: "/etc/hbase/conf"
hdfs_config_path: "/etc/hadoop/conf"
pcap_hdfs_path: "/apps/metron/pcap"
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-deployment/roles/metron_streaming/tasks/copy_bundles.yml
----------------------------------------------------------------------
diff --git a/metron-deployment/roles/metron_streaming/tasks/copy_bundles.yml b/metron-deployment/roles/metron_streaming/tasks/copy_bundles.yml
index 12dd411..ca972cb 100644
--- a/metron-deployment/roles/metron_streaming/tasks/copy_bundles.yml
+++ b/metron-deployment/roles/metron_streaming/tasks/copy_bundles.yml
@@ -35,7 +35,6 @@
src: "{{ metron_indexing_bundle_path }}"
dest: "{{ metron_directory }}"
-
- name: Copy Metron Parsers bundle
copy:
src: "{{ metron_parsers_bundle_path }}"
@@ -61,6 +60,11 @@
src: "{{ metron_maas_bundle_path }}"
dest: "{{ metron_directory }}"
+- name: Copy Metron Profiler bundle
+ copy:
+ src: "{{ metron_profiler_bundle_path }}"
+ dest: "{{ metron_directory }}"
+
- name: Unbundle Metron bundles
shell: "{{ item }}"
args:
@@ -75,4 +79,5 @@
- tar xzvf metron-common*.tar.gz
- tar xzvf metron-pcap-backend*.tar.gz
- tar xzvf metron-maas-service*.tar.gz
+ - tar xzvf metron-profiler*.tar.gz
- rm *.tar.gz
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-deployment/roles/metron_streaming/tasks/topologies.yml
----------------------------------------------------------------------
diff --git a/metron-deployment/roles/metron_streaming/tasks/topologies.yml b/metron-deployment/roles/metron_streaming/tasks/topologies.yml
index 0ca0b3d..7bb2bb1 100644
--- a/metron-deployment/roles/metron_streaming/tasks/topologies.yml
+++ b/metron-deployment/roles/metron_streaming/tasks/topologies.yml
@@ -78,3 +78,15 @@
- { regexp: "bolt.hdfs.rotation.policy=", line: "bolt.hdfs.rotation.policy={{ metron_hdfs_rotation_policy }}" }
- { regexp: "bolt.hdfs.rotation.policy.count=", line: "bolt.hdfs.rotation.policy.count={{ metron_hdfs_rotation_policy_count}}" }
- { regexp: "bolt.hdfs.rotation.policy.units=", line: "bolt.hdfs.rotation.policy.units={{ metron_hdfs_rotation_policy_units }}" }
+
+- name: Configure Profiler topology
+ lineinfile: >
+ dest={{ metron_profiler_properties_config_path }}
+ regexp="{{ item.regexp }}"
+ line="{{ item.line }}"
+ with_items:
+ - { regexp: "kafka.zk=", line: "kafka.zk={{ zookeeper_url }}" }
+ - { regexp: "kafka.broker=", line: "kafka.broker={{ kafka_broker_url }}" }
+
+
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java
index 6fed7d4..dedf681 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java
@@ -18,7 +18,6 @@
package org.apache.metron.common.bolt;
import org.apache.log4j.Logger;
-import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.ConfigurationType;
import org.apache.metron.common.configuration.ConfigurationsUtils;
import org.apache.metron.common.configuration.EnrichmentConfigurations;
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredProfilerBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredProfilerBolt.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredProfilerBolt.java
new file mode 100644
index 0000000..bc1e5d6
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredProfilerBolt.java
@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.common.bolt;
+
+import org.apache.log4j.Logger;
+import org.apache.metron.common.configuration.ConfigurationType;
+import org.apache.metron.common.configuration.ConfigurationsUtils;
+import org.apache.metron.common.configuration.profiler.ProfilerConfig;
+import org.apache.metron.common.configuration.profiler.ProfilerConfigurations;
+
+import java.io.IOException;
+
+/**
+ * A bolt used in the Profiler topology that is configured with values stored in Zookeeper.
+ */
+public abstract class ConfiguredProfilerBolt extends ConfiguredBolt<ProfilerConfigurations> {
+
+ private static final Logger LOG = Logger.getLogger(ConfiguredProfilerBolt.class);
+ protected final ProfilerConfigurations configurations = new ProfilerConfigurations();
+
+ public ConfiguredProfilerBolt(String zookeeperUrl) {
+ super(zookeeperUrl);
+ }
+
+ protected ProfilerConfig getProfilerConfig() {
+ return getConfigurations().getProfilerConfig();
+ }
+
+ @Override
+ protected ProfilerConfigurations defaultConfigurations() {
+ return new ProfilerConfigurations();
+ }
+
+ @Override
+ public void loadConfig() {
+ try {
+ ConfigurationsUtils.updateProfilerConfigsFromZookeeper(getConfigurations(), client);
+ } catch (Exception e) {
+ LOG.warn("Unable to load configs from zookeeper, but the cache should load lazily...");
+ }
+ }
+
+ @Override
+ public void updateConfig(String path, byte[] data) throws IOException {
+ if (data.length != 0) {
+ String name = path.substring(path.lastIndexOf("/") + 1);
+
+ // update the profiler configuration from zookeeper
+ if (path.startsWith(ConfigurationType.PROFILER.getZookeeperRoot())) {
+ getConfigurations().updateProfilerConfig(data);
+ reloadCallback(name, ConfigurationType.PROFILER);
+
+ // update the global configuration from zookeeper
+ } else if (ConfigurationType.GLOBAL.getZookeeperRoot().equals(path)) {
+ getConfigurations().updateGlobalConfig(data);
+ reloadCallback(name, ConfigurationType.GLOBAL);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationType.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationType.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationType.java
index 9469bf6..27478d4 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationType.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationType.java
@@ -22,15 +22,15 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Function;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.configuration.profiler.ProfilerConfig;
import org.apache.metron.common.utils.JSONUtils;
import java.io.IOException;
import java.util.Map;
public enum ConfigurationType implements Function<String, Object> {
- GLOBAL("global"
- ,"."
- , s -> {
+
+ GLOBAL("global",".", s -> {
try {
return JSONUtils.INSTANCE.load(s, new TypeReference<Map<String, Object>>() {
});
@@ -38,28 +38,36 @@ public enum ConfigurationType implements Function<String, Object> {
throw new RuntimeException("Unable to load " + s, e);
}
}),
- PARSER("parsers"
- ,"parsers"
- , s -> {
+
+ PARSER("parsers","parsers", s -> {
try {
return JSONUtils.INSTANCE.load(s, SensorParserConfig.class);
} catch (IOException e) {
throw new RuntimeException("Unable to load " + s, e);
}
}),
- ENRICHMENT("enrichments"
- ,"enrichments"
- , s -> {
+
+ ENRICHMENT("enrichments","enrichments", s -> {
try {
return JSONUtils.INSTANCE.load(s, SensorEnrichmentConfig.class);
} catch (IOException e) {
throw new RuntimeException("Unable to load " + s, e);
}
+ }),
+
+ PROFILER("profiler","profiler", s -> {
+ try {
+ return JSONUtils.INSTANCE.load(s, ProfilerConfig.class);
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to load " + s, e);
+ }
});
+
String name;
String directory;
String zookeeperRoot;
Function<String,?> deserializer;
+
ConfigurationType(String name, String directory, Function<String, ?> deserializer) {
this.name = name;
this.directory = directory;
@@ -67,16 +75,18 @@ public enum ConfigurationType implements Function<String, Object> {
this.deserializer = deserializer;
}
- public String getName() { return name; }
+ public String getName() {
+ return name;
+ }
public String getDirectory() {
return directory;
}
- public Object deserialize(String s)
- {
+ public Object deserialize(String s) {
return deserializer.apply(s);
}
+
@Override
public Object apply(String s) {
return deserialize(s);
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java
index 56ac9d9..1411f9c 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java
@@ -24,6 +24,7 @@ import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.metron.common.Constants;
import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.configuration.profiler.ProfilerConfigurations;
import org.apache.metron.common.utils.JSONUtils;
import org.apache.zookeeper.KeeperException;
@@ -36,6 +37,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.metron.common.configuration.ConfigurationType.*;
+
public class ConfigurationsUtils {
public static CuratorFramework getClient(String zookeeperUrl) {
@@ -65,8 +68,13 @@ public class ConfigurationsUtils {
}
public static void writeGlobalConfigToZookeeper(byte[] globalConfig, CuratorFramework client) throws Exception {
- ConfigurationType.GLOBAL.deserialize(new String(globalConfig));
- writeToZookeeper(ConfigurationType.GLOBAL.getZookeeperRoot(), globalConfig, client);
+ GLOBAL.deserialize(new String(globalConfig));
+ writeToZookeeper(GLOBAL.getZookeeperRoot(), globalConfig, client);
+ }
+
+ public static void writeProfilerConfigToZookeeper(byte[] config, CuratorFramework client) throws Exception {
+ PROFILER.deserialize(new String(config));
+ writeToZookeeper(PROFILER.getZookeeperRoot(), config, client);
}
public static void writeSensorParserConfigToZookeeper(String sensorType, SensorParserConfig sensorParserConfig, String zookeeperUrl) throws Exception {
@@ -85,7 +93,7 @@ public class ConfigurationsUtils {
}
public static void writeSensorParserConfigToZookeeper(String sensorType, byte[] configData, CuratorFramework client) throws Exception {
- writeToZookeeper(ConfigurationType.PARSER.getZookeeperRoot() + "/" + sensorType, configData, client);
+ writeToZookeeper(PARSER.getZookeeperRoot() + "/" + sensorType, configData, client);
}
public static void writeSensorEnrichmentConfigToZookeeper(String sensorType, SensorEnrichmentConfig sensorEnrichmentConfig, String zookeeperUrl) throws Exception {
@@ -104,7 +112,7 @@ public class ConfigurationsUtils {
}
public static void writeSensorEnrichmentConfigToZookeeper(String sensorType, byte[] configData, CuratorFramework client) throws Exception {
- writeToZookeeper(ConfigurationType.ENRICHMENT.getZookeeperRoot() + "/" + sensorType, configData, client);
+ writeToZookeeper(ENRICHMENT.getZookeeperRoot() + "/" + sensorType, configData, client);
}
public static void writeConfigToZookeeper(String name, Map<String, Object> config, String zookeeperUrl) throws Exception {
@@ -136,7 +144,7 @@ public class ConfigurationsUtils {
public static void updateParserConfigsFromZookeeper(ParserConfigurations configurations, CuratorFramework client) throws Exception {
updateConfigsFromZookeeper(configurations, client);
- List<String> sensorTypes = client.getChildren().forPath(ConfigurationType.PARSER.getZookeeperRoot());
+ List<String> sensorTypes = client.getChildren().forPath(PARSER.getZookeeperRoot());
for(String sensorType: sensorTypes) {
configurations.updateSensorParserConfig(sensorType, readSensorParserConfigBytesFromZookeeper(sensorType, client));
}
@@ -144,30 +152,38 @@ public class ConfigurationsUtils {
public static void updateEnrichmentConfigsFromZookeeper(EnrichmentConfigurations configurations, CuratorFramework client) throws Exception {
updateConfigsFromZookeeper(configurations, client);
- List<String> sensorTypes = client.getChildren().forPath(ConfigurationType.ENRICHMENT.getZookeeperRoot());
+ List<String> sensorTypes = client.getChildren().forPath(ENRICHMENT.getZookeeperRoot());
for(String sensorType: sensorTypes) {
configurations.updateSensorEnrichmentConfig(sensorType, readSensorEnrichmentConfigBytesFromZookeeper(sensorType, client));
}
}
+ public static void updateProfilerConfigsFromZookeeper(ProfilerConfigurations configurations, CuratorFramework client) throws Exception {
+ updateConfigsFromZookeeper(configurations, client);
+ }
+
public static SensorEnrichmentConfig readSensorEnrichmentConfigFromZookeeper(String sensorType, CuratorFramework client) throws Exception {
- return JSONUtils.INSTANCE.load(new ByteArrayInputStream(readFromZookeeper(ConfigurationType.ENRICHMENT.getZookeeperRoot() + "/" + sensorType, client)), SensorEnrichmentConfig.class);
+ return JSONUtils.INSTANCE.load(new ByteArrayInputStream(readFromZookeeper(ENRICHMENT.getZookeeperRoot() + "/" + sensorType, client)), SensorEnrichmentConfig.class);
}
public static SensorParserConfig readSensorParserConfigFromZookeeper(String sensorType, CuratorFramework client) throws Exception {
- return JSONUtils.INSTANCE.load(new ByteArrayInputStream(readFromZookeeper(ConfigurationType.PARSER.getZookeeperRoot() + "/" + sensorType, client)), SensorParserConfig.class);
+ return JSONUtils.INSTANCE.load(new ByteArrayInputStream(readFromZookeeper(PARSER.getZookeeperRoot() + "/" + sensorType, client)), SensorParserConfig.class);
}
public static byte[] readGlobalConfigBytesFromZookeeper(CuratorFramework client) throws Exception {
- return readFromZookeeper(ConfigurationType.GLOBAL.getZookeeperRoot(), client);
+ return readFromZookeeper(GLOBAL.getZookeeperRoot(), client);
+ }
+
+ public static byte[] readProfilerConfigBytesFromZookeeper(CuratorFramework client) throws Exception {
+ return readFromZookeeper(PROFILER.getZookeeperRoot(), client);
}
public static byte[] readSensorParserConfigBytesFromZookeeper(String sensorType, CuratorFramework client) throws Exception {
- return readFromZookeeper(ConfigurationType.PARSER.getZookeeperRoot() + "/" + sensorType, client);
+ return readFromZookeeper(PARSER.getZookeeperRoot() + "/" + sensorType, client);
}
public static byte[] readSensorEnrichmentConfigBytesFromZookeeper(String sensorType, CuratorFramework client) throws Exception {
- return readFromZookeeper(ConfigurationType.ENRICHMENT.getZookeeperRoot() + "/" + sensorType, client);
+ return readFromZookeeper(ENRICHMENT.getZookeeperRoot() + "/" + sensorType, client);
}
public static byte[] readConfigBytesFromZookeeper(String name, CuratorFramework client) throws Exception {
@@ -178,41 +194,63 @@ public class ConfigurationsUtils {
return client.getData().forPath(path);
}
- public static void uploadConfigsToZookeeper(String globalConfigPath, String parsersConfigPath, String enrichmentsConfigPath, String zookeeperUrl) throws Exception {
+ public static void uploadConfigsToZookeeper(String globalConfigPath,
+ String parsersConfigPath,
+ String enrichmentsConfigPath,
+ String profilerConfigPath,
+ String zookeeperUrl) throws Exception {
try(CuratorFramework client = getClient(zookeeperUrl)) {
client.start();
- uploadConfigsToZookeeper(globalConfigPath, parsersConfigPath, enrichmentsConfigPath, client);
+ uploadConfigsToZookeeper(globalConfigPath, parsersConfigPath, enrichmentsConfigPath, profilerConfigPath, client);
}
}
public static void uploadConfigsToZookeeper(String rootFilePath, CuratorFramework client) throws Exception {
- uploadConfigsToZookeeper(rootFilePath, rootFilePath, rootFilePath, client);
+ uploadConfigsToZookeeper(rootFilePath, rootFilePath, rootFilePath, rootFilePath, client);
}
- public static void uploadConfigsToZookeeper(String globalConfigPath, String parsersConfigPath, String enrichmentsConfigPath, CuratorFramework client) throws Exception {
+ public static void uploadConfigsToZookeeper(String globalConfigPath,
+ String parsersConfigPath,
+ String enrichmentsConfigPath,
+ String profilerConfigPath,
+ CuratorFramework client) throws Exception {
+
+ // global
if (globalConfigPath != null) {
byte[] globalConfig = readGlobalConfigFromFile(globalConfigPath);
if (globalConfig.length > 0) {
ConfigurationsUtils.writeGlobalConfigToZookeeper(readGlobalConfigFromFile(globalConfigPath), client);
}
}
+
+ // parsers
if (parsersConfigPath != null) {
Map<String, byte[]> sensorParserConfigs = readSensorParserConfigsFromFile(parsersConfigPath);
for (String sensorType : sensorParserConfigs.keySet()) {
ConfigurationsUtils.writeSensorParserConfigToZookeeper(sensorType, sensorParserConfigs.get(sensorType), client);
}
}
+
+ // enrichments
if (enrichmentsConfigPath != null) {
Map<String, byte[]> sensorEnrichmentConfigs = readSensorEnrichmentConfigsFromFile(enrichmentsConfigPath);
for (String sensorType : sensorEnrichmentConfigs.keySet()) {
ConfigurationsUtils.writeSensorEnrichmentConfigToZookeeper(sensorType, sensorEnrichmentConfigs.get(sensorType), client);
}
}
+
+ // profiler
+ if (profilerConfigPath != null) {
+ byte[] globalConfig = readProfilerConfigFromFile(profilerConfigPath);
+ if (globalConfig.length > 0) {
+ ConfigurationsUtils.writeProfilerConfigToZookeeper(readProfilerConfigFromFile(profilerConfigPath), client);
+ }
+ }
}
public static byte[] readGlobalConfigFromFile(String rootPath) throws IOException {
byte[] globalConfig = new byte[0];
- File configPath = new File(rootPath, ConfigurationType.GLOBAL.getName() + ".json");
+ File configPath = new File(rootPath, GLOBAL.getName() + ".json");
if (configPath.exists()) {
globalConfig = Files.readAllBytes(configPath.toPath());
}
@@ -220,11 +258,27 @@ public class ConfigurationsUtils {
}
public static Map<String, byte[]> readSensorParserConfigsFromFile(String rootPath) throws IOException {
- return readSensorConfigsFromFile(rootPath, ConfigurationType.PARSER);
+ return readSensorConfigsFromFile(rootPath, PARSER);
}
public static Map<String, byte[]> readSensorEnrichmentConfigsFromFile(String rootPath) throws IOException {
- return readSensorConfigsFromFile(rootPath, ConfigurationType.ENRICHMENT);
+ return readSensorConfigsFromFile(rootPath, ENRICHMENT);
+ }
+
+
+ /**
+ * Read the Profiler configuration from a file. There is only a single profiler configuration.
+ * @param rootPath Path to the Profiler configuration.
+ */
+ public static byte[] readProfilerConfigFromFile(String rootPath) throws IOException {
+
+ byte[] config = new byte[0];
+ File configPath = new File(rootPath, PROFILER.getName() + ".json");
+ if (configPath.exists()) {
+ config = Files.readAllBytes(configPath.toPath());
+ }
+
+ return config;
}
public static Map<String, byte[]> readSensorConfigsFromFile(String rootPath, ConfigurationType configType) throws IOException {
@@ -247,19 +301,24 @@ public class ConfigurationsUtils {
}
public static void visitConfigs(CuratorFramework client, ConfigurationVisitor callback) throws Exception {
- visitConfigs(client, callback, ConfigurationType.GLOBAL);
- visitConfigs(client, callback, ConfigurationType.PARSER);
- visitConfigs(client, callback, ConfigurationType.ENRICHMENT);
+ visitConfigs(client, callback, GLOBAL);
+ visitConfigs(client, callback, PARSER);
+ visitConfigs(client, callback, ENRICHMENT);
+ visitConfigs(client, callback, PROFILER);
}
public static void visitConfigs(CuratorFramework client, ConfigurationVisitor callback, ConfigurationType configType) throws Exception {
+
if (client.checkExists().forPath(configType.getZookeeperRoot()) != null) {
- if (configType.equals(ConfigurationType.GLOBAL)) {
+
+ if (configType.equals(GLOBAL)) {
byte[] globalConfigData = client.getData().forPath(configType.getZookeeperRoot());
callback.visit(configType, "global", new String(globalConfigData));
- } else if (configType.equals(ConfigurationType.PARSER) || configType.equals(ConfigurationType.ENRICHMENT)) {
+
+ } else if (configType.equals(PARSER) || configType.equals(ENRICHMENT) || configType.equals(PROFILER)) {
List<String> children = client.getChildren().forPath(configType.getZookeeperRoot());
for (String child : children) {
+
byte[] data = client.getData().forPath(configType.getZookeeperRoot() + "/" + child);
callback.visit(configType, child, new String(data));
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfileConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfileConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfileConfig.java
new file mode 100644
index 0000000..f832e1d
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfileConfig.java
@@ -0,0 +1,145 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.common.configuration.profiler;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The user defined configuration values required to generate a Profile.
+ */
+public class ProfileConfig implements Serializable {
+
+ /**
+ * The name of the profile.
+ */
+ private String profile;
+
+ /**
+ * Stella code that when executed results in the name of the entity being profiled. A
+ * profile is created 'for-each' of these; hence the name.
+ */
+ private String foreach;
+
+ /**
+ * Stella code that when executed determines whether a message should be included in this
+ * profile.
+ */
+ private String onlyif;
+
+ /**
+ * Stella code that when executed results in a single measurement that is stored with the Profile.
+ */
+ private String result;
+
+ /**
+ * Defines how the state is initialized before any messages are received.
+ */
+ private Map<String, String> init = new HashMap<>();
+
+ /**
+ * Defines how the state is updated when a new message is received.
+ */
+ private Map<String, String> update = new HashMap<>();
+
+ public String getProfile() {
+ return profile;
+ }
+
+ public void setProfile(String profile) {
+ this.profile = profile;
+ }
+
+ public String getForeach() {
+ return foreach;
+ }
+
+ public void setForeach(String foreach) {
+ this.foreach = foreach;
+ }
+
+ public String getOnlyif() {
+ return onlyif;
+ }
+
+ public void setOnlyif(String onlyif) {
+ this.onlyif = onlyif;
+ }
+
+ public String getResult() {
+ return result;
+ }
+
+ public void setResult(String result) {
+ this.result = result;
+ }
+
+ public Map<String, String> getInit() {
+ return init;
+ }
+
+ public void setInit(Map<String, String> init) {
+ this.init = init;
+ }
+
+ public Map<String, String> getUpdate() {
+ return update;
+ }
+
+ public void setUpdate(Map<String, String> update) {
+ this.update = update;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ProfileConfig that = (ProfileConfig) o;
+ if (profile != null ? !profile.equals(that.profile) : that.profile != null) return false;
+ if (foreach != null ? !foreach.equals(that.foreach) : that.foreach != null) return false;
+ if (onlyif != null ? !onlyif.equals(that.onlyif) : that.onlyif != null) return false;
+ if (result != null ? !result.equals(that.result) : that.result != null) return false;
+ if (init != null ? !init.equals(that.init) : that.init != null) return false;
+ return update != null ? update.equals(that.update) : that.update == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result1 = profile != null ? profile.hashCode() : 0;
+ result1 = 31 * result1 + (foreach != null ? foreach.hashCode() : 0);
+ result1 = 31 * result1 + (onlyif != null ? onlyif.hashCode() : 0);
+ result1 = 31 * result1 + (result != null ? result.hashCode() : 0);
+ result1 = 31 * result1 + (init != null ? init.hashCode() : 0);
+ result1 = 31 * result1 + (update != null ? update.hashCode() : 0);
+ return result1;
+ }
+
+ @Override
+ public String toString() {
+ return "ProfileConfig{" +
+ "profile='" + profile + '\'' +
+ ", foreach='" + foreach + '\'' +
+ ", onlyif='" + onlyif + '\'' +
+ ", result='" + result + '\'' +
+ ", init=" + init +
+ ", update=" + update +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfig.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfig.java
new file mode 100644
index 0000000..8263d39
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfig.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.common.configuration.profiler;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The user defined configuration values required for the Profiler.
+ */
+public class ProfilerConfig implements Serializable {
+
+ /**
+ * The input topic from which messages will be read.
+ */
+ private String inputTopic;
+
+ /**
+ * One or more profile definitions.
+ */
+ private List<ProfileConfig> profiles = new ArrayList<>();
+
+ public String getInputTopic() {
+ return inputTopic;
+ }
+
+ public void setInputTopic(String inputTopic) {
+ this.inputTopic = inputTopic;
+ }
+
+ public List<ProfileConfig> getProfiles() {
+ return profiles;
+ }
+
+ public void setProfiles(List<ProfileConfig> profiles) {
+ this.profiles = profiles;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ProfilerConfig that = (ProfilerConfig) o;
+
+ if (inputTopic != null ? !inputTopic.equals(that.inputTopic) : that.inputTopic != null) return false;
+ return profiles != null ? profiles.equals(that.profiles) : that.profiles == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = inputTopic != null ? inputTopic.hashCode() : 0;
+ result = 31 * result + (profiles != null ? profiles.hashCode() : 0);
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfigurations.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfigurations.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfigurations.java
new file mode 100644
index 0000000..19b6a5a
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/profiler/ProfilerConfigurations.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.common.configuration.profiler;
+
+import org.apache.metron.common.configuration.ConfigurationType;
+import org.apache.metron.common.configuration.Configurations;
+import org.apache.metron.common.utils.JSONUtils;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Created by nallen on 7/28/16.
+ */
+public class ProfilerConfigurations extends Configurations {
+
+ public ProfilerConfig getProfilerConfig() {
+ return (ProfilerConfig) configurations.get(getKey());
+ }
+
+ public void updateProfilerConfig(byte[] data) throws IOException {
+ updateProfilerConfig(new ByteArrayInputStream(data));
+ }
+
+ public void updateProfilerConfig(InputStream io) throws IOException {
+ ProfilerConfig config = JSONUtils.INSTANCE.load(io, ProfilerConfig.class);
+ updateProfilerConfig(config);
+ }
+
+ public void updateProfilerConfig(ProfilerConfig config) {
+ configurations.put(getKey(), config);
+ }
+
+ private String getKey() {
+ return ConfigurationType.PROFILER.getName();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
index b03c66f..a178851 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/dsl/MapVariableResolver.java
@@ -23,7 +23,9 @@ import java.util.List;
import java.util.Map;
public class MapVariableResolver implements VariableResolver {
+
List<Map> variableMappings = new ArrayList<>();
+
public MapVariableResolver(Map variableMappingOne, Map... variableMapping) {
variableMappings.add(variableMappingOne);
for(Map m : variableMapping) {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
index 0ce374b..78ba2de 100644
--- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/stellar/BaseStellarProcessor.java
@@ -61,6 +61,7 @@ public class BaseStellarProcessor<T> {
public boolean validate(String rule) throws ParseException {
return validate(rule, true);
}
+
public boolean validate(String rule, boolean throwException) throws ParseException {
try {
parse(rule, x -> null, StellarFunctions.FUNCTION_RESOLVER(), Context.EMPTY_CONTEXT());
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/HBaseBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/HBaseBolt.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/HBaseBolt.java
deleted file mode 100644
index 1eff028..0000000
--- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/bolt/HBaseBolt.java
+++ /dev/null
@@ -1,184 +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.enrichment.bolt;
-
-
-
-import java.io.IOException;
-import java.util.Map;
-
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.base.Splitter;
-import com.google.common.collect.Iterables;
-import org.apache.commons.lang.StringUtils;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.log4j.Logger;
-import org.apache.metron.hbase.Connector;
-import org.apache.metron.hbase.HTableConnector;
-import org.apache.metron.hbase.TupleTableConfig;
-import org.json.simple.JSONObject;
-
-import backtype.storm.task.OutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.IRichBolt;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Tuple;
-import backtype.storm.tuple.Values;
-
-import org.apache.metron.common.utils.ErrorUtils;
-
-/**
- * A Storm bolt for putting data into HBase.
- * <p>
- * By default works in batch mode by enabling HBase's client-side write buffer. Enabling batch mode
- * is recommended for high throughput, but it can be disabled in {@link TupleTableConfig}.
- * <p>
- * The HBase configuration is picked up from the first <tt>hbase-site.xml</tt> encountered in the
- * classpath
- * @see TupleTableConfig
- * @see HTableConnector
- */
-@SuppressWarnings("serial")
-public class HBaseBolt implements IRichBolt {
- private static final Logger LOG = Logger.getLogger(HBaseBolt.class);
- private static final String DEFAULT_ZK_PORT = "2181";
-
- protected OutputCollector collector;
- protected TupleTableConfig conf;
- protected boolean autoAck = true;
- protected Connector connector;
- private String _quorum;
- private String _port;
-
- public HBaseBolt(TupleTableConfig conf, String quorum, String port) {
- this.conf = conf;
- _quorum = quorum;
- _port = port;
- }
- public HBaseBolt(final TupleTableConfig conf, String zkConnectString) throws IOException {
- this(conf, zkConnectStringToHosts(zkConnectString), zkConnectStringToPort(zkConnectString));
- }
- public static String zkConnectStringToHosts(String connString) {
- Iterable<String> hostPortPairs = Splitter.on(',').split(connString);
- return Joiner.on(',').join(Iterables.transform(hostPortPairs, new Function<String, String>() {
-
- @Override
- public String apply(String hostPortPair) {
- return Iterables.getFirst(Splitter.on(':').split(hostPortPair), "");
- }
- }));
- }
- public static String zkConnectStringToPort(String connString) {
- String hostPortPair = Iterables.getFirst(Splitter.on(",").split(connString), "");
- return Iterables.getLast(Splitter.on(":").split(hostPortPair),DEFAULT_ZK_PORT);
- }
-
-
- public Connector createConnector() throws IOException{
- initialize();
- return new HTableConnector(conf, _quorum, _port);
- }
-
- public void initialize() {
- TupleTableConfig hbaseBoltConfig = conf;
- String allColumnFamiliesColumnQualifiers = conf.getFields();
- String[] tokenizedColumnFamiliesWithColumnQualifiers = StringUtils
- .split(allColumnFamiliesColumnQualifiers, "\\|");
- for (String tokenizedColumnFamilyWithColumnQualifiers : tokenizedColumnFamiliesWithColumnQualifiers) {
- String[] cfCqTokens = StringUtils.split( tokenizedColumnFamilyWithColumnQualifiers, ":");
- String columnFamily = cfCqTokens[0];
- String[] columnQualifiers = StringUtils.split(cfCqTokens[1], ",");
- for (String columnQualifier : columnQualifiers) {
- hbaseBoltConfig.addColumn(columnFamily, columnQualifier);
- }
- setAutoAck(true);
- }
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings("rawtypes")
-
- public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
- this.collector = collector;
-
- try {
- if(connector == null) {
- this.connector = createConnector();
- }
-
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- LOG.info("Preparing HBaseBolt for table: " + this.conf.getTableName());
- }
-
- /** {@inheritDoc} */
-
- public void execute(Tuple input) {
- try {
- Put p = conf.getPutFromTuple(input);
- this.connector.put(p);
- } catch (IOException ex) {
-
- JSONObject error = ErrorUtils.generateErrorMessage(
- "Alerts problem: " + input.toString(), ex);
- collector.emit("error", new Values(error));
-
- throw new RuntimeException(ex);
- }
-
- if (this.autoAck) {
- this.collector.ack(input);
- }
- }
-
- /** {@inheritDoc} */
-
- public void cleanup() {
- this.connector.close();
- }
-
- /** {@inheritDoc} */
-
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declareStream("error", new Fields("HBase"));
- }
-
- /** {@inheritDoc} */
-
- public Map<String, Object> getComponentConfiguration() {
- return null;
- }
-
- /**
- * @return the autoAck
- */
- public boolean isAutoAck() {
- return autoAck;
- }
-
- /**
- * @param autoAck the autoAck to set
- */
- public void setAutoAck(boolean autoAck) {
- this.autoAck = autoAck;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/components/ConfigUploadComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/components/ConfigUploadComponent.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/components/ConfigUploadComponent.java
index 5cc14ac..97effac 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/components/ConfigUploadComponent.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/components/ConfigUploadComponent.java
@@ -18,7 +18,6 @@
package org.apache.metron.enrichment.integration.components;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.metron.common.configuration.ConfigurationsUtils;
import org.apache.metron.common.configuration.SensorParserConfig;
import org.apache.metron.integration.InMemoryComponent;
import org.apache.metron.integration.UnableToStartException;
@@ -29,12 +28,15 @@ import java.util.Map;
import java.util.Optional;
import java.util.Properties;
+import static org.apache.metron.common.configuration.ConfigurationsUtils.*;
+
public class ConfigUploadComponent implements InMemoryComponent {
private Properties topologyProperties;
private String globalConfigPath;
private String parserConfigsPath;
private String enrichmentConfigsPath;
+ private String profilerConfigPath;
private Optional<String> globalConfig = Optional.empty();
private Map<String, SensorParserConfig> parserSensorConfigs = new HashMap<>();
public ConfigUploadComponent withTopologyProperties(Properties topologyProperties) {
@@ -56,6 +58,11 @@ public class ConfigUploadComponent implements InMemoryComponent {
return this;
}
+ public ConfigUploadComponent withProfilerConfigsPath(String profilerConfigsPath) {
+ this.profilerConfigPath = profilerConfigsPath;
+ return this;
+ }
+
public ConfigUploadComponent withParserSensorConfig(String name, SensorParserConfig config) {
parserSensorConfigs.put(name, config);
return this;
@@ -69,23 +76,18 @@ public class ConfigUploadComponent implements InMemoryComponent {
@Override
public void start() throws UnableToStartException {
try {
+ final String zookeeperUrl = topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY);
+
if(globalConfigPath != null) {
- ConfigurationsUtils.uploadConfigsToZookeeper(globalConfigPath
- , parserConfigsPath
- , enrichmentConfigsPath
- , topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY)
- );
+ uploadConfigsToZookeeper(globalConfigPath, parserConfigsPath, enrichmentConfigsPath, profilerConfigPath, zookeeperUrl);
}
+
for(Map.Entry<String, SensorParserConfig> kv : parserSensorConfigs.entrySet()) {
- ConfigurationsUtils.writeSensorParserConfigToZookeeper( kv.getKey()
- , kv.getValue()
- , topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY)
- );
+ writeSensorParserConfigToZookeeper(kv.getKey(), kv.getValue(), zookeeperUrl);
}
+
if(globalConfig.isPresent()) {
- ConfigurationsUtils.writeGlobalConfigToZookeeper(globalConfig.get().getBytes()
- , topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY)
- );
+ writeGlobalConfigToZookeeper(globalConfig.get().getBytes(), zookeeperUrl);
}
} catch (Exception e) {
@@ -95,10 +97,10 @@ public class ConfigUploadComponent implements InMemoryComponent {
public SensorParserConfig getSensorParserConfig(String sensorType) {
SensorParserConfig sensorParserConfig = new SensorParserConfig();
- CuratorFramework client = ConfigurationsUtils.getClient(topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY));
+ CuratorFramework client = getClient(topologyProperties.getProperty(KafkaWithZKComponent.ZOOKEEPER_PROPERTY));
client.start();
try {
- sensorParserConfig = ConfigurationsUtils.readSensorParserConfigFromZookeeper(sensorType, client);
+ sensorParserConfig = readSensorParserConfigFromZookeeper(sensorType, client);
} catch (Exception e) {
e.printStackTrace();
} finally {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/pom.xml b/metron-platform/metron-hbase/pom.xml
index 7949f83..31f2960 100644
--- a/metron-platform/metron-hbase/pom.xml
+++ b/metron-platform/metron-hbase/pom.xml
@@ -34,11 +34,50 @@
<groupId>org.apache.metron</groupId>
<artifactId>metron-common</artifactId>
<version>${project.parent.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-test-utilities</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ <version>${global_hadoop_version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ <version>${global_hadoop_version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <classifier>tests</classifier>
+ <version>${global_hadoop_version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <classifier>tests</classifier>
+ <version>${global_hadoop_version}</version>
+ <scope>provided</scope>
</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>
@@ -48,8 +87,11 @@
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
</exclusions>
- <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
@@ -65,9 +107,93 @@
<artifactId>log4j-over-slf4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
+ <exclusion>
+ <artifactId>disruptor</artifactId>
+ <groupId>com.googlecode.disruptor</groupId>
+ </exclusion>
</exclusions>
</dependency>
<dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-hbase</artifactId>
+ <version>${global_storm_version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <groupId>org.slf4j</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-hdfs</artifactId>
+ <version>${global_storm_version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-hbase</artifactId>
+ <version>${global_storm_version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${global_hbase_version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-testing-util</artifactId>
+ <version>${global_hbase_version}</version>
+ <scope>test</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>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>${global_mockito_version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/BatchHelper.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/BatchHelper.java b/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/BatchHelper.java
new file mode 100644
index 0000000..e7cf2cf
--- /dev/null
+++ b/metron-platform/metron-hbase/src/main/java/org/apache/metron/hbase/bolt/BatchHelper.java
@@ -0,0 +1,127 @@
+/*
+ *
+ * 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.hbase.bolt;
+
+import backtype.storm.Constants;
+import backtype.storm.task.OutputCollector;
+import backtype.storm.tuple.Tuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Provides functionality for handling batch writes to HBase.
+ */
+public class BatchHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BatchHelper.class);
+
+ /**
+ * The batch size. Defaults to 15,000.
+ */
+ private int batchSize = 15000;
+
+ /**
+ * A batch of tuples.
+ */
+ private List<Tuple> tupleBatch;
+
+ private boolean forceFlush = false;
+ private OutputCollector collector;
+
+ /**
+ * @param batchSize The batch size.
+ * @param collector The output collector.
+ */
+ public BatchHelper(int batchSize, OutputCollector collector) {
+ if (batchSize > 0) {
+ this.batchSize = batchSize;
+ }
+ this.collector = collector;
+ this.tupleBatch = new LinkedList<>();
+ }
+
+ /**
+ * Fail the batch.
+ * @param e The exception which caused the failure.
+ */
+ public void fail(Exception e) {
+ collector.reportError(e);
+ for (Tuple t : tupleBatch) {
+ collector.fail(t);
+ }
+ tupleBatch.clear();
+ forceFlush = false;
+ }
+
+ /**
+ * Ack the batch.
+ */
+ public void ack() {
+ for (Tuple t : tupleBatch) {
+ collector.ack(t);
+ }
+ tupleBatch.clear();
+ forceFlush = false;
+ }
+
+ public boolean shouldHandle(Tuple tuple) {
+ if (isTick(tuple)) {
+ LOG.debug("TICK received! current batch status [{}/{}]", tupleBatch.size(), batchSize);
+ forceFlush = true;
+ return false;
+ } else {
+ return true;
+ }
+ }
+
+ /**
+ * Adds a tuple to the batch.
+ * @param tuple
+ */
+ public void addBatch(Tuple tuple) {
+ tupleBatch.add(tuple);
+ if (tupleBatch.size() >= batchSize) {
+ forceFlush = true;
+ }
+ }
+
+ public List<Tuple> getBatchTuples() {
+ return this.tupleBatch;
+ }
+
+ public int getBatchSize() {
+ return this.batchSize;
+ }
+
+ public boolean shouldFlush() {
+ return forceFlush && !tupleBatch.isEmpty();
+ }
+
+ public boolean isTick(Tuple tuple) {
+ return tuple != null &&
+ Constants.SYSTEM_COMPONENT_ID.equals(tuple.getSourceComponent()) &&
+ Constants.SYSTEM_TICK_STREAM_ID.equals(tuple.getSourceStreamId());
+ }
+
+}
[3/3] incubator-metron git commit: METRON-309 Created a Normalcy
Profiler (nickwallen) closes apache/incubator-metron#208
Posted by ni...@apache.org.
METRON-309 Created a Normalcy Profiler (nickwallen) closes apache/incubator-metron#208
Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/81754864
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/81754864
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/81754864
Branch: refs/heads/master
Commit: 817548647efe736dd05a7675d33f12f415ecf46f
Parents: 1ea8d9e
Author: nickwallen <ni...@nickallen.org>
Authored: Mon Aug 22 10:47:24 2016 -0400
Committer: Nick Allen <ni...@nickallen.org>
Committed: Mon Aug 22 10:47:24 2016 -0400
----------------------------------------------------------------------
metron-analytics/metron-profiler/README.md | 250 ++++++++++++++
metron-analytics/metron-profiler/pom.xml | 337 +++++++++++++++++++
.../src/main/assembly/assembly.xml | 72 ++++
.../src/main/config/profiler.properties | 36 ++
.../src/main/flux/profiler/remote.yaml | 114 +++++++
.../metron/profiler/ProfileMeasurement.java | 133 ++++++++
.../profiler/bolt/ProfileBuilderBolt.java | 292 ++++++++++++++++
.../profiler/bolt/ProfileHBaseMapper.java | 136 ++++++++
.../profiler/bolt/ProfileSplitterBolt.java | 151 +++++++++
.../stellar/DefaultStellarExecutor.java | 117 +++++++
.../profiler/stellar/StellarExecutor.java | 64 ++++
.../src/main/scripts/start_profiler_topology.sh | 22 ++
.../zookeeper/readme-example-1/profiler.json | 17 +
.../zookeeper/readme-example-2/profiler.json | 19 ++
.../zookeeper/readme-example-3/profiler.json | 20 ++
.../profiler/bolt/ProfileBuilderBoltTest.java | 269 +++++++++++++++
.../profiler/bolt/ProfileSplitterBoltTest.java | 212 ++++++++++++
.../integration/ConfigUploadComponent.java | 119 +++++++
.../integration/ProfilerIntegrationTest.java | 226 +++++++++++++
.../util/DefaultStellarExecutorTest.java | 161 +++++++++
metron-analytics/pom.xml | 3 +-
.../roles/metron_streaming/defaults/main.yml | 9 +
.../metron_streaming/tasks/copy_bundles.yml | 7 +-
.../roles/metron_streaming/tasks/topologies.yml | 12 +
.../common/bolt/ConfiguredEnrichmentBolt.java | 1 -
.../common/bolt/ConfiguredProfilerBolt.java | 75 +++++
.../common/configuration/ConfigurationType.java | 34 +-
.../configuration/ConfigurationsUtils.java | 105 ++++--
.../configuration/profiler/ProfileConfig.java | 145 ++++++++
.../configuration/profiler/ProfilerConfig.java | 73 ++++
.../profiler/ProfilerConfigurations.java | 53 +++
.../metron/common/dsl/MapVariableResolver.java | 2 +
.../common/stellar/BaseStellarProcessor.java | 1 +
.../metron/enrichment/bolt/HBaseBolt.java | 184 ----------
.../components/ConfigUploadComponent.java | 32 +-
metron-platform/metron-hbase/pom.xml | 128 ++++++-
.../apache/metron/hbase/bolt/BatchHelper.java | 127 +++++++
.../org/apache/metron/hbase/bolt/HBaseBolt.java | 202 +++++++++++
.../apache/metron/hbase/client/HBaseClient.java | 165 +++++++++
.../java/org/apache/metron/hbase/Widget.java | 77 +++++
.../org/apache/metron/hbase/WidgetMapper.java | 55 +++
.../apache/metron/hbase/bolt/HBaseBoltTest.java | 156 +++++++++
.../metron/hbase/client/HBaseClientTest.java | 149 ++++++++
.../components/FluxTopologyComponent.java | 173 +++++-----
.../org/apache/metron/test/mock/MockHTable.java | 5 +-
metron-platform/pom.xml | 1 -
46 files changed, 4418 insertions(+), 323 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/README.md
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/README.md b/metron-analytics/metron-profiler/README.md
new file mode 100644
index 0000000..2ad9c6f
--- /dev/null
+++ b/metron-analytics/metron-profiler/README.md
@@ -0,0 +1,250 @@
+# Metron Profiler
+
+The Profiler is a feature extraction mechanism that can generate a profile describing the behavior of an entity on a network. An entity might be a server, user, subnet or application. Once a profile has been generated defining what normal behavior looks-like, models can be built that identify anomalous behavior.
+
+This is achieved by summarizing the streaming telemetry data consumed by Metron over sliding windows. A summary statistic is applied to the data received within a given window. Collecting this summary across many windows results in a time series that is useful for analysis.
+
+## Usage
+
+Any field contained within a message can be used to generate a profile. A profile can even be produced from combining fields that originate in different data sources. A user has considerable power to transform the data used in a profile by leveraging the Stellar language. A user only need configure the desired profiles in Zookeeper and ensure that the Profiler topology is running.
+
+### Configuration
+
+The Profiler configuration requires a JSON-formatted set of elements, many of which can contain Stellar code. The configuration contains the following elements.
+
+| Name | | Description |
+|--- |--- |--- |
+| profile | Required | A unique name identifying the profile. The field is treated as a string. |
+| foreach | Required | A separate profile is maintained *for each* of these. This is effectively the entity that the profile is describing. The field is expected to contain a Stellar expression whose result is the entity name. For example, if `ip_src_addr` then a separate profile would be maintained for each unique IP source address in the data; 10.0.0.1, 10.0.0.2, etc. |
+| onlyif | Optional | An expression that determines if a message should be applied to the profile. A Stellar expression is expected that when executed returns a boolean. A message is only applied to a profile if this condition is true. This allows a profile to filter the messages that it receives. |
+| init | Optional | A set of expressions that is executed at the start of a window period. A map is expected where the key is the variable name and the value is a Stellar expression. The map can contain 0 or more variables/expressions. At the start of each window period the expression is executed once and stored in a variable with the given name. |
+| update | Required | A set of expressions that is executed when a message is applied to the profile. A map is expected where the key is the variable name and the value is a Stellar expression. The map can include 0 or more variables/expressions. |
+| result | Required | A Stellar expression that is executed when the window period expires. The expression is expected to in some way summarize the messages that were applied to the profile over the window period. The expression must result in a numeric value such as a Double, Long, Float, Short, or Integer. |
+
+#### Examples
+
+Examples of the types of profiles that can be built include the following. Each shows the configuration that would be required to produce the profile. These examples assume a fictitious input messages that looks something like the following.
+
+```
+{
+ "ip_src_addr": "10.0.0.1",
+ "protocol": "HTTPS",
+ "length": "10",
+ "bytes_in": "234"
+},
+{
+ "ip_src_addr": "10.0.0.2",
+ "protocol": "HTTP",
+ "length": "20",
+ "bytes_in": "390"
+},
+{
+ "ip_src_addr": "10.0.0.3",
+ "protocol": "DNS",
+ "length": "30",
+ "bytes_in": "560"
+}
+```
+
+#### Example 1
+
+The total number of bytes of HTTP data for each host. The following configuration would be used to generate this profile.
+
+```
+{
+ "inputTopic": "indexing",
+ "profiles": [
+ {
+ "profile": "example1",
+ "foreach": "ip_src_addr",
+ "onlyif": "protocol == 'HTTP'",
+ "init": {
+ "total_bytes": 0.0
+ },
+ "update": {
+ "total_bytes": "total_bytes + bytes_in"
+ },
+ "result": "total_bytes"
+ }
+ ]
+}
+```
+
+This creates a profile...
+ * Named \u2018example1\u2019
+ * That for each IP source address
+ * Only if the 'protocol' field equals 'HTTP'
+ * Initializes a counter \u2018total_bytes\u2019 to zero
+ * Adds to \u2018total_bytes\u2019 the value of the message's \u2018bytes_in\u2019 field
+ * Returns \u2018total_bytes\u2019 as the result
+
+#### Example 2
+
+The ratio of DNS traffic to HTTP traffic for each host. The following configuration would be used to generate this profile.
+
+```
+{
+ "inputTopic": "indexing",
+ "profiles": [
+ {
+ "profile": "example2",
+ "foreach": "ip_src_addr",
+ "onlyif": "protocol == 'DNS' or protocol == 'HTTP'",
+ "init": {
+ "num_dns": 1.0,
+ "num_http": 1.0
+ },
+ "update": {
+ "num_dns": "num_dns + (if protocol == 'DNS' then 1 else 0)",
+ "num_http": "num_http + (if protocol == 'HTTP' then 1 else 0)"
+ },
+ "result": "num_dns / num_http"
+ }
+ ]
+}
+```
+
+This creates a profile...
+ * Named \u2018example2\u2019
+ * That for each IP source address
+ * Only if the 'protocol' field equals 'HTTP' or 'DNS'
+ * Accumulates the number of DNS requests
+ * Accumulates the number of HTTP requests
+ * Returns the ratio of these as the result
+
+#### Example 3
+
+The average response body length of HTTP traffic. The following configuration would be used to generate this profile.
+
+```
+{
+ "inputTopic": "indexing",
+ "profiles": [
+ {
+ "profile": "example3",
+ "foreach": "ip_src_addr",
+ "onlyif": "protocol == 'HTTP'",
+ "init": {
+ "sum": 0.0,
+ "cnt": 0.0
+ },
+ "update": {
+ "sum": "sum + resp_body_len",
+ "cnt": "cnt + 1"
+ },
+ "result": "sum / cnt"
+ }
+ ]
+}
+```
+
+This creates a profile...
+ * Named \u2018example3\u2019
+ * That for each IP source address
+ * Only if the 'protocol' field equals 'HTTP'
+ * Accumulates the sum of response body length
+ * Accumulates the number of messages
+ * Calculates the average as the result
+
+### Topology Configuration
+
+The Profiler topology also accepts the following configuration settings.
+
+| Setting | Description |
+|--- |--- |
+| profiler.workers | The number of worker processes to create for the topology. |
+| profiler.executors | The number of executors to spawn per component. |
+| profiler.input.topic | The name of the Kafka topic from which to consume data. |
+| profiler.flush.interval.seconds | The duration of a profile's sliding window before it is flushed. |
+| profiler.hbase.salt.divisor | A salt is prepended to the row key to help prevent hotspotting. This constant is used to generate the salt. Ideally, this constant should be roughly equal to the number of nodes in the Hbase cluster. |
+| profiler.hbase.table | The name of the HBase table that profiles are written to. |
+| profiler.hbase.batch | The number of puts that are written in a single batch. |
+| profiler.hbase.flush.interval.seconds | The maximum number of seconds between batch writes to HBase. |
+
+
+## Getting Started
+
+This section will describe the steps required to get your first profile running.
+
+1. Launch the 'Quick Dev' environment.
+ ```
+ $ cd metron-deployment/vagrant/quick-dev-platform/
+ $ ./run.sh
+ ```
+
+2. After the environment has been deployed, then login to the host.
+ ```
+ $ vagrant ssh
+ $ sudo su -
+ $ cd /usr/metron/0.2.0BETA/
+ ```
+
+3. Create a table within HBase that will store the profile data. The table name and column family must match the Profiler topology configuration stored at `/usr/metron/0.2.0BETA/config/profiler.properties`.
+ ```
+ $ /usr/hdp/current/hbase-client/bin/hbase shell
+ hbase(main):001:0> create 'profiler', 'P'
+ ```
+
+4. Shorten the flush intervals to more immediately see results. Edit the Profiler topology properties located at `/usr/metron/0.2.0BETA/config/profiler.properties`. Alter the following two properties.
+ ```
+ profiler.flush.interval.seconds=30
+ profiler.hbase.flush.interval.seconds=30
+ ```
+
+5. Create the Profiler definition in a file located at `/usr/metron/0.2.0BETA/config/zookeeper/profiler.json`. The following JSON will create a profile that simply counts the number of messages.
+ ```
+ {
+ "inputTopic": "indexing",
+ "profiles": [
+ {
+ "profile": "test",
+ "foreach": "ip_src_addr",
+ "onlyif": "true",
+ "init": { "sum": 0 },
+ "update": { "sum": "sum + 1" },
+ "result": "sum"
+ }
+ ]
+ }
+ ```
+
+6. Upload the Profiler definition to Zookeeper.
+ ```
+ $ bin/zk_load_configs.sh -m PUSH -i config/zookeeper/ -z node1:2181
+ ```
+
+7. Start the Profiler topology.
+ ```
+ bin/start_profiler_topology.sh
+ ```
+
+8. Ensure that test messages are being sent to the Profiler's input topic in Kafka. The Profiler will consume messages from the `inputTopic` in the Profiler definition.
+
+9. Check the HBase table to validate that the Profiler is working.
+ ```
+ $ /usr/hdp/current/hbase-client/bin/hbase shell
+ hbase(main):001:0> count 'profiler'
+ ```
+
+## Implementation
+
+The Profiler is implemented as a Storm topology using the following bolts and spouts.
+
+## Topology
+
+### KafkaSpout
+
+A spout that consumes messages from a single Kafka topic. In most cases, the Profiler topology will consume messages from the `indexing` topic. This topic contains fully enriched messages that are ready to be indexed. This ensures that profiles can take advantage of all the available data elements.
+
+### ProfileSplitterBolt
+
+The bolt responsible for filtering incoming messages and directing each to the one or more downstream bolts that are responsible for building a Profile. Each message may be needed by 0, 1 or even many Profiles. Each emitted tuple contains the 'resolved' entity name, the profile definition, and the input message.
+
+### ProfileBuilderBolt
+
+This bolt maintains all of the state required to build a Profile. When the window period expires, the data is summarized as a ProfileMeasurement, all state is flushed, and the ProfileMeasurement is emitted. Each instance of this bolt is responsible for maintaining the state for a single Profile-Entity pair.
+
+### HBaseBolt
+
+A bolt that is responsible for writing to HBase. Most profiles will be flushed every 15 minutes or so. If each ProfileBuilderBolt were responsible for writing to HBase itself, there would be little to no opportunity to optimize these writes. By aggregating the writes from multiple Profile-Entity pairs these writes can be batched, for example.
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/pom.xml
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/pom.xml b/metron-analytics/metron-profiler/pom.xml
new file mode 100644
index 0000000..b31df67
--- /dev/null
+++ b/metron-analytics/metron-profiler/pom.xml
@@ -0,0 +1,337 @@
+<?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-analytics</artifactId>
+ <version>0.2.0BETA</version>
+ </parent>
+ <artifactId>metron-profiler</artifactId>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-hbase</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-integration-test</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-test-utilities</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</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>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ <version>${global_hadoop_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ <version>${global_hadoop_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <classifier>tests</classifier>
+ <version>${global_hadoop_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <classifier>tests</classifier>
+ <version>${global_hadoop_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.adrianwalker</groupId>
+ <artifactId>multiline-string</artifactId>
+ <version>0.1.2</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>tools</artifactId>
+ <groupId>sun.jdk</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ <version>${global_storm_version}</version>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <groupId>org.slf4j</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-hdfs</artifactId>
+ <version>${global_storm_version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-hbase</artifactId>
+ <version>${global_storm_version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_2.9.2</artifactId>
+ <version>${global_kafka_version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${global_hbase_version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${global_hbase_version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-testing-util</artifactId>
+ <version>${global_hbase_version}</version>
+ <scope>test</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>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>${global_mockito_version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.tempus-fugit</groupId>
+ <artifactId>tempus-fugit</artifactId>
+ <version>1.2-20140129.191141-5</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <reporting>
+ <plugins>
+ <!-- Normally, dependency report takes time, skip it -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-project-info-reports-plugin</artifactId>
+ <version>2.7</version>
+
+ <configuration>
+ <dependencyLocationsEnabled>false</dependencyLocationsEnabled>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>emma-maven-plugin</artifactId>
+ <version>1.0-alpha-3</version>
+ <inherited>true</inherited>
+ </plugin>
+ </plugins>
+ </reporting>
+
+ <build>
+ <plugins>
+ <plugin>
+ <!-- Separates the unit tests from the integration tests. -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.12.4</version>
+ <configuration>
+ <!-- Skip the default running of this plug-in (or everything is run twice...see below) -->
+ <argLine>-Xmx2048m</argLine>
+ <skip>true</skip>
+ <!-- Show 100% of the lines from the stack trace (doesn't work) -->
+ <trimStackTrace>false</trimStackTrace>
+ </configuration>
+ <executions>
+ <execution>
+ <id>unit-tests</id>
+ <phase>test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <!-- Never skip running the tests when the test phase is invoked -->
+ <skip>false</skip>
+ <includes>
+ <!-- Include unit tests within integration-test phase. -->
+ <include>**/*Test.java</include>
+ </includes>
+ <excludes>
+ <!-- Exclude integration tests within (unit) test phase. -->
+ <exclude>**/*IntegrationTest.java</exclude>
+ </excludes>
+
+ </configuration>
+ </execution>
+ <execution>
+ <id>integration-tests</id>
+ <phase>integration-test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <!-- Never skip running the tests when the integration-test phase is invoked -->
+ <skip>false</skip>
+ <includes>
+ <!-- Include integration tests within integration-test phase. -->
+ <include>**/*IntegrationTest.java</include>
+ </includes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${global_shade_version}</version>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <relocations>
+ <relocation>
+ <pattern>com.google.common</pattern>
+ <shadedPattern>org.apache.metron.guava.metron-profiler</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.fasterxml.jackson.core</pattern>
+ <shadedPattern>com.fasterxml.jackson.core.metron.elasticsearch</shadedPattern>
+ </relocation>
+ </relocations>
+ <artifactSet>
+ <excludes>
+ <exclude>storm:storm-core:*</exclude>
+ <exclude>storm:storm-lib:*</exclude>
+ <exclude>org.slf4j.impl*</exclude>
+ <exclude>org.slf4j:slf4j-log4j*</exclude>
+ </excludes>
+ </artifactSet>
+ <transformers>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.DontIncludeResourceTransformer">
+ <resource>.yaml</resource>
+ </transformer>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass></mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <configuration>
+ <descriptor>src/main/assembly/assembly.xml</descriptor>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id> <!-- this is used for inheritance merges -->
+ <phase>package</phase> <!-- bind to the packaging phase -->
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/assembly/assembly.xml
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/assembly/assembly.xml b/metron-analytics/metron-profiler/src/main/assembly/assembly.xml
new file mode 100644
index 0000000..a5f1643
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/assembly/assembly.xml
@@ -0,0 +1,72 @@
+<!--
+ ~
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ ~
+ -->
+
+<assembly>
+ <id>archive</id>
+ <formats>
+ <format>tar.gz</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${project.basedir}/src/main/config</directory>
+ <outputDirectory>/config</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ <excludes>
+ <exclude>**/*.formatted</exclude>
+ <exclude>**/*.filtered</exclude>
+ </excludes>
+ <fileMode>0644</fileMode>
+ <lineEnding>unix</lineEnding>
+ <filtered>true</filtered>
+ </fileSet>
+ <fileSet>
+ <directory>${project.basedir}/src/main/scripts</directory>
+ <outputDirectory>/bin</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ <excludes>
+ <exclude>**/*.formatted</exclude>
+ <exclude>**/*.filtered</exclude>
+ </excludes>
+ <fileMode>0755</fileMode>
+ <lineEnding>unix</lineEnding>
+ <filtered>true</filtered>
+ </fileSet>
+ <fileSet>
+ <directory>${project.basedir}/src/main/flux</directory>
+ <outputDirectory>/flux</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ <excludes>
+ <exclude>**/*.formatted</exclude>
+ <exclude>**/*.filtered</exclude>
+ </excludes>
+ <fileMode>0644</fileMode>
+ <lineEnding>unix</lineEnding>
+ </fileSet>
+ <fileSet>
+ <directory>${project.basedir}/target</directory>
+ <includes>
+ <include>${project.artifactId}-${project.version}.jar</include>
+ </includes>
+ <outputDirectory>/lib</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ </fileSet>
+ </fileSets>
+</assembly>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/config/profiler.properties
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/config/profiler.properties b/metron-analytics/metron-profiler/src/main/config/profiler.properties
new file mode 100644
index 0000000..80fd2da
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/config/profiler.properties
@@ -0,0 +1,36 @@
+#
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+
+##### Storm #####
+
+profiler.workers=1
+profiler.executors=0
+profiler.input.topic=indexing
+profiler.flush.interval.seconds=900
+profiler.hbase.salt.divisor=1000
+profiler.hbase.table=profiler
+profiler.hbase.batch=10
+profiler.hbase.flush.interval.seconds=30
+
+##### Kafka #####
+
+kafka.zk=node1:2181
+kafka.broker=node1:6667
+kafka.start=WHERE_I_LEFT_OFF
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/flux/profiler/remote.yaml
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/flux/profiler/remote.yaml b/metron-analytics/metron-profiler/src/main/flux/profiler/remote.yaml
new file mode 100644
index 0000000..d99510a
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/flux/profiler/remote.yaml
@@ -0,0 +1,114 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: "profiler"
+
+config:
+ topology.workers: ${profiler.workers}
+ topology.acker.executors: ${profiler.executors}
+
+components:
+
+ - id: "defaultExecutor"
+ className: "org.apache.metron.profiler.stellar.DefaultStellarExecutor"
+
+ - id: "hbaseMapper"
+ className: "org.apache.metron.profiler.bolt.ProfileHBaseMapper"
+ properties:
+ - name: "saltDivisor"
+ value: ${profiler.hbase.salt.divisor}
+
+ - id: "zkHosts"
+ className: "storm.kafka.ZkHosts"
+ constructorArgs:
+ - "${kafka.zk}"
+
+ - id: "kafkaConfig"
+ className: "org.apache.metron.common.spout.kafka.SpoutConfig"
+ constructorArgs:
+ # zookeeper hosts
+ - ref: "zkHosts"
+ # topic name
+ - "${profiler.input.topic}"
+ # zk root
+ - ""
+ # id
+ - "indexing"
+ configMethods:
+ - name: "from"
+ args:
+ - "${kafka.start}"
+
+spouts:
+
+ - id: "kafkaSpout"
+ className: "storm.kafka.KafkaSpout"
+ constructorArgs:
+ - ref: "kafkaConfig"
+
+bolts:
+
+ - id: "splitterBolt"
+ className: "org.apache.metron.profiler.bolt.ProfileSplitterBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ properties:
+ - name: "executor"
+ ref: "defaultExecutor"
+
+ - id: "builderBolt"
+ className: "org.apache.metron.profiler.bolt.ProfileBuilderBolt"
+ constructorArgs:
+ - "${kafka.zk}"
+ properties:
+ - name: "executor"
+ ref: "defaultExecutor"
+ - name: "flushFrequency"
+ value: ${profiler.flush.interval.seconds}
+
+ - id: "hbaseBolt"
+ className: "org.apache.metron.hbase.bolt.HBaseBolt"
+ constructorArgs:
+ - "${profiler.hbase.table}"
+ - ref: "hbaseMapper"
+ configMethods:
+ - name: "withTableProvider"
+ args: ["${hbase.provider.impl}"]
+ - name: "withBatchSize"
+ args: [${profiler.hbase.batch}]
+ - name: "withFlushIntervalSecs"
+ args: [${profiler.hbase.flush.interval.seconds}]
+
+streams:
+
+ - name: "spout -> splitter"
+ from: "kafkaSpout"
+ to: "splitterBolt"
+ grouping:
+ type: SHUFFLE
+
+ - name: "splitter -> builder"
+ from: "splitterBolt"
+ to: "builderBolt"
+ grouping:
+ type: FIELDS
+ args: ["entity", "profile", "message"]
+
+ - name: "builder -> hbase"
+ from: "builderBolt"
+ to: "hbaseBolt"
+ grouping:
+ type: SHUFFLE
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/ProfileMeasurement.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/ProfileMeasurement.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/ProfileMeasurement.java
new file mode 100644
index 0000000..e240567
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/ProfileMeasurement.java
@@ -0,0 +1,133 @@
+/*
+ *
+ * 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.profiler;
+
+/**
+ * Represents a single data point within a Profile.
+ *
+ * A Profile is effectively a time series. To this end a Profile is composed
+ * of many ProfileMeasurement values which in aggregate form a time series.
+ */
+public class ProfileMeasurement {
+
+ /**
+ * The name of the profile that this measurement is associated with.
+ */
+ private String profileName;
+
+ /**
+ * The name of the entity being profiled.
+ */
+ private String entity;
+
+ /**
+ * When the measurement window was started in milliseconds since the epoch.
+ */
+ private long start;
+
+ /**
+ * When the measurement window closed in milliseconds since the epoch.
+ */
+ private long end;
+
+ /**
+ * The actual measurement itself.
+ */
+ private double value;
+
+ public String getProfileName() {
+ return profileName;
+ }
+
+ public void setProfileName(String profileName) {
+ this.profileName = profileName;
+ }
+
+ public String getEntity() {
+ return entity;
+ }
+
+ public void setEntity(String entity) {
+ this.entity = entity;
+ }
+
+ public long getStart() {
+ return start;
+ }
+
+ public void setStart(long start) {
+ this.start = start;
+ }
+
+ public long getEnd() {
+ return end;
+ }
+
+ public void setEnd(long end) {
+ this.end = end;
+ }
+
+ public double getValue() {
+ return value;
+ }
+
+ public void setValue(double value) {
+ this.value = value;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ProfileMeasurement that = (ProfileMeasurement) o;
+
+ if (start != that.start) return false;
+ if (end != that.end) return false;
+ if (Double.compare(that.value, value) != 0) return false;
+ if (profileName != null ? !profileName.equals(that.profileName) : that.profileName != null) return false;
+ return entity != null ? entity.equals(that.entity) : that.entity == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result;
+ long temp;
+ result = profileName != null ? profileName.hashCode() : 0;
+ result = 31 * result + (entity != null ? entity.hashCode() : 0);
+ result = 31 * result + (int) (start ^ (start >>> 32));
+ result = 31 * result + (int) (end ^ (end >>> 32));
+ temp = Double.doubleToLongBits(value);
+ result = 31 * result + (int) (temp ^ (temp >>> 32));
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return "ProfileMeasurement{" +
+ "profileName='" + profileName + '\'' +
+ ", entity='" + entity + '\'' +
+ ", start=" + start +
+ ", end=" + end +
+ ", value=" + value +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java
new file mode 100644
index 0000000..d6656ac
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileBuilderBolt.java
@@ -0,0 +1,292 @@
+/*
+ *
+ * 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.profiler.bolt;
+
+import backtype.storm.Config;
+import backtype.storm.Constants;
+import backtype.storm.task.OutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+import org.apache.metron.common.bolt.ConfiguredProfilerBolt;
+import org.apache.metron.common.configuration.profiler.ProfileConfig;
+import org.apache.metron.common.dsl.ParseException;
+import org.apache.metron.profiler.ProfileMeasurement;
+import org.apache.metron.profiler.stellar.StellarExecutor;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+import static java.lang.String.format;
+
+/**
+ * A bolt that is responsible for building a Profile.
+ *
+ * This bolt maintains the state required to build a Profile. When the window
+ * period expires, the data is summarized as a ProfileMeasurement, all state is
+ * flushed, and the ProfileMeasurement is emitted.
+ *
+ * Each instance of this bolt is responsible for maintaining the state for a single
+ * Profile-Entity pair.
+ */
+public class ProfileBuilderBolt extends ConfiguredProfilerBolt {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(ProfileBuilderBolt.class);
+
+ /**
+ * Executes Stellar code and maintains state across multiple invocations.
+ */
+ private StellarExecutor executor;
+
+ /**
+ * The number of seconds between when the Profile is flushed.
+ */
+ private int flushFrequency;
+
+ /**
+ * A ProfileMeasurement is created and emitted each window period. A Profile
+ * itself is composed of many ProfileMeasurements.
+ */
+ private transient ProfileMeasurement measurement;
+
+ /**
+ * The definition of the Profile that the bolt is building.
+ */
+ private transient ProfileConfig profileConfig;
+
+ /**
+ * Parses JSON messages.
+ */
+ private transient JSONParser parser;
+
+ private OutputCollector collector;
+
+ /**
+ * @param zookeeperUrl The Zookeeper URL that contains the configuration data.
+ */
+ public ProfileBuilderBolt(String zookeeperUrl) {
+ super(zookeeperUrl);
+ }
+
+ /**
+ * Defines the frequency at which the bolt will receive tick tuples. Tick tuples are
+ * used to control how often a profile is flushed.
+ */
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ Config conf = new Config();
+ conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, getFlushFrequency());
+ return conf;
+ }
+
+ @Override
+ public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+ super.prepare(stormConf, context, collector);
+ this.collector = collector;
+ this.parser = new JSONParser();
+ }
+
+ /**
+ * The builder emits a single field, 'measurement', which contains a ProfileMeasurement. A
+ * ProfileMeasurement is emitted when a time window expires and a flush occurs.
+ */
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ // once the time window expires, a complete ProfileMeasurement is emitted
+ declarer.declare(new Fields("measurement"));
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ try {
+ doExecute(input);
+
+ } catch (Throwable e) {
+ LOG.error(format("Unexpected failure: message='%s', tuple='%s'", e.getMessage(), input), e);
+ collector.reportError(e);
+
+ } finally {
+ collector.ack(input);
+ }
+ }
+
+ /**
+ * Update the execution environment based on data contained in the
+ * message. If the tuple is a tick tuple, then flush the profile
+ * and reset the execution environment.
+ * @param input The tuple to execute.
+ */
+ private void doExecute(Tuple input) {
+
+ if(!isTickTuple(input)) {
+
+ // if this is the first tuple in a window period, initialization is needed
+ if (!isInitialized()) {
+ init(input);
+ }
+
+ // update the profile with data from a new message
+ update(input);
+
+ } else {
+
+ // flush the profile - can only flush if it has been initialized
+ if(isInitialized()) {
+ flush(input);
+ }
+ }
+ }
+
+ /**
+ * Initialize the bolt. Occurs when the first tuple is received at the start
+ * of each window period.
+ * @param input The input tuple
+ */
+ private void init(Tuple input) {
+
+ // save the profile definition - needed later during a flush
+ profileConfig = (ProfileConfig) input.getValueByField("profile");
+
+ // create the measurement which will be saved at the end of the window period
+ measurement = new ProfileMeasurement();
+ measurement.setStart(getTimestamp());
+ measurement.setEntity(input.getStringByField("entity"));
+ measurement.setProfileName(profileConfig.getProfile());
+
+ // execute the 'init' expression
+ try {
+ JSONObject message = (JSONObject) input.getValueByField("message");
+ Map<String, String> expressions = profileConfig.getInit();
+ expressions.forEach((var, expr) -> executor.assign(var, expr, message));
+
+ } catch(ParseException e) {
+ String msg = format("Bad 'init' expression: %s, profile=%s, entity=%s, start=%d",
+ e.getMessage(), measurement.getProfileName(), measurement.getEntity(), measurement.getStart());
+ throw new ParseException(msg, e);
+ }
+ }
+
+ /**
+ * Update the Profile based on data contained in a new message.
+ * @param input The tuple containing a new message.
+ */
+ private void update(Tuple input) {
+ JSONObject message = (JSONObject) input.getValueByField("message");
+
+ // execute each of the 'update' expressions
+ try {
+ Map<String, String> expressions = profileConfig.getUpdate();
+ expressions.forEach((var, expr) -> executor.assign(var, expr, message));
+
+ } catch(ParseException e) {
+ String msg = format("Bad 'update' expression: %s, profile=%s, entity=%s, start=%d",
+ e.getMessage(), measurement.getProfileName(), measurement.getEntity(), measurement.getStart());
+ throw new ParseException(msg, e);
+ }
+ }
+
+ /**
+ * Flush the Profile.
+ *
+ * Executed on a fixed time period when a tick tuple is received. Completes
+ * and emits the ProfileMeasurement. Clears all state in preparation for
+ * the next window period.
+ */
+ private void flush(Tuple tickTuple) {
+ LOG.info(String.format("Flushing profile: profile=%s, entity=%s, start=%d",
+ measurement.getProfileName(), measurement.getEntity(), measurement.getStart()));
+
+ // execute the 'result' expression
+ Double result;
+ try {
+ String resultExpr = profileConfig.getResult();
+ result = executor.execute(resultExpr, new JSONObject(), Double.class);
+ } catch(ParseException e) {
+ throw new ParseException("Bad 'result' expression", e);
+ }
+
+ // emit the completed profile measurement
+ measurement.setEnd(getTimestamp());
+ measurement.setValue(result);
+ emit(measurement, tickTuple);
+
+ // clear the execution state to prepare for the next window
+ executor.clearState();
+
+ // reset measurement - used as a flag to indicate if initialized
+ measurement = null;
+ }
+
+ /**
+ * Emits a message containing a ProfileMeasurement.
+ * @param measurement The completed ProfileMeasurement.
+ */
+ private void emit(ProfileMeasurement measurement, Tuple anchor) {
+ collector.emit(anchor, new Values(measurement));
+ collector.ack(anchor);
+ }
+
+ /**
+ * Returns a value that can be used as the current timestamp. Allows subclasses
+ * to override, if necessary.
+ */
+ private long getTimestamp() {
+ return System.currentTimeMillis();
+ }
+
+ /**
+ * Has the Stellar execution environment already been initialized
+ * @return True, it it has been initialized.
+ */
+ private boolean isInitialized() {
+ return measurement != null;
+ }
+
+ /**
+ * Is this a tick tuple?
+ * @param tuple The tuple
+ */
+ protected static boolean isTickTuple(Tuple tuple) {
+ return Constants.SYSTEM_COMPONENT_ID.equals(tuple.getSourceComponent()) &&
+ Constants.SYSTEM_TICK_STREAM_ID.equals(tuple.getSourceStreamId());
+ }
+
+ public StellarExecutor getExecutor() {
+ return executor;
+ }
+
+ public void setExecutor(StellarExecutor executor) {
+ this.executor = executor;
+ }
+
+ public int getFlushFrequency() {
+ return flushFrequency;
+ }
+
+ public void setFlushFrequency(int flushFrequency) {
+ this.flushFrequency = flushFrequency;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileHBaseMapper.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileHBaseMapper.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileHBaseMapper.java
new file mode 100644
index 0000000..1c515be
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileHBaseMapper.java
@@ -0,0 +1,136 @@
+/*
+ *
+ * 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.profiler.bolt;
+
+import backtype.storm.tuple.Tuple;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.profiler.ProfileMeasurement;
+import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
+import org.apache.storm.hbase.common.ColumnList;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Calendar;
+
+/**
+ * An HbaseMapper that defines how a ProfileMeasurement is persisted within an HBase table.
+ */
+public class ProfileHBaseMapper implements HBaseMapper {
+
+ /**
+ * A salt is prepended to the row key to help prevent hotspotting. This constant is used
+ * to generate the salt. Ideally, this constant should be roughly equal to the number of
+ * nodes in the Hbase cluster.
+ */
+ private int saltDivisor = 10;
+
+ /**
+ * The name of the column family.
+ */
+ private String columnFamily;
+
+ public ProfileHBaseMapper() {
+ setColumnFamily("P");
+ }
+
+ /**
+ * Define the row key for a ProfileMeasurement.
+ * @param tuple The tuple containing a ProfileMeasurement.
+ * @return The Hbase row key.
+ */
+ @Override
+ public byte[] rowKey(Tuple tuple) {
+ ProfileMeasurement m = (ProfileMeasurement) tuple.getValueByField("measurement");
+
+ // create a calendar to determine day-of-week, etc
+ Calendar calendar = Calendar.getInstance();
+ calendar.setTimeInMillis(m.getStart());
+
+ return Bytes.toBytes(getSalt(m.getStart()) +
+ m.getProfileName() +
+ calendar.get(Calendar.DAY_OF_WEEK) +
+ calendar.get(Calendar.WEEK_OF_MONTH) +
+ calendar.get(Calendar.MONTH) +
+ calendar.get(Calendar.YEAR) +
+ m.getEntity() +
+ m.getStart());
+ }
+
+ /**
+ * Defines how the fields within a ProfileMeasurement are mapped to HBase.
+ * @param tuple The tuple containing the ProfileMeasurement.
+ */
+ @Override
+ public ColumnList columns(Tuple tuple) {
+ ProfileMeasurement measurement = (ProfileMeasurement) tuple.getValueByField("measurement");
+
+ byte[] cfBytes = Bytes.toBytes(columnFamily);
+ ColumnList cols = new ColumnList();
+ cols.addColumn(cfBytes, QPROFILE, Bytes.toBytes(measurement.getProfileName()));
+ cols.addColumn(cfBytes, QENTITY, Bytes.toBytes(measurement.getEntity()));
+ cols.addColumn(cfBytes, QSTART, Bytes.toBytes(measurement.getStart()));
+ cols.addColumn(cfBytes, QEND, Bytes.toBytes(measurement.getEnd()));
+ cols.addColumn(cfBytes, QVALUE, Bytes.toBytes(measurement.getValue()));
+
+ return cols;
+ }
+
+ /**
+ * Calculates a salt value that is used as part of the row key.
+ *
+ * The salt is calculated as 'md5(timestamp) % N' where N is a configurable value that ideally
+ * is close to the number of nodes in the Hbase cluster.
+ *
+ * @param epoch The timestamp in epoch millis to use in generating the salt.
+ */
+ private int getSalt(long epoch) {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("MD5");
+ byte[] hash = digest.digest(Bytes.toBytes(epoch));
+ return Bytes.toInt(hash) % saltDivisor;
+
+ } catch(NoSuchAlgorithmException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public String getColumnFamily() {
+ return columnFamily;
+ }
+
+ public void setColumnFamily(String columnFamily) {
+ this.columnFamily = columnFamily;
+ }
+
+ public int getSaltDivisor() {
+ return saltDivisor;
+ }
+
+ public void setSaltDivisor(int saltDivisor) {
+ this.saltDivisor = saltDivisor;
+ }
+
+ public static final byte[] QPROFILE = Bytes.toBytes("profile");
+ public static final byte[] QENTITY = Bytes.toBytes("entity");
+ public static final byte[] QSTART = Bytes.toBytes("start");
+ public static final byte[] QEND = Bytes.toBytes("end");
+ public static final byte[] QVALUE = Bytes.toBytes("value");
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java
new file mode 100644
index 0000000..bb19321
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/bolt/ProfileSplitterBolt.java
@@ -0,0 +1,151 @@
+/*
+ *
+ * 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.profiler.bolt;
+
+import backtype.storm.task.OutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+import org.apache.metron.common.bolt.ConfiguredProfilerBolt;
+import org.apache.metron.common.configuration.profiler.ProfileConfig;
+import org.apache.metron.common.configuration.profiler.ProfilerConfig;
+import org.apache.metron.profiler.stellar.StellarExecutor;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+import static java.lang.String.format;
+
+/**
+ * The bolt responsible for filtering incoming messages and directing
+ * each to the one or more bolts responsible for building a Profile. Each
+ * message may be needed by 0, 1 or even many Profiles.
+ */
+public class ProfileSplitterBolt extends ConfiguredProfilerBolt {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(ProfileSplitterBolt.class);
+
+ private OutputCollector collector;
+
+ /**
+ * JSON parser.
+ */
+ private transient JSONParser parser;
+
+ /**
+ * Executes Stellar code.
+ */
+ private StellarExecutor executor;
+
+ /**
+ * @param zookeeperUrl The Zookeeper URL that contains the configuration for this bolt.
+ */
+ public ProfileSplitterBolt(String zookeeperUrl) {
+ super(zookeeperUrl);
+ }
+
+ @Override
+ public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+ super.prepare(stormConf, context, collector);
+ this.collector = collector;
+ this.parser = new JSONParser();
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ try {
+ doExecute(input);
+
+ } catch (IllegalArgumentException | ParseException | UnsupportedEncodingException e) {
+ LOG.error(format("Unexpected failure: message='%s', tuple='%s'", e.getMessage(), input), e);
+ collector.reportError(e);
+
+ } finally {
+ collector.ack(input);
+ }
+ }
+
+ private void doExecute(Tuple input) throws ParseException, UnsupportedEncodingException {
+
+ // retrieve the input message
+ byte[] data = input.getBinary(0);
+ JSONObject message = (JSONObject) parser.parse(new String(data, "UTF8"));
+
+ // ensure there is a valid profiler configuration
+ ProfilerConfig config = getProfilerConfig();
+ if(config == null) {
+ throw new IllegalArgumentException("Fatal: Unable to find valid profiler definition");
+ }
+
+ // apply the message to each of the profile definitions
+ for (ProfileConfig profile: config.getProfiles()) {
+ applyProfile(profile, input, message);
+ }
+ }
+
+ /**
+ * Applies a message to a Profile definition.
+ * @param profile The profile definition.
+ * @param input The input tuple that delivered the message.
+ * @param message The message that may be needed by the profile.
+ */
+ private void applyProfile(ProfileConfig profile, Tuple input, JSONObject message) throws ParseException, UnsupportedEncodingException {
+
+ // is this message needed by this profile?
+ String onlyIf = profile.getOnlyif();
+ if (executor.execute(onlyIf, message, Boolean.class)) {
+
+ // what is the name of the entity in this message?
+ String entity = executor.execute(profile.getForeach(), message, String.class);
+
+ // emit a message for the bolt responsible for building this profile
+ collector.emit(input, new Values(entity, profile, message));
+ }
+ }
+
+ /**
+ * Each emitted tuple contains the following fields.
+ * <p><ol>
+ * <li> entity - The name of the entity. The actual result of executing the Stellar expression.
+ * <li> profile - The profile definition that the message needs applied to.
+ * <li> message - The message containing JSON-formatted data that needs applied to a profile.
+ * </ol></p>
+ */
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("entity", "profile", "message"));
+ }
+
+ public StellarExecutor getExecutor() {
+ return executor;
+ }
+
+ public void setExecutor(StellarExecutor executor) {
+ this.executor = executor;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java
new file mode 100644
index 0000000..d608222
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/DefaultStellarExecutor.java
@@ -0,0 +1,117 @@
+/*
+ *
+ * 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.profiler.stellar;
+
+import org.apache.metron.common.dsl.MapVariableResolver;
+import org.apache.metron.common.dsl.ParseException;
+import org.apache.metron.common.dsl.VariableResolver;
+import org.apache.metron.common.stellar.StellarPredicateProcessor;
+import org.apache.metron.common.stellar.StellarProcessor;
+import org.apache.metron.common.utils.ConversionUtils;
+import org.json.simple.JSONObject;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The default implementation of a StellarExecutor.
+ */
+public class DefaultStellarExecutor implements StellarExecutor, Serializable {
+
+ /**
+ * The current state of the Stellar execution environment.
+ */
+ private Map<String, Object> state;
+
+ public DefaultStellarExecutor() {
+ clearState();
+ }
+
+ /**
+ * @param initialState Initial state loaded into the execution environment.
+ */
+ public DefaultStellarExecutor(Map<String, Object> initialState) {
+ this.state = new HashMap<>(initialState);
+ }
+
+ @Override
+ public Map<String, Object> getState() {
+ return new HashMap<>(state);
+ }
+
+ /**
+ * Execute an expression and assign the result to a variable.
+ *
+ * @param variable The variable name to assign to.
+ * @param expression The expression to execute.
+ * @param message The message that provides additional context for the expression.
+ */
+ @Override
+ public void assign(String variable, String expression, JSONObject message) {
+ Object result = execute(expression, message);
+ state.put(variable, result);
+ }
+
+ /**
+ * Execute a Stellar expression and returns the result.
+ *
+ * @param expr The expression to execute.
+ * @param message The message that is accessible when Stellar is executed.
+ * @param clazz The expected class of the expression's result.
+ * @param <T> The expected class of the expression's result.
+ */
+ @Override
+ public <T> T execute(String expr, JSONObject message, Class<T> clazz) {
+ Object resultObject = execute(expr, message);
+
+ // perform type conversion, if necessary
+ T result = ConversionUtils.convert(resultObject, clazz);
+ if(result == null) {
+ throw new IllegalArgumentException(String.format("Unexpected type: expected=%s, actual=%s, expression=%s",
+ clazz.getSimpleName(), resultObject.getClass().getSimpleName(), expr));
+ }
+
+ return result;
+ }
+
+ @Override
+ public void clearState() {
+ this.state = new HashMap<>();
+ }
+
+ /**
+ * Execute a Stellar expression.
+ *
+ * @param expr The expression to execute.
+ * @param msg The message that is accessible when Stellar is executed.
+ */
+ private Object execute(String expr, JSONObject msg) {
+ try {
+ VariableResolver resolver = new MapVariableResolver(state, msg);
+ StellarProcessor processor = new StellarProcessor();
+ return processor.parse(expr, resolver);
+
+ } catch (ParseException e) {
+ throw new ParseException(String.format("Bad expression: expr=%s, msg=%s, state=%s", expr, msg, state));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java
new file mode 100644
index 0000000..9005e00
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/java/org/apache/metron/profiler/stellar/StellarExecutor.java
@@ -0,0 +1,64 @@
+/*
+ *
+ * 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.profiler.stellar;
+
+import org.json.simple.JSONObject;
+
+import java.util.Map;
+
+/**
+ * Executes Stellar expressions and maintains state across multiple invocations.
+ *
+ * There are two sets of functions in Stellar currently. One can be executed with
+ * a PredicateProcessor and the other a TransformationProcessor. This interface
+ * abstracts away that complication.
+ */
+public interface StellarExecutor {
+
+ /**
+ * Execute an expression and assign the result to a variable.
+ *
+ * @param variable The name of the variable to assign to.
+ * @param expression The expression to execute.
+ * @param message The message that provides additional context for the expression.
+ */
+ void assign(String variable, String expression, JSONObject message);
+
+ /**
+ * Execute a Stellar expression and return the result.
+ *
+ * @param expression The expression to execute.
+ * @param message The message that is accessible when Stellar is executed.
+ * @param clazz The expected class of the expression's result.
+ * @param <T> The expected class of the expression's result.
+ */
+ <T> T execute(String expression, JSONObject message, Class<T> clazz);
+
+ /**
+ * The current state of the Stellar execution environment.
+ */
+ Map<String, Object> getState();
+
+ /**
+ * Removes all state from the execution environment.
+ */
+ void clearState();
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/main/scripts/start_profiler_topology.sh
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/main/scripts/start_profiler_topology.sh b/metron-analytics/metron-profiler/src/main/scripts/start_profiler_topology.sh
new file mode 100644
index 0000000..6f78471
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/main/scripts/start_profiler_topology.sh
@@ -0,0 +1,22 @@
+#!/bin/bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+METRON_VERSION=${project.version}
+METRON_HOME=/usr/metron/$METRON_VERSION
+TOPOLOGY_JAR=${project.artifactId}-$METRON_VERSION.jar
+storm jar $METRON_HOME/lib/$TOPOLOGY_JAR org.apache.storm.flux.Flux --remote $METRON_HOME/flux/profiler/remote.yaml --filter $METRON_HOME/config/profiler.properties
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-1/profiler.json
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-1/profiler.json b/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-1/profiler.json
new file mode 100644
index 0000000..f811df7
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-1/profiler.json
@@ -0,0 +1,17 @@
+{
+ "inputTopic": "indexing",
+ "profiles": [
+ {
+ "profile": "example1",
+ "foreach": "ip_src_addr",
+ "onlyif": "protocol == 'HTTP'",
+ "init": {
+ "total_bytes": 0.0
+ },
+ "update": {
+ "total_bytes": "total_bytes + bytes_in"
+ },
+ "result": "total_bytes"
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-2/profiler.json
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-2/profiler.json b/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-2/profiler.json
new file mode 100644
index 0000000..2c6a278
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-2/profiler.json
@@ -0,0 +1,19 @@
+{
+ "inputTopic": "indexing",
+ "profiles": [
+ {
+ "profile": "example2",
+ "foreach": "ip_src_addr",
+ "onlyif": "protocol == 'DNS' or protocol == 'HTTP'",
+ "init": {
+ "num_dns": 1.0,
+ "num_http": 1.0
+ },
+ "update": {
+ "num_dns": "num_dns + (if protocol == 'DNS' then 1 else 0)",
+ "num_http": "num_http + (if protocol == 'HTTP' then 1 else 0)"
+ },
+ "result": "num_dns / num_http"
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-3/profiler.json
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-3/profiler.json b/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-3/profiler.json
new file mode 100644
index 0000000..96c5935
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/config/zookeeper/readme-example-3/profiler.json
@@ -0,0 +1,20 @@
+{
+ "inputTopic": "indexing",
+ "profiles": [
+ {
+ "profile": "example3",
+ "foreach": "ip_src_addr",
+ "onlyif": "protocol == 'HTTP'",
+ "init": {
+ "sum": 0.0,
+ "cnt": 0.0
+ },
+ "update": {
+ "sum": "sum + resp_body_len",
+ "cnt": "cnt + 1"
+ },
+ "result": "sum / cnt"
+ }
+ ]
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/81754864/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileBuilderBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileBuilderBoltTest.java b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileBuilderBoltTest.java
new file mode 100644
index 0000000..1528e31
--- /dev/null
+++ b/metron-analytics/metron-profiler/src/test/java/org/apache/metron/profiler/bolt/ProfileBuilderBoltTest.java
@@ -0,0 +1,269 @@
+/*
+ *
+ * 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.profiler.bolt;
+
+import backtype.storm.Constants;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.metron.common.configuration.profiler.ProfileConfig;
+import org.apache.metron.common.utils.JSONUtils;
+import org.apache.metron.profiler.ProfileMeasurement;
+import org.apache.metron.profiler.stellar.DefaultStellarExecutor;
+import org.apache.metron.test.bolt.BaseBoltTest;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.io.IOException;
+import java.util.HashMap;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.*;
+
+/**
+ * Tests the ProfileBuilderBolt.
+ */
+public class ProfileBuilderBoltTest extends BaseBoltTest {
+
+ /**
+ * {
+ * "ip_src_addr": "10.0.0.1",
+ * "ip_dst_addr": "10.0.0.20"
+ * }
+ */
+ @Multiline
+ private String input;
+
+ /**
+ * {
+ * "profile": "test",
+ * "foreach": "ip_src_addr",
+ * "onlyif": "true",
+ * "init": {
+ * "x": 10,
+ * "y": 20
+ * },
+ * "update": {
+ * "x": "x + 10",
+ * "y": "y + 20"
+ * },
+ * "result": "x + y"
+ * }
+ */
+ @Multiline
+ private String profile;
+
+ /**
+ * {
+ * "profile": "test",
+ * "foreach": "ip_src_addr",
+ * "onlyif": "true",
+ * "init": { "x": 10 },
+ * "update": { "x": "x + 'string'" },
+ * "result": "x"
+ * }
+ */
+ @Multiline
+ private String profileWithBadUpdate;
+
+ /**
+ * {
+ * "profile": "test",
+ * "foreach": "ip_src_addr",
+ * "onlyif": "true",
+ * "init": { "x": "10 + 'string'" },
+ * "update": { "x": "x + 2" },
+ * "result": "x"
+ * }
+ */
+ @Multiline
+ private String profileWithBadInit;
+
+ private JSONObject message;
+
+ public static Tuple mockTickTuple() {
+ return mockTuple(Constants.SYSTEM_COMPONENT_ID, Constants.SYSTEM_TICK_STREAM_ID);
+ }
+
+ public static Tuple mockTuple(String componentId, String streamId) {
+ Tuple tuple = mock(Tuple.class);
+ when(tuple.getSourceComponent()).thenReturn(componentId);
+ when(tuple.getSourceStreamId()).thenReturn(streamId);
+ return tuple;
+ }
+
+ @Before
+ public void setup() throws Exception {
+
+ // parse the input message
+ JSONParser parser = new JSONParser();
+ message = (JSONObject) parser.parse(input);
+
+ // the tuple will contain the original message
+ when(tuple.getValueByField(eq("message"))).thenReturn(message);
+
+ // the tuple will contain the 'fully resolved' name of the entity
+ when(tuple.getStringByField(eq("entity"))).thenReturn("10.0.0.1");
+
+ // the tuple will contain the profile definition
+ ProfileConfig profileConfig = JSONUtils.INSTANCE.load(profile, ProfileConfig.class);
+ when(tuple.getValueByField(eq("profile"))).thenReturn(profileConfig);
+ }
+
+ /**
+ * Create a ProfileBuilderBolt to test
+ */
+ private ProfileBuilderBolt createBolt() throws IOException {
+
+ ProfileBuilderBolt bolt = new ProfileBuilderBolt("zookeeperURL");
+ bolt.setCuratorFramework(client);
+ bolt.setTreeCache(cache);
+ bolt.setExecutor(new DefaultStellarExecutor());
+
+ bolt.prepare(new HashMap<>(), topologyContext, outputCollector);
+ return bolt;
+ }
+
+ /**
+ * Ensure that the bolt can update a profile based on new messages that it receives.
+ */
+ @Test
+ public void testUpdateProfile() throws Exception {
+
+ ProfileBuilderBolt bolt = createBolt();
+ bolt.execute(tuple);
+ bolt.execute(tuple);
+
+ // validate that x=10+10+10 y=20+20+20
+ assertEquals(10+10+10.0, bolt.getExecutor().getState().get("x"));
+ assertEquals(20+20+20.0, bolt.getExecutor().getState().get("y"));
+ }
+
+ /**
+ * Ensure that the bolt can flush the profile when a tick tuple is received.
+ */
+ @Test
+ public void testFlushProfile() throws Exception {
+
+ // setup
+ ProfileBuilderBolt bolt = createBolt();
+ bolt.execute(tuple);
+ bolt.execute(tuple);
+
+ // execute - the tick tuple triggers a flush of the profile
+ bolt.execute(mockTickTuple());
+
+ // capture the ProfileMeasurement that should be emitted
+ ArgumentCaptor<Values> arg = ArgumentCaptor.forClass(Values.class);
+ verify(outputCollector, times(1)).emit(refEq(tuple), arg.capture());
+
+ Values actual = arg.getValue();
+ ProfileMeasurement measurement = (ProfileMeasurement) actual.get(0);
+
+ // verify
+ assertThat(measurement.getValue(), equalTo(90.0));
+ assertThat(measurement.getEntity(), equalTo("10.0.0.1"));
+ assertThat(measurement.getProfileName(), equalTo("test"));
+ }
+
+ /**
+ * What happens if we try to flush, but have yet to receive any messages to
+ * apply to the profile?
+ *
+ * The ProfileBuilderBolt will not have received the data necessary from the
+ * ProfileSplitterBolt, like the entity and profile name, that is required
+ * to perform the flush. The flush has to be skipped until this information
+ * is received from the Splitter.
+ */
+ @Test
+ public void testFlushProfileWithNoMessages() throws Exception {
+
+ ProfileBuilderBolt bolt = createBolt();
+
+ // no messages have been received before a flush occurs
+ bolt.execute(mockTickTuple());
+ bolt.execute(mockTickTuple());
+ bolt.execute(mockTickTuple());
+
+ // no ProfileMeasurement should be written to the ProfileStore
+ verify(outputCollector, times(0)).emit(any(Values.class));
+ }
+
+ /**
+ * The executor's state should be cleared after a flush.
+ */
+ @Test
+ public void testStateClearedAfterFlush() throws Exception {
+
+ ProfileBuilderBolt bolt = createBolt();
+ bolt.execute(tuple);
+ bolt.execute(tuple);
+
+ // execute - should clear state from previous tuples
+ bolt.execute(mockTickTuple());
+
+ assertThat(bolt.getExecutor().getState().size(), equalTo(0));
+ }
+
+ /**
+ * What happens when the profile contains a bad Stellar expression?
+ */
+ @Test
+ public void testProfileWithBadUpdate() throws Exception {
+
+ // setup - ensure the bad profile is used
+ ProfileConfig profileConfig = JSONUtils.INSTANCE.load(profileWithBadUpdate, ProfileConfig.class);
+ when(tuple.getValueByField(eq("profile"))).thenReturn(profileConfig);
+
+ // execute
+ ProfileBuilderBolt bolt = createBolt();
+ bolt.execute(tuple);
+
+ // verify - expect the tuple to be acked and an error reported
+ verify(outputCollector, times(1)).ack(eq(tuple));
+ verify(outputCollector, times(1)).reportError(any());
+ }
+
+ /**
+ * What happens when the profile contains a bad Stellar expression?
+ */
+ @Test
+ public void testProfileWithBadInit() throws Exception {
+
+ // setup - ensure the bad profile is used
+ ProfileConfig profileConfig = JSONUtils.INSTANCE.load(profileWithBadInit, ProfileConfig.class);
+ when(tuple.getValueByField(eq("profile"))).thenReturn(profileConfig);
+
+ // execute
+ ProfileBuilderBolt bolt = createBolt();
+ bolt.execute(tuple);
+
+ // verify - expect the tuple to be acked and an error reported
+ verify(outputCollector, times(1)).ack(eq(tuple));
+ verify(outputCollector, times(1)).reportError(any());
+ }
+}