[Profiling] Allow to upgrade managed ILM policy

docs/changelog/96550.yaml

@@ -0,0 +1,5 @@
+pr: 96550
+summary: "[Profiling] Allow to upgrade managed ILM policy"
+area: Application
+type: enhancement
+issues: []

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

@@ -500,19 +500,20 @@ public void onFailure(Exception e) {
  }
 
  private void addIndexLifecyclePoliciesIfMissing(ClusterState state) {
- Optional<IndexLifecycleMetadata> maybeMeta = Optional.ofNullable(state.metadata().custom(IndexLifecycleMetadata.TYPE));
+ IndexLifecycleMetadata metadata = state.metadata().custom(IndexLifecycleMetadata.TYPE);
  for (LifecyclePolicy policy : getPolicyConfigs()) {
  final AtomicBoolean creationCheck = policyCreationsInProgress.computeIfAbsent(
  policy.getName(),
  key -> new AtomicBoolean(false)
  );
  if (creationCheck.compareAndSet(false, true)) {
- final boolean policyNeedsToBeCreated = maybeMeta.flatMap(
- ilmMeta -> Optional.ofNullable(ilmMeta.getPolicies().get(policy.getName()))
- ).isPresent() == false;
- if (policyNeedsToBeCreated) {
+ final LifecyclePolicy currentPolicy = metadata != null ? metadata.getPolicies().get(policy.getName()) : null;
+ if (Objects.isNull(currentPolicy)) {
  logger.debug("adding lifecycle policy [{}] for [{}], because it doesn't exist", policy.getName(), getOrigin());
  putPolicy(policy, creationCheck);
+ } else if (isUpgradeRequired(currentPolicy, policy)) {
+ logger.info("upgrading lifecycle policy [{}] for [{}]", policy.getName(), getOrigin());
+ putPolicy(policy, creationCheck);
  } else {
  logger.trace("not adding lifecycle policy [{}] for [{}], because it already exists", policy.getName(), getOrigin());
  creationCheck.set(false);
@@ -521,6 +522,17 @@ private void addIndexLifecyclePoliciesIfMissing(ClusterState state) {
  }
  }
 
+ /**
+ * Determines whether an index lifecycle policy should be upgraded to a newer version.
+ *
+ * @param currentPolicy The current lifecycle policy. Never null.
+ * @param newPolicy The new lifecycle policy. Never null.
+ * @return <code>true</code> if <code>newPolicy</code> should replace <code>currentPolicy</code>.
+ */
+ protected boolean isUpgradeRequired(LifecyclePolicy currentPolicy, LifecyclePolicy newPolicy) {
+ return false;
+ }
+
  private void putPolicy(final LifecyclePolicy policy, final AtomicBoolean creationCheck) {
  final Executor executor = threadPool.generic();
  executor.execute(() -> {

x-pack/plugin/core/src/main/resources/org/elasticsearch/xpack/profiler/ilm-policy/profiling-60-days.json

@@ -33,6 +33,7 @@
  },
  "_meta": {
  "description": "default policy for Elastic Universal Profiling",
- "managed": true
+ "managed": true,
+ "version": ${xpack.profiling.template.version}
  }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/template/IndexTemplateRegistryTests.java

@@ -39,14 +39,21 @@
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.core.ilm.DeleteAction;
 import org.elasticsearch.xpack.core.ilm.IndexLifecycleMetadata;
+import org.elasticsearch.xpack.core.ilm.LifecycleAction;
 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.PutLifecycleAction;
 import org.junit.After;
 import org.junit.Before;
 
+import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -92,6 +99,9 @@ public void testThatIndependentPipelinesAreAddedImmediately() throws Exception {
  if (action instanceof PutPipelineAction) {
  assertPutPipelineAction(calledTimes, action, request, listener, "custom-plugin-final_pipeline");
  return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // ignore lifecycle policies in this case
+ return AcknowledgedResponse.TRUE;
  } else {
  // the composable template is not expected to be added, as it's dependency is not available in the cluster state
  // custom-plugin-settings.json is not expected to be added as it contains a dependency on the default_pipeline
@@ -114,6 +124,9 @@ public void testThatDependentPipelinesAreAddedIfDependenciesExist() throws Excep
  if (action instanceof PutPipelineAction) {
  assertPutPipelineAction(calledTimes, action, request, listener, "custom-plugin-default_pipeline");
  return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // ignore lifecycle policies in this case
+ return AcknowledgedResponse.TRUE;
  } else {
  // the composable template is not expected to be added, as it's dependency is not available in the cluster state
  // custom-plugin-settings.json is not expected to be added as it contains a dependency on the default_pipeline
@@ -141,6 +154,9 @@ public void testThatTemplateIsAddedIfAllDependenciesExist() throws Exception {
  if (action instanceof PutComponentTemplateAction) {
  assertPutComponentTemplate(calledTimes, action, request, listener);
  return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // ignore lifecycle policies in this case
+ return AcknowledgedResponse.TRUE;
  } else {
  // the composable template is not expected to be added, as it's dependency is not available in the cluster state
  fail("client called with unexpected request: " + request.toString());
@@ -167,6 +183,9 @@ public void testThatTemplateIsNotAddedIfNotAllDependenciesExist() throws Excepti
  if (action instanceof PutPipelineAction) {
  assertPutPipelineAction(calledTimes, action, request, listener, "custom-plugin-default_pipeline");
  return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // ignore lifecycle policies in this case
+ return AcknowledgedResponse.TRUE;
  } else {
  // the template is not expected to be added, as the final pipeline is missing
  fail("client called with unexpected request: " + request.toString());
@@ -193,6 +212,9 @@ public void testThatComposableTemplateIsAddedIfDependenciesExist() throws Except
  if (action instanceof PutComposableIndexTemplateAction) {
  assertPutComposableIndexTemplateAction(calledTimes, action, request, listener);
  return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // ignore lifecycle policies in this case
+ return AcknowledgedResponse.TRUE;
  } else if (action instanceof PutPipelineAction) {
  // ignore pipelines in this case
  return AcknowledgedResponse.TRUE;
@@ -224,6 +246,9 @@ public void testThatTemplatesAreUpgradedWhenNeeded() throws Exception {
  "custom-plugin-final_pipeline"
  );
  return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // ignore lifecycle policies in this case
+ return AcknowledgedResponse.TRUE;
  } else if (action instanceof PutComponentTemplateAction) {
  assertPutComponentTemplate(calledTimes, action, request, listener);
  return AcknowledgedResponse.TRUE;
@@ -255,6 +280,9 @@ public void testThatTemplatesAreNotUpgradedWhenNotNeeded() throws Exception {
  if (action instanceof PutComposableIndexTemplateAction) {
  // ignore this
  return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ // ignore lifecycle policies in this case
+ return AcknowledgedResponse.TRUE;
  } else {
  fail("client called with unexpected request: " + request.toString());
  return null;
@@ -271,6 +299,173 @@ public void testThatTemplatesAreNotUpgradedWhenNotNeeded() throws Exception {
  assertBusy(() -> assertThat(calledTimes.get(), equalTo(0)));
  }
 
+ public void testThatNonExistingPoliciesAreAddedImmediately() throws Exception {
+ DiscoveryNode node = DiscoveryNodeUtils.create("node");
+ DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
+
+ AtomicInteger calledTimes = new AtomicInteger(0);
+ client.setVerifier((action, request, listener) -> {
+ if (action instanceof PutComposableIndexTemplateAction) {
+ // ignore this
+ return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ assertPutLifecycleAction(calledTimes, action, request, listener);
+ return AcknowledgedResponse.TRUE;
+ } else {
+ fail("client called with unexpected request: " + request.toString());
+ return null;
+ }
+ });
+
+ ClusterChangedEvent event = createClusterChangedEvent(
+ Map.of("custom-plugin-settings", 3),
+ Map.of(),
+ Map.of("custom-plugin-default_pipeline", 3, "custom-plugin-final_pipeline", 3),
+ nodes
+ );
+ registry.clusterChanged(event);
+ assertBusy(() -> assertThat(calledTimes.get(), equalTo(registry.getPolicyConfigs().size())));
+ }
+
+ public void testPolicyAlreadyExists() {
+ DiscoveryNode node = DiscoveryNodeUtils.create("node");
+ DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
+
+ Map<String, LifecyclePolicy> policyMap = new HashMap<>();
+ List<LifecyclePolicy> policies = registry.getPolicyConfigs();
+ assertThat(policies, hasSize(1));
+ policies.forEach(p -> policyMap.put(p.getName(), p));
+
+ client.setVerifier((action, request, listener) -> {
+ if (action instanceof PutComposableIndexTemplateAction) {
+ // ignore this
+ return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ fail("if the policy already exists it should not be re-put");
+ } else {
+ fail("client called with unexpected request: " + request.toString());
+ }
+ return null;
+ });
+
+ ClusterChangedEvent event = createClusterChangedEvent(
+ Map.of("custom-plugin-settings", 3),
+ policyMap,
+ Map.of("custom-plugin-default_pipeline", 3, "custom-plugin-final_pipeline", 3),
+ nodes
+ );
+
+ registry.clusterChanged(event);
+ }
+
+ public void testPolicyAlreadyExistsButDiffers() throws IOException {
+ DiscoveryNode node = DiscoveryNodeUtils.create("node");
+ DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
+
+ Map<String, LifecyclePolicy> policyMap = new HashMap<>();
+ String policyStr = "{\"phases\":{\"delete\":{\"min_age\":\"1m\",\"actions\":{\"delete\":{}}}}}";
+ List<LifecyclePolicy> policies = registry.getPolicyConfigs();
+ assertThat(policies, hasSize(1));
+ policies.forEach(p -> policyMap.put(p.getName(), p));
+
+ client.setVerifier((action, request, listener) -> {
+ if (action instanceof PutComposableIndexTemplateAction) {
+ // ignore this
+ return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ fail("if the policy already exists it should not be re-put");
+ } else {
+ fail("client called with unexpected request: " + request.toString());
+ }
+ return null;
+ });
+
+ try (
+ XContentParser parser = XContentType.JSON.xContent()
+ .createParser(
+ XContentParserConfiguration.EMPTY.withRegistry(
+ new NamedXContentRegistry(
+ List.of(
+ new NamedXContentRegistry.Entry(
+ LifecycleAction.class,
+ new ParseField(DeleteAction.NAME),
+ DeleteAction::parse
+ )
+ )
+ )
+ ),
+ policyStr
+ )
+ ) {
+ LifecyclePolicy different = LifecyclePolicy.parse(parser, policies.get(0).getName());
+ policyMap.put(policies.get(0).getName(), different);
+ ClusterChangedEvent event = createClusterChangedEvent(
+ Map.of("custom-plugin-settings", 3),
+ policyMap,
+ Map.of("custom-plugin-default_pipeline", 3, "custom-plugin-final_pipeline", 3),
+ nodes
+ );
+ registry.clusterChanged(event);
+ }
+ }
+
+ public void testPolicyUpgraded() throws Exception {
+ registry.setPolicyUpgradeRequired(true);
+ DiscoveryNode node = DiscoveryNodeUtils.create("node");
+ DiscoveryNodes nodes = DiscoveryNodes.builder().localNodeId("node").masterNodeId("node").add(node).build();
+
+ Map<String, LifecyclePolicy> policyMap = new HashMap<>();
+ String priorPolicyStr = "{\"phases\":{\"delete\":{\"min_age\":\"1m\",\"actions\":{\"delete\":{}}}}}";
+ List<LifecyclePolicy> policies = registry.getPolicyConfigs();
+ assertThat(policies, hasSize(1));
+ policies.forEach(p -> policyMap.put(p.getName(), p));
+
+ AtomicInteger calledTimes = new AtomicInteger(0);
+ client.setVerifier((action, request, listener) -> {
+ if (action instanceof PutComposableIndexTemplateAction) {
+ // ignore this
+ return AcknowledgedResponse.TRUE;
+ } else if (action instanceof PutLifecycleAction) {
+ assertPutLifecycleAction(calledTimes, action, request, listener);
+ return AcknowledgedResponse.TRUE;
+
+ } else {
+ fail("client called with unexpected request: " + request.toString());
+ }
+ return null;
+ });
+
+ try (
+ XContentParser parser = XContentType.JSON.xContent()
+ .createParser(
+ XContentParserConfiguration.EMPTY.withRegistry(
+ new NamedXContentRegistry(
+ List.of(
+ new NamedXContentRegistry.Entry(
+ LifecycleAction.class,
+ new ParseField(DeleteAction.NAME),
+ DeleteAction::parse
+ )
+ )
+ )
+ ),
+ priorPolicyStr
+ )
+ ) {
+ LifecyclePolicy priorPolicy = LifecyclePolicy.parse(parser, policies.get(0).getName());
+ policyMap.put(policies.get(0).getName(), priorPolicy);
+ ClusterChangedEvent event = createClusterChangedEvent(
+ Map.of("custom-plugin-settings", 3),
+ policyMap,
+ Map.of("custom-plugin-default_pipeline", 3, "custom-plugin-final_pipeline", 3),
+ nodes
+ );
+ registry.clusterChanged(event);
+ // we've changed one policy that should be upgraded
+ assertBusy(() -> assertThat(calledTimes.get(), equalTo(1)));
+ }
+ }
+
  private static void assertPutComponentTemplate(
  AtomicInteger calledTimes,
  ActionType<?> action,
@@ -332,6 +527,20 @@ private static void assertPutPipelineAction(
  calledTimes.incrementAndGet();
  }
 
+ private static void assertPutLifecycleAction(
+ AtomicInteger calledTimes,
+ ActionType<?> action,
+ ActionRequest request,
+ ActionListener<?> listener
+ ) {
+ assertThat(action, instanceOf(PutLifecycleAction.class));
+ assertThat(request, instanceOf(PutLifecycleAction.Request.class));
+ final PutLifecycleAction.Request putRequest = (PutLifecycleAction.Request) request;
+ assertThat(putRequest.getPolicy().getName(), equalTo("custom-plugin-policy"));
+ assertNotNull(listener);
+ calledTimes.incrementAndGet();
+ }
+
  private ClusterChangedEvent createClusterChangedEvent(Map<String, Integer> existingTemplates, DiscoveryNodes nodes) {
  return createClusterChangedEvent(existingTemplates, Collections.emptyMap(), Collections.emptyMap(), nodes);
  }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/template/TestRegistryWithCustomPlugin.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.json.JsonXContent;
+import org.elasticsearch.xpack.core.ilm.LifecyclePolicy;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -27,6 +28,8 @@ class TestRegistryWithCustomPlugin extends IndexTemplateRegistry {
  public static final int REGISTRY_VERSION = 3;
  public static final String TEMPLATE_VERSION_VARIABLE = "xpack.custom_plugin.template.version";
 
+ private boolean policyUpgradeRequired = false;
+
  TestRegistryWithCustomPlugin(
  Settings nodeSettings,
  ClusterService clusterService,
@@ -88,6 +91,24 @@ protected List<IngestPipelineConfig> getIngestPipelines() {
  );
  }
 
+ @Override
+ protected List<LifecyclePolicy> getPolicyConfigs() {
+ return List.of(
+ new LifecyclePolicyConfig("custom-plugin-policy", "/org/elasticsearch/xpack/core/template/custom-plugin-policy.json").load(
+ LifecyclePolicyConfig.DEFAULT_X_CONTENT_REGISTRY
+ )
+ );
+ }
+
+ @Override
+ protected boolean isUpgradeRequired(LifecyclePolicy currentPolicy, LifecyclePolicy newPolicy) {
+ return policyUpgradeRequired;
+ }
+
+ public void setPolicyUpgradeRequired(boolean policyUpgradeRequired) {
+ this.policyUpgradeRequired = policyUpgradeRequired;
+ }
+
  @Override
  protected String getOrigin() {
  return "test";