Move watcher history to data stream

x-pack/docs/en/rest-api/watcher/execute-watch.asciidoc

@@ -5,7 +5,7 @@
 <titleabbrev>Execute watch</titleabbrev>
 ++++
 
-Forces the execution of a stored watch. 
+Forces the execution of a stored watch.
 
 [[watcher-api-execute-watch-request]]
 ==== {api-request-title}
@@ -189,6 +189,7 @@ This is an example of the output:
 {
   "_id": "my_watch_0-2015-06-02T23:17:55.124Z", <1>
   "watch_record": { <2>
+    "@timestamp": "2015-06-02T23:17:55.124Z",
     "watch_id": "my_watch",
     "node": "my_node",
     "messages": [],
@@ -276,6 +277,7 @@ This is an example of the output:
 --------------------------------------------------
 // TESTRESPONSE[s/my_watch_0-2015-06-02T23:17:55.124Z/$body._id/]
 // TESTRESPONSE[s/"triggered_time": "2015-06-02T23:17:55.124Z"/"triggered_time": "$body.watch_record.trigger_event.triggered_time"/]
+// TESTRESPONSE[s/"@timestamp": "2015-06-02T23:17:55.124Z"/"@timestamp": "$body.watch_record.trigger_event.triggered_time"/]
 // TESTRESPONSE[s/"scheduled_time": "2015-06-02T23:17:55.124Z"/"scheduled_time": "$body.watch_record.trigger_event.manual.schedule.scheduled_time"/]
 // TESTRESPONSE[s/"execution_time": "2015-06-02T23:17:55.124Z"/"execution_time": "$body.watch_record.result.execution_time"/]
 // TESTRESPONSE[s/"timestamp": "2015-06-02T23:17:55.111Z"/"timestamp": "$body.watch_record.status.state.timestamp"/]

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

@@ -5,21 +5,10 @@
  */
 package org.elasticsearch.xpack.core.watcher.history;
 
-import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.xpack.core.watcher.support.WatcherIndexTemplateRegistryField;
 
-import java.time.ZonedDateTime;
-
 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 + "-";
-    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 final String DATA_STREAM = INDEX_PREFIX + WatcherIndexTemplateRegistryField.INDEX_TEMPLATE_VERSION;
 }

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

@@ -154,6 +154,7 @@ public final XContentBuilder toXContent(XContentBuilder builder, Params params)
         builder.field(WATCH_ID.getPreferredName(), id.watchId());
         builder.field(NODE.getPreferredName(), nodeId);
         builder.field(STATE.getPreferredName(), state.id());
+        builder.field("@timestamp", triggerEvent.triggeredTime());
 
         if (user != null) {
             builder.field(USER.getPreferredName(), user);

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
     // version 12: templates changed to composable templates
     // version 13: add `allow_auto_create` setting
+    // version 14: move watch history to data stream
     // Note: if you change this, also inform the kibana team around the watcher-ui
-    public static final int INDEX_TEMPLATE_VERSION = 13;
+    public static final int INDEX_TEMPLATE_VERSION = 14;
     public static final String HISTORY_TEMPLATE_NAME = ".watch-history-" + INDEX_TEMPLATE_VERSION;
     public static final String HISTORY_TEMPLATE_NAME_NO_ILM = ".watch-history-no-ilm-" + INDEX_TEMPLATE_VERSION;
     public static final String TRIGGERED_TEMPLATE_NAME = ".triggered_watches";

x-pack/plugin/core/src/main/resources/watch-history-no-ilm.json

@@ -1,6 +1,9 @@
 {
   "index_patterns": [ ".watcher-history-${xpack.watcher.template.version}*" ],
   "priority": 2147483646,
+  "data_stream": {
+    "hidden":true
+  },
   "template": {
     "settings": {
       "index.number_of_shards": 1,

x-pack/plugin/core/src/main/resources/watch-history.json

@@ -1,6 +1,9 @@
 {
   "index_patterns": [ ".watcher-history-${xpack.watcher.template.version}*" ],
   "priority": 2147483647,
+  "data_stream": {
+    "hidden":true
+  },
   "template": {
     "settings": {
       "index.number_of_shards": 1,

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

@@ -178,8 +178,6 @@
 import org.elasticsearch.xpack.core.watcher.transport.actions.stats.WatcherStatsAction;
 import org.elasticsearch.xpack.core.watcher.watch.Watch;
 
-import java.time.ZoneOffset;
-import java.time.ZonedDateTime;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -1569,9 +1567,7 @@ public void testWatcherAdminRole() {
 
         assertThat(role.indices().allowedIndicesMatcher(IndexAction.NAME).test(mockIndexAbstraction("foo")), is(false));
 
-        ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
-        String historyIndex = HistoryStoreField.getHistoryIndexNameForTime(now);
-        for (String index : new String[]{ Watch.INDEX, historyIndex, TriggeredWatchStoreField.INDEX_NAME }) {
+        for (String index : new String[]{ Watch.INDEX, HistoryStoreField.DATA_STREAM, TriggeredWatchStoreField.INDEX_NAME }) {
             assertOnlyReadAllowed(role, index);
         }
 
@@ -1604,9 +1600,7 @@ public void testWatcherUserRole() {
         assertThat(role.indices().allowedIndicesMatcher(IndexAction.NAME).test(mockIndexAbstraction(TriggeredWatchStoreField.INDEX_NAME)),
                 is(false));
 
-        ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
-        String historyIndex = HistoryStoreField.getHistoryIndexNameForTime(now);
-        for (String index : new String[]{ Watch.INDEX, historyIndex }) {
+        for (String index : new String[]{ Watch.INDEX, HistoryStoreField.DATA_STREAM }) {
             assertOnlyReadAllowed(role, index);
         }
 

x-pack/plugin/watcher/build.gradle

@@ -29,6 +29,7 @@ dependencies {
 
   testImplementation project(path: xpackModule('core'), configuration: 'testArtifacts')
   testImplementation project(xpackModule('ilm'))
+  testImplementation project(xpackModule('data-streams'))
 
   // watcher deps
   api 'com.googlecode.owasp-java-html-sanitizer:owasp-java-html-sanitizer:20191001.1'

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/actions/TimeThrottleIntegrationTests.java

@@ -95,9 +95,9 @@ private void assertHistoryEntryThrottled(String id) {
     }
 
     private Map<String, Object> assertLatestHistoryEntry(String id) {
-        refresh(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*");
+        refresh(HistoryStoreField.DATA_STREAM + "*");
 
-        SearchResponse searchResponse = client().prepareSearch(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*")
+        SearchResponse searchResponse = client().prepareSearch(HistoryStoreField.DATA_STREAM + "*")
                 .setSize(1)
                 .setSource(new SearchSourceBuilder().query(QueryBuilders.boolQuery()
                         .must(termQuery("watch_id", id))))
@@ -111,7 +111,7 @@ private Map<String, Object> assertLatestHistoryEntry(String id) {
     }
 
     private void assertTotalHistoryEntries(String id, long expectedCount) {
-        SearchResponse searchResponse = client().prepareSearch(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*")
+        SearchResponse searchResponse = client().prepareSearch(HistoryStoreField.DATA_STREAM + "*")
                 .setSize(0)
                 .setSource(new SearchSourceBuilder().query(QueryBuilders.boolQuery().must(termQuery("watch_id", id))))
                 .get();

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/actions/email/EmailAttachmentTests.java

@@ -187,7 +187,7 @@ public void testThatEmailAttachmentsAreSent() throws Exception {
         timeWarp().trigger("_test_id");
         refresh();
 
-        SearchResponse searchResponse = client().prepareSearch(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*")
+        SearchResponse searchResponse = client().prepareSearch(HistoryStoreField.DATA_STREAM + "*")
                 .setQuery(QueryBuilders.termQuery("watch_id", "_test_id"))
                 .execute().actionGet();
         assertHitCount(searchResponse, 1);

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryTemplateEmailMappingsTests.java

@@ -88,7 +88,7 @@ public void testEmailFields() throws Exception {
         // the action should fail as no email server is available
         assertWatchWithMinimumActionsCount("_id", ExecutionState.EXECUTED, 1);
 
-        SearchResponse response = client().prepareSearch(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*").setSource(searchSource()
+        SearchResponse response = client().prepareSearch(HistoryStoreField.DATA_STREAM + "*").setSource(searchSource()
                 .aggregation(terms("from").field("result.actions.email.message.from"))
                 .aggregation(terms("to").field("result.actions.email.message.to"))
                 .aggregation(terms("cc").field("result.actions.email.message.cc"))

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryTemplateHttpMappingsTests.java

@@ -88,7 +88,7 @@ public void testHttpFields() throws Exception {
         // the action should fail as no email server is available
         assertWatchWithMinimumActionsCount("_id", ExecutionState.EXECUTED, 1);
 
-        SearchResponse response = client().prepareSearch(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*").setSource(searchSource()
+        SearchResponse response = client().prepareSearch(HistoryStoreField.DATA_STREAM + "*").setSource(searchSource()
                 .aggregation(terms("input_result_path").field("result.input.http.request.path"))
                 .aggregation(terms("input_result_host").field("result.input.http.request.host"))
                 .aggregation(terms("webhook_path").field("result.actions.webhook.request.path")))

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryTemplateIndexActionMappingsTests.java

@@ -47,7 +47,7 @@ public void testIndexActionFields() throws Exception {
         flush();
         refresh();
 
-        SearchResponse response = client().prepareSearch(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*").setSource(searchSource()
+        SearchResponse response = client().prepareSearch(HistoryStoreField.DATA_STREAM + "*").setSource(searchSource()
                 .aggregation(terms("index_action_indices").field("result.actions.index.response.index")))
                 .get();
 

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/history/HistoryTemplateSearchInputMappingsTests.java

@@ -60,7 +60,7 @@ WatcherSearchTemplateRequest.DEFAULT_INDICES_OPTIONS, new BytesArray("{}")
         // the action should fail as no email server is available
         assertWatchWithMinimumActionsCount("_id", ExecutionState.EXECUTED, 1);
 
-        SearchResponse response = client().prepareSearch(HistoryStoreField.INDEX_PREFIX_WITH_TEMPLATE + "*").setSource(searchSource()
+        SearchResponse response = client().prepareSearch(HistoryStoreField.DATA_STREAM + "*").setSource(searchSource()
                 .aggregation(terms("input_search_type").field("result.input.search.request.search_type"))
                 .aggregation(terms("input_indices").field("result.input.search.request.indices"))
                 .aggregation(terms("input_body").field("result.input.search.request.body")))