Data node's OpenSearch upgrade UI

changelog/unreleased/pr-25860.toml

@@ -0,0 +1,5 @@
+type = "a"
+message = "Add reindexing for outdated indices (created in previous major version of OpenSearch)"
+
+issues = ["Graylog2/graylog-plugin-enterprise#14011"]
+pulls = ["25860"]

changelog/unreleased/pr-26076.toml

@@ -0,0 +1,4 @@
+type = "f"
+message = "Provide certificate authentication for internal OpenSearch maintenance calls."
+issues = ["Graylog2/graylog-plugin-enterprise#14009"]
+pulls = ["26076"]

data-node/src/main/java/org/graylog/datanode/opensearch/configuration/beans/impl/OpensearchSecurityConfigurationBean.java

@@ -34,6 +34,7 @@
 import org.graylog.datanode.process.configuration.files.OpensearchSecurityConfigurationFile;
 import org.graylog.security.certutil.csr.InMemoryKeystoreInformation;
 import org.graylog.security.certutil.csr.KeystoreInformation;
+import org.graylog2.indexer.security.IndexerAdminCertConstants;
 import org.graylog2.security.JwtSecret;
 import org.graylog2.security.TruststoreCreator;
 import org.slf4j.Logger;
@@ -192,7 +193,7 @@ private Map<String, String> commonSecurityConfig() {
 
         config.put("plugins.security.nodes_dn", "CN=*");
         config.put("plugins.security.allow_default_init_securityindex", "true");
-        //config.put("plugins.security.authcz.admin_dn", "CN=kirk,OU=client,O=client,L=test,C=de");
+        config.put("plugins.security.authcz.admin_dn", IndexerAdminCertConstants.ADMIN_DN);
 
         config.put("plugins.security.enable_snapshot_restore_privilege", "true");
         config.put("plugins.security.check_snapshot_restore_write_privileges", "true");

full-backend-tests/src/test/java/org/graylog2/indexer/indices/OutdatedIndexServiceIT.java

@@ -0,0 +1,69 @@
+/*
+ * 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.graylog2.indexer.indices;
+
+import org.graylog.testing.completebackend.FullBackendTest;
+import org.graylog.testing.completebackend.GraylogBackendConfiguration;
+import org.graylog.testing.completebackend.Lifecycle;
+import org.graylog.testing.elasticsearch.BulkIndexRequest;
+import org.graylog.testing.elasticsearch.SearchServerBaseTest;
+import org.graylog2.indexer.counts.CountsAdapter;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@GraylogBackendConfiguration(serverLifecycle = Lifecycle.CLASS)
+public class OutdatedIndexServiceIT extends SearchServerBaseTest {
+
+    OutdatedIndexService outdatedIndexService;
+    IndicesAdapter indicesAdapter;
+    CountsAdapter countsAdapter;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        indicesAdapter = searchServer().adapters().indicesAdapter();
+        countsAdapter = searchServer().adapters().countsAdapter();
+        outdatedIndexService = new OutdatedIndexService(indicesAdapter, null, null);
+    }
+
+    @FullBackendTest
+    public void testReindexingSuccessful() {
+        String toReindex = client().createRandomIndex("reindextest");
+        BulkIndexRequest bulkIndexRequest = new BulkIndexRequest();
+        long messageCount = 10L;
+        for (int i = 0; i < messageCount; i++) {
+            bulkIndexRequest.addRequest(toReindex, Map.of("foo", "bar" + i));
+        }
+        client().bulkIndex(bulkIndexRequest);
+        String originalId = indicesAdapter.getIndexId(toReindex);
+        Optional<DateTime> originalCreationDate = indicesAdapter.indexCreationDate(toReindex);
+        assertThat(countsAdapter.totalCount(List.of(toReindex))).isEqualTo(messageCount);
+        outdatedIndexService.reindex(toReindex, true);
+        assertThat(countsAdapter.totalCount(List.of(toReindex))).isEqualTo(messageCount);
+        assertThat(indicesAdapter.getIndexId(toReindex)).isNotEqualTo(originalId);
+        assertThat(indicesAdapter.indexCreationDate(toReindex).get()).isNotEqualTo(originalCreationDate.get());
+        indicesAdapter.delete(toReindex);
+    }
+
+
+}

graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/Elasticsearch7Module.java

@@ -43,6 +43,7 @@
 import org.graylog2.indexer.messages.MessagesAdapter;
 import org.graylog2.indexer.results.MultiChunkResultRetriever;
 import org.graylog2.indexer.searches.SearchesAdapter;
+import org.graylog2.indexer.security.IndexerAdminCert;
 import org.graylog2.indexer.security.SecurityAdapter;
 import org.graylog2.migrations.V20170607164210_MigrateReopenedIndicesToAliases;
 import org.graylog2.plugin.VersionAwareModule;
@@ -63,6 +64,7 @@ protected void configure() {
         bindForSupportedVersion(CountsAdapter.class).to(CountsAdapterES7.class);
         bindForSupportedVersion(ClusterAdapter.class).to(ClusterAdapterES7.class);
         bindForSupportedVersion(IndicesAdapter.class).to(IndicesAdapterES7.class);
+        bindForSupportedVersion(IndicesAdapter.class, IndexerAdminCert.class).to(IndicesAdapterES7.class);
         bindForSupportedVersion(DataStreamAdapter.class).to(DataStreamAdapterES7.class);
         if (useComposableIndexTemplates) {
             bindForSupportedVersion(IndexTemplateAdapter.class).to(ComposableIndexTemplateAdapter.class);
@@ -99,4 +101,9 @@ protected void configure() {
     private <T> LinkedBindingBuilder<T> bindForSupportedVersion(Class<T> interfaceClass) {
         return bindForVersion(supportedVersion, interfaceClass);
     }
+
+    private <T> LinkedBindingBuilder<T> bindForSupportedVersion(Class<T> interfaceClass,
+                                                                Class<? extends java.lang.annotation.Annotation> qualifier) {
+        return bindForVersion(supportedVersion, interfaceClass, qualifier);
+    }
 }

graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/IndicesAdapterES7.java

@@ -138,7 +138,7 @@ public IndicesAdapterES7(ElasticsearchClient client,
     }
 
     @Override
-    public void move(String source, String target, Consumer<IndexMoveResult> resultCallback) {
+    public void reindex(String source, String target, Consumer<IndexMoveResult> resultCallback) {
         final ReindexRequest request = new ReindexRequest();
         request.setSourceIndices(source);
         request.setDestIndex(target);

graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/IndicesAdapterOS2.java

@@ -138,7 +138,7 @@ public IndicesAdapterOS2(OpenSearchClient client,
     }
 
     @Override
-    public void move(String source, String target, Consumer<IndexMoveResult> resultCallback) {
+    public void reindex(String source, String target, Consumer<IndexMoveResult> resultCallback) {
         final ReindexRequest request = new ReindexRequest();
         request.setSourceIndices(source);
         request.setDestIndex(target);

graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java

@@ -50,6 +50,7 @@
 import org.graylog2.indexer.messages.MessagesAdapter;
 import org.graylog2.indexer.results.MultiChunkResultRetriever;
 import org.graylog2.indexer.searches.SearchesAdapter;
+import org.graylog2.indexer.security.IndexerAdminCert;
 import org.graylog2.indexer.security.SecurityAdapter;
 import org.graylog2.migrations.V20170607164210_MigrateReopenedIndicesToAliases;
 import org.graylog2.plugin.VersionAwareModule;
@@ -70,6 +71,7 @@ protected void configure() {
         bindForSupportedVersion(CountsAdapter.class).to(CountsAdapterOS2.class);
         bindForSupportedVersion(ClusterAdapter.class).to(ClusterAdapterOS2.class);
         bindForSupportedVersion(IndicesAdapter.class).to(IndicesAdapterOS2.class);
+        bindForSupportedVersion(IndicesAdapter.class, IndexerAdminCert.class).to(IndicesAdapterOS2.class);
         bindForSupportedVersion(DataStreamAdapter.class).to(DataStreamAdapterOS2.class);
         bindForSupportedVersion(SecurityAdapter.class).to(SecurityAdapterOS.class);
         if (useComposableIndexTemplates) {
@@ -113,4 +115,9 @@ protected void configure() {
     private <T> LinkedBindingBuilder<T> bindForSupportedVersion(Class<T> interfaceClass) {
         return bindForVersion(supportedVersion, interfaceClass);
     }
+
+    private <T> LinkedBindingBuilder<T> bindForSupportedVersion(Class<T> interfaceClass,
+                                                                Class<? extends java.lang.annotation.Annotation> qualifier) {
+        return bindForVersion(supportedVersion, interfaceClass, qualifier);
+    }
 }

graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/IndicesAdapterOS2IT.java

@@ -31,4 +31,5 @@ public class IndicesAdapterOS2IT extends IndicesAdapterIT {
     protected SearchServerInstance searchServer() {
         return openSearchInstance;
     }
+
 }

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/AdminIndicesAdapterProvider.java

@@ -0,0 +1,76 @@
+/*
+ * 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.storage.opensearch3;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.inject.Inject;
+import jakarta.inject.Provider;
+import org.graylog.storage.opensearch3.cluster.ClusterStateApi;
+import org.graylog.storage.opensearch3.stats.ClusterStatsApi;
+import org.graylog.storage.opensearch3.stats.IndexStatisticsBuilder;
+import org.graylog.storage.opensearch3.stats.StatsApi;
+import org.graylog2.indexer.indices.IndexTemplateAdapter;
+import org.graylog2.indexer.indices.IndicesAdapter;
+
+/**
+ * Builds an {@link IndicesAdapterOS} backed by the admin-cert OpenSearch client. Used for the
+ * {@code @AdminIndexer IndicesAdapter} binding, which routes its calls through
+ * {@link AdminOpensearchClientProvider}.
+ *
+ * <p>T\he returned adapter holds
+ * a stable client reference because {@link AdminOpensearchClientProvider} hot-swaps the
+ * underlying transport rather than replacing the client.
+ */
+public class AdminIndicesAdapterProvider implements Provider<IndicesAdapter> {
+
+    private final AdminOpensearchClientProvider adminClientProvider;
+    private final StatsApi statsApi;
+    private final ClusterStatsApi clusterStatsApi;
+    private final ClusterStateApi clusterStateApi;
+    private final IndexTemplateAdapter indexTemplateAdapter;
+    private final IndexStatisticsBuilder indexStatisticsBuilder;
+    private final ObjectMapper objectMapper;
+
+    @Inject
+    public AdminIndicesAdapterProvider(AdminOpensearchClientProvider adminClientProvider,
+                                       StatsApi statsApi,
+                                       ClusterStatsApi clusterStatsApi,
+                                       ClusterStateApi clusterStateApi,
+                                       IndexTemplateAdapter indexTemplateAdapter,
+                                       IndexStatisticsBuilder indexStatisticsBuilder,
+                                       ObjectMapper objectMapper) {
+        this.adminClientProvider = adminClientProvider;
+        this.statsApi = statsApi;
+        this.clusterStatsApi = clusterStatsApi;
+        this.clusterStateApi = clusterStateApi;
+        this.indexTemplateAdapter = indexTemplateAdapter;
+        this.indexStatisticsBuilder = indexStatisticsBuilder;
+        this.objectMapper = objectMapper;
+    }
+
+    @Override
+    public IndicesAdapter get() {
+        return new IndicesAdapterOS(
+                adminClientProvider.getAdminClient(),
+                statsApi,
+                clusterStatsApi,
+                clusterStateApi,
+                indexTemplateAdapter,
+                indexStatisticsBuilder,
+                objectMapper);
+    }
+}

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/AdminOpensearchClientProvider.java

@@ -0,0 +1,147 @@
+/*
+ * 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.storage.opensearch3;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.annotation.Nonnull;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import org.graylog.security.certutil.ClientCertSslContextFactory;
+import org.graylog.storage.opensearch3.client.CustomAsyncOpenSearchClient;
+import org.graylog.storage.opensearch3.client.CustomOpenSearchClient;
+import org.graylog2.configuration.IndexerHosts;
+import org.graylog2.indexer.security.IndexerAdminCertConstants;
+import org.opensearch.client.transport.OpenSearchTransport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.SSLContext;
+import java.net.URI;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.graylog.storage.opensearch3.OfficialOpensearchClientProvider.TransportConfig;
+
+/**
+ * Provides an {@link OfficialOpensearchClient} that authenticates via a short-lived in-memory
+ * client certificate signed by Graylog's CA. The certificate uses
+ * {@link IndexerAdminCertConstants#ADMIN_DN}, which the Data Node configures as
+ * {@code plugins.security.authcz.admin_dn} — bypassing the security plugin and granting
+ * superuser access for tasks like reindexing system indices.
+ *
+ * <p>Requires a configured CA (Data Node only). Calls fail if no CA is present.
+ *
+ * <p>The returned client is stable: the same {@link OfficialOpensearchClient} instance is
+ * returned forever, while the underlying transport is hot-swapped through a
+ * {@link DynamicTransport} when the cert nears expiry. This makes the returned client safe
+ * to cache in adapter constructors.
+ */
+@Singleton
+public class AdminOpensearchClientProvider {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AdminOpensearchClientProvider.class);
+
+    static final Duration CERT_LIFETIME = Duration.ofMinutes(15);
+    static final Duration REFRESH_BEFORE_EXPIRY = Duration.ofMinutes(1);
+
+    private final ClientCertSslContextFactory sslContextFactory;
+    private final List<URI> hosts;
+    private final OfficialOpensearchClientProvider transportProvider;
+    private final ObjectMapper objectMapper;
+    private final Clock clock;
+
+    private volatile OfficialOpensearchClient cachedClient;
+    private volatile DynamicTransport dynamicTransport;
+    private volatile ScheduledExecutorService drainScheduler;
+    private volatile Instant currentCertExpiresAt;
+
+    @Inject
+    public AdminOpensearchClientProvider(ClientCertSslContextFactory sslContextFactory,
+                                         @IndexerHosts List<URI> hosts,
+                                         OfficialOpensearchClientProvider transportProvider,
+                                         ObjectMapper objectMapper,
+                                         Clock clock) {
+        this.sslContextFactory = sslContextFactory;
+        this.hosts = hosts;
+        this.transportProvider = transportProvider;
+        this.objectMapper = objectMapper;
+        this.clock = clock;
+    }
+
+    /**
+     * Returns the admin client. The same {@link OfficialOpensearchClient} instance is returned
+     * across the lifetime of this provider; only the internal transport (and underlying cert)
+     * is rotated when the cert nears expiry.
+     */
+    @Nonnull
+    public OfficialOpensearchClient getAdminClient() {
+        if (cachedClient != null && !needsRefresh(clock.instant())) {
+            return cachedClient;
+        }
+        return initOrRefresh();
+    }
+
+    private synchronized OfficialOpensearchClient initOrRefresh() {
+        final Instant now = clock.instant();
+        if (cachedClient != null && !needsRefresh(now)) {
+            return cachedClient;
+        }
+
+        try {
+            final SSLContext sslContext = sslContextFactory.buildClientCertSslContext(
+                    IndexerAdminCertConstants.ADMIN_CN, CERT_LIFETIME);
+            final OpenSearchTransport newTransport = transportProvider.buildTransport(hosts, TransportConfig.clientCertAuth(sslContext));
+
+            if (cachedClient == null) {
+                this.drainScheduler = createDrainScheduler();
+                this.dynamicTransport = new DynamicTransport(newTransport, drainScheduler);
+                this.cachedClient = new OfficialOpensearchClient(
+                        new CustomOpenSearchClient(dynamicTransport),
+                        new CustomAsyncOpenSearchClient(dynamicTransport),
+                        objectMapper);
+                LOG.info("Built admin OpenSearch client with a {} min cert lifetime.", CERT_LIFETIME.toMinutes());
+            } else {
+                dynamicTransport.swap(newTransport);
+                LOG.debug("Rotated admin OpenSearch client certificate.");
+            }
+            this.currentCertExpiresAt = now.plus(CERT_LIFETIME);
+        } catch (RuntimeException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new IllegalStateException("Failed to build admin OpenSearch client", e);
+        }
+
+        return cachedClient;
+    }
+
+    private boolean needsRefresh(Instant now) {
+        return currentCertExpiresAt == null
+                || !now.isBefore(currentCertExpiresAt.minus(REFRESH_BEFORE_EXPIRY));
+    }
+
+    private static ScheduledExecutorService createDrainScheduler() {
+        return Executors.newSingleThreadScheduledExecutor(r -> {
+            final Thread t = new Thread(r, "admin-opensearch-transport-drain");
+            t.setDaemon(true);
+            return t;
+        });
+    }
+}