Add default ILM policies and switch to ILM for apm-data plugin (#115687) (#116742)

(cherry picked from commit 6325e46231c15da744b3ad28811279b03f1299d0)

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

docs/changelog/115687.yaml

@@ -0,0 +1,5 @@
+pr: 115687
+summary: Add default ILM policies and switch to ILM for apm-data plugin
+area: Data streams
+type: feature
+issues: []

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -820,7 +820,26 @@ public abstract class ESRestTestCase extends ESTestCase {
             ".fleet-file-tohost-meta-ilm-policy",
             ".deprecation-indexing-ilm-policy",
             ".monitoring-8-ilm-policy",
-            "behavioral_analytics-events-default_policy"
+            "behavioral_analytics-events-default_policy",
+            "logs-apm.app_logs-default_policy",
+            "logs-apm.error_logs-default_policy",
+            "metrics-apm.app_metrics-default_policy",
+            "metrics-apm.internal_metrics-default_policy",
+            "metrics-apm.service_destination_10m_metrics-default_policy",
+            "metrics-apm.service_destination_1m_metrics-default_policy",
+            "metrics-apm.service_destination_60m_metrics-default_policy",
+            "metrics-apm.service_summary_10m_metrics-default_policy",
+            "metrics-apm.service_summary_1m_metrics-default_policy",
+            "metrics-apm.service_summary_60m_metrics-default_policy",
+            "metrics-apm.service_transaction_10m_metrics-default_policy",
+            "metrics-apm.service_transaction_1m_metrics-default_policy",
+            "metrics-apm.service_transaction_60m_metrics-default_policy",
+            "metrics-apm.transaction_10m_metrics-default_policy",
+            "metrics-apm.transaction_1m_metrics-default_policy",
+            "metrics-apm.transaction_60m_metrics-default_policy",
+            "traces-apm.rum_traces-default_policy",
+            "traces-apm.sampled_traces-default_policy",
+            "traces-apm.traces-default_policy"
         );
     }
 

x-pack/plugin/apm-data/src/main/resources/component-templates/logs-apm.app-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: logs-apm.app_logs-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/logs-apm.error-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: logs-apm.error_logs-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.app-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.app_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.internal-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.internal_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_destination.10m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_destination_10m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_destination.1m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_destination_1m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_destination.60m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_destination_60m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_summary.10m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_summary_10m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_summary.1m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_summary_1m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_summary.60m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_summary_60m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_transaction.10m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_transaction_10m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_transaction.1m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_transaction_1m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.service_transaction.60m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.service_transaction_60m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.transaction.10m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.transaction_10m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.transaction.1m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.transaction_1m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/metrics-apm.transaction.60m-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: metrics-apm.transaction_60m_metrics-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/traces-apm-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: traces-apm.traces-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/traces-apm.rum-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: traces-apm.rum_traces-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/component-templates/traces-apm.sampled-fallback@ilm.yaml

@@ -8,4 +8,3 @@ template:
     index:
       lifecycle:
         name: traces-apm.sampled_traces-default_policy
-        prefer_ilm: false

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/logs-apm.app_logs-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 10d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/logs-apm.error_logs-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 10d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.app_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 90d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.internal_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 90d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_destination_10m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 14d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 180d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_destination_1m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 7d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 90d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_destination_60m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 390d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_summary_10m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 14d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 180d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_summary_1m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 7d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 90d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_summary_60m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 390d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_transaction_10m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 14d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 180d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_transaction_1m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 7d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 90d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.service_transaction_60m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 390d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.transaction_10m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 14d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 180d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.transaction_1m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 7d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 90d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/metrics-apm.transaction_60m_metrics-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 390d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/traces-apm.rum_traces-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 90d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/traces-apm.sampled_traces-default_policy.yaml

@@ -0,0 +1,13 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 1h
+  delete:
+    min_age: 1h
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/lifecycle-policies/traces-apm.traces-default_policy.yaml

@@ -0,0 +1,16 @@
+---
+_meta:
+  description: Default ILM policy for APM managed datastreams
+  managed: true
+phases:
+  hot:
+    actions:
+      rollover:
+        max_age: 30d
+        max_primary_shard_size: 50gb
+      set_priority:
+        priority: 100
+  delete:
+    min_age: 10d
+    actions:
+      delete: {}

x-pack/plugin/apm-data/src/main/resources/resources.yaml

@@ -1,7 +1,7 @@
 # "version" holds the version of the templates and ingest pipelines installed
 # by xpack-plugin apm-data. This must be increased whenever an existing template or
 # pipeline is changed, in order for it to be updated on Elasticsearch upgrade.
-version: 10
+version: 11
 
 component-templates:
   # Data lifecycle.
@@ -97,3 +97,24 @@ ingest-pipelines:
   - metrics-apm@pipeline:
       dependencies:
         - apm@pipeline
+
+lifecycle-policies:
+  - logs-apm.app_logs-default_policy
+  - logs-apm.error_logs-default_policy
+  - metrics-apm.app_metrics-default_policy
+  - metrics-apm.internal_metrics-default_policy
+  - metrics-apm.service_destination_10m_metrics-default_policy
+  - metrics-apm.service_destination_1m_metrics-default_policy
+  - metrics-apm.service_destination_60m_metrics-default_policy
+  - metrics-apm.service_summary_10m_metrics-default_policy
+  - metrics-apm.service_summary_1m_metrics-default_policy
+  - metrics-apm.service_summary_60m_metrics-default_policy
+  - metrics-apm.service_transaction_10m_metrics-default_policy
+  - metrics-apm.service_transaction_1m_metrics-default_policy
+  - metrics-apm.service_transaction_60m_metrics-default_policy
+  - metrics-apm.transaction_10m_metrics-default_policy
+  - metrics-apm.transaction_1m_metrics-default_policy
+  - metrics-apm.transaction_60m_metrics-default_policy
+  - traces-apm.rum_traces-default_policy
+  - traces-apm.sampled_traces-default_policy
+  - traces-apm.traces-default_policy

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

@@ -44,6 +44,8 @@ import org.elasticsearch.xpack.core.ilm.IndexLifecycleMetadata;
 import org.elasticsearch.xpack.core.ilm.LifecyclePolicy;
 import org.elasticsearch.xpack.core.ilm.LifecyclePolicyMetadata;
 import org.elasticsearch.xpack.core.ilm.OperationMode;
+import org.elasticsearch.xpack.core.ilm.action.ILMActions;
+import org.elasticsearch.xpack.core.ilm.action.PutLifecycleRequest;
 import org.elasticsearch.xpack.core.template.IngestPipelineConfig;
 import org.elasticsearch.xpack.stack.StackTemplateRegistry;
 import org.elasticsearch.xpack.stack.StackTemplateRegistryAccessor;
@@ -57,6 +59,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -133,6 +136,7 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         assertThat(apmIndexTemplateRegistry.getComponentTemplateConfigs().entrySet(), hasSize(0));
         assertThat(apmIndexTemplateRegistry.getComposableTemplateConfigs().entrySet(), hasSize(0));
         assertThat(apmIndexTemplateRegistry.getIngestPipelines(), hasSize(0));
+        assertThat(apmIndexTemplateRegistry.getLifecyclePolicies(), hasSize(0));
 
         client.setVerifier((a, r, l) -> {
             fail("if the registry is disabled nothing should happen");
@@ -145,6 +149,7 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         assertThat(apmIndexTemplateRegistry.getComponentTemplateConfigs().entrySet(), not(hasSize(0)));
         assertThat(apmIndexTemplateRegistry.getComposableTemplateConfigs().entrySet(), not(hasSize(0)));
         assertThat(apmIndexTemplateRegistry.getIngestPipelines(), not(hasSize(0)));
+        assertThat(apmIndexTemplateRegistry.getLifecyclePolicies(), not(hasSize(0)));
     }
 
     public void testThatIndependentTemplatesAreAddedImmediatelyIfMissing() throws Exception {
@@ -154,23 +159,26 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         AtomicInteger actualInstalledIndexTemplates = new AtomicInteger(0);
         AtomicInteger actualInstalledComponentTemplates = new AtomicInteger(0);
         AtomicInteger actualInstalledIngestPipelines = new AtomicInteger(0);
+        AtomicInteger actualILMPolicies = new AtomicInteger(0);
 
         client.setVerifier(
             (action, request, listener) -> verifyActions(
                 actualInstalledIndexTemplates,
                 actualInstalledComponentTemplates,
                 actualInstalledIngestPipelines,
+                actualILMPolicies,
                 action,
                 request,
                 listener
             )
         );
-        apmIndexTemplateRegistry.clusterChanged(createClusterChangedEvent(Map.of(), Map.of(), nodes));
+        apmIndexTemplateRegistry.clusterChanged(createClusterChangedEvent(Map.of(), Map.of(), List.of(), Map.of(), nodes));
 
         assertBusy(() -> assertThat(actualInstalledIngestPipelines.get(), equalTo(getIndependentPipelineConfigs().size())));
         assertBusy(() -> assertThat(actualInstalledComponentTemplates.get(), equalTo(getIndependentComponentTemplateConfigs().size())));
+        assertBusy(() -> assertThat(actualILMPolicies.get(), equalTo(getIndependentLifecyclePolicies().size())));
 
-        // index templates should not be installed as they are dependent in component templates and ingest pipelines
+        // index templates should not be installed as they are dependent on component templates and ingest pipelines
         assertThat(actualInstalledIndexTemplates.get(), equalTo(0));
     }
 
@@ -201,6 +209,31 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         });
     }
 
+    public void testILMLifecyclePolicies() throws Exception {
+        DiscoveryNode node = DiscoveryNodeUtils.create("node");
+        DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
+
+        final List<LifecyclePolicy> lifecyclePolicies = apmIndexTemplateRegistry.getLifecyclePolicies();
+        assertThat(lifecyclePolicies, is(not(empty())));
+
+        final Set<String> expectedILMPolicies = apmIndexTemplateRegistry.getLifecyclePolicies()
+            .stream()
+            .map(LifecyclePolicy::getName)
+            .collect(Collectors.toSet());
+        final Set<String> installedILMPolicies = ConcurrentHashMap.newKeySet(lifecyclePolicies.size());
+        client.setVerifier((a, r, l) -> {
+            if (a == ILMActions.PUT && r instanceof PutLifecycleRequest putLifecycleRequest) {
+                if (expectedILMPolicies.contains(putLifecycleRequest.getPolicy().getName())) {
+                    installedILMPolicies.add(putLifecycleRequest.getPolicy().getName());
+                }
+            }
+            return AcknowledgedResponse.TRUE;
+        });
+
+        apmIndexTemplateRegistry.clusterChanged(createClusterChangedEvent(Map.of(), Map.of(), List.of(), Map.of(), nodes));
+        assertBusy(() -> { assertThat(installedILMPolicies, equalTo(expectedILMPolicies)); });
+    }
+
     public void testComponentTemplates() throws Exception {
         DiscoveryNode node = DiscoveryNodeUtils.create("node");
         DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
@@ -208,12 +241,14 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         AtomicInteger actualInstalledIndexTemplates = new AtomicInteger(0);
         AtomicInteger actualInstalledComponentTemplates = new AtomicInteger(0);
         AtomicInteger actualInstalledIngestPipelines = new AtomicInteger(0);
+        AtomicInteger actualILMPolicies = new AtomicInteger(0);
 
         client.setVerifier(
             (action, request, listener) -> verifyActions(
                 actualInstalledIndexTemplates,
                 actualInstalledComponentTemplates,
                 actualInstalledIngestPipelines,
+                actualILMPolicies,
                 action,
                 request,
                 listener
@@ -224,6 +259,9 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
                 Map.of(),
                 Map.of(),
                 apmIndexTemplateRegistry.getIngestPipelines().stream().map(IngestPipelineConfig::getId).collect(Collectors.toList()),
+                apmIndexTemplateRegistry.getLifecyclePolicies()
+                    .stream()
+                    .collect(Collectors.toMap(LifecyclePolicy::getName, Function.identity())),
                 nodes
             )
         );
@@ -237,8 +275,10 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
 
         // ingest pipelines should not have been installed as we used a cluster state that includes them already
         assertThat(actualInstalledIngestPipelines.get(), equalTo(0));
-        // index templates should not be installed as they are dependent in component templates and ingest pipelines
+        // index templates should not be installed as they are dependent on component templates and ingest pipelines
         assertThat(actualInstalledIndexTemplates.get(), equalTo(0));
+        // ilm policies should not have been installed as we used a cluster state that includes them already
+        assertThat(actualILMPolicies.get(), equalTo(0));
     }
 
     public void testIndexTemplates() throws Exception {
@@ -248,12 +288,14 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         AtomicInteger actualInstalledIndexTemplates = new AtomicInteger(0);
         AtomicInteger actualInstalledComponentTemplates = new AtomicInteger(0);
         AtomicInteger actualInstalledIngestPipelines = new AtomicInteger(0);
+        AtomicInteger actualILMPolicies = new AtomicInteger(0);
 
         client.setVerifier(
             (action, request, listener) -> verifyActions(
                 actualInstalledIndexTemplates,
                 actualInstalledComponentTemplates,
                 actualInstalledIngestPipelines,
+                actualILMPolicies,
                 action,
                 request,
                 listener
@@ -272,6 +314,9 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
                 componentTemplates,
                 Map.of(),
                 apmIndexTemplateRegistry.getIngestPipelines().stream().map(IngestPipelineConfig::getId).collect(Collectors.toList()),
+                apmIndexTemplateRegistry.getLifecyclePolicies()
+                    .stream()
+                    .collect(Collectors.toMap(LifecyclePolicy::getName, Function.identity())),
                 nodes
             )
         );
@@ -280,9 +325,11 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
             () -> assertThat(actualInstalledIndexTemplates.get(), equalTo(apmIndexTemplateRegistry.getComposableTemplateConfigs().size()))
         );
 
-        // ingest pipelines and component templates should not have been installed as we used a cluster state that includes them already
+        // ingest pipelines, component templates, and lifecycle policies should not have been installed as we used a cluster state that
+        // includes them already
         assertThat(actualInstalledComponentTemplates.get(), equalTo(0));
         assertThat(actualInstalledIngestPipelines.get(), equalTo(0));
+        assertThat(actualILMPolicies.get(), equalTo(0));
     }
 
     public void testIndexTemplateConventions() throws Exception {
@@ -408,10 +455,18 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
             .collect(Collectors.toList());
     }
 
+    private Map<String, LifecyclePolicy> getIndependentLifecyclePolicies() {
+        // All lifecycle policies are independent
+        return apmIndexTemplateRegistry.getLifecyclePolicies()
+            .stream()
+            .collect(Collectors.toMap(LifecyclePolicy::getName, Function.identity()));
+    }
+
     private ActionResponse verifyActions(
         AtomicInteger indexTemplatesCounter,
         AtomicInteger componentTemplatesCounter,
         AtomicInteger ingestPipelinesCounter,
+        AtomicInteger ilmPolicyCounter,
         ActionType<?> action,
         ActionRequest request,
         ActionListener<?> listener
@@ -430,6 +485,9 @@ public class APMIndexTemplateRegistryTests extends ESTestCase {
         } else if (action == PutPipelineTransportAction.TYPE) {
             ingestPipelinesCounter.incrementAndGet();
             return AcknowledgedResponse.TRUE;
+        } else if (action == ILMActions.PUT) {
+            ilmPolicyCounter.incrementAndGet();
+            return AcknowledgedResponse.TRUE;
         } else {
             fail("client called with unexpected request:" + request.toString());
             return null;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/LifecyclePolicyUtils.java

@@ -23,6 +23,7 @@ import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.core.template.resources.TemplateResources;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -48,19 +49,32 @@ public class LifecyclePolicyUtils {
             source = replaceVariables(source, variables);
             validate(source);
 
-            try (
-                XContentParser parser = XContentType.JSON.xContent()
-                    .createParser(XContentParserConfiguration.EMPTY.withRegistry(xContentRegistry), source)
-            ) {
-                LifecyclePolicy policy = LifecyclePolicy.parse(parser, name);
-                policy.validate();
-                return policy;
-            }
+            return parsePolicy(source, name, xContentRegistry, XContentType.JSON);
         } catch (Exception e) {
             throw new IllegalArgumentException("unable to load policy [" + name + "] from [" + resource + "]", e);
         }
     }
 
+    /**
+     * Parses lifecycle policy based on the provided content type without doing any variable substitution.
+     * It is caller's responsibility to do any variable substitution if required.
+     */
+    public static LifecyclePolicy parsePolicy(
+        String rawPolicy,
+        String name,
+        NamedXContentRegistry xContentRegistry,
+        XContentType contentType
+    ) throws IOException {
+        try (
+            XContentParser parser = contentType.xContent()
+                .createParser(XContentParserConfiguration.EMPTY.withRegistry(xContentRegistry), rawPolicy)
+        ) {
+            LifecyclePolicy policy = LifecyclePolicy.parse(parser, name);
+            policy.validate();
+            return policy;
+        }
+    }
+
     private static String replaceVariables(String template, Map<String, String> variables) {
         for (Map.Entry<String, String> variable : variables.entrySet()) {
             template = replaceVariable(template, variable.getKey(), variable.getValue());

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/template/YamlTemplateRegistry.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.core.template;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.metadata.ComponentTemplate;
@@ -22,7 +23,10 @@ import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
+import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xcontent.yaml.YamlXContent;
+import org.elasticsearch.xpack.core.ilm.LifecyclePolicy;
+import org.elasticsearch.xpack.core.ilm.LifecyclePolicyUtils;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -48,6 +52,7 @@ public abstract class YamlTemplateRegistry extends IndexTemplateRegistry {
     private final Map<String, ComponentTemplate> componentTemplates;
     private final Map<String, ComposableIndexTemplate> composableIndexTemplates;
     private final List<IngestPipelineConfig> ingestPipelines;
+    private final List<LifecyclePolicy> lifecyclePolicies;
     private final FeatureService featureService;
     private volatile boolean enabled;
 
@@ -84,6 +89,7 @@ public abstract class YamlTemplateRegistry extends IndexTemplateRegistry {
             final List<Object> componentTemplateNames = (List<Object>) resources.get("component-templates");
             final List<Object> indexTemplateNames = (List<Object>) resources.get("index-templates");
             final List<Object> ingestPipelineConfigs = (List<Object>) resources.get("ingest-pipelines");
+            final List<Object> lifecyclePolicyConfigs = (List<Object>) resources.get("lifecycle-policies");
 
             componentTemplates = Optional.ofNullable(componentTemplateNames)
                 .orElse(Collections.emptyList())
@@ -110,9 +116,16 @@ public abstract class YamlTemplateRegistry extends IndexTemplateRegistry {
                     );
                 })
                 .collect(Collectors.toList());
+            lifecyclePolicies = Optional.ofNullable(lifecyclePolicyConfigs)
+                .orElse(Collections.emptyList())
+                .stream()
+                .map(o -> (String) o)
+                .filter(templateFilter)
+                .map(this::loadLifecyclePolicy)
+                .collect(Collectors.toList());
             this.featureService = featureService;
         } catch (IOException e) {
-            throw new RuntimeException(e);
+            throw new ElasticsearchException(e);
         }
     }
 
@@ -178,6 +191,15 @@ public abstract class YamlTemplateRegistry extends IndexTemplateRegistry {
         }
     }
 
+    @Override
+    public List<LifecyclePolicy> getLifecyclePolicies() {
+        if (enabled) {
+            return lifecyclePolicies;
+        } else {
+            return Collections.emptyList();
+        }
+    }
+
     protected abstract String getVersionProperty();
 
     private ComponentTemplate loadComponentTemplate(String name, int version) {
@@ -192,7 +214,7 @@ public abstract class YamlTemplateRegistry extends IndexTemplateRegistry {
                 return ComponentTemplate.parse(parser);
             }
         } catch (Exception e) {
-            throw new RuntimeException("failed to load " + getName() + " Ingest plugin's component template: " + name, e);
+            throw new ElasticsearchException("failed to load " + getName() + " Ingest plugin's component template: " + name, e);
         }
     }
 
@@ -208,7 +230,7 @@ public abstract class YamlTemplateRegistry extends IndexTemplateRegistry {
                 return ComposableIndexTemplate.parse(parser);
             }
         } catch (Exception e) {
-            throw new RuntimeException("failed to load " + getName() + " Ingest plugin's index template: " + name, e);
+            throw new ElasticsearchException("failed to load " + getName() + " Ingest plugin's index template: " + name, e);
         }
     }
 
@@ -226,6 +248,17 @@ public abstract class YamlTemplateRegistry extends IndexTemplateRegistry {
         );
     }
 
+    // IndexTemplateRegistry ensures that ILM lifecycle policies are not loaded
+    // when in DSL only mode.
+    private LifecyclePolicy loadLifecyclePolicy(String name) {
+        try {
+            var rawPolicy = loadResource(this.getClass(), "/lifecycle-policies/" + name + ".yaml");
+            return LifecyclePolicyUtils.parsePolicy(rawPolicy, name, LifecyclePolicyConfig.DEFAULT_X_CONTENT_REGISTRY, XContentType.YAML);
+        } catch (IOException e) {
+            throw new ElasticsearchException(e);
+        }
+    }
+
     @Override
     protected boolean applyRolloverAfterTemplateV2Update() {
         return true;