diff --git a/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java b/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java index a7f7c51bc2..a4acbbfa27 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/management/CommandLineProcess.java @@ -33,8 +33,6 @@ import java.util.concurrent.TimeoutException; class CommandLineProcess { - private static final String JAVA_HOME_ENV = "JAVA_HOME"; - private final Path executable; private final List arguments; private final ProcessListener listener; @@ -57,7 +55,7 @@ class CommandLineProcess { private static final Logger LOG = LoggerFactory.getLogger(CommandLineProcess.class); - public void start() throws IOException { + public void start() { LOG.info("Running process from " + executable.toAbsolutePath()); @@ -70,13 +68,13 @@ class CommandLineProcess { executor.setStreamHandler(new PumpStreamHandler(new LoggingOutputStream(listener::onStdOut), new LoggingOutputStream(listener::onStdErr))); executor.setWatchdog(watchDog); - executor.execute(cmdLine, environment.getEnv(), listener); try { + executor.execute(cmdLine, environment.getEnv(), listener); this.process = executor.getProcess().get(30, TimeUnit.SECONDS); listener.onStart(); } catch (TimeoutException e) { throw new RuntimeException("Failed to obtain process", e); - } catch (ExecutionException | InterruptedException e) { + } catch (ExecutionException | InterruptedException | IOException e) { throw new RuntimeException(e); } } diff --git a/data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java b/data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java index 428f74bb50..dcb1eb1c16 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/management/ManagableProcess.java @@ -21,7 +21,9 @@ import org.graylog.datanode.process.ProcessState; import java.io.IOException; -public interface ManagableProcess { +public interface ManagableProcess { + + void startWithConfig(ConfigurationType configuration); void start() throws IOException; diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchDynamicConfiguration.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchDynamicConfiguration.java new file mode 100644 index 0000000000..fc258ac9c1 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchDynamicConfiguration.java @@ -0,0 +1,20 @@ +/* + * 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.management; + +public class OpensearchDynamicConfiguration { +} diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java index cadc9aa51d..cdf5a0e68b 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java @@ -22,7 +22,7 @@ import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; import java.net.URI; import java.util.List; -public interface OpensearchProcess extends ManagableProcess { +public interface OpensearchProcess extends ManagableProcess { ProcessInfo processInfo(); diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java index 55de789a0f..cd732364aa 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java @@ -19,7 +19,6 @@ package org.graylog.datanode.management; import com.github.oxo42.stateless4j.StateMachine; import org.apache.commons.collections4.queue.CircularFifoQueue; import org.apache.commons.exec.ExecuteException; -import org.graylog.datanode.ProcessProvidingExecutor; import org.graylog.datanode.process.OpensearchConfiguration; import org.graylog.datanode.process.ProcessEvent; import org.graylog.datanode.process.ProcessInfo; @@ -36,13 +35,12 @@ import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; import java.net.URI; import java.nio.file.Path; import java.nio.file.Paths; -import java.util.ArrayList; import java.util.List; import java.util.Locale; +import java.util.Objects; import java.util.Optional; import java.util.Queue; @@ -50,6 +48,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { private static final Logger LOG = LoggerFactory.getLogger(OpensearchProcessImpl.class); private final OpensearchConfiguration configuration; + private OpensearchDynamicConfiguration dynamicConfiguration; private final RestHighLevelClient restClient; private final StateMachine processState; @@ -132,6 +131,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { } public void onEvent(ProcessEvent event) { + LOG.debug("Process event: " + event); this.processState.fire(event); } @@ -153,7 +153,23 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { } @Override - public synchronized void start() throws IOException { + public void startWithConfig(OpensearchDynamicConfiguration configuration) { + if(!Objects.equals(dynamicConfiguration, configuration)) { + this.dynamicConfiguration = configuration; + restart(); + } else { + LOG.info("Dynamic configuration not changed, ignoring"); + } + } + + @Override + public synchronized void restart() { + if(dynamicConfiguration == null) { + throw new IllegalArgumentException("Dynamic runtime configuration required but not supplied!"); + } + + stop(); // this could be an attempt to restart. If there is already one running OS process, just stop it. + final Path executable = configuration.opensearchDir().resolve(Paths.get("bin", "opensearch")); final List arguments = configuration.asMap().entrySet().stream() .map(it -> String.format(Locale.ROOT, "-E%s=%s", it.getKey(), it.getValue())).toList(); @@ -163,7 +179,9 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { @Override public synchronized void stop() { - commandLineProcess.stop(); + if(this.commandLineProcess != null) { + commandLineProcess.stop(); + } } @Override diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java index 48cd71304a..2bc5ac8a57 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java @@ -37,9 +37,9 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro } @Override - protected void startUp() throws Exception { - this.process.start(); - this.processWatchdog.start(); + protected void startUp() { + this.process.startWithConfig(new OpensearchDynamicConfiguration()); + this.processWatchdog.start(); } diff --git a/data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java b/data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java index 768f1d0edf..77f07bbf3d 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java +++ b/data-node/src/main/java/org/graylog/datanode/management/ProcessWatchdog.java @@ -76,7 +76,7 @@ public class ProcessWatchdog implements Runnable { private void restartProcess() { try { LOG.info("Detected terminated process, restarting"); - process.start(); + process.restart(); } catch (IOException e) { LOG.warn("Failed to start the process.", e); } finally { 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 4ece7ddada..07e0f3c73c 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 @@ -44,7 +44,7 @@ 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)) { + if (!process.isInState(ProcessState.TERMINATED) && !process.isInState(ProcessState.WAITING_FOR_CONFIGURATION)) { try { final MainResponse health = process.restClient() .info(RequestOptions.DEFAULT); diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java b/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java index 8a5cba0c65..cc0c0fbf15 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java @@ -20,7 +20,7 @@ public enum ProcessState { /** * Fresh created process, not started yet */ - NEW, + WAITING_FOR_CONFIGURATION, /** * The process is running on the underlying OS and has a process ID. It's not responding to the REST API yet */ 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 index 9ead799c57..c85243a472 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ProcessStateMachine.java @@ -37,10 +37,10 @@ public class ProcessStateMachine { StateMachineConfig config = new StateMachineConfig<>(); // Freshly created process, it hasn't started yet and doesn't have any pid. - config.configure(ProcessState.NEW) + config.configure(ProcessState.WAITING_FOR_CONFIGURATION) .permit(ProcessEvent.PROCESS_STARTED, ProcessState.STARTING) .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED) - .permit(ProcessEvent.HEALTH_CHECK_FAILED, 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 @@ -58,7 +58,8 @@ public class ProcessStateMachine { .onEntry(rebootCounter::resetFailuresCounter) .permitReentry(ProcessEvent.HEALTH_CHECK_OK) .permit(ProcessEvent.HEALTH_CHECK_FAILED, ProcessState.NOT_RESPONDING) - .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED); + .permit(ProcessEvent.PROCESS_TERMINATED, ProcessState.TERMINATED) + .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 @@ -83,6 +84,6 @@ public class ProcessStateMachine { .ignore(ProcessEvent.HEALTH_CHECK_FAILED) .ignore(ProcessEvent.PROCESS_TERMINATED); // final state, all following terminate events are ignored - return new StateMachine<>(ProcessState.NEW, config); + return new StateMachine<>(ProcessState.WAITING_FOR_CONFIGURATION, config); } } 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 71e45470b7..9bc9a627fb 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 @@ -16,6 +16,7 @@ */ package org.graylog.datanode.rest; +import org.graylog.datanode.management.OpensearchDynamicConfiguration; import org.graylog.datanode.management.OpensearchProcess; import org.graylog2.plugin.Version; @@ -24,6 +25,7 @@ import javax.ws.rs.GET; import javax.ws.rs.Path; import javax.ws.rs.Produces; import javax.ws.rs.core.MediaType; +import java.io.IOException; @Path("/") @Produces(MediaType.APPLICATION_JSON) @@ -45,4 +47,11 @@ public class StatusController { new StatusResponse(openSearch.opensearchVersion(), openSearch.processInfo()) ); } + + @GET + @Path("/start") + public DataNodeStatus start() throws IOException { + openSearch.startWithConfig(new OpensearchDynamicConfiguration()); + return status(); + } } diff --git a/data-node/src/test/java/org/graylog/datanode/management/ProcessWatchdogTest.java b/data-node/src/test/java/org/graylog/datanode/management/ProcessWatchdogTest.java index 935db238e5..a2e4bd8074 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/ProcessWatchdogTest.java +++ b/data-node/src/test/java/org/graylog/datanode/management/ProcessWatchdogTest.java @@ -36,7 +36,7 @@ class ProcessWatchdogTest { void testLifecycle() throws IOException, ExecutionException, RetryException { final TestableProcess process = new TestableProcess(); final ProcessWatchdog watchdog = new ProcessWatchdog(process, 100); - process.start(); + process.restart(); watchdog.start(); // both process and watchdog are running now. Let's stop the process and see if the watchdog will restart it diff --git a/data-node/src/test/java/org/graylog/datanode/management/TestableProcess.java b/data-node/src/test/java/org/graylog/datanode/management/TestableProcess.java index 97ac9a6ac9..2da9a81c74 100644 --- a/data-node/src/test/java/org/graylog/datanode/management/TestableProcess.java +++ b/data-node/src/test/java/org/graylog/datanode/management/TestableProcess.java @@ -21,18 +21,23 @@ import org.graylog.datanode.process.ProcessState; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class TestableProcess implements ManagableProcess { +public class TestableProcess implements ManagableProcess { private static final Logger LOG = LoggerFactory.getLogger(TestableProcess.class); private volatile ProcessState state; public TestableProcess() { - this.state = ProcessState.NEW; + this.state = ProcessState.WAITING_FOR_CONFIGURATION; } @Override - public void start() { + public void startWithConfig(String ignored) { + restart(); + } + + @Override + public void restart() { LOG.debug("Starting process"); this.state = ProcessState.STARTING; } 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 index e46eba1d9a..c2230e4b45 100644 --- a/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java +++ b/data-node/src/test/java/org/graylog/datanode/process/ProcessStateMachineTest.java @@ -25,7 +25,7 @@ class ProcessStateMachineTest { @Test void testOptimalScenario() { final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.NEW); + Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); machine.fire(ProcessEvent.PROCESS_STARTED); Assertions.assertEquals(ProcessState.STARTING, machine.getState()); @@ -40,7 +40,7 @@ class ProcessStateMachineTest { @Test void testRestFailing() { final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.NEW); + Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); machine.fire(ProcessEvent.PROCESS_STARTED); Assertions.assertEquals(ProcessState.STARTING, machine.getState()); @@ -65,7 +65,7 @@ class ProcessStateMachineTest { @Test void testStartupFailure() { final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.NEW); + Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); machine.fire(ProcessEvent.PROCESS_STARTED); Assertions.assertEquals(ProcessState.STARTING, machine.getState()); @@ -84,7 +84,7 @@ class ProcessStateMachineTest { @Test void testStartupFailureResolved() { final StateMachine machine = ProcessStateMachine.createNew(); - Assertions.assertEquals(machine.getState(), ProcessState.NEW); + Assertions.assertEquals(machine.getState(), ProcessState.WAITING_FOR_CONFIGURATION); machine.fire(ProcessEvent.PROCESS_STARTED); Assertions.assertEquals(ProcessState.STARTING, machine.getState());