Added a setting to disable watcher

The following node setting will disable watcher (it's enabled by default).

```
watcher.enabled: false
```

Original commit: elastic/x-pack-elasticsearch@1d0541a924

src/main/java/org/elasticsearch/watcher/WatcherPlugin.java

@@ -24,13 +24,16 @@ import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilde
 public class WatcherPlugin extends AbstractPlugin {
 
     public static final String NAME = "watcher";
+    public static final String ENABLED_SETTING = NAME + ".enabled";
 
     private final Settings settings;
     private final boolean transportClient;
+    protected final boolean enabled;
 
     public WatcherPlugin(Settings settings) {
         this.settings = settings;
         transportClient = "transport".equals(settings.get(Client.CLIENT_TYPE_SETTING));
+        enabled = watcherEnabled(settings);
     }
 
     @Override public String name() {
@@ -43,6 +46,9 @@ public class WatcherPlugin extends AbstractPlugin {
 
     @Override
     public Collection<Class<? extends Module>> modules() {
+        if (!enabled) {
+            return ImmutableList.of();
+        }
         return transportClient ?
                 ImmutableList.<Class<? extends Module>>of(TransportClientWatcherModule.class) :
                 ImmutableList.<Class<? extends Module>>of(WatcherModule.class);
@@ -50,7 +56,7 @@ public class WatcherPlugin extends AbstractPlugin {
 
     @Override
     public Collection<Class<? extends LifecycleComponent>> services() {
-        if (transportClient) {
+        if (!enabled || transportClient) {
             return ImmutableList.of();
         }
         return ImmutableList.<Class<? extends LifecycleComponent>>of(
@@ -64,7 +70,7 @@ public class WatcherPlugin extends AbstractPlugin {
 
     @Override
     public Settings additionalSettings() {
-        if (transportClient) {
+        if (!enabled || transportClient) {
             return ImmutableSettings.EMPTY;
         }
         Settings additionalSettings = settingsBuilder()
@@ -74,4 +80,8 @@ public class WatcherPlugin extends AbstractPlugin {
         return additionalSettings;
     }
 
+    public static boolean watcherEnabled(Settings settings) {
+        return settings.getAsBoolean(ENABLED_SETTING, true);
+    }
+
 }

src/test/java/org/elasticsearch/watcher/WatcherPluginDisableTests.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher;
+
+import org.apache.http.HttpStatus;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
+import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.http.HttpServerTransport;
+import org.elasticsearch.license.plugin.LicensePlugin;
+import org.elasticsearch.node.internal.InternalNode;
+import org.elasticsearch.test.ElasticsearchIntegrationTest;
+import org.elasticsearch.test.ElasticsearchIntegrationTest.ClusterScope;
+import org.elasticsearch.test.rest.client.http.HttpRequestBuilder;
+import org.elasticsearch.test.rest.client.http.HttpResponse;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.threadpool.ThreadPoolInfo;
+import org.elasticsearch.watcher.execution.InternalWatchExecutor;
+import org.junit.Test;
+
+import static org.elasticsearch.test.ElasticsearchIntegrationTest.Scope.SUITE;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+
+/**
+ *
+ */
+@ClusterScope(scope = SUITE, numClientNodes = 0, transportClientRatio = 0, randomDynamicTemplates = false, maxNumDataNodes = 3)
+public class WatcherPluginDisableTests extends ElasticsearchIntegrationTest {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return ImmutableSettings.settingsBuilder()
+                .put(super.nodeSettings(nodeOrdinal))
+                .put("plugin.types", WatcherPlugin.class.getName() + "," + LicensePlugin.class.getName())
+                .put(WatcherPlugin.ENABLED_SETTING, false)
+                .put(InternalNode.HTTP_ENABLED, true)
+                .build();
+    }
+
+    @Test
+    public void testRestEndpoints() throws Exception {
+        HttpServerTransport httpServerTransport = internalCluster().getDataNodeInstance(HttpServerTransport.class);
+        try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
+            HttpRequestBuilder request = new HttpRequestBuilder(httpClient).httpTransport(httpServerTransport).method("GET").path("/_watcher");
+            HttpResponse response = request.execute();
+            assertThat(response.getStatusCode(), is(HttpStatus.SC_BAD_REQUEST));
+        }
+    }
+
+    @Test
+    public void testThreadPools() throws Exception {
+        NodesInfoResponse nodesInfo = client().admin().cluster().prepareNodesInfo().setThreadPool(true).get();
+        for (NodeInfo nodeInfo : nodesInfo) {
+            ThreadPoolInfo threadPoolInfo = nodeInfo.getThreadPool();
+            for (ThreadPool.Info info : threadPoolInfo) {
+                assertThat(info.getName(), not(is(InternalWatchExecutor.THREAD_POOL_NAME)));
+            }
+        }
+    }
+}

src/test/java/org/elasticsearch/watcher/test/AbstractWatcherIntegrationTests.java

@@ -28,6 +28,8 @@ import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.shield.ShieldPlugin;
 import org.elasticsearch.shield.authc.esusers.ESUsersRealm;
+import org.elasticsearch.shield.authc.support.SecuredString;
+import org.elasticsearch.shield.authc.support.UsernamePasswordToken;
 import org.elasticsearch.shield.crypto.InternalCryptoService;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.elasticsearch.test.ElasticsearchIntegrationTest.ClusterScope;
@@ -602,6 +604,8 @@ public abstract class AbstractWatcherIntegrationTests extends ElasticsearchInteg
 
         public static final String TEST_USERNAME = "test";
         public static final String TEST_PASSWORD = "changeme";
+        public static final String TEST_BASIC_AUTH_TOKEN = UsernamePasswordToken.basicAuthHeaderValue(TEST_USERNAME, new SecuredString(TEST_PASSWORD.toCharArray()));
+        public static final String ADMIN_BASIC_AUTH_TOKEN = UsernamePasswordToken.basicAuthHeaderValue("admin", new SecuredString("changeme".toCharArray()));
 
         static boolean auditLogsEnabled = SystemPropertyUtil.getBoolean("tests.audit_logs", true);
         static byte[] systemKey = generateKey(); // must be the same for all nodes

src/test/java/org/elasticsearch/watcher/test/TimeWarpedWatcherPlugin.java

@@ -39,6 +39,9 @@ public class TimeWarpedWatcherPlugin extends WatcherPlugin {
 
     @Override
     public Collection<Class<? extends Module>> modules() {
+        if (!enabled) {
+            return super.modules();
+        }
         return ImmutableList.<Class<? extends Module>>of(WatcherModule.class);
     }