When Watcher starts verify that all index templates have been added, otherwise add the index templates

There maybe a small window of time just after Watcher has started that the index templates don't exist.

Original commit: elastic/x-pack-elasticsearch@9af3018be4

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

@@ -16,6 +16,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
 import org.elasticsearch.watcher.execution.ExecutionService;
+import org.elasticsearch.watcher.support.WatcherIndexTemplateRegistry;
 import org.elasticsearch.watcher.support.clock.Clock;
 import org.elasticsearch.watcher.trigger.TriggerService;
 import org.elasticsearch.watcher.watch.Watch;
@@ -39,11 +40,12 @@ public class WatcherService extends AbstractComponent {
     private final WatchStore watchStore;
     private final WatchLockService watchLockService;
     private final ExecutionService executionService;
+    private final WatcherIndexTemplateRegistry watcherIndexTemplateRegistry;
     private final AtomicReference<WatcherState> state = new AtomicReference<>(WatcherState.STOPPED);
 
     @Inject
     public WatcherService(Settings settings, Clock clock, TriggerService triggerService, WatchStore watchStore,
-                          Watch.Parser watchParser, ExecutionService executionService, WatchLockService watchLockService) {
+                          Watch.Parser watchParser, ExecutionService executionService, WatchLockService watchLockService, WatcherIndexTemplateRegistry watcherIndexTemplateRegistry) {
         super(settings);
         this.clock = clock;
         this.triggerService = triggerService;
@@ -51,12 +53,14 @@ public class WatcherService extends AbstractComponent {
         this.watchParser = watchParser;
         this.watchLockService = watchLockService;
         this.executionService = executionService;
+        this.watcherIndexTemplateRegistry = watcherIndexTemplateRegistry;
     }
 
     public void start(ClusterState clusterState) throws Exception {
         if (state.compareAndSet(WatcherState.STOPPED, WatcherState.STARTING)) {
             try {
                 logger.info("starting watch service...");
+                watcherIndexTemplateRegistry.addTemplatesIfMissing();
                 watchLockService.start();
 
                 // Try to load watch store before the execution service, b/c action depends on watch store

src/main/java/org/elasticsearch/watcher/support/WatcherIndexTemplateRegistry.java

@@ -80,9 +80,23 @@ public class WatcherIndexTemplateRegistry extends AbstractComponent implements C
             return;
         }
 
+        putTemplatesIfMissing(state);
+    }
+
+    /**
+     * Adds the registered index templates if missing to the cluster.
+     */
+    public void addTemplatesIfMissing() {
+        putTemplatesIfMissing(clusterService.state());
+    }
+
+    synchronized void putTemplatesIfMissing(ClusterState state) {
         for (TemplateConfig template : indexTemplates) {
             if (!state.metaData().getTemplates().containsKey(template.getTemplateName())) {
+                logger.debug("adding index template [{}], because it doesn't exist", template.getTemplateName());
                 putTemplate(template);
+            } else {
+                logger.trace("not adding index template [{}], because it already exists", template.getTemplateName());
             }
         }
     }

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.watcher.test;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.admin.indices.template.get.GetIndexTemplatesResponse;
 import org.elasticsearch.action.search.SearchRequestBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchType;
@@ -30,10 +31,7 @@ import org.elasticsearch.shield.crypto.InternalCryptoService;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.elasticsearch.test.ElasticsearchIntegrationTest.ClusterScope;
 import org.elasticsearch.test.TestCluster;
-import org.elasticsearch.watcher.WatcherLifeCycleService;
+import org.elasticsearch.watcher.*;
-import org.elasticsearch.watcher.WatcherPlugin;
-import org.elasticsearch.watcher.WatcherService;
-import org.elasticsearch.watcher.WatcherState;
 import org.elasticsearch.watcher.actions.email.service.Authentication;
 import org.elasticsearch.watcher.actions.email.service.Email;
 import org.elasticsearch.watcher.actions.email.service.EmailService;
@@ -71,6 +69,7 @@ import java.util.Map;
 import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
 import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
 import static org.elasticsearch.test.ElasticsearchIntegrationTest.Scope.SUITE;
+import static org.elasticsearch.watcher.WatcherModule.*;
 import static org.hamcrest.Matchers.*;
 import static org.hamcrest.core.Is.is;
 import static org.hamcrest.core.IsNot.not;
@@ -433,6 +432,11 @@ public abstract class AbstractWatcherIntegrationTests extends ElasticsearchInteg
                 }
             }
         });
+        // Verify that the index templates exist:
+        GetIndexTemplatesResponse response = client().admin().indices().prepareGetTemplates(
+                HISTORY_TEMPLATE_NAME, TRIGGERED_TEMPLATE_NAME, WATCHES_TEMPLATE_NAME
+        ).get();
+        assertThat(response.getIndexTemplates().size(), equalTo(3));
     }
 
     protected void ensureLicenseEnabled()  throws Exception {

src/test/java/org/elasticsearch/watcher/watch/WatchServiceTests.java

@@ -19,6 +19,7 @@ import org.elasticsearch.watcher.WatcherService;
 import org.elasticsearch.watcher.WatcherState;
 import org.elasticsearch.watcher.actions.ActionStatus;
 import org.elasticsearch.watcher.execution.ExecutionService;
+import org.elasticsearch.watcher.support.WatcherIndexTemplateRegistry;
 import org.elasticsearch.watcher.support.clock.ClockMock;
 import org.elasticsearch.watcher.support.clock.SystemClock;
 import org.elasticsearch.watcher.trigger.Trigger;
@@ -45,7 +46,6 @@ public class WatchServiceTests extends ElasticsearchTestCase {
     private WatchStore watchStore;
     private Watch.Parser watchParser;
     private WatcherService watcherService;
-    private ExecutionService executionService;
     private WatchLockService watchLockService;
     private ClockMock clock;
 
@@ -54,10 +54,11 @@ public class WatchServiceTests extends ElasticsearchTestCase {
         triggerService = mock(TriggerService.class);
         watchStore = mock(WatchStore.class);
         watchParser = mock(Watch.Parser.class);
-        executionService =  mock(ExecutionService.class);
+        ExecutionService executionService =  mock(ExecutionService.class);
         watchLockService = mock(WatchLockService.class);
         clock = new ClockMock();
-        watcherService = new WatcherService(Settings.EMPTY, clock, triggerService, watchStore, watchParser, executionService, watchLockService);
+        WatcherIndexTemplateRegistry watcherIndexTemplateRegistry = mock(WatcherIndexTemplateRegistry.class);
+        watcherService = new WatcherService(Settings.EMPTY, clock, triggerService, watchStore, watchParser, executionService, watchLockService, watcherIndexTemplateRegistry);
         Field field = WatcherService.class.getDeclaredField("state");
         field.setAccessible(true);
         AtomicReference<WatcherState> state = (AtomicReference<WatcherState>) field.get(watcherService);