Watcher: Allow to execute actions for each element in array

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

@@ -192,6 +192,49 @@ 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.
+
+In order to protect from long running watches, after one hundred runs with an
+foreach loop the execution is gracefully stopped.
+
+[source,js]
+--------------------------------------------------
+PUT _watcher/watch/log_event_watch
+{
+  "trigger" : {
+    "schedule" : { "interval" : "5m" }
+  },
+  "input" : {
+    "search" : {
+      "request" : {
+        "indices" : "log-events",
+        "body" : {
+          "query" : { "match" : { "status" : "error" } }
+        }
+      }
+    }
+  },
+  "condition" : {
+    "compare" : { "ctx.payload.hits.total" : { "gt" : 0 } }
+  },
+  "actions" : {
+    "log_hits" : {
+      "foreach" : "ctx.payload.hits.hits", <1>
+      "logging" : {
+        "text" : "Found id {{ctx.payload._id}} with field {{ctx.payload._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,13 +7,17 @@
 
 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;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.script.JodaCompatibleZonedDateTime;
 import org.elasticsearch.xpack.core.watcher.actions.throttler.ActionThrottler;
 import org.elasticsearch.xpack.core.watcher.actions.throttler.Throttler;
 import org.elasticsearch.xpack.core.watcher.actions.throttler.ThrottlerField;
@@ -30,29 +34,42 @@
 import java.time.Clock;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+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;
 
 public class ActionWrapper implements ToXContentObject {
 
+    private final int MAXIMUM_FOREACH_RUNS = 100;
+
     private String id;
     @Nullable
     private final ExecutableCondition condition;
     @Nullable
     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,16 +157,90 @@ 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, toMap(ctx));
+                int runs = 0;
+                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 (runs >= MAXIMUM_FOREACH_RUNS) {
+                                break;
+                            }
+                            if (o instanceof Map) {
+                                results.add(action.execute(id, ctx, new Payload.Simple((Map<String, Object>) o)));
+                            } else {
+                                results.add(action.execute(id, ctx, new Payload.Simple("_value", o)));
+                            }
+                            runs++;
+                        }
+                    }
+                } else if (object == null) {
+                    throw new ElasticsearchException("specified foreach object was null: [{}]", 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;
+                }
+
+                final int numberOfActionsExecuted = runs;
+                return new ActionWrapperResult(id, conditionResult, transformResult,
+                    new Action.Result(action.type(), status) {
+                        @Override
+                        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+                            builder.field("number_of_actions_executed", numberOfActionsExecuted);
+                            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));
+            }
         }
     }
 
+    private Map<String, Object> toMap(WatchExecutionContext ctx) {
+        Map<String, Object> model = new HashMap<>();
+        model.put("id", ctx.id().value());
+        model.put("watch_id", ctx.id().watchId());
+        model.put("execution_time", new JodaCompatibleZonedDateTime(ctx.executionTime().toInstant(), ZoneOffset.UTC));
+        model.put("trigger", ctx.triggerEvent().data());
+        model.put("metadata", ctx.watch().metadata());
+        model.put("vars", ctx.vars());
+        if (ctx.payload().data() != null) {
+            model.put("payload", ctx.payload().data());
+        }
+        return Map.of("ctx", model);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -186,6 +277,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 +292,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 +303,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 +332,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,13 @@
                 "reason" : {
                   "type" : "keyword"
                 },
+                "number_of_actions_executed": {
+                  "type": "integer"
+                },
+                "foreach" : {
+                  "type": "object",
+                  "enabled" : false
+                },
                 "email": {
                   "type": "object",
                   "dynamic": true,