Ensure that .watcher-history-11* template is in installed prior to use

[WatcherIndexTemplateRegistry](https://github.com/elastic/elasticsearch/blob/7.7/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/support/WatcherIndexTemplateRegistry.java#L74)
as of elastic#52962 requires all nodes
to be on 7.7.0 before it allows the version 11 index template to be installed.

While in a mixed cluster, nothing prevents Watcher from running on the new
host before the all of the nodes are on 7.7.0. This will result in the
.watcher-history-11* index without the proper mappings. Without the proper
mapping a single document (for a large watch) can exceed the default 1000 field
limit and cause error to show in the logs.

This commit ensures the same logic for writing to the index is applied as for
installing the template. In a mixed cluster, the `10` index template will continue
to be written. Only once all of nodes are on 7.7.0+ will the `11` index template
be installed and used.

Note - this PR targets 7.x and will be back ported to 7.7.next. This conditional
installation of the template is not present in master.

closes elastic#56732

Author: jakelandis

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

@@ -154,7 +154,7 @@ private void addTemplatesIfMissing(ClusterState state) {
             if (creationCheck.compareAndSet(false, true)) {
                 IndexTemplateMetadata currentTemplate = state.metadata().getTemplates().get(templateName);
                 if (Objects.isNull(currentTemplate)) {
-                    logger.debug("adding index template [{}] for [{}], because it doesn't exist", templateName, getOrigin());
+                    logger.info("adding index template [{}] for [{}], because it doesn't exist", templateName, getOrigin());
                     putTemplate(newTemplate, creationCheck);
                 } else if (Objects.isNull(currentTemplate.getVersion()) || newTemplate.getVersion() > currentTemplate.getVersion()) {
                     // IndexTemplateConfig now enforces templates contain a `version` property, so if the template doesn't have one we can

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/history/HistoryStoreField.java

@@ -5,6 +5,8 @@
  */
 package org.elasticsearch.xpack.core.watcher.history;
 
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.xpack.core.watcher.support.WatcherIndexTemplateRegistryField;
 
@@ -14,12 +16,19 @@ public final class HistoryStoreField {
 
     public static final String INDEX_PREFIX = ".watcher-history-";
     public static final String INDEX_PREFIX_WITH_TEMPLATE = INDEX_PREFIX + WatcherIndexTemplateRegistryField.INDEX_TEMPLATE_VERSION + "-";
+    public static final String INDEX_PREFIX_WITH_TEMPLATE_10 = INDEX_PREFIX +
+        WatcherIndexTemplateRegistryField.INDEX_TEMPLATE_VERSION_10 + "-";
     private static final DateFormatter indexTimeFormat = DateFormatter.forPattern("yyyy.MM.dd");
 
     /**
      * Calculates the correct history index name for a given time
      */
-    public static String getHistoryIndexNameForTime(ZonedDateTime time) {
-        return INDEX_PREFIX_WITH_TEMPLATE + indexTimeFormat.format(time);
+    public static String getHistoryIndexNameForTime(ZonedDateTime time, ClusterState state) {
+        //null only allowed for testing
+        if (state == null || state.nodes().getMinNodeVersion().onOrAfter(Version.V_7_7_0)) {
+            return INDEX_PREFIX_WITH_TEMPLATE + indexTimeFormat.format(time);
+        } else {
+            return INDEX_PREFIX_WITH_TEMPLATE_10 + indexTimeFormat.format(time);
+        }
     }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/support/WatcherIndexTemplateRegistryField.java

@@ -18,8 +18,9 @@ public final class WatcherIndexTemplateRegistryField {
     // version 11: watch history indices are hidden
     // Note: if you change this, also inform the kibana team around the watcher-ui
     public static final int INDEX_TEMPLATE_VERSION = 11;
+    public static final int INDEX_TEMPLATE_VERSION_10 = 10;
     public static final String HISTORY_TEMPLATE_NAME = ".watch-history-" + INDEX_TEMPLATE_VERSION;
-    public static final String HISTORY_TEMPLATE_NAME_10 = ".watch-history-10";
+    public static final String HISTORY_TEMPLATE_NAME_10 = ".watch-history-" + INDEX_TEMPLATE_VERSION_10;
     public static final String HISTORY_TEMPLATE_NAME_NO_ILM = ".watch-history-no-ilm-" + INDEX_TEMPLATE_VERSION;
     public static final String HISTORY_TEMPLATE_NAME_NO_ILM_10 = ".watch-history-no-ilm-10";
     public static final String TRIGGERED_TEMPLATE_NAME = ".triggered_watches";

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/store/ReservedRolesStoreTests.java

@@ -1395,7 +1395,7 @@ public void testWatcherAdminRole() {
         assertThat(role.indices().allowedIndicesMatcher(IndexAction.NAME).test("foo"), is(false));
 
         ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
-        String historyIndex = HistoryStoreField.getHistoryIndexNameForTime(now);
+        String historyIndex = HistoryStoreField.getHistoryIndexNameForTime(now, null);
         for (String index : new String[]{ Watch.INDEX, historyIndex, TriggeredWatchStoreField.INDEX_NAME }) {
             assertOnlyReadAllowed(role, index);
         }
@@ -1429,7 +1429,7 @@ public void testWatcherUserRole() {
         assertThat(role.indices().allowedIndicesMatcher(IndexAction.NAME).test(TriggeredWatchStoreField.INDEX_NAME), is(false));
 
         ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
-        String historyIndex = HistoryStoreField.getHistoryIndexNameForTime(now);
+        String historyIndex = HistoryStoreField.getHistoryIndexNameForTime(now, null);
         for (String index : new String[]{ Watch.INDEX, historyIndex }) {
             assertOnlyReadAllowed(role, index);
         }

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

@@ -241,10 +241,6 @@ public Watcher(final Settings settings) {
         this.settings = settings;
         this.transportClient = XPackPlugin.transportClientMode(settings);
         this.enabled = XPackSettings.WATCHER_ENABLED.get(settings);
-
-        if (enabled && transportClient == false) {
-            validAutoCreateIndex(settings, logger);
-        }
     }
 
     // overridable by tests
@@ -259,6 +255,10 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
                                                NodeEnvironment nodeEnvironment, NamedWriteableRegistry namedWriteableRegistry,
                                                IndexNameExpressionResolver expressionResolver,
                                                Supplier<RepositoriesService> repositoriesServiceSupplier) {
+        if (enabled && transportClient == false) {
+            validAutoCreateIndex(settings, logger, clusterService);
+        }
+
         if (enabled == false) {
             return Collections.emptyList();
         }
@@ -386,7 +386,7 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
             .setConcurrentRequests(SETTING_BULK_CONCURRENT_REQUESTS.get(settings))
             .build();
 
-        HistoryStore historyStore = new HistoryStore(bulkProcessor);
+        HistoryStore historyStore = new HistoryStore(bulkProcessor, clusterService);
 
         // schedulers
         final Set<Schedule.Parser> scheduleParsers = new HashSet<>();
@@ -601,7 +601,7 @@ public void onIndexModule(IndexModule module) {
         module.addIndexOperationListener(listener);
     }
 
-    static void validAutoCreateIndex(Settings settings, Logger logger) {
+    static void validAutoCreateIndex(Settings settings, Logger logger, ClusterService clusterService) {
         String value = settings.get("action.auto_create_index");
         if (value == null) {
             return;
@@ -623,14 +623,14 @@ static void validAutoCreateIndex(Settings settings, Logger logger) {
         indices.add(".watches");
         indices.add(".triggered_watches");
         ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now));
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusDays(1)));
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(1)));
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(2)));
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(3)));
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(4)));
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(5)));
-        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(6)));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now, clusterService.state()));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusDays(1), clusterService.state()));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(1), clusterService.state()));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(2), clusterService.state()));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(3), clusterService.state()));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(4), clusterService.state()));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(5), clusterService.state()));
+        indices.add(HistoryStoreField.getHistoryIndexNameForTime(now.plusMonths(6), clusterService.state()));
         for (String index : indices) {
             boolean matched = false;
             for (String match : matches) {

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/execution/ExecutionService.java

@@ -452,7 +452,7 @@ private void executeAsync(WatchExecutionContext ctx, final TriggeredWatch trigge
      * Any existing watchRecord will be overwritten.
      */
     private void forcePutHistory(WatchRecord watchRecord) {
-        String index = HistoryStoreField.getHistoryIndexNameForTime(watchRecord.triggerEvent().triggeredTime());
+        String index = HistoryStoreField.getHistoryIndexNameForTime(watchRecord.triggerEvent().triggeredTime(), clusterService.state());
         try {
             try (XContentBuilder builder = XContentFactory.jsonBuilder();
                  ThreadContext.StoredContext ignore = client.threadPool().getThreadContext().stashWithOrigin(WATCHER_ORIGIN)) {

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/history/HistoryStore.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.xpack.core.watcher.history.HistoryStoreField;
@@ -31,17 +32,19 @@ public class HistoryStore {
     private static final Logger logger = LogManager.getLogger(HistoryStore.class);
 
     private final BulkProcessor bulkProcessor;
+    private final ClusterService clusterService;
 
-    public HistoryStore(BulkProcessor bulkProcessor) {
+    public HistoryStore(BulkProcessor bulkProcessor, ClusterService clusterService) {
         this.bulkProcessor = bulkProcessor;
+        this.clusterService = clusterService;
     }
 
     /**
      * Stores the specified watchRecord.
      * If the specified watchRecord already was stored this call will fail with a version conflict.
      */
     public void put(WatchRecord watchRecord) throws Exception {
-        String index = HistoryStoreField.getHistoryIndexNameForTime(watchRecord.triggerEvent().triggeredTime());
+        String index = HistoryStoreField.getHistoryIndexNameForTime(watchRecord.triggerEvent().triggeredTime(), clusterService.state());
         try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
             watchRecord.toXContent(builder, WatcherParams.HIDE_SECRETS);
 
@@ -58,7 +61,7 @@ public void put(WatchRecord watchRecord) throws Exception {
      * Any existing watchRecord will be overwritten.
      */
     public void forcePut(WatchRecord watchRecord) {
-        String index = HistoryStoreField.getHistoryIndexNameForTime(watchRecord.triggerEvent().triggeredTime());
+        String index = HistoryStoreField.getHistoryIndexNameForTime(watchRecord.triggerEvent().triggeredTime(), clusterService.state());
             try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
                 watchRecord.toXContent(builder, WatcherParams.HIDE_SECRETS);
 
@@ -78,7 +81,7 @@ public void forcePut(WatchRecord watchRecord) {
      * @return true, if history store is ready to be started
      */
     public static boolean validate(ClusterState state) {
-        String currentIndex = HistoryStoreField.getHistoryIndexNameForTime(ZonedDateTime.now(ZoneOffset.UTC));
+        String currentIndex = HistoryStoreField.getHistoryIndexNameForTime(ZonedDateTime.now(ZoneOffset.UTC), state);
         IndexMetadata indexMetadata = WatchStoreUtils.getConcreteIndex(currentIndex, state.metadata());
         return indexMetadata == null || (indexMetadata.getState() == IndexMetadata.State.OPEN &&
             state.routingTable().index(indexMetadata.getIndex()).allPrimaryShardsActive());

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

@@ -5,7 +5,11 @@
  */
 package org.elasticsearch.xpack.watcher;
 
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.TestEnvironment;
@@ -19,7 +23,9 @@
 import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.watcher.watch.Watch;
 import org.elasticsearch.xpack.watcher.notification.NotificationService;
+import org.junit.Before;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -31,33 +37,52 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 
 public class WatcherPluginTests extends ESTestCase {
 
+    private ClusterService clusterService;
+    private ClusterState clusterState;
+    private DiscoveryNodes discoveryNodes;
+
+    @Before
+    public void init() {
+        clusterService = mock(ClusterService.class);
+        clusterState = mock(ClusterState.class);
+        discoveryNodes = mock(DiscoveryNodes.class);
+        when(clusterService.state()).thenReturn(clusterState);
+        when(clusterState.nodes()).thenReturn(discoveryNodes);
+        when(discoveryNodes.getMinNodeVersion()).thenReturn(randomFrom(Arrays.asList(Version.V_7_0_0, Version.V_7_7_0)));
+
+    }
+
     public void testValidAutoCreateIndex() {
-        Watcher.validAutoCreateIndex(Settings.EMPTY, logger);
-        Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", true).build(), logger);
+        Watcher.validAutoCreateIndex(Settings.EMPTY, logger, clusterService);
+        Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", true).build(), logger, clusterService);
 
         IllegalArgumentException exception = expectThrows(IllegalArgumentException.class,
-                () -> Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", false).build(), logger));
+                () -> Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", false).build(), logger,
+                    clusterService));
         assertThat(exception.getMessage(), containsString("[.watches,.triggered_watches,.watcher-history-*]"));
 
         Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index",
-                ".watches,.triggered_watches,.watcher-history*").build(), logger);
-        Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", "*w*").build(), logger);
-        Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", ".w*,.t*").build(), logger);
+                ".watches,.triggered_watches,.watcher-history*").build(), logger, clusterService);
+        Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", "*w*").build(), logger, clusterService);
+        Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", ".w*,.t*").build(), logger, clusterService);
 
         exception = expectThrows(IllegalArgumentException.class,
-                () -> Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", ".watches").build(), logger));
+                () -> Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", ".watches").build(), logger,
+                    clusterService));
         assertThat(exception.getMessage(), containsString("[.watches,.triggered_watches,.watcher-history-*]"));
 
         exception = expectThrows(IllegalArgumentException.class,
-                () -> Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", ".triggered_watch").build(), logger));
+                () -> Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", ".triggered_watch").build(), logger,
+                    clusterService));
         assertThat(exception.getMessage(), containsString("[.watches,.triggered_watches,.watcher-history-*]"));
 
         exception = expectThrows(IllegalArgumentException.class,
                 () -> Watcher.validAutoCreateIndex(Settings.builder().put("action.auto_create_index", ".watcher-history-*").build(),
-                        logger));
+                        logger, clusterService));
         assertThat(exception.getMessage(), containsString("[.watches,.triggered_watches,.watcher-history-*]"));
     }
 
@@ -130,7 +155,6 @@ public void testReload() {
             .build();
         NotificationService mockService = mock(NotificationService.class);
         Watcher watcher = new TestWatcher(settings, mockService);
-
         watcher.reload(settings);
         verify(mockService, times(1)).reload(settings);
     }