Merge branch 'master' into search-cluster-health-reporters

Author: patrickmann

changelog/unreleased/pr-26674.toml

@@ -0,0 +1,4 @@
+type = "f"
+message = "Fix Kafka inputs being unable to process LZ4-compressed record batches."
+
+pulls = ["26674"]

data-node/pom.xml

@@ -21,7 +21,6 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.graylog.security.certutil.ClientCertSslContextFactory;
-import org.graylog.security.certutil.ClientCertSslContextFactoryImpl;
 import org.graylog.storage.opensearch3.client.CustomAsyncOpenSearchClient;
 import org.graylog.storage.opensearch3.client.CustomOpenSearchClient;
 import org.graylog2.configuration.IndexerHosts;
@@ -53,6 +52,10 @@
  * 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.
+ *
+ * <p>Refresh is lazy and driven by actual use: the cert is only rotated when a request is
+ * about to be sent and the current cert is close to expiry (see {@link #refreshIfNeeded()},
+ * wired as the transport's pre-request hook).
  */
 @Singleton
 public class AdminOpensearchClientProvider {
@@ -70,7 +73,6 @@ public class AdminOpensearchClientProvider {
 
     private volatile OfficialOpensearchClient cachedClient;
     private volatile DynamicTransport dynamicTransport;
-    private volatile ScheduledExecutorService drainScheduler;
     private volatile Instant currentCertExpiresAt;
 
     @Inject
@@ -89,48 +91,60 @@ public AdminOpensearchClientProvider(ClientCertSslContextFactory sslContextFacto
     /**
      * 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.
+     * is rotated, lazily, when a request is sent through a client whose cert nears expiry.
      */
     @Nonnull
     public OfficialOpensearchClient getAdminClient() {
-        if (cachedClient != null && !needsRefresh(clock.instant())) {
-            return cachedClient;
-        }
         return initOrRefresh();
     }
 
+    /**
+     * Rotates the cert if it is close to expiry. Wired as the {@link DynamicTransport}
+     * pre-request hook so the cached (and therefore never re-fetched) client picks up a fresh
+     * cert on its next actual use rather than relying on a background timer.
+     */
+    void refreshIfNeeded() {
+        if (cachedClient != null && needsRefresh(clock.instant())) {
+            initOrRefresh();
+        }
+    }
+
     private synchronized OfficialOpensearchClient initOrRefresh() {
         final Instant now = clock.instant();
         if (cachedClient != null && !needsRefresh(now)) {
             return cachedClient;
         }
 
+        if (cachedClient == null) {
+            this.dynamicTransport = new DynamicTransport(buildTransport(), createDrainScheduler(), this::refreshIfNeeded);
+            this.cachedClient = new OfficialOpensearchClient(
+                    new CustomOpenSearchClient(dynamicTransport),
+                    new CustomAsyncOpenSearchClient(dynamicTransport),
+                    objectMapper);
+            this.currentCertExpiresAt = now.plus(CERT_LIFETIME);
+            LOG.info("Built admin OpenSearch client with a {} min cert lifetime.", CERT_LIFETIME.toMinutes());
+        } else {
+            dynamicTransport.swap(buildTransport());
+            this.currentCertExpiresAt = now.plus(CERT_LIFETIME);
+            LOG.debug("Rotated admin OpenSearch client certificate.");
+        }
+        return cachedClient;
+    }
+
+    /**
+     * Builds a fresh transport authenticated with a newly minted, short-lived admin cert.
+     */
+    private OpenSearchTransport buildTransport() {
         try {
-            final OpenSearchTransport newTransport = sslContextFactory.buildClientCertSslContext(IndexerAdminCertConstants.ADMIN_CN, CERT_LIFETIME)
+            return sslContextFactory.buildClientCertSslContext(IndexerAdminCertConstants.ADMIN_CN, CERT_LIFETIME)
                     .map(TransportConfig::clientCertAuth)
                     .map(certAuth -> transportProvider.buildTransport(hosts, certAuth))
-                    .orElse(transportProvider.buildTransport(hosts));
-
-            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);
+                    .orElseGet(() -> transportProvider.buildTransport(hosts));
         } 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) {

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

@@ -39,18 +39,29 @@ public class DynamicTransport implements OpenSearchTransport {
 
     private final AtomicReference<OpenSearchTransport> current;
     private final ScheduledExecutorService scheduler;
+    private final Runnable beforeRequest;
     private final AtomicLong swapGeneration = new AtomicLong(0);
     private volatile long drainedGeneration = 0;
 
     public DynamicTransport(OpenSearchTransport initial, ScheduledExecutorService scheduler) {
+        this(initial, scheduler, () -> {});
+    }
+
+    /**
+     * @param beforeRequest hook run before every request is dispatched. Lets an owner lazily
+     *                      rotate the underlying transport (e.g. to refresh an expiring client
+     *                      certificate). Must be cheap and idempotent, as it runs on the request path.
+     */
+    public DynamicTransport(OpenSearchTransport initial, ScheduledExecutorService scheduler, Runnable beforeRequest) {
         this.current = new AtomicReference<>(initial);
         this.scheduler = scheduler;
+        this.beforeRequest = beforeRequest;
     }
 
     public void swap(OpenSearchTransport newTransport) {
         final OpenSearchTransport old = current.getAndSet(newTransport);
         final long generation = swapGeneration.incrementAndGet();
-        LOG.info("OpenSearch transport swapped due to node list update (generation {}). Draining old transport.", generation);
+        LOG.info("OpenSearch transport swapped (generation {}). Draining old transport.", generation);
         try {
             scheduler.schedule(() -> {
                 drainedGeneration = generation;
@@ -80,6 +91,7 @@ public <RequestT, ResponseT, ErrorT> ResponseT performRequest(
             RequestT request,
             Endpoint<RequestT, ResponseT, ErrorT> endpoint,
             TransportOptions options) throws IOException {
+        beforeRequest.run();
         try {
             return current.get().performRequest(request, endpoint, options);
         } catch (IOException e) {
@@ -92,6 +104,11 @@ public <RequestT, ResponseT, ErrorT> CompletableFuture<ResponseT> performRequest
             RequestT request,
             Endpoint<RequestT, ResponseT, ErrorT> endpoint,
             TransportOptions options) {
+        try {
+            beforeRequest.run();
+        } catch (RuntimeException e) {
+            return CompletableFuture.failedFuture(e);
+        }
         return current.get().performRequestAsync(request, endpoint, options)
                 .exceptionally(t -> {
                     final Throwable cause = unwrapCompletionException(t);

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

@@ -114,6 +114,29 @@ void refreshesTransportWhenCertNearsExpiryButKeepsClientReference() {
         verify(transportProvider, times(2)).buildTransport(any(), any());
     }
 
+    @Test
+    void refreshIfNeededRotatesTransportOnUseOnlyWhenCertNearsExpiry() {
+        final Instant start = Instant.parse("2026-01-01T00:00:00Z");
+        final MutableClock clock = new MutableClock(start);
+        final AdminOpensearchClientProvider provider = newProvider(clock, initializedSslContextFactory());
+
+        final OfficialOpensearchClient client = provider.getAdminClient();
+        verify(transportProvider, times(1)).buildTransport(any(), any());
+
+        // A "use" while the cert is still fresh must not mint a new cert.
+        provider.refreshIfNeeded();
+        verify(transportProvider, times(1)).buildTransport(any(), any());
+
+        // Once the cert nears expiry, the next "use" rotates the transport in place.
+        clock.advance(AdminOpensearchClientProvider.CERT_LIFETIME);
+        provider.refreshIfNeeded();
+
+        verify(transportProvider, times(2)).buildTransport(any(), any());
+        assertThat(provider.getAdminClient())
+                .as("client reference must remain stable across the on-use rotation")
+                .isSameAs(client);
+    }
+
     @Test
     void requestsCertWithAdminCommonNameAndConfiguredLifetime() {
         final ClientCertSslContextFactory sslContextFactory = (commonName, certificateLifetime) -> {

graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/AdminOpensearchClientProviderTest.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
@@ -173,6 +174,42 @@ void asyncPreservesOriginalExceptionWhenNotSwapping() {
                 .isSameAs(originalException);
     }
 
+    @Test
+    @SuppressWarnings("unchecked")
+    void runsBeforeRequestHookAheadOfEachRequest() throws IOException {
+        final var delegate = mock(OpenSearchTransport.class);
+        final var endpoint = mock(Endpoint.class);
+        final var options = mock(TransportOptions.class);
+        when(delegate.performRequest(any(), any(), any())).thenReturn("result");
+        when(delegate.performRequestAsync(any(), any(), any())).thenReturn(CompletableFuture.completedFuture("result"));
+
+        final var hookCalls = new AtomicInteger();
+        final var transport = new DynamicTransport(delegate, scheduler, hookCalls::incrementAndGet);
+
+        transport.performRequest("req", endpoint, options);
+        transport.performRequestAsync("req", endpoint, options);
+
+        assertThat(hookCalls).hasValue(2);
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void asyncReturnsFailedFutureWhenBeforeRequestHookThrows() {
+        final var delegate = mock(OpenSearchTransport.class);
+        final var endpoint = mock(Endpoint.class);
+        final var options = mock(TransportOptions.class);
+        final var hookFailure = new IllegalStateException("cert refresh failed");
+
+        final var transport = new DynamicTransport(delegate, scheduler, () -> {
+            throw hookFailure;
+        });
+
+        final CompletableFuture<?> future = transport.performRequestAsync("req", endpoint, options);
+        assertThat(future).isCompletedExceptionally();
+        assertThatThrownBy(future::get).isInstanceOf(ExecutionException.class).cause().isSameAs(hookFailure);
+        verify(delegate, never()).performRequestAsync(any(), any(), any());
+    }
+
     @Test
     void closesOldTransportImmediatelyWhenSchedulerRejects() throws IOException {
         final var oldDelegate = mock(OpenSearchTransport.class);

graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/DynamicTransportTest.java

@@ -658,6 +658,11 @@
             <groupId>com.github.luben</groupId>
             <artifactId>zstd-jni</artifactId>
         </dependency>
+        <!-- Explicitly declare lz4 since kafka-clients depends on it and we exclude it from the declaration. -->
+        <dependency>
+            <groupId>at.yawk.lz4</groupId>
+            <artifactId>lz4-java</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>com.github.zafarkhaja</groupId>

graylog2-server/pom.xml

@@ -16,6 +16,7 @@
  */
 package org.graylog.plugins.views.search;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
 import org.graylog.plugins.views.search.errors.SearchError;
 import org.graylog2.indexer.indexset.MongoIndexSet;
 import org.graylog2.indexer.ranges.IndexRange;
@@ -32,7 +33,15 @@ public record SearchResult(Map<String, QueryExplainResult> queries) {
     public record QueryExplainResult(Map<String, ExplainResult> searchTypes) {
     }
 
-    public record ExplainResult(String queryString, Set<IndexRangeResult> searchedIndexRanges) {
+    public record ExplainResult(String queryString, Set<IndexRangeResult> searchedIndexRanges,
+                                @JsonInclude(JsonInclude.Include.NON_NULL) String effectiveQuery) {
+        public ExplainResult(String queryString, Set<IndexRangeResult> searchedIndexRanges) {
+            this(queryString, searchedIndexRanges, null);
+        }
+
+        public ExplainResult withEffectiveQuery(final String effectiveQuery) {
+            return new ExplainResult(queryString, searchedIndexRanges, effectiveQuery);
+        }
     }
 
     public record IndexRangeResult(String indexName, long begin, long end, boolean isWarmTiered,

graylog2-server/src/main/java/org/graylog/plugins/views/search/ExplainResults.java

@@ -27,6 +27,9 @@
 import org.graylog.plugins.views.search.QueryResult;
 import org.graylog.plugins.views.search.Search;
 import org.graylog.plugins.views.search.SearchJob;
+import org.graylog.plugins.views.search.SearchType;
+import org.graylog.plugins.views.search.searchfilters.EffectiveQueryComposer;
+import org.graylog.plugins.views.search.searchfilters.model.UsedSearchFilter;
 import org.graylog.plugins.views.search.elasticsearch.ElasticsearchQueryString;
 import org.graylog.plugins.views.search.errors.QueryError;
 import org.graylog.plugins.views.search.errors.SearchError;
@@ -38,6 +41,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -62,17 +66,20 @@ public class QueryEngine {
     private final Executor dataLakeJobsQueryPool;
     private final ElasticsearchBackendProvider elasticsearchBackendProvider;
     private final Map<String, QueryBackend<? extends GeneratedQueryContext>> unversionedBackends;
+    private final EffectiveQueryComposer effectiveQueryComposer;
 
     @Inject
     public QueryEngine(Configuration configuration,
                        ElasticsearchBackendProvider elasticsearchBackendProvider,
                        Map<String, QueryBackend<? extends GeneratedQueryContext>> unversionedBackends,
                        Set<QueryMetadataDecorator> queryMetadataDecorators,
-                       QueryParser queryParser) {
+                       QueryParser queryParser,
+                       EffectiveQueryComposer effectiveQueryComposer) {
         this.elasticsearchBackendProvider = elasticsearchBackendProvider;
         this.unversionedBackends = unversionedBackends;
         this.queryMetadataDecorators = queryMetadataDecorators;
         this.queryParser = queryParser;
+        this.effectiveQueryComposer = effectiveQueryComposer;
 
         this.indexerJobsQueryPool = createThreadPool(
                 configuration.searchQueryEngineIndexerJobsPoolSize(),
@@ -111,12 +118,31 @@ public ExplainResults explain(SearchJob searchJob, Set<SearchError> validationEr
                     var backend = getBackendForQuery(q);
                     final GeneratedQueryContext generatedQueryContext = backend.generate(q, Set.of(), timezone);
 
-                    return backend.explain(searchJob, q, generatedQueryContext);
+                    return withEffectiveQueries(q, backend.explain(searchJob, q, generatedQueryContext));
                 }));
 
         return new ExplainResults(searchJob.getSearchId(), new ExplainResults.SearchResult(queries), validationErrors);
     }
 
+    private ExplainResults.QueryExplainResult withEffectiveQueries(final Query query, final ExplainResults.QueryExplainResult queryExplainResult) {
+        final Map<String, SearchType> searchTypesById = query.searchTypes().stream()
+                .collect(Collectors.toMap(SearchType::id, s -> s, (a, b) -> a));
+        final String baseQuery = query.query().queryString();
+
+        final Map<String, ExplainResults.ExplainResult> enriched = queryExplainResult.searchTypes().entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, entry -> {
+                    final List<UsedSearchFilter> filters = new java.util.ArrayList<>(
+                            query.filters() != null ? query.filters() : List.of());
+                    final SearchType searchType = searchTypesById.get(entry.getKey());
+                    if (searchType != null && searchType.filters() != null) {
+                        filters.addAll(searchType.filters());
+                    }
+                    return entry.getValue().withEffectiveQuery(effectiveQueryComposer.compose(baseQuery, filters));
+                }));
+
+        return new ExplainResults.QueryExplainResult(enriched);
+    }
+
     @WithSpan
     public SearchJob execute(SearchJob searchJob, Set<SearchError> validationErrors, DateTimeZone timezone) {
         final Set<Query> validQueries = searchJob.getSearch().queries()

graylog2-server/src/main/java/org/graylog/plugins/views/search/engine/QueryEngine.java

@@ -0,0 +1,32 @@
+/*
+ * 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.plugins.views.search.rest;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.graylog.plugins.views.search.searchfilters.model.UsedSearchFilter;
+
+import java.util.List;
+
+public record EffectiveQueryRequest(@JsonProperty("query_string") String queryString,
+                                    @JsonProperty("filters") List<UsedSearchFilter> filters) {
+    @JsonCreator
+    public EffectiveQueryRequest {
+        queryString = queryString == null ? "" : queryString;
+        filters = filters == null ? List.of() : filters;
+    }
+}