mirror of
https://github.com/Graylog2/graylog2-server.git
synced 2026-03-13 09:32:21 +08:00
Update opensearch to 2.9.0 in matrix integration tests (#16294)
* Update opensearch to 2.9.0 in matrix integration tests * Update datanode to 2.9.0 opensearch * workaround for opensearch number_of_replicas in single-node mode * trigger the fix of replica numbers in opensearch only for 2.9.0 and higher releases * better hostname handling for DatanodeClusterIT * log all request/responses for DatanodeClusterIT * add response logging to DatanodeClusterIT * print also logs from node B in DatanodeClusterIT in case of error * fixed DatanodeClusterIT, cached datanode images, synchronized maven packaging * better handling of DatanodeClusterIT waiting for cluster status * code cleanup * log last recorded exception in case of failed waiting for datanode opensearch * log state machine transitions * increase waiting timeout for DatanodeClusterIT * print shard info in case of datanode cluster failure * safer shard info debug in DatanodeOpensearchWait * disable scluster.routing.allocation.disk.threshold for datanode integration tests (causes problems in github actions) --------- Co-authored-by: Dennis Oelkers <dennis@graylog.com>
This commit is contained in:
@@ -32,7 +32,7 @@ public interface ManagableProcess<T> {
|
||||
|
||||
void onEvent(ProcessEvent event);
|
||||
|
||||
void setStateMachineTracer(StateMachineTracer stateMachineTracer);
|
||||
void addStateMachineTracer(StateMachineTracer stateMachineTracer);
|
||||
|
||||
boolean isInState(ProcessState state);
|
||||
}
|
||||
|
||||
@@ -31,8 +31,10 @@ import java.nio.charset.Charset;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
|
||||
public class OpensearchCommandLineProcess implements Closeable {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(OpensearchCommandLineProcess.class);
|
||||
@@ -76,12 +78,25 @@ public class OpensearchCommandLineProcess implements Closeable {
|
||||
public OpensearchCommandLineProcess(OpensearchConfiguration config, ProcessListener listener) {
|
||||
fixJdkOnMac(config);
|
||||
final Path executable = config.opensearchDir().resolve(Paths.get("bin", "opensearch"));
|
||||
final List<String> arguments = config.asMap().entrySet().stream()
|
||||
final List<String> arguments = getOpensearchConfigurationArguments(config).entrySet().stream()
|
||||
.map(it -> String.format(Locale.ROOT, "-E%s=%s", it.getKey(), it.getValue())).toList();
|
||||
resultHandler = new CommandLineProcessListener(listener);
|
||||
commandLineProcess = new CommandLineProcess(executable, arguments, resultHandler, config.getEnv());
|
||||
}
|
||||
|
||||
private static Map<String, String> getOpensearchConfigurationArguments(OpensearchConfiguration config) {
|
||||
Map<String, String> allArguments = new LinkedHashMap<>(config.asMap());
|
||||
|
||||
// now copy all the environment values to the configuration arguments. Opensearch won't do it for us,
|
||||
// because we are using tar distriburion and opensearch does this only for docker dist. See opensearch-env script
|
||||
// additionally, the env variables have to be prefixed with opensearch. (e.g. "opensearch.cluster.routing.allocation.disk.threshold_enabled")
|
||||
config.getEnv().getEnv().entrySet().stream()
|
||||
.filter(entry -> entry.getKey().matches("^opensearch\\.[a-z0-9_]+(?:\\.[a-z0-9_]+)+"))
|
||||
.peek(entry -> LOG.info("Detected pass-through opensearch property {}:{}", entry.getKey().substring("opensearch.".length()), entry.getValue()))
|
||||
.forEach(entry -> allArguments.put(entry.getKey().substring("opensearch.".length()), entry.getValue()));
|
||||
return allArguments;
|
||||
}
|
||||
|
||||
public void start() {
|
||||
commandLineProcess.start();
|
||||
}
|
||||
|
||||
@@ -28,7 +28,6 @@ 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.security.certutil.CaService;
|
||||
import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient;
|
||||
import org.graylog2.security.CustomCAX509TrustManager;
|
||||
import org.slf4j.Logger;
|
||||
@@ -43,6 +42,7 @@ import java.util.Queue;
|
||||
class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(OpensearchProcessImpl.class);
|
||||
private final StateMachineTracerAggregator tracerAggregator;
|
||||
|
||||
@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
|
||||
private Optional<OpensearchConfiguration> configuration = Optional.empty();
|
||||
@@ -63,6 +63,8 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
|
||||
OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager) {
|
||||
this.datanodeConfiguration = datanodeConfiguration;
|
||||
this.processState = ProcessStateMachine.createNew();
|
||||
tracerAggregator = new StateMachineTracerAggregator();
|
||||
this.processState.setTrace(tracerAggregator);
|
||||
this.stdout = new CircularFifoQueue<>(logsCacheSize);
|
||||
this.stderr = new CircularFifoQueue<>(logsCacheSize);
|
||||
this.trustManager = trustManager;
|
||||
@@ -108,8 +110,8 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setStateMachineTracer(StateMachineTracer stateMachineTracer) {
|
||||
this.processState.setTrace(stateMachineTracer);
|
||||
public void addStateMachineTracer(StateMachineTracer stateMachineTracer) {
|
||||
this.tracerAggregator.addTracer(stateMachineTracer);
|
||||
}
|
||||
|
||||
public void setLeaderNode(boolean isLeaderNode) {
|
||||
|
||||
@@ -56,7 +56,8 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
|
||||
private OpensearchProcess createOpensearchProcess(DatanodeConfiguration datanodeConfiguration) {
|
||||
final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize(), trustManager);
|
||||
final ProcessWatchdog watchdog = new ProcessWatchdog(process, WATCHDOG_RESTART_ATTEMPTS);
|
||||
process.setStateMachineTracer(watchdog);
|
||||
process.addStateMachineTracer(watchdog);
|
||||
process.addStateMachineTracer(new StateMachineTransitionLogger());
|
||||
return process;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
/*
|
||||
* 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.management;
|
||||
|
||||
import org.graylog.datanode.process.ProcessEvent;
|
||||
import org.graylog.datanode.process.ProcessState;
|
||||
import org.graylog.datanode.process.StateMachineTracer;
|
||||
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
|
||||
public class StateMachineTracerAggregator implements StateMachineTracer {
|
||||
|
||||
private final List<StateMachineTracer> delegates = new LinkedList<>();
|
||||
|
||||
public void addTracer(StateMachineTracer tracer) {
|
||||
delegates.add(tracer);
|
||||
}
|
||||
|
||||
public void removeTracer(StateMachineTracer tracer) {
|
||||
delegates.remove(tracer);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void trigger(ProcessEvent processEvent) {
|
||||
delegates.forEach(d -> d.trigger(processEvent));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void transition(ProcessEvent processEvent, ProcessState s1, ProcessState s2) {
|
||||
delegates.forEach(d -> d.transition(processEvent, s1, s2));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
/*
|
||||
* 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.management;
|
||||
|
||||
import org.graylog.datanode.process.ProcessEvent;
|
||||
import org.graylog.datanode.process.ProcessState;
|
||||
import org.graylog.datanode.process.StateMachineTracer;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
public class StateMachineTransitionLogger implements StateMachineTracer {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(StateMachineTransitionLogger.class);
|
||||
|
||||
@Override
|
||||
public void trigger(ProcessEvent trigger) {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void transition(ProcessEvent trigger, ProcessState source, ProcessState destination) {
|
||||
LOG.debug("Triggered {}, source state: {}, destination: {}", trigger, source, destination);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user