Support for searchable snapshots in datanode (#18357)

* Support for searchable snapshots in datanode

* Added changelog

* Code cleanup, extracted s3 repository configuration

* fixed installing of s3 repo plugin

* code cleanup

* Install opensearch plugins from predownloaded zip archives

* fixed configuration key

* output datanode logs in DatanodeSearchableSnapshotsIT if startup fails

* transfer predownloaded opensearch plugins to the datanode docker image

* Set default node.search.cache.size to 10gb

* javadoc, code cleanup

* code cleanup

* fix s3 configuration enabled check, added test

* Add md5 checksum to the s3 repository download

data-node/src/main/java/org/graylog/datanode/Configuration.java

@@ -85,6 +85,9 @@ public class Configuration extends BaseConfiguration {
     @Parameter(value = "opensearch_location")
     private String opensearchDistributionRoot = "dist";
 
+    @Parameter(value = "opensearch_plugins_location", validators = DirectoryReadableValidator.class)
+    private Path opensearchPluginsDir = Path.of("dist/plugins");
+
     @Parameter(value = "opensearch_data_location", required = true, validators = DirectoryWritableValidator.class)
     private Path opensearchDataLocation = Path.of("datanode/data");
 
@@ -240,6 +243,9 @@ public class Configuration extends BaseConfiguration {
     @Parameter(value = "metrics_policy")
     private String metricsPolicy = "gl-datanode-metrics-ism";
 
+    @Parameter(value = "node_search_cache_size")
+    private String searchCacheSize = "10gb";
+  
     /**
      * https://opensearch.org/docs/latest/tuning-your-cluster/availability-and-recovery/snapshots/snapshot-restore/#shared-file-system
      */
@@ -277,6 +283,11 @@ public class Configuration extends BaseConfiguration {
         return opensearchDistributionRoot;
     }
 
+    @Nullable
+    public Path getOpensearchPluginsDir() {
+        return opensearchPluginsDir;
+    }
+
     /**
      * Use {@link DatanodeDirectories} to obtain a reference to this directory.
      */
@@ -648,6 +659,11 @@ public class Configuration extends BaseConfiguration {
         return rootPasswordSha2;
     }
 
+
+    public String getNodeSearchCacheSize() {
+        return searchCacheSize;
+    }
+  
     public List<String> getPathRepo() {
         return pathRepo;
     }

data-node/src/main/java/org/graylog/datanode/OpensearchDistribution.java

@@ -29,7 +29,12 @@ public record OpensearchDistribution(Path directory, String version, @Nullable S
         this(path, version, null, null);
     }
 
-    public Path getOpensearchExecutable() {
-        return directory.resolve(Paths.get("bin", "opensearch"));
+    public Path getOpensearchBinDirPath() {
+        return directory.resolve("bin");
     }
+
+    public Path getOpensearchExecutable() {
+        return getOpensearchBinDirPath().resolve("opensearch");
+    }
+
 }

data-node/src/main/java/org/graylog/datanode/commands/Server.java

@@ -31,8 +31,10 @@ import org.graylog.datanode.bindings.PeriodicalBindings;
 import org.graylog.datanode.bindings.ServerBindings;
 import org.graylog.datanode.bootstrap.Main;
 import org.graylog.datanode.bootstrap.ServerBootstrap;
+import org.graylog.datanode.configuration.S3RepositoryConfiguration;
 import org.graylog.datanode.rest.RestBindings;
 import org.graylog.datanode.shutdown.GracefulShutdown;
+import org.graylog.plugins.map.config.GeoIpProcessorConfig;
 import org.graylog2.bindings.MongoDBModule;
 import org.graylog2.bindings.PasswordAlgorithmBindings;
 import org.graylog2.cluster.nodes.DataNodeDto;
@@ -63,6 +65,7 @@ public class Server extends ServerBootstrap {
     private static final Logger LOG = LoggerFactory.getLogger(Server.class);
 
     protected static final Configuration configuration = new Configuration();
+    private final S3RepositoryConfiguration s3RepositoryConfiguration = new S3RepositoryConfiguration();
     private final MongoDbConfiguration mongoDbConfiguration = new MongoDbConfiguration();
     private final TLSProtocolsConfiguration tlsConfiguration = new TLSProtocolsConfiguration();
 
@@ -101,7 +104,8 @@ public class Server extends ServerBootstrap {
     protected List<Object> getCommandConfigurationBeans() {
         return Arrays.asList(configuration,
                 mongoDbConfiguration,
-                tlsConfiguration);
+                tlsConfiguration,
+                s3RepositoryConfiguration);
     }
 
     private static class ShutdownHook implements Runnable {

data-node/src/main/java/org/graylog/datanode/configuration/DatanodeDirectories.java

@@ -34,8 +34,6 @@ import java.util.Set;
 
 /**
  * This is a collection of pointers to directories used to store data, logs and configuration of the managed opensearch.
- * Each data type is additionally stored in a subdirectory named after the nodeId, to avoid unexpected collisions when
- * running more datanode instances in the same machine.
  */
 public class DatanodeDirectories {
 
@@ -45,12 +43,14 @@ public class DatanodeDirectories {
     private final Path logsTargetDir;
     private final Path configurationSourceDir;
     private final Path configurationTargetDir;
+    private final Path opensearchPluginsDir;
 
-    public DatanodeDirectories(Path dataTargetDir, Path logsTargetDir, @Nullable Path configurationSourceDir, Path configurationTargetDir) {
+    public DatanodeDirectories(Path dataTargetDir, Path logsTargetDir, @Nullable Path configurationSourceDir, Path configurationTargetDir, @Nullable Path opensearchPluginsDir) {
         this.dataTargetDir = dataTargetDir;
         this.logsTargetDir = logsTargetDir;
         this.configurationSourceDir = configurationSourceDir;
         this.configurationTargetDir = configurationTargetDir;
+        this.opensearchPluginsDir = opensearchPluginsDir;
     }
 
     public static DatanodeDirectories fromConfiguration(Configuration configuration, NodeId nodeId) {
@@ -58,7 +58,8 @@ public class DatanodeDirectories {
                 backwardsCompatible(configuration.getOpensearchDataLocation(), nodeId, "opensearch_data_location"),
                 backwardsCompatible(configuration.getOpensearchLogsLocation(), nodeId, "opensearch_logs_location"),
                 configuration.getDatanodeConfigurationLocation(),
-                backwardsCompatible(configuration.getOpensearchConfigLocation(), nodeId, "opensearch_config_location")
+                backwardsCompatible(configuration.getOpensearchConfigLocation(), nodeId, "opensearch_config_location"),
+                configuration.getOpensearchPluginsDir()
         );
 
         LOG.info("Opensearch of the node {} uses following directories as its storage: {}", nodeId.getNodeId(), directories);
@@ -165,6 +166,10 @@ public class DatanodeDirectories {
         return createRestrictedAccessFile(resolvedPath);
     }
 
+    public Optional<Path> getOpensearchPluginsDir() {
+        return Optional.ofNullable(opensearchPluginsDir);
+    }
+
     @Override
     public String toString() {
         return "DatanodeDirectories{" +

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

@@ -53,6 +53,7 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
     private final byte[] signingKey;
     private final NodeService<DataNodeDto> nodeService;
     private final PreflightConfigService preflightConfigService;
+    private final S3RepositoryConfiguration s3RepositoryConfiguration;
 
     @Inject
     public OpensearchConfigurationProvider(final Configuration localConfiguration,
@@ -62,7 +63,8 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
                                            final InSecureConfiguration inSecureConfiguration,
                                            final NodeService<DataNodeDto> nodeService,
                                            final PreflightConfigService preflightConfigService,
-                                           final @Named("password_secret") String passwordSecret) {
+                                           final @Named("password_secret") String passwordSecret,
+                                           final S3RepositoryConfiguration s3RepositoryConfiguration) {
         this.localConfiguration = localConfiguration;
         this.datanodeConfiguration = datanodeConfiguration;
         this.uploadedCertFilesSecureConfiguration = uploadedCertFilesSecureConfiguration;
@@ -71,6 +73,7 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
         this.signingKey = passwordSecret.getBytes(StandardCharsets.UTF_8);
         this.nodeService = nodeService;
         this.preflightConfigService = preflightConfigService;
+        this.s3RepositoryConfiguration = s3RepositoryConfiguration;
     }
 
     private boolean isPreflight() {
@@ -121,9 +124,11 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
                     localConfiguration.getOpensearchTransportPort(),
                     localConfiguration.getClustername(),
                     localConfiguration.getDatanodeNodeName(),
-                    List.of(),
+                    List.of("cluster_manager", "data", "ingest", "remote_cluster_client", "search"),
                     localConfiguration.getOpensearchDiscoverySeedHosts(),
                     securityConfiguration,
+                    s3RepositoryConfiguration,
+                    localConfiguration.getNodeSearchCacheSize(),
                     opensearchProperties.build()
             );
         } catch (GeneralSecurityException | KeyStoreStorageException | IOException e) {

data-node/src/main/java/org/graylog/datanode/configuration/S3RepositoryConfiguration.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
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.datanode.configuration;
+
+import com.github.joschi.jadconfig.Parameter;
+import com.github.joschi.jadconfig.converters.BooleanConverter;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+
+public class S3RepositoryConfiguration {
+
+    @Parameter(value = "s3_client_default_access_key")
+    private String s3ClientDefaultAccessKey;
+
+    @Parameter(value = "s3_client_default_secret_key")
+    private String s3ClientDefaultSecretKey;
+
+    @Parameter(value = "s3_client_default_protocol")
+    private String s3ClientDefaultProtocol = "http";
+
+    @Parameter(value = "s3_client_default_endpoint")
+    private String s3ClientDefaultEndpoint;
+
+    @Parameter(value = "s3_client_default_region")
+    private String s3ClientDefaultRegion = "us-east-2";
+
+    @Parameter(value = "s3_client_default_path_style_access", converter = BooleanConverter.class)
+    private boolean s3ClientDefaultPathStyleAccess = true;
+
+
+    /**
+     * access and secret keys are handled separately and stored in an opensearch keystore.
+     * See usages of {@link #getS3ClientDefaultAccessKey()} and {@link #getS3ClientDefaultSecretKey()}
+     */
+    public Map<String, String> toOpensearchProperties() {
+        return Map.of(
+                "s3.client.default.protocol", s3ClientDefaultProtocol,
+                "s3.client.default.endpoint", s3ClientDefaultEndpoint,
+                "s3.client.default.region", s3ClientDefaultRegion,
+                "s3.client.default.path_style_access", String.valueOf(s3ClientDefaultPathStyleAccess)
+        );
+    }
+
+    public String getS3ClientDefaultAccessKey() {
+        return s3ClientDefaultAccessKey;
+    }
+
+    public String getS3ClientDefaultSecretKey() {
+        return s3ClientDefaultSecretKey;
+    }
+
+    /**
+     * Verify that either both access and secret keys and the endpoint are configured or none of them. Partial configuration
+     * will lead to an IllegalStateException.
+     */
+    public boolean isRepositoryEnabled() {
+        if (noneBlank(s3ClientDefaultEndpoint, s3ClientDefaultAccessKey, s3ClientDefaultSecretKey)) {
+            // All the required properties are set and not blank, s3 repository is enabled
+            return true;
+        } else if (allBlank(s3ClientDefaultEndpoint, s3ClientDefaultAccessKey, s3ClientDefaultSecretKey)) {
+            // all are empty, this means repository is not configured at all
+            return false;
+        } else {
+            // One or two properties are configured, this is an incomplete configuration we can't handle this situation
+            throw new IllegalStateException("""
+                    S3 Client not configured properly, all
+                    s3_client_default_access_key, s3_client_default_secret_key and s3_client_default_endpoint
+                    have to be provided in the configuration!""");
+        }
+    }
+
+    private boolean noneBlank(String... properties) {
+        return Arrays.stream(properties).noneMatch(StringUtils::isBlank);
+    }
+
+    private boolean allBlank(String... properties) {
+        return Arrays.stream(properties).allMatch(StringUtils::isBlank);
+    }
+}

data-node/src/main/java/org/graylog/datanode/management/OpensearchCommandLineProcess.java

@@ -18,14 +18,14 @@ package org.graylog.datanode.management;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import jakarta.validation.constraints.NotNull;
 import org.apache.commons.exec.OS;
+import org.graylog.datanode.management.opensearch.cli.OpensearchCli;
 import org.graylog.datanode.process.OpensearchConfiguration;
 import org.graylog.datanode.process.ProcessInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import jakarta.validation.constraints.NotNull;
-
 import java.io.BufferedReader;
 import java.io.Closeable;
 import java.io.IOException;
@@ -90,12 +90,46 @@ public class OpensearchCommandLineProcess implements Closeable {
 
     public OpensearchCommandLineProcess(OpensearchConfiguration config, ProcessListener listener) {
         fixJdkOnMac(config);
+        configureS3RepositoryPlugin(config);
         final Path executable = config.opensearchDistribution().getOpensearchExecutable();
         writeOpenSearchConfig(config);
         resultHandler = new CommandLineProcessListener(listener);
         commandLineProcess = new CommandLineProcess(executable, List.of(), resultHandler, config.getEnv());
     }
 
+    private void configureS3RepositoryPlugin(OpensearchConfiguration config) {
+        if (config.s3RepositoryConfiguration().isRepositoryEnabled()) {
+            final OpensearchCli opensearchCli = new OpensearchCli(config);
+            installPlugin(opensearchCli, config, "repository-s3");
+            configureS3Credentials(opensearchCli, config);
+        } else {
+            LOG.info("No S3 repository configuration provided, skipping plugin initialization");
+        }
+    }
+
+    private void configureS3Credentials(OpensearchCli opensearchCli, OpensearchConfiguration config) {
+        LOG.info("Creating opensearch keystore");
+        // this always operates on newly created configuration directory, there is no existing keystore present. Otherwise
+        // the command will get stuck. There is currently no timeout, so it will just block the startup forever
+        // TODO: add watchdog and timeout to the underlying command handling.
+        final String createdMessage = opensearchCli.keystore().create();
+        LOG.info(createdMessage);
+        LOG.info("Setting opensearch s3 repository keystore secrets");
+        opensearchCli.keystore().add("s3.client.default.access_key", config.s3RepositoryConfiguration().getS3ClientDefaultAccessKey());
+        opensearchCli.keystore().add("s3.client.default.secret_key", config.s3RepositoryConfiguration().getS3ClientDefaultSecretKey());
+    }
+
+
+    private void installPlugin(OpensearchCli opensearchCli, OpensearchConfiguration config, String pluginName) {
+        final List<String> installedPlugins = opensearchCli.plugin().list();
+        if (!installedPlugins.contains(pluginName)) {
+            opensearchCli.plugin().installFromZip(pluginName, config.opensearchDistribution().version());
+            LOG.info("Successfully installed opensearch plugin " + pluginName);
+        } else {
+            LOG.info("Opensearch plugin " + pluginName + " already installed, skipping");
+        }
+    }
+
     private static Map<String, String> getOpensearchConfigurationArguments(OpensearchConfiguration config) {
         Map<String, String> allArguments = new LinkedHashMap<>(config.asMap());
 

data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java

@@ -54,12 +54,9 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
     private final OpensearchProcess process;
     private final Provider<OpensearchConfiguration> configurationProvider;
     private final EventBus eventBus;
-    private final NodeService<DataNodeDto> nodeService;
     private final NodeId nodeId;
     private final DataNodeProvisioningService dataNodeProvisioningService;
     private final IndexFieldTypesService indexFieldTypesService;
-    private final ObjectMapper objectMapper;
-    private final ProcessStateMachine processStateMachine;
     private final ClusterEventBus clusterEventBus;
 
 
@@ -78,12 +75,9 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
                                     final ClusterEventBus clusterEventBus) {
         this.configurationProvider = configurationProvider;
         this.eventBus = eventBus;
-        this.nodeService = nodeService;
         this.nodeId = nodeId;
         this.dataNodeProvisioningService = dataNodeProvisioningService;
-        this.objectMapper = objectMapper;
         this.indexFieldTypesService = indexFieldTypesService;
-        this.processStateMachine = processStateMachine;
         this.clusterEventBus = clusterEventBus;
         this.process = createOpensearchProcess(datanodeConfiguration, trustManager, configuration, nodeService, objectMapper, processStateMachine);
         eventBus.register(this);
@@ -175,6 +169,8 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro
                 original.nodeRoles(),
                 original.discoverySeedHosts(),
                 original.opensearchSecurityConfiguration(),
+                original.s3RepositoryConfiguration(),
+                original.nodeSearchCacheSize(),
                 finalAdditionalConfig);
 
         if (config.securityConfigured()) {

data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/AbstractOpensearchCli.java

@@ -0,0 +1,115 @@
+/*
+ * 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.opensearch.cli;
+
+import org.apache.commons.exec.CommandLine;
+import org.apache.commons.exec.DefaultExecuteResultHandler;
+import org.apache.commons.exec.DefaultExecutor;
+import org.apache.commons.exec.PumpStreamHandler;
+import org.graylog.datanode.process.OpensearchConfiguration;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public abstract class AbstractOpensearchCli {
+
+    private final Path configPath;
+    private final Path binPath;
+
+    /**
+     *
+     * @param configPath Opensearch CLI tools adapt configuration stored under OPENSEARCH_PATH_CONF env property.
+     *                   This is why wealways want to set this configPath for each CLI tool.
+     * @param bin location of the actual executable binary that this wrapper handles
+     */
+    private AbstractOpensearchCli(Path configPath, Path bin) {
+        this.configPath = configPath;
+        this.binPath = bin;
+    }
+
+    public AbstractOpensearchCli(OpensearchConfiguration config, String binName) {
+        this(config.datanodeDirectories().getOpensearchProcessConfigurationDir(),
+                checkExecutable(config.opensearchDistribution().getOpensearchBinDirPath().resolve(binName)));
+    }
+
+    private static Path checkExecutable(Path path) {
+        if (!Files.isExecutable(path)) {
+            throw new IllegalArgumentException("Path " + path + " doesn't point to any known opensearch cli tool");
+        }
+        return path;
+    }
+
+    protected String runBatch(String... args) {
+        return runWithStdin(Collections.emptyList(), args);
+    }
+
+    /**
+     * @param answersToPrompts Some opensearch CLI tools have to be operated in an interactive way - provide answers
+     *                         to questions the tool is asking. It's scripting unfriendly. Our way around this is to
+     *                         provide an input stream of expected responses, each delimited by \n. There is no validation
+     *                         and no logic, just the expected order of responses.
+     * @param args arguments of the command, in opensearch-keystore create, the create is the first argument
+     * @return All the STDOUT and STDERR of the process merged into one String.
+     */
+    protected String runWithStdin(List<String> answersToPrompts, String... args) {
+        final CommandLine cmd = new CommandLine(binPath.toFile());
+        Arrays.asList(args).forEach(cmd::addArgument);
+
+        final ByteArrayOutputStream stdout = new ByteArrayOutputStream();
+
+        final PumpStreamHandler pumpStreamHandler;
+        if (answersToPrompts.isEmpty()) {
+            pumpStreamHandler = new PumpStreamHandler(stdout, stdout);
+        } else {
+            InputStream input = new ByteArrayInputStream(String.join("\n", answersToPrompts).getBytes(Charset.defaultCharset()));
+            pumpStreamHandler = new PumpStreamHandler(stdout, stdout, input);
+        }
+
+        // TODO: add watchdog and timeout to the underlying command handling
+        final DefaultExecutor executor = new DefaultExecutor.Builder<>()
+                .setExecuteStreamHandler(pumpStreamHandler)
+                .get();
+
+        try {
+            final DefaultExecuteResultHandler executeResultHandler = new DefaultExecuteResultHandler();
+            final Map<String, String> env = Collections.singletonMap("OPENSEARCH_PATH_CONF", configPath.toAbsolutePath().toString());
+            executor.execute(cmd, env, executeResultHandler);
+            executeResultHandler.waitFor();
+            final int exitValue = executeResultHandler.getExitValue();
+            if (exitValue != 0) {
+                throw new RuntimeException("Failed to execute opensearch cli" + binPath + "\n" + formatOutput(stdout));
+            }
+            return stdout.toString(StandardCharsets.UTF_8).trim();
+        } catch (IOException | InterruptedException e) {
+            throw new RuntimeException("Failed to execute opensearch cli" + binPath + "\n" + formatOutput(stdout));
+        }
+    }
+
+    private String formatOutput(ByteArrayOutputStream stdout) {
+        return "STDOUT/STDERR: " + stdout.toString(StandardCharsets.UTF_8);
+    }
+}

data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchCli.java

@@ -0,0 +1,42 @@
+/*
+ * 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.opensearch.cli;
+
+import org.graylog.datanode.process.OpensearchConfiguration;
+
+/**
+ * Collection of opensearch CLI tools. All of them need to have OPENSEARCH_PATH_CONF preconfigured, so they operate
+ * on the correct version of configuration.
+ */
+public class OpensearchCli {
+
+    private final OpensearchKeystoreCli keystore;
+    private final OpensearchPluginCli plugin;
+
+    public OpensearchCli(OpensearchConfiguration config) {
+        this.keystore = new OpensearchKeystoreCli(config);
+        this.plugin = new OpensearchPluginCli(config);
+    }
+
+    public OpensearchKeystoreCli keystore() {
+        return keystore;
+    }
+
+    public OpensearchPluginCli plugin() {
+        return plugin;
+    }
+}

data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchKeystoreCli.java

@@ -0,0 +1,45 @@
+/*
+ * 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.opensearch.cli;
+
+import org.graylog.datanode.process.OpensearchConfiguration;
+
+import java.util.Collections;
+
+public class OpensearchKeystoreCli extends AbstractOpensearchCli {
+
+    OpensearchKeystoreCli(OpensearchConfiguration config) {
+        super(config, "opensearch-keystore");
+    }
+
+    /**
+     * Create a new opensearch keystore. This command expects that there is no keystore. Otherwise, it will ask for
+     * confirmation to override it and this fails, as there is currently no stdin configured.
+     * @return STDOUT/STDERR of the execution as one String
+     */
+    public String create() {
+        return runBatch("create");
+    }
+
+    /**
+     * Add secrets to the store. The command is interactive, it will ask for the secret value (to avoid recording the value
+     * in the command line history). So we have to work around that and provide the value in STDIN.
+     */
+    public void add(String key, String secretValue) {
+        runWithStdin(Collections.singletonList(secretValue), "add", key);
+    }
+}

data-node/src/main/java/org/graylog/datanode/management/opensearch/cli/OpensearchPluginCli.java

@@ -0,0 +1,66 @@
+/*
+ * 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.opensearch.cli;
+
+import org.graylog.datanode.configuration.DatanodeDirectories;
+import org.graylog.datanode.process.OpensearchConfiguration;
+import org.graylog2.shared.utilities.StringUtils;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+public class OpensearchPluginCli extends AbstractOpensearchCli {
+
+    private final Path pluginsPath;
+
+    OpensearchPluginCli(OpensearchConfiguration config) {
+        super(config, "opensearch-plugin");
+        this.pluginsPath = config.datanodeDirectories().getOpensearchPluginsDir()
+                .orElseThrow(() -> new IllegalStateException("Configuration opensearch_plugins_location not set. Please point this configuration to a directory holding plugins ZIP files."));
+    }
+
+    private Path resolveOpensearchPlugin(String pluginName, String version) {
+        final String pluginFileName = StringUtils.f("%s-%s.zip", pluginName, version);
+        final Path pluginFilePath = pluginsPath.resolve(pluginFileName);
+        if (!Files.exists(pluginFilePath)) {
+            throw new IllegalStateException("Failed to find " + pluginFileName + " plugin, tried path " + pluginFilePath);
+        }
+
+        return pluginFilePath;
+    }
+
+    /**
+     * Installs plugin by its name, will be downloaded from the internet. This is not a preferred way for distributed
+     * releases, as it requires access to the internet. We should rather predownload plugins and use {@link #installFromZip(String, String)}
+     */
+    public List<String> list() {
+        final String output = runBatch("list");
+        return Arrays.stream(output.split("\n")).toList();
+    }
+
+    /**
+     * Install plugin from predownloaded zip file. The exact location of the file is determined by {@link DatanodeDirectories#getOpensearchPluginsDir()}
+     * @param pluginName Name without version, e.g. "repository-s3"
+     * @param version Semver version of the plugin. Usually it should correspond to the opensearch version, but it may
+     *                differ for some plugins.
+     */
+    public void installFromZip(String pluginName, String version) {
+        runBatch("install", "--batch", resolveOpensearchPlugin(pluginName, version).toAbsolutePath().toUri().toString());
+    }
+}

data-node/src/main/java/org/graylog/datanode/process/OpensearchConfiguration.java

@@ -19,6 +19,7 @@ package org.graylog.datanode.process;
 import org.apache.commons.exec.OS;
 import org.graylog.datanode.OpensearchDistribution;
 import org.graylog.datanode.configuration.DatanodeDirectories;
+import org.graylog.datanode.configuration.S3RepositoryConfiguration;
 import org.graylog.datanode.configuration.variants.OpensearchSecurityConfiguration;
 import org.graylog.datanode.management.Environment;
 import org.graylog.shaded.opensearch2.org.apache.http.HttpHost;
@@ -34,9 +35,14 @@ public record OpensearchConfiguration(
         String hostname,
         int httpPort,
         int transportPort,
-        String clusterName, String nodeName, List<String> nodeRoles,
+        String clusterName,
+        String nodeName,
+        List<String> nodeRoles,
         List<String> discoverySeedHosts,
         OpensearchSecurityConfiguration opensearchSecurityConfiguration,
+        S3RepositoryConfiguration s3RepositoryConfiguration,
+
+        String nodeSearchCacheSize,
         Map<String, String> additionalConfiguration
 ) {
     public Map<String, String> asMap() {
@@ -71,6 +77,11 @@ public record OpensearchConfiguration(
 
         config.put("discovery.seed_providers", "file");
 
+        config.put("node.search.cache.size", nodeSearchCacheSize);
+        if(s3RepositoryConfiguration.isRepositoryEnabled()) {
+            config.putAll(s3RepositoryConfiguration.toOpensearchProperties());
+        }
+
         config.putAll(additionalConfiguration);
         return config;
     }

data-node/src/test/java/org/graylog/datanode/bootstrap/preflight/DatanodeDirectoriesLockfileCheckTest.java

@@ -37,7 +37,7 @@ class DatanodeDirectoriesLockfileCheckTest {
         final Path logsDirLock = logsDir.resolve(DatanodeDirectoriesLockfileCheck.DATANODE_LOCKFILE);
         final Path configDirLock = configDir.resolve(DatanodeDirectoriesLockfileCheck.DATANODE_LOCKFILE);
 
-        final PreflightCheck check = new DatanodeDirectoriesLockfileCheck("5ca1ab1e-0000-4000-a000-000000000000", new DatanodeDirectories(dataDir, logsDir, null, configDir));
+        final PreflightCheck check = new DatanodeDirectoriesLockfileCheck("5ca1ab1e-0000-4000-a000-000000000000", new DatanodeDirectories(dataDir, logsDir, null, configDir, null));
         check.runCheck();
 
         Assertions.assertThat(Files.readString(dataDirLock)).isEqualTo("5ca1ab1e-0000-4000-a000-000000000000");
@@ -59,7 +59,7 @@ class DatanodeDirectoriesLockfileCheckTest {
         Files.writeString(logsDirLock, "5ca1ab1e-0000-4000-a000-000000000000");
         Files.writeString(configDirLock, "5ca1ab1e-0000-4000-a000-000000000000");
 
-        final PreflightCheck check = new DatanodeDirectoriesLockfileCheck("5ca1ab1e-0000-4000-a000-000000000000", new DatanodeDirectories(dataDir, logsDir, null, configDir));
+        final PreflightCheck check = new DatanodeDirectoriesLockfileCheck("5ca1ab1e-0000-4000-a000-000000000000", new DatanodeDirectories(dataDir, logsDir, null, configDir, null));
         check.runCheck();
 
         Assertions.assertThat(Files.readString(dataDirLock)).isEqualTo("5ca1ab1e-0000-4000-a000-000000000000");
@@ -80,7 +80,7 @@ class DatanodeDirectoriesLockfileCheckTest {
         Files.writeString(logsDirLock, "5ca1ab1e-0000-4000-a000-000000000001");
         Files.writeString(configDirLock, "5ca1ab1e-0000-4000-a000-000000000001");
 
-        final PreflightCheck check = new DatanodeDirectoriesLockfileCheck("5ca1ab1e-0000-4000-a000-000000000000", new DatanodeDirectories(dataDir, logsDir, null, configDir));
+        final PreflightCheck check = new DatanodeDirectoriesLockfileCheck("5ca1ab1e-0000-4000-a000-000000000000", new DatanodeDirectories(dataDir, logsDir, null, configDir, null));
 
         Assertions.assertThatThrownBy(check::runCheck)
                 .isInstanceOf(DatanodeLockFileException.class)

data-node/src/test/java/org/graylog/datanode/configuration/DatanodeDirectoriesTest.java

@@ -29,8 +29,8 @@ import java.nio.file.attribute.PosixFilePermission;
 class DatanodeDirectoriesTest {
 
     @Test
-    void testConfigDirPermissions(@TempDir Path dataDir, @TempDir Path logsDir, @TempDir Path configSourceDir, @TempDir Path configTargetDir) throws IOException {
-        final DatanodeDirectories datanodeDirectories = new DatanodeDirectories(dataDir, logsDir, configSourceDir, configTargetDir);
+    void testConfigDirPermissions(@TempDir Path dataDir, @TempDir Path logsDir, @TempDir Path configSourceDir, @TempDir Path configTargetDir, @TempDir Path opensearchPluginsDir) throws IOException {
+        final DatanodeDirectories datanodeDirectories = new DatanodeDirectories(dataDir, logsDir, configSourceDir, configTargetDir, opensearchPluginsDir);
         final Path dir = datanodeDirectories.createOpensearchProcessConfigurationDir();
         Assertions.assertThat(Files.getPosixFilePermissions(dir)).
                 contains(

data-node/src/test/java/org/graylog/datanode/configuration/S3RepositoryConfigurationTest.java

@@ -0,0 +1,65 @@
+/*
+ * 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.configuration;
+
+import com.github.joschi.jadconfig.JadConfig;
+import com.github.joschi.jadconfig.RepositoryException;
+import com.github.joschi.jadconfig.ValidationException;
+import com.github.joschi.jadconfig.repositories.InMemoryRepository;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+class S3RepositoryConfigurationTest {
+
+    @Test
+    void testFullConfigurationEnabled() throws ValidationException, RepositoryException {
+        final S3RepositoryConfiguration config = initializeConfiguration(Map.of(
+                "s3_client_default_access_key", "user",
+                "s3_client_default_secret_key", "password",
+                "s3_client_default_endpoint", "http://localhost:9000"
+
+        ));
+        Assertions.assertThat(config.isRepositoryEnabled()).isTrue();
+    }
+
+    @Test
+    void testEmptyConfigurationDisabled() throws ValidationException, RepositoryException {
+        final S3RepositoryConfiguration config = initializeConfiguration(Map.of());
+        Assertions.assertThat(config.isRepositoryEnabled()).isFalse();
+    }
+
+    @Test
+    void testPartialConfigurationException() throws ValidationException, RepositoryException {
+        final S3RepositoryConfiguration config = initializeConfiguration(Map.of(
+                "s3_client_default_access_key", "user",
+                "s3_client_default_secret_key", "password"
+
+        ));
+        Assertions.assertThatThrownBy(config::isRepositoryEnabled)
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("S3 Client not configured properly");
+
+    }
+
+    private S3RepositoryConfiguration initializeConfiguration(Map<String, String> properties) throws RepositoryException, ValidationException {
+        final S3RepositoryConfiguration configuration = new S3RepositoryConfiguration();
+        new JadConfig(new InMemoryRepository(properties), configuration).process();
+        return configuration;
+    }
+}

data-node/src/test/java/org/graylog/datanode/integration/DatanodeSearchableSnapshotsIT.java

@@ -0,0 +1,136 @@
+/*
+ * 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.integration;
+
+import com.github.rholder.retry.RetryException;
+import io.restassured.http.ContentType;
+import io.restassured.response.ValidatableResponse;
+import org.graylog.datanode.restoperations.DatanodeOpensearchWait;
+import org.graylog.datanode.restoperations.RestOperationParameters;
+import org.graylog.datanode.testinfra.DatanodeContainerizedBackend;
+import org.graylog.testing.completebackend.S3MinioContainer;
+import org.graylog.testing.containermatrix.MongodbServer;
+import org.graylog.testing.mongodb.MongoDBTestService;
+import org.hamcrest.Matchers;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.Network;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+
+import java.util.concurrent.ExecutionException;
+
+import static io.restassured.RestAssured.given;
+
+public class DatanodeSearchableSnapshotsIT {
+
+
+    private Network network;
+    private MongoDBTestService mongoDB;
+    private S3MinioContainer s3Container;
+    private DatanodeContainerizedBackend datanode;
+
+    @BeforeEach
+    void setUp() {
+        network = Network.newNetwork();
+        mongoDB = MongoDBTestService.create(MongodbServer.DEFAULT_VERSION, network);
+        mongoDB.start();
+        s3Container = new S3MinioContainer(network);
+        s3Container.start();
+
+        datanode = new DatanodeContainerizedBackend(network, mongoDB, "datanode", datanodeContainer -> {
+            datanodeContainer.withEnv("GRAYLOG_DATANODE_S3_CLIENT_DEFAULT_ACCESS_KEY", s3Container.getAccessKey());
+            datanodeContainer.withEnv("GRAYLOG_DATANODE_S3_CLIENT_DEFAULT_SECRET_KEY", s3Container.getSecretKey());
+            datanodeContainer.withEnv("GRAYLOG_DATANODE_S3_CLIENT_DEFAULT_ENDPOINT", s3Container.getInternalURI().toString());
+        });
+
+        try {
+            datanode.start();
+        } catch (Exception e) {
+            System.out.println(datanode.getLogs());
+        }
+    }
+
+    @AfterEach
+    void tearDown() {
+        datanode.stop();
+        s3Container.stop();
+        mongoDB.close();
+        network.close();
+    }
+
+    @Test
+    void testSnapshotCreation() throws ExecutionException, RetryException {
+
+        final var opensearchRestPort = datanode.getOpensearchRestPort();
+        final var baseUrl = "http://localhost:" + opensearchRestPort;
+
+        // give opensearch time to start everything
+        waitForNode(opensearchRestPort);
+
+        // verify that the s3-repository plugin is installed
+        verifyInstalledPlugin(baseUrl, "repository-s3");
+
+        final String bucketName = "my-s3-bucket";
+        final String repositoryName = "my-s3-repository";
+        createS3Bucket(bucketName);
+        createRepository(baseUrl, repositoryName, bucketName);
+
+        createSnapshot(baseUrl, repositoryName)
+                .body("snapshot.state", Matchers.equalTo("SUCCESS"));
+    }
+
+    private static ValidatableResponse createRepository(String baseUrl, String repositoryName, String bucketName) {
+        final String req = """
+                {
+                  "type": "s3",
+                  "settings": {
+                    "bucket": "%s",
+                    "base_path": "my/snapshot/directory"
+                  }
+                }
+                """.formatted(bucketName);
+
+        return given()
+                .body(req)
+                .contentType(ContentType.JSON)
+                .put(baseUrl + "/_snapshot/" + repositoryName)
+                .then()
+                .statusCode(200)
+                .body("acknowledged", Matchers.equalTo(true));
+    }
+
+    private static ValidatableResponse createSnapshot(String baseUrl, String repositoryName) {
+        return given().put(baseUrl + "/_snapshot/" + repositoryName + "/1?wait_for_completion=true").then().statusCode(200);
+    }
+
+    private void createS3Bucket(String bucketName) {
+        try (final S3Client client = s3Container.getClient()) {
+            client.createBucket(CreateBucketRequest.builder().bucket(bucketName).build());
+        }
+    }
+
+    private static void verifyInstalledPlugin(String baseUrl, String pluginName) {
+        given().get(baseUrl + "/_cat/plugins").then().statusCode(200).body(Matchers.containsString(pluginName));
+    }
+
+    private void waitForNode(Integer opensearchRestPort) throws ExecutionException, RetryException {
+        // this instance is not using any security, no truststore or jwt tokens needed
+        new DatanodeOpensearchWait(RestOperationParameters.builder().port(opensearchRestPort).build()).waitForNodesCount(1);
+    }
+}

data-node/src/test/java/org/graylog/datanode/restoperations/RestOperation.java

@@ -50,25 +50,20 @@ public abstract class RestOperation {
     final ValidatableResponse validatedResponse(String url, String method, String body, String errorMessage, Predicate<ValidatableResponse>... predicates) {
 
         if (LOG.isInfoEnabled()) {
-            LOG.info("cURL request: " + formatCurlConnectionInfo(parameters.port(), url));
+            LOG.info("cURL request: " + formatCurlConnectionInfo(parameters, url));
         }
 
 
         final RequestSpecification req = RestAssured.given().accept(ContentType.JSON);
         Optional.ofNullable(parameters.truststore()).ifPresent(ts -> req.trustStore(parameters.truststore()));
         parameters.addAuthorizationHeaders(req);
-        ValidatableResponse response;
-        if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.GET)) {
-            response = req.get(formatUrl(parameters.port(), url)).then();
-        } else if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.PUT)) {
-            response = req.header(new Header("Content-Type", "application/json")).body(body).put(formatUrl(parameters.port(), url)).then();
-        } else if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.POST)) {
-            response = req.header(new Header("Content-Type", "application/json")).body(body).post(formatUrl(parameters.port(), url)).then();
-        } else if (method.toUpperCase(Locale.ENGLISH).equals(HttpMethod.DELETE)) {
-            response = req.delete(formatUrl(parameters.port(), url)).then();
-        } else {
-            throw Assertions.fail("Method not implemented");
-        }
+        ValidatableResponse response = switch (method.toUpperCase(Locale.ENGLISH)) {
+            case HttpMethod.GET -> req.get(formatUrl(parameters, url)).then();
+            case HttpMethod.PUT -> req.header(new Header("Content-Type", "application/json")).body(body).put(formatUrl(parameters, url)).then();
+            case HttpMethod.POST -> req.header(new Header("Content-Type", "application/json")).body(body).post(formatUrl(parameters, url)).then();
+            case HttpMethod.DELETE -> req.delete(formatUrl(parameters, url)).then();
+            default -> throw Assertions.fail("Method not implemented");
+        };
         final boolean responseOk = Arrays.stream(predicates).allMatch(p -> p.test(response));
         if (!responseOk) {
             LOG.error(serializeResponse(response.extract()));
@@ -78,7 +73,7 @@ public abstract class RestOperation {
         return response;
     }
 
-    String formatCurlConnectionInfo(int port, String url) {
+    String formatCurlConnectionInfo(RestOperationParameters port, String url) {
         List<String> builder = new LinkedList<>();
         builder.add("curl -k"); // trust self-signed certs
         builder.add(parameters.formatCurlAuthentication());
@@ -95,7 +90,9 @@ public abstract class RestOperation {
         return String.format(Locale.ROOT, "\nstatus code: %d\nstatus line: %s\n\nContent-type: %s\nheaders: %s\nbody: %s", statusCode, statusLine, contentType, headers, body);
     }
 
-    String formatUrl(int port, String url) {
-        return "https://localhost:" + port + url;
+    String formatUrl(RestOperationParameters params, String url) {
+        final boolean securedConnection = params.truststore() != null;
+        String procotol = securedConnection ? "https" : "http";
+        return procotol + "://localhost:" + params.port() + url;
     }
 }

data-node/src/test/java/org/graylog/datanode/restoperations/RestOperationParameters.java

@@ -27,6 +27,7 @@ public abstract class RestOperationParameters {
 
     abstract int port();
 
+    @Nullable
     abstract KeyStore truststore();
 
     @Nullable

data-node/src/test/java/org/graylog/datanode/restoperations/WaitingRestOperation.java

@@ -48,7 +48,7 @@ public abstract class WaitingRestOperation extends RestOperation {
     final ValidatableResponse waitForResponse(String url, Predicate<ValidatableResponse>... predicates) throws ExecutionException, RetryException {
 
         if (LOG.isInfoEnabled()) {
-            LOG.info("cURL request: " + formatCurlConnectionInfo(parameters.port(), url));
+            LOG.info("cURL request: " + formatCurlConnectionInfo(parameters, url));
         }
 
         //noinspection UnstableApiUsage
@@ -86,12 +86,12 @@ public abstract class WaitingRestOperation extends RestOperation {
 
                 parameters.addAuthorizationHeaders(req);
 
-                return req.get(formatUrl(parameters.port(), url))
+                return req.get(formatUrl(parameters, url))
                         .then();
             });
         } catch (Exception e) {
             if (lastRecordedResponse != null) {
-                LOG.warn("Last recorded opensearch response, waiting for {}: {}", formatUrl(parameters.port(), url), lastRecordedResponse);
+                LOG.warn("Last recorded opensearch response, waiting for {}: {}", formatUrl(parameters, url), lastRecordedResponse);
             }
             throw e;
         }

data-node/src/test/java/org/graylog/datanode/testinfra/DatanodeDevContainerBuilder.java

@@ -24,16 +24,18 @@ import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
-import org.testcontainers.images.PullPolicy;
 import org.testcontainers.images.builder.ImageFromDockerfile;
+import org.testcontainers.images.builder.dockerfile.DockerfileBuilder;
 
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Duration;
+import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import static org.graylog.datanode.testinfra.DatanodeContainerizedBackend.IMAGE_WORKING_DIR;
 import static org.graylog.testing.completebackend.DefaultPluginJarsProvider.getProjectReposPath;
@@ -141,6 +143,7 @@ public class DatanodeDevContainerBuilder implements org.graylog.testing.datanode
                 .withNetwork(network)
                 .withEnv("GRAYLOG_DATANODE_DATA_DIR", "data")
                 .withEnv("GRAYLOG_DATANODE_OPENSEARCH_LOCATION", IMAGE_WORKING_DIR)
+                .withEnv("GRAYLOG_DATANODE_OPENSEARCH_PLUGINS_LOCATION", IMAGE_WORKING_DIR + "/plugins")
                 .withEnv("GRAYLOG_DATANODE_INSECURE_STARTUP", "true")
                 .withEnv("GRAYLOG_DATANODE_CONFIG_LOCATION", IMAGE_WORKING_DIR + "/config") // this is the datanode config dir for certs
                 .withEnv("GRAYLOG_DATANODE_OPENSEARCH_DATA_LOCATION", IMAGE_WORKING_DIR + "/opensearch/data")
@@ -197,25 +200,49 @@ public class DatanodeDevContainerBuilder implements org.graylog.testing.datanode
             throw new RuntimeException("Failed to link opensearch distribution to the datanode docker image, path " + downloadedOpensearch.toAbsolutePath() + " does not exist!");
         }
 
-        return new ImageFromDockerfile("local/graylog-datanode:latest", false)
-                // the following command makes the opensearch tar.gz archive accessible in the docker build context, so it can
-                // be later used by the ADD command
-                .withFileFromPath(opensearchTarArchive, downloadedOpensearch)
-                .withDockerfileFromBuilder(builder ->
-                        builder.from("eclipse-temurin:17-jre-jammy")
-                                .workDir(IMAGE_WORKING_DIR)
-                                .run("mkdir -p opensearch/config")
-                                .run("mkdir -p opensearch/data")
-                                .run("mkdir -p opensearch/logs")
-                                .run("mkdir -p config")
+        final Path pluginsDir = getPath().resolve(Path.of("opensearch", "plugins"));
+        final List<String> pluginNames = getPluginNames(pluginsDir);
+        LOG.info("Detected following opensearch plugins: " + String.join(", ", pluginNames));
 
-                                .add(opensearchTarArchive, "./" + opensearchTarArchive + "/") // this will automatically extract the tar
-                                .run("touch datanode.conf") // create empty configuration file, required but all config comes via env props
-                                .run("useradd opensearch")
-                                .run("chown -R opensearch:opensearch " + IMAGE_WORKING_DIR)
-                                .user("opensearch")
-                                .expose(DatanodeContainerizedBackend.DATANODE_REST_PORT, DatanodeContainerizedBackend.DATANODE_OPENSEARCH_HTTP_PORT)
-                                .entryPoint("java", "-jar", "graylog-datanode.jar", "datanode", "-f", "datanode.conf")
-                                .build());
+        final ImageFromDockerfile image = new ImageFromDockerfile("local/graylog-datanode:latest", false);
+
+        // the following command makes the opensearch tar.gz archive accessible in the docker build context, so it can
+        // be later used by the ADD command
+        image.withFileFromPath(opensearchTarArchive, downloadedOpensearch);
+
+        // add plugin files to the docker build context, so they can be used by ADD command later
+        pluginNames.forEach(pluginName -> image.withFileFromPath(pluginName, pluginsDir.resolve(pluginName)));
+
+        return image.withDockerfileFromBuilder(builder ->
+        {
+            final DockerfileBuilder fileBuilder = builder.from("eclipse-temurin:17-jre-jammy")
+                    .workDir(IMAGE_WORKING_DIR)
+                    .run("mkdir -p opensearch/config")
+                    .run("mkdir -p opensearch/data")
+                    .run("mkdir -p opensearch/logs")
+                    .run("mkdir -p config")
+                    .run("mkdir -p plugins")
+                    .add(opensearchTarArchive, "./" + opensearchTarArchive + "/"); // this will automatically extract the tar
+
+            pluginNames.forEach(pluginName -> {
+                fileBuilder.add(pluginName, "./plugins/" + pluginName);
+            });
+
+            fileBuilder.run("touch datanode.conf") // create empty configuration file, required but all config comes via env props
+                    .run("useradd opensearch")
+                    .run("chown -R opensearch:opensearch " + IMAGE_WORKING_DIR)
+                    .user("opensearch")
+                    .expose(DatanodeContainerizedBackend.DATANODE_REST_PORT, DatanodeContainerizedBackend.DATANODE_OPENSEARCH_HTTP_PORT)
+                    .entryPoint("java", "-jar", "graylog-datanode.jar", "datanode", "-f", "datanode.conf");
+            builder.build();
+        });
+    }
+
+    private static List<String> getPluginNames(Path pluginsDir) {
+        try {
+            return Files.list(pluginsDir).map(p -> p.getFileName().toString()).collect(Collectors.toList());
+        } catch (IOException e) {
+            throw new RuntimeException("Failed to list opensearch plugins", e);
+        }
     }
 }