Delete legacy watcher templates

We moved watcher to run on the system indices infrastructure in elastic#67588.
This commit makes sure Watcher cleans up the legacy templates it used
before it was migrated to system indices.

Author: andreidan

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

@@ -8,16 +8,21 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.IndexTemplateMetadata;
+import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.component.LifecycleListener;
 import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.index.shard.ShardId;
@@ -26,11 +31,13 @@
 import org.elasticsearch.xpack.core.watcher.watch.Watch;
 import org.elasticsearch.xpack.watcher.watch.WatchStoreUtils;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -41,14 +48,24 @@
 public class WatcherLifeCycleService implements ClusterStateListener {
 
     private static final Logger logger = LogManager.getLogger(WatcherLifeCycleService.class);
+
+    public static final Set<String> LEGACY_WATCHER_INDEX_TEMPLATES = Set.of(
+        ".watches",
+        ".triggered_watches"
+    );
+
     private final AtomicReference<WatcherState> state = new AtomicReference<>(WatcherState.STARTED);
     private final AtomicReference<List<ShardRouting>> previousShardRoutings = new AtomicReference<>(Collections.emptyList());
     private volatile boolean shutDown = false; // indicates that the node has been shutdown and we should never start watcher after this.
+    private final ClusterService clusterService;
     private volatile WatcherService watcherService;
     private final EnumSet<WatcherState> stopStates = EnumSet.of(WatcherState.STOPPED, WatcherState.STOPPING);
+    private final AtomicBoolean legacyTemplatesDeleteInProgress = new AtomicBoolean(false);
+    private final AtomicBoolean checkForLegacyTemplates = new AtomicBoolean(true);
 
     WatcherLifeCycleService(ClusterService clusterService, WatcherService watcherService) {
         this.watcherService = watcherService;
+        this.clusterService = clusterService;
         clusterService.addListener(this);
         // Close if the indices service is being stopped, so we don't run into search failures (locally) that will
         // happen because we're shutting down and an watch is scheduled.
@@ -96,6 +113,36 @@ public void clusterChanged(ClusterChangedEvent event) {
             return;
         }
 
+        if (event.localNodeMaster() && checkForLegacyTemplates.get()) {
+            List<String> existingTemplatesToDelete = new ArrayList<>(LEGACY_WATCHER_INDEX_TEMPLATES.size());
+            ImmutableOpenMap<String, IndexTemplateMetadata> clusterLegacyTemplates = event.state().getMetadata().templates();
+            for (String legacyWatcherIndexTemplate : LEGACY_WATCHER_INDEX_TEMPLATES) {
+                if (clusterLegacyTemplates.containsKey(legacyWatcherIndexTemplate)) {
+                    existingTemplatesToDelete.add(legacyWatcherIndexTemplate);
+                }
+            }
+
+            if (existingTemplatesToDelete.isEmpty() == false) {
+                // if someone else is executing the deletion of templates (due to fast successive cluster updates), we'll skip doing so
+                if (legacyTemplatesDeleteInProgress.compareAndSet(false, true) == false) {
+                    return;
+                }
+
+                MetadataIndexTemplateService.removeTemplates(clusterService, LEGACY_WATCHER_INDEX_TEMPLATES,
+                    ActionListener.wrap(r -> {
+                        legacyTemplatesDeleteInProgress.set(false);
+                        // we've done it so we shouldn't check anymore
+                        checkForLegacyTemplates.set(false);
+                        logger.debug("deleted legacy Watcher index templates [{}]", String.join(",", LEGACY_WATCHER_INDEX_TEMPLATES));
+                    }, e -> {
+                        legacyTemplatesDeleteInProgress.set(false);
+                        logger.debug(new ParameterizedMessage("unable to delete legacy Watcher index templates [{}]",
+                            String.join(",", LEGACY_WATCHER_INDEX_TEMPLATES)), e);
+                    })
+                );
+            }
+        }
+
         boolean isWatcherStoppedManually = isWatcherStoppedManually(event.state());
         boolean isStoppedOrStopping = stopStates.contains(this.state.get());
         // if this is not a data node, we need to start it ourselves possibly

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

@@ -40,6 +40,7 @@
 
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -562,6 +563,13 @@ public void testWatcherReloadsOnNodeOutageWithWatcherShard() {
         verify(watcherService).reload(eq(event.state()), anyString());
     }
 
+    public void testLegacyWatcherTemplatesToDelete() {
+        assertThat(WatcherLifeCycleService.LEGACY_WATCHER_INDEX_TEMPLATES.size(), is(2));
+        Iterator<String> legacyTemplatesIterator = WatcherLifeCycleService.LEGACY_WATCHER_INDEX_TEMPLATES.iterator();
+        assertThat(legacyTemplatesIterator.next(), is(".watches"));
+        assertThat(legacyTemplatesIterator.next(), is(".triggered_watches"));
+    }
+
     private void startWatcher() {
         Index index = new Index(Watch.INDEX, "uuid");
         IndexRoutingTable.Builder indexRoutingTableBuilder = IndexRoutingTable.builder(index);

x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/WatcherRestartIT.java

@@ -9,10 +9,14 @@
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
 
+import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Map;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 
 public class WatcherRestartIT extends AbstractUpgradeTestCase {
@@ -46,4 +50,30 @@ private void ensureWatcherStarted() throws Exception {
             assertThat(responseBody, not(containsString("\"watcher_state\":\"stopped\"")));
         });
     }
+
+    public void testEnsureWatcherDeletesLegacyTemplates() throws Exception {
+        client().performRequest(new Request("POST", "/_watcher/_start"));
+        ensureWatcherStarted();
+
+        // All the legacy ML templates we created over the years should be deleted now they're no longer needed
+        assertBusy(() -> {
+            Request request = new Request("GET", "/_template/*watches*");
+            try {
+                Response response = client().performRequest(request);
+                Map<String, Object> responseLevel = entityAsMap(response);
+                assertNotNull(responseLevel);
+
+                assertThat(responseLevel.containsKey(".watches"), is(false));
+                assertThat(responseLevel.containsKey(".triggered_watches"), is(false));
+            } catch (ResponseException e) {
+                // Not found is fine
+                assertThat(
+                    "Unexpected failure getting templates: " + e.getResponse().getStatusLine(),
+                    e.getResponse().getStatusLine().getStatusCode(),
+                    is(404)
+                );
+            }
+        });
+    }
+
 }