Avoiding watcher validation errors when a data stream points to more …

…than one index (#85507) (#85755)

* Avoiding watcher validation errors when a data stream points to more than one index (#85507)

This commit fixes a validation bug that prevented watcher from starting if the .watcher-history-16 data stream is
pointing to more than one index. Before 8.0, .watcher-history-16 data had been an alias and had never pointed
to more than one index, so this had not been a problem.

* Fixing test for backport

docs/changelog/85507.yaml

@@ -0,0 +1,7 @@
+pr: 85507
+summary: Avoiding watcher validation errors when a data stream points to more than
+  one index
+area: Watcher
+type: bug
+issues:
+ - 85508

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/watch/WatchStoreUtils.java

@@ -9,16 +9,19 @@
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexNotFoundException;
 
+import java.util.Locale;
+
 public class WatchStoreUtils {
 
     /**
-     * Method to get indexmetadata of a index, that potentially is behind an alias.
+     * Method to get indexmetadata of a index, that potentially is behind an alias or data stream.
      *
      * @param name Name of the index or the alias
      * @param metadata Metadata to search for the name
-     * @return IndexMetadata of the concrete index
+     * @return IndexMetadata of the concrete index. If this alias or data stream has a writable index, this one is returned
      * @throws IllegalStateException If an alias points to two indices
      * @throws IndexNotFoundException If no index exists
      */
@@ -28,11 +31,24 @@ public static IndexMetadata getConcreteIndex(String name, Metadata metadata) {
             return null;
         }
 
-        if (indexAbstraction.getType() != IndexAbstraction.Type.CONCRETE_INDEX && indexAbstraction.getIndices().size() > 1) {
-            throw new IllegalStateException("Alias [" + name + "] points to more than one index");
+        if (indexAbstraction.getType() == IndexAbstraction.Type.ALIAS
+            && indexAbstraction.getIndices().size() > 1
+            && indexAbstraction.getWriteIndex() == null) {
+            throw new IllegalStateException(
+                String.format(
+                    Locale.ROOT,
+                    "Alias [%s] points to %d indices, and does not have a designated write index",
+                    name,
+                    indexAbstraction.getIndices().size()
+                )
+            );
         }
 
-        return metadata.index(indexAbstraction.getIndices().get(0));
+        Index concreteIndex = indexAbstraction.getWriteIndex();
+        if (concreteIndex == null) {
+            concreteIndex = indexAbstraction.getIndices().get(indexAbstraction.getIndices().size() - 1);
+        }
+        return metadata.index(concreteIndex);
     }
 
 }

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/execution/TriggeredWatchStoreTests.java

@@ -388,7 +388,7 @@ public void testLoadingFailsWithTwoAliases() {
         ClusterState cs = csBuilder.build();
 
         IllegalStateException e = expectThrows(IllegalStateException.class, () -> TriggeredWatchStore.validate(cs));
-        assertThat(e.getMessage(), is("Alias [.triggered_watches] points to more than one index"));
+        assertThat(e.getMessage(), is("Alias [.triggered_watches] points to 2 indices, and does not have a designated write index"));
     }
 
     // this is a special condition that could lead to an NPE in earlier versions

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/watch/WatchStoreUtilsTests.java

@@ -0,0 +1,147 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.watcher.watch;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.metadata.AliasMetadata;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.DataStreamAlias;
+import org.elasticsearch.cluster.metadata.DataStreamMetadata;
+import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class WatchStoreUtilsTests extends ESTestCase {
+
+    public void testGetConcreteIndexForDataStream() {
+        String dataStreamName = randomAlphaOfLength(20);
+        Metadata.Builder metadataBuilder = Metadata.builder();
+        ImmutableOpenMap.Builder<String, Metadata.Custom> customsBuilder = ImmutableOpenMap.builder();
+        Map<String, DataStream> dataStreams = new HashMap<>();
+        ImmutableOpenMap.Builder<String, IndexMetadata> indexMetadataMapBuilder = ImmutableOpenMap.builder();
+        List<String> indexNames = new ArrayList<>();
+        for (int i = 0; i < randomIntBetween(2, 10); i++) {
+            String indexName = dataStreamName + "_" + i;
+            indexNames.add(indexName);
+            indexMetadataMapBuilder.put(indexName, createIndexMetaData(indexName, null));
+        }
+        metadataBuilder.indices(indexMetadataMapBuilder.build());
+        dataStreams.put(
+            dataStreamName,
+            DataStreamTestHelper.newInstance(
+                dataStreamName,
+                new DataStream.TimestampField(DataStream.TimestampField.FIXED_TIMESTAMP_FIELD),
+                indexNames.stream().map(indexName -> new Index(indexName, IndexMetadata.INDEX_UUID_NA_VALUE)).collect(Collectors.toList())
+            )
+        );
+        Map<String, DataStreamAlias> dataStreamAliases = Collections.emptyMap();
+        DataStreamMetadata dataStreamMetadata = new DataStreamMetadata(dataStreams, dataStreamAliases);
+        customsBuilder.put(DataStreamMetadata.TYPE, dataStreamMetadata);
+        metadataBuilder.customs(customsBuilder.build());
+        IndexMetadata concreteIndex = WatchStoreUtils.getConcreteIndex(dataStreamName, metadataBuilder.build());
+        assertNotNull(concreteIndex);
+        assertEquals(indexNames.get(indexNames.size() - 1), concreteIndex.getIndex().getName());
+    }
+
+    public void testGetConcreteIndexForAliasWithMultipleNonWritableIndices() {
+        String aliasName = randomAlphaOfLength(20);
+        Metadata.Builder metadataBuilder = Metadata.builder();
+        AliasMetadata.Builder aliasMetadataBuilder = new AliasMetadata.Builder(aliasName);
+        aliasMetadataBuilder.writeIndex(false);
+        AliasMetadata aliasMetadata = aliasMetadataBuilder.build();
+        ImmutableOpenMap.Builder<String, IndexMetadata> indexMetadataMapBuilder = ImmutableOpenMap.builder();
+        for (int i = 0; i < randomIntBetween(2, 10); i++) {
+            String indexName = aliasName + "_" + i;
+            indexMetadataMapBuilder.put(indexName, createIndexMetaData(indexName, aliasMetadata));
+        }
+        metadataBuilder.indices(indexMetadataMapBuilder.build());
+        expectThrows(IllegalStateException.class, () -> WatchStoreUtils.getConcreteIndex(aliasName, metadataBuilder.build()));
+    }
+
+    public void testGetConcreteIndexForAliasWithMultipleIndicesWithWritable() {
+        String aliasName = randomAlphaOfLength(20);
+        Metadata.Builder metadataBuilder = Metadata.builder();
+        AliasMetadata.Builder aliasMetadataBuilder = new AliasMetadata.Builder(aliasName);
+        aliasMetadataBuilder.writeIndex(false);
+        AliasMetadata nonWritableAliasMetadata = aliasMetadataBuilder.build();
+        AliasMetadata.Builder writableAliasMetadataBuilder = new AliasMetadata.Builder(aliasName);
+        writableAliasMetadataBuilder.writeIndex(true);
+        AliasMetadata writableAliasMetadata = writableAliasMetadataBuilder.build();
+        ImmutableOpenMap.Builder<String, IndexMetadata> indexMetadataMapBuilder = ImmutableOpenMap.builder();
+        List<String> indexNames = new ArrayList<>();
+        int indexCount = randomIntBetween(2, 10);
+        int writableIndexIndex = randomIntBetween(0, indexCount - 1);
+        for (int i = 0; i < indexCount; i++) {
+            String indexName = aliasName + "_" + i;
+            indexNames.add(indexName);
+            final AliasMetadata aliasMetadata;
+            if (i == writableIndexIndex) {
+                aliasMetadata = writableAliasMetadata;
+            } else {
+                aliasMetadata = nonWritableAliasMetadata;
+            }
+            indexMetadataMapBuilder.put(indexName, createIndexMetaData(indexName, aliasMetadata));
+        }
+        metadataBuilder.indices(indexMetadataMapBuilder.build());
+        IndexMetadata concreteIndex = WatchStoreUtils.getConcreteIndex(aliasName, metadataBuilder.build());
+        assertNotNull(concreteIndex);
+        assertEquals(indexNames.get(writableIndexIndex), concreteIndex.getIndex().getName());
+    }
+
+    public void testGetConcreteIndexForAliasWithOneNonWritableIndex() {
+        String aliasName = randomAlphaOfLength(20);
+        Metadata.Builder metadataBuilder = Metadata.builder();
+        AliasMetadata.Builder aliasMetadataBuilder = new AliasMetadata.Builder(aliasName);
+        aliasMetadataBuilder.writeIndex(false);
+        AliasMetadata aliasMetadata = aliasMetadataBuilder.build();
+        ImmutableOpenMap.Builder<String, IndexMetadata> indexMetadataMapBuilder = ImmutableOpenMap.builder();
+        String indexName = aliasName + "_" + 0;
+        indexMetadataMapBuilder.put(indexName, createIndexMetaData(indexName, aliasMetadata));
+        metadataBuilder.indices(indexMetadataMapBuilder.build());
+        IndexMetadata concreteIndex = WatchStoreUtils.getConcreteIndex(aliasName, metadataBuilder.build());
+        assertNotNull(concreteIndex);
+        assertEquals(indexName, concreteIndex.getIndex().getName());
+    }
+
+    public void testGetConcreteIndexForConcreteIndex() {
+        String indexName = randomAlphaOfLength(20);
+        Metadata.Builder metadataBuilder = Metadata.builder();
+        ImmutableOpenMap.Builder<String, IndexMetadata> indexMetadataMapBuilder = ImmutableOpenMap.builder();
+        indexMetadataMapBuilder.put(indexName, createIndexMetaData(indexName, null));
+        metadataBuilder.indices(indexMetadataMapBuilder.build());
+        IndexMetadata concreteIndex = WatchStoreUtils.getConcreteIndex(indexName, metadataBuilder.build());
+        assertNotNull(concreteIndex);
+        assertEquals(indexName, concreteIndex.getIndex().getName());
+    }
+
+    private IndexMetadata createIndexMetaData(String indexName, AliasMetadata aliasMetadata) {
+        IndexMetadata.Builder indexMetadataBuilder = new IndexMetadata.Builder(indexName);
+        Settings settings = Settings.builder()
+            .put(IndexMetadata.SETTING_PRIORITY, 5)
+            .put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+            .put(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 1)
+            .put(IndexMetadata.SETTING_INDEX_VERSION_CREATED.getKey(), Version.CURRENT)
+            .build();
+        indexMetadataBuilder.settings(settings);
+        if (aliasMetadata != null) {
+            indexMetadataBuilder.putAlias(aliasMetadata);
+        }
+        return indexMetadataBuilder.build();
+    }
+}