From 91ded55629e2e73ef0f609df2ed33fe2ccdf61f5 Mon Sep 17 00:00:00 2001 From: Matthias Oesterheld <33032967+moesterheld@users.noreply.github.com> Date: Fri, 19 Apr 2024 07:23:07 +0200 Subject: [PATCH] Initial refactoring of data node process/state machine (#19101) * rename state machine to DataNodeStateMachine to make clear it is not only handling the process anymore * initial refactoring of packages and classes: clearly separate processing and opensearch-related classes * Make ManageableProcess independent from OpensearchStateMachine --- .../bindings/GenericInitializerBindings.java | 6 +- .../datanode/bindings/ServerBindings.java | 6 +- .../OpensearchConfigurationProvider.java | 2 +- .../datanode/initializers/JerseyService.java | 6 +- .../ConfigureMetricsIndexSettings.java | 14 +- .../OpensearchConfigurationChangeEvent.java | 4 +- .../OpensearchInfo.java} | 11 +- .../OpensearchProcess.java | 10 +- .../OpensearchProcessImpl.java | 41 ++--- .../OpensearchProcessService.java | 20 ++- .../opensearch/cli/AbstractOpensearchCli.java | 4 +- .../opensearch/cli/OpensearchCli.java | 4 +- .../cli}/OpensearchCommandLineProcess.java | 8 +- .../opensearch/cli/OpensearchKeystoreCli.java | 4 +- .../OpensearchConfiguration.java | 4 +- .../rest}/OpensearchRestClient.java | 4 +- .../statemachine}/FailuresCounter.java | 2 +- .../statemachine/OpensearchEvent.java} | 4 +- .../statemachine/OpensearchState.java} | 6 +- .../statemachine/OpensearchStateMachine.java | 123 ++++++++++++++ .../OpensearchStateMachineProvider.java} | 16 +- .../tracer}/ClusterNodeStateTracer.java | 11 +- .../tracer}/OpensearchRemovalTracer.java | 22 +-- .../tracer/OpensearchWatchdog.java} | 24 ++- .../tracer/StateMachineTracer.java | 28 +++ .../tracer}/StateMachineTracerAggregator.java | 11 +- .../tracer}/StateMachineTransitionLogger.java | 11 +- .../periodicals/ClusterManagerDiscovery.java | 11 +- .../periodicals/MetricsCollector.java | 8 +- .../periodicals/NodePingPeriodical.java | 12 +- .../periodicals/OpensearchNodeHeartbeat.java | 20 +-- .../CommandLineProcess.java | 17 +- .../CommandLineProcessListener.java | 2 +- .../{management => process}/Environment.java | 2 +- .../LoggingOutputStream.java | 2 +- .../ManagableProcess.java | 19 +-- .../datanode/process/OpensearchInfo.java | 20 --- .../ProcessListener.java | 2 +- .../datanode/process/ProcessStateMachine.java | 123 -------------- .../graylog/datanode/rest/LogsController.java | 2 +- .../datanode/rest/StatusController.java | 2 +- .../graylog/datanode/rest/StatusResponse.java | 2 +- data-node/src/main/resources/log4j2.xml | 2 +- .../TestableOpensearchProcess.java} | 33 ++-- .../statemachine}/FailuresCounterTest.java | 2 +- .../OpensearchStateMachineTest.java | 159 ++++++++++++++++++ .../tracer}/OpensearchRemovalTracerTest.java | 13 +- .../tracer/OpensearchWatchdogTracerTest.java} | 23 +-- .../periodicals/NodePingPeriodicalTest.java | 6 +- .../CommandLineProcessTest.java | 3 +- .../EnvironmentTest.java | 4 +- .../LoggingOutputStreamTest.java | 2 +- .../process/ProcessStateMachineTest.java | 159 ------------------ .../{management => process}/test-script.sh | 0 54 files changed, 530 insertions(+), 526 deletions(-) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch}/OpensearchConfigurationChangeEvent.java (86%) rename data-node/src/main/java/org/graylog/datanode/{process/StateMachineTracer.java => opensearch/OpensearchInfo.java} (65%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch}/OpensearchProcess.java (78%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch}/OpensearchProcessImpl.java (87%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch}/OpensearchProcessService.java (91%) rename data-node/src/main/java/org/graylog/datanode/{management => }/opensearch/cli/AbstractOpensearchCli.java (97%) rename data-node/src/main/java/org/graylog/datanode/{management => }/opensearch/cli/OpensearchCli.java (89%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch/cli}/OpensearchCommandLineProcess.java (96%) rename data-node/src/main/java/org/graylog/datanode/{management => }/opensearch/cli/OpensearchKeystoreCli.java (92%) rename data-node/src/main/java/org/graylog/datanode/{process => opensearch/configuration}/OpensearchConfiguration.java (97%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch/rest}/OpensearchRestClient.java (95%) rename data-node/src/main/java/org/graylog/datanode/{process => opensearch/statemachine}/FailuresCounter.java (96%) rename data-node/src/main/java/org/graylog/datanode/{process/ProcessEvent.java => opensearch/statemachine/OpensearchEvent.java} (91%) rename data-node/src/main/java/org/graylog/datanode/{process/ProcessState.java => opensearch/statemachine/OpensearchState.java} (94%) create mode 100644 data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachine.java rename data-node/src/main/java/org/graylog/datanode/{process/ProcessStateMachineProvider.java => opensearch/statemachine/OpensearchStateMachineProvider.java} (64%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch/statemachine/tracer}/ClusterNodeStateTracer.java (85%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch/statemachine/tracer}/OpensearchRemovalTracer.java (89%) rename data-node/src/main/java/org/graylog/datanode/{management/ProcessWatchdog.java => opensearch/statemachine/tracer/OpensearchWatchdog.java} (77%) create mode 100644 data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracer.java rename data-node/src/main/java/org/graylog/datanode/{management => opensearch/statemachine/tracer}/StateMachineTracerAggregator.java (77%) rename data-node/src/main/java/org/graylog/datanode/{management => opensearch/statemachine/tracer}/StateMachineTransitionLogger.java (75%) rename data-node/src/main/java/org/graylog/datanode/{management => process}/CommandLineProcess.java (87%) rename data-node/src/main/java/org/graylog/datanode/{management => process}/CommandLineProcessListener.java (98%) rename data-node/src/main/java/org/graylog/datanode/{management => process}/Environment.java (97%) rename data-node/src/main/java/org/graylog/datanode/{management => process}/LoggingOutputStream.java (97%) rename data-node/src/main/java/org/graylog/datanode/{management => process}/ManagableProcess.java (59%) delete mode 100644 data-node/src/main/java/org/graylog/datanode/process/OpensearchInfo.java rename data-node/src/main/java/org/graylog/datanode/{management => process}/ProcessListener.java (95%) delete mode 100644 data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java rename data-node/src/test/java/org/graylog/datanode/{management/TestableProcess.java => opensearch/TestableOpensearchProcess.java} (55%) rename data-node/src/test/java/org/graylog/datanode/{process => opensearch/statemachine}/FailuresCounterTest.java (95%) create mode 100644 data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineTest.java rename data-node/src/test/java/org/graylog/datanode/{management => opensearch/statemachine/tracer}/OpensearchRemovalTracerTest.java (90%) rename data-node/src/test/java/org/graylog/datanode/{management/ProcessWatchdogTracerTest.java => opensearch/statemachine/tracer/OpensearchWatchdogTracerTest.java} (65%) rename data-node/src/test/java/org/graylog/datanode/{management => process}/CommandLineProcessTest.java (98%) rename data-node/src/test/java/org/graylog/datanode/{management => process}/EnvironmentTest.java (91%) rename data-node/src/test/java/org/graylog/datanode/{management => process}/LoggingOutputStreamTest.java (97%) delete mode 100644 data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java rename data-node/src/test/resources/org/graylog/datanode/{management => process}/test-script.sh (100%) diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java index 48b0acb678..87ed18c30f 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java @@ -21,9 +21,9 @@ import com.google.inject.AbstractModule; import com.google.inject.multibindings.Multibinder; import org.graylog.datanode.configuration.OpensearchConfigurationProvider; import org.graylog.datanode.initializers.PeriodicalsService; -import org.graylog.datanode.management.OpensearchProcess; -import org.graylog.datanode.management.OpensearchProcessService; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.OpensearchProcess; +import org.graylog.datanode.opensearch.OpensearchProcessService; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; import org.graylog.datanode.shutdown.GracefulShutdownService; import org.graylog2.bootstrap.preflight.PreflightConfigService; import org.graylog2.bootstrap.preflight.PreflightConfigServiceImpl; diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java index 4ca67d67c8..e2069e2dfe 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java @@ -23,8 +23,8 @@ import com.google.inject.multibindings.OptionalBinder; import jakarta.ws.rs.container.DynamicFeature; import jakarta.ws.rs.ext.ExceptionMapper; import org.graylog.datanode.Configuration; -import org.graylog.datanode.process.ProcessStateMachine; -import org.graylog.datanode.process.ProcessStateMachineProvider; +import org.graylog.datanode.opensearch.statemachine.OpensearchStateMachine; +import org.graylog.datanode.opensearch.statemachine.OpensearchStateMachineProvider; import org.graylog.datanode.shared.system.activities.DataNodeActivityWriter; import org.graylog2.bindings.providers.ClusterEventBusProvider; import org.graylog2.cluster.ClusterConfigServiceImpl; @@ -68,7 +68,7 @@ public class ServerBindings extends Graylog2Module { bind(ClusterEventBus.class).toProvider(ClusterEventBusProvider.class).asEagerSingleton(); bind(EventBus.class).toProvider(EventBusProvider.class).asEagerSingleton(); bind(InputConfigurationBeanDeserializerModifier.class).toInstance(InputConfigurationBeanDeserializerModifier.withoutConfig()); - bind(ProcessStateMachine.class).toProvider(ProcessStateMachineProvider.class).asEagerSingleton(); + bind(OpensearchStateMachine.class).toProvider(OpensearchStateMachineProvider.class).asEagerSingleton(); } private void bindFactoryModules() { diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java index 611967cb72..fc74e1b064 100644 --- a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java +++ b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java @@ -27,7 +27,7 @@ import org.graylog.datanode.configuration.variants.MongoCertSecureConfiguration; import org.graylog.datanode.configuration.variants.OpensearchSecurityConfiguration; import org.graylog.datanode.configuration.variants.SecurityConfigurationVariant; import org.graylog.datanode.configuration.variants.UploadedCertFilesSecureConfiguration; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; import org.graylog.security.certutil.ca.exceptions.KeyStoreStorageException; import org.graylog2.cluster.Node; import org.graylog2.cluster.nodes.DataNodeDto; diff --git a/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java b/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java index f01dc5073a..d92e1c365f 100644 --- a/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java +++ b/data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java @@ -27,6 +27,7 @@ import com.google.common.eventbus.Subscribe; import com.google.common.net.HostAndPort; import com.google.common.util.concurrent.AbstractIdleService; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import jakarta.annotation.Nonnull; import jakarta.inject.Inject; import jakarta.inject.Named; import jakarta.ws.rs.container.ContainerRequestFilter; @@ -47,8 +48,8 @@ import org.glassfish.jersey.server.model.Resource; import org.graylog.datanode.Configuration; import org.graylog.datanode.configuration.variants.KeystoreInformation; import org.graylog.datanode.configuration.variants.OpensearchSecurityConfiguration; -import org.graylog.datanode.management.OpensearchConfigurationChangeEvent; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.OpensearchConfigurationChangeEvent; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; import org.graylog.datanode.rest.config.SecuredNodeAnnotationFilter; import org.graylog.security.certutil.CertConstants; import org.graylog2.bootstrap.preflight.web.BasicAuthFilter; @@ -58,7 +59,6 @@ import org.graylog2.rest.MoreMediaTypes; import org.graylog2.shared.rest.exceptionmappers.JacksonPropertyExceptionMapper; import org.graylog2.shared.rest.exceptionmappers.JsonProcessingExceptionMapper; import org.graylog2.shared.security.tls.KeyStoreUtils; -import jakarta.annotation.Nonnull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; 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 563efa8c8a..5b76ee20af 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 @@ -21,11 +21,11 @@ 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.opensearch.OpensearchProcess; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.opensearch.statemachine.tracer.StateMachineTracer; import org.graylog.datanode.periodicals.MetricsCollector; -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.graylog2.cluster.nodes.DataNodeDto; @@ -69,12 +69,12 @@ public class ConfigureMetricsIndexSettings implements StateMachineTracer { } @Override - public void trigger(ProcessEvent trigger) { + public void trigger(OpensearchEvent trigger) { } @Override - public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) { - if (destination == ProcessState.AVAILABLE && source == ProcessState.STARTING) { + public void transition(OpensearchEvent trigger, OpensearchState source, OpensearchState destination) { + if (destination == OpensearchState.AVAILABLE && source == OpensearchState.STARTING) { process.openSearchClient().ifPresent(client -> { final IsmApi ismApi = new IsmApi(client, objectMapper); int replicas = nodeService.allActive().size() == 1 ? 0 : 1; diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchConfigurationChangeEvent.java b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchConfigurationChangeEvent.java similarity index 86% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchConfigurationChangeEvent.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchConfigurationChangeEvent.java index c7bea34f40..deb5886366 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchConfigurationChangeEvent.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchConfigurationChangeEvent.java @@ -14,9 +14,9 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; public record OpensearchConfigurationChangeEvent(OpensearchConfiguration config) { } diff --git a/data-node/src/main/java/org/graylog/datanode/process/StateMachineTracer.java b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchInfo.java similarity index 65% rename from data-node/src/main/java/org/graylog/datanode/process/StateMachineTracer.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchInfo.java index e06489664b..275d973126 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/StateMachineTracer.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchInfo.java @@ -14,12 +14,11 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.process; +package org.graylog.datanode.opensearch; -import com.github.oxo42.stateless4j.delegates.Trace; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.process.ProcessInformation; -/** - * The tracer allows to observe triggered event (before) and transitions (after) of the {@link ProcessStateMachine} - */ -public interface StateMachineTracer extends Trace { +public record OpensearchInfo(String nodeName, OpensearchState state, boolean isLeaderNode, String restBaseUrl, + ProcessInformation process) { } diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcess.java similarity index 78% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcess.java index 691f552676..5430962103 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcess.java @@ -14,10 +14,12 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch; -import org.graylog.datanode.process.OpensearchConfiguration; -import org.graylog.datanode.process.OpensearchInfo; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.process.ManagableProcess; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; import org.graylog.storage.opensearch2.OpenSearchClient; @@ -25,7 +27,7 @@ import java.net.URI; import java.util.List; import java.util.Optional; -public interface OpensearchProcess extends ManagableProcess { +public interface OpensearchProcess extends ManagableProcess { OpensearchInfo processInfo(); diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java similarity index 87% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java index 6e9497d6a7..89e0d369f8 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java @@ -14,22 +14,25 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch; import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.oxo42.stateless4j.delegates.Trace; import org.apache.commons.collections4.queue.CircularFifoQueue; import org.apache.commons.exec.ExecuteException; import org.apache.http.client.utils.URIBuilder; import org.graylog.datanode.Configuration; import org.graylog.datanode.configuration.DatanodeConfiguration; import org.graylog.datanode.configuration.variants.OpensearchSecurityConfiguration; -import org.graylog.datanode.process.OpensearchConfiguration; -import org.graylog.datanode.process.OpensearchInfo; -import org.graylog.datanode.process.ProcessEvent; +import org.graylog.datanode.opensearch.cli.OpensearchCommandLineProcess; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; +import org.graylog.datanode.opensearch.rest.OpensearchRestClient; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.opensearch.statemachine.OpensearchStateMachine; +import org.graylog.datanode.opensearch.statemachine.tracer.StateMachineTracer; import org.graylog.datanode.process.ProcessInformation; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.ProcessStateMachine; -import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.datanode.process.ProcessListener; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; import org.graylog.storage.opensearch2.OpenSearchClient; import org.graylog2.cluster.nodes.DataNodeDto; @@ -63,7 +66,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { private Optional restClient = Optional.empty(); private Optional openSearchClient = Optional.empty(); - private final ProcessStateMachine processState; + private final OpensearchStateMachine processState; private final DatanodeConfiguration datanodeConfiguration; @@ -79,7 +82,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager, - final Configuration configuration, final NodeService nodeService, ObjectMapper objectMapper, ProcessStateMachine processState) { + final Configuration configuration, final NodeService nodeService, ObjectMapper objectMapper, OpensearchStateMachine processState) { this.datanodeConfiguration = datanodeConfiguration; this.processState = processState; this.stdout = new CircularFifoQueue<>(logsCacheSize); @@ -143,14 +146,14 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { return String.format(Locale.ROOT, "%s://%s:%d", protocol, host, port); } - public void onEvent(ProcessEvent event) { + public void onEvent(OpensearchEvent event) { LOG.debug("Process event: " + event); this.processState.fire(event); } @Override - public void addStateMachineTracer(StateMachineTracer stateMachineTracer) { - this.processState.getTracerAggregator().addTracer(stateMachineTracer); + public void addStateMachineTracer(Trace stateMachineTracer) { + this.processState.getTracerAggregator().addTracer((StateMachineTracer) stateMachineTracer); } public void setLeaderNode(boolean isLeaderNode) { @@ -162,7 +165,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { return isLeaderNode; } - public boolean isInState(ProcessState expectedState) { + public boolean isInState(OpensearchState expectedState) { return this.processState.getState().equals(expectedState); } @@ -241,7 +244,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { private void stopProcess() { if (this.commandLineProcess != null) { - onEvent(ProcessEvent.PROCESS_STOPPED); + onEvent(OpensearchEvent.PROCESS_STOPPED); commandLineProcess.close(); } } @@ -250,13 +253,13 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { public void onRemove() { LOG.info("Starting removal of OpenSearch node"); if (this.commandLineProcess != null) { - onEvent(ProcessEvent.PROCESS_REMOVE); + onEvent(OpensearchEvent.PROCESS_REMOVE); } } @Override public void onReset() { - onEvent(ProcessEvent.RESET); + onEvent(OpensearchEvent.RESET); stop(); configure(); start(); @@ -264,7 +267,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { @Override public void onStart() { - onEvent(ProcessEvent.PROCESS_STARTED); + onEvent(OpensearchEvent.PROCESS_STARTED); } @Override @@ -282,12 +285,12 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { @Override public void onProcessComplete(int exitValue) { LOG.info("Opensearch process completed with exit code {}", exitValue); - onEvent(ProcessEvent.PROCESS_TERMINATED); + onEvent(OpensearchEvent.PROCESS_TERMINATED); } @Override public void onProcessFailed(ExecuteException e) { LOG.warn("Opensearch process failed", e); - onEvent(ProcessEvent.PROCESS_TERMINATED); + onEvent(OpensearchEvent.PROCESS_TERMINATED); } } diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessService.java similarity index 91% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessService.java index 499f6b54ba..5d3b6cc552 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessService.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.eventbus.EventBus; @@ -28,8 +28,12 @@ import org.graylog.datanode.bootstrap.preflight.DatanodeDirectoriesLockfileCheck import org.graylog.datanode.configuration.DatanodeConfiguration; import org.graylog.datanode.configuration.OpensearchConfigurationProvider; import org.graylog.datanode.metrics.ConfigureMetricsIndexSettings; -import org.graylog.datanode.process.OpensearchConfiguration; -import org.graylog.datanode.process.ProcessStateMachine; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; +import org.graylog.datanode.opensearch.statemachine.OpensearchStateMachine; +import org.graylog.datanode.opensearch.statemachine.tracer.ClusterNodeStateTracer; +import org.graylog.datanode.opensearch.statemachine.tracer.OpensearchRemovalTracer; +import org.graylog.datanode.opensearch.statemachine.tracer.OpensearchWatchdog; +import org.graylog.datanode.opensearch.statemachine.tracer.StateMachineTransitionLogger; import org.graylog2.bootstrap.preflight.PreflightConfigResult; import org.graylog2.bootstrap.preflight.PreflightConfigService; import org.graylog2.cluster.nodes.DataNodeDto; @@ -75,7 +79,7 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro final NodeId nodeId, final IndexFieldTypesService indexFieldTypesService, final ObjectMapper objectMapper, - final ProcessStateMachine processStateMachine, + final OpensearchStateMachine opensearchStateMachine, final ClusterEventBus clusterEventBus, final DatanodeDirectoriesLockfileCheck lockfileCheck, final PreflightConfigService preflightConfigService) { @@ -88,14 +92,14 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro this.clusterEventBus = clusterEventBus; this.lockfileCheck = lockfileCheck; this.preflightConfigService = preflightConfigService; - this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper, processStateMachine); + this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper, opensearchStateMachine); eventBus.register(this); } private OpensearchProcess createOpensearchProcess(final DatanodeConfiguration datanodeConfiguration, final CustomCAX509TrustManager trustManager, final Configuration configuration, - final NodeService nodeService, final ObjectMapper objectMapper, final ProcessStateMachine processStateMachine) { - final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize(), trustManager, configuration, nodeService, objectMapper, processStateMachine); - final ProcessWatchdog watchdog = new ProcessWatchdog(process, WATCHDOG_RESTART_ATTEMPTS); + final NodeService nodeService, final ObjectMapper objectMapper, final OpensearchStateMachine opensearchStateMachine) { + final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize(), trustManager, configuration, nodeService, objectMapper, opensearchStateMachine); + final OpensearchWatchdog watchdog = new OpensearchWatchdog(process, WATCHDOG_RESTART_ATTEMPTS); process.addStateMachineTracer(watchdog); process.addStateMachineTracer(new StateMachineTransitionLogger()); process.addStateMachineTracer(new OpensearchRemovalTracer(process, configuration.getDatanodeNodeName(), nodeId, clusterEventBus)); diff --git a/data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/AbstractOpensearchCli.java b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/AbstractOpensearchCli.java similarity index 97% rename from data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/AbstractOpensearchCli.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/cli/AbstractOpensearchCli.java index 9671541f23..6541c95bca 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/AbstractOpensearchCli.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/AbstractOpensearchCli.java @@ -14,13 +14,13 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management.opensearch.cli; +package org.graylog.datanode.opensearch.cli; import org.apache.commons.exec.CommandLine; import org.apache.commons.exec.DefaultExecuteResultHandler; import org.apache.commons.exec.DefaultExecutor; import org.apache.commons.exec.PumpStreamHandler; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; diff --git a/data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchCli.java b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCli.java similarity index 89% rename from data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchCli.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCli.java index 697dfdb3f2..cd00589999 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchCli.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCli.java @@ -14,9 +14,9 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management.opensearch.cli; +package org.graylog.datanode.opensearch.cli; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; /** * Collection of opensearch CLI tools. All of them need to have OPENSEARCH_PATH_CONF preconfigured, so they operate diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchCommandLineProcess.java b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCommandLineProcess.java similarity index 96% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchCommandLineProcess.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCommandLineProcess.java index e9ceae341e..86eb2b60b9 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchCommandLineProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCommandLineProcess.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.cli; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; @@ -26,9 +26,11 @@ import com.github.rholder.retry.StopStrategies; import com.github.rholder.retry.WaitStrategies; import jakarta.validation.constraints.NotNull; import org.apache.commons.exec.OS; -import org.graylog.datanode.management.opensearch.cli.OpensearchCli; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; +import org.graylog.datanode.process.CommandLineProcess; +import org.graylog.datanode.process.CommandLineProcessListener; import org.graylog.datanode.process.ProcessInformation; +import org.graylog.datanode.process.ProcessListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchKeystoreCli.java b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchKeystoreCli.java similarity index 92% rename from data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchKeystoreCli.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchKeystoreCli.java index 79d3483698..bf3fb6cad6 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchKeystoreCli.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchKeystoreCli.java @@ -14,9 +14,9 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management.opensearch.cli; +package org.graylog.datanode.opensearch.cli; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; import java.util.Collections; diff --git a/data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java b/data-node/src/main/java/org/graylog/datanode/opensearch/configuration/OpensearchConfiguration.java similarity index 97% rename from data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/configuration/OpensearchConfiguration.java index 550ce48c9e..969860034f 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/configuration/OpensearchConfiguration.java @@ -14,14 +14,14 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.process; +package org.graylog.datanode.opensearch.configuration; import org.apache.commons.exec.OS; import org.graylog.datanode.OpensearchDistribution; import org.graylog.datanode.configuration.DatanodeDirectories; import org.graylog.datanode.configuration.S3RepositoryConfiguration; import org.graylog.datanode.configuration.variants.OpensearchSecurityConfiguration; -import org.graylog.datanode.management.Environment; +import org.graylog.datanode.process.Environment; import org.graylog.shaded.opensearch2.org.apache.http.HttpHost; import java.util.LinkedHashMap; diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java b/data-node/src/main/java/org/graylog/datanode/opensearch/rest/OpensearchRestClient.java similarity index 95% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/rest/OpensearchRestClient.java index ba75092d25..ac31a6e385 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/rest/OpensearchRestClient.java @@ -14,10 +14,10 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.rest; import org.graylog.datanode.configuration.DatanodeConfiguration; -import org.graylog.datanode.process.OpensearchConfiguration; +import org.graylog.datanode.opensearch.configuration.OpensearchConfiguration; import org.graylog.shaded.opensearch2.org.apache.http.HttpHost; import org.graylog.shaded.opensearch2.org.apache.http.HttpRequestInterceptor; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; diff --git a/data-node/src/main/java/org/graylog/datanode/process/FailuresCounter.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/FailuresCounter.java similarity index 96% rename from data-node/src/main/java/org/graylog/datanode/process/FailuresCounter.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/FailuresCounter.java index eb85646e8c..89d85e032e 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/FailuresCounter.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/FailuresCounter.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.process; +package org.graylog.datanode.opensearch.statemachine; public class FailuresCounter { diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchEvent.java similarity index 91% rename from data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchEvent.java index 28fb3456da..4b44c079f5 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchEvent.java @@ -14,9 +14,9 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.process; +package org.graylog.datanode.opensearch.statemachine; -public enum ProcessEvent { +public enum OpensearchEvent { PROCESS_PREPARED, PROCESS_STARTED, HEALTH_CHECK_OK, diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchState.java similarity index 94% rename from data-node/src/main/java/org/graylog/datanode/process/ProcessState.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchState.java index 2c1221f204..8b83327f23 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchState.java @@ -14,11 +14,11 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.process; +package org.graylog.datanode.opensearch.statemachine; import org.graylog2.cluster.nodes.DataNodeStatus; -public enum ProcessState { +public enum OpensearchState { /** * Fresh created process, not started yet */ @@ -63,7 +63,7 @@ public enum ProcessState { private final DataNodeStatus dataNodeStatus; - ProcessState(DataNodeStatus dataNodeStatus) { + OpensearchState(DataNodeStatus dataNodeStatus) { this.dataNodeStatus = dataNodeStatus; } diff --git a/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachine.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachine.java new file mode 100644 index 0000000000..dfce0bf634 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachine.java @@ -0,0 +1,123 @@ +/* + * 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 + * . + */ +package org.graylog.datanode.opensearch.statemachine; + +import com.github.oxo42.stateless4j.StateMachine; +import com.github.oxo42.stateless4j.StateMachineConfig; +import org.graylog.datanode.opensearch.statemachine.tracer.StateMachineTracerAggregator; + +public class OpensearchStateMachine extends StateMachine { + /** + * How many times can the OS rest api call fail before we switch to the failed state + */ + public static final int MAX_REST_TEMPORARY_FAILURES = 3; + public static final int MAX_REST_STARTUP_FAILURES = 5; + public static final int MAX_REBOOT_FAILURES = 3; + + StateMachineTracerAggregator tracerAggregator = new StateMachineTracerAggregator(); + + public OpensearchStateMachine(OpensearchState initialState, StateMachineConfig config) { + super(initialState, config); + setTrace(tracerAggregator); + } + + public static OpensearchStateMachine createNew() { + + final FailuresCounter restFailureCounter = FailuresCounter.oneBased(MAX_REST_TEMPORARY_FAILURES); + final FailuresCounter startupFailuresCounter = FailuresCounter.oneBased(MAX_REST_STARTUP_FAILURES); + final FailuresCounter rebootCounter = FailuresCounter.oneBased(MAX_REBOOT_FAILURES); + + StateMachineConfig config = new StateMachineConfig<>(); + + // Freshly created process, it hasn't started yet and doesn't have any pid. + config.configure(OpensearchState.WAITING_FOR_CONFIGURATION) + .permit(OpensearchEvent.PROCESS_PREPARED, OpensearchState.PREPARED) + // jump to started only allowed to facilitate startup with insecure config + .permit(OpensearchEvent.PROCESS_STARTED, OpensearchState.STARTING) + .ignore(OpensearchEvent.PROCESS_STOPPED) + .ignore(OpensearchEvent.HEALTH_CHECK_FAILED); + + config.configure(OpensearchState.PREPARED) + .permit(OpensearchEvent.PROCESS_STARTED, OpensearchState.STARTING) + .permit(OpensearchEvent.PROCESS_TERMINATED, OpensearchState.TERMINATED) + .permit(OpensearchEvent.PROCESS_STOPPED, OpensearchState.TERMINATED) + .ignore(OpensearchEvent.HEALTH_CHECK_FAILED); + + // the process has started already, now we have to wait for a running OS and available REST api + // the startupFailuresCounter keeps track of failed REST status calls and allow failures during the + // startup period + config.configure(OpensearchState.STARTING) + .permitDynamic(OpensearchEvent.HEALTH_CHECK_FAILED, + () -> startupFailuresCounter.failedTooManyTimes() ? OpensearchState.FAILED : OpensearchState.STARTING, + startupFailuresCounter::increment) + .permit(OpensearchEvent.HEALTH_CHECK_OK, OpensearchState.AVAILABLE) + .permit(OpensearchEvent.PROCESS_STOPPED, OpensearchState.TERMINATED) + .permit(OpensearchEvent.PROCESS_TERMINATED, OpensearchState.TERMINATED); + + // the process is running and responding to the REST status, it's available for any usage + config.configure(OpensearchState.AVAILABLE) + .onEntry(restFailureCounter::resetFailuresCounter) + .onEntry(rebootCounter::resetFailuresCounter) + .permitReentry(OpensearchEvent.HEALTH_CHECK_OK) + .permit(OpensearchEvent.HEALTH_CHECK_FAILED, OpensearchState.NOT_RESPONDING) + .permit(OpensearchEvent.PROCESS_STOPPED, OpensearchState.TERMINATED) + .permit(OpensearchEvent.PROCESS_TERMINATED, OpensearchState.TERMINATED) + .permit(OpensearchEvent.PROCESS_REMOVE, OpensearchState.REMOVING) + .ignore(OpensearchEvent.PROCESS_STARTED); + + // if the REST api is not responding, we'll jump to this state and count how many times the failure + // occurs. If it fails ttoo many times, we'll mark the process as FAILED + config.configure(OpensearchState.NOT_RESPONDING) + .permitDynamic(OpensearchEvent.HEALTH_CHECK_FAILED, + () -> restFailureCounter.failedTooManyTimes() ? OpensearchState.FAILED : OpensearchState.NOT_RESPONDING, + restFailureCounter::increment + ) + .permit(OpensearchEvent.HEALTH_CHECK_OK, OpensearchState.AVAILABLE) + .permit(OpensearchEvent.PROCESS_STOPPED, OpensearchState.TERMINATED) + .permit(OpensearchEvent.PROCESS_TERMINATED, OpensearchState.TERMINATED); + + // failed and we see the process as not recoverable. + // TODO: what to do if the process fails? Reboot? + config.configure(OpensearchState.FAILED) + .ignore(OpensearchEvent.HEALTH_CHECK_FAILED) + .permit(OpensearchEvent.HEALTH_CHECK_OK, OpensearchState.AVAILABLE) + .permit(OpensearchEvent.PROCESS_STOPPED, OpensearchState.TERMINATED) + .permit(OpensearchEvent.PROCESS_TERMINATED, OpensearchState.TERMINATED); + + // final state, the process is not alive anymore, terminated on the operating system level + config.configure(OpensearchState.TERMINATED) + .permit(OpensearchEvent.PROCESS_STARTED, OpensearchState.STARTING, rebootCounter::increment) + .ignore(OpensearchEvent.HEALTH_CHECK_FAILED) + .ignore(OpensearchEvent.PROCESS_STOPPED) + .ignore(OpensearchEvent.PROCESS_TERMINATED); // final state, all following terminate events are ignored + + config.configure(OpensearchState.REMOVING) + .ignore(OpensearchEvent.HEALTH_CHECK_OK) + .permit(OpensearchEvent.HEALTH_CHECK_FAILED, OpensearchState.FAILED) + .permit(OpensearchEvent.PROCESS_STOPPED, OpensearchState.REMOVED); + + config.configure(OpensearchState.REMOVED) + .permit(OpensearchEvent.RESET, OpensearchState.WAITING_FOR_CONFIGURATION) + .ignore(OpensearchEvent.PROCESS_STOPPED); + + return new OpensearchStateMachine(OpensearchState.WAITING_FOR_CONFIGURATION, config); + } + + public StateMachineTracerAggregator getTracerAggregator() { + return tracerAggregator; + } +} diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachineProvider.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineProvider.java similarity index 64% rename from data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachineProvider.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineProvider.java index f4bd0f77ae..8dde0e0a3a 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachineProvider.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineProvider.java @@ -14,21 +14,21 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.process; +package org.graylog.datanode.opensearch.statemachine; -import jakarta.inject.Provider; import jakarta.inject.Inject; +import jakarta.inject.Provider; -public class ProcessStateMachineProvider implements Provider { - private final ProcessStateMachine processStateMachine; +public class OpensearchStateMachineProvider implements Provider { + private final OpensearchStateMachine opensearchStateMachine; @Inject - public ProcessStateMachineProvider() { - this.processStateMachine = ProcessStateMachine.createNew(); + public OpensearchStateMachineProvider() { + this.opensearchStateMachine = OpensearchStateMachine.createNew(); } @Override - public ProcessStateMachine get() { - return processStateMachine; + public OpensearchStateMachine get() { + return opensearchStateMachine; } } diff --git a/data-node/src/main/java/org/graylog/datanode/management/ClusterNodeStateTracer.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/ClusterNodeStateTracer.java similarity index 85% rename from data-node/src/main/java/org/graylog/datanode/management/ClusterNodeStateTracer.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/ClusterNodeStateTracer.java index 81cd5b6785..e421ecaa51 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/ClusterNodeStateTracer.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/ClusterNodeStateTracer.java @@ -14,11 +14,10 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.statemachine.tracer; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.graylog2.cluster.NodeNotFoundException; import org.graylog2.cluster.nodes.DataNodeDto; import org.graylog2.cluster.nodes.NodeService; @@ -40,11 +39,11 @@ public class ClusterNodeStateTracer implements StateMachineTracer { } @Override - public void trigger(ProcessEvent processEvent) { + public void trigger(OpensearchEvent processEvent) { } @Override - public void transition(ProcessEvent processEvent, ProcessState source, ProcessState destination) { + public void transition(OpensearchEvent processEvent, OpensearchState source, OpensearchState destination) { try { if (!source.equals(destination)) { log.info("Updating cluster node {} from {} to {} (reason: {})", nodeId.getNodeId(), diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchRemovalTracer.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchRemovalTracer.java similarity index 89% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchRemovalTracer.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchRemovalTracer.java index 8bcb868bc2..c85d8cb057 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchRemovalTracer.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchRemovalTracer.java @@ -14,13 +14,13 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.statemachine.tracer; import com.google.common.eventbus.EventBus; import com.google.common.util.concurrent.ThreadFactoryBuilder; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.datanode.opensearch.OpensearchProcess; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.graylog.shaded.opensearch2.org.opensearch.OpenSearchStatusException; import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.health.ClusterHealthRequest; import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.health.ClusterHealthResponse; @@ -70,15 +70,15 @@ public class OpensearchRemovalTracer implements StateMachineTracer { @Override - public void trigger(ProcessEvent trigger) { + public void trigger(OpensearchEvent trigger) { LOG.debug("Removal tracer trigger: {}", trigger); } @Override - public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) { - if (destination == ProcessState.AVAILABLE && !allocationExcludeChecked) { + public void transition(OpensearchEvent trigger, OpensearchState source, OpensearchState destination) { + if (destination == OpensearchState.AVAILABLE && !allocationExcludeChecked) { checkAllocationEnabledStatus(); - } else if (trigger == ProcessEvent.PROCESS_REMOVE) { + } else if (trigger == OpensearchEvent.PROCESS_REMOVE) { triggerRemoval(); } } @@ -119,11 +119,11 @@ public class OpensearchRemovalTracer implements StateMachineTracer { executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("datanode-removal").build()); executorService.scheduleAtFixedRate(this::checkRemovalStatus, 10, 10, TimeUnit.SECONDS); } else { - process.onEvent(ProcessEvent.HEALTH_CHECK_FAILED); + process.onEvent(OpensearchEvent.HEALTH_CHECK_FAILED); } } catch (IOException e) { LOG.error("Failed to exclude node from cluster allocation", e); - process.onEvent(ProcessEvent.HEALTH_CHECK_FAILED); + process.onEvent(OpensearchEvent.HEALTH_CHECK_FAILED); } }); } @@ -141,7 +141,7 @@ public class OpensearchRemovalTracer implements StateMachineTracer { eventBus.post(DataNodeLifecycleEvent.create(nodeId.getNodeId(), DataNodeLifecycleTrigger.REMOVED)); } } catch (IOException | OpenSearchStatusException e) { - process.onEvent(ProcessEvent.HEALTH_CHECK_FAILED); + process.onEvent(OpensearchEvent.HEALTH_CHECK_FAILED); } } } diff --git a/data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchWatchdog.java similarity index 77% rename from data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchWatchdog.java index a509ae6880..cc2f1b6564 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchWatchdog.java @@ -14,41 +14,39 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.statemachine.tracer; -import org.graylog.datanode.process.FailuresCounter; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.datanode.opensearch.statemachine.FailuresCounter; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.process.ManagableProcess; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; - /** * This process watchdog follows transitions of the state machine and will try to restart the process in case of termination. * If the process is actually stopped, it won't restart it and will automatically deactivate itself. */ -public class ProcessWatchdog implements StateMachineTracer { +public class OpensearchWatchdog implements StateMachineTracer { - private static final Logger LOG = LoggerFactory.getLogger(ProcessWatchdog.class); + private static final Logger LOG = LoggerFactory.getLogger(OpensearchWatchdog.class); private boolean active; private final FailuresCounter restartCounter; - private final ManagableProcess process; + private final ManagableProcess process; - public ProcessWatchdog(ManagableProcess process, int restartAttemptsCount) { + public OpensearchWatchdog(ManagableProcess process, int restartAttemptsCount) { this.process = process; this.restartCounter = FailuresCounter.zeroBased(restartAttemptsCount); } @Override - public void trigger(ProcessEvent trigger) { + public void trigger(OpensearchEvent trigger) { LOG.debug("Watchdog trigger: {}", trigger); } @Override - public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) { + public void transition(OpensearchEvent trigger, OpensearchState source, OpensearchState destination) { LOG.debug("Watchdog transition event:{}, source:{}, destination:{}", trigger, source, destination); switch (trigger) { case PROCESS_STARTED -> activateWatchdog(); diff --git a/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracer.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracer.java new file mode 100644 index 0000000000..59025c6a5e --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracer.java @@ -0,0 +1,28 @@ +/* + * 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 + * . + */ +package org.graylog.datanode.opensearch.statemachine.tracer; + +import com.github.oxo42.stateless4j.delegates.Trace; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.opensearch.statemachine.OpensearchStateMachine; + +/** + * The tracer allows to observe triggered event (before) and transitions (after) of the {@link OpensearchStateMachine} + */ +public interface StateMachineTracer extends Trace { +} diff --git a/data-node/src/main/java/org/graylog/datanode/management/StateMachineTracerAggregator.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracerAggregator.java similarity index 77% rename from data-node/src/main/java/org/graylog/datanode/management/StateMachineTracerAggregator.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracerAggregator.java index 6ad92ba5e5..4473eca802 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/StateMachineTracerAggregator.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracerAggregator.java @@ -14,11 +14,10 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.statemachine.tracer; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import java.util.LinkedList; import java.util.List; @@ -36,12 +35,12 @@ public class StateMachineTracerAggregator implements StateMachineTracer { } @Override - public void trigger(ProcessEvent processEvent) { + public void trigger(OpensearchEvent processEvent) { delegates.forEach(d -> d.trigger(processEvent)); } @Override - public void transition(ProcessEvent processEvent, ProcessState s1, ProcessState s2) { + public void transition(OpensearchEvent processEvent, OpensearchState s1, OpensearchState s2) { delegates.forEach(d -> d.transition(processEvent, s1, s2)); } } diff --git a/data-node/src/main/java/org/graylog/datanode/management/StateMachineTransitionLogger.java b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTransitionLogger.java similarity index 75% rename from data-node/src/main/java/org/graylog/datanode/management/StateMachineTransitionLogger.java rename to data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTransitionLogger.java index cc97d0b2c4..0e9ae743e6 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/StateMachineTransitionLogger.java +++ b/data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTransitionLogger.java @@ -14,11 +14,10 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.statemachine.tracer; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.StateMachineTracer; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,12 +26,12 @@ public class StateMachineTransitionLogger implements StateMachineTracer { private static final Logger LOG = LoggerFactory.getLogger(StateMachineTransitionLogger.class); @Override - public void trigger(ProcessEvent trigger) { + public void trigger(OpensearchEvent trigger) { } @Override - public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) { + public void transition(OpensearchEvent trigger, OpensearchState source, OpensearchState destination) { if (!source.equals(destination)) { LOG.debug("Triggered {}, source state: {}, destination: {}", trigger, source, destination); } diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java b/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java index 2888bbd62f..d95912db9d 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java @@ -17,10 +17,12 @@ package org.graylog.datanode.periodicals; import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; import org.graylog.datanode.Configuration; import org.graylog.datanode.configuration.DatanodeConfiguration; -import org.graylog.datanode.management.OpensearchProcess; -import org.graylog.datanode.process.ProcessState; +import org.graylog.datanode.opensearch.OpensearchProcess; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.graylog.shaded.opensearch2.org.opensearch.client.Request; import org.graylog.shaded.opensearch2.org.opensearch.client.Response; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; @@ -28,9 +30,6 @@ import org.graylog2.plugin.periodical.Periodical; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - import java.io.IOException; import java.util.Optional; @@ -54,7 +53,7 @@ public class ClusterManagerDiscovery extends Periodical { @Override // This method is "synchronized" because we are also calling it directly in AutomaticLeaderElectionService public synchronized void doRun() { - if (managedOpenSearch.isInState(ProcessState.AVAILABLE)) { + if (managedOpenSearch.isInState(OpensearchState.AVAILABLE)) { final Boolean isManagerNode = getClusterStateResponse(managedOpenSearch) .map(r -> r.nodes().get(r.clusterManagerNode())) .map(managerNode -> configuration.getDatanodeNodeName().equals(managerNode.name())) diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java b/data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java index 34cdcc4a52..ff52bcf1d4 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/MetricsCollector.java @@ -18,12 +18,13 @@ package org.graylog.datanode.periodicals; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.ImmutableMap; +import jakarta.annotation.Nonnull; import jakarta.inject.Inject; import org.graylog.datanode.Configuration; -import org.graylog.datanode.management.OpensearchProcess; +import org.graylog.datanode.opensearch.OpensearchProcess; import org.graylog.datanode.metrics.ClusterStatMetricsCollector; import org.graylog.datanode.metrics.NodeMetricsCollector; -import org.graylog.datanode.process.ProcessState; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; 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; @@ -38,7 +39,6 @@ import org.graylog.shaded.opensearch2.org.opensearch.search.builder.SearchSource import org.graylog.shaded.opensearch2.org.opensearch.search.sort.SortBuilders; import org.graylog.shaded.opensearch2.org.opensearch.search.sort.SortOrder; import org.graylog2.plugin.periodical.Periodical; -import jakarta.annotation.Nonnull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -106,7 +106,7 @@ public class MetricsCollector extends Periodical { @Override public void doRun() { - if (process.isInState(ProcessState.AVAILABLE)) { + if (process.isInState(OpensearchState.AVAILABLE)) { process.restClient().ifPresent(client -> { this.nodeStatMetricsCollector = new NodeMetricsCollector(client, objectMapper); this.clusterStatMetricsCollector = new ClusterStatMetricsCollector(client, objectMapper); diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java b/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java index 48eae6d7d6..9a43082aee 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java @@ -16,9 +16,10 @@ */ 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.datanode.opensearch.OpensearchProcess; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.graylog2.cluster.nodes.DataNodeDto; import org.graylog2.cluster.nodes.DataNodeStatus; import org.graylog2.cluster.nodes.NodeService; @@ -28,9 +29,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.annotation.Nonnull; - -import jakarta.inject.Inject; - import java.net.URI; import java.util.function.Supplier; @@ -44,7 +42,7 @@ public class NodePingPeriodical extends Periodical { private final Supplier datanodeRestApiUri; private final Supplier isLeader; private final Configuration configuration; - private final Supplier processState; + private final Supplier processState; @Inject @@ -69,7 +67,7 @@ public class NodePingPeriodical extends Periodical { Supplier opensearchClusterUri, Supplier datanodeRestApiUri, Supplier isLeader, - Supplier processState + Supplier processState ) { this.nodeService = nodeService; this.nodeId = nodeId; diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java b/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java index 5b768e3bf7..59816950b5 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java @@ -16,9 +16,11 @@ */ package org.graylog.datanode.periodicals; -import org.graylog.datanode.management.OpensearchProcess; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; +import org.graylog.datanode.opensearch.OpensearchProcess; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.graylog.shaded.opensearch2.org.opensearch.OpenSearchStatusException; import org.graylog.shaded.opensearch2.org.opensearch.client.RequestOptions; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; @@ -28,10 +30,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.annotation.Nonnull; - -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - import java.io.IOException; import java.util.Optional; @@ -49,8 +47,8 @@ public class OpensearchNodeHeartbeat extends Periodical { @Override // This method is "synchronized" because we are also calling it directly in AutomaticLeaderElectionService public synchronized void doRun() { - if (!process.isInState(ProcessState.TERMINATED) && !process.isInState(ProcessState.WAITING_FOR_CONFIGURATION) - && !process.isInState(ProcessState.REMOVED)) { + if (!process.isInState(OpensearchState.TERMINATED) && !process.isInState(OpensearchState.WAITING_FOR_CONFIGURATION) + && !process.isInState(OpensearchState.REMOVED)) { final Optional restClient = process.restClient(); if (restClient.isPresent()) { @@ -66,11 +64,11 @@ public class OpensearchNodeHeartbeat extends Periodical { } private void onNodeResponse(OpensearchProcess process, MainResponse nodeResponse) { - process.onEvent(ProcessEvent.HEALTH_CHECK_OK); + process.onEvent(OpensearchEvent.HEALTH_CHECK_OK); } private void onRestError(OpensearchProcess process, Exception e) { - process.onEvent(ProcessEvent.HEALTH_CHECK_FAILED); + process.onEvent(OpensearchEvent.HEALTH_CHECK_FAILED); LOG.warn("Opensearch REST api of process {} unavailable. Cause: {}", process.processInfo().process().pid(), e.getMessage()); } diff --git a/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java b/data-node/src/main/java/org/graylog/datanode/process/CommandLineProcess.java similarity index 87% rename from data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java rename to data-node/src/main/java/org/graylog/datanode/process/CommandLineProcess.java index 2588281055..29a475dfbc 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/process/CommandLineProcess.java @@ -14,35 +14,32 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; +import jakarta.validation.constraints.NotNull; import org.apache.commons.exec.CommandLine; import org.apache.commons.exec.DefaultExecutor; import org.apache.commons.exec.ExecuteWatchdog; import org.apache.commons.exec.PumpStreamHandler; -import org.graylog.datanode.process.ProcessInformation; -import org.graylog.datanode.process.WatchdogWithProcessInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import jakarta.validation.constraints.NotNull; - import java.io.IOException; import java.nio.file.Path; import java.util.List; -class CommandLineProcess { +public class CommandLineProcess { private final Path executable; private final List arguments; private final ProcessListener listener; private final Environment environment; private final WatchdogWithProcessInfo watchDog; - CommandLineProcess(Path executable, - List arguments, - ProcessListener listener, - Environment environment) { + public CommandLineProcess(Path executable, + List arguments, + ProcessListener listener, + Environment environment) { this.executable = executable; this.arguments = arguments; this.listener = listener; diff --git a/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcessListener.java b/data-node/src/main/java/org/graylog/datanode/process/CommandLineProcessListener.java similarity index 98% rename from data-node/src/main/java/org/graylog/datanode/management/CommandLineProcessListener.java rename to data-node/src/main/java/org/graylog/datanode/process/CommandLineProcessListener.java index 6884efa1f6..19ca391757 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcessListener.java +++ b/data-node/src/main/java/org/graylog/datanode/process/CommandLineProcessListener.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; import org.apache.commons.exec.ExecuteException; import org.slf4j.Logger; diff --git a/data-node/src/main/java/org/graylog/datanode/management/Environment.java b/data-node/src/main/java/org/graylog/datanode/process/Environment.java similarity index 97% rename from data-node/src/main/java/org/graylog/datanode/management/Environment.java rename to data-node/src/main/java/org/graylog/datanode/process/Environment.java index 7a67ecf638..56738354a8 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/Environment.java +++ b/data-node/src/main/java/org/graylog/datanode/process/Environment.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; import java.util.Collections; import java.util.HashMap; diff --git a/data-node/src/main/java/org/graylog/datanode/management/LoggingOutputStream.java b/data-node/src/main/java/org/graylog/datanode/process/LoggingOutputStream.java similarity index 97% rename from data-node/src/main/java/org/graylog/datanode/management/LoggingOutputStream.java rename to data-node/src/main/java/org/graylog/datanode/process/LoggingOutputStream.java index 391ecb1afa..d69970999f 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/LoggingOutputStream.java +++ b/data-node/src/main/java/org/graylog/datanode/process/LoggingOutputStream.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; import java.io.IOException; import java.io.OutputStream; diff --git a/data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java b/data-node/src/main/java/org/graylog/datanode/process/ManagableProcess.java similarity index 59% rename from data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java rename to data-node/src/main/java/org/graylog/datanode/process/ManagableProcess.java index fffe7606e2..ee35c18919 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ManagableProcess.java @@ -14,15 +14,14 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.StateMachineTracer; +import com.github.oxo42.stateless4j.delegates.Trace; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.opensearch.statemachine.tracer.StateMachineTracer; -import java.io.IOException; - -public interface ManagableProcess { +public interface ManagableProcess { void configure(T configuration); @@ -30,10 +29,10 @@ public interface ManagableProcess { void stop(); - void onEvent(ProcessEvent event); + void onEvent(EVENT event); - void addStateMachineTracer(StateMachineTracer stateMachineTracer); + void addStateMachineTracer(Trace tracer); - boolean isInState(ProcessState state); + boolean isInState(STATE state); } diff --git a/data-node/src/main/java/org/graylog/datanode/process/OpensearchInfo.java b/data-node/src/main/java/org/graylog/datanode/process/OpensearchInfo.java deleted file mode 100644 index 3f8ca3691f..0000000000 --- a/data-node/src/main/java/org/graylog/datanode/process/OpensearchInfo.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * 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 - * . - */ -package org.graylog.datanode.process; - -public record OpensearchInfo(String nodeName, ProcessState state, boolean isLeaderNode, String restBaseUrl, ProcessInformation process) { -} diff --git a/data-node/src/main/java/org/graylog/datanode/management/ProcessListener.java b/data-node/src/main/java/org/graylog/datanode/process/ProcessListener.java similarity index 95% rename from data-node/src/main/java/org/graylog/datanode/management/ProcessListener.java rename to data-node/src/main/java/org/graylog/datanode/process/ProcessListener.java index 5e2672c43d..b2631a9752 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/ProcessListener.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ProcessListener.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; import org.apache.commons.exec.ExecuteResultHandler; diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java b/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java deleted file mode 100644 index d9c52a6574..0000000000 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java +++ /dev/null @@ -1,123 +0,0 @@ -/* - * 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 - * . - */ -package org.graylog.datanode.process; - -import com.github.oxo42.stateless4j.StateMachine; -import com.github.oxo42.stateless4j.StateMachineConfig; -import org.graylog.datanode.management.StateMachineTracerAggregator; - -public class ProcessStateMachine extends StateMachine { - /** - * How many times can the OS rest api call fail before we switch to the failed state - */ - public static final int MAX_REST_TEMPORARY_FAILURES = 3; - public static final int MAX_REST_STARTUP_FAILURES = 5; - public static final int MAX_REBOOT_FAILURES = 3; - - StateMachineTracerAggregator tracerAggregator = new StateMachineTracerAggregator(); - - public ProcessStateMachine(ProcessState initialState, StateMachineConfig config) { - super(initialState, config); - setTrace(tracerAggregator); - } - - public static ProcessStateMachine createNew() { - - final FailuresCounter restFailureCounter = FailuresCounter.oneBased(MAX_REST_TEMPORARY_FAILURES); - final FailuresCounter startupFailuresCounter = FailuresCounter.oneBased(MAX_REST_STARTUP_FAILURES); - final FailuresCounter rebootCounter = FailuresCounter.oneBased(MAX_REBOOT_FAILURES); - - StateMachineConfig config = new StateMachineConfig<>(); - - // Freshly created process, it hasn't started yet and doesn't have any pid. - config.configure(ProcessState.WAITING_FOR_CONFIGURATION) - .permit(ProcessEvent.PROCESS_PREPARED, ProcessState.PREPARED) - // jump to started only allowed to facilitate startup with insecure config - .permit(ProcessEvent.PROCESS_STARTED, ProcessState.STARTING) - .ignore(ProcessEvent.PROCESS_STOPPED) - .ignore(ProcessEvent.HEALTH_CHECK_FAILED); - - config.configure(ProcessState.PREPARED) - .permit(ProcessEvent.PROCESS_STARTED, ProcessState.STARTING) - .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED) - .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.TERMINATED) - .ignore(ProcessEvent.HEALTH_CHECK_FAILED); - - // the process has started already, now we have to wait for a running OS and available REST api - // the startupFailuresCounter keeps track of failed REST status calls and allow failures during the - // startup period - config.configure(ProcessState.STARTING) - .permitDynamic(ProcessEvent.HEALTH_CHECK_FAILED, - () -> startupFailuresCounter.failedTooManyTimes() ? ProcessState.FAILED : ProcessState.STARTING, - startupFailuresCounter::increment) - .permit(ProcessEvent.HEALTH_CHECK_OK, ProcessState.AVAILABLE) - .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.TERMINATED) - .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED); - - // the process is running and responding to the REST status, it's available for any usage - config.configure(ProcessState.AVAILABLE) - .onEntry(restFailureCounter::resetFailuresCounter) - .onEntry(rebootCounter::resetFailuresCounter) - .permitReentry(ProcessEvent.HEALTH_CHECK_OK) - .permit(ProcessEvent.HEALTH_CHECK_FAILED, ProcessState.NOT_RESPONDING) - .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.TERMINATED) - .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED) - .permit(ProcessEvent.PROCESS_REMOVE, ProcessState.REMOVING) - .ignore(ProcessEvent.PROCESS_STARTED); - - // if the REST api is not responding, we'll jump to this state and count how many times the failure - // occurs. If it fails ttoo many times, we'll mark the process as FAILED - config.configure(ProcessState.NOT_RESPONDING) - .permitDynamic(ProcessEvent.HEALTH_CHECK_FAILED, - () -> restFailureCounter.failedTooManyTimes() ? ProcessState.FAILED : ProcessState.NOT_RESPONDING, - restFailureCounter::increment - ) - .permit(ProcessEvent.HEALTH_CHECK_OK, ProcessState.AVAILABLE) - .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.TERMINATED) - .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED); - - // failed and we see the process as not recoverable. - // TODO: what to do if the process fails? Reboot? - config.configure(ProcessState.FAILED) - .ignore(ProcessEvent.HEALTH_CHECK_FAILED) - .permit(ProcessEvent.HEALTH_CHECK_OK, ProcessState.AVAILABLE) - .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.TERMINATED) - .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED); - - // final state, the process is not alive anymore, terminated on the operating system level - config.configure(ProcessState.TERMINATED) - .permit(ProcessEvent.PROCESS_STARTED, ProcessState.STARTING, rebootCounter::increment) - .ignore(ProcessEvent.HEALTH_CHECK_FAILED) - .ignore(ProcessEvent.PROCESS_STOPPED) - .ignore(ProcessEvent.PROCESS_TERMINATED); // final state, all following terminate events are ignored - - config.configure(ProcessState.REMOVING) - .ignore(ProcessEvent.HEALTH_CHECK_OK) - .permit(ProcessEvent.HEALTH_CHECK_FAILED, ProcessState.FAILED) - .permit(ProcessEvent.PROCESS_STOPPED, ProcessState.REMOVED); - - config.configure(ProcessState.REMOVED) - .permit(ProcessEvent.RESET, ProcessState.WAITING_FOR_CONFIGURATION) - .ignore(ProcessEvent.PROCESS_STOPPED); - - return new ProcessStateMachine(ProcessState.WAITING_FOR_CONFIGURATION, config); - } - - public StateMachineTracerAggregator getTracerAggregator() { - return tracerAggregator; - } -} diff --git a/data-node/src/main/java/org/graylog/datanode/rest/LogsController.java b/data-node/src/main/java/org/graylog/datanode/rest/LogsController.java index 4d1f9ff8fc..7d0f75af00 100644 --- a/data-node/src/main/java/org/graylog/datanode/rest/LogsController.java +++ b/data-node/src/main/java/org/graylog/datanode/rest/LogsController.java @@ -16,7 +16,7 @@ */ package org.graylog.datanode.rest; -import org.graylog.datanode.management.OpensearchProcess; +import org.graylog.datanode.opensearch.OpensearchProcess; import jakarta.inject.Inject; diff --git a/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java b/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java index b4e62bac85..0dea32a4d6 100644 --- a/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java +++ b/data-node/src/main/java/org/graylog/datanode/rest/StatusController.java @@ -17,7 +17,7 @@ package org.graylog.datanode.rest; import org.graylog.datanode.configuration.DatanodeConfiguration; -import org.graylog.datanode.management.OpensearchProcess; +import org.graylog.datanode.opensearch.OpensearchProcess; import org.graylog2.plugin.Version; import jakarta.inject.Inject; diff --git a/data-node/src/main/java/org/graylog/datanode/rest/StatusResponse.java b/data-node/src/main/java/org/graylog/datanode/rest/StatusResponse.java index 4b9b8fc5e1..fb85bb3936 100644 --- a/data-node/src/main/java/org/graylog/datanode/rest/StatusResponse.java +++ b/data-node/src/main/java/org/graylog/datanode/rest/StatusResponse.java @@ -16,6 +16,6 @@ */ package org.graylog.datanode.rest; -import org.graylog.datanode.process.OpensearchInfo; +import org.graylog.datanode.opensearch.OpensearchInfo; public record StatusResponse(String opensearchVersion, OpensearchInfo node) {} diff --git a/data-node/src/main/resources/log4j2.xml b/data-node/src/main/resources/log4j2.xml index 6a9428fa56..2b7bb91e79 100644 --- a/data-node/src/main/resources/log4j2.xml +++ b/data-node/src/main/resources/log4j2.xml @@ -8,7 +8,7 @@ - + diff --git a/data-node/src/test/java/org/graylog/datanode/management/TestableProcess.java b/data-node/src/test/java/org/graylog/datanode/opensearch/TestableOpensearchProcess.java similarity index 55% rename from data-node/src/test/java/org/graylog/datanode/management/TestableProcess.java rename to data-node/src/test/java/org/graylog/datanode/opensearch/TestableOpensearchProcess.java index 214578958b..6789074641 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/TestableProcess.java +++ b/data-node/src/test/java/org/graylog/datanode/opensearch/TestableOpensearchProcess.java @@ -14,56 +14,57 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch; import com.github.oxo42.stateless4j.StateMachine; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; -import org.graylog.datanode.process.ProcessStateMachine; -import org.graylog.datanode.process.StateMachineTracer; +import com.github.oxo42.stateless4j.delegates.Trace; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; +import org.graylog.datanode.opensearch.statemachine.OpensearchStateMachine; +import org.graylog.datanode.process.ManagableProcess; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class TestableProcess implements ManagableProcess { +public class TestableOpensearchProcess implements ManagableProcess { - private static final Logger LOG = LoggerFactory.getLogger(TestableProcess.class); + private static final Logger LOG = LoggerFactory.getLogger(TestableOpensearchProcess.class); - private final StateMachine stateMachine; + private final StateMachine stateMachine; - public TestableProcess() { - this.stateMachine = ProcessStateMachine.createNew(); + public TestableOpensearchProcess() { + this.stateMachine = OpensearchStateMachine.createNew(); } @Override public void configure(String ignored) { LOG.debug("Preparing process"); - onEvent(ProcessEvent.PROCESS_PREPARED); + onEvent(OpensearchEvent.PROCESS_PREPARED); } @Override public void start() { LOG.debug("Starting process"); - onEvent(ProcessEvent.PROCESS_STARTED); + onEvent(OpensearchEvent.PROCESS_STARTED); } @Override public void stop() { LOG.debug("Stopping process process"); - onEvent(ProcessEvent.PROCESS_STOPPED); + onEvent(OpensearchEvent.PROCESS_STOPPED); } @Override - public void onEvent(ProcessEvent event) { + public void onEvent(OpensearchEvent event) { stateMachine.fire(event); } @Override - public void addStateMachineTracer(StateMachineTracer stateMachineTracer) { + public void addStateMachineTracer(Trace stateMachineTracer) { stateMachine.setTrace(stateMachineTracer); } @Override - public boolean isInState(ProcessState state) { + public boolean isInState(OpensearchState state) { return this.stateMachine.isInState(state); } } diff --git a/data-node/src/test/java/org/graylog/datanode/process/FailuresCounterTest.java b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/FailuresCounterTest.java similarity index 95% rename from data-node/src/test/java/org/graylog/datanode/process/FailuresCounterTest.java rename to data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/FailuresCounterTest.java index 21666458e6..4decb60331 100644 --- a/data-node/src/test/java/org/graylog/datanode/process/FailuresCounterTest.java +++ b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/FailuresCounterTest.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.process; +package org.graylog.datanode.opensearch.statemachine; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; diff --git a/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineTest.java b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineTest.java new file mode 100644 index 0000000000..f556c59e88 --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineTest.java @@ -0,0 +1,159 @@ +/* + * 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 + * . + */ +package org.graylog.datanode.opensearch.statemachine; + +import com.github.oxo42.stateless4j.StateMachine; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class OpensearchStateMachineTest { + + @Test + void testOptimalScenario() { + final StateMachine machine = OpensearchStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), OpensearchState.WAITING_FOR_CONFIGURATION); + + machine.fire(OpensearchEvent.PROCESS_PREPARED); + Assertions.assertEquals(OpensearchState.PREPARED, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_STARTED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_OK); + Assertions.assertEquals(OpensearchState.AVAILABLE, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_TERMINATED); + Assertions.assertEquals(OpensearchState.TERMINATED, machine.getState()); + } + + @Test + void testRestFailing() { + final StateMachine machine = OpensearchStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), OpensearchState.WAITING_FOR_CONFIGURATION); + + machine.fire(OpensearchEvent.PROCESS_PREPARED); + Assertions.assertEquals(OpensearchState.PREPARED, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_STARTED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_OK); + Assertions.assertEquals(OpensearchState.AVAILABLE, machine.getState()); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + + // three failures are still accepted + Assertions.assertEquals(OpensearchState.NOT_RESPONDING, machine.getState()); + + // fourth should bring the state to FAILED + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + Assertions.assertEquals(OpensearchState.FAILED, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_OK); + Assertions.assertEquals(OpensearchState.AVAILABLE, machine.getState()); + } + + @Test + void testStartupFailure() { + final StateMachine machine = OpensearchStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), OpensearchState.WAITING_FOR_CONFIGURATION); + + machine.fire(OpensearchEvent.PROCESS_PREPARED); + Assertions.assertEquals(OpensearchState.PREPARED, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_STARTED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + // after five repetitions we give up waiting for the process start and fail + Assertions.assertEquals(OpensearchState.FAILED, machine.getState()); + } + + @Test + void testStartupFailureResolved() { + final StateMachine machine = OpensearchStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), OpensearchState.WAITING_FOR_CONFIGURATION); + + machine.fire(OpensearchEvent.PROCESS_PREPARED); + Assertions.assertEquals(OpensearchState.PREPARED, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_STARTED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + machine.fire(OpensearchEvent.HEALTH_CHECK_OK); + // succeeded just in time before we give up + Assertions.assertEquals(OpensearchState.AVAILABLE, machine.getState()); + } + + @Test + void testSuccessfullRemoval() { + final StateMachine machine = OpensearchStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), OpensearchState.WAITING_FOR_CONFIGURATION); + + machine.fire(OpensearchEvent.PROCESS_PREPARED); + Assertions.assertEquals(OpensearchState.PREPARED, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_STARTED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_OK); + Assertions.assertEquals(OpensearchState.AVAILABLE, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_REMOVE); + Assertions.assertEquals(OpensearchState.REMOVING, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_STOPPED); + Assertions.assertEquals(OpensearchState.REMOVED, machine.getState()); + + } + + @Test + void testFailingRemoval() { + final StateMachine machine = OpensearchStateMachine.createNew(); + Assertions.assertEquals(machine.getState(), OpensearchState.WAITING_FOR_CONFIGURATION); + + machine.fire(OpensearchEvent.PROCESS_PREPARED); + Assertions.assertEquals(OpensearchState.PREPARED, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_STARTED); + Assertions.assertEquals(OpensearchState.STARTING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_OK); + Assertions.assertEquals(OpensearchState.AVAILABLE, machine.getState()); + + machine.fire(OpensearchEvent.PROCESS_REMOVE); + Assertions.assertEquals(OpensearchState.REMOVING, machine.getState()); + + machine.fire(OpensearchEvent.HEALTH_CHECK_FAILED); + Assertions.assertEquals(OpensearchState.FAILED, machine.getState()); + + } + +} diff --git a/data-node/src/test/java/org/graylog/datanode/management/OpensearchRemovalTracerTest.java b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchRemovalTracerTest.java similarity index 90% rename from data-node/src/test/java/org/graylog/datanode/management/OpensearchRemovalTracerTest.java rename to data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchRemovalTracerTest.java index c8a74cd5b4..15d817750d 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/OpensearchRemovalTracerTest.java +++ b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchRemovalTracerTest.java @@ -14,11 +14,12 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.statemachine.tracer; import com.google.common.eventbus.EventBus; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; +import org.graylog.datanode.opensearch.OpensearchProcess; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.health.ClusterHealthResponse; import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterGetSettingsResponse; import org.graylog.shaded.opensearch2.org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; @@ -79,7 +80,7 @@ public class OpensearchRemovalTracerTest { .put(OpensearchRemovalTracer.CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING, NODENAME) .build(); when(clusterClient.getSettings(any(), any())).thenReturn(new ClusterGetSettingsResponse(null, settings, null)); - classUnderTest.transition(ProcessEvent.HEALTH_CHECK_OK, ProcessState.STARTING, ProcessState.AVAILABLE); + classUnderTest.transition(OpensearchEvent.HEALTH_CHECK_OK, OpensearchState.STARTING, OpensearchState.AVAILABLE); ArgumentCaptor settingsRequest = ArgumentCaptor.forClass(ClusterUpdateSettingsRequest.class); @@ -97,7 +98,7 @@ public class OpensearchRemovalTracerTest { .put(OpensearchRemovalTracer.CLUSTER_ROUTING_ALLOCATION_EXCLUDE_SETTING, "notmynodename") .build(); when(clusterClient.getSettings(any(), any())).thenReturn(new ClusterGetSettingsResponse(null, settings, null)); - classUnderTest.transition(ProcessEvent.HEALTH_CHECK_OK, ProcessState.STARTING, ProcessState.AVAILABLE); + classUnderTest.transition(OpensearchEvent.HEALTH_CHECK_OK, OpensearchState.STARTING, OpensearchState.AVAILABLE); verify(clusterClient).getSettings(any(), any()); verifyNoMoreInteractions(clusterClient); assertTrue(classUnderTest.allocationExcludeChecked); @@ -108,7 +109,7 @@ public class OpensearchRemovalTracerTest { ClusterUpdateSettingsResponse response = mock(ClusterUpdateSettingsResponse.class); when(response.isAcknowledged()).thenReturn(true); when(clusterClient.putSettings(any(), any())).thenReturn(response); - classUnderTest.transition(ProcessEvent.PROCESS_REMOVE, ProcessState.AVAILABLE, ProcessState.REMOVING); + classUnderTest.transition(OpensearchEvent.PROCESS_REMOVE, OpensearchState.AVAILABLE, OpensearchState.REMOVING); ArgumentCaptor settingsRequest = ArgumentCaptor.forClass(ClusterUpdateSettingsRequest.class); diff --git a/data-node/src/test/java/org/graylog/datanode/management/ProcessWatchdogTracerTest.java b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchWatchdogTracerTest.java similarity index 65% rename from data-node/src/test/java/org/graylog/datanode/management/ProcessWatchdogTracerTest.java rename to data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchWatchdogTracerTest.java index 1b9dae6143..eebc255547 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/ProcessWatchdogTracerTest.java +++ b/data-node/src/test/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchWatchdogTracerTest.java @@ -14,20 +14,21 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.opensearch.statemachine.tracer; import org.assertj.core.api.Assertions; -import org.graylog.datanode.process.ProcessEvent; -import org.graylog.datanode.process.ProcessState; +import org.graylog.datanode.opensearch.TestableOpensearchProcess; +import org.graylog.datanode.opensearch.statemachine.OpensearchEvent; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.junit.jupiter.api.Test; -class ProcessWatchdogTracerTest { +class OpensearchWatchdogTracerTest { @Test void testLifecycle() { - final TestableProcess process = new TestableProcess(); - final ProcessWatchdog watchdog = new ProcessWatchdog(process, 3); + final TestableOpensearchProcess process = new TestableOpensearchProcess(); + final OpensearchWatchdog watchdog = new OpensearchWatchdog(process, 3); process.addStateMachineTracer(watchdog); process.configure("ignored"); process.start(); @@ -49,14 +50,14 @@ class ProcessWatchdogTracerTest { terminateProcess(process); Assertions.assertThat(watchdog.isActive()).isFalse(); - Assertions.assertThat(process.isInState(ProcessState.TERMINATED)).isTrue(); + Assertions.assertThat(process.isInState(OpensearchState.TERMINATED)).isTrue(); } - private void terminateProcess(TestableProcess process) { - process.onEvent(ProcessEvent.PROCESS_TERMINATED); + private void terminateProcess(TestableOpensearchProcess process) { + process.onEvent(OpensearchEvent.PROCESS_TERMINATED); } - private boolean isInStartingState(TestableProcess process) { - return process.isInState(ProcessState.STARTING); + private boolean isInStartingState(TestableOpensearchProcess process) { + return process.isInState(OpensearchState.STARTING); } } diff --git a/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java b/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java index 37f6839715..5077b52695 100644 --- a/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java +++ b/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java @@ -17,7 +17,7 @@ package org.graylog.datanode.periodicals; import org.graylog.datanode.Configuration; -import org.graylog.datanode.process.ProcessState; +import org.graylog.datanode.opensearch.statemachine.OpensearchState; import org.graylog2.cluster.NodeNotFoundException; import org.graylog2.cluster.nodes.DataNodeDto; import org.graylog2.cluster.nodes.NodeService; @@ -50,7 +50,7 @@ class NodePingPeriodicalTest { () -> cluster, () -> datanodeRestApi, () -> true, - () -> ProcessState.AVAILABLE + () -> OpensearchState.AVAILABLE ); task.doRun(); @@ -61,7 +61,7 @@ class NodePingPeriodicalTest { .setTransportAddress(uri.toString()) .setClusterAddress(cluster) .setRestApiAddress(datanodeRestApi) - .setDataNodeStatus(ProcessState.AVAILABLE.getDataNodeStatus()) + .setDataNodeStatus(OpensearchState.AVAILABLE.getDataNodeStatus()) .setHostname("localhost") .build() )); diff --git a/data-node/src/test/java/org/graylog/datanode/management/CommandLineProcessTest.java b/data-node/src/test/java/org/graylog/datanode/process/CommandLineProcessTest.java similarity index 98% rename from data-node/src/test/java/org/graylog/datanode/management/CommandLineProcessTest.java rename to data-node/src/test/java/org/graylog/datanode/process/CommandLineProcessTest.java index 8af568316d..f0f7808986 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/CommandLineProcessTest.java +++ b/data-node/src/test/java/org/graylog/datanode/process/CommandLineProcessTest.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; import com.github.rholder.retry.RetryException; import com.github.rholder.retry.Retryer; @@ -23,7 +23,6 @@ import com.github.rholder.retry.StopStrategies; import com.github.rholder.retry.WaitStrategies; import org.apache.commons.exec.ExecuteException; import org.assertj.core.api.Assertions; -import org.graylog.datanode.process.ProcessEvent; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.slf4j.Logger; diff --git a/data-node/src/test/java/org/graylog/datanode/management/EnvironmentTest.java b/data-node/src/test/java/org/graylog/datanode/process/EnvironmentTest.java similarity index 91% rename from data-node/src/test/java/org/graylog/datanode/management/EnvironmentTest.java rename to data-node/src/test/java/org/graylog/datanode/process/EnvironmentTest.java index 59d821ff22..5bc742cdf0 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/EnvironmentTest.java +++ b/data-node/src/test/java/org/graylog/datanode/process/EnvironmentTest.java @@ -14,15 +14,13 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; import java.util.Map; -import static org.junit.jupiter.api.Assertions.*; - class EnvironmentTest { @Test diff --git a/data-node/src/test/java/org/graylog/datanode/management/LoggingOutputStreamTest.java b/data-node/src/test/java/org/graylog/datanode/process/LoggingOutputStreamTest.java similarity index 97% rename from data-node/src/test/java/org/graylog/datanode/management/LoggingOutputStreamTest.java rename to data-node/src/test/java/org/graylog/datanode/process/LoggingOutputStreamTest.java index 24d54b7d62..29a0c5da9d 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/LoggingOutputStreamTest.java +++ b/data-node/src/test/java/org/graylog/datanode/process/LoggingOutputStreamTest.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog.datanode.process; import org.apache.commons.exec.StreamPumper; import org.assertj.core.api.Assertions; diff --git a/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java b/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java deleted file mode 100644 index 7d91c0eff8..0000000000 --- a/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java +++ /dev/null @@ -1,159 +0,0 @@ -/* - * 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 - * . - */ -package org.graylog.datanode.process; - -import com.github.oxo42.stateless4j.StateMachine; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; - -class ProcessStateMachineTest { - - @Test - void testOptimalScenario() { - final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); - - machine.fire(ProcessEvent.PROCESS_PREPARED); - Assertions.assertEquals(ProcessState.PREPARED, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_STARTED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_OK); - Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_TERMINATED); - Assertions.assertEquals(ProcessState.TERMINATED, machine.getState()); - } - - @Test - void testRestFailing() { - final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); - - machine.fire(ProcessEvent.PROCESS_PREPARED); - Assertions.assertEquals(ProcessState.PREPARED, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_STARTED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_OK); - Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - - // three failures are still accepted - Assertions.assertEquals(ProcessState.NOT_RESPONDING, machine.getState()); - - // fourth should bring the state to FAILED - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - Assertions.assertEquals(ProcessState.FAILED, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_OK); - Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); - } - - @Test - void testStartupFailure() { - final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); - - machine.fire(ProcessEvent.PROCESS_PREPARED); - Assertions.assertEquals(ProcessState.PREPARED, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_STARTED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - // after five repetitions we give up waiting for the process start and fail - Assertions.assertEquals(ProcessState.FAILED, machine.getState()); - } - - @Test - void testStartupFailureResolved() { - final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); - - machine.fire(ProcessEvent.PROCESS_PREPARED); - Assertions.assertEquals(ProcessState.PREPARED, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_STARTED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - machine.fire(ProcessEvent.HEALTH_CHECK_OK); - // succeeded just in time before we give up - Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); - } - - @Test - void testSuccessfullRemoval() { - final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); - - machine.fire(ProcessEvent.PROCESS_PREPARED); - Assertions.assertEquals(ProcessState.PREPARED, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_STARTED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_OK); - Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_REMOVE); - Assertions.assertEquals(ProcessState.REMOVING, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_STOPPED); - Assertions.assertEquals(ProcessState.REMOVED, machine.getState()); - - } - - @Test - void testFailingRemoval() { - final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); - - machine.fire(ProcessEvent.PROCESS_PREPARED); - Assertions.assertEquals(ProcessState.PREPARED, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_STARTED); - Assertions.assertEquals(ProcessState.STARTING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_OK); - Assertions.assertEquals(ProcessState.AVAILABLE, machine.getState()); - - machine.fire(ProcessEvent.PROCESS_REMOVE); - Assertions.assertEquals(ProcessState.REMOVING, machine.getState()); - - machine.fire(ProcessEvent.HEALTH_CHECK_FAILED); - Assertions.assertEquals(ProcessState.FAILED, machine.getState()); - - } - -} diff --git a/data-node/src/test/resources/org/graylog/datanode/management/test-script.sh b/data-node/src/test/resources/org/graylog/datanode/process/test-script.sh similarity index 100% rename from data-node/src/test/resources/org/graylog/datanode/management/test-script.sh rename to data-node/src/test/resources/org/graylog/datanode/process/test-script.sh