add correlation rule layer for events-correlation-engine

Signed-off-by: Subhobrata Dey <[email protected]>

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Support for HTTP/2 (server-side) ([#3847](https://github.com/opensearch-project/OpenSearch/pull/3847))
 - Add getter for path field in NestedQueryBuilder ([#4636](https://github.com/opensearch-project/OpenSearch/pull/4636))
 - Allow mmap to use new JDK-19 preview APIs in Apache Lucene 9.4+ ([#5151](https://github.com/opensearch-project/OpenSearch/pull/5151))
+- Events Correlation Engine ([#6892](https://github.com/opensearch-project/OpenSearch/pull/6892))
 
 ### Dependencies
 - Bump `log4j-core` from 2.18.0 to 2.19.0

gradle/missing-javadoc.gradle

@@ -135,6 +135,7 @@ configure([
   project(":plugins:discovery-ec2:qa:amazon-ec2"),
   project(":plugins:discovery-gce"),
   project(":plugins:discovery-gce:qa:gce"),
+  project(":plugins:events-correlation-engine"),
   project(":plugins:ingest-attachment"),
   project(":plugins:mapper-annotated-text"),
   project(":plugins:mapper-murmur3"),

plugins/events-correlation-engine/build.gradle

@@ -0,0 +1,21 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+apply plugin: 'opensearch.java-rest-test'
+apply plugin: 'opensearch.internal-cluster-test'
+
+opensearchplugin {
+  description 'OpenSearch Events Correlation Engine.'
+  classname 'org.opensearch.plugin.correlation.EventsCorrelationPlugin'
+}
+
+dependencies {
+}

plugins/events-correlation-engine/src/internalClusterTest/java/org/opensearch/plugin/correlation/EventsCorrelationPluginTransportIT.java

@@ -0,0 +1,152 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.plugin.correlation;
+
+import org.apache.lucene.search.join.ScoreMode;
+import org.junit.Assert;
+import org.opensearch.action.admin.cluster.node.info.NodeInfo;
+import org.opensearch.action.admin.cluster.node.info.NodesInfoRequest;
+import org.opensearch.action.admin.cluster.node.info.NodesInfoResponse;
+import org.opensearch.action.admin.cluster.node.info.PluginsAndModules;
+import org.opensearch.action.search.SearchRequest;
+import org.opensearch.action.search.SearchResponse;
+import org.opensearch.index.query.NestedQueryBuilder;
+import org.opensearch.index.query.QueryBuilders;
+import org.opensearch.plugin.correlation.rules.action.IndexCorrelationRuleAction;
+import org.opensearch.plugin.correlation.rules.action.IndexCorrelationRuleRequest;
+import org.opensearch.plugin.correlation.rules.action.IndexCorrelationRuleResponse;
+import org.opensearch.plugin.correlation.rules.model.CorrelationQuery;
+import org.opensearch.plugin.correlation.rules.model.CorrelationRule;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.PluginInfo;
+import org.opensearch.rest.RestRequest;
+import org.opensearch.rest.RestStatus;
+import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class EventsCorrelationPluginTransportIT extends OpenSearchIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(EventsCorrelationPlugin.class);
+    }
+
+    public void testPluginsAreInstalled() {
+        NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
+        nodesInfoRequest.addMetric(NodesInfoRequest.Metric.PLUGINS.metricName());
+        NodesInfoResponse nodesInfoResponse = OpenSearchIntegTestCase.client().admin().cluster().nodesInfo(nodesInfoRequest).actionGet();
+        List<PluginInfo> pluginInfos = nodesInfoResponse.getNodes()
+            .stream()
+            .flatMap(
+                (Function<NodeInfo, Stream<PluginInfo>>) nodeInfo -> nodeInfo.getInfo(PluginsAndModules.class).getPluginInfos().stream()
+            )
+            .collect(Collectors.toList());
+        Assert.assertTrue(
+            pluginInfos.stream()
+                .anyMatch(pluginInfo -> pluginInfo.getName().equals("org.opensearch.plugin.correlation.EventsCorrelationPlugin"))
+        );
+    }
+
+    public void testCreatingACorrelationRule() throws ExecutionException, InterruptedException {
+        List<CorrelationQuery> correlationQueries = Arrays.asList(
+            new CorrelationQuery("s3_access_logs", "aws.cloudtrail.eventName:ReplicateObject", "@timestamp", List.of("s3")),
+            new CorrelationQuery("app_logs", "keywords:PermissionDenied", "@timestamp", List.of("others_application"))
+        );
+        CorrelationRule correlationRule = new CorrelationRule(
+            CorrelationRule.NO_ID,
+            CorrelationRule.NO_VERSION,
+            "s3 to app logs",
+            correlationQueries
+        );
+        IndexCorrelationRuleRequest request = new IndexCorrelationRuleRequest(
+            CorrelationRule.NO_ID,
+            correlationRule,
+            RestRequest.Method.POST
+        );
+
+        IndexCorrelationRuleResponse response = client().execute(IndexCorrelationRuleAction.INSTANCE, request).get();
+        Assert.assertEquals(RestStatus.CREATED, response.getStatus());
+
+        NestedQueryBuilder queryBuilder = QueryBuilders.nestedQuery(
+            "correlate",
+            QueryBuilders.matchQuery("correlate.index", "s3_access_logs"),
+            ScoreMode.None
+        );
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+        searchSourceBuilder.query(queryBuilder);
+        searchSourceBuilder.fetchSource(true);
+
+        SearchRequest searchRequest = new SearchRequest();
+        searchRequest.indices(CorrelationRule.CORRELATION_RULE_INDEX);
+        searchRequest.source(searchSourceBuilder);
+
+        SearchResponse searchResponse = client().search(searchRequest).get();
+        Assert.assertEquals(1L, searchResponse.getHits().getTotalHits().value);
+    }
+
+    public void testFilteringCorrelationRules() throws ExecutionException, InterruptedException {
+        List<CorrelationQuery> correlationQueries1 = Arrays.asList(
+            new CorrelationQuery("s3_access_logs", "aws.cloudtrail.eventName:ReplicateObject", "@timestamp", List.of("s3")),
+            new CorrelationQuery("app_logs", "keywords:PermissionDenied", "@timestamp", List.of("others_application"))
+        );
+        CorrelationRule correlationRule1 = new CorrelationRule(
+            CorrelationRule.NO_ID,
+            CorrelationRule.NO_VERSION,
+            "s3 to app logs",
+            correlationQueries1
+        );
+        IndexCorrelationRuleRequest request1 = new IndexCorrelationRuleRequest(
+            CorrelationRule.NO_ID,
+            correlationRule1,
+            RestRequest.Method.POST
+        );
+        client().execute(IndexCorrelationRuleAction.INSTANCE, request1).get();
+
+        List<CorrelationQuery> correlationQueries2 = Arrays.asList(
+            new CorrelationQuery("windows", "host.hostname:EC2AMAZ*", "@timestamp", List.of("windows")),
+            new CorrelationQuery("app_logs", "endpoint:/customer_records.txt", "@timestamp", List.of("others_application"))
+        );
+        CorrelationRule correlationRule2 = new CorrelationRule(
+            CorrelationRule.NO_ID,
+            CorrelationRule.NO_VERSION,
+            "windows to app logs",
+            correlationQueries2
+        );
+        IndexCorrelationRuleRequest request2 = new IndexCorrelationRuleRequest(
+            CorrelationRule.NO_ID,
+            correlationRule2,
+            RestRequest.Method.POST
+        );
+        client().execute(IndexCorrelationRuleAction.INSTANCE, request2).get();
+
+        NestedQueryBuilder queryBuilder = QueryBuilders.nestedQuery(
+            "correlate",
+            QueryBuilders.matchQuery("correlate.index", "s3_access_logs"),
+            ScoreMode.None
+        );
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+        searchSourceBuilder.query(queryBuilder);
+        searchSourceBuilder.fetchSource(true);
+
+        SearchRequest searchRequest = new SearchRequest();
+        searchRequest.indices(CorrelationRule.CORRELATION_RULE_INDEX);
+        searchRequest.source(searchSourceBuilder);
+
+        SearchResponse searchResponse = client().search(searchRequest).get();
+        Assert.assertEquals(1L, searchResponse.getHits().getTotalHits().value);
+    }
+}

plugins/events-correlation-engine/src/javaRestTest/java/org/opensearch/plugin/correlation/EventsCorrelationPluginRestIT.java

@@ -0,0 +1,90 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.plugin.correlation;
+
+import org.junit.Assert;
+import org.opensearch.action.search.SearchResponse;
+import org.opensearch.client.Request;
+import org.opensearch.client.Response;
+import org.opensearch.common.xcontent.LoggingDeprecationHandler;
+import org.opensearch.common.xcontent.json.JsonXContent;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.test.rest.OpenSearchRestTestCase;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+public class EventsCorrelationPluginRestIT extends OpenSearchRestTestCase {
+
+    @SuppressWarnings("unchecked")
+    public void testPluginsAreInstalled() throws IOException {
+        Request request = new Request("GET", "/_cat/plugins?s=component&h=name,component,version,description&format=json");
+        Response response = client().performRequest(request);
+        List<Object> pluginsList = JsonXContent.jsonXContent.createParser(
+            NamedXContentRegistry.EMPTY,
+            LoggingDeprecationHandler.INSTANCE,
+            response.getEntity().getContent()
+        ).list();
+        Assert.assertTrue(
+            pluginsList.stream()
+                .map(o -> (Map<String, Object>) o)
+                .anyMatch(plugin -> plugin.get("component").equals("events-correlation-engine"))
+        );
+    }
+
+    public void testCreatingACorrelationRule() throws IOException {
+        Request request = new Request("POST", "/_plugins/_correlation/rules");
+        request.setJsonEntity(sampleCorrelationRule());
+        Response response = client().performRequest(request);
+
+        Assert.assertEquals(201, response.getStatusLine().getStatusCode());
+
+        Map<String, Object> responseMap = entityAsMap(response);
+        String id = responseMap.get("_id").toString();
+
+        request = new Request("POST", "/.opensearch-correlation-rules-config/_search");
+        request.setJsonEntity(matchIdQuery(id));
+        response = client().performRequest(request);
+
+        Assert.assertEquals(200, response.getStatusLine().getStatusCode());
+        SearchResponse searchResponse = SearchResponse.fromXContent(
+            createParser(JsonXContent.jsonXContent, response.getEntity().getContent())
+        );
+        Assert.assertEquals(1L, searchResponse.getHits().getTotalHits().value);
+    }
+
+    private String sampleCorrelationRule() {
+        return "{\n"
+            + "  \"name\": \"s3 to app logs\",\n"
+            + "  \"correlate\": [\n"
+            + "    {\n"
+            + "      \"index\": \"s3_access_logs\",\n"
+            + "      \"query\": \"aws.cloudtrail.eventName:ReplicateObject\",\n"
+            + "      \"timestampField\": \"@timestamp\",\n"
+            + "      \"tags\": [\n"
+            + "        \"s3\"\n"
+            + "      ]\n"
+            + "    },\n"
+            + "    {\n"
+            + "      \"index\": \"app_logs\",\n"
+            + "      \"query\": \"keywords:PermissionDenied\",\n"
+            + "      \"timestampField\": \"@timestamp\",\n"
+            + "      \"tags\": [\n"
+            + "        \"others_application\"\n"
+            + "      ]\n"
+            + "    }\n"
+            + "  ]\n"
+            + "}";
+    }
+
+    private String matchIdQuery(String id) {
+        return "{\n" + "   \"query\" : {\n" + "     \"match\":{\n" + "        \"_id\": \"" + id + "\"\n" + "     }\n" + "   }\n" + "}";
+    }
+}

plugins/events-correlation-engine/src/main/java/org/opensearch/plugin/correlation/EventsCorrelationPlugin.java

@@ -0,0 +1,91 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.plugin.correlation;
+
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.ActionResponse;
+import org.opensearch.client.Client;
+import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
+import org.opensearch.cluster.node.DiscoveryNodes;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.io.stream.NamedWriteableRegistry;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.IndexScopedSettings;
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.settings.SettingsFilter;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.env.Environment;
+import org.opensearch.env.NodeEnvironment;
+import org.opensearch.plugin.correlation.rules.action.IndexCorrelationRuleAction;
+import org.opensearch.plugin.correlation.rules.resthandler.RestIndexCorrelationRuleAction;
+import org.opensearch.plugin.correlation.rules.transport.TransportIndexCorrelationRuleAction;
+import org.opensearch.plugin.correlation.settings.EventsCorrelationSettings;
+import org.opensearch.plugin.correlation.utils.CorrelationRuleIndices;
+import org.opensearch.plugins.ActionPlugin;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.rest.RestController;
+import org.opensearch.rest.RestHandler;
+import org.opensearch.script.ScriptService;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.watcher.ResourceWatcherService;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Supplier;
+
+public class EventsCorrelationPlugin extends Plugin implements ActionPlugin {
+
+    public static final String PLUGINS_BASE_URI = "/_plugins/_correlation";
+    public static final String CORRELATION_RULES_BASE_URI = PLUGINS_BASE_URI + "/rules";
+
+    private CorrelationRuleIndices correlationRuleIndices;
+
+    @Override
+    public Collection<Object> createComponents(
+        Client client,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ResourceWatcherService resourceWatcherService,
+        ScriptService scriptService,
+        NamedXContentRegistry xContentRegistry,
+        Environment environment,
+        NodeEnvironment nodeEnvironment,
+        NamedWriteableRegistry namedWriteableRegistry,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<RepositoriesService> repositoriesServiceSupplier
+    ) {
+        correlationRuleIndices = new CorrelationRuleIndices(client, clusterService);
+        return List.of(correlationRuleIndices);
+    }
+
+    @Override
+    public List<RestHandler> getRestHandlers(
+        Settings settings,
+        RestController restController,
+        ClusterSettings clusterSettings,
+        IndexScopedSettings indexScopedSettings,
+        SettingsFilter settingsFilter,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<DiscoveryNodes> nodesInCluster
+    ) {
+        return List.of(new RestIndexCorrelationRuleAction());
+    }
+
+    @Override
+    public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
+        return List.of(new ActionPlugin.ActionHandler<>(IndexCorrelationRuleAction.INSTANCE, TransportIndexCorrelationRuleAction.class));
+    }
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return List.of(EventsCorrelationSettings.IS_CORRELATION_INDEX_SETTING, EventsCorrelationSettings.CORRELATION_TIME_WINDOW);
+    }
+}

plugins/events-correlation-engine/src/main/java/org/opensearch/plugin/correlation/rules/action/IndexCorrelationRuleAction.java

@@ -0,0 +1,21 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.plugin.correlation.rules.action;
+
+import org.opensearch.action.ActionType;
+
+public class IndexCorrelationRuleAction extends ActionType<IndexCorrelationRuleResponse> {
+
+    public static final IndexCorrelationRuleAction INSTANCE = new IndexCorrelationRuleAction();
+    public static final String NAME = "cluster:admin/correlation/rules";
+
+    private IndexCorrelationRuleAction() {
+        super(NAME, IndexCorrelationRuleResponse::new);
+    }
+}