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 #108860

Fixes #106461

Author: andreidan

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 @@
 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 @@
  */
 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 APMIndexTemplateRegistry(
         ClusterService clusterService,
         ThreadPool threadPool,
         Client client,
-        NamedXContentRegistry xContentRegistry
+        NamedXContentRegistry xContentRegistry,
+        FeatureService featureService
     ) {
         super(nodeSettings, clusterService, threadPool, client, xContentRegistry);
 
@@ -78,6 +85,7 @@ public APMIndexTemplateRegistry(
                 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 @@ protected String getOrigin() {
         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 Collection<?> createComponents(PluginServices services) {
         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.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 void createRegistryAndClient() {
         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 void createRegistryAndClient() {
             clusterService,
             threadPool,
             client,
-            NamedXContentRegistry.EMPTY
+            NamedXContentRegistry.EMPTY,
+            featureService
         );
         apmIndexTemplateRegistry.setEnabled(true);
     }
@@ -355,6 +358,25 @@ public void testIndexTemplateConventions() throws Exception {
         }
     }
 
+    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();