From 4486b7ff58bfb98a5bc1ed8e05490fb7202bab9f Mon Sep 17 00:00:00 2001 From: cstella Date: Mon, 22 Jan 2018 17:11:18 -0500 Subject: [PATCH 01/18] Updating the stack definitions and such on the mpack to 2.6 --- .../metron-mpack/src/main/resources/mpack.json | 14 ++++++++++++++ .../roles/ambari_common/defaults/main.yml | 2 +- .../roles/ambari_config/defaults/main.yml | 2 +- .../roles/ambari_config/vars/single_node_vm.yml | 5 +++++ .../roles/ambari_config/vars/small_cluster.yml | 5 +++++ 5 files changed, 26 insertions(+), 2 deletions(-) diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/mpack.json b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/mpack.json index 7a9d8927d9..39468819ae 100644 --- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/mpack.json +++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/mpack.json @@ -38,7 +38,12 @@ { "stack_name" : "HDP", "stack_version" : "2.5" + }, + { + "stack_name" : "HDP", + "stack_version" : "2.6" } + ] }, { @@ -56,8 +61,13 @@ { "stack_name" : "HDP", "stack_version" : "2.5" + }, + { + "stack_name" : "HDP", + "stack_version" : "2.6" } + ] }, { @@ -75,6 +85,10 @@ { "stack_name" : "HDP", "stack_version" : "2.5" + }, + { + "stack_name" : "HDP", + "stack_version" : "2.6" } ] diff --git a/metron-deployment/roles/ambari_common/defaults/main.yml b/metron-deployment/roles/ambari_common/defaults/main.yml index 2c9389c63e..e7b2da77d4 100644 --- a/metron-deployment/roles/ambari_common/defaults/main.yml +++ b/metron-deployment/roles/ambari_common/defaults/main.yml @@ -17,4 +17,4 @@ --- hadoop_logrotate_frequency: daily hadoop_logrotate_retention: 30 -rhel_ambari_install_url: http://public-repo-1.hortonworks.com/ambari/centos6/2.x/updates/2.4.2.0/ambari.repo +rhel_ambari_install_url: http://public-repo-1.hortonworks.com/ambari/centos6/2.x/updates/2.5.2.0/ambari.repo diff --git a/metron-deployment/roles/ambari_config/defaults/main.yml b/metron-deployment/roles/ambari_config/defaults/main.yml index e0de145fd0..ad7ca9ed5e 100644 --- a/metron-deployment/roles/ambari_config/defaults/main.yml +++ b/metron-deployment/roles/ambari_config/defaults/main.yml @@ -34,5 +34,5 @@ mapred_reduce_java_opts : -Xmx1024m mapred_map_mem_mb : 1229 mapred_reduce_mem_mb : 1229 topology_classpath: '/etc/hbase/conf:/etc/hadoop/conf' -hdp_stack: "2.5" +hdp_stack: "2.6" elasticsearch_network_interface: _site_ diff --git a/metron-deployment/roles/ambari_config/vars/single_node_vm.yml b/metron-deployment/roles/ambari_config/vars/single_node_vm.yml index 6a6090235b..24b9297a49 100644 --- a/metron-deployment/roles/ambari_config/vars/single_node_vm.yml +++ b/metron-deployment/roles/ambari_config/vars/single_node_vm.yml @@ -87,6 +87,11 @@ configurations: supervisor.slots.ports: "[6700, 6701, 6702, 6703, 6704, 6705]" storm.local.dir: '{{ storm_local_dir }}' topology.classpath: '{{ topology_classpath }}' + storm.cluster.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer"}]' + topology.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer", "parallelism.hint": 1, "whitelist": ["kafkaOffset\\..+/", "__complete-latency", "__process-latency", "__receive\\.population$", "__sendqueue\\.population$", "__execute-count", "__emit-count", "__ack-count", "__fail-count", "memory/heap\\.usedBytes$", "memory/nonHeap\\.usedBytes$", "GC/.+\\.count$", "GC/.+\\.timeMs$"]}]' + nimbus.childopts: '-Xmx1024m _JAAS_PLACEHOLDER' + supervisor.childopts: '-Xmx256m _JAAS_PLACEHOLDER' + worker.childopts: "-Xmx768m _JAAS_PLACEHOLDER" - kafka-env: content: "{% raw %}\n#!/bin/bash\n\n# Set KAFKA specific environment variables here.\n\n# The java implementation to use.\nexport KAFKA_HEAP_OPTS=\"-Xms256M -Xmx256M\"\nexport KAFKA_JVM_PERFORMANCE_OPTS=\"-server -XX:+UseG1GC -XX:+DisableExplicitGC -Djava.awt.headless=true\"\nexport JAVA_HOME={{java64_home}}\nexport PATH=$PATH:$JAVA_HOME/bin\nexport PID_DIR={{kafka_pid_dir}}\nexport LOG_DIR={{kafka_log_dir}}\nexport KAFKA_KERBEROS_PARAMS={{kafka_kerberos_params}}\n# Add kafka sink to classpath and related depenencies\nif [ -e \"/usr/lib/ambari-metrics-kafka-sink/ambari-metrics-kafka-sink.jar\" ]; then\n export CLASSPATH=$CLASSPATH:/usr/lib/ambari-metrics-kafka-sink/ambari-metrics-kafka-sink.jar\n export CLASSPATH=$CLASSPATH:/usr/lib/ambari-metrics-kafka-sink/lib/*\nfi\nif [ -f /etc/kafka/conf/kafka-ranger-env.sh ]; then\n . /etc/kafka/conf/kafka-ranger-env.sh\nfi{% endraw %}" - kafka-broker: diff --git a/metron-deployment/roles/ambari_config/vars/small_cluster.yml b/metron-deployment/roles/ambari_config/vars/small_cluster.yml index 4ec845856b..10bfc9781a 100644 --- a/metron-deployment/roles/ambari_config/vars/small_cluster.yml +++ b/metron-deployment/roles/ambari_config/vars/small_cluster.yml @@ -85,6 +85,11 @@ configurations: supervisor.slots.ports: "[6700, 6701, 6702, 6703, 6704, 6705]" storm.local.dir: '{{ storm_local_dir | default("/hadoop/storm") }}' topology.classpath: '{{ topology_classpath }}' + storm.cluster.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer"}]' + topology.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer", "parallelism.hint": 1, "whitelist": ["kafkaOffset\\..+/", "__complete-latency", "__process-latency", "__receive\\.population$", "__sendqueue\\.population$", "__execute-count", "__emit-count", "__ack-count", "__fail-count", "memory/heap\\.usedBytes$", "memory/nonHeap\\.usedBytes$", "GC/.+\\.count$", "GC/.+\\.timeMs$"]}]' + nimbus.childopts: '-Xmx1024m _JAAS_PLACEHOLDER' + supervisor.childopts: '-Xmx256m _JAAS_PLACEHOLDER' + worker.childopts: "-Xmx768m _JAAS_PLACEHOLDER" - kafka-broker: log.dirs: '{{ kafka_log_dirs | default("/kafka-log") }}' - metron-rest-env: From 972fe01e19e28f3be94393d83b49a269dbc09e9c Mon Sep 17 00:00:00 2001 From: cstella Date: Tue, 23 Jan 2018 12:10:38 -0500 Subject: [PATCH 02/18] random service check bug --- .../ELASTICSEARCH/5.6.2/package/scripts/service_check.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/ELASTICSEARCH/5.6.2/package/scripts/service_check.py b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/ELASTICSEARCH/5.6.2/package/scripts/service_check.py index d59954f837..5b82af120b 100755 --- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/ELASTICSEARCH/5.6.2/package/scripts/service_check.py +++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/ELASTICSEARCH/5.6.2/package/scripts/service_check.py @@ -32,7 +32,7 @@ def service_check(self, env): doc = '{"name": "Ambari Smoke test"}' index = "ambari_smoke_test" - Logger.info("Running Elastic search service check", file=sys.stdout) + Logger.info("Running Elastic search service check") # Make sure the service is actually up. We can live without everything allocated. # Need both the retry and ES timeout. Can hit the URL before ES is ready at all and get no response, but can From 5d56bc1fba811630554192e2fc8614b2050d520d Mon Sep 17 00:00:00 2001 From: cstella Date: Tue, 23 Jan 2018 19:36:37 -0500 Subject: [PATCH 03/18] Updating poms to relocate jackson. --- metron-analytics/metron-profiler/pom.xml | 4 ++-- metron-platform/metron-elasticsearch/pom.xml | 4 ++-- metron-platform/metron-solr/pom.xml | 12 +++++++++++- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/metron-analytics/metron-profiler/pom.xml b/metron-analytics/metron-profiler/pom.xml index d634cef683..4d36782100 100644 --- a/metron-analytics/metron-profiler/pom.xml +++ b/metron-analytics/metron-profiler/pom.xml @@ -341,8 +341,8 @@ org.apache.metron.guava.metron-profiler - com.fasterxml.jackson.core - com.fasterxml.jackson.core.metron.elasticsearch + com.fasterxml.jackson + org.apache.metron.jackson diff --git a/metron-platform/metron-elasticsearch/pom.xml b/metron-platform/metron-elasticsearch/pom.xml index 97f4062aed..141d8aaf6b 100644 --- a/metron-platform/metron-elasticsearch/pom.xml +++ b/metron-platform/metron-elasticsearch/pom.xml @@ -264,8 +264,8 @@ org.apache.metron.guava.metron-elasticsearch - com.fasterxml.jackson.core - com.fasterxml.jackson.core.metron.elasticsearch + com.fasterxml.jackson + org.apache.metron.jackson diff --git a/metron-platform/metron-solr/pom.xml b/metron-platform/metron-solr/pom.xml index a2eee7181d..9c9c7fbd6a 100644 --- a/metron-platform/metron-solr/pom.xml +++ b/metron-platform/metron-solr/pom.xml @@ -261,7 +261,17 @@ META-INF/*.RSA - + + + + com.google.common + org.apache.metron.guava + + + com.fasterxml.jackson + org.apache.metron.jackson + + storm:storm-core:* From b89fa88ddd769aece0e211c4b87a4b0bb15aa309 Mon Sep 17 00:00:00 2001 From: cstella Date: Wed, 24 Jan 2018 11:32:05 -0500 Subject: [PATCH 04/18] Added comments. --- metron-deployment/roles/ambari_config/vars/single_node_vm.yml | 2 ++ metron-deployment/roles/ambari_config/vars/small_cluster.yml | 2 ++ 2 files changed, 4 insertions(+) diff --git a/metron-deployment/roles/ambari_config/vars/single_node_vm.yml b/metron-deployment/roles/ambari_config/vars/single_node_vm.yml index 24b9297a49..bf54fe0da8 100644 --- a/metron-deployment/roles/ambari_config/vars/single_node_vm.yml +++ b/metron-deployment/roles/ambari_config/vars/single_node_vm.yml @@ -87,8 +87,10 @@ configurations: supervisor.slots.ports: "[6700, 6701, 6702, 6703, 6704, 6705]" storm.local.dir: '{{ storm_local_dir }}' topology.classpath: '{{ topology_classpath }}' + # Storm expects ambari metrics to be available in 2.6. We do *not* install ambari metrics in full-dev, so we need to revert to the old consumer storm.cluster.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer"}]' topology.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer", "parallelism.hint": 1, "whitelist": ["kafkaOffset\\..+/", "__complete-latency", "__process-latency", "__receive\\.population$", "__sendqueue\\.population$", "__execute-count", "__emit-count", "__ack-count", "__fail-count", "memory/heap\\.usedBytes$", "memory/nonHeap\\.usedBytes$", "GC/.+\\.count$", "GC/.+\\.timeMs$"]}]' + # Storm expects ambari metrics to be available in 2.6 and ambari metrics pulls data via JMX, but since we don't use ambari metrics here, we don't have the javaagent around to use and thus that must be removed from nimbus, supervisor and worker properties nimbus.childopts: '-Xmx1024m _JAAS_PLACEHOLDER' supervisor.childopts: '-Xmx256m _JAAS_PLACEHOLDER' worker.childopts: "-Xmx768m _JAAS_PLACEHOLDER" diff --git a/metron-deployment/roles/ambari_config/vars/small_cluster.yml b/metron-deployment/roles/ambari_config/vars/small_cluster.yml index 10bfc9781a..218e2670e4 100644 --- a/metron-deployment/roles/ambari_config/vars/small_cluster.yml +++ b/metron-deployment/roles/ambari_config/vars/small_cluster.yml @@ -85,8 +85,10 @@ configurations: supervisor.slots.ports: "[6700, 6701, 6702, 6703, 6704, 6705]" storm.local.dir: '{{ storm_local_dir | default("/hadoop/storm") }}' topology.classpath: '{{ topology_classpath }}' + # Storm expects ambari metrics to be available in 2.6. We do *not* install ambari metrics in full-dev, so we need to revert to the old consumer storm.cluster.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer"}]' topology.metrics.consumer.register: '[{"class": "org.apache.storm.metric.LoggingMetricsConsumer", "parallelism.hint": 1, "whitelist": ["kafkaOffset\\..+/", "__complete-latency", "__process-latency", "__receive\\.population$", "__sendqueue\\.population$", "__execute-count", "__emit-count", "__ack-count", "__fail-count", "memory/heap\\.usedBytes$", "memory/nonHeap\\.usedBytes$", "GC/.+\\.count$", "GC/.+\\.timeMs$"]}]' + # Storm expects ambari metrics to be available in 2.6 and ambari metrics pulls data via JMX, but since we don't use ambari metrics here, we don't have the javaagent around to use and thus that must be removed from nimbus, supervisor and worker properties nimbus.childopts: '-Xmx1024m _JAAS_PLACEHOLDER' supervisor.childopts: '-Xmx256m _JAAS_PLACEHOLDER' worker.childopts: "-Xmx768m _JAAS_PLACEHOLDER" From 3969a67a24cf07825fe2cd0a2418c9654b7e78c8 Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 09:28:43 -0500 Subject: [PATCH 05/18] Normalizing security protocols to the one that kafka supports. --- .../metron/rest/config/KafkaConfig.java | 8 ++++++-- .../rest/service/impl/StormCLIWrapper.java | 3 ++- .../metron/common/utils/KafkaUtils.java | 19 +++++++++++++++++++ .../topology/ParserTopologyBuilder.java | 3 ++- 4 files changed, 29 insertions(+), 4 deletions(-) diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java index a15c48f726..7e9b46859a 100644 --- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java +++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/KafkaConfig.java @@ -22,6 +22,8 @@ import org.I0Itec.zkclient.ZkClient; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.common.protocol.SecurityProtocol; +import org.apache.metron.common.utils.KafkaUtils; import org.apache.metron.rest.MetronRestConstants; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -86,7 +88,7 @@ public Map consumerProperties() { props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); if (environment.getProperty(MetronRestConstants.KERBEROS_ENABLED_SPRING_PROPERTY, Boolean.class, false)) { - props.put("security.protocol", environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY)); + props.put("security.protocol", KafkaUtils.INSTANCE.normalizeProtocol(environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY))); } return props; } @@ -109,11 +111,13 @@ public Map producerProperties() { producerConfig.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); producerConfig.put("request.required.acks", 1); if (environment.getProperty(MetronRestConstants.KERBEROS_ENABLED_SPRING_PROPERTY, Boolean.class, false)) { - producerConfig.put("security.protocol", environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY)); + producerConfig.put("security.protocol", KafkaUtils.INSTANCE.normalizeProtocol(environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY))); } return producerConfig; } + + @Bean public KafkaProducer kafkaProducer() { return new KafkaProducer<>(producerProperties()); diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java index 463c925267..fff7390410 100644 --- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java +++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/StormCLIWrapper.java @@ -18,6 +18,7 @@ package org.apache.metron.rest.service.impl; import org.apache.commons.lang3.StringUtils; +import org.apache.metron.common.utils.KafkaUtils; import org.apache.metron.rest.MetronRestConstants; import org.apache.metron.rest.RestException; import org.slf4j.Logger; @@ -117,7 +118,7 @@ protected String[] getParserStartCommand(String name) { // kafka security protocol command.add( "-ksp"); - command.add( environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY)); + command.add(KafkaUtils.INSTANCE.normalizeProtocol(environment.getProperty(MetronRestConstants.KAFKA_SECURITY_PROTOCOL_SPRING_PROPERTY))); // extra topology options boolean kerberosEnabled = environment.getProperty(MetronRestConstants.KERBEROS_ENABLED_SPRING_PROPERTY, Boolean.class, false); diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java index a26a3bbd5e..db8acde16a 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java @@ -19,12 +19,14 @@ package org.apache.metron.common.utils; import com.fasterxml.jackson.core.type.TypeReference; +import com.google.common.base.Joiner; import com.google.common.base.Splitter; import com.google.common.collect.Iterables; import org.apache.curator.RetryPolicy; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.retry.ExponentialBackoffRetry; +import org.apache.kafka.common.protocol.SecurityProtocol; import java.util.ArrayList; import java.util.List; @@ -67,6 +69,23 @@ public List getBrokersFromZookeeper(CuratorFramework client) throws Exce return ret; } + public String normalizeProtocol(String protocol) { + if(protocol.equalsIgnoreCase("PLAINTEXTSASL") || protocol.equalsIgnoreCase("SASL_PLAINTEXT")) { + if(SecurityProtocol.getNames().contains("PLAINTEXTSASL")) { + return "PLAINTEXTSASL"; + } + else if(SecurityProtocol.getNames().contains("SASL_PLAINTEXT")) { + return "SASL_PLAINTEXT"; + } + else { + throw new IllegalStateException("Unable to find the appropriate SASL protocol, " + + "viable options are: " + Joiner.on(",").join(SecurityProtocol.getNames())); + } + } + else { + return protocol.trim(); + } + } /* The URL accepted is NOT a general URL, and is assumed to follow the format used by the Kafka structures in Zookeeper. See: https://cwiki.apache.org/confluence/display/KAFKA/Kafka+data+structures+in+Zookeeper diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java index c9187035f1..1039e56f58 100644 --- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java +++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java @@ -18,6 +18,7 @@ package org.apache.metron.parsers.topology; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.metron.common.utils.KafkaUtils; import org.apache.metron.parsers.topology.config.ValueSupplier; import org.apache.metron.storm.kafka.flux.SimpleStormKafkaBuilder; import org.apache.metron.storm.kafka.flux.SpoutConfiguration; @@ -163,7 +164,7 @@ private static StormKafkaSpout createKafkaSpout( String zkQuorum , inputTopic + "_parser" ); if(securityProtocol.isPresent()) { - kafkaSpoutConfigOptions.putIfAbsent("security.protocol", securityProtocol.get()); + kafkaSpoutConfigOptions.putIfAbsent("security.protocol", KafkaUtils.INSTANCE.normalizeProtocol(securityProtocol.get())); } return SimpleStormKafkaBuilder.create( inputTopic , zkQuorum From 2f553e85ca8cf41d35395c6e85d70aab7b39db09 Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 13:59:18 -0500 Subject: [PATCH 06/18] Isolate JSONUtils.load from exposing Jackson --- .../metron/maas/functions/MaaSFunctions.java | 4 +-- .../common/configuration/Configurations.java | 4 +-- .../GlobalConfigurationOperations.java | 4 +-- .../IndexingConfigurationOperations.java | 4 +-- .../configuration/IndexingConfigurations.java | 4 +-- .../apache/metron/common/utils/JSONUtils.java | 26 ++++++++++++++----- .../metron/common/utils/KafkaUtils.java | 5 ++-- .../configuration/StellarEnrichmentTest.java | 4 +-- .../metron/common/utils/JSONUtilsTest.java | 4 +-- .../ZKConfigurationsCacheIntegrationTest.java | 10 +++---- .../TransformFilterExtractorDecorator.java | 4 +-- .../nonbulk/geo/GeoEnrichmentLoader.java | 4 +-- .../bulk/ElasticsearchImportExport.java | 4 +-- .../ElasticsearchUpdateIntegrationTest.java | 7 +++-- .../components/ElasticSearchComponent.java | 4 +-- .../enrichment/cli/LatencySummarizer.java | 3 +-- .../EnrichmentIntegrationTest.java | 6 ++--- .../apache/metron/indexing/dao/HBaseDao.java | 3 +-- .../apache/metron/indexing/dao/IndexDao.java | 3 +-- .../metron/indexing/dao/update/Document.java | 4 +-- .../metron/indexing/dao/InMemoryDao.java | 3 +-- .../indexing/dao/SearchIntegrationTest.java | 6 ++--- .../HDFSIndexingIntegrationTest.java | 4 +-- .../integration/IndexingIntegrationTest.java | 3 +-- .../metron/parsers/bolt/ParserBolt.java | 4 +-- .../metron/parsers/json/JSONMapParser.java | 4 +-- .../parsers/topology/ParserTopologyCLI.java | 4 +-- .../topology/config/ConfigHandlers.java | 4 +-- .../parsers/integration/ParserDriver.java | 4 +-- .../WriterBoltIntegrationTest.java | 3 +-- .../benchmark/StellarMicrobenchmark.java | 4 +-- .../configuration/ConfigurationType.java | 4 +-- .../shell/DefaultStellarShellExecutor.java | 5 +--- .../common/shell/cli/StellarShell.java | 3 +-- .../stellar/common/utils/JSONUtils.java | 26 ++++++++++++++----- .../dsl/functions/StringFunctions.java | 5 ++-- .../stellar/common/utils/JSONUtilsTest.java | 4 +-- 37 files changed, 84 insertions(+), 117 deletions(-) diff --git a/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java b/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java index eacb64dc6f..a260f25062 100644 --- a/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java +++ b/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java @@ -17,7 +17,6 @@ */ package org.apache.metron.maas.functions; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import java.lang.invoke.MethodHandles; @@ -174,8 +173,7 @@ public Object apply(List args, Context context) throws ParseException { URL u = new URL(url + "/" + modelFunction); String results = RESTUtil.INSTANCE.getRESTJSONResults(u, modelArgs); - ret = JSONUtils.INSTANCE.load(results, new TypeReference>() { - }); + ret = JSONUtils.INSTANCE.load(results, JSONUtils.MAP_SUPPLIER); resultCache.put(cacheKey, ret); return ret; } catch (Exception e) { diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/Configurations.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/Configurations.java index ecf8a1b5f2..af421a97ab 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/Configurations.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/Configurations.java @@ -17,7 +17,6 @@ */ package org.apache.metron.common.configuration; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; @@ -61,8 +60,7 @@ public void updateGlobalConfig(byte[] data) throws IOException { } public void updateGlobalConfig(InputStream io) throws IOException { - Map globalConfig = JSONUtils.INSTANCE.load(io, new TypeReference>() { - }); + Map globalConfig = JSONUtils.INSTANCE.load(io, JSONUtils.MAP_SUPPLIER); updateGlobalConfig(globalConfig); } diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/GlobalConfigurationOperations.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/GlobalConfigurationOperations.java index 4842c9170a..581ec3ef0c 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/GlobalConfigurationOperations.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/GlobalConfigurationOperations.java @@ -18,7 +18,6 @@ package org.apache.metron.common.configuration; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.IOException; import java.util.Map; import org.apache.curator.framework.CuratorFramework; @@ -38,8 +37,7 @@ public String getDirectory() { @Override public Object deserialize(String s) throws IOException { - return JSONUtils.INSTANCE.load(s, new TypeReference>() { - }); + return JSONUtils.INSTANCE.load(s, JSONUtils.MAP_SUPPLIER); } @Override diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurationOperations.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurationOperations.java index a75c84ecee..7884e4da19 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurationOperations.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurationOperations.java @@ -18,7 +18,6 @@ package org.apache.metron.common.configuration; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.IOException; import java.util.Map; import org.apache.curator.framework.CuratorFramework; @@ -33,8 +32,7 @@ public String getTypeName() { @Override public Object deserialize(String s) throws IOException { - return JSONUtils.INSTANCE.load(s, new TypeReference>() { - }); + return JSONUtils.INSTANCE.load(s, JSONUtils.MAP_SUPPLIER); } @Override diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java index 003b6df6e9..7ef9f00785 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/IndexingConfigurations.java @@ -17,7 +17,6 @@ */ package org.apache.metron.common.configuration; -import com.fasterxml.jackson.core.type.TypeReference; import org.apache.metron.stellar.common.utils.ConversionUtils; import org.apache.metron.common.utils.JSONUtils; @@ -77,8 +76,7 @@ public void updateSensorIndexingConfig(String sensorType, byte[] data) throws IO } public void updateSensorIndexingConfig(String sensorType, InputStream io) throws IOException { - Map sensorIndexingConfig = JSONUtils.INSTANCE.load(io, new TypeReference>() { - }); + Map sensorIndexingConfig = JSONUtils.INSTANCE.load(io, JSONUtils.MAP_SUPPLIER); updateSensorIndexingConfig(sensorType, sensorIndexingConfig); } diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java index 280b167b2f..c413fccec1 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java @@ -29,6 +29,10 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + import org.json.simple.JSONObject; import org.json.simple.parser.JSONParser; import org.json.simple.parser.ParseException; @@ -36,6 +40,16 @@ public enum JSONUtils { INSTANCE; + public static class ReferenceSupplier implements Supplier> { + @Override + public TypeReference get() { + return new TypeReference() { }; + } + } + + public final static ReferenceSupplier> MAP_SUPPLIER = new ReferenceSupplier<>(); + public final static ReferenceSupplier> LIST_SUPPLIER = new ReferenceSupplier<>(); + private static ThreadLocal _parser = ThreadLocal.withInitial(() -> new JSONParser()); @@ -51,17 +65,17 @@ public ObjectMapper getMapper() { } - public T load(InputStream is, TypeReference ref) throws IOException { - return _mapper.get().readValue(is, ref); + public T load(InputStream is, ReferenceSupplier ref) throws IOException { + return _mapper.get().readValue(is, ref.get()); } - public T load(String is, TypeReference ref) throws IOException { - return _mapper.get().readValue(is, ref); + public T load(String is, ReferenceSupplier ref) throws IOException { + return _mapper.get().readValue(is, ref.get()); } - public T load(File f, TypeReference ref) throws IOException { + public T load(File f, ReferenceSupplier ref) throws IOException { try (InputStream is = new BufferedInputStream(new FileInputStream(f))) { - return _mapper.get().readValue(is, ref); + return _mapper.get().readValue(is, ref.get()); } } diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java index a26a3bbd5e..d54e2b8ae5 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java @@ -18,7 +18,7 @@ package org.apache.metron.common.utils; -import com.fasterxml.jackson.core.type.TypeReference; +import com.google.common.base.Joiner; import com.google.common.base.Splitter; import com.google.common.collect.Iterables; import org.apache.curator.RetryPolicy; @@ -48,8 +48,7 @@ public List getBrokersFromZookeeper(CuratorFramework client) throws Exce for(String id : client.getChildren().forPath("/brokers/ids")) { byte[] data = client.getData().forPath("/brokers/ids/" + id); String brokerInfoStr = new String(data); - Map brokerInfo = JSONUtils.INSTANCE.load(brokerInfoStr, new TypeReference>() { - }); + Map brokerInfo = JSONUtils.INSTANCE.load(brokerInfoStr, JSONUtils.MAP_SUPPLIER); String host = (String) brokerInfo.get("host"); if(host != null) { ret.add(host + ":" + brokerInfo.get("port")); diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/StellarEnrichmentTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/StellarEnrichmentTest.java index 8eb9bbca0a..60a7d82dcb 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/StellarEnrichmentTest.java +++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/configuration/StellarEnrichmentTest.java @@ -17,7 +17,6 @@ */ package org.apache.metron.common.configuration; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.collect.ImmutableList; import org.adrianwalker.multilinestring.Multiline; import org.apache.metron.common.utils.JSONUtils; @@ -221,8 +220,7 @@ public class StellarEnrichmentTest { public static String message; public static JSONObject getMessage() throws IOException { - Map ret = JSONUtils.INSTANCE.load(message, new TypeReference>() { - }); + Map ret = JSONUtils.INSTANCE.load(message, JSONUtils.MAP_SUPPLIER); return new JSONObject(ret); } diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java index 7f4846e259..aecfcfdfc5 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java +++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java @@ -21,7 +21,6 @@ import static org.hamcrest.CoreMatchers.equalTo; import static org.junit.Assert.assertThat; -import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import java.io.File; import java.io.IOException; @@ -56,8 +55,7 @@ public void loads_file_with_typeref() throws Exception { put("b", "world"); }}; Map actual = JSONUtils.INSTANCE - .load(configFile, new TypeReference>() { - }); + .load(configFile, JSONUtils.MAP_SUPPLIER); assertThat("config not equal", actual, equalTo(expected)); } diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/zookeeper/ZKConfigurationsCacheIntegrationTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/zookeeper/ZKConfigurationsCacheIntegrationTest.java index 64bf986285..ec4a98a2c0 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/zookeeper/ZKConfigurationsCacheIntegrationTest.java +++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/zookeeper/ZKConfigurationsCacheIntegrationTest.java @@ -17,7 +17,6 @@ */ package org.apache.metron.common.zookeeper; -import com.fasterxml.jackson.core.type.TypeReference; import org.adrianwalker.multilinestring.Multiline; import org.apache.commons.io.IOUtils; import org.apache.curator.framework.CuratorFramework; @@ -226,14 +225,14 @@ public void validateUpdate() throws Exception { ConfigurationsUtils.writeProfilerConfigToZookeeper( profilerConfig.getBytes(), client); //indexing { - Map expectedConfig = JSONUtils.INSTANCE.load(testIndexingConfig, new TypeReference>() {}); + Map expectedConfig = JSONUtils.INSTANCE.load(testIndexingConfig, JSONUtils.MAP_SUPPLIER); IndexingConfigurations config = cache.get( IndexingConfigurations.class); assertEventually(() -> Assert.assertEquals(expectedConfig, config.getSensorIndexingConfig("test"))); } //enrichment { SensorEnrichmentConfig expectedConfig = JSONUtils.INSTANCE.load(testEnrichmentConfig, SensorEnrichmentConfig.class); - Map expectedGlobalConfig = JSONUtils.INSTANCE.load(globalConfig, new TypeReference>() {}); + Map expectedGlobalConfig = JSONUtils.INSTANCE.load(globalConfig, JSONUtils.MAP_SUPPLIER); EnrichmentConfigurations config = cache.get( EnrichmentConfigurations.class); assertEventually(() -> Assert.assertEquals(expectedConfig, config.getSensorEnrichmentConfig("test"))); assertEventually(() -> Assert.assertEquals(expectedGlobalConfig, config.getGlobalConfig())); @@ -255,12 +254,11 @@ public void validateUpdate() throws Exception { @Test public void validateBaseWrite() throws Exception { File globalConfigFile = new File(TestConstants.SAMPLE_CONFIG_PATH + "/global.json"); - Map expectedGlobalConfig = JSONUtils.INSTANCE.load(globalConfigFile, new TypeReference>() { }); + Map expectedGlobalConfig = JSONUtils.INSTANCE.load(globalConfigFile, JSONUtils.MAP_SUPPLIER); //indexing { File inFile = new File(TestConstants.SAMPLE_CONFIG_PATH + "/indexing/test.json"); - Map expectedConfig = JSONUtils.INSTANCE.load(inFile, new TypeReference>() { - }); + Map expectedConfig = JSONUtils.INSTANCE.load(inFile, JSONUtils.MAP_SUPPLIER); IndexingConfigurations config = cache.get( IndexingConfigurations.class); assertEventually(() -> Assert.assertEquals(expectedConfig, config.getSensorIndexingConfig("test"))); assertEventually(() -> Assert.assertEquals(expectedGlobalConfig, config.getGlobalConfig())); diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/extractor/TransformFilterExtractorDecorator.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/extractor/TransformFilterExtractorDecorator.java index c47dfc61bd..b665a019df 100644 --- a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/extractor/TransformFilterExtractorDecorator.java +++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/extractor/TransformFilterExtractorDecorator.java @@ -17,7 +17,6 @@ */ package org.apache.metron.dataloads.extractor; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.ByteArrayInputStream; import java.io.IOException; import java.lang.invoke.MethodHandles; @@ -208,8 +207,7 @@ private Map getGlobalConfig(Optional zkClient) try { return JSONUtils.INSTANCE.load( new ByteArrayInputStream(ConfigurationsUtils.readGlobalConfigBytesFromZookeeper(zkClient.get())), - new TypeReference>() { - }); + JSONUtils.MAP_SUPPLIER); } catch (Exception e) { LOG.warn("Exception thrown while attempting to get global config from Zookeeper.", e); } diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/geo/GeoEnrichmentLoader.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/geo/GeoEnrichmentLoader.java index 322edecb35..b3660158aa 100644 --- a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/geo/GeoEnrichmentLoader.java +++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/geo/GeoEnrichmentLoader.java @@ -18,7 +18,6 @@ package org.apache.metron.dataloads.nonbulk.geo; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Function; import com.google.common.base.Joiner; import org.apache.commons.cli.*; @@ -200,8 +199,7 @@ protected void pushConfig(Path srcPath, Path dstPath, String zookeeper) { // Fetch the global configuration Map global = JSONUtils.INSTANCE.load( new ByteArrayInputStream(ConfigurationsUtils.readGlobalConfigBytesFromZookeeper(client)), - new TypeReference>() { - }); + JSONUtils.MAP_SUPPLIER); // Update the global config and push it back global.put(GeoLiteDatabase.GEO_HDFS_FILE, dstPath.toString() + "/" + srcPath.getName()); diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/bulk/ElasticsearchImportExport.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/bulk/ElasticsearchImportExport.java index 0a04dfc724..77e0d61c2c 100644 --- a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/bulk/ElasticsearchImportExport.java +++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/bulk/ElasticsearchImportExport.java @@ -18,7 +18,6 @@ package org.apache.metron.elasticsearch.bulk; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.FileReader; @@ -71,8 +70,7 @@ public void bulkify(Path input, Path output) throws IOException { String line; while ((line = br.readLine()) != null) { Map inDoc = JSONUtils.INSTANCE - .load(line, new TypeReference>() { - }); + .load(line, JSONUtils.MAP_SUPPLIER); Object id = inDoc.get("_id"); Object type = inDoc.get("_type"); String createRaw = String diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchUpdateIntegrationTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchUpdateIntegrationTest.java index 116ee4b2a3..325d42e668 100644 --- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchUpdateIntegrationTest.java +++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchUpdateIntegrationTest.java @@ -18,7 +18,6 @@ package org.apache.metron.elasticsearch.integration; import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.collect.Iterables; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; @@ -158,7 +157,7 @@ public void test() throws Exception { Assert.assertEquals(1, columns.size()); Assert.assertEquals(message0 , JSONUtils.INSTANCE.load(new String(columns.lastEntry().getValue()) - , new TypeReference>() {}) + , JSONUtils.MAP_SUPPLIER) ); } { @@ -195,10 +194,10 @@ public void test() throws Exception { NavigableMap columns = r.getFamilyMap(CF.getBytes()); Assert.assertEquals(2, columns.size()); Assert.assertEquals(message0, JSONUtils.INSTANCE.load(new String(columns.lastEntry().getValue()) - , new TypeReference>() {}) + , JSONUtils.MAP_SUPPLIER) ); Assert.assertNotEquals(message0, JSONUtils.INSTANCE.load(new String(columns.firstEntry().getValue()) - , new TypeReference>() {}) + , JSONUtils.MAP_SUPPLIER) ); } { diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java index f95ef215f9..e716ce1114 100644 --- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java +++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java @@ -19,7 +19,6 @@ import static java.util.Arrays.asList; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.File; import java.io.IOException; import java.util.ArrayList; @@ -210,8 +209,7 @@ public BulkResponse add(String indexName, String sensorType, Iterable do indexRequestBuilder = indexRequestBuilder.setSource(doc); Map esDoc = JSONUtils.INSTANCE - .load(doc, new TypeReference>() { - }); + .load(doc, JSONUtils.MAP_SUPPLIER); indexRequestBuilder.setId((String) esDoc.get(Constants.GUID)); Object ts = esDoc.get("timestamp"); if (ts != null) { diff --git a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/cli/LatencySummarizer.java b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/cli/LatencySummarizer.java index b40f2ad8ed..2e5e8bf63d 100644 --- a/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/cli/LatencySummarizer.java +++ b/metron-platform/metron-enrichment/src/main/java/org/apache/metron/enrichment/cli/LatencySummarizer.java @@ -17,7 +17,6 @@ */ package org.apache.metron.enrichment.cli; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Joiner; import com.google.common.base.Splitter; import com.google.common.collect.Iterables; @@ -181,7 +180,7 @@ public static void main(String... argv) throws IOException { LatencyStats statsMap = new LatencyStats(); BufferedReader reader = new BufferedReader(new InputStreamReader(System.in)); for(String line = null;(line = reader.readLine()) != null;) { - Map doc = JSONUtils.INSTANCE.load(line, new TypeReference>() {}); + Map doc = JSONUtils.INSTANCE.load(line, JSONUtils.MAP_SUPPLIER); updateStats(statsMap, doc); } statsMap.printSummary(cmd.hasOption('m')); diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java index c457e8617a..828f4e3825 100644 --- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java +++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java @@ -24,7 +24,6 @@ import static org.apache.metron.enrichment.bolt.ThreatIntelJoinBolt.THREAT_TRIAGE_RULE_SCORE; import static org.apache.metron.enrichment.bolt.ThreatIntelJoinBolt.THREAT_TRIAGE_SCORE_KEY; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Function; import com.google.common.base.Joiner; import com.google.common.base.Predicate; @@ -163,8 +162,7 @@ public void test() throws Exception { String globalConfigStr = null; { File globalConfig = new File(new File(TestConstants.SAMPLE_CONFIG_PATH), "global.json"); - Map config = JSONUtils.INSTANCE.load(globalConfig, new TypeReference>() { - }); + Map config = JSONUtils.INSTANCE.load(globalConfig, JSONUtils.MAP_SUPPLIER); config.put(SimpleHBaseEnrichmentFunctions.TABLE_PROVIDER_TYPE_CONF, MockHBaseTableProvider.class.getName()); config.put(SimpleHBaseEnrichmentFunctions.ACCESS_TRACKER_TYPE_CONF, "PERSISTENT_BLOOM"); config.put(PersistentBloomTrackerCreator.Config.PERSISTENT_BLOOM_TABLE, trackerHBaseTableName); @@ -531,7 +529,7 @@ private static List> loadMessages(List outputMessage , message -> { try { return new HashMap<>(JSONUtils.INSTANCE.load(new String(message) - , new TypeReference>() {} + , JSONUtils.MAP_SUPPLIER ) ); } catch (Exception ex) { 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 index 72f2980b10..ebb9907951 100644 --- 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 @@ -18,7 +18,6 @@ package org.apache.metron.indexing.dao; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.DataInputStream; @@ -210,7 +209,7 @@ private Document getDocumentFromResult(Result result) throws IOException { Long ts = Bytes.toLong(entry.getKey()); if(entry.getValue()!= null) { Map json = JSONUtils.INSTANCE.load(new String(entry.getValue()), - new TypeReference>() {}); + JSONUtils.MAP_SUPPLIER); try { Key k = Key.fromBytes(result.getRow()); return new Document(json, k.getGuid(), k.getSensorType(), ts); 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 03d348a661..5fbba02fba 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,7 +17,6 @@ */ package org.apache.metron.indexing.dao; -import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import java.io.IOException; import java.util.List; @@ -143,7 +142,7 @@ default Document getPatchedDocument(PatchRequest request JsonNode originalNode = JSONUtils.INSTANCE.convert(latest, JsonNode.class); JsonNode patched = JSONUtils.INSTANCE.applyPatch(request.getPatch(), originalNode); Map updated = JSONUtils.INSTANCE.getMapper() - .convertValue(patched, new TypeReference>() {}); + .convertValue(patched, JSONUtils.MAP_SUPPLIER); return new Document( updated , request.getGuid() , request.getSensorType() 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 index b5fcb025f7..6f2f7797ff 100644 --- 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 @@ -18,7 +18,6 @@ package org.apache.metron.indexing.dao.update; -import com.fasterxml.jackson.core.type.TypeReference; import org.apache.metron.common.utils.JSONUtils; import java.io.IOException; @@ -47,8 +46,7 @@ public Document(String document, String guid, String sensorType) throws IOExcept } private static Map convertDoc(String document) throws IOException { - return JSONUtils.INSTANCE.load(document, new TypeReference>() { - }); + return JSONUtils.INSTANCE.load(document, JSONUtils.MAP_SUPPLIER); } public String getSensorType() { 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 002ec28dbd..79c3bb31a7 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 @@ -17,7 +17,6 @@ */ package org.apache.metron.indexing.dao; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Splitter; import com.google.common.collect.ComparisonChain; import com.google.common.collect.Iterables; @@ -193,7 +192,7 @@ private static boolean isMatch(String query, Map doc) { public static Map parse(String doc) { try { - return JSONUtils.INSTANCE.load(doc, new TypeReference>() {}); + return JSONUtils.INSTANCE.load(doc, JSONUtils.MAP_SUPPLIER); } catch (IOException e) { throw new IllegalStateException(e.getMessage(), e); } diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java index f381688e7e..14b985835c 100644 --- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java +++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java @@ -18,7 +18,6 @@ */ package org.apache.metron.indexing.dao; -import com.fasterxml.jackson.core.type.TypeReference; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -464,8 +463,7 @@ public void find_one_guid() throws Exception { @Test public void get_all_latest_guid() throws Exception { - List request = JSONUtils.INSTANCE.load(getAllLatestQuery, new TypeReference>() { - }); + List request = JSONUtils.INSTANCE.load(getAllLatestQuery, new ReferenceSupplier>()); Map docs = new HashMap<>(); for(Document doc : dao.getAllLatest(request)) { @@ -945,4 +943,4 @@ public static void stop() throws Exception { protected abstract IndexDao createDao() throws Exception; protected abstract InMemoryComponent startIndex() throws Exception; protected abstract void loadTestData() throws Exception; -} \ No newline at end of file +} diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/HDFSIndexingIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/HDFSIndexingIntegrationTest.java index ab5cc3ff0f..b8af6a3eee 100644 --- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/HDFSIndexingIntegrationTest.java +++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/HDFSIndexingIntegrationTest.java @@ -18,7 +18,6 @@ package org.apache.metron.indexing.integration; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Function; import com.google.common.collect.Iterables; import org.apache.metron.common.interfaces.FieldNameConverter; @@ -75,8 +74,7 @@ public static List> readDocsFromDisk(String hdfsDirStr) thro Iterables.addAll(ret, Iterables.transform(data, bytes -> { String s = new String(bytes); try { - return JSONUtils.INSTANCE.load(s, new TypeReference>() { - }); + return JSONUtils.INSTANCE.load(s,JSONUtils.MAP_SUPPLIER); } catch (IOException e) { throw new RuntimeException(e); } diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java index b0b6cc2b57..1671ab3273 100644 --- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java +++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java @@ -18,7 +18,6 @@ package org.apache.metron.indexing.integration; -import com.fasterxml.jackson.core.type.TypeReference; import org.apache.curator.framework.CuratorFramework; import org.apache.metron.TestConstants; import org.apache.metron.common.Constants; @@ -81,7 +80,7 @@ public void test() throws Exception { }}); List> inputDocs = new ArrayList<>(); for(byte[] b : inputMessages) { - Map m = JSONUtils.INSTANCE.load(new String(b), new TypeReference>() {}); + Map m = JSONUtils.INSTANCE.load(new String(b), JSONUtils.MAP_SUPPLIER); inputDocs.add(m); } diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java index 99785b2d41..6fc4ed7c7c 100644 --- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java +++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java @@ -19,7 +19,6 @@ import static org.apache.metron.common.Constants.METADATA_PREFIX; -import com.fasterxml.jackson.core.type.TypeReference; import java.io.IOException; import java.io.Serializable; import java.lang.invoke.MethodHandles; @@ -146,8 +145,7 @@ private Map getMetadata(Tuple t, boolean readMetadata) { try { keyStr = keyObj == null?null:new String(keyObj); if(!StringUtils.isEmpty(keyStr)) { - Map metadata = JSONUtils.INSTANCE.load(keyStr, new TypeReference>() { - }); + Map metadata = JSONUtils.INSTANCE.load(keyStr,JSONUtils.MAP_SUPPLIER); for(Map.Entry kv : metadata.entrySet()) { ret.put(METADATA_PREFIX + kv.getKey(), kv.getValue()); } diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/json/JSONMapParser.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/json/JSONMapParser.java index 5d824561dc..7e5468faba 100644 --- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/json/JSONMapParser.java +++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/json/JSONMapParser.java @@ -17,7 +17,6 @@ */ package org.apache.metron.parsers.json; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Joiner; import com.google.common.collect.ImmutableList; import org.apache.metron.common.utils.JSONUtils; @@ -96,8 +95,7 @@ public List parse(byte[] rawMessage) { try { String originalString = new String(rawMessage); //convert the JSON blob into a String -> Object map - Map rawMap = JSONUtils.INSTANCE.load(originalString, new TypeReference>() { - }); + Map rawMap = JSONUtils.INSTANCE.load(originalString, JSONUtils.MAP_SUPPLIER); JSONObject ret = normalizeJSON(rawMap); ret.put("original_string", originalString ); if(!ret.containsKey("timestamp")) { diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java index 4ce05083ce..3824212b7d 100644 --- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java +++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java @@ -24,7 +24,6 @@ import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; import org.apache.storm.utils.Utils; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Joiner; import org.apache.commons.cli.*; import org.apache.commons.io.FileUtils; @@ -492,8 +491,7 @@ private static Map readJSONMapFromFile(File inputFile) { throw new IllegalArgumentException("Unable to load JSON file at " + inputFile.getAbsolutePath()); } try { - return JSONUtils.INSTANCE.load(json, new TypeReference>() { - }); + return JSONUtils.INSTANCE.load(json, JSONUtils.MAP_SUPPLIER); } catch (IOException e) { throw new IllegalStateException("Unable to process JSON.", e); } diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ConfigHandlers.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ConfigHandlers.java index f6c04a6c5d..8d98922888 100644 --- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ConfigHandlers.java +++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/config/ConfigHandlers.java @@ -19,7 +19,6 @@ package org.apache.metron.parsers.topology.config; import org.apache.storm.Config; -import com.fasterxml.jackson.core.type.TypeReference; import org.apache.commons.io.FileUtils; import org.apache.metron.common.utils.JSONUtils; @@ -81,8 +80,7 @@ public Config apply(Arg arg) { json = arg.getArg(); } try { - arg.getConfig().putAll(JSONUtils.INSTANCE.load(json, new TypeReference>() { - })); + arg.getConfig().putAll(JSONUtils.INSTANCE.load(json, JSONUtils.MAP_SUPPLIER)); } catch (IOException e) { throw new IllegalStateException("Unable to process JSON snippet.", e); } diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java index b844104026..deb0217901 100644 --- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java +++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserDriver.java @@ -17,7 +17,6 @@ */ package org.apache.metron.parsers.integration; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.collect.ImmutableList; import org.apache.metron.common.configuration.ConfigurationsUtils; import org.apache.metron.common.configuration.FieldValidator; @@ -147,8 +146,7 @@ public ProcessorResult> getResults() { public ParserDriver(String sensorType, String parserConfig, String globalConfig) throws IOException { config = SensorParserConfig.fromBytes(parserConfig.getBytes()); this.sensorType = sensorType; - this.globalConfig = JSONUtils.INSTANCE.load(globalConfig, new TypeReference>() { - }); + this.globalConfig = JSONUtils.INSTANCE.load(globalConfig, JSONUtils.MAP_SUPPLIER); } public ProcessorResult> run(List in) { diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java index d4d1c02d86..0cfaae3ca2 100644 --- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java +++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java @@ -17,7 +17,6 @@ */ package org.apache.metron.writers.integration; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Function; import com.google.common.collect.Iterables; import org.adrianwalker.multilinestring.Multiline; @@ -171,7 +170,7 @@ private static List loadMessages(List outputMessages) { , message -> { try { return new JSONObject(JSONUtils.INSTANCE.load(new String(message) - , new TypeReference>() {} + ,JSONUtils.MAP_SUPPLIER ) ); } catch (Exception ex) { diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/benchmark/StellarMicrobenchmark.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/benchmark/StellarMicrobenchmark.java index 35fa673ac3..914105ebfe 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/benchmark/StellarMicrobenchmark.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/benchmark/StellarMicrobenchmark.java @@ -17,7 +17,6 @@ */ package org.apache.metron.stellar.common.benchmark; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Joiner; import com.google.common.base.Splitter; import com.google.common.io.Files; @@ -231,8 +230,7 @@ public static void main(String... argv) throws IOException { List lines = Files.readLines(expressionsFile, Charset.defaultCharset()); Map variables = new HashMap<>(); if(variablesFile.isPresent()) { - variables = JSONUtils.INSTANCE.load(new FileInputStream(variablesFile.get()), new TypeReference>() { - }); + variables = JSONUtils.INSTANCE.load(new FileInputStream(variablesFile.get()), JSONUtils.MAP_SUPPLIER); } int numTimes = DEFAULT_NUM_TIMES; if(BenchmarkOptions.NUM_TIMES.has(cli)) { diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/configuration/ConfigurationType.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/configuration/ConfigurationType.java index 16cde833b4..99859b29ce 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/configuration/ConfigurationType.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/configuration/ConfigurationType.java @@ -18,7 +18,6 @@ package org.apache.metron.stellar.common.configuration; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Function; import org.apache.metron.stellar.common.Constants; import org.apache.metron.stellar.common.utils.JSONUtils; @@ -30,8 +29,7 @@ public enum ConfigurationType implements Function { GLOBAL("global",".", s -> { try { - return JSONUtils.INSTANCE.load(s, new TypeReference>() { - }); + return JSONUtils.INSTANCE.load(s, JSONUtils.MAP_SUPPLIER); } catch (IOException e) { throw new RuntimeException("Unable to load " + s, e); } diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/DefaultStellarShellExecutor.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/DefaultStellarShellExecutor.java index f83bb9eb92..3f2c4958d7 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/DefaultStellarShellExecutor.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/DefaultStellarShellExecutor.java @@ -19,7 +19,6 @@ */ package org.apache.metron.stellar.common.shell; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.collect.Maps; import org.apache.commons.collections.map.UnmodifiableMap; import org.apache.commons.lang3.StringUtils; @@ -370,9 +369,7 @@ private Context createContext(Properties properties, Optional */ private Map fetchGlobalConfig(CuratorFramework zkClient) throws Exception { byte[] raw = readGlobalConfigBytesFromZookeeper(zkClient); - return JSONUtils.INSTANCE.load( - new ByteArrayInputStream(raw), - new TypeReference>() {}); + return JSONUtils.INSTANCE.load( new ByteArrayInputStream(raw), JSONUtils.MAP_SUPPLIER); } /** diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java index 44ad28c347..c81df61aa3 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java @@ -20,7 +20,6 @@ package org.apache.metron.stellar.common.shell.cli; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Splitter; import com.google.common.collect.Iterables; import org.apache.commons.cli.CommandLine; @@ -221,7 +220,7 @@ private static void loadVariables( String variablePath = commandLine.getOptionValue("v"); Map variables = JSONUtils.INSTANCE.load( new File(variablePath), - new TypeReference>() {}); + JSONUtils.MAP_SUPPLIER); // for each variable... for(Map.Entry kv : variables.entrySet()) { diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java index d7e90cfb55..2a846aef12 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java @@ -27,6 +27,10 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + import org.json.simple.JSONObject; import org.json.simple.parser.JSONParser; import org.json.simple.parser.ParseException; @@ -35,23 +39,33 @@ public enum JSONUtils { INSTANCE; + public static class ReferenceSupplier implements Supplier> { + @Override + public TypeReference get() { + return new TypeReference() { }; + } + } + + public final static ReferenceSupplier> MAP_SUPPLIER = new ReferenceSupplier<>(); + public final static ReferenceSupplier> LIST_SUPPLIER = new ReferenceSupplier<>(); + private static ThreadLocal _parser = ThreadLocal.withInitial(() -> new JSONParser()); private static ThreadLocal _mapper = ThreadLocal.withInitial(() -> new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL)); - public T load(InputStream is, TypeReference ref) throws IOException { - return _mapper.get().readValue(is, ref); + public T load(InputStream is, ReferenceSupplier ref) throws IOException { + return _mapper.get().readValue(is, ref.get()); } - public T load(String is, TypeReference ref) throws IOException { - return _mapper.get().readValue(is, ref); + public T load(String is, ReferenceSupplier ref) throws IOException { + return _mapper.get().readValue(is, ref.get()); } - public T load(File f, TypeReference ref) throws IOException { + public T load(File f, ReferenceSupplier ref) throws IOException { try (InputStream is = new BufferedInputStream(new FileInputStream(f))) { - return _mapper.get().readValue(is, ref); + return _mapper.get().readValue(is, ref.get()); } } diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java index 4dc4790c9d..9c5880efb7 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java @@ -20,7 +20,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Joiner; import com.google.common.base.Splitter; import com.google.common.collect.Iterables; @@ -570,7 +569,7 @@ public Object apply(List strings) { } // Return parsed JSON Object as a HashMap try { - return JSONUtils.INSTANCE.load((String) strings.get(0), new TypeReference>(){}); + return JSONUtils.INSTANCE.load((String) strings.get(0), JSONUtils.MAP_SUPPLIER); } catch (JsonProcessingException ex) { throw new ParseException("Valid JSON string not supplied", ex); } catch (IOException e) { @@ -607,7 +606,7 @@ public Object apply(List strings) { } // Return parsed JSON Object as a List try { - return (List) JSONUtils.INSTANCE.load((String) strings.get(0), new TypeReference>(){}); + return (List) JSONUtils.INSTANCE.load((String) strings.get(0), JSONUtils.LIST_SUPPLIER); } catch (JsonProcessingException ex) { throw new ParseException("Valid JSON string not supplied", ex); } catch (IOException e) { diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/common/utils/JSONUtilsTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/common/utils/JSONUtilsTest.java index f4692e39f7..e28bfb7fce 100644 --- a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/common/utils/JSONUtilsTest.java +++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/common/utils/JSONUtilsTest.java @@ -17,7 +17,6 @@ */ package org.apache.metron.stellar.common.utils; -import com.fasterxml.jackson.core.type.TypeReference; import org.adrianwalker.multilinestring.Multiline; import org.junit.Assert; import org.junit.BeforeClass; @@ -54,8 +53,7 @@ public void loads_file_with_typeref() throws Exception { put("a", "hello"); put("b", "world"); }}; - Map actual = JSONUtils.INSTANCE.load(configFile, new TypeReference>() { - }); + Map actual = JSONUtils.INSTANCE.load(configFile, JSONUtils.MAP_SUPPLIER); Assert.assertThat("config not equal", actual, equalTo(expected)); } From 9c8b3bd821f8feb2f4d39ca640706ebb7a89dbe6 Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 14:32:43 -0500 Subject: [PATCH 07/18] Updating JSONUtils to avoid exposing jackson --- .../configuration/ConfigurationsUtils.java | 5 +---- .../apache/metron/common/utils/JSONUtils.java | 14 ++++++++------ .../metron/common/utils/JSONUtilsTest.java | 10 ++++------ .../org/apache/metron/indexing/dao/IndexDao.java | 16 ++++++++++++---- .../indexing/dao/SearchIntegrationTest.java | 2 +- 5 files changed, 26 insertions(+), 21 deletions(-) diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java index 5183788816..c18d9f9281 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java @@ -596,10 +596,7 @@ public static void applyConfigPatchToZookeeper( byte[] patchData, CuratorFramework client) throws Exception { byte[] configData = readConfigBytesFromZookeeper(configurationType, configName, client); - JsonNode source = JSONUtils.INSTANCE.readTree(configData); - JsonNode patch = JSONUtils.INSTANCE.readTree(patchData); - JsonNode patchedConfig = JSONUtils.INSTANCE.applyPatch(patch, source); - byte[] prettyPatchedConfig = JSONUtils.INSTANCE.toJSONPretty(patchedConfig); + byte[] prettyPatchedConfig = JSONUtils.INSTANCE.applyPatch(configData, patchData); // ensure the patch produces a valid result; otherwise exception thrown during deserialization String prettyPatchedConfigStr = new String(prettyPatchedConfig); diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java index c413fccec1..f76019aa51 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java @@ -122,7 +122,7 @@ public JSONObject toJSONObject(Object o) throws JsonProcessingException, ParseEx * @param json JSON value to deserialize * @return deserialized JsonNode Object */ - public JsonNode readTree(String json) throws IOException { + JsonNode readTree(String json) throws IOException { return _mapper.get().readTree(json); } @@ -132,7 +132,7 @@ public JsonNode readTree(String json) throws IOException { * @param json JSON value to deserialize * @return deserialized JsonNode Object */ - public JsonNode readTree(byte[] json) throws IOException { + JsonNode readTree(byte[] json) throws IOException { return _mapper.get().readTree(json); } @@ -152,14 +152,16 @@ public JsonNode readTree(byte[] json) throws IOException { * @param source Source JSON to apply patch to * @return new json after applying the patch */ - public JsonNode applyPatch(String patch, String source) throws IOException { + public byte[] applyPatch(String patch, String source) throws IOException { JsonNode patchNode = readTree(patch); JsonNode sourceNode = readTree(source); - return applyPatch(patchNode, sourceNode); + return toJSONPretty(JsonPatch.apply(patchNode, sourceNode)); } - public JsonNode applyPatch(JsonNode patch, JsonNode source) throws IOException { - return JsonPatch.apply(patch, source); + public byte[] applyPatch(byte[] patch, byte[] source) throws IOException { + JsonNode patchNode = readTree(patch); + JsonNode sourceNode = readTree(source); + return toJSONPretty(JsonPatch.apply(patchNode, sourceNode)); } } diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java b/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java index aecfcfdfc5..46765913c9 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java +++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/JSONUtilsTest.java @@ -121,9 +121,8 @@ public TestConfig setB(String b) { @Test public void applyPatch_modifies_source_json_doc() throws IOException { - JsonNode actual = JSONUtils.INSTANCE.applyPatch(patchJson, sourceJson); - JsonNode expected = JSONUtils.INSTANCE.readTree(expectedJson); - assertThat(actual, equalTo(expected)); + String actual = new String(JSONUtils.INSTANCE.applyPatch(patchJson, sourceJson)); + assertThat(JSONUtils.INSTANCE.load(actual, JSONUtils.MAP_SUPPLIER), equalTo(JSONUtils.INSTANCE.load(expectedJson, JSONUtils.MAP_SUPPLIER))); } /** @@ -158,9 +157,8 @@ public void applyPatch_modifies_source_json_doc() throws IOException { @Test public void applyPatch_modifies_complex_source_json_doc() throws IOException { - JsonNode actual = JSONUtils.INSTANCE.applyPatch(patchComplexJson, complexJson); - JsonNode expected = JSONUtils.INSTANCE.readTree(expectedComplexJson); - assertThat(actual, equalTo(expected)); + String actual = new String(JSONUtils.INSTANCE.applyPatch(patchComplexJson, complexJson)); + assertThat(JSONUtils.INSTANCE.load(actual, JSONUtils.MAP_SUPPLIER), equalTo(JSONUtils.INSTANCE.load(expectedComplexJson, JSONUtils.MAP_SUPPLIER))); } } 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 5fbba02fba..2c057d481a 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,11 +17,16 @@ */ package org.apache.metron.indexing.dao; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import java.io.IOException; import java.util.List; import java.util.Map; import java.util.Optional; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.flipkart.zjsonpatch.JsonPatch; import org.apache.metron.common.utils.JSONUtils; import org.apache.metron.indexing.dao.search.FieldType; import org.apache.metron.indexing.dao.search.GetRequest; @@ -41,6 +46,9 @@ */ public interface IndexDao { + public static ThreadLocal _mapper = ThreadLocal.withInitial(() -> + new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL)); + /** * Return search response based on the search request * @@ -139,10 +147,10 @@ default Document getPatchedDocument(PatchRequest request 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 = JSONUtils.INSTANCE.applyPatch(request.getPatch(), originalNode); - Map updated = JSONUtils.INSTANCE.getMapper() - .convertValue(patched, JSONUtils.MAP_SUPPLIER); + JsonNode originalNode = _mapper.get().convertValue(latest, JsonNode.class); + JsonNode patched = JsonPatch.apply(request.getPatch(), originalNode); + Map updated = _mapper.get() + .convertValue(patched, new TypeReference>() {}); return new Document( updated , request.getGuid() , request.getSensorType() diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java index 14b985835c..ae648a3628 100644 --- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java +++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java @@ -463,7 +463,7 @@ public void find_one_guid() throws Exception { @Test public void get_all_latest_guid() throws Exception { - List request = JSONUtils.INSTANCE.load(getAllLatestQuery, new ReferenceSupplier>()); + List request = JSONUtils.INSTANCE.load(getAllLatestQuery, new JSONUtils.ReferenceSupplier<>()); Map docs = new HashMap<>(); for(Document doc : dao.getAllLatest(request)) { From 55b641a910454dc6a9561c2bd69b1e59302334de Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 15:37:00 -0500 Subject: [PATCH 08/18] Updating string functions to be more explicit --- .../apache/metron/stellar/dsl/functions/StringFunctions.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java index 9c5880efb7..7c87973667 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java @@ -610,9 +610,11 @@ public Object apply(List strings) { } catch (JsonProcessingException ex) { throw new ParseException("Valid JSON string not supplied", ex); } catch (IOException e) { - e.printStackTrace(); throw new ParseException("Valid JSON string not supplied", e); } + catch (ClassCastException e) { + throw new ParseException("JSON String does not represent a list", e); + } } } } From 6d4c2c7267962f2c77e6f4005454a3d9d30befca Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 16:23:17 -0500 Subject: [PATCH 09/18] Fixed tests. --- .../dsl/functions/StringFunctions.java | 26 +++++++++++-------- .../dsl/functions/StringFunctionsTest.java | 9 ++++--- 2 files changed, 20 insertions(+), 15 deletions(-) diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java index 7c87973667..17566d2c6e 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java @@ -568,15 +568,18 @@ public Object apply(List strings) { throw new ParseException("Valid JSON string not supplied"); } // Return parsed JSON Object as a HashMap + String in = (String)strings.get(0); try { - return JSONUtils.INSTANCE.load((String) strings.get(0), JSONUtils.MAP_SUPPLIER); + return (Map)JSONUtils.INSTANCE.load(in, JSONUtils.MAP_SUPPLIER); } catch (JsonProcessingException ex) { - throw new ParseException("Valid JSON string not supplied", ex); - } catch (IOException e) { - e.printStackTrace(); + throw new ParseException(String.format("{} is not a valid JSON string", in), ex); + } catch (IOException ex) { + throw new ParseException(String.format("{} is not a valid JSON string", in), ex); + } + catch (ClassCastException ex) { + throw new ParseException(String.format("{} is not a valid JSON string, expected a map", in), ex); } } - return new ParseException("Unable to parse JSON string"); } } @@ -605,15 +608,16 @@ public Object apply(List strings) { throw new ParseException("Valid JSON string not supplied"); } // Return parsed JSON Object as a List + String in = (String)strings.get(0); try { - return (List) JSONUtils.INSTANCE.load((String) strings.get(0), JSONUtils.LIST_SUPPLIER); + return (List) JSONUtils.INSTANCE.load(in, JSONUtils.LIST_SUPPLIER); } catch (JsonProcessingException ex) { - throw new ParseException("Valid JSON string not supplied", ex); - } catch (IOException e) { - throw new ParseException("Valid JSON string not supplied", e); + throw new ParseException(String.format("{} is not a valid JSON string", in), ex); + } catch (IOException ex) { + throw new ParseException(String.format("{} is not a valid JSON string", in), ex); } - catch (ClassCastException e) { - throw new ParseException("JSON String does not represent a list", e); + catch (ClassCastException ex) { + throw new ParseException(String.format("{} is not a valid JSON string, expected a list", in), ex); } } } diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java index 418bf2de5b..9c5db15bd3 100644 --- a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java +++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java @@ -617,7 +617,8 @@ public void testToJsonMap() throws Exception { //Simple Arrays boolean thrown = false; try { - run("TO_JSON_MAP(msg)", ImmutableMap.of("msg", string3)); + Object o = run("TO_JSON_MAP(msg)", ImmutableMap.of("msg", string3)); + System.out.println(string3 + " == " + o); } catch (ParseException pe) { thrown = true; } @@ -656,7 +657,7 @@ public void testToJsonMap() throws Exception { run("TO_JSON_MAP('123, 456')", new HashedMap<>()); } catch (ParseException pe) { thrown = true; - Assert.assertTrue(pe.getMessage().contains("Valid JSON string not supplied")); + Assert.assertTrue(pe.getMessage().contains("is not a valid JSON string")); } Assert.assertTrue(thrown); thrown = false; @@ -730,7 +731,7 @@ public void testToJsonList() throws Exception { run("TO_JSON_LIST('123, 456')", new HashedMap<>()); } catch (ParseException pe) { thrown = true; - Assert.assertTrue(pe.getMessage().contains("Valid JSON string not supplied")); + Assert.assertTrue(pe.getMessage().contains("is not a valid JSON string")); } Assert.assertTrue(thrown); @@ -740,7 +741,7 @@ public void testToJsonList() throws Exception { run("TO_JSON_LIST('{\"foo\" : 2')", new HashedMap<>()); } catch (ParseException pe) { thrown = true; - Assert.assertTrue(pe.getMessage().contains("Valid JSON string not supplied")); + Assert.assertTrue(pe.getMessage().contains("is not a valid JSON string")); } Assert.assertTrue(thrown); } From c1267c7a04ac59c187f19c540e6e9d9b6a63344b Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 16:26:44 -0500 Subject: [PATCH 10/18] updating format --- .../metron/stellar/dsl/functions/StringFunctions.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java index 17566d2c6e..d1a53b23f9 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java @@ -612,12 +612,12 @@ public Object apply(List strings) { try { return (List) JSONUtils.INSTANCE.load(in, JSONUtils.LIST_SUPPLIER); } catch (JsonProcessingException ex) { - throw new ParseException(String.format("{} is not a valid JSON string", in), ex); + throw new ParseException(String.format("%s is not a valid JSON string", in), ex); } catch (IOException ex) { - throw new ParseException(String.format("{} is not a valid JSON string", in), ex); + throw new ParseException(String.format("%s is not a valid JSON string", in), ex); } catch (ClassCastException ex) { - throw new ParseException(String.format("{} is not a valid JSON string, expected a list", in), ex); + throw new ParseException(String.format("%s is not a valid JSON string, expected a list", in), ex); } } } From babd82a0646b46a1ba246d055edb4cfa14ec3fdc Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 16:28:28 -0500 Subject: [PATCH 11/18] missed one --- .../metron/stellar/dsl/functions/StringFunctions.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java index d1a53b23f9..2865ea745d 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/StringFunctions.java @@ -572,12 +572,12 @@ public Object apply(List strings) { try { return (Map)JSONUtils.INSTANCE.load(in, JSONUtils.MAP_SUPPLIER); } catch (JsonProcessingException ex) { - throw new ParseException(String.format("{} is not a valid JSON string", in), ex); + throw new ParseException(String.format("%s is not a valid JSON string", in), ex); } catch (IOException ex) { - throw new ParseException(String.format("{} is not a valid JSON string", in), ex); + throw new ParseException(String.format("%s is not a valid JSON string", in), ex); } catch (ClassCastException ex) { - throw new ParseException(String.format("{} is not a valid JSON string, expected a map", in), ex); + throw new ParseException(String.format("%s is not a valid JSON string, expected a map", in), ex); } } } From 66494ec17afc5a22b8ad80c671eb3666529dcd52 Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 16:45:11 -0500 Subject: [PATCH 12/18] updating string functions test. --- .../metron/stellar/dsl/functions/StringFunctionsTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java index 9c5db15bd3..9f78fe1afa 100644 --- a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java +++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/StringFunctionsTest.java @@ -667,7 +667,7 @@ public void testToJsonMap() throws Exception { run("TO_JSON_MAP('{\"foo\" : 2')", new HashedMap<>()); } catch (ParseException pe) { thrown = true; - Assert.assertTrue(pe.getMessage().contains("Valid JSON string not supplied")); + Assert.assertTrue(pe.getMessage().contains("is not a valid JSON string")); } Assert.assertTrue(thrown); thrown = false; From 09cbb1fbda7712b918f3444d9fd7c220058da00e Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 17:10:55 -0500 Subject: [PATCH 13/18] Swapped args --- .../apache/metron/common/configuration/ConfigurationsUtils.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java index c18d9f9281..a89db63553 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/configuration/ConfigurationsUtils.java @@ -596,7 +596,7 @@ public static void applyConfigPatchToZookeeper( byte[] patchData, CuratorFramework client) throws Exception { byte[] configData = readConfigBytesFromZookeeper(configurationType, configName, client); - byte[] prettyPatchedConfig = JSONUtils.INSTANCE.applyPatch(configData, patchData); + byte[] prettyPatchedConfig = JSONUtils.INSTANCE.applyPatch(patchData, configData); // ensure the patch produces a valid result; otherwise exception thrown during deserialization String prettyPatchedConfigStr = new String(prettyPatchedConfig); From 17585f1b30437f12848749f4cf972e0d93e5fcea Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 18:51:01 -0500 Subject: [PATCH 14/18] Updating JSONUtils to actually function correctly. --- .../apache/metron/common/utils/JSONUtils.java | 32 +++++++++++++++---- .../ElasticsearchSearchIntegrationTest.java | 6 ++++ .../indexing/dao/SearchIntegrationTest.java | 2 +- .../stellar/common/utils/JSONUtils.java | 25 ++++++++++++--- 4 files changed, 54 insertions(+), 11 deletions(-) diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java index f76019aa51..135546da25 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/JSONUtils.java @@ -29,10 +29,15 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.function.Supplier; +import com.google.common.reflect.TypeToken; import org.json.simple.JSONObject; import org.json.simple.parser.JSONParser; import org.json.simple.parser.ParseException; @@ -41,14 +46,29 @@ public enum JSONUtils { INSTANCE; public static class ReferenceSupplier implements Supplier> { + Type type; + protected ReferenceSupplier() { + Type superClass = this.getClass().getGenericSuperclass(); + if(superClass instanceof Class) { + throw new IllegalArgumentException("Internal error: ReferenceSupplier constructed without actual type information"); + } else { + this.type = ((ParameterizedType)superClass).getActualTypeArguments()[0]; + } + } + @Override public TypeReference get() { - return new TypeReference() { }; + return new TypeReference() { + @Override + public Type getType() { + return type; + } + }; } } - public final static ReferenceSupplier> MAP_SUPPLIER = new ReferenceSupplier<>(); - public final static ReferenceSupplier> LIST_SUPPLIER = new ReferenceSupplier<>(); + public final static ReferenceSupplier> MAP_SUPPLIER = new ReferenceSupplier>() {}; + public final static ReferenceSupplier> LIST_SUPPLIER = new ReferenceSupplier>(){}; private static ThreadLocal _parser = ThreadLocal.withInitial(() -> new JSONParser()); @@ -66,16 +86,16 @@ public ObjectMapper getMapper() { public T load(InputStream is, ReferenceSupplier ref) throws IOException { - return _mapper.get().readValue(is, ref.get()); + return _mapper.get().readValue(is, (TypeReference)ref.get()); } public T load(String is, ReferenceSupplier ref) throws IOException { - return _mapper.get().readValue(is, ref.get()); + return _mapper.get().readValue(is, (TypeReference)ref.get()); } public T load(File f, ReferenceSupplier ref) throws IOException { try (InputStream is = new BufferedInputStream(new FileInputStream(f))) { - return _mapper.get().readValue(is, ref.get()); + return _mapper.get().readValue(is, (TypeReference)ref.get()); } } diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java index f86a04d9a3..1bc5b6e235 100644 --- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java +++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java @@ -21,13 +21,16 @@ import java.io.File; import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.concurrent.ExecutionException; import org.adrianwalker.multilinestring.Multiline; +import org.apache.metron.common.utils.JSONUtils; import org.apache.metron.elasticsearch.dao.ElasticsearchDao; import org.apache.metron.elasticsearch.integration.components.ElasticSearchComponent; import org.apache.metron.indexing.dao.AccessConfig; import org.apache.metron.indexing.dao.IndexDao; import org.apache.metron.indexing.dao.SearchIntegrationTest; +import org.apache.metron.indexing.dao.search.GetRequest; import org.apache.metron.integration.InMemoryComponent; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; @@ -37,6 +40,7 @@ import org.json.simple.JSONObject; import org.json.simple.parser.JSONParser; import org.json.simple.parser.ParseException; +import org.junit.Test; public class ElasticsearchSearchIntegrationTest extends SearchIntegrationTest { @@ -233,4 +237,6 @@ protected void loadTestData() throw new RuntimeException("Failed to index test data"); } } + + } diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java index ae648a3628..b40db46b0d 100644 --- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java +++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java @@ -463,7 +463,7 @@ public void find_one_guid() throws Exception { @Test public void get_all_latest_guid() throws Exception { - List request = JSONUtils.INSTANCE.load(getAllLatestQuery, new JSONUtils.ReferenceSupplier<>()); + List request = JSONUtils.INSTANCE.load(getAllLatestQuery, new JSONUtils.ReferenceSupplier>(){}); Map docs = new HashMap<>(); for(Document doc : dao.getAllLatest(request)) { diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java index 2a846aef12..be775d3d7c 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java @@ -27,6 +27,8 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; import java.util.List; import java.util.Map; import java.util.function.Supplier; @@ -40,9 +42,24 @@ public enum JSONUtils { INSTANCE; public static class ReferenceSupplier implements Supplier> { + Type type; + protected ReferenceSupplier() { + Type superClass = this.getClass().getGenericSuperclass(); + if(superClass instanceof Class) { + throw new IllegalArgumentException("Internal error: ReferenceSupplier constructed without actual type information"); + } else { + this.type = ((ParameterizedType)superClass).getActualTypeArguments()[0]; + } + } + @Override public TypeReference get() { - return new TypeReference() { }; + return new TypeReference() { + @Override + public Type getType() { + return type; + } + }; } } @@ -56,16 +73,16 @@ public TypeReference get() { new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL)); public T load(InputStream is, ReferenceSupplier ref) throws IOException { - return _mapper.get().readValue(is, ref.get()); + return _mapper.get().readValue(is, (TypeReference)ref.get()); } public T load(String is, ReferenceSupplier ref) throws IOException { - return _mapper.get().readValue(is, ref.get()); + return _mapper.get().readValue(is, (TypeReference)ref.get()); } public T load(File f, ReferenceSupplier ref) throws IOException { try (InputStream is = new BufferedInputStream(new FileInputStream(f))) { - return _mapper.get().readValue(is, ref.get()); + return _mapper.get().readValue(is, (TypeReference)ref.get()); } } From e289ad6e1c5b8a81719e598e78c2325e24e9160c Mon Sep 17 00:00:00 2001 From: cstella Date: Thu, 25 Jan 2018 18:51:55 -0500 Subject: [PATCH 15/18] missed one --- .../org/apache/metron/stellar/common/utils/JSONUtils.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java index be775d3d7c..9fb1c3f53e 100644 --- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java +++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/utils/JSONUtils.java @@ -63,8 +63,8 @@ public Type getType() { } } - public final static ReferenceSupplier> MAP_SUPPLIER = new ReferenceSupplier<>(); - public final static ReferenceSupplier> LIST_SUPPLIER = new ReferenceSupplier<>(); + public final static ReferenceSupplier> MAP_SUPPLIER = new ReferenceSupplier>(){}; + public final static ReferenceSupplier> LIST_SUPPLIER = new ReferenceSupplier>(){}; private static ThreadLocal _parser = ThreadLocal.withInitial(() -> new JSONParser()); From 8c07921e0c9993b51e48cf500f234f9488cce500 Mon Sep 17 00:00:00 2001 From: cstella Date: Mon, 29 Jan 2018 09:23:24 -0500 Subject: [PATCH 16/18] Normalizing protocol in the spout as well. --- .../metron/storm/kafka/flux/SimpleStormKafkaBuilder.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java b/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java index 1bcee9a009..5afe6f34f2 100644 --- a/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java +++ b/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java @@ -203,6 +203,9 @@ public SimpleStormKafkaBuilder( Map kafkaProps , createDeserializer(Optional.ofNullable((String)kafkaProps.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)), DEFAULT_DESERIALIZER) , subscription ); + if(kafkaProps.containsKey("security.protocol")) { + kafkaProps.put("security.protocol", KafkaUtils.INSTANCE.normalizeProtocol((String) kafkaProps.get("security.protocol"))); + } setProp(kafkaProps); setRecordTranslator(new SpoutRecordTranslator<>(FieldsConfiguration.toList(fieldsConfiguration))); } From ebad0ae557cd3166422e690f94061f12e2bf143c Mon Sep 17 00:00:00 2001 From: cstella Date: Mon, 29 Jan 2018 13:45:09 -0500 Subject: [PATCH 17/18] normalizing for the producer as well as consumer. --- .../java/org/apache/metron/common/utils/KafkaUtils.java | 9 +++++++++ .../metron/storm/kafka/flux/SimpleStormKafkaBuilder.java | 5 ++--- .../java/org/apache/metron/writer/kafka/KafkaWriter.java | 1 + 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java index f5ea49abfa..796bc42d8c 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/KafkaUtils.java @@ -33,6 +33,7 @@ public enum KafkaUtils { INSTANCE; + public static final String SECURITY_PROTOCOL = "security.protocol"; public List getBrokersFromZookeeper(String zkQuorum) throws Exception { RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); CuratorFramework framework = CuratorFrameworkFactory.newClient(zkQuorum, retryPolicy); @@ -67,6 +68,14 @@ public List getBrokersFromZookeeper(CuratorFramework client) throws Exce return ret; } + public Map normalizeProtocol(Map configs) { + if(configs.containsKey(SECURITY_PROTOCOL)) { + String protocol = normalizeProtocol((String)configs.get(SECURITY_PROTOCOL)); + configs.put(SECURITY_PROTOCOL, protocol); + } + return configs; + } + public String normalizeProtocol(String protocol) { if(protocol.equalsIgnoreCase("PLAINTEXTSASL") || protocol.equalsIgnoreCase("SASL_PLAINTEXT")) { if(SecurityProtocol.getNames().contains("PLAINTEXTSASL")) { diff --git a/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java b/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java index 5afe6f34f2..f99e5492d7 100644 --- a/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java +++ b/metron-platform/metron-storm-kafka/src/main/java/org/apache/metron/storm/kafka/flux/SimpleStormKafkaBuilder.java @@ -203,9 +203,8 @@ public SimpleStormKafkaBuilder( Map kafkaProps , createDeserializer(Optional.ofNullable((String)kafkaProps.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)), DEFAULT_DESERIALIZER) , subscription ); - if(kafkaProps.containsKey("security.protocol")) { - kafkaProps.put("security.protocol", KafkaUtils.INSTANCE.normalizeProtocol((String) kafkaProps.get("security.protocol"))); - } + + kafkaProps = KafkaUtils.INSTANCE.normalizeProtocol(kafkaProps); setProp(kafkaProps); setRecordTranslator(new SpoutRecordTranslator<>(FieldsConfiguration.toList(fieldsConfiguration))); } diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java index 7ce9b9bce6..f73e0f4625 100644 --- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java +++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java @@ -158,6 +158,7 @@ public Map createProducerConfigs() { producerConfig.put("value.serializer", valueSerializer); producerConfig.put("request.required.acks", requiredAcks); producerConfig.putAll(producerConfigs == null?new HashMap<>():producerConfigs); + producerConfig = KafkaUtils.INSTANCE.normalizeProtocol(producerConfig); return producerConfig; } From b281da4abcc6eb419ebebbd278c1095fae8ce1a9 Mon Sep 17 00:00:00 2001 From: Nick Allen Date: Mon, 29 Jan 2018 18:54:01 -0500 Subject: [PATCH 18/18] Fix for embedded handlebars --- .../ambari_gather_facts/defaults/main.yml | 19 ++++++ .../roles/ambari_gather_facts/tasks/main.yml | 67 +++++++++++-------- 2 files changed, 58 insertions(+), 28 deletions(-) create mode 100644 metron-deployment/ansible/roles/ambari_gather_facts/defaults/main.yml diff --git a/metron-deployment/ansible/roles/ambari_gather_facts/defaults/main.yml b/metron-deployment/ansible/roles/ambari_gather_facts/defaults/main.yml new file mode 100644 index 0000000000..5351a60ef0 --- /dev/null +++ b/metron-deployment/ansible/roles/ambari_gather_facts/defaults/main.yml @@ -0,0 +1,19 @@ +# +# 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. +# +--- +curl: "curl -s -u {{ ambari_user }}:{{ ambari_password }} -X GET -H \"X-Requested-By: ambari\"" +parse_json: "import sys, json; print json.load(sys.stdin)" diff --git a/metron-deployment/ansible/roles/ambari_gather_facts/tasks/main.yml b/metron-deployment/ansible/roles/ambari_gather_facts/tasks/main.yml index 2b37eec15b..25f0982f9f 100644 --- a/metron-deployment/ansible/roles/ambari_gather_facts/tasks/main.yml +++ b/metron-deployment/ansible/roles/ambari_gather_facts/tasks/main.yml @@ -32,55 +32,55 @@ cluster_name: "{{ (cluster_name_response.content | from_json)['items'][0].Clusters.cluster_name }}" when: cluster_name is undefined +- set_fact: + base_url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}" + # # namenode_host # - name: "Ask Ambari: namenode_host" - uri: - url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/services/HDFS/components/NAMENODE" - user: "{{ ambari_user }}" - password: "{{ ambari_password }}" - force_basic_auth: yes - return_content: yes - register: namenode_hosts_response + shell: > + {{ curl }} '{{ base_url }}/services/HDFS/components/NAMENODE' \ + | python -c '{{ parse_json }}["host_components"][0]["HostRoles"]["host_name"]' + args: + warn: false + register: namenode_host_response when: namenode_host is undefined - set_fact: - namenode_host: "{{ (namenode_hosts_response.content | from_json).host_components[0].HostRoles.host_name }}" + namenode_host: "{{ namenode_host_response.stdout_lines[0] }}" when: namenode_host is undefined # # core_site_tag # - name: "Ask Ambari: core_site_tag" - uri: - url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/hosts/{{ namenode_host }}/host_components/NAMENODE" - user: "{{ ambari_user }}" - password: "{{ ambari_password }}" - force_basic_auth: yes - return_content: yes + shell: > + {{ curl }} '{{ base_url }}/hosts/{{ namenode_host }}/host_components/NAMENODE' \ + | python -c '{{ parse_json }}["HostRoles"]["actual_configs"]["core-site"]["default"]' + args: + warn: false register: core_site_tag_response when: core_site_tag is undefined - set_fact: - core_site_tag: "{{ (core_site_tag_response.content | from_json).HostRoles.actual_configs['core-site'].default }}" + core_site_tag: "{{ core_site_tag_response.stdout_lines[0] }}" when: core_site_tag is undefined # # hdfs_url # - name: "Ask Ambari: hdfs_url" - uri: - url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/configurations?type=core-site&tag={{ core_site_tag }}" - user: "{{ ambari_user }}" - password: "{{ ambari_password }}" - force_basic_auth: yes - return_content: yes - register: core_site_response + shell: > + {{ curl }} '{{ base_url }}/configurations?type=core-site&tag={{ core_site_tag }}' \ + | python -c '{{ parse_json }}["items"][0]["properties"]["fs.defaultFS"]' + args: + warn: false + register: hdfs_url_response when: hdfs_url is undefined - set_fact: - hdfs_url: "{{ (core_site_response.content | from_json)['items'][0].properties['fs.defaultFS'] }}" + hdfs_url: "{{ hdfs_url_response.stdout_lines[0] }}" when: hdfs_url is undefined # @@ -88,7 +88,7 @@ # - name: "Ask Ambari: kafka_broker_hosts" uri: - url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/services/KAFKA/components/KAFKA_BROKER" + url: "{{ base_url }}/services/KAFKA/components/KAFKA_BROKER" user: "{{ ambari_user }}" password: "{{ ambari_password }}" force_basic_auth: yes @@ -105,7 +105,7 @@ # - name: "Ask Ambari: kafka_broker_tag" uri: - url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/hosts/{{ kafka_broker_hosts[0] }}/host_components/KAFKA_BROKER" + url: "{{ base_url }}/hosts/{{ kafka_broker_hosts[0] }}/host_components/KAFKA_BROKER" user: "{{ ambari_user }}" password: "{{ ambari_password }}" force_basic_auth: yes @@ -122,7 +122,8 @@ # - name: "Ask Ambari: kafka_broker_port" shell: > - curl -s -u {{ ambari_user }}:{{ ambari_password }} -X GET -H "X-Requested-By: ambari" "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/configurations?type=kafka-broker&tag={{ kafka_broker_tag }}" | python -c 'import sys, json; print json.load(sys.stdin)["items"][0]["properties"]["listeners"]' + {{ curl }} '{{ base_url }}/configurations?type=kafka-broker&tag={{ kafka_broker_tag }}' \ + | python -c '{{ parse_json }}["items"][0]["properties"]["listeners"]' args: warn: false register: kafka_broker_port_response @@ -191,6 +192,9 @@ zookeeper_url: "{% for host in zookeeper_hosts %}{% if loop.index != 1 %},{% endif %}{{ host }}:{{ zookeeper_port }}{% endfor %}" when: zookeeper_url is undefined +# +# metron_hosts +# - name: "Ask Ambari: metron_hosts" uri: url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/services/METRON/components/METRON_INDEXING" @@ -205,6 +209,9 @@ metron_hosts: "{{ (metron_hosts_response.content | from_json).host_components | map(attribute='HostRoles.host_name') | list }}" when: metron_hosts is undefined +# +# kibana hosts +# - name: "Ask Ambari: kibana_hosts" uri: url: "http://{{ groups.ambari_master[0] }}:{{ ambari_port }}/api/v1/clusters/{{ cluster_name }}/services/KIBANA/components/KIBANA_MASTER" @@ -225,10 +232,14 @@ # - name: debug debug: - msg: "zookeeper_port = {{ zookeeper_port }}, + msg: "cluster_name = {{ cluster_name }}, + namenode_host = {{ namenode_host }}, + hdfs_url = {{ hdfs_url }}, + zookeeper_port = {{ zookeeper_port }}, zookeeper_hosts = {{ zookeeper_hosts }}, zookeeper_url = {{ zookeeper_url }}, kafka_broker_port = {{ kafka_broker_port }}, kafka_broker_hosts = {{ kafka_broker_hosts }}, kafka_broker_url = {{ kafka_broker_url }}, - metron_hosts = {{ metron_hosts }}" + metron_hosts = {{ metron_hosts }}, + kibana_hosts = {{ kibana_hosts }}"