From ff7962ab0747ad005845c3d8a577d1472c651bd5 Mon Sep 17 00:00:00 2001
From: Matthias Oesterheld <33032967+moesterheld@users.noreply.github.com>
Date: Mon, 25 Mar 2024 15:10:11 +0100
Subject: [PATCH] fix duplicate startup of data node (#18772)

* fix duplicate startup of data node

* always recreate dataStreamService to ensure correct OS client on configuration change
---
 .../management/OpensearchProcessImpl.java     | 32 +++++++++++++------
 .../management/OpensearchProcessService.java  |  4 +--
 .../ConfigureMetricsIndexSettings.java        | 16 ++++------
 3 files changed, 31 insertions(+), 21 deletions(-)

diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java
index 898848e59a..5b70ca288e 100644
--- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java
+++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java
@@ -17,7 +17,6 @@
 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;
 import org.apache.http.client.utils.URIBuilder;
@@ -180,8 +179,21 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
                     trustManager.refresh();
                     // refresh the seed hosts
                     writeSeedHostsList();
+
+                    boolean startedPreviously = Objects.nonNull(commandLineProcess) && commandLineProcess.processInfo().alive();
+                    if (startedPreviously) {
+                        stop();
+                    }
+
                     commandLineProcess = new OpensearchCommandLineProcess(config, this);
+
+                    if (startedPreviously) {
+                        commandLineProcess.start();
+                    }
+
                     restClient = Optional.of(createRestClient(config));
+                    openSearchClient = restClient.map(c -> new OpenSearchClient(c, objectMapper));
+
                 }),
                 () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");}
         );
@@ -199,14 +211,16 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
     }
     @Override
     public synchronized void start() {
-        opensearchConfiguration.ifPresentOrElse(
-                (config -> {
-                    commandLineProcess.start();
-                    restClient = Optional.of(createRestClient(config));
-                    openSearchClient = restClient.map(c -> new OpenSearchClient(c, objectMapper));
-                }),
-                () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");}
-        );
+        if (Objects.isNull(commandLineProcess) || !commandLineProcess.processInfo().alive()) {
+            opensearchConfiguration.ifPresentOrElse(
+                    (config -> {
+                        commandLineProcess.start();
+                        restClient = Optional.of(createRestClient(config));
+                        openSearchClient = restClient.map(c -> new OpenSearchClient(c, objectMapper));
+                    }),
+                    () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");}
+            );
+        }
     }
 
     @Override
diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java
index 24f8f5e38d..6aa5518fbb 100644
--- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java
+++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java
@@ -34,12 +34,10 @@ import org.graylog2.cluster.nodes.NodeService;
 import org.graylog2.cluster.preflight.DataNodeProvisioningConfig;
 import org.graylog2.cluster.preflight.DataNodeProvisioningService;
 import org.graylog2.cluster.preflight.DataNodeProvisioningStateChangeEvent;
-import org.graylog2.configuration.IndexSetsDefaultConfiguration;
 import org.graylog2.datanode.DataNodeLifecycleEvent;
 import org.graylog2.datanode.RemoteReindexAllowlistEvent;
 import org.graylog2.events.ClusterEventBus;
 import org.graylog2.indexer.fieldtypes.IndexFieldTypesService;
-import org.graylog2.indexer.indexset.IndexSetConfig;
 import org.graylog2.plugin.cluster.ClusterConfigService;
 import org.graylog2.plugin.system.NodeId;
 import org.graylog2.security.CustomCAX509TrustManager;
@@ -126,7 +124,7 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
     @SuppressWarnings("unused")
     public void handlePreflightConfigEvent(DataNodeProvisioningStateChangeEvent event) {
         switch (event.state()) {
-            case STARTUP_REQUESTED -> startUp();
+            case STARTUP_REQUESTED -> this.process.start();
             case STORED -> {
                 configure();
                 dataNodeProvisioningService.changeState(event.nodeId(), DataNodeProvisioningConfig.State.STARTUP_PREPARED);
diff --git a/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java b/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java
index 1401d1504e..563efa8c8a 100644
--- a/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java
+++ b/data-node/src/main/java/org/graylog/datanode/metrics/ConfigureMetricsIndexSettings.java
@@ -76,15 +76,13 @@ public class ConfigureMetricsIndexSettings implements StateMachineTracer {
     public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) {
         if (destination == ProcessState.AVAILABLE && source == ProcessState.STARTING) {
             process.openSearchClient().ifPresent(client -> {
-                if (dataStreamService == null) {
-                    final IsmApi ismApi = new IsmApi(client, objectMapper);
-                    int replicas = nodeService.allActive().size() == 1 ? 0 : 1;
-                    dataStreamService = new DataStreamServiceImpl(
-                            new DataStreamAdapterOS2(client, objectMapper, ismApi),
-                            indexFieldTypesService,
-                            replicas
-                    );
-                }
+                final IsmApi ismApi = new IsmApi(client, objectMapper);
+                int replicas = nodeService.allActive().size() == 1 ? 0 : 1;
+                dataStreamService = new DataStreamServiceImpl(
+                        new DataStreamAdapterOS2(client, objectMapper, ismApi),
+                        indexFieldTypesService,
+                        replicas
+                );
                 dataStreamService.createDataStream(configuration.getMetricsStream(),
                         configuration.getMetricsTimestamp(),
                         createMappings(),