mirror of
https://github.com/Graylog2/graylog2-server.git
synced 2026-03-13 09:32:21 +08:00
Async cluster restart during remote reindex migration (#19080)
* Async cluster restart during remote reindex migration, configuration cleanup * Safer and faster way to restart and wait for datanode cluster during remote reindex --------- Co-authored-by: Matthias Oesterheld <33032967+moesterheld@users.noreply.github.com>
This commit is contained in:
@@ -280,6 +280,11 @@ public class Configuration extends BaseConfiguration {
|
||||
return opensearchAuditLog;
|
||||
}
|
||||
|
||||
/**
|
||||
* The insecure flag causes problems on many places. We should replace it with autosecurity option, that would
|
||||
* configure all the CA and certs automatically.
|
||||
*/
|
||||
@Deprecated
|
||||
public boolean isInsecureStartup() {
|
||||
return insecureStartup;
|
||||
}
|
||||
|
||||
@@ -17,6 +17,10 @@
|
||||
package org.graylog.datanode.configuration;
|
||||
|
||||
import com.google.common.collect.ImmutableMap;
|
||||
import jakarta.inject.Inject;
|
||||
import jakarta.inject.Named;
|
||||
import jakarta.inject.Provider;
|
||||
import jakarta.inject.Singleton;
|
||||
import org.graylog.datanode.Configuration;
|
||||
import org.graylog.datanode.configuration.variants.InSecureConfiguration;
|
||||
import org.graylog.datanode.configuration.variants.MongoCertSecureConfiguration;
|
||||
@@ -25,22 +29,17 @@ import org.graylog.datanode.configuration.variants.SecurityConfigurationVariant;
|
||||
import org.graylog.datanode.configuration.variants.UploadedCertFilesSecureConfiguration;
|
||||
import org.graylog.datanode.process.OpensearchConfiguration;
|
||||
import org.graylog.security.certutil.ca.exceptions.KeyStoreStorageException;
|
||||
import org.graylog2.bootstrap.preflight.PreflightConfigResult;
|
||||
import org.graylog2.bootstrap.preflight.PreflightConfigService;
|
||||
import org.graylog2.cluster.Node;
|
||||
import org.graylog2.cluster.nodes.DataNodeDto;
|
||||
import org.graylog2.cluster.nodes.NodeService;
|
||||
|
||||
import jakarta.inject.Inject;
|
||||
import jakarta.inject.Named;
|
||||
import jakarta.inject.Provider;
|
||||
import jakarta.inject.Singleton;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.GeneralSecurityException;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
@Singleton
|
||||
@@ -52,9 +51,13 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
|
||||
private final DatanodeConfiguration datanodeConfiguration;
|
||||
private final byte[] signingKey;
|
||||
private final NodeService<DataNodeDto> nodeService;
|
||||
private final PreflightConfigService preflightConfigService;
|
||||
private final S3RepositoryConfiguration s3RepositoryConfiguration;
|
||||
|
||||
/**
|
||||
* This configuration won't survive datanode restart. But it can be repeatedly provided to the managed opensearch
|
||||
*/
|
||||
private final Map<String, Object> transientConfiguration = new ConcurrentHashMap<>();
|
||||
|
||||
@Inject
|
||||
public OpensearchConfigurationProvider(final Configuration localConfiguration,
|
||||
final DatanodeConfiguration datanodeConfiguration,
|
||||
@@ -62,7 +65,6 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
|
||||
final MongoCertSecureConfiguration mongoCertSecureConfiguration,
|
||||
final InSecureConfiguration inSecureConfiguration,
|
||||
final NodeService<DataNodeDto> nodeService,
|
||||
final PreflightConfigService preflightConfigService,
|
||||
final @Named("password_secret") String passwordSecret,
|
||||
final S3RepositoryConfiguration s3RepositoryConfiguration) {
|
||||
this.localConfiguration = localConfiguration;
|
||||
@@ -72,7 +74,6 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
|
||||
this.inSecureConfiguration = inSecureConfiguration;
|
||||
this.signingKey = passwordSecret.getBytes(StandardCharsets.UTF_8);
|
||||
this.nodeService = nodeService;
|
||||
this.preflightConfigService = preflightConfigService;
|
||||
this.s3RepositoryConfiguration = s3RepositoryConfiguration;
|
||||
}
|
||||
|
||||
@@ -159,7 +160,12 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
|
||||
// enable admin access via the REST API
|
||||
config.put("plugins.security.restapi.admin.enabled", "true");
|
||||
|
||||
config.putAll(transientConfiguration);
|
||||
|
||||
return config.build();
|
||||
}
|
||||
|
||||
public void setTransientConfiguration(String key, Object value) {
|
||||
this.transientConfiguration.put(key, value);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,7 +26,7 @@ import jakarta.inject.Singleton;
|
||||
import org.graylog.datanode.Configuration;
|
||||
import org.graylog.datanode.bootstrap.preflight.DatanodeDirectoriesLockfileCheck;
|
||||
import org.graylog.datanode.configuration.DatanodeConfiguration;
|
||||
import org.graylog.datanode.configuration.variants.InSecureConfiguration;
|
||||
import org.graylog.datanode.configuration.OpensearchConfigurationProvider;
|
||||
import org.graylog.datanode.metrics.ConfigureMetricsIndexSettings;
|
||||
import org.graylog.datanode.process.OpensearchConfiguration;
|
||||
import org.graylog.datanode.process.ProcessStateMachine;
|
||||
@@ -41,15 +41,11 @@ import org.graylog2.datanode.DataNodeLifecycleEvent;
|
||||
import org.graylog2.datanode.RemoteReindexAllowlistEvent;
|
||||
import org.graylog2.events.ClusterEventBus;
|
||||
import org.graylog2.indexer.fieldtypes.IndexFieldTypesService;
|
||||
import org.graylog2.plugin.cluster.ClusterConfigService;
|
||||
import org.graylog2.plugin.system.NodeId;
|
||||
import org.graylog2.security.CustomCAX509TrustManager;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
@Singleton
|
||||
public class OpensearchProcessService extends AbstractIdleService implements Provider<OpensearchProcess> {
|
||||
|
||||
@@ -57,21 +53,20 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
|
||||
|
||||
private static final int WATCHDOG_RESTART_ATTEMPTS = 3;
|
||||
private final OpensearchProcess process;
|
||||
private final Provider<OpensearchConfiguration> configurationProvider;
|
||||
private final OpensearchConfigurationProvider configurationProvider;
|
||||
private final EventBus eventBus;
|
||||
private final NodeId nodeId;
|
||||
private final DataNodeProvisioningService dataNodeProvisioningService;
|
||||
private final IndexFieldTypesService indexFieldTypesService;
|
||||
private final ClusterEventBus clusterEventBus;
|
||||
private final DatanodeDirectoriesLockfileCheck lockfileCheck;
|
||||
private final ClusterConfigService clusterConfigService;
|
||||
private final PreflightConfigService preflightConfigService;
|
||||
private final Configuration configuration;
|
||||
|
||||
|
||||
@Inject
|
||||
public OpensearchProcessService(final DatanodeConfiguration datanodeConfiguration,
|
||||
final Provider<OpensearchConfiguration> configurationProvider,
|
||||
final OpensearchConfigurationProvider configurationProvider,
|
||||
final EventBus eventBus,
|
||||
final CustomCAX509TrustManager trustManager,
|
||||
final NodeService<DataNodeDto> nodeService,
|
||||
@@ -83,7 +78,6 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
|
||||
final ProcessStateMachine processStateMachine,
|
||||
final ClusterEventBus clusterEventBus,
|
||||
final DatanodeDirectoriesLockfileCheck lockfileCheck,
|
||||
final ClusterConfigService clusterConfigService,
|
||||
final PreflightConfigService preflightConfigService) {
|
||||
this.configurationProvider = configurationProvider;
|
||||
this.configuration = configuration;
|
||||
@@ -93,7 +87,6 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
|
||||
this.indexFieldTypesService = indexFieldTypesService;
|
||||
this.clusterEventBus = clusterEventBus;
|
||||
this.lockfileCheck = lockfileCheck;
|
||||
this.clusterConfigService = clusterConfigService;
|
||||
this.preflightConfigService = preflightConfigService;
|
||||
this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper, processStateMachine);
|
||||
eventBus.register(this);
|
||||
@@ -117,7 +110,8 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
|
||||
switch (event.action()) {
|
||||
case ADD -> {
|
||||
this.process.stop();
|
||||
configure(Map.of("reindex.remote.allowlist", event.allowlist())); // , "action.auto_create_index", "false"));
|
||||
this.configurationProvider.setTransientConfiguration("reindex.remote.allowlist", event.allowlist());
|
||||
configure(); // , "action.auto_create_index", "false"));
|
||||
this.process.start();
|
||||
}
|
||||
case REMOVE -> {
|
||||
@@ -178,33 +172,8 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
|
||||
}
|
||||
}
|
||||
|
||||
protected void configure() {
|
||||
this.configure(Map.of());
|
||||
}
|
||||
|
||||
private void configure(Map<String, Object> additionalConfig) {
|
||||
final OpensearchConfiguration original = configurationProvider.get();
|
||||
|
||||
final var finalAdditionalConfig = new HashMap<String, Object>();
|
||||
finalAdditionalConfig.putAll(original.additionalConfiguration());
|
||||
finalAdditionalConfig.putAll(additionalConfig);
|
||||
|
||||
final var config = new OpensearchConfiguration(
|
||||
original.opensearchDistribution(),
|
||||
original.datanodeDirectories(),
|
||||
original.bindAddress(),
|
||||
original.hostname(),
|
||||
original.httpPort(),
|
||||
original.transportPort(),
|
||||
original.clusterName(),
|
||||
original.nodeName(),
|
||||
original.nodeRoles(),
|
||||
original.discoverySeedHosts(),
|
||||
original.opensearchSecurityConfiguration(),
|
||||
original.s3RepositoryConfiguration(),
|
||||
original.nodeSearchCacheSize(),
|
||||
finalAdditionalConfig);
|
||||
|
||||
private void configure() {
|
||||
final OpensearchConfiguration config = configurationProvider.get();
|
||||
if (config.securityConfigured()) {
|
||||
this.process.configure(config);
|
||||
} else {
|
||||
|
||||
Reference in New Issue
Block a user