Install the apm templates only if DSL available (#109166)

The APM module installs component and index templates that make use of
data stream lifecycle. This makes sure the templates are installed only
once the cluster has the data stream lifecycle feature.

Follow-up to https://github.com/elastic/elasticsearch/pull/108860

Fixes #106461

x-pack/plugin/apm-data/build.gradle

@@ -20,6 +20,7 @@ dependencies {
   compileOnly project(path: xpackModule('core'))
   testImplementation project(path: ':x-pack:plugin:stack')
   testImplementation(testArtifact(project(xpackModule('core'))))
+  testImplementation project(':modules:data-streams')
   clusterModules project(':modules:data-streams')
   clusterModules project(':modules:ingest-common')
   clusterModules project(':modules:ingest-geoip')

x-pack/plugin/apm-data/src/main/java/org/elasticsearch/xpack/apmdata/APMIndexTemplateRegistry.java

@@ -10,12 +10,15 @@ package org.elasticsearch.xpack.apmdata;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.metadata.ComponentTemplate;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.features.FeatureService;
+import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
@@ -39,12 +42,15 @@ import static org.elasticsearch.xpack.apmdata.ResourceUtils.loadVersionedResourc
  */
 public class APMIndexTemplateRegistry extends IndexTemplateRegistry {
     private static final Logger logger = LogManager.getLogger(APMIndexTemplateRegistry.class);
-
+    // this node feature is a redefinition of {@link DataStreamFeatures#DATA_STREAM_LIFECYCLE} and it's meant to avoid adding a
+    // dependency to the data-streams module just for this
+    public static final NodeFeature DATA_STREAM_LIFECYCLE = new NodeFeature("data_stream.lifecycle");
     private final int version;
 
     private final Map<String, ComponentTemplate> componentTemplates;
     private final Map<String, ComposableIndexTemplate> composableIndexTemplates;
     private final List<IngestPipelineConfig> ingestPipelines;
+    private final FeatureService featureService;
     private volatile boolean enabled;
 
     @SuppressWarnings("unchecked")
@@ -53,7 +59,8 @@ public class APMIndexTemplateRegistry extends IndexTemplateRegistry {
         ClusterService clusterService,
         ThreadPool threadPool,
         Client client,
-        NamedXContentRegistry xContentRegistry
+        NamedXContentRegistry xContentRegistry,
+        FeatureService featureService
     ) {
         super(nodeSettings, clusterService, threadPool, client, xContentRegistry);
 
@@ -78,6 +85,7 @@ public class APMIndexTemplateRegistry extends IndexTemplateRegistry {
                 Map.Entry<String, Map<String, Object>> pipelineConfig = map.entrySet().iterator().next();
                 return loadIngestPipeline(pipelineConfig.getKey(), version, (List<String>) pipelineConfig.getValue().get("dependencies"));
             }).collect(Collectors.toList());
+            this.featureService = featureService;
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
@@ -105,6 +113,13 @@ public class APMIndexTemplateRegistry extends IndexTemplateRegistry {
         return ClientHelper.APM_ORIGIN;
     }
 
+    @Override
+    protected boolean isClusterReady(ClusterChangedEvent event) {
+        // Ensure current version of the components are installed only after versions that support data stream lifecycle
+        // due to the use of the feature in all the `@lifecycle` component templates
+        return featureService.clusterHasFeature(event.state(), DATA_STREAM_LIFECYCLE);
+    }
+
     @Override
     protected boolean requiresMasterNode() {
         return true;

x-pack/plugin/apm-data/src/main/java/org/elasticsearch/xpack/apmdata/APMPlugin.java

@@ -48,7 +48,14 @@ public class APMPlugin extends Plugin implements ActionPlugin {
         Settings settings = services.environment().settings();
         ClusterService clusterService = services.clusterService();
         registry.set(
-            new APMIndexTemplateRegistry(settings, clusterService, services.threadPool(), services.client(), services.xContentRegistry())
+            new APMIndexTemplateRegistry(
+                settings,
+                clusterService,
+                services.threadPool(),
+                services.client(),
+                services.xContentRegistry(),
+                services.featureService()
+            )
         );
         if (enabled) {
             APMIndexTemplateRegistry registryInstance = registry.get();

x-pack/plugin/apm-data/src/test/java/org/elasticsearch/xpack/apmdata/APMIndexTemplateRegistryTests.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.apmdata;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
@@ -29,6 +30,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.datastreams.DataStreamFeatures;
 import org.elasticsearch.features.FeatureService;
 import org.elasticsearch.ingest.IngestMetadata;
 import org.elasticsearch.ingest.PipelineConfiguration;
@@ -88,7 +90,7 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         threadPool = new TestThreadPool(this.getClass().getName());
         client = new VerifyingClient(threadPool);
         clusterService = ClusterServiceUtils.createClusterService(threadPool, clusterSettings);
-        FeatureService featureService = new FeatureService(List.of());
+        FeatureService featureService = new FeatureService(List.of(new DataStreamFeatures()));
         stackTemplateRegistryAccessor = new StackTemplateRegistryAccessor(
             new StackTemplateRegistry(Settings.EMPTY, clusterService, threadPool, client, NamedXContentRegistry.EMPTY, featureService)
         );
@@ -98,7 +100,8 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
             clusterService,
             threadPool,
             client,
-            NamedXContentRegistry.EMPTY
+            NamedXContentRegistry.EMPTY,
+            featureService
         );
         apmIndexTemplateRegistry.setEnabled(true);
     }
@@ -355,6 +358,25 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         }
     }
 
+    public void testThatNothingIsInstalledWhenAllNodesAreNotUpdated() {
+        DiscoveryNode updatedNode = DiscoveryNodeUtils.create("updatedNode");
+        DiscoveryNode outdatedNode = DiscoveryNodeUtils.create("outdatedNode", ESTestCase.buildNewFakeTransportAddress(), Version.V_8_10_0);
+        DiscoveryNodes nodes = DiscoveryNodes.builder()
+            .localNodeId("updatedNode")
+            .masterNodeId("updatedNode")
+            .add(updatedNode)
+            .add(outdatedNode)
+            .build();
+
+        client.setVerifier((a, r, l) -> {
+            fail("if some cluster mode are not updated to at least v.8.11.0 nothing should happen");
+            return null;
+        });
+
+        ClusterChangedEvent event = createClusterChangedEvent(Map.of(), Map.of(), nodes);
+        apmIndexTemplateRegistry.clusterChanged(event);
+    }
+
     private Map<String, ComponentTemplate> getIndependentComponentTemplateConfigs() {
         return apmIndexTemplateRegistry.getComponentTemplateConfigs().entrySet().stream().filter(template -> {
             Settings settings = template.getValue().template().settings();