Data stream lifecycle service refactor

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

@@ -47,6 +47,7 @@
 import org.elasticsearch.datastreams.action.TransportPromoteDataStreamAction;
 import org.elasticsearch.datastreams.action.TransportUpdateDataStreamMappingsAction;
 import org.elasticsearch.datastreams.action.TransportUpdateDataStreamSettingsAction;
+import org.elasticsearch.datastreams.lifecycle.AdditionalDataStreamLifecycleActions;
 import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleErrorStore;
 import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService;
 import org.elasticsearch.datastreams.lifecycle.action.DeleteDataStreamLifecycleAction;
@@ -86,6 +87,7 @@
 import org.elasticsearch.health.HealthIndicatorService;
 import org.elasticsearch.index.IndexSettingProvider;
 import org.elasticsearch.plugins.ActionPlugin;
+import org.elasticsearch.plugins.ExtensiblePlugin;
 import org.elasticsearch.plugins.HealthPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
@@ -102,7 +104,7 @@
 
 import static org.elasticsearch.cluster.metadata.DataStreamLifecycle.DATA_STREAM_LIFECYCLE_ORIGIN;
 
-public class DataStreamsPlugin extends Plugin implements ActionPlugin, HealthPlugin {
+public class DataStreamsPlugin extends Plugin implements ActionPlugin, HealthPlugin, ExtensiblePlugin {
 
     public static final Setting<TimeValue> TIME_SERIES_POLL_INTERVAL = Setting.timeSetting(
         "time_series.poll_interval",
@@ -153,6 +155,7 @@ public static TimeValue getLookAheadTime(Settings settings) {
     private final SetOnce<DataStreamLifecycleHealthInfoPublisher> dataStreamLifecycleErrorsPublisher = new SetOnce<>();
     private final SetOnce<DataStreamLifecycleHealthIndicatorService> dataStreamLifecycleHealthIndicatorService = new SetOnce<>();
     private final Settings settings;
+    private AdditionalDataStreamLifecycleActions additionalDataStreamLifecycleActions;
 
     public DataStreamsPlugin(Settings settings) {
         this.settings = settings;
@@ -220,7 +223,8 @@ public Collection<?> createComponents(PluginServices services) {
                 errorStoreInitialisationService.get(),
                 services.allocationService(),
                 dataStreamLifecycleErrorsPublisher.get(),
-                services.dataStreamGlobalRetentionSettings()
+                services.dataStreamGlobalRetentionSettings(),
+                additionalDataStreamLifecycleActions
             )
         );
         dataLifecycleInitialisationService.get().init();
@@ -314,4 +318,15 @@ public void close() throws IOException {
     public Collection<HealthIndicatorService> getHealthIndicatorServices() {
         return List.of(dataStreamLifecycleHealthIndicatorService.get());
     }
+
+    @Override
+    public void loadExtensions(ExtensionLoader loader) {
+        List<AdditionalDataStreamLifecycleActions> dataStreamLifecycleActions = loader.loadExtensions(
+            AdditionalDataStreamLifecycleActions.class
+        );
+        assert dataStreamLifecycleActions.size() <= 1;
+        if (dataStreamLifecycleActions.isEmpty() == false) {
+            this.additionalDataStreamLifecycleActions = dataStreamLifecycleActions.getFirst();
+        }
+    }
 }

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

@@ -0,0 +1,16 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.datastreams.lifecycle;
+
+import java.util.List;
+
+public interface AdditionalDataStreamLifecycleActions {
+    List<DataStreamLifecycleAction> getDataStreamLifecycleActions();
+}

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.datastreams.lifecycle;
+
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.ProjectState;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.index.Index;
+
+import java.util.Set;
+
+@FunctionalInterface
+public interface DataStreamLifecycleAction {
+    /**
+     * This takes some action on the data stream. The action is expected to be fast, or run asynchronously. It returns a set of indices
+     * that ought to be ignored by subsequent actions in the current pass.
+     *
+     * @param projectState                    The current ProjectState
+     * @param dataStream                      The data stream to be acted upon
+     * @param indicesToExcludeForRemainingRun A set of indices that ought to be ignored by this action.
+     */
+    Set<Index> apply(
+        ProjectState projectState,
+        DataStream dataStream,
+        Set<Index> indicesToExcludeForRemainingRun,
+        Client client,
+        DataStreamLifecycleErrorStore errorStore
+    );
+}