Remove the index type from internal watcher indexes (elastic#39761)

This commit removes the "doc" type from watcher internal indexes.
The template still carries the "_doc" type since that is needed for
the internal representation.

This impacts the .watches, .triggered-watches, and .watch-history indexes.

External consumers do not need any changes since all external calls
go through the _watcher API, and should not interact with the the .index directly.

Relates elastic#38637

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/execution/TriggeredWatchStoreField.java

@@ -8,5 +8,4 @@
 public final class TriggeredWatchStoreField {
 
     public static final String INDEX_NAME = ".triggered_watches";
-    public static final String DOC_TYPE = "doc";
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/watch/Watch.java

@@ -26,7 +26,6 @@ public class Watch implements ToXContentObject {
 
     public static final String INCLUDE_STATUS_KEY = "include_status";
     public static final String INDEX = ".watches";
-    public static final String DOC_TYPE = "doc";
 
     private final String id;
     private final Trigger trigger;

x-pack/plugin/core/src/main/resources/triggered-watches.json

@@ -9,7 +9,7 @@
     "index.priority": 900
   },
   "mappings": {
-    "doc": {
+    "_doc": {
       "dynamic" : "strict",
       "properties": {
         "trigger_event": {

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

@@ -9,7 +9,7 @@
     "index.format": 6
   },
   "mappings": {
-    "doc": {
+    "_doc": {
       "_meta": {
         "watcher-history-version": "${xpack.watcher.template.version}"
       },

x-pack/plugin/core/src/main/resources/watches.json

@@ -9,7 +9,7 @@
     "index.priority": 800
   },
   "mappings": {
-    "doc": {
+    "_doc": {
       "dynamic" : "strict",
       "properties": {
         "status": {

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

@@ -101,7 +101,7 @@ void setConfiguration(Configuration configuration) {
      */
     @Override
     public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
-        if (isWatchDocument(shardId.getIndexName(), operation.type())) {
+        if (isWatchDocument(shardId.getIndexName())) {
             ZonedDateTime now = Instant.ofEpochMilli(clock.millis()).atZone(ZoneOffset.UTC);
             try {
                 Watch watch = parser.parseWithSecrets(operation.id(), true, operation.source(), now, XContentType.JSON,
@@ -150,7 +150,7 @@ public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
      */
     @Override
     public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
-        if (isWatchDocument(shardId.getIndexName(), index.type())) {
+        if (isWatchDocument(shardId.getIndexName())) {
             logger.debug(() -> new ParameterizedMessage("removing watch [{}] from trigger", index.id()), ex);
             triggerService.remove(index.id());
         }
@@ -166,7 +166,7 @@ public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
      */
     @Override
     public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
-        if (isWatchDocument(shardId.getIndexName(), delete.type())) {
+        if (isWatchDocument(shardId.getIndexName())) {
             triggerService.remove(delete.id());
         }
 
@@ -177,11 +177,10 @@ public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
      * Check if a supplied index and document matches the current configuration for watcher
      *
      * @param index   The index to check for
-     * @param docType The document type
      * @return true if this is a watch in the active watcher index, false otherwise
      */
-    private boolean isWatchDocument(String index, String docType) {
-        return configuration.isIndexAndActive(index) && docType.equals(Watch.DOC_TYPE);
+    private boolean isWatchDocument(String index) {
+        return configuration.isIndexAndActive(index);
     }
 
     /**

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

@@ -34,10 +34,18 @@ public static EmailAction.Builder emailAction(EmailTemplate email) {
         return EmailAction.builder(email);
     }
 
+    /**
+     * Types are deprecated and should not be used. use {@link #indexAction(String)}
+     */
+    @Deprecated
     public static IndexAction.Builder indexAction(String index, String type) {
         return IndexAction.builder(index, type);
     }
 
+    public static IndexAction.Builder indexAction(String index) {
+        return IndexAction.builder(index);
+    }
+
     public static JiraAction.Builder jiraAction(String account, MapBuilder<String, Object> fields) {
         return jiraAction(account, fields.immutableMap());
     }

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/actions/index/IndexAction.java

@@ -29,7 +29,7 @@ public class IndexAction implements Action {
 
     public static final String TYPE = "index";
 
-    @Nullable final String docType;
+    @Nullable @Deprecated final String docType;
     @Nullable final String index;
     @Nullable final String docId;
     @Nullable final String executionTimeField;
@@ -40,6 +40,15 @@ public class IndexAction implements Action {
     private static final DeprecationLogger deprecationLogger = new DeprecationLogger(LogManager.getLogger(IndexAction.class));
     public static final String TYPES_DEPRECATION_MESSAGE = "[types removal] Specifying types in a watcher index action is deprecated.";
 
+    public IndexAction(@Nullable String index, @Nullable String docId,
+                       @Nullable String executionTimeField,
+                       @Nullable TimeValue timeout, @Nullable ZoneId dynamicNameTimeZone, @Nullable RefreshPolicy refreshPolicy) {
+        this(index, null, docId, executionTimeField, timeout, dynamicNameTimeZone, refreshPolicy);
+    }
+    /**
+     * Document types are deprecated, use constructor without docType
+     */
+    @Deprecated
     public IndexAction(@Nullable String index, @Nullable String docType, @Nullable String docId,
                        @Nullable String executionTimeField,
                        @Nullable TimeValue timeout, @Nullable ZoneId dynamicNameTimeZone, @Nullable RefreshPolicy refreshPolicy) {
@@ -188,10 +197,18 @@ public static IndexAction parse(String watchId, String actionId, XContentParser
         return new IndexAction(index, docType, docId, executionTimeField, timeout, dynamicNameTimeZone, refreshPolicy);
     }
 
+    /**
+     * Document types are deprecated, use {@link #builder(java.lang.String)}
+     */
+    @Deprecated
     public static Builder builder(String index, String docType) {
         return new Builder(index, docType);
     }
 
+    public static Builder builder(String index) {
+        return new Builder(index);
+    }
+
     public static class Result extends Action.Result {
 
         private final XContentSource response;
@@ -278,11 +295,20 @@ public static class Builder implements Action.Builder<IndexAction> {
         ZoneId dynamicNameTimeZone;
         RefreshPolicy refreshPolicy;
 
+        /**
+         * Document types are deprecated and should not be used. Use: {@link Builder#Builder(java.lang.String)}
+         */
+        @Deprecated
         private Builder(String index, String docType) {
             this.index = index;
             this.docType = docType;
         }
 
+        private Builder(String index) {
+            this.index = index;
+            this.docType = null;
+        }
+
         public Builder setDocId(String docId) {
             this.docId = docId;
             return this;

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

@@ -352,7 +352,7 @@ public void updateWatchStatus(Watch watch) throws IOException {
             .field(WatchField.STATUS.getPreferredName(), watch.status(), params)
             .endObject();
 
-        UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, watch.id());
+        UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, watch.id());
         updateRequest.doc(source);
         updateRequest.setIfSeqNo(watch.getSourceSeqNo());
         updateRequest.setIfPrimaryTerm(watch.getSourcePrimaryTerm());
@@ -501,7 +501,7 @@ public void executeTriggeredWatches(Collection<TriggeredWatch> triggeredWatches)
      */
     private GetResponse getWatch(String id) {
         try (ThreadContext.StoredContext ignore = stashWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN)) {
-            GetRequest getRequest = new GetRequest(Watch.INDEX, Watch.DOC_TYPE, id).preference(Preference.LOCAL.type()).realtime(true);
+            GetRequest getRequest = new GetRequest(Watch.INDEX, id).preference(Preference.LOCAL.type()).realtime(true);
             PlainActionFuture<GetResponse> future = PlainActionFuture.newFuture();
             client.get(getRequest, future);
             return future.actionGet();

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

@@ -96,8 +96,7 @@ public BulkResponse putAll(final List<TriggeredWatch> triggeredWatches) throws I
     private BulkRequest createBulkRequest(final List<TriggeredWatch> triggeredWatches) throws IOException {
         BulkRequest request = new BulkRequest();
         for (TriggeredWatch triggeredWatch : triggeredWatches) {
-            IndexRequest indexRequest = new IndexRequest(TriggeredWatchStoreField.INDEX_NAME, TriggeredWatchStoreField.DOC_TYPE,
-                triggeredWatch.id().value());
+            IndexRequest indexRequest = new IndexRequest(TriggeredWatchStoreField.INDEX_NAME).id(triggeredWatch.id().value());
             try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
                 triggeredWatch.toXContent(builder, ToXContent.EMPTY_PARAMS);
                 indexRequest.source(builder);
@@ -115,7 +114,7 @@ private BulkRequest createBulkRequest(final List<TriggeredWatch> triggeredWatche
      * @param wid The ID os the triggered watch id
      */
     public void delete(Wid wid) {
-        DeleteRequest request = new DeleteRequest(TriggeredWatchStoreField.INDEX_NAME, TriggeredWatchStoreField.DOC_TYPE, wid.value());
+        DeleteRequest request = new DeleteRequest(TriggeredWatchStoreField.INDEX_NAME, wid.value());
         bulkProcessor.add(request);
     }
 

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

@@ -28,8 +28,6 @@
 
 public class HistoryStore {
 
-    public static final String DOC_TYPE = "doc";
-
     private static final Logger logger = LogManager.getLogger(HistoryStore.class);
 
     private final BulkProcessor bulkProcessor;
@@ -47,7 +45,7 @@ public void put(WatchRecord watchRecord) throws Exception {
         try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
             watchRecord.toXContent(builder, WatcherParams.HIDE_SECRETS);
 
-            IndexRequest request = new IndexRequest(index, DOC_TYPE, watchRecord.id().value()).source(builder);
+            IndexRequest request = new IndexRequest(index).id(watchRecord.id().value()).source(builder);
             request.opType(IndexRequest.OpType.CREATE);
             bulkProcessor.add(request);
         } catch (IOException ioe) {
@@ -64,7 +62,7 @@ public void forcePut(WatchRecord watchRecord) {
             try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
                 watchRecord.toXContent(builder, WatcherParams.HIDE_SECRETS);
 
-                IndexRequest request = new IndexRequest(index, DOC_TYPE, watchRecord.id().value()).source(builder);
+                IndexRequest request = new IndexRequest(index).id(watchRecord.id().value()).source(builder);
                 bulkProcessor.add(request);
         } catch (IOException ioe) {
             final WatchRecord wr = watchRecord;

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/ack/TransportAckWatchAction.java

@@ -75,7 +75,7 @@ protected void doExecute(AckWatchRequest request, ActionListener<AckWatchRespons
                 listener.onFailure(new ElasticsearchStatusException("watch[{}] is running currently, cannot ack until finished",
                     RestStatus.CONFLICT, request.getWatchId()));
             } else {
-                GetRequest getRequest = new GetRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId())
+                GetRequest getRequest = new GetRequest(Watch.INDEX, request.getWatchId())
                     .preference(Preference.LOCAL.type()).realtime(true);
 
                 executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, getRequest,
@@ -99,7 +99,7 @@ protected void doExecute(AckWatchRequest request, ActionListener<AckWatchRespons
                                 return;
                             }
 
-                            UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId());
+                            UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, request.getWatchId());
                             // this may reject this action, but prevents concurrent updates from a watch execution
                             updateRequest.setIfSeqNo(getResponse.getSeqNo());
                             updateRequest.setIfPrimaryTerm(getResponse.getPrimaryTerm());

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/activate/TransportActivateWatchAction.java

@@ -61,7 +61,7 @@ public TransportActivateWatchAction(TransportService transportService, ActionFil
     protected void doExecute(ActivateWatchRequest request, ActionListener<ActivateWatchResponse> listener) {
         try {
             ZonedDateTime now = clock.instant().atZone(ZoneOffset.UTC);
-            UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId());
+            UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, request.getWatchId());
             updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
             XContentBuilder builder = activateWatchBuilder(request.isActivate(), now);
             updateRequest.doc(builder);
@@ -72,7 +72,7 @@ protected void doExecute(ActivateWatchRequest request, ActionListener<ActivateWa
 
             executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, updateRequest,
                     ActionListener.<UpdateResponse>wrap(updateResponse -> {
-                GetRequest getRequest = new GetRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId())
+                GetRequest getRequest = new GetRequest(Watch.INDEX, request.getWatchId())
                         .preference(Preference.LOCAL.type()).realtime(true);
 
                 executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, getRequest,

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/delete/TransportDeleteWatchAction.java

@@ -42,7 +42,7 @@ public TransportDeleteWatchAction(TransportService transportService, ActionFilte
 
     @Override
     protected void doExecute(Task task, DeleteWatchRequest request, ActionListener<DeleteWatchResponse> listener) {
-        DeleteRequest deleteRequest = new DeleteRequest(Watch.INDEX, Watch.DOC_TYPE, request.getId());
+        DeleteRequest deleteRequest = new DeleteRequest(Watch.INDEX, request.getId());
         deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
         executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, deleteRequest,
                 ActionListener.<DeleteResponse>wrap(deleteResponse -> {

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/execute/TransportExecuteWatchAction.java

@@ -81,7 +81,7 @@ public TransportExecuteWatchAction(TransportService transportService, ThreadPool
     @Override
     protected void doExecute(ExecuteWatchRequest request, ActionListener<ExecuteWatchResponse> listener) {
         if (request.getId() != null) {
-            GetRequest getRequest = new GetRequest(Watch.INDEX, Watch.DOC_TYPE, request.getId())
+            GetRequest getRequest = new GetRequest(Watch.INDEX, request.getId())
                     .preference(Preference.LOCAL.type()).realtime(true);
 
             executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, getRequest,

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/get/TransportGetWatchAction.java

@@ -52,7 +52,7 @@ public TransportGetWatchAction(TransportService transportService, ActionFilters
 
     @Override
     protected void doExecute(GetWatchRequest request, ActionListener<GetWatchResponse> listener) {
-        GetRequest getRequest = new GetRequest(Watch.INDEX, Watch.DOC_TYPE, request.getId())
+        GetRequest getRequest = new GetRequest(Watch.INDEX, request.getId())
                 .preference(Preference.LOCAL.type()).realtime(true);
 
         executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, getRequest,

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/put/TransportPutWatchAction.java

@@ -94,7 +94,7 @@ protected void doExecute(PutWatchRequest request, ActionListener<PutWatchRespons
                 watch.toXContent(builder, DEFAULT_PARAMS);
 
                 if (isUpdate) {
-                    UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, request.getId());
+                    UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, request.getId());
                     if (request.getIfSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                         updateRequest.setIfSeqNo(request.getIfSeqNo());
                         updateRequest.setIfPrimaryTerm(request.getIfPrimaryTerm());
@@ -112,7 +112,7 @@ protected void doExecute(PutWatchRequest request, ActionListener<PutWatchRespons
                             }, listener::onFailure),
                             client::update);
                 } else {
-                    IndexRequest indexRequest = new IndexRequest(Watch.INDEX, Watch.DOC_TYPE, request.getId());
+                    IndexRequest indexRequest = new IndexRequest(Watch.INDEX).id(request.getId());
                     indexRequest.source(builder);
                     indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
                     executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, indexRequest,

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

@@ -34,7 +34,7 @@ public void testCanUseAnyConcreteIndexName() throws Exception {
         createIndex(watchResultsIndex);
 
         stopWatcher();
-        replaceWatcherIndexWithRandomlyNamedIndex(Watch.INDEX, newWatcherIndexName, Watch.DOC_TYPE);
+        replaceWatcherIndexWithRandomlyNamedIndex(Watch.INDEX, newWatcherIndexName);
         startWatcher();
 
         PutWatchResponse putWatchResponse = watcherClient().preparePutWatch("mywatch").setSource(watchBuilder()