Introduce data stream options and failure store configuration classes (…

…#109515)

In order to facilitate enabling and disabling the failure store &
component template composition, we introduce new metadata classes that
can support a more extensible failure store configuration.

We would like to introduce **data stream options**. Data stream options
capture the configuration of data stream level (smaller and larger)
features, such the failure store and in the future data stream
lifecycle. They are different than settings because they are applied on
a data stream level and not per backing index.

This PR is only setting the basic classes to enable follow up PRs that
will actually use them. 

Examples, these are not final, they are only used to help visualise a
potential direction:

```
GET _data_stream/my-*/_options
{
  "data_streams": [
    {
      "name": "my-non-opinionated-ds",
      "options": {
      }
    },
    {
      "name": "my-fs",
      "options": {
        "failure_store": {
          "enabled": true
        }
      }
    },
    {
      "name": "my-no-fs",
      "options": {
        "failure_store": {
          "enabled": false
        }
      }
    }
  ]
}

// If we decide to add lifecycle here too:
PUT _data_stream/my-fs/_options
{
   "failure_store": {
     "enabled": true
   },
   "lifecycle": {
   }
}
```

What we see above are 3 data streams:

- `my-fs` with the failure store explicitly enabled
- `my-no-fs` with the failure store explicitly disabled, and
- `my-non-opinionated-ds` which does not specify what to do with the failure store, so for now it means failure store disabled but that could change in the future.

Template composition examples pending

server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamFailureStore.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.cluster.metadata;
+
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.cluster.SimpleDiffable;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+/**
+ * Holds the data stream failure store metadata that enable or disable the failure store of a data stream. Currently, it
+ * supports the following configurations:
+ * - enabled
+ */
+public record DataStreamFailureStore(boolean enabled) implements SimpleDiffable<DataStreamFailureStore>, ToXContentObject {
+
+    public static final ParseField ENABLED_FIELD = new ParseField("enabled");
+
+    public static final ConstructingObjectParser<DataStreamFailureStore, Void> PARSER = new ConstructingObjectParser<>(
+        "failure_store",
+        false,
+        (args, unused) -> new DataStreamFailureStore(args[0] == null || (Boolean) args[0])
+    );
+
+    static {
+        PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), ENABLED_FIELD);
+    }
+
+    public DataStreamFailureStore() {
+        this(true);
+    }
+
+    public DataStreamFailureStore(StreamInput in) throws IOException {
+        this(in.readBoolean());
+    }
+
+    public static Diff<DataStreamFailureStore> readDiffFrom(StreamInput in) throws IOException {
+        return SimpleDiffable.readDiffFrom(DataStreamFailureStore::new, in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeBoolean(enabled);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this, true, true);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(ENABLED_FIELD.getPreferredName(), enabled);
+        builder.endObject();
+        return builder;
+    }
+
+    public static DataStreamFailureStore fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+}

server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamOptions.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.cluster.metadata;
+
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.cluster.SimpleDiffable;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+/**
+ * Holds data stream dedicated configuration options such as failure store, (in the future lifecycle). Currently, it
+ * supports the following configurations:
+ * - failure store
+ */
+public record DataStreamOptions(@Nullable DataStreamFailureStore failureStore)
+    implements
+        SimpleDiffable<DataStreamOptions>,
+        ToXContentObject {
+
+    public static final ParseField FAILURE_STORE_FIELD = new ParseField("failure_store");
+
+    public static final ConstructingObjectParser<DataStreamOptions, Void> PARSER = new ConstructingObjectParser<>(
+        "options",
+        false,
+        (args, unused) -> new DataStreamOptions((DataStreamFailureStore) args[0])
+    );
+
+    static {
+        PARSER.declareField(
+            ConstructingObjectParser.optionalConstructorArg(),
+            (p, c) -> DataStreamFailureStore.fromXContent(p),
+            FAILURE_STORE_FIELD,
+            ObjectParser.ValueType.OBJECT_OR_NULL
+        );
+    }
+
+    public DataStreamOptions() {
+        this(null);
+    }
+
+    public static DataStreamOptions read(StreamInput in) throws IOException {
+        return new DataStreamOptions(in.readOptionalWriteable(DataStreamFailureStore::new));
+    }
+
+    @Nullable
+    public DataStreamFailureStore getFailureStore() {
+        return failureStore;
+    }
+
+    public static Diff<DataStreamOptions> readDiffFrom(StreamInput in) throws IOException {
+        return SimpleDiffable.readDiffFrom(DataStreamOptions::read, in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeOptionalWriteable(failureStore);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this, true, true);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        if (failureStore != null) {
+            builder.field(FAILURE_STORE_FIELD.getPreferredName(), failureStore);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    public static DataStreamOptions fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+}

server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamFailureStoreTests.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.cluster.metadata;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+public class DataStreamFailureStoreTests extends AbstractXContentSerializingTestCase<DataStreamFailureStore> {
+
+    @Override
+    protected Writeable.Reader<DataStreamFailureStore> instanceReader() {
+        return DataStreamFailureStore::new;
+    }
+
+    @Override
+    protected DataStreamFailureStore createTestInstance() {
+        return randomFailureStore();
+    }
+
+    @Override
+    protected DataStreamFailureStore mutateInstance(DataStreamFailureStore instance) throws IOException {
+        return new DataStreamFailureStore(instance.enabled() == false);
+    }
+
+    @Override
+    protected DataStreamFailureStore doParseInstance(XContentParser parser) throws IOException {
+        return DataStreamFailureStore.fromXContent(parser);
+    }
+
+    static DataStreamFailureStore randomFailureStore() {
+        return new DataStreamFailureStore(randomBoolean());
+    }
+}

server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamOptionsTests.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.cluster.metadata;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+public class DataStreamOptionsTests extends AbstractXContentSerializingTestCase<DataStreamOptions> {
+
+    @Override
+    protected Writeable.Reader<DataStreamOptions> instanceReader() {
+        return DataStreamOptions::read;
+    }
+
+    @Override
+    protected DataStreamOptions createTestInstance() {
+        return new DataStreamOptions(randomBoolean() ? null : DataStreamFailureStoreTests.randomFailureStore());
+    }
+
+    @Override
+    protected DataStreamOptions mutateInstance(DataStreamOptions instance) throws IOException {
+        var failureStore = instance.getFailureStore();
+        if (failureStore == null) {
+            failureStore = DataStreamFailureStoreTests.randomFailureStore();
+        } else {
+            failureStore = randomBoolean() ? null : new DataStreamFailureStore(failureStore.enabled() == false);
+        }
+        return new DataStreamOptions(failureStore);
+    }
+
+    @Override
+    protected DataStreamOptions doParseInstance(XContentParser parser) throws IOException {
+        return DataStreamOptions.fromXContent(parser);
+    }
+}