Watcher: Allow to execute actions for each element in array

This adds the ability to execute an action for each element that occurs
in an array, for example you could sent a dedicated slack action for
each search hit returned from a search.

Relates elastic#34546

Author: spinscale

x-pack/docs/en/watcher/actions.asciidoc

@@ -192,6 +192,48 @@ of a watch during its execution:
 image::images/action-throttling.jpg[align="center"]
 
 
+[[action-foreach]]
+=== Running an action for each element in an array
+
+You can use the `foreach` field in an action to trigger the configured action
+for every element within that array. Each element within that array has to be a
+map.
+
+[source,js]
+--------------------------------------------------
+PUT _watcher/watch/log_event_watch
+{
+  "trigger" : {
+    "schedule" : { "interval" : "5m" }
+  },
+  "input" : {
+    "search" : {
+      "request" : {
+        "indices" : "log-events",
+        "body" : {
+          "size" : 0,
+          "query" : { "match" : { "status" : "error" } }
+        }
+      }
+    }
+  },
+  "condition" : {
+    "compare" : { "ctx.payload.hits.total.value" : { "gt" : 0 } }
+  },
+  "actions" : {
+    "log_hits" : {
+      "foreach" : "ctx.payload.hits.hits", <1>
+      "logging" : {
+        "text" : "Found id {{_id}} with field {{_source.my_field}}"
+      }
+    }
+  }
+}
+--------------------------------------------------
+// CONSOLE
+
+<1> The logging statement will be executed for each of the returned search hits.
+
 [[action-conditions]]
 === Adding conditions to actions
 

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

@@ -7,9 +7,12 @@
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.ObjectPath;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
@@ -30,7 +33,13 @@
 import java.time.Clock;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.unit.TimeValue.timeValueMillis;
 
@@ -43,16 +52,20 @@ public class ActionWrapper implements ToXContentObject {
     private final ExecutableTransform<Transform, Transform.Result> transform;
     private final ActionThrottler throttler;
     private final ExecutableAction<? extends Action> action;
+    @Nullable
+    private String path;
 
     public ActionWrapper(String id, ActionThrottler throttler,
                          @Nullable ExecutableCondition condition,
                          @Nullable ExecutableTransform<Transform, Transform.Result> transform,
-                         ExecutableAction<? extends Action> action) {
+                         ExecutableAction<? extends Action> action,
+                         @Nullable String path) {
         this.id = id;
         this.condition = condition;
         this.throttler = throttler;
         this.transform = transform;
         this.action = action;
+        this.path = path;
     }
 
     public String id() {
@@ -140,13 +153,64 @@ public ActionWrapperResult execute(WatchExecutionContext ctx) {
                 return new ActionWrapperResult(id, conditionResult, null, new Action.Result.FailureWithException(action.type(), e));
             }
         }
-        try {
-            Action.Result actionResult = action.execute(id, ctx, payload);
-            return new ActionWrapperResult(id, conditionResult, transformResult, actionResult);
-        } catch (Exception e) {
-            action.logger().error(
+        if (Strings.isEmpty(path)) {
+            try {
+                Action.Result actionResult = action.execute(id, ctx, payload);
+                return new ActionWrapperResult(id, conditionResult, transformResult, actionResult);
+            } catch (Exception e) {
+                action.logger().error(
+                        (Supplier<?>) () -> new ParameterizedMessage("failed to execute action [{}/{}]", ctx.watch().id(), id), e);
+                return new ActionWrapperResult(id, new Action.Result.FailureWithException(action.type(), e));
+            }
+        } else {
+            try {
+                List<Action.Result> results = new ArrayList<>();
+                Object object = ObjectPath.eval(path, ctx.payload().data());
+                if (object instanceof Collection) {
+                    Collection collection = Collection.class.cast(object);
+                    if (collection.isEmpty()) {
+                        throw new ElasticsearchException("foreach object [{}] was an empty list, could not run any action", path);
+                    } else {
+                        for (Object o : collection) {
+                            if (o instanceof Map) {
+                                results.add(action.execute(id, ctx, new Payload.Simple((Map<String, Object>) o)));
+                            } else {
+                                throw new ElasticsearchException("item in foreach [{}] object was not a map", path);
+                            }
+                        }
+                    }
+                } else {
+                    throw new ElasticsearchException("specified foreach object was not a an array/collection: [{}]", path);
+                }
+
+                // check if we have mixed results, then set to partial failure
+                final Set<Action.Result.Status> statuses = results.stream().map(Action.Result::status).collect(Collectors.toSet());
+                Action.Result.Status status;
+                if (statuses.size() == 1) {
+                    status = statuses.iterator().next();
+                } else {
+                    status = Action.Result.Status.PARTIAL_FAILURE;
+                }
+
+                return new ActionWrapperResult(id, conditionResult, transformResult,
+                    new Action.Result(action.type(), status) {
+                        @Override
+                        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+                            builder.startArray(WatchField.FOREACH.getPreferredName());
+                            for (Action.Result result : results) {
+                                builder.startObject();
+                                result.toXContent(builder, params);
+                                builder.endObject();
+                            }
+                            builder.endArray();
+                            return builder;
+                        }
+                });
+            } catch (Exception e) {
+                action.logger().error(
                     (Supplier<?>) () -> new ParameterizedMessage("failed to execute action [{}/{}]", ctx.watch().id(), id), e);
-            return new ActionWrapperResult(id, new Action.Result.FailureWithException(action.type(), e));
+                return new ActionWrapperResult(id, new Action.Result.FailureWithException(action.type(), e));
+            }
         }
     }
 
@@ -186,6 +250,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                     .field(transform.type(), transform, params)
                     .endObject();
         }
+        if (Strings.isEmpty(path) == false) {
+            builder.field(WatchField.FOREACH.getPreferredName(), path);
+        }
         builder.field(action.type(), action, params);
         return builder.endObject();
     }
@@ -198,6 +265,7 @@ static ActionWrapper parse(String watchId, String actionId, XContentParser parse
         ExecutableCondition condition = null;
         ExecutableTransform<Transform, Transform.Result> transform = null;
         TimeValue throttlePeriod = null;
+        String path = null;
         ExecutableAction<? extends Action> action = null;
 
         String currentFieldName = null;
@@ -208,6 +276,8 @@ static ActionWrapper parse(String watchId, String actionId, XContentParser parse
             } else {
                 if (WatchField.CONDITION.match(currentFieldName, parser.getDeprecationHandler())) {
                     condition = actionRegistry.getConditionRegistry().parseExecutable(watchId, parser);
+                } else if (WatchField.FOREACH.match(currentFieldName, parser.getDeprecationHandler())) {
+                    path = parser.text();
                 } else if (Transform.TRANSFORM.match(currentFieldName, parser.getDeprecationHandler())) {
                     transform = actionRegistry.getTransformRegistry().parse(watchId, parser);
                 } else if (ThrottlerField.THROTTLE_PERIOD.match(currentFieldName, parser.getDeprecationHandler())) {
@@ -235,7 +305,7 @@ static ActionWrapper parse(String watchId, String actionId, XContentParser parse
         }
 
         ActionThrottler throttler = new ActionThrottler(clock, throttlePeriod, licenseState);
-        return new ActionWrapper(actionId, throttler, condition, transform, action);
+        return new ActionWrapper(actionId, throttler, condition, transform, action, path);
     }
 
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/client/WatchSourceBuilder.java

@@ -101,7 +101,7 @@ public WatchSourceBuilder addAction(String id, TimeValue throttlePeriod, Transfo
     }
 
     public WatchSourceBuilder addAction(String id, TimeValue throttlePeriod, Transform transform, Action action) {
-        actions.put(id, new TransformedAction(id, action, throttlePeriod, null, transform));
+        actions.put(id, new TransformedAction(id, action, throttlePeriod, null, transform, null));
         return this;
     }
 
@@ -111,7 +111,13 @@ public WatchSourceBuilder addAction(String id, TimeValue throttlePeriod, Conditi
     }
 
     public WatchSourceBuilder addAction(String id, TimeValue throttlePeriod, Condition condition, Transform transform, Action action) {
-        actions.put(id, new TransformedAction(id, action, throttlePeriod, condition, transform));
+        actions.put(id, new TransformedAction(id, action, throttlePeriod, condition, transform, null));
+        return this;
+    }
+
+    public WatchSourceBuilder addAction(String id, TimeValue throttlePeriod, Condition condition, Transform transform, String path,
+                                        Action action) {
+        actions.put(id, new TransformedAction(id, action, throttlePeriod, condition, transform, path));
         return this;
     }
 
@@ -186,16 +192,18 @@ public final BytesReference buildAsBytes(XContentType contentType) {
     static class TransformedAction implements ToXContentObject {
 
         private final Action action;
+        @Nullable private String path;
         @Nullable private final TimeValue throttlePeriod;
         @Nullable private final Condition condition;
         @Nullable private final Transform transform;
 
         TransformedAction(String id, Action action, @Nullable TimeValue throttlePeriod,
-                          @Nullable Condition condition, @Nullable Transform transform) {
+                          @Nullable Condition condition, @Nullable Transform transform, @Nullable String path) {
             this.throttlePeriod = throttlePeriod;
             this.condition = condition;
             this.transform = transform;
             this.action = action;
+            this.path = path;
         }
 
         @Override
@@ -215,6 +223,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                         .field(transform.type(), transform, params)
                         .endObject();
             }
+            if (path != null) {
+                builder.field("foreach", path);
+            }
             builder.field(action.type(), action, params);
             return builder.endObject();
         }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/support/WatcherIndexTemplateRegistryField.java

@@ -14,8 +14,9 @@ public final class WatcherIndexTemplateRegistryField {
     // version 7: add full exception stack traces for better debugging
     // version 8: fix slack attachment property not to be dynamic, causing field type issues
     // version 9: add a user field defining which user executed the watch
+    // version 10: add support for foreach path in actions
     // Note: if you change this, also inform the kibana team around the watcher-ui
-    public static final String INDEX_TEMPLATE_VERSION = "9";
+    public static final String INDEX_TEMPLATE_VERSION = "10";
     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/java/org/elasticsearch/xpack/core/watcher/watch/WatchField.java

@@ -13,6 +13,7 @@ public final class WatchField {
     public static final ParseField CONDITION = new ParseField("condition");
     public static final ParseField ACTIONS = new ParseField("actions");
     public static final ParseField TRANSFORM = new ParseField("transform");
+    public static final ParseField FOREACH = new ParseField("foreach");
     public static final ParseField THROTTLE_PERIOD = new ParseField("throttle_period_in_millis");
     public static final ParseField THROTTLE_PERIOD_HUMAN = new ParseField("throttle_period");
     public static final ParseField METADATA = new ParseField("metadata");

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

@@ -264,6 +264,10 @@
                 "reason" : {
                   "type" : "keyword"
                 },
+                "foreach" : {
+                  "type": "object",
+                  "enabled" : false
+                },
                 "email": {
                   "type": "object",
                   "dynamic": true,

x-pack/plugin/src/test/resources/rest-api-spec/test/watcher/execute_watch/80_foreach.yml

@@ -0,0 +1,55 @@
+---
+setup:
+  - do:
+      cluster.health:
+          wait_for_status: yellow
+
+---
+teardown:
+  - do:
+      watcher.delete_watch:
+        id: "test_watch"
+        ignore: 404
+
+---
+"Test execute watch api with foreach action":
+  - do:
+      watcher.execute_watch:
+        body:  >
+          {
+            "watch" : {
+              "trigger": {
+                "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+              },
+              "input": {
+                "simple": {
+                  "my_array" : [
+                    { "_doc" : { "_index" : "my_test_index", "_id" : "first", "key" : "first" } },
+                    { "_doc" : { "_index" : "my_test_index", "_id" : "second", "key" : "second" } }
+                  ]
+                }
+              },
+              "actions": {
+                "indexme" : {
+                  "foreach" : "my_array",
+                  "index" : {}
+                }
+              }
+            }
+          }
+
+  - match: { watch_record.trigger_event.type: "manual" }
+  - match: { watch_record.state: "executed" }
+  - match: { watch_record.status.execution_state: "executed" }
+
+  - do:
+      mget:
+        body:
+          docs:
+            - { _index: my_test_index, _id: "first"}
+            - { _index: my_test_index, _id: "second"}
+
+  - is_true:  docs.0.found
+  - match:  { docs.0._source: { "key": "first" }}
+  - is_true:  docs.1.found
+  - match:  { docs.1._source: { "key": "second" }}