opensource/graylog2-server
1
0
Fork 0
mirror of https://github.com/Graylog2/graylog2-server.git synced 2026-03-13 09:32:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix duplicate startup of data node (#18772)

Browse Source 
* fix duplicate startup of data node

* always recreate dataStreamService to ensure correct OS client on configuration change
This commit is contained in:
Matthias Oesterheld
2024-03-25 15:10:11 +01:00
committed by GitHub
parent 265c3a1774
commit ff7962ab07
3 changed files with 31 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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

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

@@ -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);

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

@@ -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(),