Watcher: Execute Watch API returns better error message (elastic/x-pack-elasticsearch#4228)

Many users take the JSON from an PUT watch API and put it into the
execute watch API and then start to wonder why there is an error, as
they forget to wrap the watch inside a 'watch' field.

This commit adds a better error message in this case hinting at the user
to add a 'watch' field.

Original commit: elastic/x-pack-elasticsearch@5b56b4abad

plugin/src/test/resources/rest-api-spec/test/watcher/execute_watch/70_invalid.yml

@@ -0,0 +1,47 @@
+---
+"Test execute watch api returns proper error message with watch directly in the body":
+  - do:
+      catch: /please wrap watch including field \[trigger\] inside a \"watch\" field/
+      xpack.watcher.execute_watch:
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            }
+          }
+  - do:
+      catch: /please wrap watch including field \[input\] inside a \"watch\" field/
+      xpack.watcher.execute_watch:
+        body:  >
+          {
+            "input": {
+              "simple": {
+                "foo": "bar"
+              }
+            }
+          }
+  - do:
+      catch: /please wrap watch including field \[condition\] inside a \"watch\" field/
+      xpack.watcher.execute_watch:
+        body:  >
+          {
+            "condition": {
+              "always": {}
+            }
+          }
+  - do:
+      catch: /please wrap watch including field \[actions\] inside a \"watch\" field/
+      xpack.watcher.execute_watch:
+        body:  >
+          {
+            "actions": {
+              "indexme" : {
+                "index" : {
+                  "index" : "my_test_index",
+                  "doc_type" : "my-type",
+                  "doc_id": "my-id"
+                }
+              }
+            }
+          }
+

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/rest/action/RestExecuteWatchAction.java

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.watcher.rest.action;
 
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.set.Sets;
@@ -26,10 +27,13 @@ import org.elasticsearch.xpack.core.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.xpack.core.watcher.transport.actions.execute.ExecuteWatchRequest;
 import org.elasticsearch.xpack.core.watcher.transport.actions.execute.ExecuteWatchRequestBuilder;
 import org.elasticsearch.xpack.core.watcher.transport.actions.execute.ExecuteWatchResponse;
+import org.elasticsearch.xpack.core.watcher.watch.WatchField;
 import org.elasticsearch.xpack.watcher.rest.WatcherRestHandler;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.POST;
@@ -39,6 +43,13 @@ import static org.elasticsearch.xpack.watcher.rest.action.RestExecuteWatchAction
 
 public class RestExecuteWatchAction extends WatcherRestHandler implements RestRequestFilter {
 
+    private static final List<String> RESERVED_FIELD_NAMES = Arrays.asList(WatchField.TRIGGER.getPreferredName(),
+            WatchField.INPUT.getPreferredName(), WatchField.CONDITION.getPreferredName(),
+            WatchField.ACTIONS.getPreferredName(), WatchField.TRANSFORM.getPreferredName(),
+            WatchField.THROTTLE_PERIOD.getPreferredName(), WatchField.THROTTLE_PERIOD_HUMAN.getPreferredName(),
+            WatchField.METADATA.getPreferredName(), WatchField.STATUS.getPreferredName(),
+            WatchField.VERSION.getPreferredName());
+
     public RestExecuteWatchAction(Settings settings, RestController controller) {
         super(settings);
         controller.registerHandler(POST, URI_BASE + "/watch/{id}/_execute", this);
@@ -104,9 +115,10 @@ public class RestExecuteWatchAction extends WatcherRestHandler implements RestRe
                     } else if (Field.TRIGGER_DATA.match(currentFieldName, parser.getDeprecationHandler())) {
                         builder.setTriggerData(parser.map());
                     } else if (Field.WATCH.match(currentFieldName, parser.getDeprecationHandler())) {
-                        XContentBuilder watcherSource = XContentBuilder.builder(parser.contentType().xContent());
+                        try (XContentBuilder watcherSource = XContentBuilder.builder(parser.contentType().xContent())) {
                             watcherSource.generator().copyCurrentStructure(parser);
                             builder.setWatchSource(BytesReference.bytes(watcherSource), parser.contentType());
+                        }
                     } else if (Field.ACTION_MODES.match(currentFieldName, parser.getDeprecationHandler())) {
                         while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                             if (token == XContentParser.Token.FIELD_NAME) {
@@ -124,10 +136,15 @@ public class RestExecuteWatchAction extends WatcherRestHandler implements RestRe
                                         currentFieldName);
                             }
                         }
+                    } else {
+                        if (RESERVED_FIELD_NAMES.contains(currentFieldName)) {
+                            throw new ElasticsearchParseException("please wrap watch including field [{}] inside a \"watch\" field",
+                                    currentFieldName);
                         } else {
                             throw new ElasticsearchParseException("could not parse watch execution request. unexpected object field [{}]",
                                     currentFieldName);
                         }
+                    }
                 } else {
                     throw new ElasticsearchParseException("could not parse watch execution request. unexpected token [{}]", token);
                 }