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 2016/04/07 14:56:39 UTC
[6/6] incubator-metron git commit: METRON-93: Generalize the HBase
threat intel infrastructure to support enrichments closes
apache/incubator-metron#64
METRON-93: Generalize the HBase threat intel infrastructure to support enrichments closes apache/incubator-metron#64
Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/9621c55e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/9621c55e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/9621c55e
Branch: refs/heads/master
Commit: 9621c55ec61dc3f013ddbdff7135442c25033321
Parents: e59b1a3
Author: cestella <ce...@gmail.com>
Authored: Thu Apr 7 08:56:23 2016 -0400
Committer: cstella <ce...@gmail.com>
Committed: Thu Apr 7 08:56:23 2016 -0400
----------------------------------------------------------------------
LICENSE | 5 +
deployment/amazon-ec2/conf/defaults.yml | 7 +-
.../inventory/metron_example/group_vars/all | 3 +-
.../inventory/multinode-vagrant/group_vars/all | 3 +-
.../inventory/singlenode-vagrant/group_vars/all | 3 +-
deployment/roles/hadoop_setup/tasks/main.yml | 3 +-
.../files/config/sensors/bro.json | 7 +-
.../files/config/sensors/pcap.json | 7 +-
.../files/config/sensors/snort.json | 10 +-
.../files/config/sensors/yaf.json | 7 +-
.../roles/metron_streaming/files/extractor.json | 1 +
.../roles/metron_streaming/tasks/main.yml | 12 +-
.../metron_streaming/tasks/threat_intel.yml | 2 +-
.../main/java/org/apache/metron/Constants.java | 3 +-
.../metron/bolt/BulkMessageWriterBolt.java | 3 +-
.../org/apache/metron/bolt/ConfiguredBolt.java | 12 +
.../java/org/apache/metron/bolt/JoinBolt.java | 6 +
.../metron/domain/SensorEnrichmentConfig.java | 60 +++
.../metron/enrichment/EnrichmentConfig.java | 203 ++++++++++
.../hbase/converters/AbstractConverter.java | 95 ++---
.../enrichment/EnrichmentConverter.java | 38 ++
.../converters/enrichment/EnrichmentKey.java | 119 ++++++
.../converters/enrichment/EnrichmentValue.java | 107 +++++
.../threatintel/ThreatIntelConverter.java | 39 --
.../converters/threatintel/ThreatIntelKey.java | 89 ----
.../threatintel/ThreatIntelValue.java | 110 -----
.../metron/hbase/lookup/EnrichmentLookup.java | 108 +++++
.../apache/metron/reference/lookup/Lookup.java | 93 +++--
.../accesstracker/BloomAccessTracker.java | 1 +
.../lookup/accesstracker/NoopAccessTracker.java | 65 +++
.../reference/lookup/handler/Handler.java | 6 +-
.../metron/threatintel/ThreatIntelResults.java | 71 ----
.../threatintel/hbase/ThreatIntelLookup.java | 74 ----
.../metron/utils/ConfigurationsUtils.java | 41 +-
.../java/org/apache/metron/utils/JSONUtils.java | 10 +-
.../metron/enrichment/EnrichmentConfigTest.java | 197 +++++++++
.../enrichment/EnrichmentConverterTest.java | 34 ++
metron-streaming/Metron-DataLoads/README.md | 254 ++++++++++--
.../src/main/bash/flatfile_loader.sh | 39 ++
.../src/main/bash/threatintel_taxii_load.sh | 2 +-
.../metron/dataloads/ThreatIntelLoader.java | 195 ---------
.../dataloads/bulk/LeastRecentlyUsedPruner.java | 2 +-
.../dataloads/bulk/ThreatIntelBulkLoader.java | 371 +++++++++--------
.../metron/dataloads/extractor/Extractor.java | 4 +-
.../dataloads/extractor/ExtractorHandler.java | 12 +-
.../dataloads/extractor/csv/CSVExtractor.java | 174 ++++----
.../extractor/csv/LookupConverter.java | 2 +-
.../extractor/csv/LookupConverters.java | 13 +-
.../dataloads/extractor/stix/StixExtractor.java | 2 -
.../stix/types/AbstractObjectTypeHandler.java | 4 +-
.../extractor/stix/types/AddressHandler.java | 101 +++--
.../extractor/stix/types/DomainHandler.java | 72 ++--
.../extractor/stix/types/HostnameHandler.java | 57 ++-
.../extractor/stix/types/ObjectTypeHandler.java | 8 +-
.../dataloads/hbase/mr/BulkLoadMapper.java | 4 -
.../metron/dataloads/hbase/mr/PrunerMapper.java | 2 +-
.../SimpleEnrichmentFlatFileLoader.java | 263 ++++++++++++
.../dataloads/nonbulk/taxii/ConnectionType.java | 23 ++
.../dataloads/nonbulk/taxii/TableInfo.java | 71 ++++
.../nonbulk/taxii/TaxiiConnectionConfig.java | 222 ++++++++++
.../dataloads/nonbulk/taxii/TaxiiHandler.java | 406 +++++++++++++++++++
.../dataloads/nonbulk/taxii/TaxiiLoader.java | 208 ++++++++++
.../metron/dataloads/taxii/ConnectionType.java | 23 --
.../metron/dataloads/taxii/TableInfo.java | 72 ----
.../dataloads/taxii/TaxiiConnectionConfig.java | 196 ---------
.../metron/dataloads/taxii/TaxiiHandler.java | 403 ------------------
.../metron/dataloads/taxii/TaxiiLoader.java | 180 --------
.../dataloads/extractor/ExtractorTest.java | 20 +-
.../extractor/csv/CSVExtractorTest.java | 130 ++----
.../extractor/stix/StixExtractorTest.java | 152 +++----
.../hbase/HBaseEnrichmentConverterTest.java | 74 ++++
.../hbase/HBaseThreatIntelConverterTest.java | 76 ----
.../hbase/mr/BulkLoadMapperIntegrationTest.java | 32 +-
.../dataloads/hbase/mr/BulkLoadMapperTest.java | 34 +-
.../LeastRecentlyUsedPrunerIntegrationTest.java | 34 +-
.../nonbulk/taxii/MockTaxiiService.java | 94 +++++
.../nonbulk/taxii/TaxiiIntegrationTest.java | 119 ++++++
.../dataloads/taxii/MockTaxiiService.java | 99 -----
.../dataloads/taxii/TaxiiIntegrationTest.java | 121 ------
metron-streaming/Metron-Elasticsearch/pom.xml | 5 +
.../metron/writer/ElasticsearchWriter.java | 2 +-
.../ElasticsearchEnrichmentIntegrationTest.java | 2 +-
.../Metron-EnrichmentAdapters/pom.xml | 3 +-
.../adapters/cif/AbstractCIFAdapter.java | 3 +-
.../adapters/cif/CIFHbaseAdapter.java | 9 +-
.../enrichment/adapters/geo/GeoAdapter.java | 12 +-
.../adapters/host/AbstractHostAdapter.java | 5 +-
.../adapters/host/HostFromJSONListAdapter.java | 7 +-
.../host/HostFromPropertiesFileAdapter.java | 9 +-
.../enrichment/adapters/jdbc/JdbcAdapter.java | 3 +-
.../simplehbase/SimpleHBaseAdapter.java | 119 ++++++
.../adapters/simplehbase/SimpleHBaseConfig.java | 55 +++
.../adapters/threat/AbstractThreatAdapter.java | 41 --
.../adapters/threat/ThreatHbaseAdapter.java | 140 -------
.../threatintel/ThreatIntelAdapter.java | 135 ++++++
.../adapters/threatintel/ThreatIntelConfig.java | 108 +++++
.../adapters/whois/WhoisHBaseAdapter.java | 11 +-
.../apache/metron/enrichment/bolt/CacheKey.java | 73 ++++
.../enrichment/bolt/EnrichmentJoinBolt.java | 24 +-
.../enrichment/bolt/GenericEnrichmentBolt.java | 44 +-
.../enrichment/bolt/ThreatIntelJoinBolt.java | 12 +-
.../enrichment/utils/EnrichmentUtils.java | 49 +++
.../metron/threatintel/ThreatIntelAdapter.java | 108 -----
.../metron/threatintel/ThreatIntelConfig.java | 125 ------
.../adapters/cif/CIFHbaseAdapterTest.java | 9 +-
.../adapters/geo/GeoMysqlAdapterTest.java | 3 +-
.../adapters/whois/WhoisHBaseAdapterTest.java | 5 +-
metron-streaming/Metron-MessageParsers/pom.xml | 14 +-
metron-streaming/Metron-Solr/pom.xml | 5 +
.../integration/EnrichmentIntegrationTest.java | 95 +++--
.../integration/util/EnrichmentHelper.java | 39 ++
.../integration/util/mock/MockGeoAdapter.java | 7 +-
.../util/threatintel/ThreatIntelHelper.java | 39 --
.../resources/sample/config/sensors/bro.json | 7 +-
.../resources/sample/config/sensors/snort.json | 7 +-
.../resources/sample/config/sensors/yaf.json | 15 +-
metron-streaming/Metron-Topologies/pom.xml | 15 +
.../src/main/bash/zk_load_configs.sh | 2 +-
.../topologies/enrichment/remote.yaml | 99 ++++-
.../topologies/enrichment/test.yaml | 98 ++++-
metron-streaming/pom.xml | 27 +-
121 files changed, 4512 insertions(+), 3102 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/LICENSE
----------------------------------------------------------------------
diff --git a/LICENSE b/LICENSE
index 4628dd1..da4e5a0 100644
--- a/LICENSE
+++ b/LICENSE
@@ -768,6 +768,11 @@ The above copyright notice and this permission notice shall be included in all c
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-----------------------------------------------------------------------
+This bundles the following released to the Public Domain:
+
+- Java Multiline String (https://github.com/benelog/multiline)
+
+-----------------------------------------------------------------------
This product bundles normalize.css which is available under a "MIT" license. For details, see metron-ui/lib/public/vendor/LICENSE.normalize
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/amazon-ec2/conf/defaults.yml
----------------------------------------------------------------------
diff --git a/deployment/amazon-ec2/conf/defaults.yml b/deployment/amazon-ec2/conf/defaults.yml
index 77e841d..5f581f6 100644
--- a/deployment/amazon-ec2/conf/defaults.yml
+++ b/deployment/amazon-ec2/conf/defaults.yml
@@ -36,7 +36,8 @@ cluster_type: small_cluster
# hbase
pcap_hbase_table: pcap
tracker_hbase_table: access_tracker
-threatintel_ip_hbase_table: malicious_ip
+threatintel_hbase_table: threatintel
+enrichment_hbase_table: enrichment
# kafka
num_partitions: 3
@@ -74,5 +75,5 @@ elasticsearch_data_dir: "/data1/elasticsearch,/data2/elasticsearch"
install_elasticsearch: True
install_solr: False
elasticsearch_transport_port: 9300
-elasticsearch_network_interface: eth1
-elasticsearch_web_port: 9200
\ No newline at end of file
+elasticsearch_network_interface: eth0
+elasticsearch_web_port: 9200
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/inventory/metron_example/group_vars/all
----------------------------------------------------------------------
diff --git a/deployment/inventory/metron_example/group_vars/all b/deployment/inventory/metron_example/group_vars/all
index c5d81b0..4eba586 100644
--- a/deployment/inventory/metron_example/group_vars/all
+++ b/deployment/inventory/metron_example/group_vars/all
@@ -29,7 +29,8 @@ cluster_type: small_cluster
# hbase
pcap_hbase_table: pcap
tracker_hbase_table: access_tracker
-threatintel_ip_hbase_table: malicious_ip
+threatintel_hbase_table: threatintel
+enrichment_hbase_table: enrichment
# metron variables
metron_version: 0.1BETA
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/inventory/multinode-vagrant/group_vars/all
----------------------------------------------------------------------
diff --git a/deployment/inventory/multinode-vagrant/group_vars/all b/deployment/inventory/multinode-vagrant/group_vars/all
index e486296..8f4e5ff 100644
--- a/deployment/inventory/multinode-vagrant/group_vars/all
+++ b/deployment/inventory/multinode-vagrant/group_vars/all
@@ -26,7 +26,8 @@ cluster_type: multi_vagrant_cluster
# hbase
pcap_hbase_table: pcap
tracker_hbase_table: access_tracker
-threatintel_ip_hbase_table: malicious_ip
+threatintel_hbase_table: threatintel
+enrichment_hbase_table: enrichment
#elasticsearch
elasticsearch_transport_port: 9300
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/inventory/singlenode-vagrant/group_vars/all
----------------------------------------------------------------------
diff --git a/deployment/inventory/singlenode-vagrant/group_vars/all b/deployment/inventory/singlenode-vagrant/group_vars/all
index 364598f..6d96d5d 100644
--- a/deployment/inventory/singlenode-vagrant/group_vars/all
+++ b/deployment/inventory/singlenode-vagrant/group_vars/all
@@ -26,7 +26,8 @@ cluster_type: single_node_vm
# hbase
pcap_hbase_table: pcap
tracker_hbase_table: access_tracker
-threatintel_ip_hbase_table: malicious_ip
+threatintel_hbase_table: threatintel
+enrichment_hbase_table: enrichment
# metron variables
metron_version: 0.1BETA
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/hadoop_setup/tasks/main.yml
----------------------------------------------------------------------
diff --git a/deployment/roles/hadoop_setup/tasks/main.yml b/deployment/roles/hadoop_setup/tasks/main.yml
index 5b6c47c..de01abf 100644
--- a/deployment/roles/hadoop_setup/tasks/main.yml
+++ b/deployment/roles/hadoop_setup/tasks/main.yml
@@ -22,7 +22,8 @@
with_items:
- "{{ pcap_hbase_table }}"
- "{{ tracker_hbase_table }}"
- - "{{ threatintel_ip_hbase_table }}"
+ - "{{ threatintel_hbase_table }}"
+ - "{{ enrichment_hbase_table }}"
#if kafka topic
- name: Create Kafka topics
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/metron_streaming/files/config/sensors/bro.json
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/files/config/sensors/bro.json b/deployment/roles/metron_streaming/files/config/sensors/bro.json
index 34109b8..2b534b4 100644
--- a/deployment/roles/metron_streaming/files/config/sensors/bro.json
+++ b/deployment/roles/metron_streaming/files/config/sensors/bro.json
@@ -8,7 +8,12 @@
},
"threatIntelFieldMap":
{
- "ip": ["ip_dst_addr", "ip_src_addr"]
+ "hbaseThreatIntel": ["ip_src_addr", "ip_dst_addr"]
+ },
+ "fieldToThreatIntelTypeMap":
+ {
+ "ip_src_addr" : ["malicious_ip"],
+ "ip_dst_addr" : ["malicious_ip"]
}
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/metron_streaming/files/config/sensors/pcap.json
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/files/config/sensors/pcap.json b/deployment/roles/metron_streaming/files/config/sensors/pcap.json
index 4b9c639..7792165 100644
--- a/deployment/roles/metron_streaming/files/config/sensors/pcap.json
+++ b/deployment/roles/metron_streaming/files/config/sensors/pcap.json
@@ -8,7 +8,12 @@
},
"threatIntelFieldMap":
{
- "ip": ["ip_src_addr", "ip_dst_addr"]
+ "hbaseThreatIntel": ["ip_dst_addr", "ip_src_addr"]
+ },
+ "fieldToThreatIntelTypeMap":
+ {
+ "ip_dst_addr" : [ "malicious_ip" ]
+ ,"ip_src_addr" : [ "malicious_ip" ]
}
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/metron_streaming/files/config/sensors/snort.json
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/files/config/sensors/snort.json b/deployment/roles/metron_streaming/files/config/sensors/snort.json
index 1208637..c5b6dcc 100644
--- a/deployment/roles/metron_streaming/files/config/sensors/snort.json
+++ b/deployment/roles/metron_streaming/files/config/sensors/snort.json
@@ -6,9 +6,13 @@
"geo": ["ip_dst_addr", "ip_src_addr"],
"host": ["host"]
},
- "threatIntelFieldMap":
+ "threatIntelFieldMap":
{
- "ip": ["ip_dst_addr", "ip_src_addr"]
+ "hbaseThreatIntel": ["ip_src_addr", "ip_dst_addr"]
+ },
+ "fieldToThreatIntelTypeMap":
+ {
+ "ip_src_addr" : ["malicious_ip"],
+ "ip_dst_addr" : ["malicious_ip"]
}
}
-
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/metron_streaming/files/config/sensors/yaf.json
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/files/config/sensors/yaf.json b/deployment/roles/metron_streaming/files/config/sensors/yaf.json
index 65de961..2b46c9a 100644
--- a/deployment/roles/metron_streaming/files/config/sensors/yaf.json
+++ b/deployment/roles/metron_streaming/files/config/sensors/yaf.json
@@ -8,7 +8,12 @@
},
"threatIntelFieldMap":
{
- "ip": ["ip_dst_addr", "ip_src_addr"]
+ "hbaseThreatIntel": ["ip_src_addr", "ip_dst_addr"]
+ },
+ "fieldToThreatIntelTypeMap":
+ {
+ "ip_src_addr" : ["malicious_ip"],
+ "ip_dst_addr" : ["malicious_ip"]
}
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/metron_streaming/files/extractor.json
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/files/extractor.json b/deployment/roles/metron_streaming/files/extractor.json
index 81429e8..545202a 100644
--- a/deployment/roles/metron_streaming/files/extractor.json
+++ b/deployment/roles/metron_streaming/files/extractor.json
@@ -4,6 +4,7 @@
"ip": 0
},
"indicator_column": "ip",
+ "type" : "malicious_ip",
"separator": ","
},
"extractor": "CSV"
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/metron_streaming/tasks/main.yml
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/tasks/main.yml b/deployment/roles/metron_streaming/tasks/main.yml
index 863ecd8..b913011 100644
--- a/deployment/roles/metron_streaming/tasks/main.yml
+++ b/deployment/roles/metron_streaming/tasks/main.yml
@@ -83,8 +83,10 @@
- { regexp: "bolt.hbase.table.name=", line: "bolt.hbase.table.name={{ pcap_hbase_table }}" }
- { regexp: "threat.intel.tracker.table=", line: "threat.intel.tracker.table={{ tracker_hbase_table }}" }
- { regexp: "threat.intel.tracker.cf=", line: "threat.intel.tracker.cf=t" }
- - { regexp: "threat.intel.ip.table=", line: "threat.intel.ip.table={{ threatintel_ip_hbase_table }}" }
- - { regexp: "threat.intel.ip.cf=", line: "threat.intel.ip.cf=t" }
+ - { regexp: "threat.intel.simple.hbase.table=", line: "threat.intel.simple.hbase.table={{ threatintel_hbase_table }}" }
+ - { regexp: "threat.intel.simple.hbase.cf=", line: "threat.intel.simple.hbase.cf=t" }
+ - { regexp: "enrichment.simple.hbase.table=", line: "enrichment.simple.hbase.table={{ enrichment_hbase_table }}" }
+ - { regexp: "enrichment.simple.hbase.cf=", line: "enrichment.simple.hbase.cf=t" }
- { regexp: "mysql.ip=", line: "mysql.ip={{ groups.mysql[0] }}" }
- { regexp: "mysql.password=", line: "mysql.password={{ mysql_root_password }}" }
- { regexp: "index.hdfs.output=", line: "index.hdfs.output={{ metron_hdfs_output_dir }}/enrichment/indexed" }
@@ -109,8 +111,10 @@
- { regexp: "bolt.hbase.table.name=", line: "bolt.hbase.table.name={{ pcap_hbase_table }}" }
- { regexp: "threat.intel.tracker.table=", line: "threat.intel.tracker.table={{ tracker_hbase_table }}" }
- { regexp: "threat.intel.tracker.cf=", line: "threat.intel.tracker.cf=t" }
- - { regexp: "threat.intel.ip.table=", line: "threat.intel.ip.table={{ threatintel_ip_hbase_table }}" }
- - { regexp: "threat.intel.ip.cf=", line: "threat.intel.ip.cf=t" }
+ - { regexp: "threat.intel.simple.hbase.table=", line: "threat.intel.simple.hbase.table={{ threatintel_hbase_table }}" }
+ - { regexp: "threat.intel.simple.hbase.cf=", line: "threat.intel.simple.hbase.cf=t" }
+ - { regexp: "enrichment.simple.hbase.table=", line: "enrichment.simple.hbase.table={{ enrichment_hbase_table }}" }
+ - { regexp: "enrichment.simple.hbase.cf=", line: "enrichment.simple.hbase.cf=t" }
- { regexp: "mysql.ip=", line: "mysql.ip={{ groups.mysql[0] }}" }
- { regexp: "mysql.password=", line: "mysql.password={{ mysql_root_password }}" }
- { regexp: "index.hdfs.output=", line: "index.hdfs.output={{ metron_hdfs_output_dir }}/enrichment/indexed" }
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/deployment/roles/metron_streaming/tasks/threat_intel.yml
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/tasks/threat_intel.yml b/deployment/roles/metron_streaming/tasks/threat_intel.yml
index 6798319..6dd8144 100644
--- a/deployment/roles/metron_streaming/tasks/threat_intel.yml
+++ b/deployment/roles/metron_streaming/tasks/threat_intel.yml
@@ -37,7 +37,7 @@
command: "hdfs dfs -put {{ threat_intel_work_dir }}/{{ threat_intel_csv_filename }} ."
- name: Run Threat Intel Bulk Load
- shell: "{{ threat_intel_bin }} -f t --table malicious_ip -e {{ threat_intel_work_dir }}/extractor.json -i /user/root && touch {{ threat_intel_work_dir }}/loaded"
+ shell: "{{ threat_intel_bin }} -f t --table {{threatintel_hbase_table}} -e {{ threat_intel_work_dir }}/extractor.json -i /user/root && touch {{ threat_intel_work_dir }}/loaded"
args:
creates: "{{ threat_intel_work_dir }}/loaded"
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/Constants.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/Constants.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/Constants.java
index 78efa5c..32f1441 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/Constants.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/Constants.java
@@ -29,6 +29,7 @@ public class Constants {
public static final String SENSOR_TYPE = "source.type";
public static final String ENRICHMENT_TOPIC = "enrichments";
public static final String ERROR_STREAM = "error";
-
+ public static final String SIMPLE_HBASE_ENRICHMENT = "hbaseEnrichment";
+ public static final String SIMPLE_HBASE_THREAT_INTEL = "hbaseThreatIntel";
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/BulkMessageWriterBolt.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/BulkMessageWriterBolt.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/BulkMessageWriterBolt.java
index 1f12f7a..f1f79f0 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/BulkMessageWriterBolt.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/BulkMessageWriterBolt.java
@@ -81,8 +81,7 @@ public class BulkMessageWriterBolt extends ConfiguredBolt {
sensorMessageMap.put(sensorType, messageList);
} else {
try {
- String esType = sensorType + "_doc";
- bulkMessageWriter.write(esType, configurations, tupleList, messageList);
+ bulkMessageWriter.write(sensorType, configurations, tupleList, messageList);
for(Tuple t: tupleList) {
collector.ack(t);
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/ConfiguredBolt.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/ConfiguredBolt.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/ConfiguredBolt.java
index a832ebb..3d22fe9 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/ConfiguredBolt.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/ConfiguredBolt.java
@@ -50,6 +50,9 @@ public abstract class ConfiguredBolt extends BaseRichBolt {
this.zookeeperUrl = zookeeperUrl;
}
+ protected void reloadCallback() {
+ }
+
public ConfiguredBolt withTimeout(long timeout) {
this.timeout = timeout;
return this;
@@ -70,12 +73,21 @@ public abstract class ConfiguredBolt extends BaseRichBolt {
String path = event.getData().getPath();
byte[] data = event.getData().getData();
updateConfig(path, data);
+ reloadCallback();
}
}
};
cache.getListenable().addListener(listener);
+ try {
+ ConfigurationsUtils.updateConfigsFromZookeeper(configurations, client);
+ }
+ catch(Exception e) {
+ LOG.warn("Unable to load configs from zookeeper, but the cache should load lazily...");
+ }
+
cache.start();
} catch (Exception e) {
+ LOG.error(e.getMessage(), e);
throw new RuntimeException(e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/JoinBolt.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/JoinBolt.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/JoinBolt.java
index 653eade..9c2c67e 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/JoinBolt.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/bolt/JoinBolt.java
@@ -23,6 +23,7 @@ import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
+import com.google.common.base.Joiner;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
@@ -104,6 +105,11 @@ public abstract class JoinBolt<V> extends ConfiguredBolt {
cache.invalidate(key);
} else {
cache.put(key, streamMessageMap);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(getClass().getSimpleName() + ": Missed joining portions for "+ key + ". Expected " + Joiner.on(",").join(streamIds)
+ + " != " + Joiner.on(",").join(streamMessageKeys)
+ );
+ }
}
} catch (ExecutionException e) {
collector.reportError(e);
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/domain/SensorEnrichmentConfig.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/domain/SensorEnrichmentConfig.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/domain/SensorEnrichmentConfig.java
index b24e8a8..e37c6d0 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/domain/SensorEnrichmentConfig.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/domain/SensorEnrichmentConfig.java
@@ -17,6 +17,11 @@
*/
package org.apache.metron.domain;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.metron.utils.JSONUtils;
+
+import java.io.IOException;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -25,6 +30,8 @@ public class SensorEnrichmentConfig {
private String index;
private Map<String, List<String>> enrichmentFieldMap;
private Map<String, List<String>> threatIntelFieldMap;
+ private Map<String, List<String>> fieldToEnrichmentTypeMap = new HashMap<>();
+ private Map<String, List<String>> fieldToThreatIntelTypeMap = new HashMap<>();
private int batchSize;
public String getIndex() {
@@ -51,6 +58,20 @@ public class SensorEnrichmentConfig {
this.threatIntelFieldMap = threatIntelFieldMap;
}
+ public Map<String, List<String>> getFieldToEnrichmentTypeMap() {
+ return fieldToEnrichmentTypeMap;
+ }
+
+ public Map<String, List<String>> getFieldToThreatIntelTypeMap() {
+ return fieldToThreatIntelTypeMap;
+ }
+ public void setFieldToEnrichmentTypeMap(Map<String, List<String>> fieldToEnrichmentTypeMap) {
+ this.fieldToEnrichmentTypeMap = fieldToEnrichmentTypeMap;
+ }
+
+ public void setFieldToThreatIntelTypeMap(Map<String, List<String>> fieldToThreatIntelTypeMap) {
+ this.fieldToThreatIntelTypeMap= fieldToThreatIntelTypeMap;
+ }
public int getBatchSize() {
return batchSize;
}
@@ -59,4 +80,43 @@ public class SensorEnrichmentConfig {
this.batchSize = batchSize;
}
+ public static SensorEnrichmentConfig fromBytes(byte[] config) throws IOException {
+ return JSONUtils.INSTANCE.load(new String(config), SensorEnrichmentConfig.class);
+ }
+ public String toJSON(boolean pretty) throws JsonProcessingException {
+ return JSONUtils.INSTANCE.toJSON(this, pretty);
+ }
+ public String toJSON() throws JsonProcessingException {
+ return JSONUtils.INSTANCE.toJSON(this, true);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ SensorEnrichmentConfig that = (SensorEnrichmentConfig) o;
+
+ if (getBatchSize() != that.getBatchSize()) return false;
+ if (getIndex() != null ? !getIndex().equals(that.getIndex()) : that.getIndex() != null) return false;
+ if (getEnrichmentFieldMap() != null ? !getEnrichmentFieldMap().equals(that.getEnrichmentFieldMap()) : that.getEnrichmentFieldMap() != null)
+ return false;
+ if (getThreatIntelFieldMap() != null ? !getThreatIntelFieldMap().equals(that.getThreatIntelFieldMap()) : that.getThreatIntelFieldMap() != null)
+ return false;
+ if (getFieldToEnrichmentTypeMap() != null ? !getFieldToEnrichmentTypeMap().equals(that.getFieldToEnrichmentTypeMap()) : that.getFieldToEnrichmentTypeMap() != null)
+ return false;
+ return getFieldToThreatIntelTypeMap() != null ? getFieldToThreatIntelTypeMap().equals(that.getFieldToThreatIntelTypeMap()) : that.getFieldToThreatIntelTypeMap() == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = getIndex() != null ? getIndex().hashCode() : 0;
+ result = 31 * result + (getEnrichmentFieldMap() != null ? getEnrichmentFieldMap().hashCode() : 0);
+ result = 31 * result + (getThreatIntelFieldMap() != null ? getThreatIntelFieldMap().hashCode() : 0);
+ result = 31 * result + (getFieldToEnrichmentTypeMap() != null ? getFieldToEnrichmentTypeMap().hashCode() : 0);
+ result = 31 * result + (getFieldToThreatIntelTypeMap() != null ? getFieldToThreatIntelTypeMap().hashCode() : 0);
+ result = 31 * result + getBatchSize();
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/enrichment/EnrichmentConfig.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/enrichment/EnrichmentConfig.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/enrichment/EnrichmentConfig.java
new file mode 100644
index 0000000..92913d9
--- /dev/null
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/enrichment/EnrichmentConfig.java
@@ -0,0 +1,203 @@
+/**
+ * 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;
+
+import com.google.common.base.Joiner;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.metron.Constants;
+import org.apache.metron.domain.SensorEnrichmentConfig;
+import org.apache.metron.utils.ConfigurationsUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.*;
+
+public class EnrichmentConfig {
+ public static enum Type {
+ THREAT_INTEL
+ ,ENRICHMENT
+ }
+
+ protected static final Logger _LOG = LoggerFactory.getLogger(EnrichmentConfig.class);
+ public static class FieldList {
+ Type type;
+ Map<String, List<String>> fieldToEnrichmentTypes;
+
+ public Type getType() {
+ return type;
+ }
+
+ public void setType(Type type) {
+ this.type = type;
+ }
+
+ public Map<String, List<String>> getFieldToEnrichmentTypes() {
+ return fieldToEnrichmentTypes;
+ }
+
+ public void setFieldToEnrichmentTypes(Map<String, List<String>> fieldToEnrichmentTypes) {
+ this.fieldToEnrichmentTypes = fieldToEnrichmentTypes;
+ }
+ }
+ public String zkQuorum;
+ public Map<String, FieldList> sensorToFieldList;
+
+ public String getZkQuorum() {
+ return zkQuorum;
+ }
+
+ public void setZkQuorum(String zkQuorum) {
+ this.zkQuorum = zkQuorum;
+ }
+
+ public Map<String, FieldList> getSensorToFieldList() {
+ return sensorToFieldList;
+ }
+
+ public void setSensorToFieldList(Map<String, FieldList> sensorToFieldList) {
+ this.sensorToFieldList = sensorToFieldList;
+ }
+
+ public void updateSensorConfigs( ) throws Exception {
+ CuratorFramework client = ConfigurationsUtils.getClient(getZkQuorum());
+ try {
+ client.start();
+ updateSensorConfigs(new ZKSourceConfigHandler(client), sensorToFieldList);
+ }
+ finally {
+ client.close();
+ }
+ }
+
+ public static interface SourceConfigHandler {
+ SensorEnrichmentConfig readConfig(String sensor) throws Exception;
+ void persistConfig(String sensor, SensorEnrichmentConfig config) throws Exception;
+ }
+
+ public static class ZKSourceConfigHandler implements SourceConfigHandler {
+ CuratorFramework client;
+ public ZKSourceConfigHandler(CuratorFramework client) {
+ this.client = client;
+ }
+ @Override
+ public SensorEnrichmentConfig readConfig(String sensor) throws Exception {
+ return SensorEnrichmentConfig.fromBytes(ConfigurationsUtils.readSensorEnrichmentConfigBytesFromZookeeper(sensor, client));
+ }
+
+ @Override
+ public void persistConfig(String sensor, SensorEnrichmentConfig config) throws Exception {
+ ConfigurationsUtils.writeSensorEnrichmentConfigToZookeeper(sensor, config.toJSON().getBytes(), client);
+ }
+ }
+
+ public static void updateSensorConfigs( SourceConfigHandler scHandler
+ , Map<String, FieldList> sensorToFieldList
+ ) throws Exception
+ {
+ Map<String, SensorEnrichmentConfig> sourceConfigsChanged = new HashMap<>();
+ for (Map.Entry<String, FieldList> kv : sensorToFieldList.entrySet()) {
+ SensorEnrichmentConfig config = sourceConfigsChanged.get(kv.getKey());
+ if(config == null) {
+ config = scHandler.readConfig(kv.getKey());
+ if(_LOG.isDebugEnabled()) {
+ _LOG.debug(config.toJSON());
+ }
+ }
+ Map<String, List<String> > fieldMap = null;
+ Map<String, List<String> > fieldToTypeMap = null;
+ List<String> fieldList = null;
+ if(kv.getValue().type == Type.THREAT_INTEL) {
+ fieldMap = config.getThreatIntelFieldMap();
+ if(fieldMap!= null) {
+ fieldList = fieldMap.get(Constants.SIMPLE_HBASE_THREAT_INTEL);
+ }
+ if(fieldList == null) {
+ fieldList = new ArrayList<>();
+ fieldMap.put(Constants.SIMPLE_HBASE_THREAT_INTEL, fieldList);
+ }
+ fieldToTypeMap = config.getFieldToThreatIntelTypeMap();
+ if(fieldToTypeMap == null) {
+ fieldToTypeMap = new HashMap<>();
+ config.setFieldToThreatIntelTypeMap(fieldToTypeMap);
+ }
+ }
+ else if(kv.getValue().type == Type.ENRICHMENT) {
+ fieldMap = config.getEnrichmentFieldMap();
+ if(fieldMap!= null) {
+ fieldList = fieldMap.get(Constants.SIMPLE_HBASE_ENRICHMENT);
+ }
+ if(fieldList == null) {
+ fieldList = new ArrayList<>();
+ fieldMap.put(Constants.SIMPLE_HBASE_ENRICHMENT, fieldList);
+ }
+ fieldToTypeMap = config.getFieldToEnrichmentTypeMap();
+ if(fieldToTypeMap == null) {
+ fieldToTypeMap = new HashMap<>();
+ config.setFieldToEnrichmentTypeMap(fieldToTypeMap);
+ }
+ }
+ if(fieldToTypeMap == null || fieldMap == null) {
+ _LOG.debug("fieldToTypeMap is null or fieldMap is null, so skipping");
+ continue;
+ }
+ //Add the additional fields to the field list associated with the hbase adapter
+ {
+ HashSet<String> fieldSet = new HashSet<>(fieldList);
+ List<String> additionalFields = new ArrayList<>();
+ for (String field : kv.getValue().getFieldToEnrichmentTypes().keySet()) {
+ if (!fieldSet.contains(field)) {
+ additionalFields.add(field);
+ }
+ }
+ //adding only the ones that we don't already have to the field list
+ if (additionalFields.size() > 0) {
+ _LOG.debug("Adding additional fields: " + Joiner.on(',').join(additionalFields));
+ fieldList.addAll(additionalFields);
+ sourceConfigsChanged.put(kv.getKey(), config);
+ }
+ }
+ //Add the additional enrichment types to the mapping between the fields
+ {
+ for(Map.Entry<String, List<String>> fieldToType : kv.getValue().getFieldToEnrichmentTypes().entrySet()) {
+ String field = fieldToType.getKey();
+ final HashSet<String> types = new HashSet<>(fieldToType.getValue());
+ int sizeBefore = 0;
+ if(fieldToTypeMap.containsKey(field)) {
+ List<String> typeList = fieldToTypeMap.get(field);
+ sizeBefore = new HashSet<>(typeList).size();
+ types.addAll(typeList);
+ }
+ int sizeAfter = types.size();
+ boolean changed = sizeBefore != sizeAfter;
+ if(changed) {
+ fieldToTypeMap.put(field, new ArrayList<String>() {{
+ addAll(types);
+ }});
+ sourceConfigsChanged.put(kv.getKey(), config);
+ }
+ }
+ }
+ }
+ for(Map.Entry<String, SensorEnrichmentConfig> kv : sourceConfigsChanged.entrySet()) {
+ scHandler.persistConfig(kv.getKey(), kv.getValue());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/AbstractConverter.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/AbstractConverter.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/AbstractConverter.java
index 9072c22..c58dc22 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/AbstractConverter.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/AbstractConverter.java
@@ -35,59 +35,62 @@ import java.util.*;
public abstract class AbstractConverter<KEY_T extends LookupKey, VALUE_T extends LookupValue> implements HbaseConverter<KEY_T,VALUE_T> {
- public static Function<Cell, Map.Entry<byte[], byte[]>> CELL_TO_ENTRY = new Function<Cell, Map.Entry<byte[], byte[]>>() {
+ public static Function<Cell, Map.Entry<byte[], byte[]>> CELL_TO_ENTRY = new Function<Cell, Map.Entry<byte[], byte[]>>() {
- @Nullable
- @Override
- public Map.Entry<byte[], byte[]> apply(@Nullable Cell cell) {
- return new AbstractMap.SimpleEntry<byte[], byte[]>(cell.getQualifier(), cell.getValue());
- }
- };
+ @Nullable
@Override
- public Put toPut(String columnFamily, KEY_T key, VALUE_T values) throws IOException {
- Put put = new Put(key.toBytes());
- byte[] cf = Bytes.toBytes(columnFamily);
- for(Map.Entry<byte[], byte[]> kv : values.toColumns()) {
- put.add(cf, kv.getKey(), kv.getValue());
- }
- return put;
+ public Map.Entry<byte[], byte[]> apply(@Nullable Cell cell) {
+ return new AbstractMap.SimpleEntry<>(cell.getQualifier(), cell.getValue());
}
-
- public LookupKV<KEY_T, VALUE_T> fromPut(Put put, String columnFamily, KEY_T key, VALUE_T value) throws IOException {
- key.fromBytes(put.getRow());
- byte[] cf = Bytes.toBytes(columnFamily);
- value.fromColumns(Iterables.transform(put.getFamilyCellMap().get(cf), CELL_TO_ENTRY));
- return new LookupKV<>(key, value);
+ };
+ @Override
+ public Put toPut(String columnFamily, KEY_T key, VALUE_T values) throws IOException {
+ Put put = new Put(key.toBytes());
+ byte[] cf = Bytes.toBytes(columnFamily);
+ for(Map.Entry<byte[], byte[]> kv : values.toColumns()) {
+ put.add(cf, kv.getKey(), kv.getValue());
}
+ return put;
+ }
- @Override
- public Result toResult(String columnFamily, KEY_T key, VALUE_T values) throws IOException {
- Put put = toPut(columnFamily, key, values);
- return Result.create(put.getFamilyCellMap().get(Bytes.toBytes(columnFamily)));
- }
+ public LookupKV<KEY_T, VALUE_T> fromPut(Put put, String columnFamily, KEY_T key, VALUE_T value) throws IOException {
+ key.fromBytes(put.getRow());
+ byte[] cf = Bytes.toBytes(columnFamily);
+ value.fromColumns(Iterables.transform(put.getFamilyCellMap().get(cf), CELL_TO_ENTRY));
+ return new LookupKV<>(key, value);
+ }
- public LookupKV<KEY_T, VALUE_T> fromResult(Result result, String columnFamily, KEY_T key, VALUE_T value) throws IOException {
- key.fromBytes(result.getRow());
- byte[] cf = Bytes.toBytes(columnFamily);
- NavigableMap<byte[], byte[]> cols = result.getFamilyMap(cf);
- value.fromColumns(cols.entrySet());
- return new LookupKV<>(key, value);
- }
- @Override
- public Get toGet(String columnFamily, KEY_T key) {
- Get ret = new Get(key.toBytes());
- ret.addFamily(Bytes.toBytes(columnFamily));
- return ret;
+ @Override
+ public Result toResult(String columnFamily, KEY_T key, VALUE_T values) throws IOException {
+ Put put = toPut(columnFamily, key, values);
+ return Result.create(put.getFamilyCellMap().get(Bytes.toBytes(columnFamily)));
+ }
+
+ public LookupKV<KEY_T, VALUE_T> fromResult(Result result, String columnFamily, KEY_T key, VALUE_T value) throws IOException {
+ if(result == null || result.getRow() == null) {
+ return null;
}
+ key.fromBytes(result.getRow());
+ byte[] cf = Bytes.toBytes(columnFamily);
+ NavigableMap<byte[], byte[]> cols = result.getFamilyMap(cf);
+ value.fromColumns(cols.entrySet());
+ return new LookupKV<>(key, value);
+ }
+ @Override
+ public Get toGet(String columnFamily, KEY_T key) {
+ Get ret = new Get(key.toBytes());
+ ret.addFamily(Bytes.toBytes(columnFamily));
+ return ret;
+ }
- public static Iterable<Map.Entry<byte[], byte[]>> toEntries(byte[]... kvs) {
- if(kvs.length % 2 != 0) {
- throw new IllegalStateException("Must be an even size");
- }
- List<Map.Entry<byte[], byte[]>> ret = new ArrayList<>(kvs.length/2);
- for(int i = 0;i < kvs.length;i += 2) {
- ret.add(new AbstractMap.SimpleImmutableEntry<>(kvs[i], kvs[i+1])) ;
- }
- return ret;
+ public static Iterable<Map.Entry<byte[], byte[]>> toEntries(byte[]... kvs) {
+ if(kvs.length % 2 != 0) {
+ throw new IllegalStateException("Must be an even size");
+ }
+ List<Map.Entry<byte[], byte[]>> ret = new ArrayList<>(kvs.length/2);
+ for(int i = 0;i < kvs.length;i += 2) {
+ ret.add(new AbstractMap.SimpleImmutableEntry<>(kvs[i], kvs[i+1])) ;
}
+ return ret;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverter.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverter.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverter.java
new file mode 100644
index 0000000..a044498
--- /dev/null
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverter.java
@@ -0,0 +1,38 @@
+/**
+ * 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.converters.enrichment;
+
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.metron.hbase.converters.AbstractConverter;
+import org.apache.metron.reference.lookup.LookupKV;
+
+import java.io.IOException;
+
+public class EnrichmentConverter extends AbstractConverter<EnrichmentKey, EnrichmentValue> {
+
+ @Override
+ public LookupKV<EnrichmentKey, EnrichmentValue> fromPut(Put put, String columnFamily) throws IOException {
+ return fromPut(put, columnFamily, new EnrichmentKey(), new EnrichmentValue());
+ }
+
+ @Override
+ public LookupKV<EnrichmentKey, EnrichmentValue> fromResult(Result result, String columnFamily) throws IOException {
+ return fromResult(result, columnFamily, new EnrichmentKey(), new EnrichmentValue());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentKey.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentKey.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentKey.java
new file mode 100644
index 0000000..b688ad3
--- /dev/null
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentKey.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.hbase.converters.enrichment;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
+import com.google.common.hash.HashFunction;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.reference.lookup.LookupKey;
+
+import java.io.*;
+
+public class EnrichmentKey implements LookupKey {
+ private static final int SEED = 0xDEADBEEF;
+ private static final int HASH_PREFIX_SIZE=16;
+ ThreadLocal<HashFunction> hFunction= new ThreadLocal<HashFunction>() {
+ @Override
+ protected HashFunction initialValue() {
+ return Hashing.murmur3_128(SEED);
+ }
+ };
+
+ public String indicator;
+ public String type;
+
+ public EnrichmentKey() {
+
+ }
+ public EnrichmentKey(String type, String indicator) {
+ this.indicator = indicator;
+ this.type = type;
+ }
+
+ private byte[] typedIndicatorToBytes() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream w = new DataOutputStream(baos);
+ w.writeUTF(type);
+ w.writeUTF(indicator);
+ w.flush();
+ return baos.toByteArray();
+ }
+
+ @Override
+ public byte[] toBytes() {
+ byte[] indicatorBytes = new byte[0];
+ try {
+ indicatorBytes = typedIndicatorToBytes();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to convert type and indicator to bytes", e);
+ }
+ Hasher hasher = hFunction.get().newHasher();
+ hasher.putBytes(Bytes.toBytes(indicator));
+ byte[] prefix = hasher.hash().asBytes();
+ byte[] val = new byte[indicatorBytes.length + prefix.length];
+ int offset = 0;
+ System.arraycopy(prefix, 0, val, offset, prefix.length);
+ offset += prefix.length;
+ System.arraycopy(indicatorBytes, 0, val, offset, indicatorBytes.length);
+ return val;
+ }
+
+ @Override
+ public void fromBytes(byte[] row) {
+ ByteArrayInputStream baos = new ByteArrayInputStream(row);
+ baos.skip(HASH_PREFIX_SIZE);
+ DataInputStream w = new DataInputStream(baos);
+ try {
+ type = w.readUTF();
+ indicator = w.readUTF();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to convert type and indicator from bytes", e);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ EnrichmentKey that = (EnrichmentKey) o;
+
+ if (indicator != null ? !indicator.equals(that.indicator) : that.indicator != null) return false;
+ return type != null ? type.equals(that.type) : that.type == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = indicator != null ? indicator.hashCode() : 0;
+ result = 31 * result + (type != null ? type.hashCode() : 0);
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return "EnrichmentKey{" +
+ "indicator='" + indicator + '\'' +
+ ", type='" + type + '\'' +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentValue.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentValue.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentValue.java
new file mode 100644
index 0000000..f733f8e
--- /dev/null
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/enrichment/EnrichmentValue.java
@@ -0,0 +1,107 @@
+/**
+ * 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.converters.enrichment;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.metron.hbase.converters.AbstractConverter;
+import org.apache.metron.reference.lookup.LookupValue;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.type.TypeReference;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class EnrichmentValue implements LookupValue {
+ private static final ThreadLocal<ObjectMapper> _mapper = new ThreadLocal<ObjectMapper>() {
+ @Override
+ protected ObjectMapper initialValue() {
+ return new ObjectMapper();
+ }
+ };
+ public static final String VALUE_COLUMN_NAME = "v";
+ public static final byte[] VALUE_COLUMN_NAME_B = Bytes.toBytes(VALUE_COLUMN_NAME);
+
+ private Map<String, String> metadata = null;
+
+ public EnrichmentValue()
+ {
+
+ }
+
+ public EnrichmentValue(Map<String, String> metadata) {
+ this.metadata = metadata;
+ }
+
+
+
+ public Map<String, String> getMetadata() {
+ return metadata;
+ }
+
+ @Override
+ public Iterable<Map.Entry<byte[], byte[]>> toColumns() {
+ return AbstractConverter.toEntries( VALUE_COLUMN_NAME_B, Bytes.toBytes(valueToString(metadata))
+ );
+ }
+
+ @Override
+ public void fromColumns(Iterable<Map.Entry<byte[], byte[]>> values) {
+ for(Map.Entry<byte[], byte[]> cell : values) {
+ if(Bytes.equals(cell.getKey(), VALUE_COLUMN_NAME_B)) {
+ metadata = stringToValue(Bytes.toString(cell.getValue()));
+ }
+ }
+ }
+ public Map<String, String> stringToValue(String s){
+ try {
+ return _mapper.get().readValue(s, new TypeReference<Map<String, String>>(){});
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to convert string to metadata: " + s);
+ }
+ }
+ public String valueToString(Map<String, String> value) {
+ try {
+ return _mapper.get().writeValueAsString(value);
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to convert metadata to string: " + value);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ EnrichmentValue that = (EnrichmentValue) o;
+
+ return getMetadata() != null ? getMetadata().equals(that.getMetadata()) : that.getMetadata() == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ return getMetadata() != null ? getMetadata().hashCode() : 0;
+ }
+
+ @Override
+ public String toString() {
+ return "EnrichmentValue{" +
+ "metadata=" + metadata +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelConverter.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelConverter.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelConverter.java
deleted file mode 100644
index d534a52..0000000
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelConverter.java
+++ /dev/null
@@ -1,39 +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.hbase.converters.threatintel;
-
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.metron.hbase.converters.AbstractConverter;
-import org.apache.metron.reference.lookup.LookupKV;
-
-import java.io.IOException;
-
-public class ThreatIntelConverter extends AbstractConverter<ThreatIntelKey, ThreatIntelValue> {
-
- @Override
- public LookupKV<ThreatIntelKey, ThreatIntelValue> fromPut(Put put, String columnFamily) throws IOException {
- return fromPut(put, columnFamily, new ThreatIntelKey(), new ThreatIntelValue());
- }
-
- @Override
- public LookupKV<ThreatIntelKey, ThreatIntelValue> fromResult(Result result, String columnFamily) throws IOException {
- return fromResult(result, columnFamily, new ThreatIntelKey(), new ThreatIntelValue());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelKey.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelKey.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelKey.java
deleted file mode 100644
index 3d898d9..0000000
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelKey.java
+++ /dev/null
@@ -1,89 +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.hbase.converters.threatintel;
-
-import com.google.common.hash.HashFunction;
-import com.google.common.hash.Hasher;
-import com.google.common.hash.Hashing;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.metron.reference.lookup.LookupKey;
-
-public class ThreatIntelKey implements LookupKey{
- private static final int SEED = 0xDEADBEEF;
- private static final int HASH_PREFIX_SIZE=16;
- ThreadLocal<HashFunction> hFunction= new ThreadLocal<HashFunction>() {
- @Override
- protected HashFunction initialValue() {
- return Hashing.murmur3_128(SEED);
- }
- };
- public ThreatIntelKey() {
-
- }
- public ThreatIntelKey(String indicator) {
- this.indicator = indicator;
- }
-
- public String indicator;
-
- @Override
- public byte[] toBytes() {
- byte[] indicatorBytes = Bytes.toBytes(indicator);
- Hasher hasher = hFunction.get().newHasher();
- hasher.putBytes(Bytes.toBytes(indicator));
- byte[] prefix = hasher.hash().asBytes();
- byte[] val = new byte[indicatorBytes.length + prefix.length];
- int pos = 0;
- for(int i = 0;pos < prefix.length;++pos,++i) {
- val[pos] = prefix[i];
- }
- for(int i = 0;i < indicatorBytes.length;++pos,++i) {
- val[pos] = indicatorBytes[i];
- }
- return val;
- }
-
- @Override
- public void fromBytes(byte[] row) {
- ThreatIntelKey key = this;
- key.indicator = Bytes.toString(row, HASH_PREFIX_SIZE, row.length - HASH_PREFIX_SIZE);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- ThreatIntelKey that = (ThreatIntelKey) o;
-
- return indicator != null ? indicator.equals(that.indicator) : that.indicator == null;
-
- }
-
- @Override
- public int hashCode() {
- return indicator != null ? indicator.hashCode() : 0;
- }
-
- @Override
- public String toString() {
- return "ThreatIntelKey{" +
- "indicator='" + indicator + '\'' +
- '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelValue.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelValue.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelValue.java
deleted file mode 100644
index 97f0762..0000000
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/converters/threatintel/ThreatIntelValue.java
+++ /dev/null
@@ -1,110 +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.hbase.converters.threatintel;
-
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.metron.hbase.converters.AbstractConverter;
-import org.apache.metron.reference.lookup.LookupValue;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.type.TypeReference;
-
-import java.io.IOException;
-import java.util.Map;
-
-public class ThreatIntelValue implements LookupValue {
- private static final ThreadLocal<ObjectMapper> _mapper = new ThreadLocal<ObjectMapper>() {
- @Override
- protected ObjectMapper initialValue() {
- return new ObjectMapper();
- }
- };
- public static final String VALUE_COLUMN_NAME = "v";
- public static final byte[] VALUE_COLUMN_NAME_B = Bytes.toBytes(VALUE_COLUMN_NAME);
- public static final String LAST_SEEN_COLUMN_NAME = "t";
- public static final byte[] LAST_SEEN_COLUMN_NAME_B = Bytes.toBytes(LAST_SEEN_COLUMN_NAME);
-
- private Map<String, String> metadata = null;
-
- public ThreatIntelValue()
- {
-
- }
-
- public ThreatIntelValue(Map<String, String> metadata) {
- this.metadata = metadata;
- }
-
-
-
- public Map<String, String> getMetadata() {
- return metadata;
- }
-
- @Override
- public Iterable<Map.Entry<byte[], byte[]>> toColumns() {
- return AbstractConverter.toEntries( VALUE_COLUMN_NAME_B, Bytes.toBytes(valueToString(metadata))
- );
- }
-
- @Override
- public void fromColumns(Iterable<Map.Entry<byte[], byte[]>> values) {
- for(Map.Entry<byte[], byte[]> cell : values) {
- if(Bytes.equals(cell.getKey(), VALUE_COLUMN_NAME_B)) {
- metadata = stringToValue(Bytes.toString(cell.getValue()));
- }
- }
- }
- public Map<String, String> stringToValue(String s){
- try {
- return _mapper.get().readValue(s, new TypeReference<Map<String, String>>(){});
- } catch (IOException e) {
- throw new RuntimeException("Unable to convert string to metadata: " + s);
- }
- }
- public String valueToString(Map<String, String> value) {
- try {
- return _mapper.get().writeValueAsString(value);
- } catch (IOException e) {
- throw new RuntimeException("Unable to convert metadata to string: " + value);
- }
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- ThreatIntelValue that = (ThreatIntelValue) o;
-
- return getMetadata() != null ? getMetadata().equals(that.getMetadata()) : that.getMetadata() == null;
-
- }
-
- @Override
- public int hashCode() {
- return getMetadata() != null ? getMetadata().hashCode() : 0;
- }
-
- @Override
- public String toString() {
- return "ThreatIntelValue{" +
- "metadata=" + metadata +
- '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/lookup/EnrichmentLookup.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/lookup/EnrichmentLookup.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/lookup/EnrichmentLookup.java
new file mode 100644
index 0000000..059d6a6
--- /dev/null
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/hbase/lookup/EnrichmentLookup.java
@@ -0,0 +1,108 @@
+/**
+ * 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.lookup;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.metron.hbase.converters.HbaseConverter;
+import org.apache.metron.hbase.converters.enrichment.EnrichmentConverter;
+import org.apache.metron.hbase.converters.enrichment.EnrichmentKey;
+import org.apache.metron.hbase.converters.enrichment.EnrichmentValue;
+import org.apache.metron.reference.lookup.Lookup;
+import org.apache.metron.reference.lookup.LookupKV;
+import org.apache.metron.reference.lookup.accesstracker.AccessTracker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+
+public class EnrichmentLookup extends Lookup<HTableInterface, EnrichmentKey, LookupKV<EnrichmentKey,EnrichmentValue>> implements AutoCloseable {
+
+ public static class Handler implements org.apache.metron.reference.lookup.handler.Handler<HTableInterface,EnrichmentKey,LookupKV<EnrichmentKey,EnrichmentValue>> {
+ String columnFamily;
+ HbaseConverter<EnrichmentKey, EnrichmentValue> converter = new EnrichmentConverter();
+ public Handler(String columnFamily) {
+ this.columnFamily = columnFamily;
+ }
+ @Override
+ public boolean exists(EnrichmentKey key, HTableInterface table, boolean logAccess) throws IOException {
+ return table.exists(converter.toGet(columnFamily, key));
+ }
+
+ @Override
+ public LookupKV<EnrichmentKey, EnrichmentValue> get(EnrichmentKey key, HTableInterface table, boolean logAccess) throws IOException {
+ return converter.fromResult(table.get(converter.toGet(columnFamily, key)), columnFamily);
+ }
+
+ private List<Get> keysToGets(Iterable<EnrichmentKey> keys) {
+ List<Get> ret = new ArrayList<>();
+ for(EnrichmentKey key : keys) {
+ ret.add(converter.toGet(columnFamily, key));
+ }
+ return ret;
+ }
+
+ @Override
+ public Iterable<Boolean> exists(Iterable<EnrichmentKey> key, HTableInterface table, boolean logAccess) throws IOException {
+ List<Boolean> ret = new ArrayList<>();
+ for(boolean b : table.existsAll(keysToGets(key))) {
+ ret.add(b);
+ }
+ return ret;
+ }
+
+ @Override
+ public Iterable<LookupKV<EnrichmentKey, EnrichmentValue>> get( Iterable<EnrichmentKey> keys
+ , HTableInterface table
+ , boolean logAccess
+ ) throws IOException
+ {
+ List<LookupKV<EnrichmentKey, EnrichmentValue>> ret = new ArrayList<>();
+ for(Result result : table.get(keysToGets(keys))) {
+ ret.add(converter.fromResult(result, columnFamily));
+ }
+ return ret;
+ }
+
+
+ @Override
+ public void close() throws Exception {
+
+ }
+ }
+ private HTableInterface table;
+ public EnrichmentLookup(HTableInterface table, String columnFamily, AccessTracker tracker) {
+ this.table = table;
+ this.setLookupHandler(new Handler(columnFamily));
+ this.setAccessTracker(tracker);
+ }
+
+ public HTableInterface getTable() {
+ return table;
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ table.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/Lookup.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/Lookup.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/Lookup.java
index e93134e..bf871d1 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/Lookup.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/Lookup.java
@@ -23,53 +23,74 @@ import org.apache.metron.reference.lookup.handler.Handler;
import java.io.IOException;
public class Lookup<CONTEXT_T, KEY_T extends LookupKey, RESULT_T> implements Handler<CONTEXT_T, KEY_T, RESULT_T> {
- private String name;
- private AccessTracker accessTracker;
- private Handler<CONTEXT_T, KEY_T, RESULT_T> lookupHandler;
+ private String name;
+ private AccessTracker accessTracker;
+ private Handler<CONTEXT_T, KEY_T, RESULT_T> lookupHandler;
- public String getName() {
- return name;
- }
+ public String getName() {
+ return name;
+ }
- public void setName(String name) {
- this.name = name;
- }
+ public void setName(String name) {
+ this.name = name;
+ }
- public AccessTracker getAccessTracker() {
- return accessTracker;
- }
+ public AccessTracker getAccessTracker() {
+ return accessTracker;
+ }
- public void setAccessTracker(AccessTracker accessTracker) {
- this.accessTracker = accessTracker;
- }
+ public void setAccessTracker(AccessTracker accessTracker) {
+ this.accessTracker = accessTracker;
+ }
- public Handler< CONTEXT_T, KEY_T, RESULT_T > getLookupHandler() {
- return lookupHandler;
- }
+ public Handler< CONTEXT_T, KEY_T, RESULT_T > getLookupHandler() {
+ return lookupHandler;
+ }
+
+ public void setLookupHandler(Handler< CONTEXT_T, KEY_T, RESULT_T > lookupHandler) {
+ this.lookupHandler = lookupHandler;
+ }
- public void setLookupHandler(Handler< CONTEXT_T, KEY_T, RESULT_T > lookupHandler) {
- this.lookupHandler = lookupHandler;
+ @Override
+ public boolean exists(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException {
+ if(logAccess) {
+ accessTracker.logAccess(key);
}
+ return lookupHandler.exists(key, context, logAccess);
+ }
- @Override
- public boolean exists(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException {
- if(logAccess) {
- accessTracker.logAccess(key);
- }
- return lookupHandler.exists(key, context, logAccess);
+ @Override
+ public RESULT_T get(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException {
+ if(logAccess) {
+ accessTracker.logAccess(key);
}
+ return lookupHandler.get(key, context, logAccess);
+ }
- @Override
- public RESULT_T get(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException {
- if(logAccess) {
- accessTracker.logAccess(key);
- }
- return lookupHandler.get(key, context, logAccess);
+ @Override
+ public Iterable<Boolean> exists(Iterable<KEY_T> key, CONTEXT_T context, boolean logAccess) throws IOException {
+ if(logAccess) {
+ for (KEY_T k : key) {
+ accessTracker.logAccess(k);
+ }
}
+ return lookupHandler.exists(key, context, logAccess);
+ }
- @Override
- public void close() throws Exception {
- accessTracker.cleanup();
- lookupHandler.close();
+
+ @Override
+ public Iterable<RESULT_T> get(Iterable<KEY_T> key, CONTEXT_T context, boolean logAccess) throws IOException {
+ if(logAccess) {
+ for (KEY_T k : key) {
+ accessTracker.logAccess(k);
+ }
}
+ return lookupHandler.get(key, context, logAccess);
+ }
+
+ @Override
+ public void close() throws Exception {
+ accessTracker.cleanup();
+ lookupHandler.close();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/BloomAccessTracker.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/BloomAccessTracker.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/BloomAccessTracker.java
index d2700d6..2bb8414 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/BloomAccessTracker.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/BloomAccessTracker.java
@@ -43,6 +43,7 @@ public class BloomAccessTracker implements AccessTracker {
public boolean equals(Object obj) {
return this.getClass().equals(obj.getClass());
}
+
}
private static Funnel<LookupKey> LOOKUPKEY_FUNNEL = new LookupKeyFunnel();
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/NoopAccessTracker.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/NoopAccessTracker.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/NoopAccessTracker.java
new file mode 100644
index 0000000..fb6d3ba
--- /dev/null
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/accesstracker/NoopAccessTracker.java
@@ -0,0 +1,65 @@
+/**
+ * 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.reference.lookup.accesstracker;
+
+import org.apache.metron.reference.lookup.LookupKey;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class NoopAccessTracker implements AccessTracker {
+ @Override
+ public void logAccess(LookupKey key) {
+
+ }
+
+ @Override
+ public void configure(Map<String, Object> config) {
+
+ }
+
+ @Override
+ public boolean hasSeen(LookupKey key) {
+ return false;
+ }
+
+ @Override
+ public String getName() {
+ return "noop";
+ }
+
+ @Override
+ public AccessTracker union(AccessTracker tracker) {
+ return null;
+ }
+
+ @Override
+ public void reset() {
+
+ }
+
+ @Override
+ public boolean isFull() {
+ return false;
+ }
+
+ @Override
+ public void cleanup() throws IOException {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/handler/Handler.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/handler/Handler.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/handler/Handler.java
index 9af8fbd..198f90e 100644
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/handler/Handler.java
+++ b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/reference/lookup/handler/Handler.java
@@ -22,6 +22,8 @@ import org.apache.metron.reference.lookup.LookupKey;
import java.io.IOException;
public interface Handler<CONTEXT_T, KEY_T extends LookupKey, RESULT_T> extends AutoCloseable{
- boolean exists(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException;
- RESULT_T get(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException;
+ boolean exists(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException;
+ RESULT_T get(KEY_T key, CONTEXT_T context, boolean logAccess) throws IOException;
+ Iterable<Boolean> exists(Iterable<KEY_T> key, CONTEXT_T context, boolean logAccess) throws IOException;
+ Iterable<RESULT_T> get(Iterable<KEY_T> key, CONTEXT_T context, boolean logAccess) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/ThreatIntelResults.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/ThreatIntelResults.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/ThreatIntelResults.java
deleted file mode 100644
index 8186b38..0000000
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/ThreatIntelResults.java
+++ /dev/null
@@ -1,71 +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.threatintel;
-
-import org.apache.metron.hbase.converters.threatintel.ThreatIntelKey;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public class ThreatIntelResults {
- private ThreatIntelKey key;
- private Map<String, String> value;
- public ThreatIntelResults() {
- key = new ThreatIntelKey();
- value = new HashMap<>();
- }
- public ThreatIntelResults(ThreatIntelKey key, Map<String, String> value) {
- this.key = key;
- this.value = value;
- }
-
- public ThreatIntelKey getKey() {
- return key;
- }
-
- public Map<String, String> getValue() {
- return value;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- ThreatIntelResults that = (ThreatIntelResults) o;
-
- if (getKey() != null ? !getKey().equals(that.getKey()) : that.getKey() != null) return false;
- return getValue() != null ? getValue().equals(that.getValue()) : that.getValue() == null;
-
- }
-
- @Override
- public int hashCode() {
- int result = getKey() != null ? getKey().hashCode() : 0;
- result = 31 * result + (getValue() != null ? getValue().hashCode() : 0);
- return result;
- }
-
- @Override
- public String toString() {
- return "ThreatIntelResults{" +
- "key=" + key +
- ", value=" + value +
- '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/9621c55e/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/hbase/ThreatIntelLookup.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/hbase/ThreatIntelLookup.java b/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/hbase/ThreatIntelLookup.java
deleted file mode 100644
index 13efc42..0000000
--- a/metron-streaming/Metron-Common/src/main/java/org/apache/metron/threatintel/hbase/ThreatIntelLookup.java
+++ /dev/null
@@ -1,74 +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.threatintel.hbase;
-
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.metron.hbase.converters.HbaseConverter;
-import org.apache.metron.hbase.converters.threatintel.ThreatIntelConverter;
-import org.apache.metron.hbase.converters.threatintel.ThreatIntelKey;
-import org.apache.metron.hbase.converters.threatintel.ThreatIntelValue;
-import org.apache.metron.reference.lookup.Lookup;
-import org.apache.metron.reference.lookup.LookupKV;
-import org.apache.metron.reference.lookup.accesstracker.AccessTracker;
-
-import java.io.IOException;
-
-
-public class ThreatIntelLookup extends Lookup<HTableInterface, ThreatIntelKey, LookupKV<ThreatIntelKey,ThreatIntelValue>> implements AutoCloseable {
-
-
-
- public static class Handler implements org.apache.metron.reference.lookup.handler.Handler<HTableInterface,ThreatIntelKey,LookupKV<ThreatIntelKey,ThreatIntelValue>> {
- String columnFamily;
- HbaseConverter<ThreatIntelKey, ThreatIntelValue> converter = new ThreatIntelConverter();
- public Handler(String columnFamily) {
- this.columnFamily = columnFamily;
- }
- @Override
- public boolean exists(ThreatIntelKey key, HTableInterface table, boolean logAccess) throws IOException {
- return table.exists(converter.toGet(columnFamily, key));
- }
-
- @Override
- public LookupKV<ThreatIntelKey, ThreatIntelValue> get(ThreatIntelKey key, HTableInterface table, boolean logAccess) throws IOException {
- return converter.fromResult(table.get(converter.toGet(columnFamily, key)), columnFamily);
- }
-
-
- @Override
- public void close() throws Exception {
-
- }
- }
- private HTableInterface table;
- public ThreatIntelLookup(HTableInterface table, String columnFamily, AccessTracker tracker) {
- this.table = table;
- this.setLookupHandler(new Handler(columnFamily));
- this.setAccessTracker(tracker);
- }
-
- public HTableInterface getTable() {
- return table;
- }
-
- @Override
- public void close() throws Exception {
- super.close();
- table.close();
- }
-}