You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ce...@apache.org on 2017/08/07 18:14:58 UTC
[2/4] metron git commit: METRON-1051: Enable the ability to update
indexed messages closes apache/incubator-metron#666
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/mock/MockHTable.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/mock/MockHTable.java b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/mock/MockHTable.java
new file mode 100644
index 0000000..e75b533
--- /dev/null
+++ b/metron-platform/metron-hbase/src/test/java/org/apache/metron/hbase/mock/MockHTable.java
@@ -0,0 +1,701 @@
+/**
+ * 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.mock;
+
+
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.Message;
+import com.google.protobuf.Service;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.hbase.TableProvider;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.NavigableSet;
+import java.util.NoSuchElementException;
+import java.util.TreeMap;
+
+/**
+ * MockHTable.
+ *
+ * This implementation is a selected excerpt from https://gist.github.com/agaoglu/613217
+ */
+public class MockHTable implements HTableInterface {
+
+
+ private final String tableName;
+ private final List<String> columnFamilies = new ArrayList<>();
+ private HColumnDescriptor[] descriptors;
+ private final List<Put> putLog;
+ private NavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>>> data
+ = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+
+ private static List<KeyValue> toKeyValue(byte[] row, NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> rowdata, int maxVersions) {
+ return toKeyValue(row, rowdata, 0, Long.MAX_VALUE, maxVersions);
+ }
+
+ private static List<KeyValue> toKeyValue(byte[] row, NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> rowdata, long timestampStart, long timestampEnd, int maxVersions) {
+ List<KeyValue> ret = new ArrayList<KeyValue>();
+ for (byte[] family : rowdata.keySet())
+ for (byte[] qualifier : rowdata.get(family).keySet()) {
+ int versionsAdded = 0;
+ for (Map.Entry<Long, byte[]> tsToVal : rowdata.get(family).get(qualifier).descendingMap().entrySet()) {
+ if (versionsAdded++ == maxVersions)
+ break;
+ Long timestamp = tsToVal.getKey();
+ if (timestamp < timestampStart)
+ continue;
+ if (timestamp > timestampEnd)
+ continue;
+ byte[] value = tsToVal.getValue();
+ ret.add(new KeyValue(row, family, qualifier, timestamp, value));
+ }
+ }
+ return ret;
+ }
+ public MockHTable(String tableName) {
+ this.tableName = tableName;
+ this.putLog = new ArrayList<>();
+ }
+
+ public MockHTable(String tableName, String... columnFamilies) {
+ this(tableName);
+ for(String cf : columnFamilies) {
+ addColumnFamily(cf);
+ }
+ }
+
+ public int size() {
+ return data.size();
+ }
+
+ public void addColumnFamily(String columnFamily) {
+ this.columnFamilies.add(columnFamily);
+ descriptors = new HColumnDescriptor[columnFamilies.size()];
+ int i = 0;
+ for(String cf : columnFamilies) {
+ descriptors[i++] = new HColumnDescriptor(cf);
+ }
+ }
+
+ @Override
+ public byte[] getTableName() {
+ return Bytes.toBytes(tableName);
+ }
+
+ @Override
+ public TableName getName() {
+ return TableName.valueOf(tableName);
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return HBaseConfiguration.create();
+ }
+
+ @Override
+ public HTableDescriptor getTableDescriptor() throws IOException {
+ HTableDescriptor ret = new HTableDescriptor(tableName);
+ for(HColumnDescriptor c : descriptors) {
+ ret.addFamily(c);
+ }
+ return ret;
+ }
+
+ @Override
+ public boolean exists(Get get) throws IOException {
+ if(get.getFamilyMap() == null || get.getFamilyMap().size() == 0) {
+ return data.containsKey(get.getRow());
+ } else {
+ byte[] row = get.getRow();
+ if(!data.containsKey(row)) {
+ return false;
+ }
+ for(byte[] family : get.getFamilyMap().keySet()) {
+ if(!data.get(row).containsKey(family)) {
+ return false;
+ } else {
+ return true;
+ }
+ }
+ return true;
+ }
+ }
+
+ /**
+ * Test for the existence of columns in the table, as specified by the Gets.
+ *
+ * <p>This will return an array of booleans. Each value will be true if the related Get matches
+ * one or more keys, false if not.
+ *
+ * <p>This is a server-side call so it prevents any data from being transferred to
+ * the client.
+ *
+ * @param gets the Gets
+ * @return Array of boolean. True if the specified Get matches one or more keys, false if not.
+ * @throws IOException e
+ */
+ @Override
+ public boolean[] existsAll(List<Get> gets) throws IOException {
+ boolean[] ret = new boolean[gets.size()];
+ int i = 0;
+ for(boolean b : exists(gets)) {
+ ret[i++] = b;
+ }
+ return ret;
+ }
+
+ @Override
+ public Boolean[] exists(List<Get> list) throws IOException {
+ Boolean[] ret = new Boolean[list.size()];
+ int i = 0;
+ for(Get g : list) {
+ ret[i++] = exists(g);
+ }
+ return ret;
+ }
+
+ @Override
+ public void batch(List<? extends Row> list, Object[] objects) throws IOException, InterruptedException {
+ Object[] results = batch(list);
+ System.arraycopy(results, 0, objects, 0, results.length);
+ }
+
+ /**
+ * @param actions
+ * @deprecated
+ */
+ @Deprecated
+ @Override
+ public Object[] batch(List<? extends Row> actions) throws IOException, InterruptedException {
+ List<Result> results = new ArrayList<Result>();
+ for (Row r : actions) {
+ if (r instanceof Delete) {
+ delete((Delete) r);
+ continue;
+ }
+ if (r instanceof Put) {
+ put((Put) r);
+ continue;
+ }
+ if (r instanceof Get) {
+ results.add(get((Get) r));
+ }
+ }
+ return results.toArray();
+ }
+
+ @Override
+ public <R> void batchCallback(List<? extends Row> list, Object[] objects, Batch.Callback<R> callback) throws IOException, InterruptedException {
+ throw new UnsupportedOperationException();
+
+ }
+
+ /**
+ * @param list
+ * @param callback
+ * @deprecated
+ */
+ @Deprecated
+ @Override
+ public <R> Object[] batchCallback(List<? extends Row> list, Batch.Callback<R> callback) throws IOException, InterruptedException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Result get(Get get) throws IOException {
+ if (!data.containsKey(get.getRow()))
+ return new Result();
+ byte[] row = get.getRow();
+ List<KeyValue> kvs = new ArrayList<KeyValue>();
+ if (!get.hasFamilies()) {
+ kvs = toKeyValue(row, data.get(row), get.getMaxVersions());
+ } else {
+ for (byte[] family : get.getFamilyMap().keySet()){
+ if (data.get(row).get(family) == null)
+ continue;
+ NavigableSet<byte[]> qualifiers = get.getFamilyMap().get(family);
+ if (qualifiers == null || qualifiers.isEmpty())
+ qualifiers = data.get(row).get(family).navigableKeySet();
+ for (byte[] qualifier : qualifiers){
+ if (qualifier == null)
+ qualifier = "".getBytes();
+ if (!data.get(row).containsKey(family) ||
+ !data.get(row).get(family).containsKey(qualifier) ||
+ data.get(row).get(family).get(qualifier).isEmpty())
+ continue;
+ Map.Entry<Long, byte[]> timestampAndValue = data.get(row).get(family).get(qualifier).lastEntry();
+ kvs.add(new KeyValue(row,family, qualifier, timestampAndValue.getKey(), timestampAndValue.getValue()));
+ }
+ }
+ }
+ Filter filter = get.getFilter();
+ if (filter != null) {
+ filter.reset();
+ List<KeyValue> nkvs = new ArrayList<KeyValue>(kvs.size());
+ for (KeyValue kv : kvs) {
+ if (filter.filterAllRemaining()) {
+ break;
+ }
+ if (filter.filterRowKey(kv.getBuffer(), kv.getRowOffset(), kv.getRowLength())) {
+ continue;
+ }
+ if (filter.filterKeyValue(kv) == Filter.ReturnCode.INCLUDE) {
+ nkvs.add(kv);
+ }
+ // ignoring next key hint which is a optimization to reduce file system IO
+ }
+ if (filter.hasFilterRow()) {
+ filter.filterRow();
+ }
+ kvs = nkvs;
+ }
+
+ return new Result(kvs);
+ }
+
+ @Override
+ public Result[] get(List<Get> list) throws IOException {
+ Result[] ret = new Result[list.size()];
+ int i = 0;
+ for(Get g : list) {
+ ret[i++] = get(g);
+ }
+ return ret;
+ }
+
+ /**
+ * @param bytes
+ * @param bytes1
+ * @deprecated
+ */
+ @Deprecated
+ @Override
+ public Result getRowOrBefore(byte[] bytes, byte[] bytes1) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ResultScanner getScanner(Scan scan) throws IOException {
+ final List<Result> ret = new ArrayList<Result>();
+ byte[] st = scan.getStartRow();
+ byte[] sp = scan.getStopRow();
+ Filter filter = scan.getFilter();
+
+ for (byte[] row : data.keySet()){
+ // if row is equal to startRow emit it. When startRow (inclusive) and
+ // stopRow (exclusive) is the same, it should not be excluded which would
+ // happen w/o this control.
+ if (st != null && st.length > 0 &&
+ Bytes.BYTES_COMPARATOR.compare(st, row) != 0) {
+ // if row is before startRow do not emit, pass to next row
+ if (st != null && st.length > 0 &&
+ Bytes.BYTES_COMPARATOR.compare(st, row) > 0)
+ continue;
+ // if row is equal to stopRow or after it do not emit, stop iteration
+ if (sp != null && sp.length > 0 &&
+ Bytes.BYTES_COMPARATOR.compare(sp, row) <= 0)
+ break;
+ }
+
+ List<KeyValue> kvs = null;
+ if (!scan.hasFamilies()) {
+ kvs = toKeyValue(row, data.get(row), scan.getTimeRange().getMin(), scan.getTimeRange().getMax(), scan.getMaxVersions());
+ } else {
+ kvs = new ArrayList<KeyValue>();
+ for (byte[] family : scan.getFamilyMap().keySet()){
+ if (data.get(row).get(family) == null)
+ continue;
+ NavigableSet<byte[]> qualifiers = scan.getFamilyMap().get(family);
+ if (qualifiers == null || qualifiers.isEmpty())
+ qualifiers = data.get(row).get(family).navigableKeySet();
+ for (byte[] qualifier : qualifiers){
+ if (data.get(row).get(family).get(qualifier) == null)
+ continue;
+ for (Long timestamp : data.get(row).get(family).get(qualifier).descendingKeySet()){
+ if (timestamp < scan.getTimeRange().getMin())
+ continue;
+ if (timestamp > scan.getTimeRange().getMax())
+ continue;
+ byte[] value = data.get(row).get(family).get(qualifier).get(timestamp);
+ kvs.add(new KeyValue(row, family, qualifier, timestamp, value));
+ if(kvs.size() == scan.getMaxVersions()) {
+ break;
+ }
+ }
+ }
+ }
+ }
+ if (filter != null) {
+ filter.reset();
+ List<KeyValue> nkvs = new ArrayList<KeyValue>(kvs.size());
+ for (KeyValue kv : kvs) {
+ if (filter.filterAllRemaining()) {
+ break;
+ }
+ if (filter.filterRowKey(kv.getBuffer(), kv.getRowOffset(), kv.getRowLength())) {
+ continue;
+ }
+ Filter.ReturnCode filterResult = filter.filterKeyValue(kv);
+ if (filterResult == Filter.ReturnCode.INCLUDE) {
+ nkvs.add(kv);
+ } else if (filterResult == Filter.ReturnCode.NEXT_ROW) {
+ break;
+ }
+ // ignoring next key hint which is a optimization to reduce file system IO
+ }
+ if (filter.hasFilterRow()) {
+ filter.filterRow();
+ }
+ kvs = nkvs;
+ }
+ if (!kvs.isEmpty()) {
+ ret.add(new Result(kvs));
+ }
+ }
+
+ return new ResultScanner() {
+ private final Iterator<Result> iterator = ret.iterator();
+ @Override
+ public Iterator<Result> iterator() {
+ return iterator;
+ }
+ @Override
+ public Result[] next(int nbRows) throws IOException {
+ ArrayList<Result> resultSets = new ArrayList<Result>(nbRows);
+ for(int i = 0; i < nbRows; i++) {
+ Result next = next();
+ if (next != null) {
+ resultSets.add(next);
+ } else {
+ break;
+ }
+ }
+ return resultSets.toArray(new Result[resultSets.size()]);
+ }
+ @Override
+ public Result next() throws IOException {
+ try {
+ return iterator().next();
+ } catch (NoSuchElementException e) {
+ return null;
+ }
+ }
+ @Override
+ public void close() {}
+ };
+ }
+ @Override
+ public ResultScanner getScanner(byte[] family) throws IOException {
+ Scan scan = new Scan();
+ scan.addFamily(family);
+ return getScanner(scan);
+ }
+
+ @Override
+ public ResultScanner getScanner(byte[] family, byte[] qualifier)
+ throws IOException {
+ Scan scan = new Scan();
+ scan.addColumn(family, qualifier);
+ return getScanner(scan);
+ }
+
+ public List<Put> getPutLog() {
+ synchronized (putLog) {
+ return ImmutableList.copyOf(putLog);
+ }
+ }
+
+ public void addToPutLog(Put put) {
+ synchronized(putLog) {
+ putLog.add(put);
+ }
+ }
+
+ public void clear() {
+ synchronized (putLog) {
+ putLog.clear();
+ }
+ data.clear();
+ }
+
+ @Override
+ public void put(Put put) throws IOException {
+ addToPutLog(put);
+
+ byte[] row = put.getRow();
+ NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> rowData = forceFind(data, row, new TreeMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>>(Bytes.BYTES_COMPARATOR));
+ for (byte[] family : put.getFamilyMap().keySet()){
+ NavigableMap<byte[], NavigableMap<Long, byte[]>> familyData = forceFind(rowData, family, new TreeMap<byte[], NavigableMap<Long, byte[]>>(Bytes.BYTES_COMPARATOR));
+ for (KeyValue kv : put.getFamilyMap().get(family)){
+ kv.updateLatestStamp(Bytes.toBytes(System.currentTimeMillis()));
+ byte[] qualifier = kv.getQualifier();
+ NavigableMap<Long, byte[]> qualifierData = forceFind(familyData, qualifier, new TreeMap<Long, byte[]>());
+ qualifierData.put(kv.getTimestamp(), kv.getValue());
+ }
+ }
+ }
+
+ /**
+ * Helper method to find a key in a map. If key is not found, newObject is
+ * added to map and returned
+ *
+ * @param map
+ * map to extract value from
+ * @param key
+ * key to look for
+ * @param newObject
+ * set key to this if not found
+ * @return found value or newObject if not found
+ */
+ private <K, V> V forceFind(NavigableMap<K, V> map, K key, V newObject){
+ V data = map.get(key);
+ if (data == null){
+ data = newObject;
+ map.put(key, data);
+ }
+ return data;
+ }
+
+ @Override
+ public void put(List<Put> puts) throws IOException {
+ for (Put put : puts)
+ put(put);
+ }
+
+ @Override
+ public boolean checkAndPut(byte[] bytes, byte[] bytes1, byte[] bytes2, byte[] bytes3, Put put) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Atomically checks if a row/family/qualifier value matches the expected
+ * value. If it does, it adds the put. If the passed value is null, the check
+ * is for the lack of column (ie: non-existance)
+ *
+ * @param row to check
+ * @param family column family to check
+ * @param qualifier column qualifier to check
+ * @param compareOp comparison operator to use
+ * @param value the expected value
+ * @param put data to put if check succeeds
+ * @return true if the new put was executed, false otherwise
+ * @throws IOException e
+ */
+ @Override
+ public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier, CompareFilter.CompareOp compareOp, byte[] value, Put put) throws IOException {
+ return false;
+ }
+
+ @Override
+ public void delete(Delete delete) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void delete(List<Delete> list) throws IOException {
+ throw new UnsupportedOperationException();
+
+ }
+
+ @Override
+ public boolean checkAndDelete(byte[] bytes, byte[] bytes1, byte[] bytes2, byte[] bytes3, Delete delete) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Atomically checks if a row/family/qualifier value matches the expected
+ * value. If it does, it adds the delete. If the passed value is null, the
+ * check is for the lack of column (ie: non-existance)
+ *
+ * @param row to check
+ * @param family column family to check
+ * @param qualifier column qualifier to check
+ * @param compareOp comparison operator to use
+ * @param value the expected value
+ * @param delete data to delete if check succeeds
+ * @return true if the new delete was executed, false otherwise
+ * @throws IOException e
+ */
+ @Override
+ public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier, CompareFilter.CompareOp compareOp, byte[] value, Delete delete) throws IOException {
+ return false;
+ }
+
+ @Override
+ public void mutateRow(RowMutations rowMutations) throws IOException {
+ throw new UnsupportedOperationException();
+
+ }
+
+ @Override
+ public Result append(Append append) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Result increment(Increment increment) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long incrementColumnValue(byte[] bytes, byte[] bytes1, byte[] bytes2, long l) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long incrementColumnValue(byte[] bytes, byte[] bytes1, byte[] bytes2, long l, Durability durability) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * @param bytes
+ * @param bytes1
+ * @param bytes2
+ * @param l
+ * @param b
+ * @deprecated
+ */
+ @Deprecated
+ @Override
+ public long incrementColumnValue(byte[] bytes, byte[] bytes1, byte[] bytes2, long l, boolean b) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isAutoFlush() {
+ return autoflush;
+ }
+
+ @Override
+ public void flushCommits() throws IOException {
+
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+
+ @Override
+ public CoprocessorRpcChannel coprocessorService(byte[] bytes) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> aClass, byte[] bytes, byte[] bytes1, Batch.Call<T, R> call) throws ServiceException, Throwable {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T extends Service, R> void coprocessorService(Class<T> aClass, byte[] bytes, byte[] bytes1, Batch.Call<T, R> call, Batch.Callback<R> callback) throws ServiceException, Throwable {
+ throw new UnsupportedOperationException();
+ }
+
+ boolean autoflush = true;
+
+ /**
+ * @param b
+ * @deprecated
+ */
+ @Deprecated
+ @Override
+ public void setAutoFlush(boolean b) {
+ autoflush = b;
+ }
+
+ @Override
+ public void setAutoFlush(boolean b, boolean b1) {
+ autoflush = b;
+ }
+
+ @Override
+ public void setAutoFlushTo(boolean b) {
+ autoflush = b;
+ }
+
+ long writeBufferSize = 0;
+ @Override
+ public long getWriteBufferSize() {
+ return writeBufferSize;
+ }
+
+ @Override
+ public void setWriteBufferSize(long l) throws IOException {
+ writeBufferSize = l;
+ }
+
+ @Override
+ public <R extends Message> Map<byte[], R> batchCoprocessorService(Descriptors.MethodDescriptor methodDescriptor, Message message, byte[] bytes, byte[] bytes1, R r) throws ServiceException, Throwable {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <R extends Message> void batchCoprocessorService(Descriptors.MethodDescriptor methodDescriptor, Message message, byte[] bytes, byte[] bytes1, R r, Batch.Callback<R> callback) throws ServiceException, Throwable {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Atomically checks if a row/family/qualifier value matches the expected value.
+ * If it does, it performs the row mutations. If the passed value is null, the check
+ * is for the lack of column (ie: non-existence)
+ *
+ * @param row to check
+ * @param family column family to check
+ * @param qualifier column qualifier to check
+ * @param compareOp the comparison operator
+ * @param value the expected value
+ * @param mutation mutations to perform if check succeeds
+ * @return true if the new put was executed, false otherwise
+ * @throws IOException e
+ */
+ @Override
+ public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier, CompareFilter.CompareOp compareOp, byte[] value, RowMutations mutation) throws IOException {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/README.md b/metron-platform/metron-indexing/README.md
index 0cf3a66..2095d0f 100644
--- a/metron-platform/metron-indexing/README.md
+++ b/metron-platform/metron-indexing/README.md
@@ -111,6 +111,30 @@ Storm console. e.g.:
* hdfs writer
* disabled
+# Updates to Indexed Data
+
+There are clear usecases where we would want to incorporate the capability to update indexed data.
+Thus far, we have limited capabilities provided to support this use-case:
+* Updates to the random access index (e.g. Elasticsearch and Solr) should be supported
+* Updates to the cold storage index (e.g. HDFS) is not supported currently, however to support the batch
+use-case updated documents will be provided in a NoSQL write-ahead log (e.g. a HBase table) and an Java API
+will be provided to retrieve those updates scalably (i.e. a scan-free architecture).
+
+Put simply, the random access index will be always up-to-date, but the HDFS index will need to be
+joined to the NoSQL write-ahead log to get current updates.
+
+## The `IndexDao` Abstraction
+
+The indices mentioned above as part of Update should be pluggable by the developer so that
+new write-ahead logs or real-time indices can be supported by providing an implementation supporting
+the data access patterns.
+
+To support a new index, one would need to implement the `org.apache.metron.indexing.dao.IndexDao` abstraction
+and provide update and search capabilities. IndexDaos may be composed and updates will be performed
+in parallel. This enables a flexible strategy for specifying your backing store for updates at runtime.
+For instance, currently the REST API supports the update functionality and may be configured with a list of
+IndexDao implementations to use to support the updates.
+
# Notes on Performance Tuning
Default installed Metron is untuned for production deployment. By far
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/pom.xml b/metron-platform/metron-indexing/pom.xml
index b0127bb..c64c374 100644
--- a/metron-platform/metron-indexing/pom.xml
+++ b/metron-platform/metron-indexing/pom.xml
@@ -37,6 +37,32 @@
<version>${project.parent.version}</version>
</dependency>
<dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>metron-hbase</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${global_hbase_version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>${global_hadoop_version}</version>
@@ -47,7 +73,27 @@
</exclusion>
</exclusions>
<scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.flipkart.zjsonpatch</groupId>
+ <artifactId>zjsonpatch</artifactId>
+ <version>0.3.1</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
+
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java
index dd68484..ddb88e5 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java
@@ -17,13 +17,35 @@
*/
package org.apache.metron.indexing.dao;
+import org.apache.metron.hbase.HTableProvider;
+import org.apache.metron.hbase.TableProvider;
+
import java.util.HashMap;
import java.util.Map;
+import java.util.function.Supplier;
public class AccessConfig {
private Integer maxSearchResults;
+ private Supplier<Map<String, Object>> globalConfigSupplier;
private Map<String, String> optionalSettings = new HashMap<>();
+ private TableProvider tableProvider = null;
+
+ /**
+ * A supplier which will return the current global config.
+ * @return
+ */
+ public Supplier<Map<String, Object>> getGlobalConfigSupplier() {
+ return globalConfigSupplier;
+ }
+
+ public void setGlobalConfigSupplier(Supplier<Map<String, Object>> globalConfigSupplier) {
+ this.globalConfigSupplier = globalConfigSupplier;
+ }
+ /**
+ * The maximum search result.
+ * @return
+ */
public Integer getMaxSearchResults() {
return maxSearchResults;
}
@@ -32,6 +54,10 @@ public class AccessConfig {
this.maxSearchResults = maxSearchResults;
}
+ /**
+ * Get optional settings for initializing indices.
+ * @return
+ */
public Map<String, String> getOptionalSettings() {
return optionalSettings;
}
@@ -39,4 +65,17 @@ public class AccessConfig {
public void setOptionalSettings(Map<String, String> optionalSettings) {
this.optionalSettings = optionalSettings;
}
+
+ /**
+ * Return the table provider to use for NoSql DAOs
+ * @return
+ */
+ public TableProvider getTableProvider() {
+ return tableProvider;
+ }
+
+ public void setTableProvider(TableProvider tableProvider) {
+ this.tableProvider = tableProvider;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java
new file mode 100644
index 0000000..a1cf398
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java
@@ -0,0 +1,135 @@
+/**
+ * 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.indexing.dao;
+
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.common.configuration.writer.WriterConfiguration;
+import org.apache.metron.common.utils.JSONUtils;
+import org.apache.metron.indexing.dao.search.FieldType;
+import org.apache.metron.indexing.dao.search.InvalidSearchException;
+import org.apache.metron.indexing.dao.search.SearchRequest;
+import org.apache.metron.indexing.dao.search.SearchResponse;
+import org.apache.metron.indexing.dao.update.Document;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Optional;
+
+/**
+ * The HBaseDao is an index dao which only supports the following actions:
+ * * Update
+ * * Get document
+ *
+ * The mechanism here is that updates to documents will be added to a HBase Table as a write-ahead log.
+ * The Key for a row supporting a given document will be the GUID, which should be sufficiently distributed.
+ * Every new update will have a column added (column qualifier will be the timestamp of the update).
+ * Upon retrieval, the most recent column will be returned.
+ *
+ */
+public class HBaseDao implements IndexDao {
+ public static String HBASE_TABLE = "update.hbase.table";
+ public static String HBASE_CF = "update.hbase.cf";
+ private HTableInterface tableInterface;
+ private byte[] cf;
+ private AccessConfig config;
+ public HBaseDao() {
+
+ }
+
+ @Override
+ public synchronized SearchResponse search(SearchRequest searchRequest) throws InvalidSearchException {
+ return null;
+ }
+
+ @Override
+ public synchronized void init(AccessConfig config) {
+ if(this.tableInterface == null) {
+ this.config = config;
+ Map<String, Object> globalConfig = config.getGlobalConfigSupplier().get();
+ if(globalConfig == null) {
+ throw new IllegalStateException("Cannot find the global config.");
+ }
+ String table = (String)globalConfig.get(HBASE_TABLE);
+ String cf = (String) config.getGlobalConfigSupplier().get().get(HBASE_CF);
+ if(table == null || cf == null) {
+ throw new IllegalStateException("You must configure " + HBASE_TABLE + " and " + HBASE_CF + " in the global config.");
+ }
+ try {
+ tableInterface = config.getTableProvider().getTable(HBaseConfiguration.create(), table);
+ this.cf = cf.getBytes();
+ } catch (IOException e) {
+ throw new IllegalStateException("Unable to initialize HBaseDao: " + e.getMessage(), e);
+ }
+ }
+ }
+
+ public HTableInterface getTableInterface() {
+ if(tableInterface == null) {
+ init(config);
+ }
+ return tableInterface;
+ }
+
+ @Override
+ public synchronized Document getLatest(String guid, String sensorType) throws IOException {
+ Get get = new Get(guid.getBytes());
+ get.addFamily(cf);
+ Result result = getTableInterface().get(get);
+ NavigableMap<byte[], byte[]> columns = result.getFamilyMap( cf);
+ if(columns == null || columns.size() == 0) {
+ return null;
+ }
+ Map.Entry<byte[], byte[]> entry= columns.lastEntry();
+ Long ts = Bytes.toLong(entry.getKey());
+ if(entry.getValue()!= null) {
+ String json = new String(entry.getValue());
+ return new Document(json, guid, sensorType, ts);
+ }
+ else {
+ return null;
+ }
+ }
+
+ @Override
+ public synchronized void update(Document update, Optional<String> index) throws IOException {
+ Put put = new Put(update.getGuid().getBytes());
+ long ts = update.getTimestamp() == null?System.currentTimeMillis():update.getTimestamp();
+ byte[] columnQualifier = Bytes.toBytes(ts);
+ byte[] doc = JSONUtils.INSTANCE.toJSON(update.getDocument());
+ put.addColumn(cf, columnQualifier, doc);
+ getTableInterface().put(put);
+ }
+
+ @Override
+ public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> indices) throws IOException {
+ return null;
+ }
+
+ @Override
+ public Map<String, FieldType> getCommonColumnMetadata(List<String> indices) throws IOException {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java
index 31fe74e..350e402 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java
@@ -17,18 +17,129 @@
*/
package org.apache.metron.indexing.dao;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.flipkart.zjsonpatch.JsonPatch;
+import org.apache.metron.common.utils.JSONUtils;
+import org.apache.metron.indexing.dao.search.GetRequest;
import org.apache.metron.indexing.dao.search.InvalidSearchException;
import org.apache.metron.indexing.dao.search.SearchRequest;
import org.apache.metron.indexing.dao.search.SearchResponse;
+import org.apache.metron.indexing.dao.update.Document;
+import org.apache.metron.indexing.dao.update.PatchRequest;
+import org.apache.metron.indexing.dao.update.ReplaceRequest;
+import org.apache.metron.indexing.dao.update.OriginalNotFoundException;
+
+import java.io.IOException;
import org.apache.metron.indexing.dao.search.FieldType;
import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public interface IndexDao {
+
+ /**
+ * Return search response based on the search request
+ *
+ * @param searchRequest
+ * @return
+ * @throws InvalidSearchException
+ */
SearchResponse search(SearchRequest searchRequest) throws InvalidSearchException;
- void init(Map<String, Object> globalConfig, AccessConfig config);
+
+ /**
+ * Initialize the DAO with the AccessConfig object.
+ * @param config
+ */
+ void init(AccessConfig config);
+
+ /**
+ * Return the latest version of a document given the GUID and the sensor type.
+ *
+ * @param guid The GUID for the document
+ * @param sensorType The sensor type of the document
+ * @return The Document matching or null if not available.
+ * @throws IOException
+ */
+ Document getLatest(String guid, String sensorType) throws IOException;
+
+ /**
+ * Return the latest version of a document given a GetRequest.
+ * @param request The GetRequest which indicates the GUID and sensor type.
+ * @return Optionally the document (dependent upon existence in the index).
+ * @throws IOException
+ */
+ default Optional<Map<String, Object>> getLatestResult(GetRequest request) throws IOException {
+ Document ret = getLatest(request.getGuid(), request.getSensorType());
+ if(ret == null) {
+ return Optional.empty();
+ }
+ else {
+ return Optional.ofNullable(ret.getDocument());
+ }
+ }
+
+ /**
+ * Update given a Document and optionally the index where the document exists.
+ *
+ * @param update The document to replace from the index.
+ * @param index The index where the document lives.
+ * @throws IOException
+ */
+ void update(Document update, Optional<String> index) throws IOException;
+
+
+ /**
+ * Update a document in an index given a JSON Patch (see RFC 6902 at https://tools.ietf.org/html/rfc6902)
+ * @param request The patch request
+ * @param timestamp Optionally a timestamp to set. If not specified then current time is used.
+ * @throws OriginalNotFoundException If the original is not found, then it cannot be patched.
+ * @throws IOException
+ */
+ default void patch( PatchRequest request
+ , Optional<Long> timestamp
+ ) throws OriginalNotFoundException, IOException {
+ Map<String, Object> latest = request.getSource();
+ if(latest == null) {
+ Document latestDoc = getLatest(request.getGuid(), request.getSensorType());
+ if(latestDoc.getDocument() != null) {
+ latest = latestDoc.getDocument();
+ }
+ else {
+ throw new OriginalNotFoundException("Unable to patch an document that doesn't exist and isn't specified.");
+ }
+ }
+ JsonNode originalNode = JSONUtils.INSTANCE.convert(latest, JsonNode.class);
+ JsonNode patched = JsonPatch.apply(request.getPatch(), originalNode);
+ Map<String, Object> updated = JSONUtils.INSTANCE.getMapper()
+ .convertValue(patched, new TypeReference<Map<String, Object>>() {});
+ Document d = new Document( updated
+ , request.getGuid()
+ , request.getSensorType()
+ , timestamp.orElse(System.currentTimeMillis())
+ );
+ update(d, Optional.ofNullable(request.getIndex()));
+ }
+
+ /**
+ * Replace a document in an index.
+ * @param request The replacement request.
+ * @param timestamp The timestamp (optional) of the update. If not specified, then current time will be used.
+ * @throws IOException
+ */
+ default void replace( ReplaceRequest request
+ , Optional<Long> timestamp
+ ) throws IOException {
+ Document d = new Document(request.getReplacement()
+ , request.getGuid()
+ , request.getSensorType()
+ , timestamp.orElse(System.currentTimeMillis())
+ );
+ update(d, Optional.ofNullable(request.getIndex()));
+ }
+
Map<String, Map<String, FieldType>> getColumnMetadata(List<String> indices) throws IOException;
Map<String, FieldType> getCommonColumnMetadata(List<String> indices) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDaoFactory.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDaoFactory.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDaoFactory.java
index 9c2de0e..e8df0b7 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDaoFactory.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDaoFactory.java
@@ -17,14 +17,42 @@
*/
package org.apache.metron.indexing.dao;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
+
import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
+import java.util.function.Function;
public class IndexDaoFactory {
- public static IndexDao create(String daoImpl, Map<String, Object> globalConfig, AccessConfig config) throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
- Class<? extends IndexDao> clazz = (Class<? extends IndexDao>) Class.forName(daoImpl);
- IndexDao instance = clazz.getConstructor().newInstance();
- instance.init(globalConfig, config);
- return instance;
+ public static List<IndexDao> create( String daoImpls
+ , AccessConfig config
+ ) throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException
+ {
+ List<IndexDao> ret = new ArrayList<>();
+ for(String daoImpl : Splitter.on(",").split(daoImpls)) {
+ Class<? extends IndexDao> clazz = (Class<? extends IndexDao>) Class.forName(daoImpl);
+ IndexDao instance = clazz.getConstructor().newInstance();
+ instance.init(config);
+ ret.add(instance);
+ }
+ return ret;
+ }
+
+ public static IndexDao combine(Iterable<IndexDao> daos) throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
+ return combine(daos, x -> x);
+ }
+
+ public static IndexDao combine(Iterable<IndexDao> daos, Function<IndexDao, IndexDao> daoTransformation) throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
+ int numDaos = Iterables.size(daos);
+ if(numDaos == 0) {
+ throw new IllegalArgumentException("Trying to combine 0 dao's into a DAO is not a supported configuration.");
+ }
+ if( numDaos == 1) {
+ return daoTransformation.apply(Iterables.getFirst(daos, null));
+ }
+ return new MultiIndexDao(daos, daoTransformation);
}
}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexUpdateCallback.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexUpdateCallback.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexUpdateCallback.java
new file mode 100644
index 0000000..4cb4d7e
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexUpdateCallback.java
@@ -0,0 +1,27 @@
+/**
+ * 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.indexing.dao;
+
+import org.apache.metron.indexing.dao.update.Document;
+
+import java.io.IOException;
+
+public interface IndexUpdateCallback {
+ void postUpdate(IndexDao dao, Document doc) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java
new file mode 100644
index 0000000..e9a4a9a
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java
@@ -0,0 +1,159 @@
+/**
+ * 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.indexing.dao;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Iterables;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.metron.indexing.dao.search.FieldType;
+import org.apache.metron.indexing.dao.search.InvalidSearchException;
+import org.apache.metron.indexing.dao.search.SearchRequest;
+import org.apache.metron.indexing.dao.search.SearchResponse;
+import org.apache.metron.indexing.dao.update.Document;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class MultiIndexDao implements IndexDao {
+ private List<IndexDao> indices;
+
+ public MultiIndexDao( IndexDao... composedDao) {
+ indices = new ArrayList<>();
+ Collections.addAll(indices, composedDao);
+ }
+
+ public MultiIndexDao(Iterable<IndexDao> composedDao) {
+ this.indices = new ArrayList<>();
+ Iterables.addAll(indices, composedDao);
+ }
+
+ public MultiIndexDao(Iterable<IndexDao> composedDao, Function<IndexDao, IndexDao> decoratorTransformation) {
+ this(Iterables.transform(composedDao, x -> decoratorTransformation.apply(x)));
+ }
+
+ @Override
+ public void update(final Document update, Optional<String> index) throws IOException {
+ List<String> exceptions =
+ indices.parallelStream().map(dao -> {
+ try {
+ dao.update(update, index);
+ return null;
+ } catch (Throwable e) {
+ return dao.getClass() + ": " + e.getMessage() + "\n" + ExceptionUtils.getStackTrace(e);
+ }
+ }).filter(e -> e != null).collect(Collectors.toList());
+ if(exceptions.size() > 0) {
+ throw new IOException(Joiner.on("\n").join(exceptions));
+ }
+ }
+
+ @Override
+ public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> in) throws IOException {
+ for(IndexDao dao : indices) {
+ Map<String, Map<String, FieldType>> r = dao.getColumnMetadata(in);
+ if(r != null) {
+ return r;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public Map<String, FieldType> getCommonColumnMetadata(List<String> in) throws IOException {
+ for(IndexDao dao : indices) {
+ Map<String, FieldType> r = dao.getCommonColumnMetadata(in);
+ if(r != null) {
+ return r;
+ }
+ }
+ return null;
+ }
+
+ private static class DocumentContainer {
+ private Optional<Document> d = Optional.empty();
+ private Optional<Throwable> t = Optional.empty();
+ public DocumentContainer(Document d) {
+ this.d = Optional.ofNullable(d);
+ }
+ public DocumentContainer(Throwable t) {
+ this.t = Optional.ofNullable(t);
+ }
+
+ public Optional<Document> getDocument() {
+ return d;
+ }
+ public Optional<Throwable> getException() {
+ return t;
+ }
+
+ }
+
+ @Override
+ public SearchResponse search(SearchRequest searchRequest) throws InvalidSearchException {
+ for(IndexDao dao : indices) {
+ SearchResponse s = dao.search(searchRequest);
+ if(s != null) {
+ return s;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public void init(AccessConfig config) {
+ for(IndexDao dao : indices) {
+ dao.init(config);
+ }
+ }
+
+ @Override
+ public Document getLatest(final String guid, String sensorType) throws IOException {
+ Document ret = null;
+ List<DocumentContainer> output =
+ indices.parallelStream().map(dao -> {
+ try {
+ return new DocumentContainer(dao.getLatest(guid, sensorType));
+ } catch (Throwable e) {
+ return new DocumentContainer(e);
+ }
+ }).collect(Collectors.toList());
+
+ List<String> error = new ArrayList<>();
+ for(DocumentContainer dc : output) {
+ if(dc.getException().isPresent()) {
+ Throwable e = dc.getException().get();
+ error.add(e.getMessage() + "\n" + ExceptionUtils.getStackTrace(e));
+ }
+ else {
+ if(dc.getDocument().isPresent()) {
+ Document d = dc.getDocument().get();
+ if(ret == null || ret.getTimestamp() < d.getTimestamp()) {
+ ret = d;
+ }
+ }
+ }
+ }
+ if(error.size() > 0) {
+ throw new IOException(Joiner.on("\n").join(error));
+ }
+ return ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/GetRequest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/GetRequest.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/GetRequest.java
new file mode 100644
index 0000000..eb255dc
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/GetRequest.java
@@ -0,0 +1,47 @@
+/**
+ * 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.indexing.dao.search;
+
+public class GetRequest {
+ String guid;
+ String sensorType;
+
+ /**
+ * The GUID of the document
+ * @return
+ */
+ public String getGuid() {
+ return guid;
+ }
+
+ public void setGuid(String guid) {
+ this.guid = guid;
+ }
+
+ /**
+ * The sensor type of the indices that you're searching.
+ * @return
+ */
+ public String getSensorType() {
+ return sensorType;
+ }
+
+ public void setSensorType(String sensorType) {
+ this.sensorType = sensorType;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchRequest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchRequest.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchRequest.java
index b92b36d..897f918 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchRequest.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchRequest.java
@@ -39,6 +39,10 @@ public class SearchRequest {
facetFields = new ArrayList<>();
}
+ /**
+ * The list of indices to search.
+ * @return
+ */
public List<String> getIndices() {
return indices;
}
@@ -47,6 +51,10 @@ public class SearchRequest {
this.indices = indices;
}
+ /**
+ * The query to use to search the index
+ * @return
+ */
public String getQuery() {
return query;
}
@@ -55,6 +63,10 @@ public class SearchRequest {
this.query = query;
}
+ /**
+ * The size of the results returned.
+ * @return
+ */
public int getSize() {
return size;
}
@@ -63,6 +75,10 @@ public class SearchRequest {
this.size = size;
}
+ /**
+ * The index to start the search from.
+ * @return
+ */
public int getFrom() {
return from;
}
@@ -71,6 +87,10 @@ public class SearchRequest {
this.from = from;
}
+ /**
+ * The search order by field.
+ * @return
+ */
public List<SortField> getSort() {
return sort;
}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResponse.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResponse.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResponse.java
index 2416357..aad489a 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResponse.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResponse.java
@@ -29,6 +29,10 @@ public class SearchResponse {
private List<SearchResult> results = new ArrayList<>();
private Map<String, Map<String, Long>> facetCounts;
+ /**
+ * The total number of results
+ * @return
+ */
public long getTotal() {
return total;
}
@@ -37,6 +41,10 @@ public class SearchResponse {
this.total = total;
}
+ /**
+ * The list of results
+ * @return
+ */
public List<SearchResult> getResults() {
return results;
}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
index ae4f9bd..9c00bea 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
@@ -24,7 +24,24 @@ public class SearchResult {
private String id;
private Map<String, Object> source;
private float score;
+ private String index;
+ /**
+ * The index that the result comes from
+ * @return
+ */
+ public String getIndex() {
+ return index;
+ }
+
+ public void setIndex(String index) {
+ this.index = index;
+ }
+
+ /**
+ * The ID of the document from the index.
+ * @return
+ */
public String getId() {
return id;
}
@@ -33,6 +50,10 @@ public class SearchResult {
this.id = id;
}
+ /**
+ * The source (the actual result).
+ * @return
+ */
public Map<String, Object> getSource() {
return source;
}
@@ -41,6 +62,10 @@ public class SearchResult {
this.source = source;
}
+ /**
+ * The score from the index.
+ * @return
+ */
public float getScore() {
return score;
}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/Document.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/Document.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/Document.java
new file mode 100644
index 0000000..85c079f
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/Document.java
@@ -0,0 +1,88 @@
+/**
+ * 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.indexing.dao.update;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.metron.common.utils.JSONUtils;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+
+public class Document {
+ Long timestamp;
+ Map<String, Object> document;
+ String guid;
+ String sensorType;
+
+ public Document(Map<String, Object> document, String guid, String sensorType, Long timestamp) {
+ setDocument(document);
+ setGuid(guid);
+ setTimestamp(timestamp);
+ setSensorType(sensorType);
+ }
+
+
+ public Document(String document, String guid, String sensorType, Long timestamp) throws IOException {
+ this(convertDoc(document), guid, sensorType, timestamp);
+ }
+
+ public Document(String document, String guid, String sensorType) throws IOException {
+ this( document, guid, sensorType, null);
+ }
+
+ private static Map<String, Object> convertDoc(String document) throws IOException {
+ return JSONUtils.INSTANCE.load(document, new TypeReference<Map<String, Object>>() {
+ });
+ }
+
+ public String getSensorType() {
+ return sensorType;
+ }
+
+ public void setSensorType(String sensorType) {
+ this.sensorType = sensorType;
+ }
+
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(Long timestamp) {
+ this.timestamp = timestamp != null?timestamp:System.currentTimeMillis();
+ }
+
+ public Map<String, Object> getDocument() {
+ return document;
+ }
+
+ public void setDocument(Map<String, Object> document) {
+ this.document = document;
+ }
+
+ public String getGuid() {
+ return guid;
+ }
+
+ public void setGuid(String guid) {
+ this.guid = guid;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/OriginalNotFoundException.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/OriginalNotFoundException.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/OriginalNotFoundException.java
new file mode 100644
index 0000000..87a03ae
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/OriginalNotFoundException.java
@@ -0,0 +1,28 @@
+/**
+ * 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.indexing.dao.update;
+
+public class OriginalNotFoundException extends Exception {
+ public OriginalNotFoundException(String s) {
+ super(s);
+ }
+
+ public OriginalNotFoundException(String s, Exception e) {
+ super(s, e);
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/PatchRequest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/PatchRequest.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/PatchRequest.java
new file mode 100644
index 0000000..77f5958
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/PatchRequest.java
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.indexing.dao.update;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.Map;
+
+public class PatchRequest {
+ JsonNode patch;
+ Map<String, Object> source;
+ String guid;
+ String sensorType;
+ String index;
+
+ /**
+ * The index of the document to be updated. This is optional, but could result in a performance gain if specified.
+ * @return
+ */
+ public String getIndex() {
+ return index;
+ }
+
+ public void setIndex(String index) {
+ this.index = index;
+ }
+
+ /**
+ * The patch. This is in the form of a list of RFC 6902 patches.
+ * For example:
+ * <pre>
+ * [
+ * {
+ * "op": "add"
+ * , "path": "/project"
+ * , "value": "metron"
+ * }
+ * ]
+ * </pre>
+ * @return
+ */
+ public JsonNode getPatch() {
+ return patch;
+ }
+
+ public void setPatch(JsonNode patch) {
+ this.patch = patch;
+ }
+
+ /**
+ * The source document. If this is specified, then it will be used as the basis of the patch rather than the current
+ * document in the index.
+ * @return
+ */
+ public Map<String, Object> getSource() {
+ return source;
+ }
+
+ public void setSource(Map<String, Object> source) {
+ this.source = source;
+ }
+
+ /**
+ * The GUID of the document to be patched.
+ * @return
+ */
+ public String getGuid() {
+ return guid;
+ }
+
+ public void setGuid(String guid) {
+ this.guid = guid;
+ }
+
+ /**
+ * The sensor type of the document.
+ * @return
+ */
+ public String getSensorType() {
+ return sensorType;
+ }
+
+ public void setSensorType(String sensorType) {
+ this.sensorType = sensorType;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/ReplaceRequest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/ReplaceRequest.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/ReplaceRequest.java
new file mode 100644
index 0000000..96bca30
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/update/ReplaceRequest.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.indexing.dao.update;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.Map;
+
+public class ReplaceRequest {
+ Map<String, Object> replacement;
+ String guid;
+ String sensorType;
+ String index;
+
+ /**
+ * Return the index of the request. This is optional, but could result in better performance if specified.
+ * @return
+ */
+ public String getIndex() {
+ return index;
+ }
+
+ public void setIndex(String index) {
+ this.index = index;
+ }
+
+ /**
+ * The sensor type of the request. This is mandatory.
+ * @return
+ */
+ public String getSensorType() {
+ return sensorType;
+ }
+
+ public void setSensorType(String sensorType) {
+ this.sensorType = sensorType;
+ }
+
+ /**
+ * The replacement document. This is mandatory.
+ * @return
+ */
+ public Map<String, Object> getReplacement() {
+ return replacement;
+ }
+
+ public void setReplacement(Map<String, Object> replacement) {
+ this.replacement = replacement;
+ }
+
+ /**
+ * The GUID of the document to replace. This is mandatory.
+ * @return
+ */
+ public String getGuid() {
+ return guid;
+ }
+
+ public void setGuid(String guid) {
+ this.guid = guid;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/813adf2d/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
index ab83c7e..2d146e0 100644
--- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
+++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
@@ -22,8 +22,10 @@ import com.google.common.base.Splitter;
import com.google.common.collect.ComparisonChain;
import com.google.common.collect.Iterables;
import org.apache.metron.common.Constants;
+import org.apache.metron.common.configuration.writer.WriterConfiguration;
import org.apache.metron.common.utils.JSONUtils;
import org.apache.metron.indexing.dao.search.*;
+import org.apache.metron.indexing.dao.update.Document;
import java.io.IOException;
import java.util.*;
@@ -129,11 +131,41 @@ public class InMemoryDao implements IndexDao {
}
@Override
- public void init(Map<String, Object> globalConfig, AccessConfig config) {
+ public void init(AccessConfig config) {
this.config = config;
}
@Override
+ public Document getLatest(String guid, String sensorType) throws IOException {
+ for(Map.Entry<String, List<String>> kv: BACKING_STORE.entrySet()) {
+ if(kv.getKey().startsWith(sensorType)) {
+ for(String doc : kv.getValue()) {
+ Map<String, Object> docParsed = parse(doc);
+ if(docParsed.getOrDefault(Constants.GUID, "").equals(guid)) {
+ return new Document(doc, guid, sensorType, 0L);
+ }
+ }
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public void update(Document update, Optional<String> index) throws IOException {
+ for (Map.Entry<String, List<String>> kv : BACKING_STORE.entrySet()) {
+ if (kv.getKey().startsWith(update.getSensorType())) {
+ for (Iterator<String> it = kv.getValue().iterator(); it.hasNext(); ) {
+ String doc = it.next();
+ Map<String, Object> docParsed = parse(doc);
+ if (docParsed.getOrDefault(Constants.GUID, "").equals(update.getGuid())) {
+ it.remove();
+ }
+ }
+ kv.getValue().add(JSONUtils.INSTANCE.toJSON(update.getDocument(), true));
+ }
+ }
+ }
+
public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> indices) throws IOException {
Map<String, Map<String, FieldType>> columnMetadata = new HashMap<>();
for(String index: indices) {