Add functionality to create data streams (#17910)

* Create Data Stream template, prepare ism policy

* add ism

* create ism and collect jvm heap as first metric

* change lookup and term query to use data streams

* start moving data stream functionality to graylog

* move data stream management into graylog

* code cleanup, add tests

* better validation, clean up after tests

* additional clean up after IT

* don't remove `index-management` plugin from datanode

* inject ObjectMapper

* change number of replicas from default 1 to 0 for test

* set `number_of_replicas` to 0 for .opendistro-ism-config in test

* replace javax with jakarta classes

* fix integration test

* remove data streams on test cleanup

* set high priority for data stream template

data-node/src/main/java/org/graylog/datanode/Configuration.java

@@ -213,6 +213,26 @@ public class Configuration extends BaseConfiguration {
     @Parameter(value = "http_allow_embedding")
     private boolean httpAllowEmbedding = false;
 
+    @Parameter(value = "metrics_timestamp")
+    private String metricsTimestamp = "timestamp";
+
+    @Parameter(value = "metrics_template")
+    private String metricsTemplate = "gl-datanode-metrics-template";
+    @Parameter(value = "metrics_stream")
+    private String metricsStream = "gl-datanode-metrics";
+
+    @Parameter(value = "metrics_retention")
+    private String metricsRetention = "14d";
+
+    @Parameter(value = "metrics_daily_retention")
+    private String metricsDailyRetention = "365d";
+
+    @Parameter(value = "metrics_daily_index")
+    private String metricsDailyIndex = "gl-datanode-metrics-daily";
+
+    @Parameter(value = "metrics_policy")
+    private String metricsPolicy = "gl-datanode-metrics-ism";
+
     public boolean isInsecureStartup() {
         return insecureStartup;
     }
@@ -361,6 +381,35 @@ public class Configuration extends BaseConfiguration {
         return clustername;
     }
 
+
+    public String getMetricsTimestamp() {
+        return metricsTimestamp;
+    }
+
+    public String getMetricsTemplate() {
+        return metricsTemplate;
+    }
+
+    public String getMetricsStream() {
+        return metricsStream;
+    }
+
+    public String getMetricsRetention() {
+        return metricsRetention;
+    }
+
+    public String getMetricsDailyRetention() {
+        return metricsDailyRetention;
+    }
+
+    public String getMetricsDailyIndex() {
+        return metricsDailyIndex;
+    }
+
+    public String getMetricsPolicy() {
+        return metricsPolicy;
+    }
+
     public static class NodeIdFileValidator implements Validator<String> {
         @Override
         public void validate(String name, String path) throws ValidationException {
@@ -582,4 +631,5 @@ public class Configuration extends BaseConfiguration {
     public String getRootPasswordSha2() {
         return rootPasswordSha2;
     }
+
 }

data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java

@@ -20,6 +20,7 @@ import com.google.inject.AbstractModule;
 import com.google.inject.multibindings.Multibinder;
 import org.graylog.datanode.bootstrap.preflight.DataNodeConfigurationPeriodical;
 import org.graylog.datanode.periodicals.ClusterManagerDiscovery;
+import org.graylog.datanode.periodicals.MetricsCollector;
 import org.graylog.datanode.periodicals.NodePingPeriodical;
 import org.graylog.datanode.periodicals.OpensearchNodeHeartbeat;
 import org.graylog2.events.ClusterEventCleanupPeriodical;
@@ -37,5 +38,6 @@ public class PeriodicalBindings extends AbstractModule {
 //        periodicalBinder.addBinding().to(UserSessionTerminationPeriodical.class);
         periodicalBinder.addBinding().to(NodePingPeriodical.class);
         periodicalBinder.addBinding().to(DataNodeConfigurationPeriodical.class);
+        periodicalBinder.addBinding().to(MetricsCollector.class);
     }
 }

data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java

@@ -19,6 +19,7 @@ package org.graylog.datanode.management;
 import org.graylog.datanode.process.OpensearchConfiguration;
 import org.graylog.datanode.process.OpensearchInfo;
 import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient;
+import org.graylog.storage.opensearch2.OpenSearchClient;
 
 import java.net.URI;
 import java.util.List;
@@ -30,6 +31,8 @@ public interface OpensearchProcess extends ManagableProcess<OpensearchConfigurat
 
     Optional<RestHighLevelClient> restClient();
 
+    Optional<OpenSearchClient> openSearchClient();
+
     boolean isLeaderNode();
     void setLeaderNode(boolean isManagerNode);
     List<String> stdOutLogs();
@@ -42,4 +45,5 @@ public interface OpensearchProcess extends ManagableProcess<OpensearchConfigurat
     void onRemove();
 
     void onReset();
+
 }

data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java

@@ -16,6 +16,7 @@
  */
 package org.graylog.datanode.management;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.oxo42.stateless4j.StateMachine;
 import org.apache.commons.collections4.queue.CircularFifoQueue;
 import org.apache.commons.exec.ExecuteException;
@@ -30,6 +31,7 @@ import org.graylog.datanode.process.ProcessState;
 import org.graylog.datanode.process.ProcessStateMachine;
 import org.graylog.datanode.process.StateMachineTracer;
 import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient;
+import org.graylog.storage.opensearch2.OpenSearchClient;
 import org.graylog2.cluster.nodes.DataNodeDto;
 import org.graylog2.cluster.nodes.NodeService;
 import org.graylog2.security.CustomCAX509TrustManager;
@@ -60,6 +62,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
     private Optional<OpensearchConfiguration> opensearchConfiguration = Optional.empty();
     @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
     private Optional<RestHighLevelClient> restClient = Optional.empty();
+    private Optional<OpenSearchClient> openSearchClient = Optional.empty();
 
     private final StateMachine<ProcessState, ProcessEvent> processState;
 
@@ -73,9 +76,11 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
     private final CustomCAX509TrustManager trustManager;
     private final NodeService<DataNodeDto> nodeService;
     private final Configuration configuration;
+    private final ObjectMapper objectMapper;
 
 
-    OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager, final Configuration configuration, final NodeService<DataNodeDto> nodeService) {
+    OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager,
+                          final Configuration configuration, final NodeService<DataNodeDto> nodeService, ObjectMapper objectMapper) {
         this.datanodeConfiguration = datanodeConfiguration;
         this.processState = ProcessStateMachine.createNew();
         tracerAggregator = new StateMachineTracerAggregator();
@@ -85,6 +90,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
         this.trustManager = trustManager;
         this.nodeService = nodeService;
         this.configuration = configuration;
+        this.objectMapper = objectMapper;
     }
 
     private RestHighLevelClient createRestClient(OpensearchConfiguration configuration) {
@@ -105,6 +111,10 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
         return restClient;
     }
 
+    public Optional<OpenSearchClient> openSearchClient() {
+        return openSearchClient;
+    }
+
     public OpensearchInfo processInfo() {
         return new OpensearchInfo(configuration.getDatanodeNodeName(), processState.getState(), isLeaderNode, getOpensearchBaseUrl().toString(), commandLineProcess != null ? commandLineProcess.processInfo() : ProcessInformation.empty());
     }
@@ -185,6 +195,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
                     commandLineProcess = new OpensearchCommandLineProcess(config, this);
                     commandLineProcess.start();
                     restClient = Optional.of(createRestClient(config));
+                    openSearchClient = restClient.map(c -> new OpenSearchClient(c, objectMapper));
                 }),
                 () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");}
         );

data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java

@@ -16,6 +16,7 @@
  */
 package org.graylog.datanode.management;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.eventbus.EventBus;
 import com.google.common.eventbus.Subscribe;
 import com.google.common.util.concurrent.AbstractIdleService;
@@ -24,6 +25,7 @@ import jakarta.inject.Provider;
 import jakarta.inject.Singleton;
 import org.graylog.datanode.Configuration;
 import org.graylog.datanode.configuration.DatanodeConfiguration;
+import org.graylog.datanode.metrics.ConfigureMetricsIndexSettings;
 import org.graylog.datanode.process.OpensearchConfiguration;
 import org.graylog2.cluster.nodes.DataNodeDto;
 import org.graylog2.cluster.nodes.NodeService;
@@ -48,6 +50,7 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
     private final Provider<OpensearchConfiguration> configurationProvider;
     private final EventBus eventBus;
     private final NodeId nodeId;
+    private final ObjectMapper objectMapper;
 
     @Inject
     public OpensearchProcessService(final DatanodeConfiguration datanodeConfiguration,
@@ -56,20 +59,23 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
                                     final CustomCAX509TrustManager trustManager,
                                     final NodeService<DataNodeDto> nodeService,
                                     final Configuration configuration,
-                                    final NodeId nodeId) {
+                                    final NodeId nodeId, ObjectMapper objectMapper) {
         this.configurationProvider = configurationProvider;
         this.eventBus = eventBus;
         this.nodeId = nodeId;
-        this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService);
+        this.objectMapper = objectMapper;
+        this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper);
         eventBus.register(this);
     }
 
-    private OpensearchProcess createOpensearchProcess(final DatanodeConfiguration datanodeConfiguration, final CustomCAX509TrustManager trustManager, final Configuration configuration, final NodeService<DataNodeDto> nodeService) {
-        final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize(), trustManager, configuration, nodeService);
+    private OpensearchProcess createOpensearchProcess(final DatanodeConfiguration datanodeConfiguration, final CustomCAX509TrustManager trustManager, final Configuration configuration,
+                                                      final NodeService<DataNodeDto> nodeService, final ObjectMapper objectMapper) {
+        final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize(), trustManager, configuration, nodeService, objectMapper);
         final ProcessWatchdog watchdog = new ProcessWatchdog(process, WATCHDOG_RESTART_ATTEMPTS);
         process.addStateMachineTracer(watchdog);
         process.addStateMachineTracer(new StateMachineTransitionLogger());
         process.addStateMachineTracer(new OpensearchRemovalTracer(process, configuration.getDatanodeNodeName()));
+        process.addStateMachineTracer(new ConfigureMetricsIndexSettings(process, configuration, objectMapper));
         return process;
     }
 

data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.datanode.metrics;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.graylog.datanode.Configuration;
+import org.graylog.datanode.management.OpensearchProcess;
+import org.graylog.datanode.process.ProcessEvent;
+import org.graylog.datanode.process.ProcessState;
+import org.graylog.datanode.process.StateMachineTracer;
+import org.graylog.storage.opensearch2.DataStreamAdapterOS2;
+import org.graylog.storage.opensearch2.ism.IsmApi;
+import org.graylog.storage.opensearch2.ism.policy.IsmPolicy;
+import org.graylog.storage.opensearch2.ism.policy.Policy;
+import org.graylog.storage.opensearch2.ism.policy.actions.Action;
+import org.graylog.storage.opensearch2.ism.policy.actions.DeleteAction;
+import org.graylog.storage.opensearch2.ism.policy.actions.RolloverAction;
+import org.graylog.storage.opensearch2.ism.policy.actions.RollupAction;
+import org.graylog2.indexer.datastream.DataStreamAdapter;
+import org.graylog2.indexer.indices.Template;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ConfigureMetricsIndexSettings implements StateMachineTracer {
+
+    private final Logger log = LoggerFactory.getLogger(ConfigureMetricsIndexSettings.class);
+
+    private final OpensearchProcess process;
+    private final Configuration configuration;
+    private final ObjectMapper objectMapper;
+    private DataStreamAdapter dataStreamAdapter;
+
+    public ConfigureMetricsIndexSettings(OpensearchProcess process, Configuration configuration, ObjectMapper objectMapper) {
+        this.process = process;
+        this.configuration = configuration;
+        this.objectMapper = objectMapper;
+    }
+
+    @Override
+    public void trigger(ProcessEvent trigger) {
+    }
+
+    @Override
+    public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) {
+        if (destination == ProcessState.AVAILABLE && source == ProcessState.STARTING) {
+            process.openSearchClient().ifPresent(client -> {
+                if (dataStreamAdapter == null) {
+                    final IsmApi ismApi = new IsmApi(client, objectMapper);
+                    dataStreamAdapter = new DataStreamAdapterOS2(client, objectMapper, ismApi);
+                }
+                updateDataStreamTemplate();
+                dataStreamAdapter.createDataStream(configuration.getMetricsStream());
+                configureMetricsIsm(configuration);
+            });
+
+
+        }
+    }
+
+    private void updateDataStreamTemplate() {
+        Map<String, Object> mappings = new HashMap<>();
+        mappings.put(configuration.getMetricsTimestamp(), ImmutableMap.of(
+                "type", "date",
+                "format", "yyyy-MM-dd HH:mm:ss.SSS||strict_date_optional_time||epoch_millis")
+        );
+        mappings.put("node", ImmutableMap.of("type", "keyword"));
+        Template template = new Template(List.of(configuration.getMetricsStream() + "*"),
+                new Template.Mappings(ImmutableMap.of("properties", mappings)), 99999L, new Template.Settings(Map.of()));
+        dataStreamAdapter.ensureDataStreamTemplate(configuration.getMetricsTemplate(), template, configuration.getMetricsTimestamp());
+    }
+
+    private void configureMetricsIsm(Configuration configuration) {
+        dataStreamAdapter.applyIsmPolicy(configuration.getMetricsStream(),
+                createPolicy(configuration));
+    }
+
+    private IsmPolicy createPolicy(Configuration configuration) {
+        // states defined from last to first to use name in previous step
+        Policy.State stateDelete = ismDeleteState();
+        Policy.State stateRollup = ismRollupState(stateDelete.name(), configuration);
+        Policy.State stateOpen = ismOpenState(stateRollup.name());
+
+        Policy policy = new Policy(null,
+                "Manages rollover, rollup and deletion of data note metrics indices",
+                null,
+                stateOpen.name(), ImmutableList.of(stateOpen, stateRollup, stateDelete));
+
+        try {
+            log.debug("Creating ISM configuration for metrics data stream {}",
+                    objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(policy));
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+
+        return new IsmPolicy(configuration.getMetricsPolicy(), policy);
+    }
+
+    private Policy.State ismDeleteState() {
+        final List<Action> actions = ImmutableList.of(new Action(new DeleteAction()));
+        final List<Policy.Transition> transitions = ImmutableList.of();
+        return new Policy.State("delete", actions, transitions);
+    }
+
+    private Policy.State ismRollupState(String nextState, Configuration configuration) {
+
+        final RollupAction.IsmRollup ismRollup = new RollupAction.IsmRollup(
+                configuration.getMetricsDailyIndex(),
+                "Rollup Data Stream Index",
+                1000,
+                ImmutableList.of(
+                        new RollupAction.IsmRollup.DateHistogram(
+                                configuration.getMetricsTimestamp(),
+                                "60m", "UTC")
+                ),
+                ImmutableList.of( //TODO dynamically rollup all configured metrics
+                        new RollupAction.IsmRollup.Metric(
+                                "jvm_heap",
+                                ImmutableList.of(new RollupAction.IsmRollup.AvgMetric())
+                        )
+                )
+        );
+        RollupAction rollupAction = new RollupAction(ismRollup);
+        final List<Action> actions = ImmutableList.of(new Action(rollupAction));
+        final List<Policy.Transition> transitions = ImmutableList.of(new Policy.Transition(nextState, new Policy.Condition("13d")));
+        return new Policy.State("rollup", actions, transitions);
+    }
+
+    private Policy.State ismOpenState(String nextState) {
+        final List<Action> actions = ImmutableList.of(new Action(new RolloverAction("1d")));
+        final List<Policy.Transition> transitions = ImmutableList.of(new Policy.Transition(nextState, null));
+        return new Policy.State("open", actions, transitions);
+    }
+
+}

data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.datanode.periodicals;
+
+import jakarta.inject.Inject;
+import org.graylog.datanode.Configuration;
+import org.graylog.datanode.management.OpensearchProcess;
+import org.graylog.datanode.process.ProcessState;
+import org.graylog.shaded.opensearch2.org.joda.time.DateTime;
+import org.graylog.shaded.opensearch2.org.joda.time.DateTimeZone;
+import org.graylog.shaded.opensearch2.org.opensearch.action.index.IndexRequest;
+import org.graylog.shaded.opensearch2.org.opensearch.action.index.IndexResponse;
+import org.graylog.shaded.opensearch2.org.opensearch.client.RequestOptions;
+import org.graylog.shaded.opensearch2.org.opensearch.core.action.ActionListener;
+import org.graylog2.plugin.periodical.Periodical;
+import org.graylog2.plugin.system.NodeId;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class MetricsCollector extends Periodical {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MetricsCollector.class);
+    private final OpensearchProcess process;
+    private final Configuration configuration;
+    private final NodeId nodeId;
+
+    @Inject
+    public MetricsCollector(OpensearchProcess process, Configuration configuration, NodeId nodeId) {
+        this.process = process;
+        this.configuration = configuration;
+        this.nodeId = nodeId;
+    }
+
+    @Override
+    public boolean runsForever() {
+        return false;
+    }
+
+    @Override
+    public boolean stopOnGracefulShutdown() {
+        return false;
+    }
+
+    @Override
+    public boolean startOnThisNode() {
+        return true;
+    }
+
+    @Override
+    public boolean isDaemon() {
+        return true;
+    }
+
+    @Override
+    public int getInitialDelaySeconds() {
+        return 0;
+    }
+
+    @Override
+    public int getPeriodSeconds() {
+        return 60;
+    }
+
+    @NotNull
+    @Override
+    protected Logger getLogger() {
+        return LOG;
+    }
+
+    @Override
+    public void doRun() {
+        if (process.isInState(ProcessState.AVAILABLE)) {
+            process.restClient().ifPresent(client -> {
+                final IndexRequest indexRequest = new IndexRequest(configuration.getMetricsStream());
+                Map<String, Object> metrics = new HashMap<String, Object>();
+                metrics.put(configuration.getMetricsTimestamp(), new DateTime(DateTimeZone.UTC));
+                metrics.put("node", configuration.getHostname());
+                metrics.put("jvm_heap", Runtime.getRuntime().totalMemory());
+                indexRequest.source(metrics);
+                client.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
+                    @Override
+                    public void onResponse(IndexResponse indexResponse) {
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        LOG.error("Error indexing metrics", e);
+                    }
+                });
+            });
+        }
+    }
+}