metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmerri...@apache.org
Subject metron git commit: METRON-1355 Convert metron-elasticsearch to new infrastructure (merrimanr) closes apache/metron#941
Date Mon, 26 Mar 2018 13:24:29 GMT
Repository: metron
Updated Branches:
  refs/heads/feature/METRON-1344-test-infrastructure fc4ce0104 -> 24633978b


METRON-1355 Convert metron-elasticsearch to new infrastructure (merrimanr) closes apache/metron#941


Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/24633978
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/24633978
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/24633978

Branch: refs/heads/feature/METRON-1344-test-infrastructure
Commit: 24633978bedbcfe35c040d0025859e31f455488e
Parents: fc4ce01
Author: merrimanr <merrimanr@gmail.com>
Authored: Mon Mar 26 08:24:11 2018 -0500
Committer: merrimanr <merrimanr@gmail.com>
Committed: Mon Mar 26 08:24:11 2018 -0500

----------------------------------------------------------------------
 .travis.yml                                     | 10 +-
 metron-contrib/metron-docker-e2e/README.md      | 94 +++++++++++++++++++
 .../compose/docker-compose.yml                  | 13 ++-
 .../compose/metron-rest/Dockerfile              |  4 +-
 .../compose/metron-rest/bin/start.sh            |  2 +-
 metron-contrib/metron-docker-e2e/pom.xml        |  2 +-
 .../scripts/create-docker-machine.sh            | 21 +++++
 .../scripts/wait_for_elasticsearch.sh           | 29 ++++++
 .../metron-alerts/e2e/utils/e2e_util.ts         | 29 +-----
 metron-platform/metron-elasticsearch/README.md  |  7 ++
 .../dao/ElasticsearchMetaAlertDao.java          | 56 +++++------
 .../ElasticsearchMetaAlertIntegrationTest.java  | 99 +++++++++-----------
 .../ElasticsearchSearchIntegrationTest.java     | 77 ++++-----------
 .../ElasticsearchUpdateIntegrationTest.java     | 89 ++++++++++--------
 .../utils/ElasticsearchTestUtils.java           | 95 +++++++++++++++++++
 .../indexing/dao/SearchIntegrationTest.java     | 29 +-----
 16 files changed, 415 insertions(+), 241 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index 073dc1e..de5f6a9 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -33,6 +33,7 @@ language: java
 jdk:
   - oraclejdk8
 before_install:
+  - sudo sysctl -w vm.max_map_count=262144
   - sudo rm /usr/local/bin/docker-compose
   - curl -L https://github.com/docker/compose/releases/download/${DOCKER_COMPOSE_VERSION}/docker-compose-`uname -s`-`uname -m` > docker-compose
   - chmod +x docker-compose
@@ -43,16 +44,19 @@ before_install:
   - export PATH=$M2_HOME/bin:$PATH
   - npm config set cache $HOME/.npm-cache --global
   - npm config set prefix $HOME/.npm-prefix --global
-  - if [ -f ${DOCKER_METRON_CENTOS} ]; then gunzip -c ${DOCKER_METRON_CENTOS} | docker load; else docker build metron-contrib/metron-docker-e2e/compose/metron-centos/ -t "metron-centos"; fi
+  #- if [ -f ${DOCKER_METRON_CENTOS} ]; then gunzip -c ${DOCKER_METRON_CENTOS} | docker load; else docker build metron-contrib/metron-docker-e2e/compose/metron-centos/ -t "metron-centos"; fi
 
 install:
   - time mvn -q -T 2C -DskipTests clean install
   - cd $E2E_COMPOSE_HOME && docker-compose up -d
-  - if [ ! -f ${DOCKER_METRON_CENTOS} ]; then docker save metron-centos | gzip > ${DOCKER_METRON_CENTOS}; fi
+  #- if [ ! -f ${DOCKER_METRON_CENTOS} ]; then docker save metron-centos | gzip > ${DOCKER_METRON_CENTOS}; fi
   - cd $TRAVIS_BUILD_DIR
 
 script:
-  - time mvn -q -T 2C surefire:test@unit-tests && time mvn -q surefire:test@integration-tests && time mvn -q test --projects metron-interface/metron-config && time dev-utilities/build-utils/verify_licenses.sh
+  #- time mvn -q -T 2C surefire:test@unit-tests && time mvn -q surefire:test@integration-tests && time mvn -q test --projects metron-interface/metron-config && time dev-utilities/build-utils/verify_licenses.sh
+  - metron-contrib/metron-docker-e2e/scripts/wait_for_elasticsearch.sh localhost 9210
+  - time mvn -q -T 2C surefire:test@unit-tests --projects=metron-platform/metron-elasticsearch
+  - time mvn -q surefire:test@integration-tests --projects=metron-platform/metron-elasticsearch
 
 before_cache:
   - rm -rf $HOME/.m2/repository/org/apache/metron

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-contrib/metron-docker-e2e/README.md
----------------------------------------------------------------------
diff --git a/metron-contrib/metron-docker-e2e/README.md b/metron-contrib/metron-docker-e2e/README.md
new file mode 100644
index 0000000..98dff60
--- /dev/null
+++ b/metron-contrib/metron-docker-e2e/README.md
@@ -0,0 +1,94 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+# Metron Docker
+
+Metron Docker E2E is a [Docker Compose](https://docs.docker.com/compose/overview/) application that serves as a backend to integration tests.
+
+Metron Docker includes these images that have been customized for Metron:
+
+  - Kafka
+  - Zookeeper
+  - Elasticsearch
+  - Metron REST
+  - Metron UIs
+
+Setup
+-----
+
+Install [Docker for Mac](https://docs.docker.com/docker-for-mac/) or [Docker for Windows](https://docs.docker.com/docker-for-windows/).  The following versions have been tested:
+
+  - Docker version 17.12.0-ce
+  - docker-machine version 0.13.0
+  - docker-compose version 1.18.0
+
+Build Metron from the top level directory with:
+```
+$ cd $METRON_HOME
+$ mvn clean install -DskipTests
+```
+
+Create a Docker machine:
+```
+$ export METRON_DOCKER_E2E_HOME=$METRON_HOME/metron-contrib/metron-docker-e2e
+$ cd $METRON_DOCKER_E2E_HOME
+$ ./scripts/create-docker-machine.sh
+```
+
+This will create a host called "metron-machine".  Anytime you want to run Docker commands against this host, make sure you run this first to set the Docker environment variables:
+```
+$ eval "$(docker-machine env metron-machine)"
+```
+
+If you wish to use a local docker-engine install, please set an environment variable BROKER_IP_ADDR to the IP address of your host machine. This cannot be the loopback address.
+
+Usage
+-----
+
+Navigate to the compose application root:
+```
+$ cd $METRON_DOCKER_E2E_HOME/compose/
+```
+
+The Metron Docker environment lifecycle is controlled by the [docker-compose](https://docs.docker.com/compose/reference/overview/) command.  The service names can be found in the docker-compose.yml file.  For example, to build and start the environment run this command:
+```
+$ eval "$(docker-machine env metron-machine)"
+$ docker-compose up -d
+```
+
+After all services have started list the containers and ensure their status is 'Up':
+```
+$ docker-compose ps
+         Name                       Command               State                       Ports                     
+----------------------------------------------------------------------------------------------------------------
+metron_elasticsearch_1   /bin/bash bin/es-docker          Up      0.0.0.0:9210->9200/tcp, 0.0.0.0:9310->9300/tcp
+metron_kafka_1           start-kafka.sh                   Up      0.0.0.0:9092->9092/tcp                        
+metron_metron-rest_1     /bin/sh -c ./bin/start.sh        Up      0.0.0.0:8082->8082/tcp                        
+metron_metron-ui_1       /bin/sh -c ./bin/start.sh        Up      0.0.0.0:4201->4201/tcp                        
+metron_zookeeper_1       /docker-entrypoint.sh zkSe ...   Up      0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp    
+```
+
+Various services are exposed through http on the Docker host.  Get the host ip from the URL property:
+```
+$ docker-machine ls
+NAME             ACTIVE   DRIVER       STATE     URL                         SWARM   DOCKER    ERRORS
+metron-machine   *        virtualbox   Running   tcp://192.168.99.100:2376           v1.12.5
+```
+
+The various integration tests can now be run against this environment.
+
+TODO: document how to set docker machine ip address for e2e tests

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-contrib/metron-docker-e2e/compose/docker-compose.yml
----------------------------------------------------------------------
diff --git a/metron-contrib/metron-docker-e2e/compose/docker-compose.yml b/metron-contrib/metron-docker-e2e/compose/docker-compose.yml
index e5a8134..69a6073 100644
--- a/metron-contrib/metron-docker-e2e/compose/docker-compose.yml
+++ b/metron-contrib/metron-docker-e2e/compose/docker-compose.yml
@@ -31,7 +31,18 @@ services:
     depends_on:
       - zookeeper
   elasticsearch:
-    image: elasticsearch:2.3
+    image: docker.elastic.co/elasticsearch/elasticsearch:5.6.8
+    environment:
+      - xpack.security.enabled=false
+      - xpack.monitoring.enabled=false
+      - xpack.ml.enabled=false
+      - xpack.graph.enabled=false
+      - xpack.watcher.enabled=false
+      - cluster.name=elasticsearch
+      - bootstrap.memory_lock=true
+      - ES_JAVA_OPTS=-Xms512m -Xmx512m
+      - discovery.zen.minimum_master_nodes=1
+      - discovery.type=single-node
     ports:
       - "9210:9200"
       - "9310:9300"

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-contrib/metron-docker-e2e/compose/metron-rest/Dockerfile
----------------------------------------------------------------------
diff --git a/metron-contrib/metron-docker-e2e/compose/metron-rest/Dockerfile b/metron-contrib/metron-docker-e2e/compose/metron-rest/Dockerfile
index a76cf57..56ddc9f 100644
--- a/metron-contrib/metron-docker-e2e/compose/metron-rest/Dockerfile
+++ b/metron-contrib/metron-docker-e2e/compose/metron-rest/Dockerfile
@@ -14,7 +14,9 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
-FROM metron-centos
+FROM centos
+
+RUN yum install -y java-1.8.0-openjdk lsof
 
 ARG METRON_VERSION
 

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-contrib/metron-docker-e2e/compose/metron-rest/bin/start.sh
----------------------------------------------------------------------
diff --git a/metron-contrib/metron-docker-e2e/compose/metron-rest/bin/start.sh b/metron-contrib/metron-docker-e2e/compose/metron-rest/bin/start.sh
index f307e16..5a2da33 100755
--- a/metron-contrib/metron-docker-e2e/compose/metron-rest/bin/start.sh
+++ b/metron-contrib/metron-docker-e2e/compose/metron-rest/bin/start.sh
@@ -18,6 +18,6 @@
 #
 $METRON_HOME/bin/zk_load_configs.sh -z zookeeper:2181 -m PUSH -i $METRON_HOME/config/zookeeper
 
-METRON_REST_CLASSPATH="$METRON_HOME/lib/metron-rest-$METRON_VERSION.jar:$METRON_HOME/lib/metron-elasticsearch-$METRON_VERSION-uber.jar"
+METRON_REST_CLASSPATH="$METRON_HOME/lib/metron-rest-$METRON_VERSION.jar:$METRON_HOME/lib/metron-parsers-$METRON_VERSION-uber.jar:$METRON_HOME/lib/metron-elasticsearch-$METRON_VERSION-uber.jar"
 
 java -cp $METRON_REST_CLASSPATH org.apache.metron.rest.MetronRestApplication --spring.config.location=$METRON_HOME/config/application-docker.yml --spring.profiles.active=dev --server.port=8082

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-contrib/metron-docker-e2e/pom.xml
----------------------------------------------------------------------
diff --git a/metron-contrib/metron-docker-e2e/pom.xml b/metron-contrib/metron-docker-e2e/pom.xml
index ebca4c3..6d5ee6c 100644
--- a/metron-contrib/metron-docker-e2e/pom.xml
+++ b/metron-contrib/metron-docker-e2e/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>org.apache.metron</groupId>
         <artifactId>metron-contrib</artifactId>
-        <version>0.4.2</version>
+        <version>0.4.3</version>
     </parent>
     <description>Metron Docker</description>
     <url>https://metron.apache.org/</url>

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-contrib/metron-docker-e2e/scripts/create-docker-machine.sh
----------------------------------------------------------------------
diff --git a/metron-contrib/metron-docker-e2e/scripts/create-docker-machine.sh b/metron-contrib/metron-docker-e2e/scripts/create-docker-machine.sh
new file mode 100755
index 0000000..1465dd9
--- /dev/null
+++ b/metron-contrib/metron-docker-e2e/scripts/create-docker-machine.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+#
+#  Licensed to the Apache Software Foundation (ASF) under one or more
+#  contributor license agreements.  See the NOTICE file distributed with
+#  this work for additional information regarding copyright ownership.
+#  The ASF licenses this file to You under the Apache License, Version 2.0
+#  (the "License"); you may not use this file except in compliance with
+#  the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+docker-machine create --driver virtualbox --virtualbox-disk-size "30000" --virtualbox-memory "4096" --virtualbox-cpu-count "2" metron-machine
+
+# Necessary for Elasticsearch Docker image: https://www.elastic.co/guide/en/elasticsearch/reference/current/docker.html#docker-cli-run-prod-mode
+docker-machine ssh metron-machine sudo sysctl -w vm.max_map_count=262144

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-contrib/metron-docker-e2e/scripts/wait_for_elasticsearch.sh
----------------------------------------------------------------------
diff --git a/metron-contrib/metron-docker-e2e/scripts/wait_for_elasticsearch.sh b/metron-contrib/metron-docker-e2e/scripts/wait_for_elasticsearch.sh
new file mode 100755
index 0000000..ec9647f
--- /dev/null
+++ b/metron-contrib/metron-docker-e2e/scripts/wait_for_elasticsearch.sh
@@ -0,0 +1,29 @@
+#!/bin/bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+ELASTICSEARCH_HOST=$1
+ELASTICSEARCH_PORT=$2
+DELAY=5
+MAX_ATTEMPTS=24
+COUNTER=0
+while [ $COUNTER -lt $MAX_ATTEMPTS ]; do
+  curl -s --output /dev/null -XGET http://$ELASTICSEARCH_HOST:$ELASTICSEARCH_PORT && echo Elasticsearch is up after waiting "$(($DELAY * $COUNTER))" seconds && break
+  sleep $DELAY
+  let COUNTER=COUNTER+1
+done
+if [ $COUNTER -eq $MAX_ATTEMPTS  ]; then echo Could not reach REST after "$(($DELAY * $COUNTER))" seconds; fi

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-interface/metron-alerts/e2e/utils/e2e_util.ts
----------------------------------------------------------------------
diff --git a/metron-interface/metron-alerts/e2e/utils/e2e_util.ts b/metron-interface/metron-alerts/e2e/utils/e2e_util.ts
index 9304a5c..8ae1de1 100644
--- a/metron-interface/metron-alerts/e2e/utils/e2e_util.ts
+++ b/metron-interface/metron-alerts/e2e/utils/e2e_util.ts
@@ -46,28 +46,7 @@ export function waitForStalenessOf (_element ) {
 }
 
 export function loadTestData() {
-  request.delete('http://user:password@' + browser.params.rest.url + '/api/v1/sensor/indexing/config/alerts_ui_e2e', function (e, response, body) {
-    request.post({url:'http://user:password@' + browser.params.rest.url + '/api/v1/sensor/indexing/config/alerts_ui_e2e', json:
-    {
-      "hdfs": {
-        "index": "alerts_ui_e2e",
-        "batchSize": 5,
-        "enabled": true
-      },
-      "elasticsearch": {
-        "index": "alerts_ui_e2e",
-        "batchSize": 5,
-        "enabled": true
-      },
-      "solr": {
-        "index": "alerts_ui_e2e",
-        "batchSize": 5,
-        "enabled": true
-      }
-    }
-    }, function (e, response, body) {
-    });
-  });
+  deleteTestData();
 
   let template = fs.readFileSync('e2e/mock-data/alerts_ui_e2e_index.template', 'utf8');
   request({
@@ -89,9 +68,7 @@ export function loadTestData() {
 }
 
 export function deleteTestData() {
-  request.delete('http://' + browser.params.elasticsearch.url + '/alerts_ui_e2e_index*');
-  request.delete('http://user:password@' + browser.params.rest.url + '/api/v1/sensor/indexing/config/alerts_ui_e2e', function (e, response, body) {
-  });
+  request.delete('http://node1:9200/alerts_ui_e2e_index*');
 }
 
 export function createMetaAlertsIndex() {
@@ -108,6 +85,6 @@ export function createMetaAlertsIndex() {
 }
 
 export function deleteMetaAlertsIndex() {
-  request.delete('http://' + browser.params.elasticsearch.url + '/metaalert_index*');
+  request.delete('http://node1:9200/metaalert_index*');
 }
 

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-platform/metron-elasticsearch/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/README.md b/metron-platform/metron-elasticsearch/README.md
index 1e15018..a7358dc 100644
--- a/metron-platform/metron-elasticsearch/README.md
+++ b/metron-platform/metron-elasticsearch/README.md
@@ -25,6 +25,7 @@ limitations under the License.
 * [Type Mappings](#type-mappings)
 * [Using Metron with Elasticsearch 5.6.2](#using-metron-with-elasticsearch-562)
 * [Installing Elasticsearch Templates](#installing-elasticsearch-templates)
+* [Integration Testing](#integration-testing)
 
 ## Introduction
 
@@ -332,3 +333,9 @@ Ambari UI -> Services -> Metron -> Service Actions -> Elasticsearch Template Ins
 _Method 2_ - Stop the Metron Indexing service, and start it again from Ambari UI. Note that the Metron Indexing service tracks if it has successfully installed the Elasticsearch templates, and will attempt to do so each time it is Started until successful.
 
 > Note: If you have made any customization to your index templates, then installing Elasticsearch templates afresh will lead to overwriting your existing changes. Please exercise caution.
+
+## Integration Testing
+
+Integration tests depend on Elasticsearch running inside a Docker container.  Due to limitations with Docker on Mac, docker-machine must be used to run Docker in a virtual machine.  
+
+If running integration tests in this scenario, the "es.ip" setting in the global config returned by ElasticsearchTestUtils.getGlobalConfig() must be changed from "localhost" to the ip address of your Docker machine.  The ip address can be found by running `docker-machine env metron-machine`.

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDao.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDao.java
index 2311a2b..1bb1618 100644
--- a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDao.java
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDao.java
@@ -18,19 +18,6 @@
 
 package org.apache.metron.elasticsearch.dao;
 
-import static org.apache.metron.common.Constants.GUID;
-import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
-import static org.elasticsearch.index.query.QueryBuilders.constantScoreQuery;
-import static org.elasticsearch.index.query.QueryBuilders.existsQuery;
-import static org.elasticsearch.index.query.QueryBuilders.nestedQuery;
-import static org.elasticsearch.index.query.QueryBuilders.termQuery;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import java.io.IOException;
-import java.util.*;
-import java.util.Map.Entry;
-import java.util.stream.Collectors;
-import org.apache.commons.collections4.SetUtils;
 import org.apache.lucene.search.join.ScoreMode;
 import org.apache.metron.common.Constants;
 import org.apache.metron.indexing.dao.AccessConfig;
@@ -51,29 +38,36 @@ import org.apache.metron.indexing.dao.search.SearchRequest;
 import org.apache.metron.indexing.dao.search.SearchResponse;
 import org.apache.metron.indexing.dao.search.SearchResult;
 import org.apache.metron.indexing.dao.update.Document;
-import org.elasticsearch.action.get.GetResponse;
-import org.elasticsearch.action.get.MultiGetItemResponse;
-import org.elasticsearch.action.get.MultiGetRequest.Item;
-import org.elasticsearch.action.get.MultiGetRequestBuilder;
-import org.elasticsearch.action.get.MultiGetResponse;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.search.SearchRequestBuilder;
-import org.elasticsearch.action.support.replication.ReplicationResponse.ShardInfo;
-import org.elasticsearch.action.update.UpdateRequest;
-import org.elasticsearch.action.update.UpdateResponse;
-import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.index.query.InnerHitBuilder;
-import org.elasticsearch.index.query.QueryBuilder;
-import org.elasticsearch.index.query.QueryBuilders;
-import org.elasticsearch.index.query.QueryStringQueryBuilder;
-import org.elasticsearch.search.SearchHit;
 import org.apache.metron.indexing.dao.update.OriginalNotFoundException;
 import org.apache.metron.indexing.dao.update.PatchRequest;
 import org.apache.metron.stellar.common.utils.ConversionUtils;
 import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.index.query.InnerHitBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.query.QueryStringQueryBuilder;
+import org.elasticsearch.search.SearchHit;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import static org.apache.metron.common.Constants.GUID;
+import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
+import static org.elasticsearch.index.query.QueryBuilders.constantScoreQuery;
+import static org.elasticsearch.index.query.QueryBuilders.existsQuery;
+import static org.elasticsearch.index.query.QueryBuilders.nestedQuery;
+import static org.elasticsearch.index.query.QueryBuilders.termQuery;
 
 public class ElasticsearchMetaAlertDao implements MetaAlertDao {
 
@@ -200,7 +194,7 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
 
     // Start a list of updates / inserts we need to run
     Map<Document, Optional<String>> updates = new HashMap<>();
-    updates.put(metaAlert, Optional.of(MetaAlertDao.METAALERTS_INDEX));
+    updates.put(metaAlert, Optional.of(this.index));
 
     try {
       // We need to update the associated alerts with the new meta alerts, making sure existing
@@ -430,7 +424,7 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
       // Each meta alert needs to be updated with the new alert
       for (Document metaAlert : metaAlerts) {
         replaceAlertInMetaAlert(metaAlert, update);
-        updates.put(metaAlert, Optional.of(METAALERTS_INDEX));
+        updates.put(metaAlert, Optional.of(this.index));
       }
 
       // Run the alert's update

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchMetaAlertIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchMetaAlertIntegrationTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchMetaAlertIntegrationTest.java
index 9e74fb6..d7880d1 100644
--- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchMetaAlertIntegrationTest.java
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchMetaAlertIntegrationTest.java
@@ -19,21 +19,20 @@
 package org.apache.metron.elasticsearch.integration;
 
 import static org.apache.metron.indexing.dao.MetaAlertDao.ALERT_FIELD;
-import static org.apache.metron.indexing.dao.MetaAlertDao.METAALERTS_INDEX;
 import static org.apache.metron.indexing.dao.MetaAlertDao.METAALERT_FIELD;
 import static org.apache.metron.indexing.dao.MetaAlertDao.METAALERT_TYPE;
 import static org.apache.metron.indexing.dao.MetaAlertDao.STATUS_FIELD;
+import static org.apache.metron.indexing.dao.MetaAlertDao.THREAT_FIELD_DEFAULT;
+import static org.apache.metron.indexing.dao.MetaAlertDao.THREAT_SORT_DEFAULT;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.base.Joiner;
 import com.google.common.collect.Iterables;
-import java.io.File;
+
 import java.io.IOException;
-import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -46,7 +45,8 @@ import org.apache.metron.common.Constants;
 import org.apache.metron.common.utils.JSONUtils;
 import org.apache.metron.elasticsearch.dao.ElasticsearchDao;
 import org.apache.metron.elasticsearch.dao.ElasticsearchMetaAlertDao;
-import org.apache.metron.elasticsearch.integration.components.ElasticSearchComponent;
+import org.apache.metron.elasticsearch.integration.utils.ElasticsearchTestUtils;
+import org.apache.metron.elasticsearch.utils.ElasticsearchUtils;
 import org.apache.metron.indexing.dao.AccessConfig;
 import org.apache.metron.indexing.dao.IndexDao;
 import org.apache.metron.indexing.dao.MetaAlertDao;
@@ -66,6 +66,7 @@ import org.apache.metron.indexing.dao.search.SortField;
 import org.apache.metron.indexing.dao.update.Document;
 import org.apache.metron.indexing.dao.update.OriginalNotFoundException;
 import org.apache.metron.indexing.dao.update.PatchRequest;
+import org.elasticsearch.client.Client;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assert;
@@ -74,20 +75,17 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 public class ElasticsearchMetaAlertIntegrationTest {
-
+  private static final String namespace = ElasticsearchMetaAlertIntegrationTest.class.getSimpleName().toLowerCase();
+  protected static final String METAALERT_INDEX = namespace + "_metaalert_index";
   private static final int MAX_RETRIES = 10;
   private static final int SLEEP_MS = 500;
   private static final String SENSOR_NAME = "test";
-  private static final String INDEX_DIR = "target/elasticsearch_meta";
-  private static final String DATE_FORMAT = "yyyy.MM.dd.HH";
-  private static final String INDEX =
-      SENSOR_NAME + "_index_" + new SimpleDateFormat(DATE_FORMAT).format(new Date());
+  private static final String INDEX = namespace + "_" + SENSOR_NAME + "_index";
   private static final String NEW_FIELD = "new-field";
   private static final String NAME_FIELD = "name";
 
   private static IndexDao esDao;
   private static MetaAlertDao metaDao;
-  private static ElasticSearchComponent es;
 
   /**
    {
@@ -104,7 +102,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
   /**
    {
      "guid": "meta_alert",
-     "index": "metaalert_index",
+     "index": "elasticsearchmetaalertintegrationtest_metaalert_index",
      "patch": [
        {
          "op": "add",
@@ -121,7 +119,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
   /**
    {
      "guid": "meta_alert",
-     "index": "metaalert_index",
+     "index": "elasticsearchmetaalertintegrationtest_metaalert_index",
      "patch": [
        {
          "op": "add",
@@ -143,7 +141,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
   /**
    {
      "guid": "meta_alert",
-     "index": "metaalert_index",
+     "index": "elasticsearchmetaalertintegrationtest_metaalert_index",
      "patch": [
        {
          "op": "add",
@@ -177,6 +175,12 @@ public class ElasticsearchMetaAlertIntegrationTest {
            },
            "alert" : {
              "type" : "nested"
+           },
+           "threat:triage:score": {
+             "type": "float"
+           },
+           "count" : {
+             "type" : "integer"
            }
          }
        }
@@ -185,52 +189,41 @@ public class ElasticsearchMetaAlertIntegrationTest {
   @Multiline
   public static String template;
 
+  private static Client client;
+
   @BeforeClass
   public static void setupBefore() throws Exception {
-    // setup the client
-    es = new ElasticSearchComponent.Builder()
-        .withHttpPort(9211)
-        .withIndexDir(new File(INDEX_DIR))
-        .build();
-    es.start();
+    client = ElasticsearchUtils.getClient(ElasticsearchTestUtils.getGlobalConfig(), null);
+    ElasticsearchTestUtils.clearIndices(client, METAALERT_INDEX, INDEX);
 
     AccessConfig accessConfig = new AccessConfig();
-    Map<String, Object> globalConfig = new HashMap<String, Object>() {
-      {
-        put("es.clustername", "metron");
-        put("es.port", "9300");
-        put("es.ip", "localhost");
-        put("es.date.format", DATE_FORMAT);
-      }
-    };
     accessConfig.setMaxSearchResults(1000);
-    accessConfig.setGlobalConfigSupplier(() -> globalConfig);
+    accessConfig.setGlobalConfigSupplier(ElasticsearchTestUtils::getGlobalConfig);
     accessConfig.setMaxSearchGroups(100);
 
     esDao = new ElasticsearchDao();
     esDao.init(accessConfig);
-    metaDao = new ElasticsearchMetaAlertDao(esDao);
+    metaDao = new ElasticsearchMetaAlertDao(esDao, METAALERT_INDEX, THREAT_FIELD_DEFAULT, THREAT_SORT_DEFAULT);
   }
 
   @Before
   public void setup() throws IOException {
-    es.createIndexWithMapping(METAALERTS_INDEX, MetaAlertDao.METAALERT_DOC, template.replace("%MAPPING_NAME%", "metaalert"));
-    es.createIndexWithMapping(INDEX, "index_doc", template.replace("%MAPPING_NAME%", "index"));
+    client.admin().indices().prepareCreate(METAALERT_INDEX)
+            .addMapping(METAALERT_TYPE + "_doc", template.replace("%MAPPING_NAME%", METAALERT_TYPE)).get();
+    client.admin().indices().prepareCreate(INDEX)
+            .addMapping("index_doc", template.replace("%MAPPING_NAME%", "index")).get();
   }
 
   @AfterClass
   public static void teardown() {
-    if (es != null) {
-      es.stop();
-    }
+    ElasticsearchTestUtils.clearIndices(client, METAALERT_INDEX, INDEX);
   }
 
   @After
   public void reset() {
-    es.reset();
+    ElasticsearchTestUtils.clearIndices(client, METAALERT_INDEX, INDEX);
   }
 
-
   @Test
   public void shouldGetAllMetaAlertsForAlert() throws Exception {
     // Load alerts
@@ -245,7 +238,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     metaAlerts.add(buildMetaAlert("meta_inactive", MetaAlertStatus.INACTIVE,
         Optional.of(Arrays.asList(alerts.get(0), alerts.get(2)))));
     // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
-    elasticsearchAdd(metaAlerts, METAALERTS_INDEX, MetaAlertDao.METAALERT_TYPE);
+    elasticsearchAdd(metaAlerts, METAALERT_INDEX, MetaAlertDao.METAALERT_TYPE);
 
     // Verify load was successful
     List<GetRequest> createdDocs = metaAlerts.stream().map(metaAlert ->
@@ -352,7 +345,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     // Load metaAlert
     Map<String, Object> metaAlert = buildMetaAlert("meta_alert", MetaAlertStatus.ACTIVE,
         Optional.of(Collections.singletonList(alerts.get(0))));
-    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERTS_INDEX, METAALERT_TYPE);
+    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERT_INDEX, METAALERT_TYPE);
 
     // Verify load was successful
     findCreatedDocs(Arrays.asList(
@@ -432,7 +425,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     // Load metaAlert
     Map<String, Object> metaAlert = buildMetaAlert("meta_alert", MetaAlertStatus.ACTIVE,
         Optional.of(Arrays.asList(alerts.get(0), alerts.get(1), alerts.get(2), alerts.get(3))));
-    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERTS_INDEX, METAALERT_TYPE);
+    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERT_INDEX, METAALERT_TYPE);
 
     // Verify load was successful
     findCreatedDocs(Arrays.asList(
@@ -522,7 +515,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     // Load metaAlert
     Map<String, Object> metaAlert = buildMetaAlert("meta_alert", MetaAlertStatus.INACTIVE,
         Optional.of(Collections.singletonList(alerts.get(0))));
-    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERTS_INDEX, METAALERT_TYPE);
+    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERT_INDEX, METAALERT_TYPE);
 
     // Verify load was successful
     findCreatedDocs(Arrays.asList(
@@ -572,7 +565,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     Map<String, Object> metaAlert = buildMetaAlert("meta_alert", MetaAlertStatus.ACTIVE,
         Optional.of(childAlerts));
     // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
-    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERTS_INDEX,
+    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERT_INDEX,
         MetaAlertDao.METAALERT_TYPE);
 
     List<GetRequest> requests = new ArrayList<>();
@@ -661,7 +654,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
 
 
     // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
-    elasticsearchAdd(Arrays.asList(activeMetaAlert, inactiveMetaAlert), METAALERTS_INDEX, MetaAlertDao.METAALERT_TYPE);
+    elasticsearchAdd(Arrays.asList(activeMetaAlert, inactiveMetaAlert), METAALERT_INDEX, MetaAlertDao.METAALERT_TYPE);
 
     // Verify load was successful
     findCreatedDocs(Arrays.asList(
@@ -671,7 +664,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     SearchResponse searchResponse = metaDao.search(new SearchRequest() {
       {
         setQuery("*");
-        setIndices(Collections.singletonList(MetaAlertDao.METAALERT_TYPE));
+        setIndices(Collections.singletonList(namespace + "_metaalert"));
         setFrom(0);
         setSize(5);
         setSort(Collections.singletonList(new SortField() {{
@@ -715,7 +708,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     Map<String, Object> inactiveMetaAlert = buildMetaAlert("meta_inactive", MetaAlertStatus.INACTIVE,
         Optional.of(Arrays.asList(alerts.get(2), alerts.get(3))));
     // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
-    elasticsearchAdd(Arrays.asList(activeMetaAlert, inactiveMetaAlert), METAALERTS_INDEX, MetaAlertDao.METAALERT_TYPE);
+    elasticsearchAdd(Arrays.asList(activeMetaAlert, inactiveMetaAlert), METAALERT_INDEX, MetaAlertDao.METAALERT_TYPE);
 
     // Verify load was successful
     findCreatedDocs(Arrays.asList(
@@ -851,7 +844,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     Map<String, Object> inactiveMetaAlert = buildMetaAlert("meta_inactive", MetaAlertStatus.INACTIVE,
         Optional.of(Collections.singletonList(alerts.get(0))));
     // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
-    elasticsearchAdd(Arrays.asList(activeMetaAlert, inactiveMetaAlert), METAALERTS_INDEX, METAALERT_TYPE);
+    elasticsearchAdd(Arrays.asList(activeMetaAlert, inactiveMetaAlert), METAALERT_INDEX, METAALERT_TYPE);
 
     // Verify load was successful
     findCreatedDocs(Arrays.asList(
@@ -865,7 +858,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
       Map<String, Object> message0 = new HashMap<String, Object>(alerts.get(0)) {
         {
           put(NEW_FIELD, "metron");
-          put(MetaAlertDao.THREAT_FIELD_DEFAULT, "10");
+          put(THREAT_FIELD_DEFAULT, "10");
         }
       };
       String guid = "" + message0.get(Constants.GUID);
@@ -951,7 +944,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
     Map<String, Object> metaAlert = buildMetaAlert("meta_alert", MetaAlertStatus.ACTIVE,
         Optional.of(Arrays.asList(alerts.get(0), alerts.get(1))));
     // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
-    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERTS_INDEX, MetaAlertDao.METAALERT_TYPE);
+    elasticsearchAdd(Collections.singletonList(metaAlert), METAALERT_INDEX, MetaAlertDao.METAALERT_TYPE);
 
     // Verify load was successful
     findCreatedDocs(Arrays.asList(
@@ -1005,7 +998,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
   protected long getMatchingAlertCount(String fieldName, Object fieldValue) throws IOException, InterruptedException {
     long cnt = 0;
     for (int t = 0; t < MAX_RETRIES && cnt == 0; ++t, Thread.sleep(SLEEP_MS)) {
-      List<Map<String, Object>> docs = es.getAllIndexedDocs(INDEX, SENSOR_NAME + "_doc");
+      List<Map<String, Object>> docs = ElasticsearchTestUtils.getAllIndexedDocs(client, INDEX, SENSOR_NAME + "_doc");
       cnt = docs
           .stream()
           .filter(d -> {
@@ -1019,7 +1012,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
   protected long getMatchingMetaAlertCount(String fieldName, String fieldValue) throws IOException, InterruptedException {
     long cnt = 0;
     for (int t = 0; t < MAX_RETRIES && cnt == 0; ++t, Thread.sleep(SLEEP_MS)) {
-      List<Map<String, Object>> docs = es.getAllIndexedDocs(METAALERTS_INDEX, MetaAlertDao.METAALERT_DOC);
+      List<Map<String, Object>> docs = ElasticsearchTestUtils.getAllIndexedDocs(client, METAALERT_INDEX, MetaAlertDao.METAALERT_DOC);
       cnt = docs
           .stream()
           .filter(d -> {
@@ -1085,7 +1078,7 @@ public class ElasticsearchMetaAlertIntegrationTest {
       Map<String, Object> alerts = new HashMap<>();
       alerts.put(Constants.GUID, guid);
       alerts.put("source:type", SENSOR_NAME);
-      alerts.put(MetaAlertDao.THREAT_FIELD_DEFAULT, i);
+      alerts.put(THREAT_FIELD_DEFAULT, i);
       alerts.put("timestamp", System.currentTimeMillis());
       inputData.add(alerts);
     }
@@ -1113,9 +1106,9 @@ public class ElasticsearchMetaAlertIntegrationTest {
     return metaAlert;
   }
 
-  protected void elasticsearchAdd(List<Map<String, Object>> inputData, String index, String docType)
+  protected void elasticsearchAdd(List<Map<String, Object>> inputData, String index, String sensorType)
       throws IOException {
-    es.add(index, docType, inputData.stream().map(m -> {
+    ElasticsearchTestUtils.add(client, index, sensorType + "_doc", inputData.stream().map(m -> {
           try {
             return JSONUtils.INSTANCE.toJSON(m, true);
           } catch (JsonProcessingException e) {

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java
----------------------------------------------------------------------
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 7089033..6791277 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
@@ -17,49 +17,26 @@
  */
 package org.apache.metron.elasticsearch.integration;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.concurrent.ExecutionException;
-import java.util.Map;
 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.utils.ElasticsearchTestUtils;
 import org.apache.metron.elasticsearch.utils.ElasticsearchUtils;
 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;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.support.WriteRequest;
-import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.Client;
-import org.elasticsearch.index.IndexNotFoundException;
-import org.elasticsearch.index.query.QueryBuilders;
-import org.elasticsearch.search.SearchHit;
 import org.json.simple.JSONArray;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.json.simple.parser.ParseException;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.concurrent.ExecutionException;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 
 public class ElasticsearchSearchIntegrationTest extends SearchIntegrationTest {
-  private static String host = "localhost";
-  private static String port = "9310";
-  private static String dateFormat = "yyyy.MM.dd.HH";
-  private static final int MAX_RETRIES = 10;
-  private static final int SLEEP_MS = 500;
-
   /**
    * {
    * "searchintegrationtest_bro_doc": {
@@ -186,30 +163,17 @@ public class ElasticsearchSearchIntegrationTest extends SearchIntegrationTest {
   @Multiline
   private static String broDefaultStringMappings;
 
-  private static Map<String, Object> globalConfig;
   private static Client client;
 
   @BeforeClass
   public static void start() {
-    globalConfig = new HashMap<String, Object>() {{
-      put("es.clustername", "elasticsearch");
-      put("es.port", port);
-      put("es.ip", host);
-      put("es.date.format", dateFormat);
-    }};
-    client = ElasticsearchUtils.getClient(globalConfig, null);
-    clearIndices();
+    client = ElasticsearchUtils.getClient(ElasticsearchTestUtils.getGlobalConfig(), null);
+    ElasticsearchTestUtils.clearIndices(client, broIndex, snortIndex);
   }
 
   @AfterClass
   public static void stop() throws Exception {
-    clearIndices();
-  }
-
-  private static void clearIndices() {
-    try {
-      client.admin().indices().prepareDelete(broIndex, snortIndex, metaAlertIndex).get();
-    } catch (IndexNotFoundException infe) {}
+    ElasticsearchTestUtils.clearIndices(client, broIndex, snortIndex);
   }
 
   @Override
@@ -217,7 +181,7 @@ public class ElasticsearchSearchIntegrationTest extends SearchIntegrationTest {
     AccessConfig config = new AccessConfig();
     config.setMaxSearchResults(100);
     config.setMaxSearchGroups(100);
-    config.setGlobalConfigSupplier( () -> globalConfig);
+    config.setGlobalConfigSupplier(ElasticsearchTestUtils::getGlobalConfig);
 
     IndexDao dao = new ElasticsearchDao();
     dao.init(config);
@@ -226,33 +190,32 @@ public class ElasticsearchSearchIntegrationTest extends SearchIntegrationTest {
 
   @Override
   protected void loadTestData()
-      throws ParseException, IOException, ExecutionException, InterruptedException {
+      throws ParseException {
     client.admin().indices().prepareCreate(broIndex)
             .addMapping(broType, broTypeMappings).get();
     client.admin().indices().prepareCreate(snortIndex)
             .addMapping(snortType, snortTypeMappings).get();
 
-    BulkRequestBuilder bulkRequest = client.prepareBulk().setRefresh(true);
-    JSONArray broArray = (JSONArray) new JSONParser().parse(broData);
-    for(Object o: broArray) {
-      JSONObject jsonObject = (JSONObject) o;
-      IndexRequestBuilder indexRequestBuilder = client.prepareIndex(broIndex, broType);
-      indexRequestBuilder = indexRequestBuilder.setSource(jsonObject.toJSONString());
-      indexRequestBuilder = indexRequestBuilder.setTimestamp(jsonObject.get("timestamp").toString());
-      bulkRequest.add(indexRequestBuilder);
+    BulkRequestBuilder bulkRequest = client.prepareBulk().setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
+    addIndexRequest(bulkRequest, broData, broIndex, broType);
+    addIndexRequest(bulkRequest, snortData, snortIndex, snortType);
+    BulkResponse bulkResponse = bulkRequest.execute().actionGet();
+    if (bulkResponse.hasFailures()) {
+      throw new RuntimeException("Failed to index test data");
     }
-    JSONArray snortArray = (JSONArray) new JSONParser().parse(snortData);
-    for(Object o: snortArray) {
+  }
+
+  private void addIndexRequest(BulkRequestBuilder bulkRequest, String data, String index, String docType) throws ParseException {
+    JSONArray dataArray = (JSONArray) new JSONParser().parse(data);
+    for(Object o: dataArray) {
       JSONObject jsonObject = (JSONObject) o;
-      IndexRequestBuilder indexRequestBuilder = client.prepareIndex(snortIndex, snortType);
+      IndexRequestBuilder indexRequestBuilder = client.prepareIndex(index, docType);
+      indexRequestBuilder = indexRequestBuilder.setId((String) jsonObject.get("guid"));
       indexRequestBuilder = indexRequestBuilder.setSource(jsonObject.toJSONString());
       indexRequestBuilder = indexRequestBuilder.setTimestamp(jsonObject.get("timestamp").toString());
       bulkRequest.add(indexRequestBuilder);
     }
-    BulkResponse bulkResponse = bulkRequest.execute().actionGet();
-    if (bulkResponse.hasFailures()) {
-      throw new RuntimeException("Failed to index test data");
-    }
   }
 
+
 }

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchUpdateIntegrationTest.java
----------------------------------------------------------------------
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 325d42e..3c7831b 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,7 @@
 package org.apache.metron.elasticsearch.integration;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.google.common.collect.Iterables;
+import org.adrianwalker.multilinestring.Multiline;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.client.Get;
@@ -26,38 +26,59 @@ import org.apache.hadoop.hbase.client.Result;
 import org.apache.metron.common.Constants;
 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.elasticsearch.integration.utils.ElasticsearchTestUtils;
+import org.apache.metron.elasticsearch.utils.ElasticsearchUtils;
 import org.apache.metron.hbase.mock.MockHTable;
 import org.apache.metron.hbase.mock.MockHBaseTableProvider;
 import org.apache.metron.indexing.dao.*;
 import org.apache.metron.indexing.dao.update.Document;
 import org.apache.metron.indexing.dao.update.ReplaceRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.client.Client;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
-import java.io.File;
-import java.text.SimpleDateFormat;
 import java.util.*;
+import java.util.stream.Collectors;
 
 
 public class ElasticsearchUpdateIntegrationTest {
+  private static final String namespace = ElasticsearchUpdateIntegrationTest.class.getSimpleName().toLowerCase();
   private static final int MAX_RETRIES = 10;
   private static final int SLEEP_MS = 500;
   private static final String SENSOR_NAME= "test";
   private static final String TABLE_NAME = "modifications";
   private static final String CF = "p";
-  private static String indexDir = "target/elasticsearch_mutation";
-  private static String dateFormat = "yyyy.MM.dd.HH";
-  private static String index = SENSOR_NAME + "_index_" + new SimpleDateFormat(dateFormat).format(new Date());
+  private static String index = namespace + "_" + SENSOR_NAME + "_index";
   private static MockHTable table;
   private static IndexDao esDao;
   private static IndexDao hbaseDao;
   private static MultiIndexDao dao;
-  private static ElasticSearchComponent es;
+  private static Client client;
+
+  /**
+   * {
+       "test_doc" : {
+         "properties" : {
+           "guid" : {
+             "type" : "keyword"
+           },
+           "ip_src_addr" : {
+             "type" : "keyword"
+           },
+           "score" : {
+             "type" : "integer"
+           },
+           "alert" : {
+             "type" : "nested"
+           }
+         }
+       }
+     }
+   */
+  @Multiline
+  public static String testTypeMappings;
 
   @BeforeClass
   public static void setup() throws Exception {
@@ -66,23 +87,15 @@ public class ElasticsearchUpdateIntegrationTest {
     tableProvider.addToCache(TABLE_NAME, CF);
     table = (MockHTable)tableProvider.getTable(config, TABLE_NAME);
     // setup the client
-    es = new ElasticSearchComponent.Builder()
-            .withHttpPort(9211)
-            .withIndexDir(new File(indexDir))
-            .build();
-    es.start();
+    client = ElasticsearchUtils.getClient(ElasticsearchTestUtils.getGlobalConfig(), null);
+    client.admin().indices().prepareCreate(index).addMapping("test_doc", testTypeMappings).get();
 
     hbaseDao = new HBaseDao();
     AccessConfig accessConfig = new AccessConfig();
     accessConfig.setTableProvider(tableProvider);
-    Map<String, Object> globalConfig = new HashMap<String, Object>() {{
-      put("es.clustername", "metron");
-      put("es.port", "9300");
-      put("es.ip", "localhost");
-      put("es.date.format", dateFormat);
-      put(HBaseDao.HBASE_TABLE, TABLE_NAME);
-      put(HBaseDao.HBASE_CF, CF);
-    }};
+    Map<String, Object> globalConfig = ElasticsearchTestUtils.getGlobalConfig();
+    globalConfig.put(HBaseDao.HBASE_TABLE, TABLE_NAME);
+    globalConfig.put(HBaseDao.HBASE_CF, CF);
     accessConfig.setGlobalConfigSupplier(() -> globalConfig);
 
     esDao = new ElasticsearchDao();
@@ -94,9 +107,7 @@ public class ElasticsearchUpdateIntegrationTest {
 
   @AfterClass
   public static void teardown() {
-    if(es != null) {
-      es.stop();
-    }
+    ElasticsearchTestUtils.clearIndices(client, index);
   }
 
 
@@ -115,20 +126,18 @@ public class ElasticsearchUpdateIntegrationTest {
               }}
                              );
     }
-    es.add(index, SENSOR_NAME
-          , Iterables.transform(inputData,
-                    m -> {
-                      try {
-                        return JSONUtils.INSTANCE.toJSON(m, true);
-                      } catch (JsonProcessingException e) {
-                        throw new IllegalStateException(e.getMessage(), e);
-                      }
-                    }
-                    )
-    );
+    ElasticsearchTestUtils.add(client, index, SENSOR_NAME + "_doc"
+            , inputData.stream().map(m -> {
+              try {
+                return JSONUtils.INSTANCE.toJSON(m, true);
+              } catch (JsonProcessingException e) {
+                throw new IllegalStateException(e.getMessage(), e);
+              }
+            }).collect(Collectors.toList()));
+
     List<Map<String,Object>> docs = null;
     for(int t = 0;t < MAX_RETRIES;++t, Thread.sleep(SLEEP_MS)) {
-      docs = es.getAllIndexedDocs(index, SENSOR_NAME + "_doc");
+      docs = ElasticsearchTestUtils.getAllIndexedDocs(client, index, SENSOR_NAME + "_doc");
       if(docs.size() >= 10) {
         break;
       }
@@ -164,7 +173,7 @@ public class ElasticsearchUpdateIntegrationTest {
         //ensure ES is up-to-date
         long cnt = 0;
         for (int t = 0; t < MAX_RETRIES && cnt == 0; ++t, Thread.sleep(SLEEP_MS)) {
-          docs = es.getAllIndexedDocs(index, SENSOR_NAME + "_doc");
+          docs = ElasticsearchTestUtils.getAllIndexedDocs(client, index, SENSOR_NAME + "_doc");
           cnt = docs
                   .stream()
                   .filter(d -> message0.get("new-field").equals(d.get("new-field")))
@@ -204,7 +213,7 @@ public class ElasticsearchUpdateIntegrationTest {
         //ensure ES is up-to-date
         long cnt = 0;
         for (int t = 0; t < MAX_RETRIES && cnt == 0; ++t,Thread.sleep(SLEEP_MS)) {
-          docs = es.getAllIndexedDocs(index, SENSOR_NAME + "_doc");
+          docs = ElasticsearchTestUtils.getAllIndexedDocs(client, index, SENSOR_NAME + "_doc");
           cnt = docs
                   .stream()
                   .filter(d -> message0.get("new-field").equals(d.get("new-field")))

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/utils/ElasticsearchTestUtils.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/utils/ElasticsearchTestUtils.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/utils/ElasticsearchTestUtils.java
new file mode 100644
index 0000000..08c870f
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/utils/ElasticsearchTestUtils.java
@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.elasticsearch.integration.utils;
+
+import org.apache.metron.common.Constants;
+import org.apache.metron.common.utils.JSONUtils;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.search.SearchHit;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ElasticsearchTestUtils {
+
+  public static Map<String, Object> getGlobalConfig() {
+    return new HashMap<String, Object>() {{
+      put("es.clustername", "elasticsearch");
+      put("es.port", "9310");
+      put("es.ip", "localhost");
+      put("es.date.format", "yyyy.MM.dd.HH");
+    }};
+  }
+
+  public static void clearIndices(Client client, String... indices) {
+    for (String index: indices) {
+      try {
+        client.admin().indices().prepareDelete(index).get();
+      } catch (IndexNotFoundException infe) {
+      }
+    }
+  }
+
+  public static List<Map<String, Object>> getAllIndexedDocs(Client client, String index, String sourceType) throws IOException {
+    client.admin().indices().refresh(new RefreshRequest());
+    SearchResponse response = client.prepareSearch(index)
+            .setTypes(sourceType)
+            .setFrom(0)
+            .setSize(1000)
+            .execute().actionGet();
+    List<Map<String, Object>> ret = new ArrayList<Map<String, Object>>();
+    for (SearchHit hit : response.getHits()) {
+      Object o = hit.getSource();
+      ret.add((Map<String, Object>) (o));
+    }
+    return ret;
+  }
+
+  public static BulkResponse add(Client client, String indexName, String docType, Iterable<String> docs)
+          throws IOException {
+    BulkRequestBuilder bulkRequest = client.prepareBulk();
+    for (String doc : docs) {
+      IndexRequestBuilder indexRequestBuilder = client.prepareIndex(indexName, docType);
+
+      indexRequestBuilder = indexRequestBuilder.setSource(doc);
+      Map<String, Object> esDoc = JSONUtils.INSTANCE
+              .load(doc, JSONUtils.MAP_SUPPLIER);
+      indexRequestBuilder.setId((String) esDoc.get(Constants.GUID));
+      Object ts = esDoc.get("timestamp");
+      if (ts != null) {
+        indexRequestBuilder = indexRequestBuilder.setTimestamp(ts.toString());
+      }
+      bulkRequest.add(indexRequestBuilder);
+    }
+
+    BulkResponse response = bulkRequest.execute().actionGet();
+    if (response.hasFailures()) {
+      throw new IOException(response.buildFailureMessage());
+    }
+    return response;
+  }
+}

http://git-wip-us.apache.org/repos/asf/metron/blob/24633978/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java
----------------------------------------------------------------------
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 d7aa7c7..511d09f 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
@@ -1,4 +1,3 @@
-
 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -50,10 +49,8 @@ public abstract class SearchIntegrationTest {
   private static final String namespace = SearchIntegrationTest.class.getSimpleName().toLowerCase();
   protected static final String broIndex = namespace + "_bro_index";
   protected static final String snortIndex = namespace + "_snort_index";
-  protected static final String metaAlertIndex = namespace + "_metaalert_index";
   protected static final String broType = namespace + "_bro_doc";
   protected static final String snortType = namespace + "_snort_doc";
-  protected static final String metaAlertType = namespace + "_metaalert_doc";
 
   /**
    * [
@@ -113,7 +110,7 @@ public abstract class SearchIntegrationTest {
    * },
    * {
    * "guid": "snort_2",
-   * "sensorType": "searchintegrationtest_bro"
+   * "sensorType": "searchintegrationtest_snort"
    * }
    * ]
    */
@@ -543,7 +540,6 @@ public abstract class SearchIntegrationTest {
     for (int i = 2; i < 10; i++) {
       Assert.assertFalse(results.get(i).getSource().containsKey("threat:triage:score"));
     }
-<<<<<<< HEAD
   }
 
   @Test
@@ -570,27 +566,6 @@ public abstract class SearchIntegrationTest {
     for (int i = 5, j = 0; i > 0; i--, j++) {
       Assert.assertEquals("bro", results.get(j).getSource().get("source:type"));
       Assert.assertEquals(i, results.get(j).getSource().get("timestamp"));
-=======
-    // getColumnMetadata with multiple indices
-    {
-      Map<String, FieldType> fieldTypes = dao.getColumnMetadata(Arrays.asList("searchintegrationtest_bro", "searchintegrationtest_snort"));
-      Assert.assertEquals(15, fieldTypes.size());
-      Assert.assertEquals(FieldType.STRING, fieldTypes.get("guid"));
-      Assert.assertEquals(FieldType.STRING, fieldTypes.get("source:type"));
-      Assert.assertEquals(FieldType.IP, fieldTypes.get("ip_src_addr"));
-      Assert.assertEquals(FieldType.INTEGER, fieldTypes.get("ip_src_port"));
-      Assert.assertEquals(FieldType.LONG, fieldTypes.get("long_field"));
-      Assert.assertEquals(FieldType.DATE, fieldTypes.get("timestamp"));
-      Assert.assertEquals(FieldType.FLOAT, fieldTypes.get("latitude"));
-      Assert.assertEquals(FieldType.DOUBLE, fieldTypes.get("score"));
-      Assert.assertEquals(FieldType.BOOLEAN, fieldTypes.get("is_alert"));
-      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("location_point"));
-      Assert.assertEquals(FieldType.STRING, fieldTypes.get("bro_field"));
-      Assert.assertEquals(FieldType.INTEGER, fieldTypes.get("snort_field"));
-      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("duplicate_name_field"));
-      Assert.assertEquals(FieldType.FLOAT, fieldTypes.get("threat:triage:score"));
-      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("alert"));
->>>>>>> upstream/feature/METRON-1344-test-infrastructure
     }
   }
 
@@ -694,7 +669,7 @@ public abstract class SearchIntegrationTest {
 
   @Test
   public void returns_column_data_for_multiple_indices() throws Exception {
-    Map<String, FieldType> fieldTypes = dao.getColumnMetadata(Arrays.asList("bro", "snort"));
+    Map<String, FieldType> fieldTypes = dao.getColumnMetadata(Arrays.asList("searchintegrationtest_bro", "searchintegrationtest_snort"));
     Assert.assertEquals(15, fieldTypes.size());
     Assert.assertEquals(FieldType.KEYWORD, fieldTypes.get("guid"));
     Assert.assertEquals(FieldType.TEXT, fieldTypes.get("source:type"));


Mime
View raw message