Add 'verbose' flag retrieving maximum_timestamp for get data stream API

This commit adds support for the `verbose` querystring parameter to the get data stream API (`GET
/_data_stream/{name}`). The flag defaults to "false". When set to true, the `maximum_timestamp` for
the data stream will be retrieved and returned for each data stream retrieved. This is the same
information available from the data stream stats API (and internally uses the same action to
retrieval).

Author: dakrone

modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java

@@ -35,10 +35,10 @@
 import org.elasticsearch.datastreams.action.CreateDataStreamTransportAction;
 import org.elasticsearch.datastreams.action.DataStreamsStatsTransportAction;
 import org.elasticsearch.datastreams.action.DeleteDataStreamTransportAction;
-import org.elasticsearch.datastreams.action.GetDataStreamsTransportAction;
 import org.elasticsearch.datastreams.action.MigrateToDataStreamTransportAction;
 import org.elasticsearch.datastreams.action.ModifyDataStreamsTransportAction;
 import org.elasticsearch.datastreams.action.PromoteDataStreamTransportAction;
+import org.elasticsearch.datastreams.action.TransportGetDataStreamsAction;
 import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleErrorStore;
 import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService;
 import org.elasticsearch.datastreams.lifecycle.action.DeleteDataStreamLifecycleAction;
@@ -218,7 +218,7 @@ public Collection<?> createComponents(PluginServices services) {
         List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> actions = new ArrayList<>();
         actions.add(new ActionHandler<>(CreateDataStreamAction.INSTANCE, CreateDataStreamTransportAction.class));
         actions.add(new ActionHandler<>(DeleteDataStreamAction.INSTANCE, DeleteDataStreamTransportAction.class));
-        actions.add(new ActionHandler<>(GetDataStreamAction.INSTANCE, GetDataStreamsTransportAction.class));
+        actions.add(new ActionHandler<>(GetDataStreamAction.INSTANCE, TransportGetDataStreamsAction.class));
         actions.add(new ActionHandler<>(DataStreamsStatsAction.INSTANCE, DataStreamsStatsTransportAction.class));
         actions.add(new ActionHandler<>(MigrateToDataStreamAction.INSTANCE, MigrateToDataStreamTransportAction.class));
         actions.add(new ActionHandler<>(PromoteDataStreamAction.INSTANCE, PromoteDataStreamTransportAction.class));

modules/data-streams/src/main/java/org/elasticsearch/datastreams/action/GetDataStreamsTransportAction.java renamed to modules/data-streams/src/main/java/org/elasticsearch/datastreams/action/TransportGetDataStreamsAction.java

@@ -11,11 +11,13 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.datastreams.DataStreamsActionUtil;
+import org.elasticsearch.action.datastreams.DataStreamsStatsAction;
 import org.elasticsearch.action.datastreams.GetDataStreamAction;
 import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.IndexProperties;
 import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.ManagedBy;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
+import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -30,7 +32,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexMode;
@@ -43,31 +45,35 @@
 
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.index.IndexSettings.PREFER_ILM_SETTING;
 
-public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction<
+public class TransportGetDataStreamsAction extends TransportMasterNodeReadAction<
     GetDataStreamAction.Request,
     GetDataStreamAction.Response> {
 
-    private static final Logger LOGGER = LogManager.getLogger(GetDataStreamsTransportAction.class);
+    private static final Logger LOGGER = LogManager.getLogger(TransportGetDataStreamsAction.class);
     private final SystemIndices systemIndices;
     private final ClusterSettings clusterSettings;
     private final DataStreamGlobalRetentionSettings globalRetentionSettings;
+    private final Client client;
 
     @Inject
-    public GetDataStreamsTransportAction(
+    public TransportGetDataStreamsAction(
         TransportService transportService,
         ClusterService clusterService,
         ThreadPool threadPool,
         ActionFilters actionFilters,
         IndexNameExpressionResolver indexNameExpressionResolver,
         SystemIndices systemIndices,
-        DataStreamGlobalRetentionSettings globalRetentionSettings
+        DataStreamGlobalRetentionSettings globalRetentionSettings,
+        Client client
     ) {
         super(
             GetDataStreamAction.NAME,
@@ -78,11 +84,12 @@ public GetDataStreamsTransportAction(
             GetDataStreamAction.Request::new,
             indexNameExpressionResolver,
             GetDataStreamAction.Response::new,
-            EsExecutors.DIRECT_EXECUTOR_SERVICE
+            transportService.getThreadPool().executor(ThreadPool.Names.MANAGEMENT)
         );
         this.systemIndices = systemIndices;
         this.globalRetentionSettings = globalRetentionSettings;
         clusterSettings = clusterService.getClusterSettings();
+        this.client = client;
     }
 
     @Override
@@ -92,9 +99,42 @@ protected void masterOperation(
         ClusterState state,
         ActionListener<GetDataStreamAction.Response> listener
     ) throws Exception {
-        listener.onResponse(
-            innerOperation(state, request, indexNameExpressionResolver, systemIndices, clusterSettings, globalRetentionSettings)
-        );
+        if (request.verbose()) {
+            DataStreamsStatsAction.Request req = new DataStreamsStatsAction.Request();
+            req.indices(request.indices());
+            client.execute(DataStreamsStatsAction.INSTANCE, req, new ActionListener<>() {
+                @Override
+                public void onResponse(DataStreamsStatsAction.Response response) {
+                    final Map<String, Long> maxTimestamps = Arrays.stream(response.getDataStreams())
+                        .collect(
+                            Collectors.toMap(
+                                DataStreamsStatsAction.DataStreamStats::getDataStream,
+                                DataStreamsStatsAction.DataStreamStats::getMaximumTimestamp
+                            )
+                        );
+                    listener.onResponse(
+                        innerOperation(
+                            state,
+                            request,
+                            indexNameExpressionResolver,
+                            systemIndices,
+                            clusterSettings,
+                            globalRetentionSettings,
+                            maxTimestamps
+                        )
+                    );
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    listener.onFailure(e);
+                }
+            });
+        } else {
+            listener.onResponse(
+                innerOperation(state, request, indexNameExpressionResolver, systemIndices, clusterSettings, globalRetentionSettings, null)
+            );
+        }
     }
 
     static GetDataStreamAction.Response innerOperation(
@@ -103,7 +143,8 @@ static GetDataStreamAction.Response innerOperation(
         IndexNameExpressionResolver indexNameExpressionResolver,
         SystemIndices systemIndices,
         ClusterSettings clusterSettings,
-        DataStreamGlobalRetentionSettings globalRetentionSettings
+        DataStreamGlobalRetentionSettings globalRetentionSettings,
+        @Nullable Map<String, Long> maxTimestamps
     ) {
         List<DataStream> dataStreams = getDataStreams(state, indexNameExpressionResolver, request);
         List<GetDataStreamAction.Response.DataStreamInfo> dataStreamInfos = new ArrayList<>(dataStreams.size());
@@ -216,7 +257,8 @@ public int compareTo(IndexInfo o) {
                     ilmPolicyName,
                     timeSeries,
                     backingIndicesSettingsValues,
-                    indexTemplatePreferIlmValue
+                    indexTemplatePreferIlmValue,
+                    maxTimestamps == null ? null : maxTimestamps.get(dataStream.getName())
                 )
             );
         }

modules/data-streams/src/main/java/org/elasticsearch/datastreams/rest/RestGetDataStreamsAction.java

@@ -45,6 +45,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         );
         getDataStreamsRequest.includeDefaults(request.paramAsBoolean("include_defaults", false));
         getDataStreamsRequest.indicesOptions(IndicesOptions.fromRequest(request, getDataStreamsRequest.indicesOptions()));
+        getDataStreamsRequest.verbose(request.paramAsBoolean("verbose", false));
         return channel -> client.execute(GetDataStreamAction.INSTANCE, getDataStreamsRequest, new RestToXContentListener<>(channel));
     }
 
@@ -57,4 +58,19 @@ public boolean allowSystemIndexAccessByDefault() {
     public Set<String> supportedCapabilities() {
         return Set.of(DataStreamLifecycle.EFFECTIVE_RETENTION_REST_API_CAPABILITY);
     }
+
+    @Override
+    public Set<String> supportedQueryParameters() {
+        return Set.of(
+            "name",
+            "include_defaults",
+            "timeout",
+            "master_timeout",
+            IndicesOptions.WildcardOptions.EXPAND_WILDCARDS,
+            IndicesOptions.ConcreteTargetOptions.IGNORE_UNAVAILABLE,
+            IndicesOptions.WildcardOptions.ALLOW_NO_INDICES,
+            IndicesOptions.GatekeeperOptions.IGNORE_THROTTLED,
+            "verbose"
+        );
+    }
 }

modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsRequestTests.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.datastreams.action;
 
 import org.elasticsearch.action.datastreams.GetDataStreamAction.Request;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 
@@ -20,7 +21,7 @@ protected Writeable.Reader<Request> instanceReader() {
 
     @Override
     protected Request createTestInstance() {
-        return new Request(TEST_REQUEST_TIMEOUT, switch (randomIntBetween(1, 4)) {
+        var req = new Request(TEST_REQUEST_TIMEOUT, switch (randomIntBetween(1, 4)) {
             case 1 -> generateRandomStringArray(3, 8, false, false);
             case 2 -> {
                 String[] parameters = generateRandomStringArray(3, 8, false, false);
@@ -32,11 +33,40 @@ protected Request createTestInstance() {
             case 3 -> new String[] { "*" };
             default -> null;
         });
+        req.verbose(randomBoolean());
+        return req;
     }
 
     @Override
     protected Request mutateInstance(Request instance) {
-        return null;// TODO implement https://github.com/elastic/elasticsearch/issues/25929
+        var indices = instance.indices();
+        var indicesOpts = instance.indicesOptions();
+        var includeDefaults = instance.includeDefaults();
+        var verbose = instance.verbose();
+        switch (randomIntBetween(0, 3)) {
+            case 0 -> indices = randomValueOtherThan(indices, () -> generateRandomStringArray(3, 8, false, false));
+            case 1 -> indicesOpts = randomValueOtherThan(
+                indicesOpts,
+                () -> IndicesOptions.fromOptions(
+                    randomBoolean(),
+                    randomBoolean(),
+                    randomBoolean(),
+                    randomBoolean(),
+                    randomBoolean(),
+                    randomBoolean(),
+                    randomBoolean(),
+                    randomBoolean(),
+                    randomBoolean()
+                )
+            );
+            case 2 -> includeDefaults = includeDefaults == false;
+            case 3 -> verbose = verbose == false;
+        }
+        var newReq = new Request(instance.masterNodeTimeout(), indices);
+        newReq.includeDefaults(includeDefaults);
+        newReq.indicesOptions(indicesOpts);
+        newReq.verbose(verbose);
+        return newReq;
     }
 
 }

modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsResponseTests.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentFactory;
@@ -104,7 +105,8 @@ public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Excepti
                 null,
                 null,
                 indexSettingsValues,
-                false
+                false,
+                null
             );
             Response response = new Response(List.of(dataStreamInfo));
             XContentBuilder contentBuilder = XContentFactory.jsonBuilder();
@@ -206,7 +208,8 @@ public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Excepti
                 null,
                 null,
                 indexSettingsValues,
-                false
+                false,
+                null
             );
             Response response = new Response(List.of(dataStreamInfo));
             XContentBuilder contentBuilder = XContentFactory.jsonBuilder();
@@ -283,7 +286,8 @@ private Response.DataStreamInfo mutateInstance(Response.DataStreamInfo instance)
         var timeSeries = instance.getTimeSeries();
         var indexSettings = instance.getIndexSettingsValues();
         var templatePreferIlm = instance.templatePreferIlmValue();
-        switch (randomIntBetween(0, 6)) {
+        var maximumTimestamp = instance.getMaximumTimestamp();
+        switch (randomIntBetween(0, 7)) {
             case 0 -> dataStream = randomValueOtherThan(dataStream, DataStreamTestHelper::randomInstance);
             case 1 -> status = randomValueOtherThan(status, () -> randomFrom(ClusterHealthStatus.values()));
             case 2 -> indexTemplate = randomBoolean() && indexTemplate != null ? null : randomAlphaOfLengthBetween(2, 10);
@@ -305,8 +309,20 @@ private Response.DataStreamInfo mutateInstance(Response.DataStreamInfo instance)
                     )
             );
             case 6 -> templatePreferIlm = templatePreferIlm ? false : true;
+            case 7 -> maximumTimestamp = (maximumTimestamp == null)
+                ? randomNonNegativeLong()
+                : (usually() ? randomValueOtherThan(maximumTimestamp, ESTestCase::randomNonNegativeLong) : null);
         }
-        return new Response.DataStreamInfo(dataStream, status, indexTemplate, ilmPolicyName, timeSeries, indexSettings, templatePreferIlm);
+        return new Response.DataStreamInfo(
+            dataStream,
+            status,
+            indexTemplate,
+            ilmPolicyName,
+            timeSeries,
+            indexSettings,
+            templatePreferIlm,
+            maximumTimestamp
+        );
     }
 
     private List<Tuple<Instant, Instant>> generateRandomTimeSeries() {
@@ -342,7 +358,8 @@ private Response.DataStreamInfo generateRandomDataStreamInfo() {
             randomAlphaOfLengthBetween(2, 10),
             timeSeries != null ? new Response.TimeSeries(timeSeries) : null,
             generateRandomIndexSettingsValues(),
-            randomBoolean()
+            randomBoolean(),
+            usually() ? randomNonNegativeLong() : null
         );
     }
 }