diff --git a/data-node/src/main/java/org/graylog/datanode/Configuration.java b/data-node/src/main/java/org/graylog/datanode/Configuration.java index 7c9d6fd206..6f73a4ffc9 100644 --- a/data-node/src/main/java/org/graylog/datanode/Configuration.java +++ b/data-node/src/main/java/org/graylog/datanode/Configuration.java @@ -87,7 +87,7 @@ public class Configuration extends BaseConfiguration { private String configLocation; @Parameter(value = "process_logs_buffer_size") - private Integer logs = 500; + private Integer opensearchProcessLogsBufferSize = 500; @Parameter(value = "node_name") @@ -172,7 +172,7 @@ public class Configuration extends BaseConfiguration { } public Integer getProcessLogsBufferSize() { - return logs; + return opensearchProcessLogsBufferSize; } @Parameter(value = "rest_api_username") diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/OpensearchDistributionBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/DatanodeConfigurationBindings.java similarity index 72% rename from data-node/src/main/java/org/graylog/datanode/bindings/OpensearchDistributionBindings.java rename to data-node/src/main/java/org/graylog/datanode/bindings/DatanodeConfigurationBindings.java index f35856baf4..474abc9946 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/OpensearchDistributionBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/DatanodeConfigurationBindings.java @@ -17,12 +17,12 @@ package org.graylog.datanode.bindings; import com.google.inject.AbstractModule; -import org.graylog.datanode.OpensearchDistribution; -import org.graylog.datanode.OpensearchDistributionProvider; +import org.graylog.datanode.configuration.DatanodeConfiguration; +import org.graylog.datanode.configuration.DatanodeConfigurationProvider; -public class OpensearchDistributionBindings extends AbstractModule { +public class DatanodeConfigurationBindings extends AbstractModule { @Override protected void configure() { - bind(OpensearchDistribution.class).toProvider(OpensearchDistributionProvider.class); + bind(DatanodeConfiguration.class).toProvider(DatanodeConfigurationProvider.class); } } diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java index 8b412a1936..6dc5791901 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/GenericInitializerBindings.java @@ -19,12 +19,14 @@ package org.graylog.datanode.bindings; import com.google.common.util.concurrent.Service; import com.google.inject.AbstractModule; import com.google.inject.multibindings.Multibinder; -import org.graylog.datanode.configuration.DataNodeConfigProvider; -import org.graylog.datanode.configuration.DataNodeConfig; +import org.graylog.datanode.configuration.DataNodeClusterWideConfigProvider; +import org.graylog.datanode.configuration.DataNodeClusterWideConfig; +import org.graylog.datanode.configuration.DatanodeConfiguration; +import org.graylog.datanode.configuration.DatanodeConfigurationProvider; import org.graylog.datanode.initializers.JerseyService; import org.graylog.datanode.management.OpensearchProcessService; import org.graylog.datanode.initializers.PeriodicalsService; -import org.graylog.datanode.configuration.ConfigurationProvider; +import org.graylog.datanode.configuration.OpensearchConfigurationProvider; import org.graylog.datanode.process.OpensearchConfiguration; import org.graylog.datanode.management.OpensearchProcess; import org.graylog.datanode.shutdown.GracefulShutdownService; @@ -33,10 +35,8 @@ public class GenericInitializerBindings extends AbstractModule { @Override protected void configure() { // bind(ProcessingStatusRecorder.class).to(MongoDBProcessingStatusRecorderService.class).asEagerSingleton(); - - - bind(OpensearchConfiguration.class).toProvider(ConfigurationProvider.class); - bind(DataNodeConfig.class).toProvider(DataNodeConfigProvider.class); + bind(OpensearchConfiguration.class).toProvider(OpensearchConfigurationProvider.class); + bind(DataNodeClusterWideConfig.class).toProvider(DataNodeClusterWideConfigProvider.class); Multibinder serviceBinder = Multibinder.newSetBinder(binder(), Service.class); diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java index 558cd0d9fe..3c15455fa6 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java @@ -16,6 +16,7 @@ */ package org.graylog.datanode.bindings; +import com.google.common.eventbus.EventBus; import com.google.inject.multibindings.Multibinder; import com.google.inject.multibindings.OptionalBinder; import org.graylog.datanode.Configuration; @@ -34,6 +35,7 @@ import org.graylog2.plugin.cluster.RandomUUIDClusterIdFactory; import org.graylog2.plugin.inject.Graylog2Module; import org.graylog2.plugin.system.FilePersistedNodeIdProvider; import org.graylog2.plugin.system.NodeId; +import org.graylog2.shared.bindings.providers.EventBusProvider; import org.graylog2.shared.system.activities.ActivityWriter; import javax.ws.rs.container.DynamicFeature; @@ -65,6 +67,7 @@ public class ServerBindings extends Graylog2Module { private void bindProviders() { bind(ClusterEventBus.class).toProvider(ClusterEventBusProvider.class).asEagerSingleton(); + bind(EventBus.class).toProvider(EventBusProvider.class).asEagerSingleton(); bind(NodeId.class).toProvider(FilePersistedNodeIdProvider.class).asEagerSingleton(); bind(InputConfigurationBeanDeserializerModifier.class).toInstance(InputConfigurationBeanDeserializerModifier.withoutConfig()); } diff --git a/data-node/src/main/java/org/graylog/datanode/bootstrap/ServerBootstrap.java b/data-node/src/main/java/org/graylog/datanode/bootstrap/ServerBootstrap.java index 2673c06875..34b60c772c 100644 --- a/data-node/src/main/java/org/graylog/datanode/bootstrap/ServerBootstrap.java +++ b/data-node/src/main/java/org/graylog/datanode/bootstrap/ServerBootstrap.java @@ -27,9 +27,9 @@ import com.google.inject.Module; import com.google.inject.ProvisionException; import org.graylog.datanode.Configuration; import org.graylog.datanode.bindings.ConfigurationModule; +import org.graylog.datanode.bindings.DatanodeConfigurationBindings; import org.graylog.datanode.bindings.GenericBindings; import org.graylog.datanode.bindings.GenericInitializerBindings; -import org.graylog.datanode.bindings.OpensearchDistributionBindings; import org.graylog.datanode.bindings.PreflightChecksBindings; import org.graylog.datanode.bindings.SchedulerBindings; import org.graylog2.bootstrap.preflight.MongoDBPreflightCheck; @@ -157,7 +157,7 @@ public abstract class ServerBootstrap extends CmdLineTool { new IsDevelopmentBindings(), new NamedConfigParametersModule(jadConfig.getConfigurationBeans()), new ConfigurationModule(configuration), - new OpensearchDistributionBindings() + new DatanodeConfigurationBindings() ); } @@ -168,7 +168,7 @@ public abstract class ServerBootstrap extends CmdLineTool { new NamedConfigParametersModule(jadConfig.getConfigurationBeans()), new ConfigurationModule(configuration), new PreflightChecksBindings(), - new OpensearchDistributionBindings(), + new DatanodeConfigurationBindings(), new Module() { @Override public void configure(Binder binder) { @@ -281,7 +281,7 @@ public abstract class ServerBootstrap extends CmdLineTool { // result.add(new SharedPeriodicalBindings()); result.add(new SchedulerBindings()); result.add(new GenericInitializerBindings()); - result.add(new OpensearchDistributionBindings()); + result.add(new DatanodeConfigurationBindings()); // result.add(new SystemStatsModule(configuration.isDisableNativeSystemStatsCollector())); return result; diff --git a/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodePreflightGeneratePeriodical.java b/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodePreflightGeneratePeriodical.java index 4f5b9b665b..15e906109e 100644 --- a/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodePreflightGeneratePeriodical.java +++ b/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodePreflightGeneratePeriodical.java @@ -30,8 +30,8 @@ import org.graylog.security.certutil.keystore.storage.location.KeystoreMongoColl import org.graylog.security.certutil.keystore.storage.location.KeystoreMongoLocation; import org.graylog.security.certutil.privatekey.PrivateKeyEncryptedFileStorage; import org.graylog2.cluster.NodeNotFoundException; -import org.graylog2.cluster.NodePreflightConfig; -import org.graylog2.cluster.NodePreflightConfigService; +import org.graylog2.cluster.preflight.NodePreflightConfig; +import org.graylog2.cluster.preflight.NodePreflightConfigService; import org.graylog2.cluster.NodeService; import org.graylog2.plugin.periodical.Periodical; import org.graylog2.plugin.system.NodeId; diff --git a/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/OpensearchBinPreflightCheck.java b/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/OpensearchBinPreflightCheck.java index dc7afdec2d..68aa047009 100644 --- a/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/OpensearchBinPreflightCheck.java +++ b/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/OpensearchBinPreflightCheck.java @@ -16,7 +16,7 @@ */ package org.graylog.datanode.bootstrap.preflight; -import org.graylog.datanode.OpensearchDistribution; +import org.graylog.datanode.configuration.DatanodeConfiguration; import org.graylog2.bootstrap.preflight.PreflightCheck; import org.graylog2.bootstrap.preflight.PreflightCheckException; import org.slf4j.Logger; @@ -37,8 +37,8 @@ public class OpensearchBinPreflightCheck implements PreflightCheck { private static final Logger LOG = LoggerFactory.getLogger(OpensearchBinPreflightCheck.class); @Inject - public OpensearchBinPreflightCheck(OpensearchDistribution opensearchDistribution) { - this(opensearchDistribution.directory()); + public OpensearchBinPreflightCheck(DatanodeConfiguration datanodeConfiguration) { + this(datanodeConfiguration.opensearchDistribution().directory()); } public OpensearchBinPreflightCheck(Path opensearchBaseDirectory) { diff --git a/data-node/src/main/java/org/graylog/datanode/commands/Server.java b/data-node/src/main/java/org/graylog/datanode/commands/Server.java index 3cc03e6f01..336069ece4 100644 --- a/data-node/src/main/java/org/graylog/datanode/commands/Server.java +++ b/data-node/src/main/java/org/graylog/datanode/commands/Server.java @@ -34,6 +34,7 @@ import org.graylog.datanode.rest.RestBindings; import org.graylog.datanode.shutdown.GracefulShutdown; import org.graylog2.bindings.MongoDBModule; import org.graylog2.bindings.PasswordAlgorithmBindings; +import org.graylog2.cluster.preflight.PreflightConfigBindings; import org.graylog2.configuration.MongoDbConfiguration; import org.graylog2.configuration.TLSProtocolsConfiguration; import org.graylog2.featureflag.FeatureFlags; @@ -84,6 +85,7 @@ public class Server extends ServerBootstrap { new MongoDBModule(), new ServerBindings(configuration, isMigrationCommand()), new RestBindings(), + new PreflightConfigBindings(), new PeriodicalBindings(), // new InitializerBindings(), new ObjectMapperModule(chainingClassLoader), diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/ConfigurationProvider.java b/data-node/src/main/java/org/graylog/datanode/configuration/ConfigurationProvider.java deleted file mode 100644 index a74df26306..0000000000 --- a/data-node/src/main/java/org/graylog/datanode/configuration/ConfigurationProvider.java +++ /dev/null @@ -1,88 +0,0 @@ -/* - * Copyright (C) 2020 Graylog, Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the Server Side Public License, version 1, - * as published by MongoDB, Inc. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * Server Side Public License for more details. - * - * You should have received a copy of the Server Side Public License - * along with this program. If not, see - * . - */ -package org.graylog.datanode.configuration; - -import org.graylog.datanode.Configuration; -import org.graylog.datanode.OpensearchDistribution; -import org.graylog.datanode.configuration.variants.InSecureConfiguration; -import org.graylog.datanode.configuration.variants.MongoCertSecureConfiguration; -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 javax.inject.Inject; -import javax.inject.Provider; -import javax.inject.Singleton; -import java.io.IOException; -import java.nio.file.Path; -import java.security.GeneralSecurityException; -import java.util.Collections; -import java.util.List; -import java.util.Map; - -@Singleton -public class ConfigurationProvider implements Provider { - - private final OpensearchConfiguration configuration; - - @Inject - public ConfigurationProvider(Configuration localConfiguration, - UploadedCertFilesSecureConfiguration uploadedCertFilesSecureConfiguration, - MongoCertSecureConfiguration mongoCertSecureConfiguration, - InSecureConfiguration inSecureConfiguration, - DataNodeConfig sharedConfiguration, - OpensearchDistribution opensearchDistribution - ) throws IOException, GeneralSecurityException, KeyStoreStorageException { - final var cfg = sharedConfiguration.test(); - final String opensearchConfigLocation = localConfiguration.getOpensearchConfigLocation(); - - //TODO: at some point bind the whole list, for now there is too much experiments with order and prerequisites - List securityConfigurationTypes = List.of( - uploadedCertFilesSecureConfiguration, - mongoCertSecureConfiguration, - inSecureConfiguration //TODO: in final version, this configuration is tried first, not last - ); - - SecurityConfigurationVariant chosenSecurityConfigurationVariant = securityConfigurationTypes.stream() - .filter(s -> s.checkPrerequisites(localConfiguration)) - .findFirst() - .orElseThrow(() -> new RuntimeException("No valid option to start up OpenSearch")); - - final Map config = chosenSecurityConfigurationVariant.configure(localConfiguration); - configuration = new OpensearchConfiguration( - opensearchDistribution.version(), - opensearchDistribution.directory(), - Path.of(opensearchConfigLocation), - localConfiguration.getOpensearchHttpPort(), - localConfiguration.getOpensearchTransportPort(), - localConfiguration.getRestApiUsername(), - localConfiguration.getRestApiPassword(), - "datanode-cluster", - localConfiguration.getDatanodeNodeName(), - Collections.emptyList(), - localConfiguration.getOpensearchDiscoverySeedHosts(), - config - ); - } - - @Override - public OpensearchConfiguration get() { - return configuration; - } - -} diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeConfig.java b/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeClusterWideConfig.java similarity index 92% rename from data-node/src/main/java/org/graylog/datanode/configuration/DataNodeConfig.java rename to data-node/src/main/java/org/graylog/datanode/configuration/DataNodeClusterWideConfig.java index f4cacbb5ef..d05df4c31a 100644 --- a/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeConfig.java +++ b/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeClusterWideConfig.java @@ -16,5 +16,5 @@ */ package org.graylog.datanode.configuration; -public record DataNodeConfig(String test) { +public record DataNodeClusterWideConfig(String test) { } diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeConfigProvider.java b/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeClusterWideConfigProvider.java similarity index 75% rename from data-node/src/main/java/org/graylog/datanode/configuration/DataNodeConfigProvider.java rename to data-node/src/main/java/org/graylog/datanode/configuration/DataNodeClusterWideConfigProvider.java index 4a5c1234aa..b0bea88a07 100644 --- a/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeConfigProvider.java +++ b/data-node/src/main/java/org/graylog/datanode/configuration/DataNodeClusterWideConfigProvider.java @@ -25,29 +25,29 @@ import javax.inject.Provider; import javax.validation.constraints.NotNull; import java.util.Optional; -public class DataNodeConfigProvider implements Provider { - private static final Logger LOG = LoggerFactory.getLogger(DataNodeConfigProvider.class); +public class DataNodeClusterWideConfigProvider implements Provider { + private static final Logger LOG = LoggerFactory.getLogger(DataNodeClusterWideConfigProvider.class); private final ClusterConfigService clusterConfigService; @Inject - public DataNodeConfigProvider(ClusterConfigService clusterConfigService) { + public DataNodeClusterWideConfigProvider(ClusterConfigService clusterConfigService) { this.clusterConfigService = clusterConfigService; } @Override - public DataNodeConfig get() { + public DataNodeClusterWideConfig get() { return loadFromDatabase().orElse(getDefaultConfig()); } - public DataNodeConfig getDefaultConfig() { - return new DataNodeConfig(null); + public DataNodeClusterWideConfig getDefaultConfig() { + return new DataNodeClusterWideConfig(null); } @NotNull - public Optional loadFromDatabase() { + public Optional loadFromDatabase() { try { - return Optional.ofNullable(clusterConfigService.get(DataNodeConfig.class)); + return Optional.ofNullable(clusterConfigService.get(DataNodeClusterWideConfig.class)); } catch (Exception e) { LOG.error("Failed to fetch datanode configuration from database", e); return Optional.empty(); diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/DatanodeConfiguration.java b/data-node/src/main/java/org/graylog/datanode/configuration/DatanodeConfiguration.java new file mode 100644 index 0000000000..b570005be6 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/configuration/DatanodeConfiguration.java @@ -0,0 +1,30 @@ +/* + * 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.configuration; + +import org.graylog.datanode.OpensearchDistribution; + +/** + * DatanodeConfiguration holds the static configuration as parsed during the datanode startup, either from the + * config file or from the ENV properties. + */ +public record DatanodeConfiguration( + OpensearchDistribution opensearchDistribution, + String nodeName, + int processLogsBufferSize +) { +} diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/DatanodeConfigurationProvider.java b/data-node/src/main/java/org/graylog/datanode/configuration/DatanodeConfigurationProvider.java new file mode 100644 index 0000000000..7964735919 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/configuration/DatanodeConfigurationProvider.java @@ -0,0 +1,51 @@ +/* + * 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.configuration; + +import org.graylog.datanode.Configuration; +import org.graylog.datanode.OpensearchDistribution; + +import javax.inject.Inject; +import javax.inject.Provider; +import javax.inject.Singleton; +import java.io.IOException; +import java.nio.file.Path; + +@Singleton +public class DatanodeConfigurationProvider implements Provider { + + private final DatanodeConfiguration datanodeConfiguration; + + @Inject + public DatanodeConfigurationProvider(final Configuration localConfiguration) throws IOException { + final OpensearchDistribution opensearchDistribution = detectOpensearchDistribution(localConfiguration); + datanodeConfiguration = new DatanodeConfiguration( + opensearchDistribution, + localConfiguration.getDatanodeNodeName(), + localConfiguration.getProcessLogsBufferSize() + ); + } + + @Override + public DatanodeConfiguration get() { + return datanodeConfiguration; + } + + private static OpensearchDistribution detectOpensearchDistribution(final Configuration configuration) throws IOException { + return OpensearchDistribution.detectInDirectory(Path.of(configuration.getOpensearchDistributionRoot())); + } +} diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationException.java b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationException.java new file mode 100644 index 0000000000..3556bec37e --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationException.java @@ -0,0 +1,27 @@ +/* + * 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.configuration; + +public class OpensearchConfigurationException extends RuntimeException { + public OpensearchConfigurationException(String message) { + super(message); + } + + public OpensearchConfigurationException(Exception cause) { + super(cause); + } +} diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java new file mode 100644 index 0000000000..1288e5eca8 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java @@ -0,0 +1,96 @@ +/* + * 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.configuration; + +import org.graylog.datanode.Configuration; +import org.graylog.datanode.configuration.variants.InSecureConfiguration; +import org.graylog.datanode.configuration.variants.MongoCertSecureConfiguration; +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 javax.inject.Inject; +import javax.inject.Provider; +import javax.inject.Singleton; +import java.io.IOException; +import java.nio.file.Path; +import java.security.GeneralSecurityException; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +@Singleton +public class OpensearchConfigurationProvider implements Provider { + + + private final Configuration localConfiguration; + private final UploadedCertFilesSecureConfiguration uploadedCertFilesSecureConfiguration; + private final MongoCertSecureConfiguration mongoCertSecureConfiguration; + private final InSecureConfiguration inSecureConfiguration; + private final DatanodeConfiguration datanodeConfiguration; + + @Inject + public OpensearchConfigurationProvider(Configuration localConfiguration, + DatanodeConfiguration datanodeConfiguration, + UploadedCertFilesSecureConfiguration uploadedCertFilesSecureConfiguration, + MongoCertSecureConfiguration mongoCertSecureConfiguration, + InSecureConfiguration inSecureConfiguration) { + this.localConfiguration = localConfiguration; + this.datanodeConfiguration = datanodeConfiguration; + this.uploadedCertFilesSecureConfiguration = uploadedCertFilesSecureConfiguration; + this.mongoCertSecureConfiguration = mongoCertSecureConfiguration; + this.inSecureConfiguration = inSecureConfiguration; + } + + @Override + public OpensearchConfiguration get() { + final String opensearchConfigLocation = localConfiguration.getOpensearchConfigLocation(); + + //TODO: at some point bind the whole list, for now there is too much experiments with order and prerequisites + List securityConfigurationTypes = List.of( + uploadedCertFilesSecureConfiguration, + mongoCertSecureConfiguration, + inSecureConfiguration //TODO: in final version, this configuration is tried first, not last + ); + + SecurityConfigurationVariant chosenSecurityConfigurationVariant = securityConfigurationTypes.stream() + .filter(s -> s.checkPrerequisites(localConfiguration)) + .findFirst() + .orElseThrow(() -> new OpensearchConfigurationException("No valid option to start up OpenSearch")); + + try { + final Map config = chosenSecurityConfigurationVariant.configure(localConfiguration); + return new OpensearchConfiguration( + datanodeConfiguration.opensearchDistribution().directory(), + Path.of(opensearchConfigLocation), + localConfiguration.getOpensearchHttpPort(), + localConfiguration.getOpensearchTransportPort(), + localConfiguration.getRestApiUsername(), + localConfiguration.getRestApiPassword(), + "datanode-cluster", + localConfiguration.getDatanodeNodeName(), + Collections.emptyList(), + localConfiguration.getOpensearchDiscoverySeedHosts(), + config + ); + } catch (GeneralSecurityException | KeyStoreStorageException | IOException e) { + throw new OpensearchConfigurationException(e); + } + } + +} 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 cdf5a0e68b..c72bdf6b38 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 @@ -16,25 +16,22 @@ */ package org.graylog.datanode.management; +import org.graylog.datanode.process.OpensearchConfiguration; import org.graylog.datanode.process.ProcessInfo; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; import java.net.URI; import java.util.List; +import java.util.Optional; -public interface OpensearchProcess extends ManagableProcess { +public interface OpensearchProcess extends ManagableProcess { ProcessInfo processInfo(); - RestHighLevelClient restClient(); - - Object nodeName(); + Optional restClient(); boolean isLeaderNode(); void setLeaderNode(boolean isManagerNode); - - String opensearchVersion(); - List stdOutLogs(); List stdErrLogs(); 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 d0da1f09d0..85a6d352d2 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,7 @@ 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.apache.commons.io.IOUtils; +import org.graylog.datanode.configuration.DatanodeConfiguration; import org.graylog.datanode.process.OpensearchConfiguration; import org.graylog.datanode.process.ProcessEvent; import org.graylog.datanode.process.ProcessInfo; @@ -43,28 +43,30 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.util.List; import java.util.Locale; -import java.util.Objects; import java.util.Optional; import java.util.Queue; 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; + + @SuppressWarnings("OptionalUsedAsFieldOrParameterType") + private Optional configuration = Optional.empty(); + @SuppressWarnings("OptionalUsedAsFieldOrParameterType") + private Optional restClient = Optional.empty(); private final StateMachine processState; + private final DatanodeConfiguration datanodeConfiguration; + private boolean isLeaderNode; private CommandLineProcess commandLineProcess; private final Queue stdout; private final Queue stderr; - OpensearchProcessImpl(OpensearchConfiguration configuration, int logsCacheSize) { - this.configuration = configuration; - this.restClient = createRestClient(configuration); + OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize) { + this.datanodeConfiguration = datanodeConfiguration; this.processState = ProcessStateMachine.createNew(); this.stdout = new CircularFifoQueue<>(logsCacheSize); this.stderr = new CircularFifoQueue<>(logsCacheSize); @@ -73,7 +75,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { private RestHighLevelClient createRestClient(OpensearchConfiguration configuration) { final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); - final HttpHost host = getRestBaseUrl(configuration); + final HttpHost host = configuration.getRestBaseUrl(); RestClientBuilder builder = RestClient.builder(host); if ("https".equals(host.getSchemeName())) { @@ -85,10 +87,6 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { return new RestHighLevelClient(builder); } - public String opensearchVersion() { - return configuration.opensearchVersion(); - } - @Override public List stdOutLogs() { return stdout.stream().toList(); @@ -99,7 +97,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { return stderr.stream().toList(); } - public RestHighLevelClient restClient() { + public Optional restClient() { return restClient; } @@ -110,24 +108,22 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { public ProcessInfo processInfo() { return process().map(process -> new ProcessInfo( process.pid(), - configuration.nodeName(), processState.getState(), + datanodeConfiguration.nodeName(), processState.getState(), isLeaderNode, process.info().startInstant().orElse(null), process.info().totalCpuDuration().orElse(null), process.info().user().orElse(null), - getRestBaseUrl(configuration).toString() + configuration.map(OpensearchConfiguration::getRestBaseUrl).map(HttpHost::toString).orElse(null) - )).orElseGet(() -> new ProcessInfo(-1, configuration.nodeName(), processState.getState(), false, null, null, null, null)); + )).orElseGet(() -> new ProcessInfo(-1, datanodeConfiguration.nodeName(), processState.getState(), false, null, null, null, null)); } @Override public URI getOpensearchBaseUrl() { - return URI.create(getRestBaseUrl(configuration).toURI()); - } - - private HttpHost getRestBaseUrl(OpensearchConfiguration config) { - final boolean sslEnabled = Boolean.parseBoolean(config.asMap().getOrDefault("plugins.security.ssl.http.enabled", "false")); - return new HttpHost("localhost", config.httpPort(), sslEnabled ? "https" : "http"); + final String baseUrl = configuration.map(OpensearchConfiguration::getRestBaseUrl) + .map(HttpHost::toURI) + .orElse(""); // TODO: will this cause problems in the nodeservice? + return URI.create(baseUrl); } private Optional process() { @@ -153,37 +149,30 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { return isLeaderNode; } - public String nodeName() { - return configuration.nodeName(); - } - public boolean isInState(ProcessState expectedState) { return this.getStatus().equals(expectedState); } @Override - public void startWithConfig(OpensearchDynamicConfiguration configuration) { - if(!Objects.equals(dynamicConfiguration, configuration)) { - this.dynamicConfiguration = configuration; - restart(); - } else { - LOG.info("Dynamic configuration not changed, ignoring"); - } + public void startWithConfig(OpensearchConfiguration configuration) { + this.configuration = Optional.of(configuration); + restart(); + this.restClient = Optional.of(createRestClient(configuration)); } @Override - public synchronized void restart() { - if (dynamicConfiguration == null) { - throw new IllegalArgumentException("Dynamic runtime configuration required but not supplied!"); - } - - stopProcess(); - - 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(); - commandLineProcess = new CommandLineProcess(executable, arguments, this, configuration.getEnv()); - commandLineProcess.start(); + public synchronized void restart() { + configuration.ifPresentOrElse( + (config -> { + stopProcess(); + final Path executable = config.opensearchDir().resolve(Paths.get("bin", "opensearch")); + final List arguments = config.asMap().entrySet().stream() + .map(it -> String.format(Locale.ROOT, "-E%s=%s", it.getKey(), it.getValue())).toList(); + commandLineProcess = new CommandLineProcess(executable, arguments, this, config.getEnv()); + commandLineProcess.start(); + }), + () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");} + ); } @Override @@ -194,11 +183,13 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { } private void stopRestClient() { - try { - restClient().close(); - } catch (IOException e) { - LOG.warn("Failed to close rest client", e); - } + restClient().ifPresent(client -> { + try { + client.close(); + } catch (IOException e) { + LOG.warn("Failed to close rest client", e); + } + }); } private void stopProcess() { 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 a4f8b75298..bc05aa3781 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 @@ -16,37 +16,58 @@ */ package org.graylog.datanode.management; +import com.google.common.eventbus.EventBus; +import com.google.common.eventbus.Subscribe; import com.google.common.util.concurrent.AbstractIdleService; +import org.graylog2.cluster.preflight.NodePreflightStateChangeEvent; +import org.graylog.datanode.configuration.DatanodeConfiguration; +import org.graylog.datanode.configuration.OpensearchConfigurationException; import org.graylog.datanode.process.OpensearchConfiguration; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import javax.inject.Inject; -import javax.inject.Named; import javax.inject.Provider; import javax.inject.Singleton; @Singleton public class OpensearchProcessService extends AbstractIdleService implements Provider { + private static final Logger LOG = LoggerFactory.getLogger(OpensearchProcessService.class); + private static final int WATCHDOG_RESTART_ATTEMPTS = 3; private final OpensearchProcess process; + private final Provider configurationProvider; @Inject - public OpensearchProcessService(OpensearchConfiguration config, @Named(value = "process_logs_buffer_size") int logsSize) { - this.process = createOpensearchProcess(config, logsSize); + public OpensearchProcessService(DatanodeConfiguration datanodeConfiguration, Provider configurationProvider, EventBus eventBus) { + this.configurationProvider = configurationProvider; + this.process = createOpensearchProcess(datanodeConfiguration); + eventBus.register(this); } - private OpensearchProcess createOpensearchProcess(OpensearchConfiguration config, int logsSize) { - final OpensearchProcessImpl process = new OpensearchProcessImpl(config, logsSize); + private OpensearchProcess createOpensearchProcess(DatanodeConfiguration datanodeConfiguration) { + final OpensearchProcessImpl process = new OpensearchProcessImpl(datanodeConfiguration, datanodeConfiguration.processLogsBufferSize()); final ProcessWatchdog watchdog = new ProcessWatchdog(process, WATCHDOG_RESTART_ATTEMPTS); process.setStateMachineTracer(watchdog); return process; } + @Subscribe + public void handlePreflightConfigEvent(NodePreflightStateChangeEvent event) { + switch (event.state()) { + case STORED -> this.process.startWithConfig(configurationProvider.get()); + } + } + @Override protected void startUp() { - // todo: this should be moved to a configuration provider that will know the right moment when to start the - // opensearch process. - this.process.startWithConfig(new OpensearchDynamicConfiguration()); + try { + final OpensearchConfiguration config = configurationProvider.get(); + this.process.startWithConfig(config); + } catch (OpensearchConfigurationException e) { + LOG.warn("Failed to obtain opensearch configuration. Adapt your datanode configuration or use the preflight web interface", e); + } } diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java b/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java index c1e970318f..6658f4c20e 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/ClusterManagerDiscovery.java @@ -17,27 +17,32 @@ package org.graylog.datanode.periodicals; import com.fasterxml.jackson.databind.ObjectMapper; +import org.graylog.datanode.configuration.DatanodeConfiguration; import org.graylog.datanode.management.OpensearchProcess; import org.graylog.datanode.process.ProcessState; 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; import org.graylog2.plugin.periodical.Periodical; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.inject.Inject; import javax.inject.Singleton; +import java.io.IOException; import java.util.Optional; @Singleton public class ClusterManagerDiscovery extends Periodical { private static final Logger LOG = LoggerFactory.getLogger(ClusterManagerDiscovery.class); + private final DatanodeConfiguration datanodeConfiguration; private final OpensearchProcess managedOpenSearch; private final ObjectMapper objectMapper; @Inject - public ClusterManagerDiscovery(OpensearchProcess managedOpenSearch, ObjectMapper objectMapper) { + public ClusterManagerDiscovery(DatanodeConfiguration datanodeConfiguration, OpensearchProcess managedOpenSearch, ObjectMapper objectMapper) { + this.datanodeConfiguration = datanodeConfiguration; this.managedOpenSearch = managedOpenSearch; this.objectMapper = objectMapper; } @@ -48,7 +53,7 @@ public class ClusterManagerDiscovery extends Periodical { if (managedOpenSearch.isInState(ProcessState.AVAILABLE)) { final Boolean isManagerNode = getClusterStateResponse(managedOpenSearch) .map(r -> r.nodes().get(r.clusterManagerNode())) - .map(managerNode -> managedOpenSearch.nodeName().equals(managerNode.name())) + .map(managerNode -> datanodeConfiguration.nodeName().equals(managerNode.name())) .orElse(false); managedOpenSearch.setLeaderNode(isManagerNode); } @@ -56,14 +61,15 @@ public class ClusterManagerDiscovery extends Periodical { private Optional getClusterStateResponse(OpensearchProcess process) { - try { - final Response response = process.restClient() - .getLowLevelClient() - .performRequest(new Request("GET", "_cluster/state/")); + return process.restClient().flatMap(this::requestClusterState); + } - final ClusterStateResponse clusterState = objectMapper.readValue(response.getEntity().getContent(), ClusterStateResponse.class); - return Optional.of(clusterState); - } catch (Exception e) { + private Optional requestClusterState(RestHighLevelClient client) { + try { + final Response response = client.getLowLevelClient().performRequest(new Request("GET", "_cluster/state/")); + final ClusterStateResponse state = objectMapper.readValue(response.getEntity().getContent(), ClusterStateResponse.class); + return Optional.of(state); + } catch (IOException e) { LOG.warn("Failed to obtain cluster state response", e); return Optional.empty(); } diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java b/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java index f47d485c97..8fbdd84121 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java @@ -35,19 +35,19 @@ public class NodePingPeriodical extends Periodical { private static final Logger LOG = LoggerFactory.getLogger(NodePingPeriodical.class); private final NodeService nodeService; private final NodeId nodeId; - private final URI opensearchBaseUri; + private final Supplier opensearchBaseUri; private final Supplier isLeader; @Inject public NodePingPeriodical(NodeService nodeService, NodeId nodeId, OpensearchProcess managedOpenSearch) { - this(nodeService, nodeId, managedOpenSearch.getOpensearchBaseUrl(), managedOpenSearch::isLeaderNode); + this(nodeService, nodeId, managedOpenSearch::getOpensearchBaseUrl, managedOpenSearch::isLeaderNode); } NodePingPeriodical( NodeService nodeService, NodeId nodeId, - URI opensearchBaseUri, + Supplier opensearchBaseUri, Supplier isLeader ) { this.nodeService = nodeService; @@ -100,7 +100,7 @@ public class NodePingPeriodical extends Periodical { @Override public void doRun() { try { - nodeService.markAsAlive(nodeId, isLeader.get(), opensearchBaseUri); + nodeService.markAsAlive(nodeId, isLeader.get(), opensearchBaseUri.get()); } catch (NodeNotFoundException e) { LOG.warn("Did not find meta info of this node. Re-registering."); registerServer(); @@ -110,7 +110,7 @@ public class NodePingPeriodical extends Periodical { private void registerServer() { final boolean registrationSucceeded = nodeService.registerServer(nodeId.getNodeId(), isLeader.get(), - opensearchBaseUri, + opensearchBaseUri.get(), Tools.getLocalCanonicalHostname()); if (!registrationSucceeded) { 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 f896bbeb91..3cca8e5886 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 @@ -21,6 +21,7 @@ import org.graylog.datanode.process.ProcessEvent; import org.graylog.datanode.process.ProcessState; 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; import org.graylog.shaded.opensearch2.org.opensearch.client.core.MainResponse; import org.graylog2.plugin.periodical.Periodical; import org.slf4j.Logger; @@ -30,6 +31,7 @@ import javax.annotation.Nonnull; import javax.inject.Inject; import javax.inject.Singleton; import java.io.IOException; +import java.util.Optional; @Singleton public class OpensearchNodeHeartbeat extends Periodical { @@ -46,12 +48,16 @@ public class OpensearchNodeHeartbeat extends Periodical { // 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)) { - try { - final MainResponse health = process.restClient() - .info(RequestOptions.DEFAULT); - onNodeResponse(process, health); - } catch (IOException | OpenSearchStatusException e) { - onRestError(process, e); + + final Optional restClient = process.restClient(); + if(restClient.isPresent()) { + try { + final MainResponse health = restClient.get() + .info(RequestOptions.DEFAULT); + onNodeResponse(process, health); + } catch (IOException | OpenSearchStatusException e) { + onRestError(process, e); + } } } } diff --git a/data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java b/data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java index 5e5cf60a73..6aaa88c364 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java +++ b/data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java @@ -17,6 +17,7 @@ package org.graylog.datanode.process; import org.graylog.datanode.management.Environment; +import org.graylog.shaded.opensearch2.org.apache.http.HttpHost; import java.nio.file.Path; import java.util.LinkedHashMap; @@ -24,7 +25,6 @@ import java.util.List; import java.util.Map; public record OpensearchConfiguration( - String opensearchVersion, Path opensearchDir, Path opensearchConfigDir, int httpPort, @@ -64,4 +64,9 @@ public record OpensearchConfiguration( env.put("OPENSEARCH_PATH_CONF", opensearchConfigDir.resolve("opensearch").toAbsolutePath().toString()); return env; } + + public HttpHost getRestBaseUrl() { + final boolean sslEnabled = Boolean.parseBoolean(asMap().getOrDefault("plugins.security.ssl.http.enabled", "false")); + return new HttpHost("localhost", httpPort(), sslEnabled ? "https" : "http"); + } } 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 9bc9a627fb..668a521007 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,7 +16,7 @@ */ package org.graylog.datanode.rest; -import org.graylog.datanode.management.OpensearchDynamicConfiguration; +import org.graylog.datanode.configuration.DatanodeConfiguration; import org.graylog.datanode.management.OpensearchProcess; import org.graylog2.plugin.Version; @@ -25,7 +25,6 @@ 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) @@ -33,10 +32,12 @@ public class StatusController { private final Version version = Version.CURRENT_CLASSPATH; + private final DatanodeConfiguration datanodeConfiguration; private final OpensearchProcess openSearch; @Inject - public StatusController(OpensearchProcess openSearch) { + public StatusController(DatanodeConfiguration datanodeConfiguration, OpensearchProcess openSearch) { + this.datanodeConfiguration = datanodeConfiguration; this.openSearch = openSearch; } @@ -44,14 +45,7 @@ public class StatusController { public DataNodeStatus status() { return new DataNodeStatus( version, - new StatusResponse(openSearch.opensearchVersion(), openSearch.processInfo()) + new StatusResponse(datanodeConfiguration.opensearchDistribution().version(), 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/periodicals/NodePingPeriodicalTest.java b/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java index d867dc59bc..0cc7c5254d 100644 --- a/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java +++ b/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java @@ -36,7 +36,7 @@ class NodePingPeriodicalTest { final NodePingPeriodical task = new NodePingPeriodical( nodeService, nodeID, - uri, + () -> uri, () -> true ); @@ -62,7 +62,7 @@ class NodePingPeriodicalTest { final NodePingPeriodical task = new NodePingPeriodical( nodeService, nodeID, - uri, + () -> uri, () -> true ); diff --git a/graylog2-server/src/main/java/org/graylog/security/certutil/cert/storage/CertMongoStorage.java b/graylog2-server/src/main/java/org/graylog/security/certutil/cert/storage/CertMongoStorage.java index e893e6cf3e..be54af01d8 100644 --- a/graylog2-server/src/main/java/org/graylog/security/certutil/cert/storage/CertMongoStorage.java +++ b/graylog2-server/src/main/java/org/graylog/security/certutil/cert/storage/CertMongoStorage.java @@ -21,7 +21,7 @@ import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter; import org.bouncycastle.openssl.PEMParser; import org.bouncycastle.openssl.jcajce.JcaPEMWriter; import org.bouncycastle.operator.OperatorCreationException; -import org.graylog2.cluster.NodePreflightConfigService; +import org.graylog2.cluster.preflight.NodePreflightConfigService; import javax.inject.Inject; import java.io.BufferedReader; diff --git a/graylog2-server/src/main/java/org/graylog/security/certutil/csr/storage/CsrMongoStorage.java b/graylog2-server/src/main/java/org/graylog/security/certutil/csr/storage/CsrMongoStorage.java index eaa140426a..7616092dff 100644 --- a/graylog2-server/src/main/java/org/graylog/security/certutil/csr/storage/CsrMongoStorage.java +++ b/graylog2-server/src/main/java/org/graylog/security/certutil/csr/storage/CsrMongoStorage.java @@ -20,8 +20,8 @@ import org.bouncycastle.openssl.PEMParser; import org.bouncycastle.openssl.jcajce.JcaPEMWriter; import org.bouncycastle.operator.OperatorException; import org.bouncycastle.pkcs.PKCS10CertificationRequest; -import org.graylog2.cluster.NodePreflightConfig; -import org.graylog2.cluster.NodePreflightConfigService; +import org.graylog2.cluster.preflight.NodePreflightConfig; +import org.graylog2.cluster.preflight.NodePreflightConfigService; import javax.inject.Inject; import java.io.BufferedReader; diff --git a/graylog2-server/src/main/java/org/graylog2/bootstrap/ServerBootstrap.java b/graylog2-server/src/main/java/org/graylog2/bootstrap/ServerBootstrap.java index 9f1cd338df..d0973c8b37 100644 --- a/graylog2-server/src/main/java/org/graylog2/bootstrap/ServerBootstrap.java +++ b/graylog2-server/src/main/java/org/graylog2/bootstrap/ServerBootstrap.java @@ -41,6 +41,7 @@ import org.graylog2.bootstrap.preflight.PreflightCheckService; import org.graylog2.bootstrap.preflight.PreflightWebModule; import org.graylog2.bootstrap.preflight.ServerPreflightChecksModule; import org.graylog2.bootstrap.preflight.web.PreflightBoot; +import org.graylog2.cluster.preflight.PreflightConfigBindings; import org.graylog2.configuration.IndexerDiscoveryModule; import org.graylog2.configuration.PathConfiguration; import org.graylog2.configuration.TLSProtocolsConfiguration; @@ -167,6 +168,7 @@ public abstract class ServerBootstrap extends CmdLineTool { private void runPreflightWeb(List preflightCheckModules) { List modules = new ArrayList<>(preflightCheckModules); + modules.add(new PreflightConfigBindings()); modules.add(new PreflightWebModule()); modules.add(new ObjectMapperModule(chainingClassLoader)); modules.add(new SchedulerBindings()); diff --git a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogPreflightGeneratePeriodical.java b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogPreflightGeneratePeriodical.java index 144a3e4e33..ff3cb8d572 100644 --- a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogPreflightGeneratePeriodical.java +++ b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogPreflightGeneratePeriodical.java @@ -20,8 +20,8 @@ import org.graylog.security.certutil.cert.storage.CertMongoStorage; import org.graylog.security.certutil.cert.storage.CertStorage; import org.graylog.security.certutil.csr.CsrSigner; import org.graylog.security.certutil.csr.storage.CsrMongoStorage; -import org.graylog2.cluster.NodePreflightConfig; -import org.graylog2.cluster.NodePreflightConfigService; +import org.graylog2.cluster.preflight.NodePreflightConfig; +import org.graylog2.cluster.preflight.NodePreflightConfigService; import org.graylog2.plugin.periodical.Periodical; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java index 248fbd6f9d..eb4d528301 100644 --- a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java +++ b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java @@ -26,7 +26,6 @@ import org.graylog2.bootstrap.preflight.web.PreflightBoot; import org.graylog2.bootstrap.preflight.web.resources.PreflightAssetsResource; import org.graylog2.bootstrap.preflight.web.resources.PreflightResource; import org.graylog2.bootstrap.preflight.web.resources.PreflightStatusResource; -import org.graylog2.cluster.NodePreflightConfigService; import org.graylog2.cluster.NodeService; import org.graylog2.cluster.NodeServiceImpl; import org.graylog2.database.MongoConnection; @@ -45,7 +44,6 @@ public class PreflightWebModule extends Graylog2Module { bind(ServiceManager.class).toProvider(ServiceManagerProvider.class).asEagerSingleton(); bind(MongoConnection.class).toProvider(MongoConnectionProvider.class); bind(NodeService.class).to(NodeServiceImpl.class); - bind(NodePreflightConfigService.class).asEagerSingleton(); bind(PreflightConfigService.class).to(PreflightConfigServiceImpl.class); bind(PreflightBoot.class).asEagerSingleton(); diff --git a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java index 73e7f59d13..261833ad47 100644 --- a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java +++ b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java @@ -22,8 +22,8 @@ import org.graylog2.bootstrap.preflight.web.resources.model.CA; import org.graylog2.bootstrap.preflight.web.resources.model.CAType; import org.graylog2.bootstrap.preflight.web.resources.model.CertParameters; import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodePreflightConfig; -import org.graylog2.cluster.NodePreflightConfigService; +import org.graylog2.cluster.preflight.NodePreflightConfig; +import org.graylog2.cluster.preflight.NodePreflightConfigService; import org.graylog2.cluster.NodeService; import javax.inject.Inject; diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/NodePreflightConfig.java b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfig.java similarity index 98% rename from graylog2-server/src/main/java/org/graylog2/cluster/NodePreflightConfig.java rename to graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfig.java index 074bea8c76..9b5fb0e4b5 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/NodePreflightConfig.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfig.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog2.cluster; +package org.graylog2.cluster.preflight; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigBusEvents.java b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigBusEvents.java new file mode 100644 index 0000000000..962b4c8ea6 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigBusEvents.java @@ -0,0 +1,86 @@ +/* + * 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.graylog2.cluster.preflight; + +import com.google.common.eventbus.EventBus; + +import javax.inject.Inject; +import javax.inject.Named; +import java.util.Optional; +import java.util.stream.Stream; + +public class NodePreflightConfigBusEvents implements NodePreflightConfigService { + + public static final String DELEGATE_NAME = "impl"; + + private final NodePreflightConfigService delegate; + private final EventBus eventBus; + + @Inject + public NodePreflightConfigBusEvents(@Named(DELEGATE_NAME) NodePreflightConfigService delegate, EventBus eventBus) { + this.delegate = delegate; + this.eventBus = eventBus; + } + + @Override + public void changeState(String nodeId, NodePreflightConfig.State state) { + delegate.changeState(nodeId, state); + eventBus.post(new NodePreflightStateChangeEvent(nodeId, state)); + } + + @Override + public NodePreflightConfig save(NodePreflightConfig config) { + final NodePreflightConfig saved = delegate.save(config); + eventBus.post(new NodePreflightStateChangeEvent(config.nodeId(), config.state())); + return saved; + } + + @Override + public NodePreflightConfig getPreflightConfigFor(String nodeId) { + return delegate.getPreflightConfigFor(nodeId); + } + + @Override + public void writeCsr(String nodeId, String csr) { + delegate.writeCsr(nodeId, csr); + } + + @Override + public void writeCert(String nodeId, String cert) { + delegate.writeCert(nodeId, cert); + } + + @Override + public Optional readCert(String nodeId) { + return delegate.readCert(nodeId); + } + + @Override + public Stream streamAll() { + return delegate.streamAll(); + } + + @Override + public int delete(String id) { + return delegate.delete(id); + } + + @Override + public void deleteAll() { + delegate.deleteAll(); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigService.java b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigService.java new file mode 100644 index 0000000000..e292e3324b --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigService.java @@ -0,0 +1,39 @@ +/* + * 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.graylog2.cluster.preflight; + +import java.util.Optional; +import java.util.stream.Stream; + +public interface NodePreflightConfigService { + NodePreflightConfig getPreflightConfigFor(String nodeId); + + void writeCsr(String nodeId, String csr); + + void writeCert(String nodeId, String cert); + + Optional readCert(String nodeId); + + void changeState(String nodeId, NodePreflightConfig.State state); + + NodePreflightConfig save(NodePreflightConfig config); + + Stream streamAll(); + + int delete(String id); + void deleteAll(); +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/NodePreflightConfigService.java b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigServiceImpl.java similarity index 65% rename from graylog2-server/src/main/java/org/graylog2/cluster/NodePreflightConfigService.java rename to graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigServiceImpl.java index a4a66fc1e5..24bac3776a 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/NodePreflightConfigService.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightConfigServiceImpl.java @@ -14,57 +14,49 @@ * along with this program. If not, see * . */ -package org.graylog2.cluster; +package org.graylog2.cluster.preflight; -import org.graylog.security.certutil.csr.CsrGenerator; import org.graylog2.bindings.providers.MongoJackObjectMapperProvider; import org.graylog2.database.MongoConnection; import org.graylog2.database.PaginatedDbService; import org.graylog2.database.indices.MongoDbIndexTools; +import org.graylog2.shared.utilities.StringUtils; import org.mongojack.DBQuery; import org.mongojack.DBUpdate; import org.mongojack.JacksonDBCollection; import org.mongojack.WriteResult; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.inject.Inject; import java.util.Optional; -import static org.graylog2.cluster.NodePreflightConfig.FIELD_CERTIFICATE; -import static org.graylog2.cluster.NodePreflightConfig.FIELD_CSR; -import static org.graylog2.cluster.NodePreflightConfig.FIELD_NODEID; -import static org.graylog2.cluster.NodePreflightConfig.FIELD_STATE; -import static org.graylog2.cluster.NodePreflightConfig.State.CSR; -import static org.graylog2.cluster.NodePreflightConfig.State.SIGNED; +import static org.graylog2.cluster.preflight.NodePreflightConfig.FIELD_CERTIFICATE; +import static org.graylog2.cluster.preflight.NodePreflightConfig.FIELD_CSR; +import static org.graylog2.cluster.preflight.NodePreflightConfig.FIELD_NODEID; +import static org.graylog2.cluster.preflight.NodePreflightConfig.FIELD_STATE; +import static org.graylog2.cluster.preflight.NodePreflightConfig.State.CSR; +import static org.graylog2.cluster.preflight.NodePreflightConfig.State.SIGNED; -public class NodePreflightConfigService extends PaginatedDbService { +public class NodePreflightConfigServiceImpl extends PaginatedDbService implements NodePreflightConfigService { public static final String COLLECTION_NAME = "node_preflight_config"; - private static final Logger LOG = LoggerFactory.getLogger(NodePreflightConfigService.class); private final JacksonDBCollection dbCollection; - private final CsrGenerator csrGenerator; @Inject - public NodePreflightConfigService(final MongoJackObjectMapperProvider mongoJackObjectMapperProvider, - final MongoConnection mongoConnection, - final CsrGenerator csrGenerator) { + public NodePreflightConfigServiceImpl(final MongoJackObjectMapperProvider mongoJackObjectMapperProvider, + final MongoConnection mongoConnection) { super(mongoConnection, mongoJackObjectMapperProvider, NodePreflightConfig.class, COLLECTION_NAME); this.dbCollection = JacksonDBCollection.wrap(mongoConnection.getDatabase().getCollection(COLLECTION_NAME), NodePreflightConfig.class, String.class, mongoJackObjectMapperProvider.get()); new MongoDbIndexTools(db).createUniqueIndex(FIELD_NODEID); - this.csrGenerator = csrGenerator; - } - - public NodePreflightConfig getPreflightConfigFor(Node node) { - return getPreflightConfigFor(node.getNodeId()); } + @Override public NodePreflightConfig getPreflightConfigFor(String nodeId) { return dbCollection.findOne(DBQuery.is(FIELD_NODEID, nodeId)); } - public boolean writeCsr(final String nodeId, - final String csr) { + @Override + public void writeCsr(final String nodeId, + final String csr) { final WriteResult result = dbCollection.update( DBQuery.is(FIELD_NODEID, nodeId), new DBUpdate.Builder() @@ -73,11 +65,14 @@ public class NodePreflightConfigService extends PaginatedDbService 0; + if (result.getN() != 1) { + throw new RuntimeException(StringUtils.f("Failed to write node %s certificate", nodeId)); + } } - public boolean writeCert(final String nodeId, - final String cert) { + @Override + public void writeCert(final String nodeId, + final String cert) { final WriteResult result = dbCollection.update( DBQuery.is(FIELD_NODEID, nodeId), new DBUpdate.Builder() @@ -86,9 +81,12 @@ public class NodePreflightConfigService extends PaginatedDbService 0; + if (result.getN() != 1) { + throw new RuntimeException(StringUtils.f("Failed to write node %s certificate", nodeId)); + } } + @Override public Optional readCert(final String nodeId) { final NodePreflightConfig config = dbCollection.findOne( DBQuery.is(FIELD_NODEID, nodeId) @@ -100,7 +98,8 @@ public class NodePreflightConfigService extends PaginatedDbService result = dbCollection.update( DBQuery.is(FIELD_NODEID, nodeId), new DBUpdate.Builder() @@ -108,7 +107,9 @@ public class NodePreflightConfigService extends PaginatedDbService 0; + if (result.getN() != 1) { + throw new RuntimeException(StringUtils.f("Failed to change node %s state", nodeId)); + } } diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchDynamicConfiguration.java b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightStateChangeEvent.java similarity index 82% rename from data-node/src/main/java/org/graylog/datanode/management/OpensearchDynamicConfiguration.java rename to graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightStateChangeEvent.java index fc258ac9c1..ff3dd16497 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchDynamicConfiguration.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/NodePreflightStateChangeEvent.java @@ -14,7 +14,7 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.management; +package org.graylog2.cluster.preflight; -public class OpensearchDynamicConfiguration { +public record NodePreflightStateChangeEvent(String nodeId, NodePreflightConfig.State state) { } diff --git a/data-node/src/main/java/org/graylog/datanode/OpensearchDistributionProvider.java b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/PreflightConfigBindings.java similarity index 50% rename from data-node/src/main/java/org/graylog/datanode/OpensearchDistributionProvider.java rename to graylog2-server/src/main/java/org/graylog2/cluster/preflight/PreflightConfigBindings.java index f9e242cd43..f51bd4b903 100644 --- a/data-node/src/main/java/org/graylog/datanode/OpensearchDistributionProvider.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/preflight/PreflightConfigBindings.java @@ -14,26 +14,19 @@ * along with this program. If not, see * . */ -package org.graylog.datanode; +package org.graylog2.cluster.preflight; -import javax.inject.Inject; -import javax.inject.Provider; -import javax.inject.Singleton; -import java.io.IOException; -import java.nio.file.Path; +import com.google.inject.AbstractModule; +import com.google.inject.name.Names; -@Singleton -public class OpensearchDistributionProvider implements Provider { - - private final OpensearchDistribution opensearchDistribution; - - @Inject - public OpensearchDistributionProvider(Configuration configuration) throws IOException { - this.opensearchDistribution = OpensearchDistribution.detectInDirectory(Path.of(configuration.getOpensearchDistributionRoot())); - } +public class PreflightConfigBindings extends AbstractModule { @Override - public OpensearchDistribution get() { - return opensearchDistribution; + protected void configure() { + // this wires the NodePreflightConfigServiceImpl delegate into the NodePreflightConfigBusEvents from above + bind(NodePreflightConfigService.class).annotatedWith(Names.named(NodePreflightConfigBusEvents.DELEGATE_NAME)).to(NodePreflightConfigServiceImpl.class); + + // this is the generic dependency used by callers + bind(NodePreflightConfigService.class).to(NodePreflightConfigBusEvents.class); } } diff --git a/graylog2-server/src/main/java/org/graylog2/shared/bindings/providers/EventBusProvider.java b/graylog2-server/src/main/java/org/graylog2/shared/bindings/providers/EventBusProvider.java index 7acec0e18c..a297089883 100644 --- a/graylog2-server/src/main/java/org/graylog2/shared/bindings/providers/EventBusProvider.java +++ b/graylog2-server/src/main/java/org/graylog2/shared/bindings/providers/EventBusProvider.java @@ -21,10 +21,10 @@ import com.codahale.metrics.MetricRegistry; import com.google.common.eventbus.AsyncEventBus; import com.google.common.eventbus.EventBus; import com.google.common.util.concurrent.ThreadFactoryBuilder; -import org.graylog2.plugin.BaseConfiguration; import org.graylog2.shared.events.DeadEventLoggingListener; import javax.inject.Inject; +import javax.inject.Named; import javax.inject.Provider; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -33,18 +33,18 @@ import java.util.concurrent.ThreadFactory; import static com.codahale.metrics.MetricRegistry.name; public class EventBusProvider implements Provider { - private final BaseConfiguration configuration; + private final int asyncEventbusProcessors; private final MetricRegistry metricRegistry; @Inject - public EventBusProvider(final BaseConfiguration configuration, final MetricRegistry metricRegistry) { - this.configuration = configuration; + public EventBusProvider(final @Named("async_eventbus_processors") int asyncEventbusProcessors, final MetricRegistry metricRegistry) { + this.asyncEventbusProcessors = asyncEventbusProcessors; this.metricRegistry = metricRegistry; } @Override public EventBus get() { - final EventBus eventBus = new AsyncEventBus("graylog-eventbus", executorService(configuration.getAsyncEventbusProcessors())); + final EventBus eventBus = new AsyncEventBus("graylog-eventbus", executorService(asyncEventbusProcessors)); eventBus.register(new DeadEventLoggingListener()); return eventBus;