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

Initial refactoring of data node process/state machine (#19101)

Browse Source 
* 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
This commit is contained in:
Matthias Oesterheld
2024-04-19 07:23:07 +02:00
committed by GitHub
parent 6c4f720107
commit 91ded55629
54 changed files with 530 additions and 526 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java
View File

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

6
data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java
View File

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

2
data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java
View File

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

6
data-node/src/main/java/org/graylog/datanode/initializers/JerseyService.java
View File

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

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

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

4
data-node/src/main/java/org/graylog/datanode/management/OpensearchConfigurationChangeEvent.java → data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchConfigurationChangeEvent.java
View File

@@ -14,9 +14,9 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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) {
}

11
data-node/src/main/java/org/graylog/datanode/process/StateMachineTracer.java → data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchInfo.java
View File

@@ -14,12 +14,11 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<ProcessState, ProcessEvent> {
public record OpensearchInfo(String nodeName, OpensearchState state, boolean isLeaderNode, String restBaseUrl,
ProcessInformation process) {
}

10
data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java → data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcess.java
View File

@@ -14,10 +14,12 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<OpensearchConfiguration> {
public interface OpensearchProcess extends ManagableProcess<OpensearchConfiguration, OpensearchEvent, OpensearchState> {
OpensearchInfo processInfo();

41
data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java → data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessImpl.java
View File

@@ -14,22 +14,25 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<RestHighLevelClient> restClient = Optional.empty();
private Optional<OpenSearchClient> 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<DataNodeDto> nodeService, ObjectMapper objectMapper, ProcessStateMachine processState) {
final Configuration configuration, final NodeService<DataNodeDto> 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<OpensearchState, OpensearchEvent> 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);
}
}

20
data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java → data-node/src/main/java/org/graylog/datanode/opensearch/OpensearchProcessService.java
View File

@@ -14,7 +14,7 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<DataNodeDto> 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<DataNodeDto> 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));

4
data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/AbstractOpensearchCli.java → data-node/src/main/java/org/graylog/datanode/opensearch/cli/AbstractOpensearchCli.java
View File

@@ -14,13 +14,13 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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;

4
data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchCli.java → data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCli.java
View File

@@ -14,9 +14,9 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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

8
data-node/src/main/java/org/graylog/datanode/management/OpensearchCommandLineProcess.java → data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchCommandLineProcess.java
View File

@@ -14,7 +14,7 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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;

4
data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchKeystoreCli.java → data-node/src/main/java/org/graylog/datanode/opensearch/cli/OpensearchKeystoreCli.java
View File

@@ -14,9 +14,9 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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;

4
data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java → data-node/src/main/java/org/graylog/datanode/opensearch/configuration/OpensearchConfiguration.java
View File

@@ -14,14 +14,14 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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;

4
data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java → data-node/src/main/java/org/graylog/datanode/opensearch/rest/OpensearchRestClient.java
View File

@@ -14,10 +14,10 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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;

2
data-node/src/main/java/org/graylog/datanode/process/FailuresCounter.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/FailuresCounter.java
View File

@@ -14,7 +14,7 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.process;
package org.graylog.datanode.opensearch.statemachine;
public class FailuresCounter {

4
data-node/src/main/java/org/graylog/datanode/process/ProcessEvent.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchEvent.java
View File

@@ -14,9 +14,9 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.process;
package org.graylog.datanode.opensearch.statemachine;
public enum ProcessEvent {
public enum OpensearchEvent {
PROCESS_PREPARED,
PROCESS_STARTED,
HEALTH_CHECK_OK,

6
data-node/src/main/java/org/graylog/datanode/process/ProcessState.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchState.java
View File

@@ -14,11 +14,11 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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;
}

123
data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachine.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<OpensearchState, OpensearchEvent> {
/**
* 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<OpensearchState, OpensearchEvent> 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<OpensearchState, OpensearchEvent> 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;
}
}

16
data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachineProvider.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/OpensearchStateMachineProvider.java
View File

@@ -14,21 +14,21 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<ProcessStateMachine> {
private final ProcessStateMachine processStateMachine;
public class OpensearchStateMachineProvider implements Provider<OpensearchStateMachine> {
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;
}
}

11
data-node/src/main/java/org/graylog/datanode/management/ClusterNodeStateTracer.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/ClusterNodeStateTracer.java
View File

@@ -14,11 +14,10 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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(),

22
data-node/src/main/java/org/graylog/datanode/management/OpensearchRemovalTracer.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchRemovalTracer.java
View File

@@ -14,13 +14,13 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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);
}
}
}

24
data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/OpensearchWatchdog.java
View File

@@ -14,41 +14,39 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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();

28
data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracer.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<OpensearchState, OpensearchEvent> {
}

11
data-node/src/main/java/org/graylog/datanode/management/StateMachineTracerAggregator.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTracerAggregator.java
View File

@@ -14,11 +14,10 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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));
}
}

11
data-node/src/main/java/org/graylog/datanode/management/StateMachineTransitionLogger.java → data-node/src/main/java/org/graylog/datanode/opensearch/statemachine/tracer/StateMachineTransitionLogger.java
View File

@@ -14,11 +14,10 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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);
}

11
data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java
View File

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

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

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

12
data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java
View File

@@ -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<String> datanodeRestApiUri;
private final Supplier<Boolean> isLeader;
private final Configuration configuration;
private final Supplier<ProcessState> processState;
private final Supplier<OpensearchState> processState;
@Inject
@@ -69,7 +67,7 @@ public class NodePingPeriodical extends Periodical {
Supplier<String> opensearchClusterUri,
Supplier<String> datanodeRestApiUri,
Supplier<Boolean> isLeader,
Supplier<ProcessState> processState
Supplier<OpensearchState> processState
) {
this.nodeService = nodeService;
this.nodeId = nodeId;

20
data-node/src/main/java/org/graylog/datanode/periodicals/OpensearchNodeHeartbeat.java
View File

@@ -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<RestHighLevelClient> 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());
}

17
data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java → data-node/src/main/java/org/graylog/datanode/process/CommandLineProcess.java
View File

@@ -14,35 +14,32 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<String> arguments;
private final ProcessListener listener;
private final Environment environment;
private final WatchdogWithProcessInfo watchDog;
CommandLineProcess(Path executable,
List<String> arguments,
ProcessListener listener,
Environment environment) {
public CommandLineProcess(Path executable,
List<String> arguments,
ProcessListener listener,
Environment environment) {
this.executable = executable;
this.arguments = arguments;
this.listener = listener;

2
data-node/src/main/java/org/graylog/datanode/management/CommandLineProcessListener.java → data-node/src/main/java/org/graylog/datanode/process/CommandLineProcessListener.java
View File

@@ -14,7 +14,7 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.management;
package org.graylog.datanode.process;
import org.apache.commons.exec.ExecuteException;
import org.slf4j.Logger;

2
data-node/src/main/java/org/graylog/datanode/management/Environment.java → data-node/src/main/java/org/graylog/datanode/process/Environment.java
View File

@@ -14,7 +14,7 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.management;
package org.graylog.datanode.process;
import java.util.Collections;
import java.util.HashMap;

2
data-node/src/main/java/org/graylog/datanode/management/LoggingOutputStream.java → data-node/src/main/java/org/graylog/datanode/process/LoggingOutputStream.java
View File

@@ -14,7 +14,7 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.management;
package org.graylog.datanode.process;
import java.io.IOException;
import java.io.OutputStream;

19
data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java → data-node/src/main/java/org/graylog/datanode/process/ManagableProcess.java
View File

@@ -14,15 +14,14 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<T> {
public interface ManagableProcess<T, EVENT, STATE> {
void configure(T configuration);
@@ -30,10 +29,10 @@ public interface ManagableProcess<T> {
void stop();
void onEvent(ProcessEvent event);
void onEvent(EVENT event);
void addStateMachineTracer(StateMachineTracer stateMachineTracer);
void addStateMachineTracer(Trace<STATE, EVENT> tracer);
boolean isInState(ProcessState state);
boolean isInState(STATE state);
}

20
data-node/src/main/java/org/graylog/datanode/process/OpensearchInfo.java
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.process;
public record OpensearchInfo(String nodeName, ProcessState state, boolean isLeaderNode, String restBaseUrl, ProcessInformation process) {
}

2
data-node/src/main/java/org/graylog/datanode/management/ProcessListener.java → data-node/src/main/java/org/graylog/datanode/process/ProcessListener.java
View File

@@ -14,7 +14,7 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.management;
package org.graylog.datanode.process;
import org.apache.commons.exec.ExecuteResultHandler;

123
data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<ProcessState, ProcessEvent> {
/**
* 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<ProcessState, ProcessEvent> 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<ProcessState, ProcessEvent> 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;
}
}

2
data-node/src/main/java/org/graylog/datanode/rest/LogsController.java
View File

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

2
data-node/src/main/java/org/graylog/datanode/rest/StatusController.java
View File

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

2
data-node/src/main/java/org/graylog/datanode/rest/StatusResponse.java
View File

@@ -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) {}