Watcher: Make start/stop cycle more predictable and synchronous (#30118)

The current implementation starts/stops watcher using an executor. This
can result in our of order operations.

This commit reduces those executor calls to an absolute minimum in order
to be able to do state changes within the cluster state listener method,
which runs in sequence.

When a state change occurs that forces the watcher service to pause
(like no watcher index, no master node, no local shards), the service is
now in a paused state.

Pausing is a super lightweight operation, which marks the
ExecutionService as paused and waits for the currently executing watches
to finish in the background via an executor. The same applies for
stopping, the potentially long running operation is outsourced in to an
executor, as waiting for executed watches is decoupled from the current
state.

The only other long running operation is starting, where watches need to
be loaded. This is also done via an executor, but has an additional
protection by checking the cluster state version it was started with. If
another cluster state version was trying to load the watches, then this
loading will not take effect.

This PR also cleans up some unused states, like the a simple boolean in
the HistoryStore/TriggeredWatchStore marking it as started or stopped,
as this can now be caught in the execution service.

Another advantage of this approach is the fact, that now only triggered
watches are not getting executed, while watches that are run via the
Execute Watch API will still be executed regardless if watcher is
stopped or not.

Lastly the TickerScheduleTriggerEngine thread now only starts on data nodes.

Author: spinscale

docs/CHANGELOG.asciidoc

@@ -379,6 +379,11 @@ more information, see <<notification-settings>>.
 
 //[float]
 //=== Enhancements
+The cluster state listener to decide if watcher should be
+stopped/started/paused now runs far less code in an executor but is more
+synchronous and predictable. Also the trigger engine thread is only started on
+data nodes. And the Execute Watch API can be triggered regardless is watcher is
+started or stopped. ({pull}30118[#30118])
 
 [float]
 === Bug Fixes

x-pack/plugin/src/test/java/org/elasticsearch/xpack/test/rest/XPackRestIT.java

@@ -8,7 +8,9 @@
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 import org.apache.http.HttpStatus;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.client.Response;
 import org.elasticsearch.common.CheckedFunction;
+import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
@@ -17,6 +19,7 @@
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestResponse;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+import org.elasticsearch.test.rest.yaml.ObjectPath;
 import org.elasticsearch.xpack.core.ml.MlMetaIndex;
 import org.elasticsearch.xpack.core.ml.integration.MlRestTestStateCleaner;
 import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
@@ -83,7 +86,6 @@ private void waitForTemplates() throws Exception {
             templates.addAll(Arrays.asList(AuditorField.NOTIFICATIONS_INDEX, MlMetaIndex.INDEX_NAME,
                     AnomalyDetectorsIndex.jobStateIndexName(),
                     AnomalyDetectorsIndex.jobResultsIndexPrefix()));
-            templates.addAll(Arrays.asList(WatcherIndexTemplateRegistryField.TEMPLATE_NAMES));
 
             for (String template : templates) {
                 awaitCallApi("indices.exists_template", singletonMap("name", template), emptyList(),
@@ -97,19 +99,46 @@ private void waitForWatcher() throws Exception {
         // ensure watcher is started, so that a test can stop watcher and everything still works fine
         if (isWatcherTest()) {
             assertBusy(() -> {
-                try {
-                    ClientYamlTestResponse response =
-                            getAdminExecutionContext().callApi("xpack.watcher.stats", emptyMap(), emptyList(), emptyMap());
-                    String state = (String) response.evaluate("stats.0.watcher_state");
-                    if ("started".equals(state) == false) {
-                        getAdminExecutionContext().callApi("xpack.watcher.start", emptyMap(), emptyList(), emptyMap());
-                    }
-                    // assertion required to exit the assertBusy lambda
-                    assertThat(state, is("started"));
-                } catch (IOException e) {
-                    throw new AssertionError(e);
+                ClientYamlTestResponse response =
+                    getAdminExecutionContext().callApi("xpack.watcher.stats", emptyMap(), emptyList(), emptyMap());
+                String state = (String) response.evaluate("stats.0.watcher_state");
+
+                switch (state) {
+                    case "stopped":
+                        ClientYamlTestResponse startResponse =
+                            getAdminExecutionContext().callApi("xpack.watcher.start", emptyMap(), emptyList(), emptyMap());
+                        boolean isAcknowledged = (boolean) startResponse.evaluate("acknowledged");
+                        assertThat(isAcknowledged, is(true));
+                        break;
+                    case "stopping":
+                        throw new AssertionError("waiting until stopping state reached stopped state to start again");
+                    case "starting":
+                        throw new AssertionError("waiting until starting state reached started state");
+                    case "started":
+                        // all good here, we are done
+                        break;
+                    default:
+                        throw new AssertionError("unknown state[" + state + "]");
                 }
             });
+
+            for (String template : WatcherIndexTemplateRegistryField.TEMPLATE_NAMES) {
+                awaitCallApi("indices.exists_template", singletonMap("name", template), emptyList(),
+                    response -> true,
+                    () -> "Exception when waiting for [" + template + "] template to be created");
+            }
+
+            Map<String, String> params = MapBuilder.<String, String>newMapBuilder().put("size", "1000").put("ignore", "404").map();
+            Response response = adminClient().performRequest("GET", ".watches/_search", params);
+            ObjectPath objectPathResponse = ObjectPath.createFromResponse(response);
+            Integer totalHits = objectPathResponse.evaluate("hits.total");
+            if (totalHits != null && totalHits > 0) {
+                List<Map<String, Object>> hits = objectPathResponse.evaluate("hits.hits");
+                for (Map<String, Object> hit : hits) {
+                    String id = (String) hit.get("_id");
+                    assertOK(adminClient().performRequest("DELETE", "_xpack/watcher/watch/" + id));
+                }
+            }
         }
     }
 

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

@@ -370,7 +370,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
         final WatchParser watchParser = new WatchParser(settings, triggerService, registry, inputRegistry, cryptoService, getClock());
 
         final ExecutionService executionService = new ExecutionService(settings, historyStore, triggeredWatchStore, watchExecutor,
-                getClock(), watchParser, clusterService, client);
+                getClock(), watchParser, clusterService, client, threadPool.generic());
 
         final Consumer<Iterable<TriggerEvent>> triggerEngineListener = getTriggerEngineListener(executionService);
         triggerService.register(triggerEngineListener);
@@ -379,7 +379,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
                 watchParser, client);
 
         final WatcherLifeCycleService watcherLifeCycleService =
-                new WatcherLifeCycleService(settings, threadPool, clusterService, watcherService);
+                new WatcherLifeCycleService(settings, clusterService, watcherService);
 
         listener = new WatcherIndexingListener(settings, watchParser, getClock(), triggerService);
         clusterService.addListener(listener);

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

@@ -123,10 +123,10 @@ public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
                 boolean shouldBeTriggered = shardAllocationConfiguration.shouldBeTriggered(watch.id());
                 if (shouldBeTriggered) {
                     if (watch.status().state().isActive()) {
-                        logger.debug("adding watch [{}] to trigger", watch.id());
+                        logger.debug("adding watch [{}] to trigger service", watch.id());
                         triggerService.add(watch);
                     } else {
-                        logger.debug("removing watch [{}] to trigger", watch.id());
+                        logger.debug("removing watch [{}] to trigger service", watch.id());
                         triggerService.remove(watch.id());
                     }
                 } else {