You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by le...@apache.org on 2017/10/26 14:37:00 UTC
metron git commit: METRON-1272 Hide child alerts from searches and
grouping if they belong to meta alerts (justinleet) closes apache/metron#811
Repository: metron
Updated Branches:
refs/heads/master 5243366c4 -> 131a15ef7
METRON-1272 Hide child alerts from searches and grouping if they belong to meta alerts (justinleet) closes apache/metron#811
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/131a15ef
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/131a15ef
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/131a15ef
Branch: refs/heads/master
Commit: 131a15ef73a4c9d2b6568887135dbbed5ba5ebfc
Parents: 5243366
Author: justinleet <ju...@gmail.com>
Authored: Thu Oct 26 10:34:53 2017 -0400
Committer: leet <le...@apache.org>
Committed: Thu Oct 26 10:34:53 2017 -0400
----------------------------------------------------------------------
.../CURRENT/package/files/meta_index.mapping | 4 +
metron-interface/metron-alerts/README.md | 7 +
.../elasticsearch/dao/ElasticsearchDao.java | 159 +++--
.../dao/ElasticsearchMetaAlertDao.java | 257 +++++++-
.../dao/ElasticsearchMetaAlertDaoTest.java | 107 +++-
.../ElasticsearchMetaAlertIntegrationTest.java | 620 ++++++++++++++-----
.../components/ElasticSearchComponent.java | 6 +
metron-platform/metron-indexing/README.md | 2 +
.../apache/metron/indexing/dao/HBaseDao.java | 36 +-
.../apache/metron/indexing/dao/IndexDao.java | 9 +-
.../metron/indexing/dao/MetaAlertDao.java | 1 +
.../metron/indexing/dao/MultiIndexDao.java | 29 +-
.../indexing/dao/search/SearchResult.java | 32 +
.../apache/metron/indexing/dao/InMemoryDao.java | 7 +
.../indexing/dao/InMemoryMetaAlertDao.java | 5 +
15 files changed, 1036 insertions(+), 245 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/meta_index.mapping
----------------------------------------------------------------------
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/meta_index.mapping b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/meta_index.mapping
index c42343e..9da0554 100644
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/meta_index.mapping
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/meta_index.mapping
@@ -35,6 +35,10 @@
},
"alert": {
"type": "nested"
+ },
+ "source:type": {
+ "type": "string",
+ "index": "not_analyzed"
}
}
}
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-interface/metron-alerts/README.md
----------------------------------------------------------------------
diff --git a/metron-interface/metron-alerts/README.md b/metron-interface/metron-alerts/README.md
index 3d43e3e..1e1acb9 100644
--- a/metron-interface/metron-alerts/README.md
+++ b/metron-interface/metron-alerts/README.md
@@ -15,6 +15,13 @@ Alert GUIDs must be double-quoted when being searched on to ensure correctness o
### Search for Comments
Users cannot search for the contents of the comment's in the Alerts-UI
+### Meta alerts
+Grouping/faceting requests and other aggregations do not return meta alerts. This is because it's not clear what the intended results should be when there are multiple matching items.
+
+Sorting has a similar caveat, in that if we are matching on multiple alerts, there is no well defined sort.
+
+Alerts that are contained in a a meta alert are generally excluded from search results, because a user has already grouped them in a meaningful way.
+
## Prerequisites
* The Metron REST application should be up and running and Elasticsearch should have some alerts populated by Metron topologies
* The Management UI should be installed (which includes [Express](https://expressjs.com/))
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java
index f2f1b38..e423e56 100644
--- a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java
@@ -17,8 +17,23 @@
*/
package org.apache.metron.elasticsearch.dao;
+import static org.apache.metron.elasticsearch.utils.ElasticsearchUtils.INDEX_NAME_DELIMITER;
+
import com.google.common.base.Splitter;
import com.google.common.collect.Iterables;
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import org.apache.metron.elasticsearch.utils.ElasticsearchUtils;
import org.apache.metron.indexing.dao.AccessConfig;
import org.apache.metron.indexing.dao.IndexDao;
@@ -37,6 +52,8 @@ import org.apache.metron.indexing.dao.search.SortOrder;
import org.apache.metron.indexing.dao.update.Document;
import org.elasticsearch.action.ActionWriteResponse.ShardInfo;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchPhaseExecutionException;
import org.elasticsearch.action.search.SearchRequestBuilder;
@@ -63,22 +80,6 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import static org.apache.metron.elasticsearch.utils.ElasticsearchUtils.INDEX_NAME_DELIMITER;
-
public class ElasticsearchDao implements IndexDao {
@@ -174,21 +175,34 @@ public class ElasticsearchDao implements IndexDao {
@Override
public GroupResponse group(GroupRequest groupRequest) throws InvalidSearchException {
- if(client == null) {
+ return group(groupRequest, new QueryStringQueryBuilder(groupRequest.getQuery()));
+ }
+
+ /**
+ * Defers to a provided {@link org.elasticsearch.index.query.QueryBuilder} for the query.
+ * @param groupRequest The request defining the parameters of the grouping
+ * @param queryBuilder The actual query to be run. Intended for if the SearchRequest requires wrapping
+ * @return The results of the query
+ * @throws InvalidSearchException When the query is malformed or the current state doesn't allow search
+ */
+ protected GroupResponse group(GroupRequest groupRequest, QueryBuilder queryBuilder)
+ throws InvalidSearchException {
+ if (client == null) {
throw new InvalidSearchException("Uninitialized Dao! You must call init() prior to use.");
}
if (groupRequest.getGroups() == null || groupRequest.getGroups().size() == 0) {
throw new InvalidSearchException("At least 1 group must be provided.");
}
final SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
- searchSourceBuilder.query(new QueryStringQueryBuilder(groupRequest.getQuery()));
+ searchSourceBuilder.query(queryBuilder);
searchSourceBuilder.aggregation(getGroupsTermBuilder(groupRequest, 0));
String[] wildcardIndices = wildcardIndices(groupRequest.getIndices());
org.elasticsearch.action.search.SearchRequest request;
org.elasticsearch.action.search.SearchResponse response;
try {
- request = new org.elasticsearch.action.search.SearchRequest(wildcardIndices).source(searchSourceBuilder);
+ request = new org.elasticsearch.action.search.SearchRequest(wildcardIndices)
+ .source(searchSourceBuilder);
response = client.search(request).actionGet();
} catch (SearchPhaseExecutionException e) {
throw new InvalidSearchException("Could not execute search", e);
@@ -197,11 +211,14 @@ public class ElasticsearchDao implements IndexDao {
try {
commonColumnMetadata = getCommonColumnMetadata(groupRequest.getIndices());
} catch (IOException e) {
- throw new InvalidSearchException(String.format("Could not get common column metadata for indices %s", Arrays.toString(groupRequest.getIndices().toArray())));
+ throw new InvalidSearchException(String
+ .format("Could not get common column metadata for indices %s",
+ Arrays.toString(groupRequest.getIndices().toArray())));
}
GroupResponse groupResponse = new GroupResponse();
groupResponse.setGroupedBy(groupRequest.getGroups().get(0).getField());
- groupResponse.setGroupResults(getGroupResults(groupRequest, 0, response.getAggregations(), commonColumnMetadata));
+ groupResponse.setGroupResults(
+ getGroupResults(groupRequest, 0, response.getAggregations(), commonColumnMetadata));
return groupResponse;
}
@@ -246,7 +263,12 @@ public class ElasticsearchDao implements IndexDao {
*/
<T> Optional<T> searchByGuid(String guid, String sensorType,
Function<SearchHit, Optional<T>> callback) {
- QueryBuilder query = QueryBuilders.idsQuery(sensorType + "_doc").ids(guid);
+ QueryBuilder query;
+ if (sensorType != null) {
+ query = QueryBuilders.idsQuery(sensorType + "_doc").ids(guid);
+ } else {
+ query = QueryBuilders.idsQuery().ids(guid);
+ }
SearchRequestBuilder request = client.prepareSearch()
.setQuery(query)
.setSource("message")
@@ -272,30 +294,18 @@ public class ElasticsearchDao implements IndexDao {
@Override
public void update(Document update, Optional<String> index) throws IOException {
- String indexPostfix = ElasticsearchUtils.getIndexFormat(accessConfig.getGlobalConfigSupplier().get()).format(new Date());
+ String indexPostfix = ElasticsearchUtils
+ .getIndexFormat(accessConfig.getGlobalConfigSupplier().get()).format(new Date());
String sensorType = update.getSensorType();
String indexName = ElasticsearchUtils.getIndexName(sensorType, indexPostfix, null);
+ String existingIndex = calculateExistingIndex(update, index, indexPostfix);
- String type = sensorType + "_doc";
- Object ts = update.getTimestamp();
- IndexRequest indexRequest = new IndexRequest(indexName, type, update.getGuid())
- .source(update.getDocument())
- ;
- if(ts != null) {
- indexRequest = indexRequest.timestamp(ts.toString());
- }
- String existingIndex = index.orElse(
- searchByGuid(update.getGuid()
- , sensorType
- , hit -> Optional.ofNullable(hit.getIndex())
- ).orElse(indexName)
- );
- UpdateRequest updateRequest = new UpdateRequest(existingIndex, type, update.getGuid())
- .doc(update.getDocument())
- .upsert(indexRequest)
- ;
-
- org.elasticsearch.action.search.SearchResponse result = client.prepareSearch("test*").setFetchSource(true).setQuery(QueryBuilders.matchAllQuery()).get();
+ UpdateRequest updateRequest = buildUpdateRequest(update, sensorType, indexName, existingIndex);
+
+ org.elasticsearch.action.search.SearchResponse result = client.prepareSearch("test*")
+ .setFetchSource(true)
+ .setQuery(QueryBuilders.matchAllQuery())
+ .get();
result.getHits();
try {
UpdateResponse response = client.update(updateRequest).get();
@@ -303,13 +313,74 @@ public class ElasticsearchDao implements IndexDao {
ShardInfo shardInfo = response.getShardInfo();
int failed = shardInfo.getFailed();
if (failed > 0) {
- throw new IOException("ElasticsearchDao upsert failed: " + Arrays.toString(shardInfo.getFailures()));
+ throw new IOException(
+ "ElasticsearchDao upsert failed: " + Arrays.toString(shardInfo.getFailures()));
}
} catch (Exception e) {
throw new IOException(e.getMessage(), e);
}
}
+ @Override
+ public void batchUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ String indexPostfix = ElasticsearchUtils
+ .getIndexFormat(accessConfig.getGlobalConfigSupplier().get()).format(new Date());
+
+ BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
+
+ // Get the indices we'll actually be using for each Document.
+ for (Map.Entry<Document, Optional<String>> updateEntry : updates.entrySet()) {
+ Document update = updateEntry.getKey();
+ String sensorType = update.getSensorType();
+ String indexName = ElasticsearchUtils.getIndexName(sensorType, indexPostfix, null);
+ String existingIndex = calculateExistingIndex(update, updateEntry.getValue(), indexPostfix);
+ UpdateRequest updateRequest = buildUpdateRequest(
+ update,
+ sensorType,
+ indexName,
+ existingIndex
+ );
+
+ bulkRequestBuilder.add(updateRequest);
+ }
+
+ BulkResponse bulkResponse = bulkRequestBuilder.get();
+ if (bulkResponse.hasFailures()) {
+ LOG.error("Bulk Request has failures: {}", bulkResponse.buildFailureMessage());
+ throw new IOException(
+ "ElasticsearchDao upsert failed: " + bulkResponse.buildFailureMessage());
+ }
+ }
+
+ protected String calculateExistingIndex(Document update, Optional<String> index,
+ String indexPostFix) {
+ String sensorType = update.getSensorType();
+ String indexName = ElasticsearchUtils.getIndexName(sensorType, indexPostFix, null);
+
+ return index.orElse(
+ searchByGuid(update.getGuid(),
+ sensorType,
+ hit -> Optional.ofNullable(hit.getIndex())
+ ).orElse(indexName)
+ );
+ }
+
+ protected UpdateRequest buildUpdateRequest(Document update, String sensorType, String indexName,
+ String existingIndex) {
+ String type = sensorType + "_doc";
+ Object ts = update.getTimestamp();
+ IndexRequest indexRequest = new IndexRequest(indexName, type, update.getGuid())
+ .source(update.getDocument())
+ ;
+ if(ts != null) {
+ indexRequest = indexRequest.timestamp(ts.toString());
+ }
+
+ return new UpdateRequest(existingIndex, type, update.getGuid())
+ .doc(update.getDocument())
+ .upsert(indexRequest);
+ }
+
@SuppressWarnings("unchecked")
@Override
public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> indices) throws IOException {
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/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 3409973..eef134f 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
@@ -29,12 +29,15 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
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 org.apache.commons.collections4.SetUtils;
import org.apache.metron.common.Constants;
import org.apache.metron.indexing.dao.AccessConfig;
import org.apache.metron.indexing.dao.IndexDao;
@@ -59,16 +62,19 @@ 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.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.index.query.support.QueryInnerHitBuilder;
import org.elasticsearch.search.SearchHit;
public class ElasticsearchMetaAlertDao implements MetaAlertDao {
+ private static final String SOURCE_TYPE = Constants.SENSOR_TYPE.replace('.', ':');
private IndexDao indexDao;
private ElasticsearchDao elasticsearchDao;
private String index = METAALERTS_INDEX;
@@ -150,6 +156,7 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
}
@Override
+ @SuppressWarnings("unchecked")
public MetaAlertCreateResponse createMetaAlert(MetaAlertCreateRequest request)
throws InvalidCreateException, IOException {
if (request.getGuidToIndices().isEmpty()) {
@@ -159,12 +166,44 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
throw new InvalidCreateException("MetaAlertCreateRequest must contain UI groups");
}
- // Retrieve the documents going into the meta alert
+ // Retrieve the documents going into the meta alert and build it
MultiGetResponse multiGetResponse = getDocumentsByGuid(request);
Document createDoc = buildCreateDocument(multiGetResponse, request.getGroups());
+ MetaScores metaScores = calculateMetaScores(createDoc);
+ createDoc.getDocument().putAll(metaScores.getMetaScores());
+ createDoc.getDocument().put(threatTriageField, metaScores.getMetaScores().get(threatSort));
+ // Add source type to be consistent with other sources and allow filtering
+ createDoc.getDocument().put("source:type", MetaAlertDao.METAALERT_TYPE);
+
+ // Start a list of updates / inserts we need to run
+ Map<Document, Optional<String>> updates = new HashMap<>();
+ updates.put(createDoc, Optional.of(MetaAlertDao.METAALERTS_INDEX));
try {
- handleMetaUpdate(createDoc, Optional.of(METAALERTS_INDEX));
+ // We need to update the associated alerts with the new meta alerts, making sure existing
+ // links are maintained.
+ List<String> metaAlertField;
+ for (MultiGetItemResponse itemResponse : multiGetResponse) {
+ metaAlertField = new ArrayList<>();
+ GetResponse response = itemResponse.getResponse();
+ if (response.isExists()) {
+ List<String> alertField = (List<String>) response.getSourceAsMap()
+ .get(MetaAlertDao.METAALERT_FIELD);
+ if (alertField != null) {
+ metaAlertField.addAll(alertField);
+ }
+ }
+ metaAlertField.add(createDoc.getGuid());
+
+ Document alertUpdate = buildAlertUpdate(response.getId(),
+ (String) response.getSource().get(SOURCE_TYPE), metaAlertField,
+ (Long) response.getSourceAsMap().get("_timestamp"));
+ updates.put(alertUpdate, Optional.of(itemResponse.getIndex()));
+ }
+
+ // Kick off any updates.
+ indexDaoUpdate(updates);
+
MetaAlertCreateResponse createResponse = new MetaAlertCreateResponse();
createResponse.setCreated(true);
createResponse.setGuid(createDoc.getGuid());
@@ -186,11 +225,13 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
)
)
)
- // Ensures that it's a meta alert with active status or that it's an alert (signified by having no status field)
+ // Ensures that it's a meta alert with active status or that it's an alert (signified by
+ // having no status field)
.must(boolQuery()
.should(termQuery(MetaAlertDao.STATUS_FIELD, MetaAlertStatus.ACTIVE.getStatusString()))
.should(boolQuery().mustNot(existsQuery(MetaAlertDao.STATUS_FIELD)))
)
+ .mustNot(existsQuery(MetaAlertDao.METAALERT_FIELD))
);
return elasticsearchDao.search(searchRequest, qb);
}
@@ -204,7 +245,7 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
public void update(Document update, Optional<String> index) throws IOException {
if (METAALERT_TYPE.equals(update.getSensorType())) {
// We've been passed an update to the meta alert.
- handleMetaUpdate(update, index);
+ handleMetaUpdate(update);
} else {
// We need to update an alert itself. Only that portion of the update can be delegated.
// We still need to get meta alerts potentially associated with it and update.
@@ -222,6 +263,11 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
}
}
+ @Override
+ public void batchUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ throw new UnsupportedOperationException("Meta alerts do not allow for bulk updates");
+ }
+
/**
* Given an alert GUID, retrieve all associated meta alerts.
* @param guid The GUID of the child alert
@@ -295,19 +341,199 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
/**
* Process an update to a meta alert itself.
* @param update The update Document to be applied
- * @param index The optional index to update to
* @throws IOException If there's a problem running the update
*/
- protected void handleMetaUpdate(Document update, Optional<String> index) throws IOException {
- // We have an update to a meta alert itself
- // If we've updated the alerts field (i.e add/remove), recalculate meta alert scores.
+ protected void handleMetaUpdate(Document update) throws IOException {
+ Map<Document, Optional<String>> updates = new HashMap<>();
+
+ if (update.getDocument().containsKey(MetaAlertDao.STATUS_FIELD)) {
+ // Update all associated alerts to maintain the meta alert link properly
+ updates.putAll(buildStatusAlertUpdates(update));
+ }
if (update.getDocument().containsKey(MetaAlertDao.ALERT_FIELD)) {
- MetaScores metaScores = calculateMetaScores(update);
- update.getDocument().putAll(metaScores.getMetaScores());
- update.getDocument().put(threatTriageField, metaScores.getMetaScores().get(threatSort));
+ // If the alerts field changes (i.e. add/remove alert), update all affected alerts to
+ // maintain the meta alert link properly.
+ updates.putAll(buildAlertFieldUpdates(update));
+ }
+
+ // Run meta alert update.
+ updates.put(update, Optional.of(index));
+ indexDaoUpdate(updates);
+ }
+
+ /**
+ * Calls the single update variant if there's only one update, otherwise calls batch.
+ * @param updates The list of updates to run
+ * @throws IOException If there's an update error
+ */
+ protected void indexDaoUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ if (updates.size() == 1) {
+ Entry<Document, Optional<String>> singleUpdate = updates.entrySet().iterator().next();
+ indexDao.update(singleUpdate.getKey(), singleUpdate.getValue());
+ } else if (updates.size() > 1) {
+ indexDao.batchUpdate(updates);
+ } // else we have no updates, so don't do anything
+ }
+
+ protected Map<Document, Optional<String>> buildStatusAlertUpdates(Document update)
+ throws IOException {
+ Map<Document, Optional<String>> updates = new HashMap<>();
+ List<Map<String, Object>> alerts = getAllAlertsForMetaAlert(update);
+ for (Map<String, Object> alert : alerts) {
+ // Retrieve the associated alert, so we can update the array
+ List<String> metaAlertField = new ArrayList<>();
+ @SuppressWarnings("unchecked")
+ List<String> alertField = (List<String>) alert.get(MetaAlertDao.METAALERT_FIELD);
+ if (alertField != null) {
+ metaAlertField.addAll(alertField);
+ }
+ String status = (String) update.getDocument().get(MetaAlertDao.STATUS_FIELD);
+
+ Document alertUpdate = null;
+ String alertGuid = (String) alert.get(Constants.GUID);
+ // If we're making it active add add the meta alert guid for every alert.
+ if (MetaAlertStatus.ACTIVE.getStatusString().equals(status)
+ && !metaAlertField.contains(update.getGuid())) {
+ metaAlertField.add(update.getGuid());
+ alertUpdate = buildAlertUpdate(
+ alertGuid,
+ (String) alert.get(SOURCE_TYPE),
+ metaAlertField,
+ (Long) alert.get("_timestamp")
+ );
+ }
+
+ // If we're making it inactive, remove the meta alert guid from every alert.
+ if (MetaAlertStatus.INACTIVE.getStatusString().equals(status)
+ && metaAlertField.remove(update.getGuid())) {
+ alertUpdate = buildAlertUpdate(
+ alertGuid,
+ (String) alert.get(SOURCE_TYPE),
+ metaAlertField,
+ (Long) alert.get("_timestamp")
+ );
+ }
+
+ // Only run an alert update if we have an actual update.
+ if (alertUpdate != null) {
+ updates.put(alertUpdate, Optional.empty());
+ }
+ }
+ return updates;
+ }
+
+ protected Map<Document, Optional<String>> buildAlertFieldUpdates(Document update)
+ throws IOException {
+ Map<Document, Optional<String>> updates = new HashMap<>();
+ // If we've updated the alerts field (i.e add/remove), recalculate meta alert scores and
+ // the metaalerts fields for updating the children alerts.
+ MetaScores metaScores = calculateMetaScores(update);
+ update.getDocument().putAll(metaScores.getMetaScores());
+ update.getDocument().put(threatTriageField, metaScores.getMetaScores().get(threatSort));
+
+ // Get the set of GUIDs that are in the new version.
+ Set<String> updateGuids = new HashSet<>();
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> updateAlerts = (List<Map<String, Object>>) update.getDocument()
+ .get(MetaAlertDao.ALERT_FIELD);
+ for (Map<String, Object> alert : updateAlerts) {
+ updateGuids.add((String) alert.get(Constants.GUID));
+ }
+
+ // Get the set of GUIDs from the old version
+ List<Map<String, Object>> alerts = getAllAlertsForMetaAlert(update);
+ Set<String> currentGuids = new HashSet<>();
+ for (Map<String, Object> alert : alerts) {
+ currentGuids.add((String) alert.get(Constants.GUID));
}
- indexDao.update(update, index);
+ // Get both set differences, so we know what's been added and removed.
+ Set<String> removedGuids = SetUtils.difference(currentGuids, updateGuids);
+ Set<String> addedGuids = SetUtils.difference(updateGuids, currentGuids);
+
+ Document alertUpdate;
+
+ // Handle any removed GUIDs
+ for (String guid : removedGuids) {
+ // Retrieve the associated alert, so we can update the array
+ Document alert = elasticsearchDao.getLatest(guid, null);
+ List<String> metaAlertField = new ArrayList<>();
+ @SuppressWarnings("unchecked")
+ List<String> alertField = (List<String>) alert.getDocument()
+ .get(MetaAlertDao.METAALERT_FIELD);
+ if (alertField != null) {
+ metaAlertField.addAll(alertField);
+ }
+ if (metaAlertField.remove(update.getGuid())) {
+ alertUpdate = buildAlertUpdate(guid, alert.getSensorType(), metaAlertField,
+ alert.getTimestamp());
+ updates.put(alertUpdate, Optional.empty());
+ }
+ }
+
+ // Handle any added GUIDs
+ for (String guid : addedGuids) {
+ // Retrieve the associated alert, so we can update the array
+ Document alert = elasticsearchDao.getLatest(guid, null);
+ List<String> metaAlertField = new ArrayList<>();
+ @SuppressWarnings("unchecked")
+ List<String> alertField = (List<String>) alert.getDocument()
+ .get(MetaAlertDao.METAALERT_FIELD);
+ if (alertField != null) {
+ metaAlertField.addAll(alertField);
+ }
+ metaAlertField.add(update.getGuid());
+ alertUpdate = buildAlertUpdate(guid, alert.getSensorType(), metaAlertField,
+ alert.getTimestamp());
+ updates.put(alertUpdate, Optional.empty());
+ }
+
+ return updates;
+ }
+
+ @SuppressWarnings("unchecked")
+ protected List<Map<String, Object>> getAllAlertsForMetaAlert(Document update) throws IOException {
+ Document latest = indexDao.getLatest(update.getGuid(), MetaAlertDao.METAALERT_TYPE);
+ if (latest == null) {
+ return new ArrayList<>();
+ }
+ List<String> guids = new ArrayList<>();
+ List<Map<String, Object>> latestAlerts = (List<Map<String, Object>>) latest.getDocument()
+ .get(MetaAlertDao.ALERT_FIELD);
+ for (Map<String, Object> alert : latestAlerts) {
+ guids.add((String) alert.get(Constants.GUID));
+ }
+
+ List<Map<String, Object>> alerts = new ArrayList<>();
+ QueryBuilder query = QueryBuilders.idsQuery().ids(guids);
+ SearchRequestBuilder request = elasticsearchDao.getClient().prepareSearch()
+ .setQuery(query);
+ org.elasticsearch.action.search.SearchResponse response = request.get();
+ for (SearchHit hit : response.getHits().getHits()) {
+ alerts.add(hit.sourceAsMap());
+ }
+ return alerts;
+ }
+
+ /**
+ * Builds an update Document for updating the meta alerts list.
+ * @param alertGuid The GUID of the alert to update
+ * @param sensorType The sensor type to update
+ * @param metaAlertField The new metaAlertList to use
+ * @return The update Document
+ */
+ protected Document buildAlertUpdate(String alertGuid, String sensorType,
+ List<String> metaAlertField, Long timestamp) {
+ Document alertUpdate;
+ Map<String, Object> document = new HashMap<>();
+ document.put(MetaAlertDao.METAALERT_FIELD, metaAlertField);
+ alertUpdate = new Document(
+ document,
+ alertGuid,
+ sensorType,
+ timestamp
+ );
+ return alertUpdate;
}
/**
@@ -360,7 +586,11 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
@Override
public GroupResponse group(GroupRequest groupRequest) throws InvalidSearchException {
- return indexDao.group(groupRequest);
+ // Wrap the query to hide any alerts already contained in meta alerts
+ QueryBuilder qb = QueryBuilders.boolQuery()
+ .must(new QueryStringQueryBuilder(groupRequest.getQuery()))
+ .mustNot(existsQuery(MetaAlertDao.METAALERT_FIELD));
+ return elasticsearchDao.group(groupRequest, qb);
}
/**
@@ -406,6 +636,7 @@ public class ElasticsearchMetaAlertDao implements MetaAlertDao {
builder.startArray(ALERT_FIELD);
Map<String, Object> hitAlerts = hit.sourceAsMap();
+ @SuppressWarnings("unchecked")
List<Map<String, Object>> alertHits = (List<Map<String, Object>>) hitAlerts.get(ALERT_FIELD);
for (Map<String, Object> alertHit : alertHits) {
Map<String, Object> docMap = alertHit;
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDaoTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDaoTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDaoTest.java
index 5d6f4e0..9a02854 100644
--- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDaoTest.java
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchMetaAlertDaoTest.java
@@ -169,7 +169,8 @@ public class ElasticsearchMetaAlertDaoTest {
innerAlertSourceTwo.put(MetaAlertDao.THREAT_FIELD_DEFAULT, threatValueTwo);
Map<String, Object> innerHits = new HashMap<>();
- innerHits.put(MetaAlertDao.ALERT_FIELD, Arrays.asList(innerAlertSourceOne, innerAlertSourceTwo));
+ innerHits
+ .put(MetaAlertDao.ALERT_FIELD, Arrays.asList(innerAlertSourceOne, innerAlertSourceTwo));
when(metaHit.sourceAsMap()).thenReturn(innerHits);
// Construct the updated Document
@@ -218,6 +219,10 @@ public class ElasticsearchMetaAlertDaoTest {
}
@Override
+ public void batchUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ }
+
+ @Override
public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> indices)
throws IOException {
return null;
@@ -393,47 +398,95 @@ public class ElasticsearchMetaAlertDaoTest {
}
@Test
- public void testHandleMetaUpdateNonAlert() throws IOException {
- ElasticsearchDao mockEsDao= mock(ElasticsearchDao.class);
+ public void testHandleMetaUpdateNonAlertNonStatus() throws IOException {
+ ElasticsearchDao mockEsDao = mock(ElasticsearchDao.class);
Map<String, Object> docMap = new HashMap<>();
- docMap.put(MetaAlertDao.STATUS_FIELD, MetaAlertStatus.ACTIVE.getStatusString());
+ docMap.put("test", "value");
Document update = new Document(docMap, "guid", MetaAlertDao.METAALERT_TYPE, 0L);
ElasticsearchMetaAlertDao metaAlertDao = new ElasticsearchMetaAlertDao(mockEsDao);
- metaAlertDao.handleMetaUpdate(update, Optional.of(MetaAlertDao.METAALERTS_INDEX));
+ metaAlertDao.handleMetaUpdate(update);
verify(mockEsDao, times(1))
.update(update, Optional.of(MetaAlertDao.METAALERTS_INDEX));
}
@Test
public void testHandleMetaUpdateAlert() throws IOException {
- ElasticsearchDao mockEsDao= mock(ElasticsearchDao.class);
-
- Map<String, Object> alertMap = new HashMap<>();
- alertMap.put(MetaAlertDao.THREAT_FIELD_DEFAULT, 10.0d);
+ // The child alert of the meta alert
+ Map<String, Object> alertMapBefore = new HashMap<>();
+ alertMapBefore.put(MetaAlertDao.THREAT_FIELD_DEFAULT, 10.0d);
+ String guidAlert = "guid_alert";
+ alertMapBefore.put(Constants.GUID, guidAlert);
List<Map<String, Object>> alertList = new ArrayList<>();
- alertList.add(alertMap);
+ alertList.add(alertMapBefore);
+ String alertSensorType = "alert_sensor";
+ Document alertBefore = new Document(
+ alertMapBefore,
+ guidAlert,
+ alertSensorType,
+ 0L
+ );
- Map<String, Object> docMapBefore = new HashMap<>();
- docMapBefore.put(MetaAlertDao.ALERT_FIELD, alertList);
- Document before = new Document(docMapBefore, "guid", MetaAlertDao.METAALERT_TYPE, 0L);
-
- Map<String, Object> docMapAfter = new HashMap<>();
- docMapAfter.putAll(docMapBefore);
- docMapAfter.put("average", 10.0d);
- docMapAfter.put("min", 10.0d);
- docMapAfter.put("median", 10.0d);
- docMapAfter.put("max", 10.0d);
- docMapAfter.put("count", 1L);
- docMapAfter.put("sum", 10.0d);
- docMapAfter.put(MetaAlertDao.THREAT_FIELD_DEFAULT, 10.0d);
- Document after = new Document(docMapAfter, "guid", MetaAlertDao.METAALERT_TYPE, 0L);
+ // The original meta alert. It contains the alert we previously constructed.
+ Map<String, Object> metaMapBefore = new HashMap<>();
+ String metaGuid = "guid_meta";
+ metaMapBefore.putAll(alertBefore.getDocument());
+ metaMapBefore.put(MetaAlertDao.ALERT_FIELD, alertList);
+ metaMapBefore.put(Constants.GUID, metaGuid);
+ Document metaBefore = new Document(
+ metaMapBefore,
+ metaGuid,
+ MetaAlertDao.METAALERT_TYPE,
+ 0L
+ );
+
+ // Build the Documents we expect to see from updates
+ // Build the after alert. Don't add the original fields: This is only an update.
+ // The new field is the link to the meta alert.
+ Map<String, Object> alertMapAfter = new HashMap<>();
+ List<String> metaAlertField = new ArrayList<>();
+ metaAlertField.add(metaGuid);
+ alertMapAfter.put(MetaAlertDao.METAALERT_FIELD, metaAlertField);
+ Document alertAfter = new Document(
+ alertMapAfter,
+ guidAlert,
+ alertSensorType,
+ 0L
+ );
+ // Build the meta alert after. This'll be a replace, so add the original fields plus the
+ // threat fields
+ Map<String, Object> metaMapAfter = new HashMap<>();
+ metaMapAfter.putAll(metaMapBefore);
+ metaMapAfter.put("average", 10.0d);
+ metaMapAfter.put("min", 10.0d);
+ metaMapAfter.put("median", 10.0d);
+ metaMapAfter.put("max", 10.0d);
+ metaMapAfter.put("count", 1L);
+ metaMapAfter.put("sum", 10.0d);
+ metaMapAfter.put(MetaAlertDao.THREAT_FIELD_DEFAULT, 10.0d);
+
+ Document metaAfter = new Document(
+ metaMapAfter,
+ metaGuid,
+ MetaAlertDao.METAALERT_TYPE,
+ 0L
+ );
+
+ // Build the method calls we'd expect to see.
+ Map<Document, Optional<String>> updates = new HashMap<>();
+ updates.put(metaAfter, Optional.of(MetaAlertDao.METAALERTS_INDEX));
+ updates.put(alertAfter, Optional.empty());
+
+ // Build a mock ElasticsearchDao to track interactions. Actual runs are in integration tests
+ ElasticsearchDao mockEsDao = mock(ElasticsearchDao.class);
ElasticsearchMetaAlertDao metaAlertDao = new ElasticsearchMetaAlertDao(mockEsDao);
- metaAlertDao.handleMetaUpdate(before, Optional.of(MetaAlertDao.METAALERTS_INDEX));
+ when(mockEsDao.getLatest(guidAlert, null)).thenReturn(alertBefore);
+ metaAlertDao.handleMetaUpdate(metaBefore);
- verify(mockEsDao, times(1))
- .update(after, Optional.of(MetaAlertDao.METAALERTS_INDEX));
+ // Validate we're calling what we need to with what we expect.
+ verify(mockEsDao, times(1)).getLatest(guidAlert, null);
+ verify(mockEsDao, times(1)).batchUpdate(updates);
}
}
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/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 b13032f..27e5566 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
@@ -30,6 +30,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -46,14 +47,21 @@ import org.apache.metron.indexing.dao.IndexDao;
import org.apache.metron.indexing.dao.MetaAlertDao;
import org.apache.metron.indexing.dao.metaalert.MetaAlertCreateRequest;
import org.apache.metron.indexing.dao.metaalert.MetaAlertCreateResponse;
+import org.apache.metron.indexing.dao.search.Group;
+import org.apache.metron.indexing.dao.search.GroupRequest;
+import org.apache.metron.indexing.dao.search.GroupResponse;
+import org.apache.metron.indexing.dao.search.GroupResult;
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.search.SortField;
import org.apache.metron.indexing.dao.update.Document;
import org.apache.metron.indexing.dao.update.PatchRequest;
import org.apache.metron.indexing.dao.update.ReplaceRequest;
+import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -72,8 +80,166 @@ public class ElasticsearchMetaAlertIntegrationTest {
private static MetaAlertDao metaDao;
private static ElasticSearchComponent es;
+ /**
+ {
+ "guid": "update_metaalert_alert_0",
+ "source:type": "test",
+ "field": "value 0"
+ }
+ */
+ @Multiline
+ public static String updateMetaAlertAlert0;
+
+ /**
+ {
+ "guid": "update_metaalert_alert_1",
+ "source:type": "test",
+ "field":"value 1"
+ }
+ */
+ @Multiline
+ public static String updateMetaAlertAlert1;
+
+ /**
+ {
+ "guid": "update_metaalert_alert_0",
+ "patch": [
+ {
+ "op": "add",
+ "path": "/field",
+ "value": "patched value 0"
+ }
+ ],
+ "sensorType": "test"
+ }
+ */
+ @Multiline
+ public static String updateMetaAlertPatchRequest;
+
+ /**
+ {
+ "guid": "update_metaalert_alert_0",
+ "replacement": {
+ "guid": "update_metaalert_alert_0",
+ "source:type": "test",
+ "field": "replaced value 0"
+ },
+ "sensorType": "test"
+ }
+ */
+ @Multiline
+ public static String updateMetaAlertReplaceRequest;
+
+ /**
+ {
+ "guid": "active_metaalert",
+ "source:type": "metaalert",
+ "alert": [],
+ "status": "active",
+ "timestamp": 0
+ }
+ */
+ @Multiline
+ public static String activeMetaAlert;
+
+ /**
+ {
+ "guid": "inactive_metaalert",
+ "source:type": "metaalert",
+ "alert": [],
+ "status": "inactive"
+ }
+ */
+ @Multiline
+ public static String inactiveMetaAlert;
+
+ /**
+ {
+ "guid": "search_by_nested_alert_active_0",
+ "source:type": "test",
+ "ip_src_addr": "192.168.1.1",
+ "ip_src_port": 8010,
+ "metaalerts": ["active_metaalert"],
+ "timestamp": 0
+ }
+ */
+ @Multiline
+ public static String searchByNestedAlertActive0;
+
+ /**
+ {
+ "guid": "search_by_nested_alert_active_1",
+ "source:type": "test",
+ "ip_src_addr": "192.168.1.2",
+ "ip_src_port": 8009,
+ "metaalerts": ["active_metaalert"],
+ "timestamp": 0
+ }
+ */
+ @Multiline
+ public static String searchByNestedAlertActive1;
+
+ /**
+ {
+ "guid": "search_by_nested_alert_inactive_0",
+ "source:type": "test",
+ "ip_src_addr": "192.168.1.3",
+ "ip_src_port": 8008
+ }
+ */
+ @Multiline
+ public static String searchByNestedAlertInactive0;
+
+ /**
+ {
+ "guid": "search_by_nested_alert_inactive_1",
+ "source:type": "test",
+ "ip_src_addr": "192.168.1.4",
+ "ip_src_port": 8007
+ }
+ */
+ @Multiline
+ public static String searchByNestedAlertInactive1;
+
+ /**
+ {
+ "properties": {
+ "alert": {
+ "type": "nested"
+ }
+ }
+ }
+ */
+ @Multiline
+ public static String nestedAlertMapping;
+
+ /**
+ {
+ "guid": "group_by_child_alert",
+ "source:type": "test",
+ "ip_src_addr": "192.168.1.1",
+ "ip_src_port": 8010,
+ "score_field": 1,
+ "metaalerts": ["active_metaalert"]
+ }
+ */
+ @Multiline
+ public static String groupByChildAlert;
+
+ /**
+ {
+ "guid": "group_by_standalone_alert",
+ "source:type": "test",
+ "ip_src_addr": "192.168.1.1",
+ "ip_src_port": 8010,
+ "score_field": 10
+ }
+ */
+ @Multiline
+ public static String groupByStandaloneAlert;
+
@BeforeClass
- public static void setup() throws Exception {
+ public static void setupBefore() throws Exception {
// setup the client
es = new ElasticSearchComponent.Builder()
.withHttpPort(9211)
@@ -81,9 +247,6 @@ public class ElasticsearchMetaAlertIntegrationTest {
.build();
es.start();
- es.createIndexWithMapping(MetaAlertDao.METAALERTS_INDEX, MetaAlertDao.METAALERT_DOC,
- buildMetaMappingSource());
-
AccessConfig accessConfig = new AccessConfig();
Map<String, Object> globalConfig = new HashMap<String, Object>() {
{
@@ -95,12 +258,19 @@ public class ElasticsearchMetaAlertIntegrationTest {
};
accessConfig.setMaxSearchResults(1000);
accessConfig.setGlobalConfigSupplier(() -> globalConfig);
+ accessConfig.setMaxSearchGroups(100);
esDao = new ElasticsearchDao();
esDao.init(accessConfig);
metaDao = new ElasticsearchMetaAlertDao(esDao);
}
+ @Before
+ public void setup() throws IOException {
+ es.createIndexWithMapping(MetaAlertDao.METAALERTS_INDEX, MetaAlertDao.METAALERT_DOC,
+ buildMetaMappingSource());
+ }
+
@AfterClass
public static void teardown() {
if (es != null) {
@@ -108,6 +278,11 @@ public class ElasticsearchMetaAlertIntegrationTest {
}
}
+ @After
+ public void reset() {
+ es.reset();
+ }
+
protected static String buildMetaMappingSource() throws IOException {
return jsonBuilder().prettyPrint()
.startObject()
@@ -313,34 +488,17 @@ public class ElasticsearchMetaAlertIntegrationTest {
}
}
- /**
- {
- "guid": "active_metaalert",
- "source:type": "metaalert",
- "alert": [],
- "status": "active"
- }
- */
- @Multiline
- public static String activeMetaAlert;
-
- /**
- {
- "guid": "inactive_metaalert",
- "source:type": "metaalert",
- "alert": [],
- "status": "inactive"
- }
- */
- @Multiline
- public static String inactiveMetaAlert;
@Test
public void shouldSearchByStatus() throws Exception {
List<Map<String, Object>> metaInputData = new ArrayList<>();
- Map<String, Object> activeMetaAlertJSON = JSONUtils.INSTANCE.load(activeMetaAlert, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> activeMetaAlertJSON = JSONUtils.INSTANCE
+ .load(activeMetaAlert, new TypeReference<Map<String, Object>>() {
+ });
metaInputData.add(activeMetaAlertJSON);
- Map<String, Object> inactiveMetaAlertJSON = JSONUtils.INSTANCE.load(inactiveMetaAlert, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> inactiveMetaAlertJSON = JSONUtils.INSTANCE
+ .load(inactiveMetaAlert, new TypeReference<Map<String, Object>>() {
+ });
metaInputData.add(inactiveMetaAlertJSON);
// We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
@@ -354,189 +512,355 @@ public class ElasticsearchMetaAlertIntegrationTest {
setIndices(Collections.singletonList(MetaAlertDao.METAALERT_TYPE));
setFrom(0);
setSize(5);
- setSort(Collections.singletonList(new SortField(){{ setField(Constants.GUID); }}));
+ setSort(Collections.singletonList(new SortField() {{
+ setField(Constants.GUID);
+ }}));
}
});
Assert.assertEquals(1, searchResponse.getTotal());
- Assert.assertEquals(MetaAlertStatus.ACTIVE.getStatusString(), searchResponse.getResults().get(0).getSource().get(MetaAlertDao.STATUS_FIELD));
+ Assert.assertEquals(MetaAlertStatus.ACTIVE.getStatusString(),
+ searchResponse.getResults().get(0).getSource().get(MetaAlertDao.STATUS_FIELD));
}
- /**
- {
- "guid": "search_by_nested_alert_active_0",
- "source:type": "test",
- "ip_src_addr": "192.168.1.1",
- "ip_src_port": 8010
- }
- */
- @Multiline
- public static String searchByNestedAlertActive0;
-
- /**
- {
- "guid": "search_by_nested_alert_inactive_1",
- "source:type": "test",
- "ip_src_addr": "192.168.1.2",
- "ip_src_port": 8009
- }
- */
- @Multiline
- public static String searchByNestedAlertActive1;
-
- /**
- {
- "guid": "search_by_nested_alert_inactive_0",
- "source:type": "test",
- "ip_src_addr": "192.168.1.3",
- "ip_src_port": 8008
- }
- */
- @Multiline
- public static String searchByNestedAlertInactive0;
-
- /**
- {
- "guid": "search_by_nested_alert_inactive_1",
- "source:type": "test",
- "ip_src_addr": "192.168.1.4",
- "ip_src_port": 8007
- }
- */
- @Multiline
- public static String searchByNestedAlertInactive1;
@Test
public void shouldSearchByNestedAlert() throws Exception {
// Create alerts
List<Map<String, Object>> alerts = new ArrayList<>();
- Map<String, Object> searchByNestedAlertActive0JSON = JSONUtils.INSTANCE.load(searchByNestedAlertActive0, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> searchByNestedAlertActive0JSON = JSONUtils.INSTANCE
+ .load(searchByNestedAlertActive0, new TypeReference<Map<String, Object>>() {
+ });
alerts.add(searchByNestedAlertActive0JSON);
- Map<String, Object> searchByNestedAlertActive1JSON = JSONUtils.INSTANCE.load(searchByNestedAlertActive1, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> searchByNestedAlertActive1JSON = JSONUtils.INSTANCE
+ .load(searchByNestedAlertActive1, new TypeReference<Map<String, Object>>() {
+ });
alerts.add(searchByNestedAlertActive1JSON);
- Map<String, Object> searchByNestedAlertInactive0JSON = JSONUtils.INSTANCE.load(searchByNestedAlertInactive0, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> searchByNestedAlertInactive0JSON = JSONUtils.INSTANCE
+ .load(searchByNestedAlertInactive0, new TypeReference<Map<String, Object>>() {
+ });
alerts.add(searchByNestedAlertInactive0JSON);
- Map<String, Object> searchByNestedAlertInactive1JSON = JSONUtils.INSTANCE.load(searchByNestedAlertInactive1, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> searchByNestedAlertInactive1JSON = JSONUtils.INSTANCE
+ .load(searchByNestedAlertInactive1, new TypeReference<Map<String, Object>>() {
+ });
alerts.add(searchByNestedAlertInactive1JSON);
elasticsearchAdd(alerts, INDEX, SENSOR_NAME);
// Wait for updates to persist
- findUpdatedDoc(searchByNestedAlertInactive1JSON, "search_by_nested_alert_inactive_1", SENSOR_NAME);
+ findUpdatedDoc(searchByNestedAlertActive0JSON, "search_by_nested_alert_active_0",
+ SENSOR_NAME);
+ findUpdatedDoc(searchByNestedAlertActive1JSON, "search_by_nested_alert_active_1",
+ SENSOR_NAME);
+ findUpdatedDoc(searchByNestedAlertInactive0JSON, "search_by_nested_alert_inactive_0",
+ SENSOR_NAME);
+ findUpdatedDoc(searchByNestedAlertInactive1JSON, "search_by_nested_alert_inactive_1",
+ SENSOR_NAME);
+
+ // Put the nested type into the test index, so that it'll match appropriately
+ ((ElasticsearchDao) esDao).getClient().admin().indices().preparePutMapping(INDEX)
+ .setType("test_doc")
+ .setSource(nestedAlertMapping)
+ .get();
// Create metaalerts
- Map<String, Object> activeMetaAlertJSON = JSONUtils.INSTANCE.load(activeMetaAlert, new TypeReference<Map<String, Object>>() {});
- activeMetaAlertJSON.put("alert", Arrays.asList(searchByNestedAlertActive0JSON, searchByNestedAlertActive1JSON));
- Map<String, Object> inactiveMetaAlertJSON = JSONUtils.INSTANCE.load(inactiveMetaAlert, new TypeReference<Map<String, Object>>() {});
- inactiveMetaAlertJSON.put("alert", Arrays.asList(searchByNestedAlertInactive0JSON, searchByNestedAlertInactive1JSON));
+ Map<String, Object> activeMetaAlertJSON = JSONUtils.INSTANCE
+ .load(activeMetaAlert, new TypeReference<Map<String, Object>>() {
+ });
+ activeMetaAlertJSON.put("alert",
+ Arrays.asList(searchByNestedAlertActive0JSON, searchByNestedAlertActive1JSON));
+ Map<String, Object> inactiveMetaAlertJSON = JSONUtils.INSTANCE
+ .load(inactiveMetaAlert, new TypeReference<Map<String, Object>>() {
+ });
+ inactiveMetaAlertJSON.put("alert",
+ Arrays.asList(searchByNestedAlertInactive0JSON, searchByNestedAlertInactive1JSON));
// We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
- elasticsearchAdd(Arrays.asList(activeMetaAlertJSON, inactiveMetaAlertJSON), MetaAlertDao.METAALERTS_INDEX, MetaAlertDao.METAALERT_TYPE);
+ elasticsearchAdd(Arrays.asList(activeMetaAlertJSON, inactiveMetaAlertJSON),
+ MetaAlertDao.METAALERTS_INDEX, MetaAlertDao.METAALERT_TYPE);
// Wait for updates to persist
findUpdatedDoc(activeMetaAlertJSON, "active_metaalert", MetaAlertDao.METAALERT_TYPE);
SearchResponse searchResponse = metaDao.search(new SearchRequest() {
{
- setQuery("(ip_src_addr:192.168.1.1 AND ip_src_port:8009) OR (alert.ip_src_addr:192.168.1.1 AND alert.ip_src_port:8009)");
+ setQuery(
+ "(ip_src_addr:192.168.1.1 AND ip_src_port:8009) OR (alert.ip_src_addr:192.168.1.1 AND alert.ip_src_port:8009)");
setIndices(Collections.singletonList(MetaAlertDao.METAALERT_TYPE));
setFrom(0);
setSize(5);
- setSort(Collections.singletonList(new SortField(){{ setField(Constants.GUID); }}));
+ setSort(Collections.singletonList(new SortField() {
+ {
+ setField(Constants.GUID);
+ }
+ }));
}
});
// Should not have results because nested alerts shouldn't be flattened
Assert.assertEquals(0, searchResponse.getTotal());
+ // Query against all indices. Only the single active meta alert should be returned.
+ // The child alerts should be hidden.
searchResponse = metaDao.search(new SearchRequest() {
{
- setQuery("(ip_src_addr:192.168.1.1 AND ip_src_port:8010) OR (alert.ip_src_addr:192.168.1.1 AND alert.ip_src_port:8010)");
- setIndices(Collections.singletonList(MetaAlertDao.METAALERT_TYPE));
+ setQuery(
+ "(ip_src_addr:192.168.1.1 AND ip_src_port:8010)"
+ + " OR (alert.ip_src_addr:192.168.1.1 AND alert.ip_src_port:8010)");
+ setIndices(Collections.singletonList("*"));
setFrom(0);
setSize(5);
- setSort(Collections.singletonList(new SortField(){{ setField(Constants.GUID); }}));
+ setSort(Collections.singletonList(new SortField() {
+ {
+ setField(Constants.GUID);
+ }
+ }));
}
});
+
// Nested query should match a nested alert
Assert.assertEquals(1, searchResponse.getTotal());
- Assert.assertEquals("active_metaalert", searchResponse.getResults().get(0).getSource().get("guid"));
+ Assert.assertEquals("active_metaalert",
+ searchResponse.getResults().get(0).getSource().get("guid"));
+
+ // Query against all indices. The child alert has no actual attached meta alerts, and should
+ // be returned on its own.
+ searchResponse = metaDao.search(new SearchRequest() {
+ {
+ setQuery(
+ "(ip_src_addr:192.168.1.3 AND ip_src_port:8008)"
+ + " OR (alert.ip_src_addr:192.168.1.3 AND alert.ip_src_port:8008)");
+ setIndices(Collections.singletonList("*"));
+ setFrom(0);
+ setSize(5);
+ setSort(Collections.singletonList(new SortField() {
+ {
+ setField(Constants.GUID);
+ }
+ }));
+ }
+ });
+
+ // Nested query should match a plain alert
+ Assert.assertEquals(1, searchResponse.getTotal());
+ Assert.assertEquals("search_by_nested_alert_inactive_0",
+ searchResponse.getResults().get(0).getSource().get("guid"));
}
- /**
- {
- "guid": "update_metaalert_alert_0",
- "source:type": "test",
- "field": "value 0"
- }
- */
- @Multiline
- public static String updateMetaAlertAlert0;
+ @Test
+ public void shouldGroupHidesAlert() throws Exception {
+ // Create alerts
+ List<Map<String, Object>> alerts = new ArrayList<>();
+ Map<String, Object> groupByChildAlertJson = JSONUtils.INSTANCE
+ .load(groupByChildAlert, new TypeReference<Map<String, Object>>() {
+ });
+ alerts.add(groupByChildAlertJson);
+ Map<String, Object> groupByStandaloneAlertJson = JSONUtils.INSTANCE
+ .load(groupByStandaloneAlert, new TypeReference<Map<String, Object>>() {
+ });
+ alerts.add(groupByStandaloneAlertJson);
+ elasticsearchAdd(alerts, INDEX, SENSOR_NAME);
+ // Wait for updates to persist
+ findUpdatedDoc(groupByChildAlertJson, "group_by_child_alert",
+ SENSOR_NAME);
+ findUpdatedDoc(groupByStandaloneAlertJson, "group_by_standalone_alert",
+ SENSOR_NAME);
+
+ // Put the nested type into the test index, so that it'll match appropriately
+ ((ElasticsearchDao) esDao).getClient().admin().indices().preparePutMapping(INDEX)
+ .setType("test_doc")
+ .setSource(nestedAlertMapping)
+ .get();
+
+ // Don't need any meta alerts to actually exist, since we've populated the field on the alerts.
+
+ // Build our group request
+ Group searchGroup = new Group();
+ searchGroup.setField("ip_src_addr");
+ List<Group> groupList = new ArrayList<>();
+ groupList.add(searchGroup);
+ GroupResponse groupResponse = metaDao.group(new GroupRequest() {
+ {
+ setQuery("ip_src_addr:192.168.1.1");
+ setIndices(Collections.singletonList("*"));
+ setScoreField("score_field");
+ setGroups(groupList);
+ }});
- /**
- {
- "guid": "update_metaalert_alert_1",
- "source:type": "test",
- "field":"value 1"
- }
- */
- @Multiline
- public static String updateMetaAlertAlert1;
+ // Should only return the standalone alert in the group
+ GroupResult result = groupResponse.getGroupResults().get(0);
+ Assert.assertEquals(1, result.getTotal());
+ Assert.assertEquals("192.168.1.1", result.getKey());
+ // No delta, since no ops happen
+ Assert.assertEquals(10.0d, result.getScore(), 0.0d);
+ }
- /**
- {
- "guid": "update_metaalert_alert_0",
- "patch": [
- {
- "op": "add",
- "path": "/field",
- "value": "patched value 0"
- }
- ],
- "sensorType": "test"
- }
- */
- @Multiline
- public static String updateMetaAlertPatchRequest;
+ @Test
+ public void testStatusChanges() throws Exception {
+ // Create alerts
+ List<Map<String, Object>> alerts = new ArrayList<>();
+ Map<String, Object> searchByNestedAlertActive0Json = JSONUtils.INSTANCE
+ .load(searchByNestedAlertActive0, new TypeReference<Map<String, Object>>() {
+ });
+ alerts.add(searchByNestedAlertActive0Json);
+ Map<String, Object> searchByNestedAlertActive1Json = JSONUtils.INSTANCE
+ .load(searchByNestedAlertActive1, new TypeReference<Map<String, Object>>() {
+ });
+ alerts.add(searchByNestedAlertActive1Json);
+ elasticsearchAdd(alerts, INDEX, SENSOR_NAME);
+ // Wait for updates to persist
+ findUpdatedDoc(searchByNestedAlertActive0Json, "search_by_nested_alert_active_0",
+ SENSOR_NAME);
+ findUpdatedDoc(searchByNestedAlertActive1Json, "search_by_nested_alert_active_1",
+ SENSOR_NAME);
- /**
- {
- "guid": "update_metaalert_alert_0",
- "replacement": {
- "guid": "update_metaalert_alert_0",
- "source:type": "test",
- "field": "replaced value 0"
- },
- "sensorType": "test"
- }
- */
- @Multiline
- public static String updateMetaAlertReplaceRequest;
+ // Put the nested type into the test index, so that it'll match appropriately
+ ((ElasticsearchDao) esDao).getClient().admin().indices().preparePutMapping(INDEX)
+ .setType("test_doc")
+ .setSource(nestedAlertMapping)
+ .get();
+
+ // Create metaalerts
+ Map<String, Object> activeMetaAlertJSON = JSONUtils.INSTANCE
+ .load(activeMetaAlert, new TypeReference<Map<String, Object>>() {
+ });
+ activeMetaAlertJSON.put("alert",
+ Arrays.asList(searchByNestedAlertActive0Json, searchByNestedAlertActive1Json));
+
+ // We pass MetaAlertDao.METAALERT_TYPE, because the "_doc" gets appended automatically.
+ elasticsearchAdd(Collections.singletonList(activeMetaAlertJSON),
+ MetaAlertDao.METAALERTS_INDEX, MetaAlertDao.METAALERT_TYPE);
+ // Wait for updates to persist
+ findUpdatedDoc(activeMetaAlertJSON, "active_metaalert", MetaAlertDao.METAALERT_TYPE);
+
+ // Build our update request to inactive status
+ Map<String, Object> documentMap = new HashMap<>();
+
+ documentMap.put("status", MetaAlertStatus.INACTIVE.getStatusString());
+ Document document = new Document(documentMap, "active_metaalert", MetaAlertDao.METAALERT_TYPE,
+ 0L);
+ metaDao.update(document, Optional.of(MetaAlertDao.METAALERTS_INDEX));
+
+ Map<String, Object> expectedMetaDoc = new HashMap<>();
+ expectedMetaDoc.putAll(activeMetaAlertJSON);
+ expectedMetaDoc.put("status", MetaAlertStatus.INACTIVE.getStatusString());
+
+ // Make sure the update has gone through on the meta alert and the child alerts.
+ Assert.assertTrue(
+ findUpdatedDoc(expectedMetaDoc, "active_metaalert", MetaAlertDao.METAALERT_TYPE));
+
+ Map<String, Object> expectedAlertDoc0 = new HashMap<>();
+ expectedAlertDoc0.putAll(searchByNestedAlertActive0Json);
+ expectedAlertDoc0.put("metaalerts", new ArrayList<>());
+ Assert.assertTrue(
+ findUpdatedDoc(expectedAlertDoc0, "search_by_nested_alert_active_0", SENSOR_NAME));
+
+ Map<String, Object> expectedAlertDoc1 = new HashMap<>();
+ expectedAlertDoc1.putAll(searchByNestedAlertActive1Json);
+ expectedAlertDoc1.put("metaalerts", new ArrayList<>());
+ Assert.assertTrue(
+ findUpdatedDoc(expectedAlertDoc1, "search_by_nested_alert_active_1", SENSOR_NAME));
+
+ // Search against the indices. Should return the two alerts, but not the inactive metaalert.
+ SearchRequest searchRequest = new SearchRequest();
+ ArrayList<String> indices = new ArrayList<>();
+ indices.add(SENSOR_NAME);
+ indices.add(MetaAlertDao.METAALERT_TYPE);
+ searchRequest.setIndices(indices);
+ searchRequest.setSize(5);
+ searchRequest.setQuery("*");
+
+ // Validate our results
+ SearchResult expected0 = new SearchResult();
+ expected0.setId((String) expectedAlertDoc0.get(Constants.GUID));
+ expected0.setIndex(INDEX);
+ expected0.setSource(expectedAlertDoc0);
+ expected0.setScore(1.0f);
+
+ SearchResult expected1 = new SearchResult();
+ expected1.setId((String) expectedAlertDoc1.get(Constants.GUID));
+ expected1.setIndex(INDEX);
+ expected1.setSource(expectedAlertDoc1);
+ expected1.setScore(1.0f);
+
+ ArrayList<SearchResult> expectedResults = new ArrayList<>();
+ expectedResults.add(expected0);
+ expectedResults.add(expected1);
+
+ SearchResponse result = metaDao.search(searchRequest);
+ Assert.assertEquals(2, result.getTotal());
+ // Use set comparison to avoid ordering issues. We already checked counts.
+ Assert.assertEquals(new HashSet<>(expectedResults), new HashSet<>(result.getResults()));
+
+ // Build our update request back to active status
+ documentMap.put("status", MetaAlertStatus.ACTIVE.getStatusString());
+ document = new Document(documentMap, "active_metaalert", MetaAlertDao.METAALERT_TYPE, 0L);
+ metaDao.update(document, Optional.of(MetaAlertDao.METAALERTS_INDEX));
+
+ expectedMetaDoc = new HashMap<>();
+ expectedMetaDoc.putAll(activeMetaAlertJSON);
+
+ // Make sure the update has gone through on the meta alert and the child alerts.
+ Assert.assertTrue(
+ findUpdatedDoc(expectedMetaDoc, "active_metaalert", MetaAlertDao.METAALERT_TYPE));
+
+ expectedAlertDoc0 = new HashMap<>();
+ expectedAlertDoc0.putAll(searchByNestedAlertActive0Json);
+ Assert.assertTrue(
+ findUpdatedDoc(expectedAlertDoc0, "search_by_nested_alert_active_0", SENSOR_NAME));
+
+ expectedAlertDoc1 = new HashMap<>();
+ expectedAlertDoc1.putAll(searchByNestedAlertActive1Json);
+ Assert.assertTrue(
+ findUpdatedDoc(expectedAlertDoc1, "search_by_nested_alert_active_1", SENSOR_NAME));
+
+ // Search against the indices. Should return just the active metaalert.
+ SearchResult expectedMeta = new SearchResult();
+ expectedMeta.setId((String) activeMetaAlertJSON.get(Constants.GUID));
+ expectedMeta.setIndex(MetaAlertDao.METAALERTS_INDEX);
+ expectedMeta.setSource(activeMetaAlertJSON);
+ expectedMeta.setScore(1.0f);
+
+ expectedResults = new ArrayList<>();
+ expectedResults.add(expectedMeta);
+
+ result = metaDao.search(searchRequest);
+ Assert.assertEquals(1, result.getTotal());
+ Assert.assertEquals(expectedResults, result.getResults());
+ }
@Test
+ @SuppressWarnings("unchecked")
public void shouldUpdateMetaAlertOnAlertPatchOrReplace() throws Exception {
List<Map<String, Object>> inputData = new ArrayList<>();
- Map<String, Object> updateMetaAlertAlert0JSON = JSONUtils.INSTANCE.load(updateMetaAlertAlert0, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> updateMetaAlertAlert0JSON = JSONUtils.INSTANCE
+ .load(updateMetaAlertAlert0, new TypeReference<Map<String, Object>>() {
+ });
inputData.add(updateMetaAlertAlert0JSON);
- Map<String, Object> updateMetaAlertAlert1JSON = JSONUtils.INSTANCE.load(updateMetaAlertAlert1, new TypeReference<Map<String, Object>>() {});
+ Map<String, Object> updateMetaAlertAlert1JSON = JSONUtils.INSTANCE
+ .load(updateMetaAlertAlert1, new TypeReference<Map<String, Object>>() {
+ });
inputData.add(updateMetaAlertAlert1JSON);
elasticsearchAdd(inputData, INDEX, SENSOR_NAME);
// Wait for updates to persist
findUpdatedDoc(updateMetaAlertAlert1JSON, "update_metaalert_alert_1", SENSOR_NAME);
- MetaAlertCreateResponse metaAlertCreateResponse = metaDao.createMetaAlert(new MetaAlertCreateRequest() {{
- setGuidToIndices(new HashMap<String, String>() {{
- put("update_metaalert_alert_0", INDEX);
- put("update_metaalert_alert_1", INDEX);
- }});
- setGroups(Collections.singletonList("group"));
- }});
+ MetaAlertCreateResponse metaAlertCreateResponse = metaDao
+ .createMetaAlert(new MetaAlertCreateRequest() {{
+ setGuidToIndices(new HashMap<String, String>() {{
+ put("update_metaalert_alert_0", INDEX);
+ put("update_metaalert_alert_1", INDEX);
+ }});
+ setGroups(Collections.singletonList("group"));
+ }});
// Wait for updates to persist
findCreatedDoc(metaAlertCreateResponse.getGuid(), MetaAlertDao.METAALERT_TYPE);
// Patch alert
- metaDao.patch(JSONUtils.INSTANCE.load(updateMetaAlertPatchRequest, PatchRequest.class), Optional.empty());
+ metaDao.patch(JSONUtils.INSTANCE.load(updateMetaAlertPatchRequest, PatchRequest.class),
+ Optional.empty());
// Wait for updates to persist
updateMetaAlertAlert0JSON.put("field", "patched value 0");
findUpdatedDoc(updateMetaAlertAlert0JSON, "update_metaalert_alert_0", SENSOR_NAME);
- Map<String, Object> metaalert = metaDao.getLatest(metaAlertCreateResponse.getGuid(), MetaAlertDao.METAALERT_TYPE).getDocument();
+ Map<String, Object> metaalert = metaDao
+ .getLatest(metaAlertCreateResponse.getGuid(), MetaAlertDao.METAALERT_TYPE).getDocument();
List<Map<String, Object>> alerts = (List<Map<String, Object>>) metaalert.get("alert");
Assert.assertEquals(2, alerts.size());
Assert.assertEquals("update_metaalert_alert_1", alerts.get(0).get("guid"));
@@ -545,13 +869,15 @@ public class ElasticsearchMetaAlertIntegrationTest {
Assert.assertEquals("patched value 0", alerts.get(1).get("field"));
// Replace alert
- metaDao.replace(JSONUtils.INSTANCE.load(updateMetaAlertReplaceRequest, ReplaceRequest.class), Optional.empty());
+ metaDao.replace(JSONUtils.INSTANCE.load(updateMetaAlertReplaceRequest, ReplaceRequest.class),
+ Optional.empty());
// Wait for updates to persist
updateMetaAlertAlert0JSON.put("field", "replaced value 0");
findUpdatedDoc(updateMetaAlertAlert0JSON, "update_metaalert_alert_0", SENSOR_NAME);
- metaalert = metaDao.getLatest(metaAlertCreateResponse.getGuid(), MetaAlertDao.METAALERT_TYPE).getDocument();
+ metaalert = metaDao.getLatest(metaAlertCreateResponse.getGuid(), MetaAlertDao.METAALERT_TYPE)
+ .getDocument();
alerts = (List<Map<String, Object>>) metaalert.get("alert");
Assert.assertEquals(2, alerts.size());
Assert.assertEquals("update_metaalert_alert_1", alerts.get(0).get("guid"));
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/components/ElasticSearchComponent.java
----------------------------------------------------------------------
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 171b6ab..3ef9379 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
@@ -28,6 +28,7 @@ import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
@@ -233,4 +234,9 @@ public class ElasticSearchComponent implements InMemoryComponent {
node = null;
client = null;
}
+
+ @Override
+ public void reset() {
+ client.admin().indices().delete(new DeleteIndexRequest("*")).actionGet();
+ }
}
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/README.md b/metron-platform/metron-indexing/README.md
index 6dbcb98..86c8d37 100644
--- a/metron-platform/metron-indexing/README.md
+++ b/metron-platform/metron-indexing/README.md
@@ -46,6 +46,8 @@ If unspecified, or set to `0`, it defaults to a system-determined duration which
parameter `topology.message.timeout.secs`. Ignored if batchSize is `1`, since this disables batching.
* `enabled` : Whether the writer is enabled (default `true`).
+### Meta Alerts
+Alerts can be grouped, after appropriate searching, into a set of alerts called a meta alert. A meta alert is useful for maintaining the context of searching and grouping during further investigations. Standard searches can return meta alerts, but grouping and other aggregation or sorting requests will not, because there's not a clear way to aggregate in many cases if there are multiple alerts contained in the meta alert. All meta alerts will have the source type of metaalert, regardless of the contained alert's origins.
### Elasticsearch
Metron comes with built-in templates for the default sensors for Elasticsearch. When adding a new sensor, it will be necessary to add a new template defining the output fields appropriately. In addition, there is a requirement for a field `alert` of type `nested` for Elasticsearch 2.x installs. This is detailed at [Using Metron with Elasticsearch 2.x](../metron-elasticsearch/README.md#using-metron-with-elasticsearch-2x)
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java
index 32bfab9..1775018 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,6 +18,13 @@
package org.apache.metron.indexing.dao;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Optional;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
@@ -33,12 +40,6 @@ import org.apache.metron.indexing.dao.search.SearchRequest;
import org.apache.metron.indexing.dao.search.SearchResponse;
import org.apache.metron.indexing.dao.update.Document;
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.Optional;
-
/**
* The HBaseDao is an index dao which only supports the following actions:
* * Update
@@ -121,12 +122,31 @@ public class HBaseDao implements IndexDao {
@Override
public synchronized void update(Document update, Optional<String> index) throws IOException {
+ Put put = buildPut(update);
+ getTableInterface().put(put);
+ }
+
+
+
+ @Override
+ public void batchUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ List<Put> puts = new ArrayList<>();
+ for (Map.Entry<Document, Optional<String>> updateEntry : updates.entrySet()) {
+ Document update = updateEntry.getKey();
+
+ Put put = buildPut(update);
+ puts.add(put);
+ }
+ getTableInterface().put(puts);
+ }
+
+ protected Put buildPut(Document update) throws JsonProcessingException {
Put put = new Put(update.getGuid().getBytes());
- long ts = update.getTimestamp() == null?System.currentTimeMillis():update.getTimestamp();
+ long ts = update.getTimestamp() == null ? System.currentTimeMillis() : update.getTimestamp();
byte[] columnQualifier = Bytes.toBytes(ts);
byte[] doc = JSONUtils.INSTANCE.toJSONPretty(update.getDocument());
put.addColumn(cf, columnQualifier, doc);
- getTableInterface().put(put);
+ return put;
}
@Override
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/IndexDao.java
index be5d4fe..4b7829e 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
@@ -90,6 +90,13 @@ public interface IndexDao {
*/
void update(Document update, Optional<String> index) throws IOException;
+ /**
+ * Update given a Document and optionally the index where the document exists.
+ *
+ * @param updates A map of the documents to update to the index where they live.
+ * @throws IOException
+ */
+ void batchUpdate(Map<Document, Optional<String>> updates) throws IOException;
/**
* Update a document in an index given a JSON Patch (see RFC 6902 at https://tools.ietf.org/html/rfc6902)
@@ -104,7 +111,7 @@ public interface IndexDao {
Map<String, Object> latest = request.getSource();
if(latest == null) {
Document latestDoc = getLatest(request.getGuid(), request.getSensorType());
- if(latestDoc.getDocument() != null) {
+ if(latestDoc != null && latestDoc.getDocument() != null) {
latest = latestDoc.getDocument();
}
else {
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MetaAlertDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MetaAlertDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MetaAlertDao.java
index e9f047b..de12f22 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MetaAlertDao.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MetaAlertDao.java
@@ -29,6 +29,7 @@ public interface MetaAlertDao extends IndexDao {
String METAALERTS_INDEX = "metaalert_index";
String METAALERT_TYPE = "metaalert";
+ String METAALERT_FIELD = "metaalerts";
String METAALERT_DOC = METAALERT_TYPE + "_doc";
String THREAT_FIELD_DEFAULT = "threat:triage:score";
String THREAT_SORT_DEFAULT = "sum";
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java
index 2df06fc..779e6c6 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/MultiIndexDao.java
@@ -20,6 +20,14 @@ package org.apache.metron.indexing.dao;
import com.google.common.base.Joiner;
import com.google.common.collect.Iterables;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.metron.indexing.dao.search.FieldType;
import org.apache.metron.indexing.dao.search.GroupRequest;
@@ -29,11 +37,6 @@ import org.apache.metron.indexing.dao.search.SearchRequest;
import org.apache.metron.indexing.dao.search.SearchResponse;
import org.apache.metron.indexing.dao.update.Document;
-import java.io.IOException;
-import java.util.*;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
public class MultiIndexDao implements IndexDao {
private List<IndexDao> indices;
@@ -68,6 +71,22 @@ public class MultiIndexDao implements IndexDao {
}
@Override
+ public void batchUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ List<String> exceptions =
+ indices.parallelStream().map(dao -> {
+ try {
+ dao.batchUpdate(updates);
+ return null;
+ } catch (Throwable e) {
+ return dao.getClass() + ": " + e.getMessage() + "\n" + ExceptionUtils.getStackTrace(e);
+ }
+ }).filter(e -> e != null).collect(Collectors.toList());
+ if (exceptions.size() > 0) {
+ throw new IOException(Joiner.on("\n").join(exceptions));
+ }
+ }
+
+ @Override
public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> in) throws IOException {
for(IndexDao dao : indices) {
Map<String, Map<String, FieldType>> r = dao.getColumnMetadata(in);
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
index da4fac1..8ed5934 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/search/SearchResult.java
@@ -83,4 +83,36 @@ public class SearchResult {
", index='" + index + '\'' +
'}';
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ SearchResult that = (SearchResult) o;
+
+ if (Float.compare(that.getScore(), getScore()) != 0) {
+ return false;
+ }
+ if (getId() != null ? !getId().equals(that.getId()) : that.getId() != null) {
+ return false;
+ }
+ if (getSource() != null ? !getSource().equals(that.getSource()) : that.getSource() != null) {
+ return false;
+ }
+ return getIndex() != null ? getIndex().equals(that.getIndex()) : that.getIndex() == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = getId() != null ? getId().hashCode() : 0;
+ result = 31 * result + (getSource() != null ? getSource().hashCode() : 0);
+ result = 31 * result + (getScore() != +0.0f ? Float.floatToIntBits(getScore()) : 0);
+ result = 31 * result + (getIndex() != null ? getIndex().hashCode() : 0);
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
index c83f6aa..f48187e 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
@@ -214,6 +214,13 @@ public class InMemoryDao implements IndexDao {
}
}
+ @Override
+ public void batchUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ for (Map.Entry<Document, Optional<String>> update : updates.entrySet()) {
+ update(update.getKey(), update.getValue());
+ }
+ }
+
public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> indices) throws IOException {
Map<String, Map<String, FieldType>> columnMetadata = new HashMap<>();
for(String index: indices) {
http://git-wip-us.apache.org/repos/asf/metron/blob/131a15ef/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryMetaAlertDao.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryMetaAlertDao.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryMetaAlertDao.java
index 39c0001..cb7635e 100644
--- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryMetaAlertDao.java
+++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryMetaAlertDao.java
@@ -99,6 +99,11 @@ public class InMemoryMetaAlertDao implements MetaAlertDao {
}
@Override
+ public void batchUpdate(Map<Document, Optional<String>> updates) throws IOException {
+ throw new UnsupportedOperationException("InMemoryMetaAlertDao can't do bulk updates");
+ }
+
+ @Override
public Map<String, Map<String, FieldType>> getColumnMetadata(List<String> indices)
throws IOException {
return indexDao.getColumnMetadata(indices);