Include SLM policy name in Snapshot metadata (#43132)

Keep track of which SLM policy in the metadata field of the Snapshots
taken by SLM. This allows users to more easily understand where the
snapshot came from, and will enable future SLM features such as
retention policies.

Author: gwbrown

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/create/CreateSnapshotRequest.java

@@ -163,7 +163,7 @@ public ActionRequestValidationException validate() {
         return validationException;
     }
 
-    private static int metadataSize(Map<String, Object> userMetadata) {
+    public static int metadataSize(Map<String, Object> userMetadata) {
         if (userMetadata == null) {
             return 0;
         }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/snapshotlifecycle/SnapshotLifecyclePolicy.java

@@ -30,6 +30,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -57,6 +58,8 @@ public class SnapshotLifecyclePolicy extends AbstractDiffable<SnapshotLifecycleP
     private static final ParseField CONFIG = new ParseField("config");
     private static final IndexNameExpressionResolver.DateMathExpressionResolver DATE_MATH_RESOLVER =
         new IndexNameExpressionResolver.DateMathExpressionResolver();
+    private static final String POLICY_ID_METADATA_FIELD = "policy";
+    private static final String METADATA_FIELD_NAME = "metadata";
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<SnapshotLifecyclePolicy, String> PARSER =
@@ -169,6 +172,30 @@ public ActionRequestValidationException validate() {
             }
         }
 
+        if (configuration.containsKey(METADATA_FIELD_NAME)) {
+            if (configuration.get(METADATA_FIELD_NAME) instanceof Map == false) {
+                err.addValidationError("invalid configuration." + METADATA_FIELD_NAME + " [" + configuration.get(METADATA_FIELD_NAME) +
+                    "]: must be an object if present");
+            } else {
+                @SuppressWarnings("unchecked")
+                Map<String, Object> metadata = (Map<String, Object>) configuration.get(METADATA_FIELD_NAME);
+                if (metadata.containsKey(POLICY_ID_METADATA_FIELD)) {
+                    err.addValidationError("invalid configuration." + METADATA_FIELD_NAME + ": field name [" + POLICY_ID_METADATA_FIELD +
+                        "] is reserved and will be added automatically");
+                } else {
+                    Map<String, Object> metadataWithPolicyField = addPolicyNameToMetadata(metadata);
+                    int serializedSizeOriginal = CreateSnapshotRequest.metadataSize(metadata);
+                    int serializedSizeWithMetadata = CreateSnapshotRequest.metadataSize(metadataWithPolicyField);
+                    int policyNameAddedBytes = serializedSizeWithMetadata - serializedSizeOriginal;
+                    if (serializedSizeWithMetadata > CreateSnapshotRequest.MAXIMUM_METADATA_BYTES) {
+                        err.addValidationError("invalid configuration." + METADATA_FIELD_NAME + ": must be smaller than [" +
+                            (CreateSnapshotRequest.MAXIMUM_METADATA_BYTES - policyNameAddedBytes) +
+                            "] bytes, but is [" + serializedSizeOriginal + "] bytes");
+                    }
+                }
+            }
+        }
+
         // Repository validation, validation of whether the repository actually exists happens
         // elsewhere as it requires cluster state
         if (Strings.hasText(repository) == false) {
@@ -178,6 +205,17 @@ public ActionRequestValidationException validate() {
         return err.validationErrors().size() == 0 ? null : err;
     }
 
+    private Map<String, Object> addPolicyNameToMetadata(final Map<String, Object> metadata) {
+        Map<String, Object> newMetadata;
+        if (metadata == null) {
+            newMetadata = new HashMap<>();
+        } else {
+            newMetadata = new HashMap<>(metadata);
+        }
+        newMetadata.put(POLICY_ID_METADATA_FIELD, this.id);
+        return newMetadata;
+    }
+
     /**
      * Since snapshots need to be uniquely named, this method will resolve any date math used in
      * the provided name, as well as appending a unique identifier so expressions that may overlap
@@ -198,7 +236,12 @@ public String generateSnapshotName(Context context) {
      */
     public CreateSnapshotRequest toRequest() {
         CreateSnapshotRequest req = new CreateSnapshotRequest(repository, generateSnapshotName(new ResolverContext()));
-        req.source(configuration);
+        @SuppressWarnings("unchecked")
+        Map<String, Object> metadata = (Map<String, Object>) configuration.get("metadata");
+        Map<String, Object> metadataWithAddedPolicyName = addPolicyNameToMetadata(metadata);
+        Map<String, Object> mergedConfiguration = new HashMap<>(configuration);
+        mergedConfiguration.put("metadata", metadataWithAddedPolicyName);
+        req.source(mergedConfiguration);
         req.waitForCompletion(false);
         return req;
     }

x-pack/plugin/ilm/qa/multi-node/src/test/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotLifecycleIT.java

@@ -85,6 +85,10 @@ public void testFullPolicySnapshot() throws Exception {
             Map<String, Object> snapResponse = ((List<Map<String, Object>>) snapshotResponseMap.get("snapshots")).get(0);
             assertThat(snapResponse.get("snapshot").toString(), startsWith("snap-"));
             assertThat((List<String>)snapResponse.get("indices"), equalTo(Collections.singletonList(indexName)));
+            Map<String, Object> metadata = (Map<String, Object>) snapResponse.get("metadata");
+            assertNotNull(metadata);
+            assertThat(metadata.get("policy"), equalTo(policyName));
+            assertHistoryIsPresent(policyName, true, repoId);
 
             // Check that the last success date was written to the cluster state
             Request getReq = new Request("GET", "/_slm/policy/" + policyName);
@@ -194,6 +198,9 @@ public void testPolicyManualExecution() throws Exception {
                         snapshotResponseMap = XContentHelper.convertToMap(XContentType.JSON.xContent(), is, true);
                     }
                     assertThat(snapshotResponseMap.size(), greaterThan(0));
+                    final Map<String, Object> metadata = extractMetadata(snapshotResponseMap, snapshotName);
+                    assertNotNull(metadata);
+                    assertThat(metadata.get("policy"), equalTo(policyName));
                     assertHistoryIsPresent(policyName, true, repoId);
                 } catch (ResponseException e) {
                     fail("expected snapshot to exist but it does not: " + EntityUtils.toString(e.getResponse().getEntity()));
@@ -211,6 +218,16 @@ public void testPolicyManualExecution() throws Exception {
         });
     }
 
+    @SuppressWarnings("unchecked")
+    private static Map<String, Object> extractMetadata(Map<String, Object> snapshotResponseMap, String snapshotPrefix) {
+        List<Map<String, Object>> snapshots = ((List<Map<String, Object>>) snapshotResponseMap.get("snapshots"));
+        return snapshots.stream()
+            .filter(snapshot -> ((String) snapshot.get("snapshot")).startsWith(snapshotPrefix))
+            .map(snapshot -> (Map<String, Object>) snapshot.get("metadata"))
+            .findFirst()
+            .orElse(null);
+    }
+
     // This method should be called inside an assertBusy, it has no retry logic of its own
     private void assertHistoryIsPresent(String policyName, boolean success, String repository) throws IOException {
         final Request historySearchRequest = new Request("GET", ".slm-history*/_search");
@@ -263,6 +280,14 @@ private void createSnapshotPolicy(String policyName, String snapshotNamePattern,
         Map<String, Object> snapConfig = new HashMap<>();
         snapConfig.put("indices", Collections.singletonList(indexPattern));
         snapConfig.put("ignore_unavailable", ignoreUnavailable);
+        if (randomBoolean()) {
+            Map<String, Object> metadata = new HashMap<>();
+            int fieldCount = randomIntBetween(2,5);
+            for (int i = 0; i < fieldCount; i++) {
+                metadata.put(randomValueOtherThanMany(key -> "policy".equals(key) || metadata.containsKey(key),
+                    () -> randomAlphaOfLength(5)), randomAlphaOfLength(4));
+            }
+        }
         SnapshotLifecyclePolicy policy = new SnapshotLifecyclePolicy(policyName, snapshotNamePattern, schedule, repoId, snapConfig);
 
         Request putLifecycle = new Request("PUT", "/_slm/policy/" + policyName);

x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotLifecyclePolicyTests.java

@@ -17,6 +17,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
@@ -71,6 +72,53 @@ public void testValidation() {
                 "invalid schedule [ ]: must not be empty"));
     }
 
+    public void testMetadataValidation() {
+        {
+            Map<String, Object> configuration = new HashMap<>();
+            final String metadataString = randomAlphaOfLength(10);
+            configuration.put("metadata", metadataString);
+
+            SnapshotLifecyclePolicy policy = new SnapshotLifecyclePolicy("mypolicy", "<mysnapshot-{now/M}>",
+                "1 * * * * ?", "myrepo", configuration);
+            ValidationException e = policy.validate();
+            assertThat(e.validationErrors(), contains("invalid configuration.metadata [" + metadataString +
+                "]: must be an object if present"));
+        }
+
+        {
+            Map<String, Object> metadata = new HashMap<>();
+            metadata.put("policy", randomAlphaOfLength(5));
+            Map<String, Object> configuration = new HashMap<>();
+            configuration.put("metadata", metadata);
+
+            SnapshotLifecyclePolicy policy = new SnapshotLifecyclePolicy("mypolicy", "<mysnapshot-{now/M}>",
+                "1 * * * * ?", "myrepo", configuration);
+            ValidationException e = policy.validate();
+            assertThat(e.validationErrors(), contains("invalid configuration.metadata: field name [policy] is reserved and " +
+                "will be added automatically"));
+        }
+
+        {
+            Map<String, Object> metadata = new HashMap<>();
+            final int fieldCount = randomIntBetween(67, 100); // 67 is the smallest field count with these sizes that causes an error
+            final int keyBytes = 5; // chosen arbitrarily
+            final int valueBytes = 4; // chosen arbitrarily
+            int totalBytes = fieldCount * (keyBytes + valueBytes + 6 /* bytes of overhead per key/value pair */) + 1;
+            for (int i = 0; i < fieldCount; i++) {
+                metadata.put(randomValueOtherThanMany(key -> "policy".equals(key) || metadata.containsKey(key),
+                    () -> randomAlphaOfLength(keyBytes)), randomAlphaOfLength(valueBytes));
+            }
+            Map<String, Object> configuration = new HashMap<>();
+            configuration.put("metadata", metadata);
+
+            SnapshotLifecyclePolicy policy = new SnapshotLifecyclePolicy("mypolicy", "<mysnapshot-{now/M}>",
+                "1 * * * * ?", "myrepo", configuration);
+            ValidationException e = policy.validate();
+            assertThat(e.validationErrors(), contains("invalid configuration.metadata: must be smaller than [1004] bytes, but is [" +
+                totalBytes + "] bytes"));
+        }
+    }
+
     @Override
     protected SnapshotLifecyclePolicy doParseInstance(XContentParser parser) throws IOException {
         return SnapshotLifecyclePolicy.parse(parser, id);