Ensure SLM stats does not block an in-place upgrade from 7.4 (#48367)

7.5+ for SLM requires [stats] object to exist in the cluster state.
When doing an in-place upgrade from 7.4 to 7.5+ [stats] does not exist
in cluster state, result in an exception on startup [1].

This commit moves the [stats] to be an optional object in the parser
and if not found will default to an empty stats object.

[1] Caused by: java.lang.IllegalArgumentException: Required [stats]

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

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/slm/SnapshotLifecycleMetadata.java

@@ -61,7 +61,7 @@ public class SnapshotLifecycleMetadata implements MetaData.Custom {
                 throw new IllegalArgumentException("ordered " + POLICIES_FIELD.getPreferredName() + " are not supported");
             }, POLICIES_FIELD);
         PARSER.declareString(ConstructingObjectParser.constructorArg(), OPERATION_MODE_FIELD);
-        PARSER.declareObject(ConstructingObjectParser.constructorArg(), (v, o) -> SnapshotLifecycleStats.parse(v), STATS_FIELD);
+        PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (v, o) -> SnapshotLifecycleStats.parse(v), STATS_FIELD);
     }
 
     private final Map<String, SnapshotLifecyclePolicyMetadata> snapshotConfigurations;
@@ -73,7 +73,7 @@ public class SnapshotLifecycleMetadata implements MetaData.Custom {
                                      SnapshotLifecycleStats slmStats) {
         this.snapshotConfigurations = new HashMap<>(snapshotConfigurations);
         this.operationMode = operationMode;
-        this.slmStats = slmStats;
+        this.slmStats = slmStats != null ? slmStats : new SnapshotLifecycleStats();
     }
 
     public SnapshotLifecycleMetadata(StreamInput in) throws IOException {

x-pack/qa/full-cluster-restart/src/test/java/org/elasticsearch/xpack/restart/FullClusterRestartIT.java

@@ -10,20 +10,30 @@ import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.common.xcontent.DeprecationHandler;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.ObjectPath;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.search.RestSearchAction;
 import org.elasticsearch.test.StreamsUtils;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.upgrades.AbstractFullClusterRestartTestCase;
+import org.elasticsearch.xpack.core.slm.SnapshotLifecyclePolicy;
+import org.elasticsearch.xpack.slm.SnapshotLifecycleStats;
 import org.hamcrest.Matcher;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -265,6 +275,39 @@ public class FullClusterRestartIT extends AbstractFullClusterRestartTestCase {
             assertRollUpJob("rollup-job-test");
         }
     }
+    
+    public void testSlmStats() throws IOException {
+        SnapshotLifecyclePolicy slmPolicy = new SnapshotLifecyclePolicy("test-policy", "test-policy", "* * * 31 FEB ? *", "test-repo",
+            Collections.singletonMap("indices", Collections.singletonList("*")), null);
+        if (isRunningAgainstOldCluster() && getOldClusterVersion().onOrAfter(Version.V_7_4_0)) {
+            Request createRepoRequest = new Request("PUT", "_snapshot/test-repo");
+            String repoCreateJson = "{" +
+                " \"type\": \"fs\"," +
+                " \"settings\": {" +
+                "   \"location\": \"test-repo\"" +
+                "  }" +
+                "}";
+            createRepoRequest.setJsonEntity(repoCreateJson);
+            Request createSlmPolicyRequest = new Request("PUT", "_slm/policy/test-policy");
+            try (XContentBuilder builder = JsonXContent.contentBuilder()) {
+                String createSlmPolicyJson = Strings.toString(slmPolicy.toXContent(builder, null));
+                createSlmPolicyRequest.setJsonEntity(createSlmPolicyJson);
+            }
+
+            client().performRequest(createRepoRequest);
+            client().performRequest(createSlmPolicyRequest);
+        }
+
+        if (isRunningAgainstOldCluster() == false || getOldClusterVersion().onOrAfter(Version.V_7_5_0)) {
+            Response response = client().performRequest(new Request("GET", "_slm/stats"));
+            XContentType xContentType = XContentType.fromMediaTypeOrFormat(response.getEntity().getContentType().getValue());
+            try (XContentParser parser = xContentType.xContent().createParser(NamedXContentRegistry.EMPTY,
+                DeprecationHandler.THROW_UNSUPPORTED_OPERATION, response.getEntity().getContent())) {
+                assertEquals(new SnapshotLifecycleStats(), SnapshotLifecycleStats.parse(parser));
+            }
+        }
+
+    }
 
     private String loadWatch(String watch) throws IOException {
         return StreamsUtils.copyToStringFromClasspath("/org/elasticsearch/xpack/restart/" + watch);