Add compare condition to handle arrays

This commit adds a new compare condition called “array_compare”. This
condition enables comparing a single resolved value to an array of
resolved values. The value can be compared for equality, non-equality,
and strict and non-strict ordering; the array compare condition will
evaluate to true if the value compares to true with respect to the
specified operator against all (“all”) or at least one (“some”) of the
values in the array specified by “array_path”. Each value in the array
can be resolved to a value using “path” (e.g., “array_path”:
“cx.payload.aggregations.some_field.buckets” and “path”: “doc_count”
would resolve each value in the buckets array to its “doc_count”).

Closes elastic/elasticsearch#345

Original commit: elastic/x-pack-elasticsearch@0d74b4dc11

qa/smoke-test-watcher-with-shield/pom.xml

@@ -37,7 +37,7 @@
     <elasticsearch.integ.antfile>${project.basedir}/integration-tests.xml</elasticsearch.integ.antfile>
     <tests.rest.load_packaged>false</tests.rest.load_packaged>
     <xplugins.list>license,shield,watcher</xplugins.list>
-    <tests.rest.blacklist>hijack/10_basic/*</tests.rest.blacklist>
+    <tests.rest.blacklist>hijack/10_basic/*,array_compare_watch/10_basic/Basic array_compare watch</tests.rest.blacklist>
   </properties>
 
   <dependencies>

watcher/docs/reference/condition.asciidoc

@@ -8,8 +8,8 @@ Watcher supports four condition types: <<condition-always, `always`>>, <<conditi
 NOTE: If you omit the condition definition from a watch, the condition defaults to `always`. 
 
 When a condition is evaluated, it has full access to the watch execution context, including the watch payload (`ctx.payload.*`). 
-The <<condition-script, script>> and <<condition-compare, compare>> conditions can use the data in 
-the payload to determine whether or not  the necessary conditions have been met. 
+The <<condition-script, script>>, <<condition-compare, compare>>  and <<condition-array-compare, array-compare>>
+conditions can use the data in the payload to determine whether or not the necessary conditions have been met.
 
 include::condition/always.asciidoc[]
 
@@ -18,3 +18,5 @@ include::condition/never.asciidoc[]
 include::condition/script.asciidoc[]
 
 include::condition/compare.asciidoc[]
+
+include::condition/array-compare.asciidoc[]

watcher/docs/reference/condition/array-compare.asciidoc

@@ -0,0 +1,68 @@
+[[array-condition-compare]]
+==== Array Compare Condition
+
+A watch <<condition, condition>> that compares an array of values in the <<watch-execution-context, Watch Execution Context Model>>
+to a given value. The values in the model are identified by a path within that model.
+
+==== Using an Array Compare Condition
+
+The following snippet configures an `array_compare` condition that returns `true` if there is at least one bucket in the
+aggregations buckets that has a `doc_count` higher than or equal to 25:
+
+[source,json]
+--------------------------------------------------
+{
+  ...
+
+  "condition": {
+    "array_compare": {
+      "ctx.payload.aggregations.top_tweeters.buckets" : { <1>
+        "path": "doc_count" <2>,
+        "gte": { <3>
+          "value": 25, <4>
+          "quantifier": "some" <5>
+        }
+      }
+    }
+  }
+  ...
+}
+--------------------------------------------------
+<1> The field name is the path to the array (array path) in the execution context model
+<2> The value of the field `path` (here `doc_count`) is the path to the value for each element of the array that the
+comparison operator will be applied to
+<3> The field name (here `gte`) is the name of the comparison operator
+<4> The value of the field `value` in the comparison operator object is the comparison value
+<5> The value of the field `quantifier` (`all` or `some`) specifies whether the comparison must be true for all or for
+at least one of the values to evaluate the comparison to true
+
+NOTE: The `path` element is optional and will default to `""` if not specified.
+
+NOTE: The `quantifier` element is optional and will default to `"some"` if not specified.
+
+The array path is a "dot-notation" expression that can reference the following variables in the watch context:
+
+[options="header"]
+|======
+| Name                                  | Description
+| `ctx.metadata.*`                      | Any metadata associated with the watch.
+| `ctx.payload.*`                       | The payload data loaded by the watch's input.
+|======
+
+This array path must resolve to an array.
+
+The comparison operator can be any one of the operators from [[condition-compare]].
+
+The quantifier operator can be any one of the following:
+
+[options="header"]
+|======
+| Name             | Description
+| `all`             | Returns `true` when the resolved value compares `true` according to the comparison operator for all the elements in the array
+| `some`            | Returns `true` when the resolved value compares `true` according to the comparison operator for at least one element in the array
+|======
+
+NOTE: If the array is empty, `all` causes the comparison operator to evaluate to `true` and `some` causes the comparison
+operator to evaluate to `false`.
+
+NOTE: It is also possible to use date math expressions and values in the context model as in [[condition-compare]].

watcher/rest-api-spec/test/array_compare_watch/10_basic.yaml

@@ -0,0 +1,148 @@
+---
+"Basic array_compare watch":
+  - do:
+      cluster.health:
+          wait_for_status: green
+
+
+  - do: {watcher.stats:{}}
+  - match: { "watcher_state": "started" }
+  - match: { "watch_count": 0 }
+
+  - do:
+      watcher.put_watch:
+        id: "array-compare-watch"
+        body:  >
+          {
+            "trigger": {
+              "schedule": {
+                "interval": "1s"
+              }
+            },
+            "input": {
+              "search": {
+                "request": {
+                  "indices": [ "test_1" ],
+                  "body": {
+                    "query": {
+                      "filtered": {
+                        "query": {
+                          "match_all": {}
+                        }
+                      }
+                    },
+                    "aggs": {
+                      "top_foos": {
+                        "terms": {
+                          "field": "foo",
+                          "size": 1
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            },
+            "condition": {
+              "array_compare": {
+                "ctx.payload.aggregations.top_foos.buckets": {
+                  "path": "doc_count",
+                  "gte": {
+                    "value": 3,
+                    "quantifier": "some"
+                  }
+                }
+              }
+            },
+            "actions": {
+              "log": {
+                "logging": {
+                  "text": "executed at {{ctx.execution_time}}"
+                }
+              }
+            }
+          }
+  - match: { _id: "array-compare-watch" }
+  - match: { created: true }
+
+  - do:
+      index:
+          index:  test_1
+          type:   test
+          id:     1
+          body:   { foo: bar }
+
+  - do:
+      index:
+          index:  test_1
+          type:   test
+          id:     2
+          body:   { foo: bar }
+
+  - do:
+      index:
+          index:  test_1
+          type:   test
+          id:     3
+          body:   { foo: bar }
+
+  - do:
+      index:
+          index:  test_1
+          type:   test
+          id:     4
+          body:   { foo: baz }
+  - do:
+      indices.refresh: {}
+
+  - do: {watcher.stats:{}}
+  - match: { "watch_count": 1 }
+
+  # Simulate a Thread.sleep()
+  - do:
+      catch: request_timeout
+      cluster.health:
+        wait_for_nodes: 99
+        timeout: 10s
+  - match: { "timed_out": true }
+
+  - do:
+      indices.refresh:
+        index: .watch_history-*
+
+  - do:
+      search:
+        index: .watch_history-*
+        body: >
+          {
+            "query": {
+              "bool": {
+                "must" : [
+                  {
+                    "term": {
+                      "watch_id": {
+                        "value": "array-compare-watch"
+                      }
+                    }
+                  },
+                  {
+                    "term": {
+                       "result.condition.met": {
+                         "value": "true"
+                       }
+                    }
+                  }
+                ]
+              }
+            }
+          }
+  - gte:  { hits.total: 1 }
+
+  - do:
+      watcher.delete_watch:
+        id: "array-compare-watch"
+  - match: { found: true }
+
+
+  - do: {watcher.stats:{}}
+  - match: { "watch_count": 0 }

watcher/src/main/java/org/elasticsearch/watcher/condition/ConditionBuilders.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.watcher.condition;
 
 import org.elasticsearch.watcher.condition.always.AlwaysCondition;
+import org.elasticsearch.watcher.condition.compare.array.ArrayCompareCondition;
 import org.elasticsearch.watcher.condition.never.NeverCondition;
 import org.elasticsearch.watcher.condition.script.ScriptCondition;
 import org.elasticsearch.watcher.support.Script;
@@ -37,4 +38,8 @@ public final class ConditionBuilders {
     public static ScriptCondition.Builder scriptCondition(Script script) {
         return ScriptCondition.builder(script);
     }
+
+    public static ArrayCompareCondition.Builder arrayCompareCondition(String arrayPath, String path, ArrayCompareCondition.Op op, Object value, ArrayCompareCondition.Quantifier quantifier) {
+        return ArrayCompareCondition.builder(arrayPath, path, op, value, quantifier);
+    }
 }

watcher/src/main/java/org/elasticsearch/watcher/condition/ConditionModule.java

@@ -11,6 +11,8 @@ import org.elasticsearch.watcher.condition.always.AlwaysCondition;
 import org.elasticsearch.watcher.condition.always.AlwaysConditionFactory;
 import org.elasticsearch.watcher.condition.compare.CompareCondition;
 import org.elasticsearch.watcher.condition.compare.CompareConditionFactory;
+import org.elasticsearch.watcher.condition.compare.array.ArrayCompareCondition;
+import org.elasticsearch.watcher.condition.compare.array.ArrayCompareConditionFactory;
 import org.elasticsearch.watcher.condition.never.NeverCondition;
 import org.elasticsearch.watcher.condition.never.NeverConditionFactory;
 import org.elasticsearch.watcher.condition.script.ScriptCondition;
@@ -47,6 +49,9 @@ public class ConditionModule extends AbstractModule {
         bind(CompareConditionFactory.class).asEagerSingleton();
         factoriesBinder.addBinding(CompareCondition.TYPE).to(CompareConditionFactory.class);
 
+        bind(ArrayCompareConditionFactory.class).asEagerSingleton();
+        factoriesBinder.addBinding(ArrayCompareCondition.TYPE).to(ArrayCompareConditionFactory.class);
+
         for (Map.Entry<String, Class<? extends ConditionFactory>> entry : factories.entrySet()) {
             bind(entry.getValue()).asEagerSingleton();
             factoriesBinder.addBinding(entry.getKey()).to(entry.getValue());

watcher/src/main/java/org/elasticsearch/watcher/condition/compare/AbstractExecutableCompareCondition.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.condition.compare;
+
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.watcher.condition.Condition;
+import org.elasticsearch.watcher.condition.ExecutableCondition;
+import org.elasticsearch.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.watcher.support.Variables;
+import org.elasticsearch.watcher.support.WatcherDateTimeUtils;
+import org.elasticsearch.watcher.support.clock.Clock;
+import org.elasticsearch.watcher.support.xcontent.ObjectPath;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public abstract class AbstractExecutableCompareCondition<C extends Condition, R extends Condition.Result> extends ExecutableCondition<C, R> {
+    static final Pattern DATE_MATH_PATTERN = Pattern.compile("<\\{(.+)\\}>");
+    static final Pattern PATH_PATTERN = Pattern.compile("\\{\\{(.+)\\}\\}");
+
+    private final Clock clock;
+
+    public AbstractExecutableCompareCondition(C condition, ESLogger logger, Clock clock) {
+        super(condition, logger);
+        this.clock = clock;
+    }
+
+    @Override
+    public R execute(WatchExecutionContext ctx) {
+        Map<String, Object> resolvedValues = new HashMap<>();
+        try {
+            Map<String, Object> model = Variables.createCtxModel(ctx, ctx.payload());
+            return doExecute(model, resolvedValues);
+        } catch (Exception e) {
+            logger.error("failed to execute [{}] condition for [{}]", e, type(), ctx.id());
+            if (resolvedValues.isEmpty()) {
+                resolvedValues = null;
+            }
+            return doFailure(resolvedValues, e);
+        }
+    }
+
+    protected Object resolveConfiguredValue(Map<String, Object> resolvedValues, Map<String, Object> model, Object configuredValue) {
+        if (configuredValue instanceof String) {
+
+            // checking if the given value is a date math expression
+            Matcher matcher = DATE_MATH_PATTERN.matcher((String) configuredValue);
+            if (matcher.matches()) {
+                String dateMath = matcher.group(1);
+                configuredValue = WatcherDateTimeUtils.parseDateMath(dateMath, DateTimeZone.UTC, clock);
+                resolvedValues.put(dateMath, WatcherDateTimeUtils.formatDate((DateTime) configuredValue));
+            } else {
+                // checking if the given value is a path expression
+                matcher = PATH_PATTERN.matcher((String) configuredValue);
+                if (matcher.matches()) {
+                    String configuredPath = matcher.group(1);
+                    configuredValue = ObjectPath.eval(configuredPath, model);
+                    resolvedValues.put(configuredPath, configuredValue);
+                }
+            }
+        }
+        return configuredValue;
+    }
+
+    protected abstract R doExecute(Map<String, Object> model, Map<String, Object> resolvedValues) throws Exception;
+
+    protected abstract R doFailure(Map<String, Object> resolvedValues, Exception e);
+}

watcher/src/main/java/org/elasticsearch/watcher/condition/compare/CompareCondition.java

@@ -11,15 +11,11 @@ import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.watcher.condition.Condition;
-import org.elasticsearch.watcher.support.WatcherDateTimeUtils;
 import org.elasticsearch.watcher.support.xcontent.WatcherXContentUtils;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 
 import java.io.IOException;
 import java.util.Locale;
 import java.util.Map;
-import java.util.Objects;
 
 /**
  *
@@ -134,9 +130,9 @@ public class CompareCondition implements Condition {
             this.resolveValues = resolveValues;
         }
 
-        Result(@Nullable Map<String, Object> resolveValues, Exception e) {
+        Result(@Nullable Map<String, Object> resolvedValues, Exception e) {
             super(TYPE, e);
-            this.resolveValues = resolveValues;
+            this.resolveValues = resolvedValues;
         }
 
         public Map<String, Object> getResolveValues() {
@@ -159,7 +155,7 @@ public class CompareCondition implements Condition {
         EQ() {
             @Override
             public boolean eval(Object v1, Object v2) {
-                Integer compVal = compare(v1, v2);
+                Integer compVal = LenientCompare.compare(v1, v2);
                 return compVal != null && compVal == 0;
             }
 
@@ -171,7 +167,7 @@ public class CompareCondition implements Condition {
         NOT_EQ() {
             @Override
             public boolean eval(Object v1, Object v2) {
-                Integer compVal = compare(v1, v2);
+                Integer compVal = LenientCompare.compare(v1, v2);
                 return compVal == null || compVal != 0;
             }
 
@@ -183,28 +179,28 @@ public class CompareCondition implements Condition {
         LT() {
             @Override
             public boolean eval(Object v1, Object v2) {
-                Integer compVal = compare(v1, v2);
+                Integer compVal = LenientCompare.compare(v1, v2);
                 return compVal != null && compVal < 0;
             }
         },
         LTE() {
             @Override
             public boolean eval(Object v1, Object v2) {
-                Integer compVal = compare(v1, v2);
+                Integer compVal = LenientCompare.compare(v1, v2);
                 return compVal != null && compVal <= 0;
             }
         },
         GT() {
             @Override
             public boolean eval(Object v1, Object v2) {
-                Integer compVal = compare(v1, v2);
+                Integer compVal = LenientCompare.compare(v1, v2);
                 return compVal != null && compVal > 0;
             }
         },
         GTE() {
             @Override
             public boolean eval(Object v1, Object v2) {
-                Integer compVal = compare(v1, v2);
+                Integer compVal = LenientCompare.compare(v1, v2);
                 return compVal != null && compVal >= 0;
             }
         };
@@ -215,73 +211,6 @@ public class CompareCondition implements Condition {
             return false;
         }
 
-        // this method performs lenient comparison, potentially between different types. The second argument
-        // type (v2) determines the type of comparison (this is because the second argument is configured by the
-        // user while the first argument is the dynamic path that is evaluated at runtime. That is, if the user configures
-        // a number, it expects a number, therefore the comparison will be based on numeric comparison). If the
-        // comparison is numeric, other types (e.g. strings) will converted to numbers if possible, if not, the comparison
-        // will fail and `false` will be returned.
-        //
-        // may return `null` indicating v1 simply doesn't equal v2 (without any order association)
-        static Integer compare(Object v1, Object v2) {
-            if (Objects.equals(v1, v2)) {
-                return 0;
-            }
-            if (v1 == null || v2 == null) {
-                return null;
-            }
-
-            // special case for numbers. If v1 is not a number, we'll try to convert it to a number
-            if (v2 instanceof Number) {
-                if (!(v1 instanceof Number)) {
-                    try {
-                        v1 = Double.valueOf(String.valueOf(v1));
-                    } catch (NumberFormatException nfe) {
-                        // could not convert to number
-                        return null;
-                    }
-                }
-                return ((Number) v1).doubleValue() > ((Number) v2).doubleValue() ? 1 :
-                        ((Number) v1).doubleValue() < ((Number) v2).doubleValue() ? -1 : 0;
-            }
-
-            // special case for strings. If v1 is not a string, we'll convert it to a string
-            if (v2 instanceof String) {
-                v1 = String.valueOf(v1);
-                return ((String) v1).compareTo((String) v2);
-            }
-
-            // special case for date/times. If v1 is not a dateTime, we'll try to convert it to a datetime
-            if (v2 instanceof DateTime) {
-                if (v1 instanceof DateTime) {
-                    return ((DateTime) v1).compareTo((DateTime) v2);
-                }
-                if (v1 instanceof String) {
-                    try {
-                        v1 = WatcherDateTimeUtils.parseDate((String) v1);
-                    } catch (Exception e) {
-                        return null;
-                    }
-                } else if (v1 instanceof Number){
-                    v1 = new DateTime(((Number) v1).longValue(), DateTimeZone.UTC);
-                } else {
-                    // cannot convert to date...
-                    return null;
-                }
-                return ((DateTime) v1).compareTo((DateTime) v2);
-            }
-
-            if (v1.getClass() != v2.getClass() || Comparable.class.isAssignableFrom(v1.getClass())) {
-                return null;
-            }
-
-            try {
-                return ((Comparable) v1).compareTo(v2);
-            } catch (Exception e) {
-                return null;
-            }
-        }
-
         public String id() {
             return name().toLowerCase(Locale.ROOT);
         }

watcher/src/main/java/org/elasticsearch/watcher/condition/compare/ExecutableCompareCondition.java

@@ -5,81 +5,33 @@
  */
 package org.elasticsearch.watcher.condition.compare;
 
-import org.joda.time.DateTime;
 import org.elasticsearch.common.logging.ESLogger;
-import org.elasticsearch.watcher.actions.email.DataAttachment;
-import org.elasticsearch.watcher.condition.ExecutableCondition;
-import org.elasticsearch.watcher.execution.WatchExecutionContext;
-import org.elasticsearch.watcher.support.Variables;
-import org.elasticsearch.watcher.support.WatcherDateTimeUtils;
 import org.elasticsearch.watcher.support.clock.Clock;
 import org.elasticsearch.watcher.support.xcontent.ObjectPath;
-import org.joda.time.DateTimeZone;
 
-import java.io.IOException;
-import java.util.HashMap;
 import java.util.Map;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 
 /**
  *
  */
-public class ExecutableCompareCondition extends ExecutableCondition<CompareCondition, CompareCondition.Result> {
-
-    static final Pattern DATE_MATH_PATTERN = Pattern.compile("<\\{(.+)\\}>");
-    static final Pattern PATH_PATTERN = Pattern.compile("\\{\\{(.+)\\}\\}");
-
-
-    private final Clock clock;
-
+public class ExecutableCompareCondition extends AbstractExecutableCompareCondition<CompareCondition, CompareCondition.Result> {
     public ExecutableCompareCondition(CompareCondition condition, ESLogger logger, Clock clock) {
-        super(condition, logger);
-        this.clock = clock;
+        super(condition, logger, clock);
     }
 
     @Override
-    public CompareCondition.Result execute(WatchExecutionContext ctx) {
-        Map<String, Object> resolvedValues = new HashMap<>();
-        try {
-            return doExecute(ctx, resolvedValues);
-        } catch (Exception e) {
-            logger.error("failed to execute [{}] condition for [{}]", e, CompareCondition.TYPE, ctx.id());
-            if (resolvedValues.isEmpty()) {
-                resolvedValues = null;
-            }
-            return new CompareCondition.Result(resolvedValues, e);
-        }
-    }
-
-    public CompareCondition.Result doExecute(WatchExecutionContext ctx, Map<String, Object> resolvedValues) throws Exception {
-        Map<String, Object> model = Variables.createCtxModel(ctx, ctx.payload());
-
-        Object configuredValue = condition.getValue();
-
-        if (configuredValue instanceof String) {
-
-            // checking if the given value is a date math expression
-            Matcher matcher = DATE_MATH_PATTERN.matcher((String) configuredValue);
-            if (matcher.matches()) {
-                String dateMath = matcher.group(1);
-                configuredValue = WatcherDateTimeUtils.parseDateMath(dateMath, DateTimeZone.UTC, clock);
-                resolvedValues.put(dateMath, WatcherDateTimeUtils.formatDate((DateTime) configuredValue));
-            } else {
-                // checking if the given value is a path expression
-                matcher = PATH_PATTERN.matcher((String) configuredValue);
-                if (matcher.matches()) {
-                    String configuredPath = matcher.group(1);
-                    configuredValue = ObjectPath.eval(configuredPath, model);
-                    resolvedValues.put(configuredPath, configuredValue);
-                }
-            }
-        }
+    protected CompareCondition.Result doExecute(Map<String, Object> model, Map<String, Object> resolvedValues) throws Exception {
+        Object configuredValue = resolveConfiguredValue(resolvedValues, model, condition.getValue());
 
         Object resolvedValue = ObjectPath.eval(condition.getPath(), model);
         resolvedValues.put(condition.getPath(), resolvedValue);
 
         return new CompareCondition.Result(resolvedValues, condition.getOp().eval(resolvedValue, configuredValue));
     }
+
+    @Override
+    protected CompareCondition.Result doFailure(Map<String, Object> resolvedValues, Exception e) {
+        return new CompareCondition.Result(resolvedValues, e);
+    }
 }

watcher/src/main/java/org/elasticsearch/watcher/condition/compare/LenientCompare.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.condition.compare;
+
+import org.elasticsearch.watcher.support.WatcherDateTimeUtils;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+
+import java.util.Objects;
+
+public class LenientCompare {
+    // this method performs lenient comparison, potentially between different types. The second argument
+    // type (v2) determines the type of comparison (this is because the second argument is configured by the
+    // user while the first argument is the dynamic path that is evaluated at runtime. That is, if the user configures
+    // a number, it expects a number, therefore the comparison will be based on numeric comparison). If the
+    // comparison is numeric, other types (e.g. strings) will converted to numbers if possible, if not, the comparison
+    // will fail and `false` will be returned.
+    //
+    // may return `null` indicating v1 simply doesn't equal v2 (without any order association)
+    public static Integer compare(Object v1, Object v2) {
+        if (Objects.equals(v1, v2)) {
+            return 0;
+        }
+        if (v1 == null || v2 == null) {
+            return null;
+        }
+
+        // special case for numbers. If v1 is not a number, we'll try to convert it to a number
+        if (v2 instanceof Number) {
+            if (!(v1 instanceof Number)) {
+                try {
+                    v1 = Double.valueOf(String.valueOf(v1));
+                } catch (NumberFormatException nfe) {
+                    // could not convert to number
+                    return null;
+                }
+            }
+            return ((Number) v1).doubleValue() > ((Number) v2).doubleValue() ? 1 :
+                    ((Number) v1).doubleValue() < ((Number) v2).doubleValue() ? -1 : 0;
+        }
+
+        // special case for strings. If v1 is not a string, we'll convert it to a string
+        if (v2 instanceof String) {
+            v1 = String.valueOf(v1);
+            return ((String) v1).compareTo((String) v2);
+        }
+
+        // special case for date/times. If v1 is not a dateTime, we'll try to convert it to a datetime
+        if (v2 instanceof DateTime) {
+            if (v1 instanceof DateTime) {
+                return ((DateTime) v1).compareTo((DateTime) v2);
+            }
+            if (v1 instanceof String) {
+                try {
+                    v1 = WatcherDateTimeUtils.parseDate((String) v1);
+                } catch (Exception e) {
+                    return null;
+                }
+            } else if (v1 instanceof Number) {
+                v1 = new DateTime(((Number) v1).longValue(), DateTimeZone.UTC);
+            } else {
+                // cannot convert to date...
+                return null;
+            }
+            return ((DateTime) v1).compareTo((DateTime) v2);
+        }
+
+        if (v1.getClass() != v2.getClass() || Comparable.class.isAssignableFrom(v1.getClass())) {
+            return null;
+        }
+
+        try {
+            return ((Comparable) v1).compareTo(v2);
+        } catch (Exception e) {
+            return null;
+        }
+    }
+
+}

watcher/src/main/java/org/elasticsearch/watcher/condition/compare/array/ArrayCompareCondition.java

@@ -0,0 +1,347 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.condition.compare.array;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.ParseFieldMatcher;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.watcher.condition.Condition;
+import org.elasticsearch.watcher.condition.compare.LenientCompare;
+import org.elasticsearch.watcher.support.xcontent.WatcherXContentUtils;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+public class ArrayCompareCondition implements Condition {
+    public static final String TYPE = "array_compare";
+
+    private String arrayPath;
+    private String path;
+    private Op op;
+    private Object value;
+    private Quantifier quantifier;
+
+    public ArrayCompareCondition(String arrayPath, String path, Op op, Object value, Quantifier quantifier) {
+        this.arrayPath = arrayPath;
+        this.path = path;
+        this.op = op;
+        this.value = value;
+        this.quantifier = quantifier;
+    }
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    public String getArrayPath() {
+        return arrayPath;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public Op getOp() {
+        return op;
+    }
+
+    public Object getValue() {
+        return value;
+    }
+
+    public Quantifier getQuantifier() {
+        return quantifier;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ArrayCompareCondition that = (ArrayCompareCondition) o;
+        return Objects.equals(getArrayPath(), that.getArrayPath()) &&
+                Objects.equals(getPath(), that.getPath()) &&
+                Objects.equals(getOp(), that.getOp()) &&
+                Objects.equals(getValue(), that.getValue()) &&
+                Objects.equals(getQuantifier(), that.getQuantifier());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getArrayPath(), getPath(), getOp(), getValue(), getQuantifier());
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return
+                builder
+                        .startObject()
+                            .startObject(arrayPath)
+                                .field(Field.PATH.getPreferredName(), path)
+                                .startObject(op.id())
+                                    .field(Field.VALUE.getPreferredName(), value)
+                                    .field(Field.QUANTIFIER.getPreferredName(), quantifier.id())
+                                .endObject()
+                            .endObject()
+                        .endObject();
+    }
+
+    public static ArrayCompareCondition parse(String watchId, XContentParser parser) throws IOException {
+        if (parser.currentToken() != XContentParser.Token.START_OBJECT) {
+            throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. expected an object but found [{}] instead", TYPE, watchId, parser.currentToken());
+        }
+        String arrayPath = null;
+        String path = null;
+        Op op = null;
+        Object value = null;
+        boolean haveValue = false;
+        Quantifier quantifier = null;
+
+        XContentParser.Token token;
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                arrayPath = parser.currentName();
+            } else if (arrayPath == null) {
+                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. expected a field indicating the compared path, but found [{}] instead", TYPE, watchId, token);
+            } else if (token == XContentParser.Token.START_OBJECT) {
+                while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+                    if (token == XContentParser.Token.FIELD_NAME) {
+                        if (ParseFieldMatcher.STRICT.match(parser.currentName(), Field.PATH)) {
+                            parser.nextToken();
+                            path = parser.text();
+                        } else {
+                            if (op != null) {
+                                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. encountered duplicate comparison operator, but already saw [{}].", TYPE, watchId, parser.currentName(), op.id());
+                            }
+                            try {
+                                op = Op.resolve(parser.currentName());
+                            } catch (IllegalArgumentException iae) {
+                                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. unknown comparison operator [{}]", TYPE, watchId, parser.currentName(), iae);
+                            }
+                            token = parser.nextToken();
+                            if (token == XContentParser.Token.START_OBJECT) {
+                                while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+                                    if (token == XContentParser.Token.FIELD_NAME) {
+                                        if (ParseFieldMatcher.STRICT.match(parser.currentName(), Field.VALUE)) {
+                                            if (haveValue) {
+                                                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. encountered duplicate field \"value\", but already saw value [{}].", TYPE, watchId, value);
+                                            }
+                                            token = parser.nextToken();
+                                            if (!op.supportsStructures() && !token.isValue() && token != XContentParser.Token.VALUE_NULL) {
+                                                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. compared value for [{}] with operation [{}] must either be a numeric, string, boolean or null value, but found [{}] instead", TYPE, watchId, path, op.name().toLowerCase(Locale.ROOT), token);
+                                            }
+                                            value = WatcherXContentUtils.readValue(parser, token);
+                                            haveValue = true;
+                                        } else if (ParseFieldMatcher.STRICT.match(parser.currentName(), Field.QUANTIFIER)) {
+                                            if (quantifier != null) {
+                                                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. encountered duplicate field \"quantifier\", but already saw quantifier [{}].", TYPE, watchId, quantifier.id());
+                                            }
+                                            parser.nextToken();
+                                            try {
+                                                quantifier = Quantifier.resolve(parser.text());
+                                            } catch (IllegalArgumentException iae) {
+                                                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. unknown comparison quantifier [{}]", TYPE, watchId, parser.text(), iae);
+                                            }
+                                        } else {
+                                            throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. expected a field indicating the comparison value or comparison quantifier, but found [{}] instead", TYPE, watchId, parser.currentName());
+                                        }
+                                    } else {
+                                        throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. expected a field indicating the comparison value or comparison quantifier, but found [{}] instead", TYPE, watchId, token);
+                                    }
+                                }
+                            } else {
+                                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. expected an object for field [{}] but found [{}] instead", TYPE, watchId, op.id(), token);
+                            }
+                        }
+                    } else {
+                        throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. expected a field indicating the compared path or a comparison operator, but found [{}] instead", TYPE, watchId, token);
+                    }
+                }
+            } else {
+                throw new ElasticsearchParseException("could not parse [{}] condition for watch [{}]. expected an object for field [{}] but found [{}] instead", TYPE, watchId, path, token);
+            }
+        }
+
+        if (path == null) {
+            path = "";
+        }
+        if (quantifier == null) {
+            quantifier = Quantifier.SOME;
+        }
+
+        return new ArrayCompareCondition(arrayPath, path, op, value, quantifier);
+    }
+
+    public static class Result extends Condition.Result {
+        private final @Nullable Map<String, Object> resolvedValues;
+
+        Result(Map<String, Object> resolvedValues, boolean met) {
+            super(TYPE, met);
+            this.resolvedValues = resolvedValues;
+        }
+
+        Result(@Nullable Map<String, Object> resolvedValues, Exception e) {
+            super(TYPE, e);
+            this.resolvedValues = resolvedValues;
+        }
+
+        public Map<String, Object> getResolvedValues() {
+            return resolvedValues;
+        }
+
+        @Override
+        protected XContentBuilder typeXContent(XContentBuilder builder, Params params) throws IOException {
+            if (resolvedValues == null) {
+                return builder;
+            }
+            return builder.startObject(type)
+                        .field(Field.RESOLVED_VALUES.getPreferredName(), resolvedValues)
+                    .endObject();
+        }
+
+        public interface Field extends Condition.Field {
+            ParseField RESOLVED_VALUES = new ParseField("resolved_values");
+        }
+    }
+
+    public enum Op {
+        EQ() {
+            @Override
+            public boolean comparison(int x) {
+                return x == 0;
+            }
+
+            @Override
+            public boolean supportsStructures() {
+                return true;
+            }
+        },
+        NOT_EQ() {
+            @Override
+            public boolean comparison(int x) {
+                return x != 0;
+            }
+
+            @Override
+            public boolean supportsStructures() {
+                return true;
+            }
+        },
+        GTE() {
+            @Override
+            public boolean comparison(int x) {
+                return x >= 0;
+            }
+        },
+        GT() {
+            @Override
+            public boolean comparison(int x) {
+                return x > 0;
+            }
+        },
+        LTE() {
+            @Override
+            public boolean comparison(int x) {
+                return x <= 0;
+            }
+        },
+        LT() {
+            @Override
+            public boolean comparison(int x) {
+                return x < 0;
+            }
+        };
+
+        public abstract boolean comparison(int x);
+
+        public boolean supportsStructures() {
+            return false;
+        }
+
+        public String id() {
+            return name().toLowerCase(Locale.ROOT);
+        }
+
+        public static Op resolve(String id) {
+            return Op.valueOf(id.toUpperCase(Locale.ROOT));
+        }
+    }
+
+    public enum Quantifier {
+        ALL() {
+            @Override
+            public boolean eval(List<Object> values, Object configuredValue, Op op) {
+                for (Object value : values) {
+                    Integer compare = LenientCompare.compare(value, configuredValue);
+                    boolean comparison = compare != null && op.comparison(compare);
+                    if (!comparison) {
+                        return false;
+                    }
+                }
+                return true;
+            }
+        },
+        SOME() {
+            @Override
+            public boolean eval(List<Object> values, Object configuredValue, Op op) {
+                for (Object value : values) {
+                    Integer compare = LenientCompare.compare(value, configuredValue);
+                    boolean comparison = compare != null && op.comparison(compare);
+                    if (comparison) {
+                        return true;
+                    }
+                }
+                return false;
+            }
+        };
+
+        public abstract boolean eval(List<Object> values, Object configuredValue, Op op);
+
+        public static Quantifier resolve(String id) {
+            return Quantifier.valueOf(id.toUpperCase(Locale.ROOT));
+        }
+
+        public String id() {
+            return name().toLowerCase(Locale.ROOT);
+        }
+    }
+
+    public static Builder builder(String arrayPath, String path, Op op, Object value, Quantifier quantifier) {
+        return new Builder(arrayPath, path, op, value, quantifier);
+    }
+
+    public static class Builder implements Condition.Builder<ArrayCompareCondition> {
+        private String arrayPath;
+        private String path;
+        private Op op;
+        private Object value;
+        private Quantifier quantifier;
+
+        private Builder(String arrayPath, String path, Op op, Object value, Quantifier quantifier) {
+            this.arrayPath = arrayPath;
+            this.path = path;
+            this.op = op;
+            this.value = value;
+            this.quantifier = quantifier;
+        }
+
+        public ArrayCompareCondition build() {
+            return new ArrayCompareCondition(arrayPath, path, op, value, quantifier);
+        }
+    }
+
+    interface Field {
+        ParseField PATH = new ParseField("path");
+        ParseField VALUE = new ParseField("value");
+        ParseField QUANTIFIER = new ParseField("quantifier");
+    }
+}

watcher/src/main/java/org/elasticsearch/watcher/condition/compare/array/ArrayCompareConditionFactory.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.condition.compare.array;
+
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.watcher.condition.ConditionFactory;
+import org.elasticsearch.watcher.support.clock.Clock;
+
+import java.io.IOException;
+
+public class ArrayCompareConditionFactory extends ConditionFactory<ArrayCompareCondition, ArrayCompareCondition.Result, ExecutableArrayCompareCondition> {
+
+    private final Clock clock;
+
+    @Inject
+    public ArrayCompareConditionFactory(Settings settings, Clock clock) {
+        super(Loggers.getLogger(ExecutableArrayCompareCondition.class, settings));
+        this.clock = clock;
+    }
+
+    @Override
+    public String type() {
+        return ArrayCompareCondition.TYPE;
+    }
+
+    @Override
+    public ArrayCompareCondition parseCondition(String watchId, XContentParser parser) throws IOException {
+        return ArrayCompareCondition.parse(watchId, parser);
+    }
+
+    @Override
+    public ExecutableArrayCompareCondition createExecutable(ArrayCompareCondition condition) {
+        return new ExecutableArrayCompareCondition(condition, conditionLogger, clock);
+    }
+}

watcher/src/main/java/org/elasticsearch/watcher/condition/compare/array/ExecutableArrayCompareCondition.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.condition.compare.array;
+
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.watcher.condition.compare.AbstractExecutableCompareCondition;
+import org.elasticsearch.watcher.support.clock.Clock;
+import org.elasticsearch.watcher.support.xcontent.ObjectPath;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class ExecutableArrayCompareCondition extends AbstractExecutableCompareCondition<ArrayCompareCondition, ArrayCompareCondition.Result> {
+
+    public ExecutableArrayCompareCondition(ArrayCompareCondition condition, ESLogger logger, Clock clock) {
+        super(condition, logger, clock);
+    }
+
+    @SuppressWarnings("unchecked")
+    public ArrayCompareCondition.Result doExecute(Map<String, Object> model, Map<String, Object> resolvedValues) throws Exception {
+        Object configuredValue = resolveConfiguredValue(resolvedValues, model, condition.getValue());
+
+        Object object = ObjectPath.eval(condition.getArrayPath(), model);
+        if (object != null && !(object instanceof List)) {
+            throw new IllegalStateException("array path " + condition.getArrayPath() + " did not evaluate to array, was " + object);
+        }
+
+        List<Object> resolvedArray = object != null ? (List<Object>)object : Collections.emptyList();
+
+        List<Object> resolvedValue = new ArrayList<>(resolvedArray.size());
+        for (int i = 0; i < resolvedArray.size(); i++) {
+            resolvedValue.add(ObjectPath.eval(condition.getPath(), resolvedArray.get(i)));
+        }
+        resolvedValues.put(condition.getArrayPath(), resolvedArray);
+
+        return new ArrayCompareCondition.Result(resolvedValues, condition.getQuantifier().eval(resolvedValue, configuredValue, condition.getOp()));
+    }
+
+    @Override
+    protected ArrayCompareCondition.Result doFailure(Map<String, Object> resolvedValues, Exception e) {
+        return new ArrayCompareCondition.Result(resolvedValues, e);
+    }
+}

watcher/src/main/resources/watch_history.json

@@ -185,6 +185,10 @@
                   "type" : "object",
                   "enabled" : false
                 },
+                "array_compare" : {
+                  "type" : "object",
+                  "enabled" : false
+                },
                 "script" : {
                   "type" : "object",
                   "enabled" : false

watcher/src/test/java/org/elasticsearch/watcher/condition/compare/array/ArrayCompareConditionSearchTests.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.condition.compare.array;
+
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.watcher.support.clock.SystemClock;
+import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTestCase;
+import org.elasticsearch.watcher.watch.Payload;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.watcher.test.WatcherTestUtils.mockExecutionContext;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.collection.IsMapContaining.hasEntry;
+
+public class ArrayCompareConditionSearchTests extends AbstractWatcherIntegrationTestCase {
+
+    @Test
+    public void testExecuteWithAggs() throws Exception {
+        String index = "test-index";
+        String type = "test-type";
+        client().admin().indices().prepareCreate(index)
+                .addMapping(type)
+                .get();
+
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        int numberOfDocuments = randomIntBetween(1, 100);
+        int numberOfDocumentsWatchingFor = 1 + numberOfDocuments;
+        for (int i = 0; i < numberOfDocuments; i++) {
+            client().prepareIndex(index, type).setSource(source("elastic", "you know, for search", i)).get();
+            client().prepareIndex(index, type).setSource(source("fights_for_the_users", "you know, for the users", i)).get();
+        }
+
+        refresh();
+
+        SearchResponse response = client().prepareSearch(index)
+                .addAggregation(AggregationBuilders.terms("top_tweeters").field("user.screen_name").size(3)).get();
+
+
+        ExecutableArrayCompareCondition condition = new ExecutableArrayCompareCondition(
+                new ArrayCompareCondition("ctx.payload.aggregations.top_tweeters.buckets" , "doc_count", op, numberOfDocumentsWatchingFor, quantifier),
+                logger,
+                SystemClock.INSTANCE
+        );
+
+        WatchExecutionContext ctx = mockExecutionContext("_name", new Payload.XContent(response));
+        ArrayCompareCondition.Result result = condition.execute(ctx);
+
+        boolean met = quantifier.eval(Arrays.<Object>asList(numberOfDocuments, numberOfDocuments), numberOfDocumentsWatchingFor, op);
+        assertEquals(met, result.met());
+
+        Map<String, Object> resolvedValues = result.getResolvedValues();
+        assertThat(resolvedValues, notNullValue());
+        assertThat(resolvedValues.size(), is(1));
+        Map<String, Object> elastic = new HashMap<>();
+        elastic.put("doc_count", numberOfDocuments);
+        elastic.put("key", "elastic");
+        Map<String, Object> fightsForTheUsers = new HashMap<>();
+        fightsForTheUsers.put("doc_count", numberOfDocuments);
+        fightsForTheUsers.put("key", "fights_for_the_users");
+        assertThat(resolvedValues, hasEntry("ctx.payload.aggregations.top_tweeters.buckets", (Object) Arrays.asList(elastic, fightsForTheUsers)));
+
+        client().prepareIndex(index, type).setSource(source("fights_for_the_users", "you know, for the users", numberOfDocuments)).get();
+        refresh();
+
+        response = client().prepareSearch(index)
+                .addAggregation(AggregationBuilders.terms("top_tweeters").field("user.screen_name").size(3)).get();
+
+        ctx = mockExecutionContext("_name", new Payload.XContent(response));
+        result = condition.execute(ctx);
+
+        met = quantifier.eval(Arrays.<Object>asList(numberOfDocumentsWatchingFor, numberOfDocuments), numberOfDocumentsWatchingFor, op);
+        assertEquals(met, result.met());
+
+        resolvedValues = result.getResolvedValues();
+        assertThat(resolvedValues, notNullValue());
+        assertThat(resolvedValues.size(), is(1));
+        fightsForTheUsers.put("doc_count", numberOfDocumentsWatchingFor);
+        assertThat(resolvedValues, hasEntry("ctx.payload.aggregations.top_tweeters.buckets", (Object) Arrays.asList(fightsForTheUsers, elastic)));
+    }
+
+    private XContentBuilder source(String screenName, String tweet, int i) throws IOException {
+        return jsonBuilder().startObject()
+                    .startObject("user")
+                        .field("screen_name", screenName)
+                    .endObject()
+                    .field("tweet", tweet + " " + i)
+                .endObject();
+    }
+}

watcher/src/test/java/org/elasticsearch/watcher/condition/compare/array/ArrayCompareConditionTests.java

@@ -0,0 +1,363 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.condition.compare.array;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.watcher.support.clock.ClockMock;
+import org.elasticsearch.watcher.support.clock.SystemClock;
+import org.elasticsearch.watcher.watch.Payload;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.watcher.test.WatcherTestUtils.mockExecutionContext;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.Matchers.is;
+
+public class ArrayCompareConditionTests extends ESTestCase {
+
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
+    @Test
+    public void testOpEvalEQ() throws Exception {
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 1), 1, ArrayCompareCondition.Op.EQ), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 2, ArrayCompareCondition.Op.EQ), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 1, ArrayCompareCondition.Op.EQ), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 2, ArrayCompareCondition.Op.EQ), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.EQ), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.EQ), is(false));
+    }
+
+    @Test
+    public void testOpEvalNOT_EQ() throws Exception {
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 1), 3, ArrayCompareCondition.Op.NOT_EQ), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 1, ArrayCompareCondition.Op.NOT_EQ), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 1, ArrayCompareCondition.Op.NOT_EQ), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 1), 1, ArrayCompareCondition.Op.NOT_EQ), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.NOT_EQ), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.NOT_EQ), is(false));
+    }
+
+    @Test
+    public void testOpEvalGTE() throws Exception {
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 1, ArrayCompareCondition.Op.GTE), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 2, ArrayCompareCondition.Op.GTE), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 2, ArrayCompareCondition.Op.GTE), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 4, ArrayCompareCondition.Op.GTE), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.GTE), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.GTE), is(false));
+    }
+
+    @Test
+    public void testOpEvalGT() throws Exception {
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 0, ArrayCompareCondition.Op.GT), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 1, ArrayCompareCondition.Op.GT), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 2, ArrayCompareCondition.Op.GT), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 4, ArrayCompareCondition.Op.GT), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.GT), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.GT), is(false));
+    }
+
+    @Test
+    public void testOpEvalLTE() throws Exception {
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 3, ArrayCompareCondition.Op.LTE), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 0, ArrayCompareCondition.Op.LTE), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 3, ArrayCompareCondition.Op.LTE), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 0, ArrayCompareCondition.Op.LTE), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.LTE), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.LTE), is(false));
+    }
+
+    @Test
+    public void testOpEvalLT() throws Exception {
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 4, ArrayCompareCondition.Op.LT), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Arrays.<Object>asList(1, 3), 3, ArrayCompareCondition.Op.LT), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 2, ArrayCompareCondition.Op.LT), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Arrays.<Object>asList(1, 3), 0, ArrayCompareCondition.Op.LT), is(false));
+        assertThat(ArrayCompareCondition.Quantifier.ALL.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.LT), is(true));
+        assertThat(ArrayCompareCondition.Quantifier.SOME.eval(Collections.emptyList(), 1, ArrayCompareCondition.Op.LT), is(false));
+    }
+
+    @Test
+    public void testExecute() {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        int value = randomInt(10);
+        int numberOfValues = randomIntBetween(0, 3);
+        List<Object> values = new ArrayList<>(numberOfValues);
+        for (int i = 0; i < numberOfValues; i++) {
+            values.add(randomInt(10));
+        }
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        boolean met = quantifier.eval(values, value, op);
+
+        logger.debug("op [{}]", op);
+        logger.debug("value [{}]", value);
+        logger.debug("numberOfValues [{}]", numberOfValues);
+        logger.debug("values [{}]", values);
+        logger.debug("quantifier [{}]", quantifier);
+        logger.debug("met [{}]", met);
+
+        ExecutableArrayCompareCondition condition = new ExecutableArrayCompareCondition(new ArrayCompareCondition("ctx.payload.value", "", op, value, quantifier), logger, SystemClock.INSTANCE);
+        WatchExecutionContext ctx = mockExecutionContext("_name", new Payload.Simple("value", values));
+        assertThat(condition.execute(ctx).met(), is(met));
+    }
+
+    @Test
+    public void testExecutePath() {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        int value = randomInt(10);
+        int numberOfValues = randomIntBetween(0, 3);
+        List<Object> docCounts = new ArrayList<>(numberOfValues);
+        for (int i = 0; i < numberOfValues; i++) {
+            docCounts.add(randomInt(10));
+        }
+        List<Object> values = new ArrayList<>(numberOfValues);
+        for (int i = 0; i < numberOfValues; i++) {
+            Map<String, Object> map = new HashMap<>(1);
+            map.put("doc_count", docCounts.get(i));
+            values.add(map);
+        }
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        boolean met = quantifier.eval(docCounts, value, op);
+
+        logger.debug("op [{}]", op);
+        logger.debug("value [{}]", value);
+        logger.debug("numberOfValues [{}]", numberOfValues);
+        logger.debug("values [{}]", values);
+        logger.debug("quantifier [{}]", quantifier);
+        logger.debug("met [{}]", met);
+
+        ExecutableArrayCompareCondition condition = new ExecutableArrayCompareCondition(new ArrayCompareCondition("ctx.payload.value", "doc_count", op, value, quantifier), logger, SystemClock.INSTANCE);
+        WatchExecutionContext ctx = mockExecutionContext("_name", new Payload.Simple("value", values));
+        assertThat(condition.execute(ctx).met(), is(met));
+    }
+
+    @Test
+    public void testExecuteDateMath() {
+        ClockMock clock = new ClockMock();
+        boolean met = randomBoolean();
+        ArrayCompareCondition.Op op = met ? randomFrom(ArrayCompareCondition.Op.GT, ArrayCompareCondition.Op.GTE, ArrayCompareCondition.Op.NOT_EQ) : randomFrom(ArrayCompareCondition.Op.LT, ArrayCompareCondition.Op.LTE, ArrayCompareCondition.Op.EQ);
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.ALL, ArrayCompareCondition.Quantifier.SOME);
+        String value = "<{now-1d}>";
+        int numberOfValues = randomIntBetween(1, 10);
+        List<Object> values = new ArrayList<>(numberOfValues);
+        for (int i = 0; i < numberOfValues; i++) {
+            clock.fastForwardSeconds(1);
+            values.add(clock.nowUTC());
+        }
+
+        ExecutableArrayCompareCondition condition = new ExecutableArrayCompareCondition(new ArrayCompareCondition("ctx.payload.value", "", op, value, quantifier), logger, clock);
+        WatchExecutionContext ctx = mockExecutionContext("_name", new Payload.Simple("value", values));
+        assertThat(condition.execute(ctx).met(), is(met));
+    }
+
+    @Test
+    public void testParse() throws IOException {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        Object value = randomFrom("value", 1, null);
+        ArrayCompareConditionFactory factory = new ArrayCompareConditionFactory(Settings.EMPTY, SystemClock.INSTANCE);
+        XContentBuilder builder =
+                jsonBuilder().startObject()
+                    .startObject("key1.key2")
+                        .field("path", "key3.key4")
+                        .startObject(op.id())
+                            .field("value", value)
+                            .field("quantifier", quantifier.id())
+                        .endObject()
+                    .endObject()
+                .endObject();
+
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        parser.nextToken();
+
+        ArrayCompareCondition condition = factory.parseCondition("_id", parser);
+
+        assertThat(condition, notNullValue());
+        assertThat(condition.getArrayPath(), is("key1.key2"));
+        assertThat(condition.getOp(), is(op));
+        assertThat(condition.getValue(), is(value));
+        assertThat(condition.getPath(), is("key3.key4"));
+        assertThat(condition.getQuantifier(), is(quantifier));
+    }
+
+    @Test
+    public void testParseContainsDuplicateOperator() throws IOException {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        Object value = randomFrom("value", 1, null);
+        ArrayCompareConditionFactory factory = new ArrayCompareConditionFactory(Settings.EMPTY, SystemClock.INSTANCE);
+        XContentBuilder builder =
+                jsonBuilder().startObject()
+                    .startObject("key1.key2")
+                        .field("path", "key3.key4")
+                        .startObject(op.id())
+                            .field("value", value)
+                            .field("quantifier", quantifier.id())
+                        .endObject()
+                        .startObject(op.id())
+                            .field("value", value)
+                            .field("quantifier", quantifier.id())
+                        .endObject()
+                    .endObject()
+                .endObject();
+
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        parser.nextToken();
+
+        expectedException.expect(ElasticsearchParseException.class);
+        expectedException.expectMessage("duplicate comparison operator");
+
+        factory.parseCondition("_id", parser);
+    }
+
+    @Test
+    public void testParseContainsUnknownOperator() throws IOException {
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        Object value = randomFrom("value", 1, null);
+        ArrayCompareConditionFactory factory = new ArrayCompareConditionFactory(Settings.EMPTY, SystemClock.INSTANCE);
+        XContentBuilder builder =
+                jsonBuilder().startObject()
+                    .startObject("key1.key2")
+                        .field("path", "key3.key4")
+                        .startObject("unknown")
+                            .field("value", value)
+                            .field("quantifier", quantifier.id())
+                        .endObject()
+                    .endObject()
+                .endObject();
+
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        parser.nextToken();
+
+        expectedException.expect(ElasticsearchParseException.class);
+        expectedException.expectMessage("unknown comparison operator");
+
+        factory.parseCondition("_id", parser);
+    }
+
+    @Test
+    public void testParseContainsDuplicateValue() throws IOException {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        Object value = randomFrom("value", 1, null);
+        ArrayCompareConditionFactory factory = new ArrayCompareConditionFactory(Settings.EMPTY, SystemClock.INSTANCE);
+        XContentBuilder builder =
+                jsonBuilder().startObject()
+                    .startObject("key1.key2")
+                        .field("path", "key3.key4")
+                        .startObject(op.id())
+                            .field("value", value)
+                            .field("value", value)
+                            .field("quantifier", quantifier.id())
+                        .endObject()
+                    .endObject()
+                .endObject();
+
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        parser.nextToken();
+
+        expectedException.expect(ElasticsearchParseException.class);
+        expectedException.expectMessage("duplicate field \"value\"");
+
+        factory.parseCondition("_id", parser);
+    }
+
+    @Test
+    public void testParseContainsDuplicateQuantifier() throws IOException {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        Object value = randomFrom("value", 1, null);
+        ArrayCompareConditionFactory factory = new ArrayCompareConditionFactory(Settings.EMPTY, SystemClock.INSTANCE);
+        XContentBuilder builder =
+                jsonBuilder().startObject()
+                    .startObject("key1.key2")
+                        .field("path", "key3.key4")
+                        .startObject(op.id())
+                            .field("value", value)
+                            .field("quantifier", quantifier.id())
+                            .field("quantifier", quantifier.id())
+                        .endObject()
+                    .endObject()
+                .endObject();
+
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        parser.nextToken();
+
+        expectedException.expect(ElasticsearchParseException.class);
+        expectedException.expectMessage("duplicate field \"quantifier\"");
+
+        factory.parseCondition("_id", parser);
+    }
+
+    @Test
+    public void testParseContainsUnknownQuantifier() throws IOException {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        Object value = randomFrom("value", 1, null);
+        ArrayCompareConditionFactory factory = new ArrayCompareConditionFactory(Settings.EMPTY, SystemClock.INSTANCE);
+        XContentBuilder builder =
+                jsonBuilder().startObject()
+                    .startObject("key1.key2")
+                        .field("path", "key3.key4")
+                        .startObject(op.id())
+                            .field("value", value)
+                            .field("quantifier", "unknown")
+                        .endObject()
+                    .endObject()
+                .endObject();
+
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        parser.nextToken();
+
+        expectedException.expect(ElasticsearchParseException.class);
+        expectedException.expectMessage("unknown comparison quantifier");
+
+        factory.parseCondition("_id", parser);
+    }
+
+    @Test
+    public void testParseContainsUnexpectedFieldInComparisonOperator() throws IOException {
+        ArrayCompareCondition.Op op = randomFrom(ArrayCompareCondition.Op.values());
+        ArrayCompareCondition.Quantifier quantifier = randomFrom(ArrayCompareCondition.Quantifier.values());
+        Object value = randomFrom("value", 1, null);
+        ArrayCompareConditionFactory factory = new ArrayCompareConditionFactory(Settings.EMPTY, SystemClock.INSTANCE);
+        XContentBuilder builder =
+                jsonBuilder().startObject()
+                    .startObject("key1.key2")
+                        .field("path", "key3.key4")
+                        .startObject(op.id())
+                            .field("value", value)
+                            .field("quantifier", quantifier.id())
+                            .field("unexpected", "unexpected")
+                        .endObject()
+                    .endObject()
+                .endObject();
+
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        parser.nextToken();
+
+        expectedException.expect(ElasticsearchParseException.class);
+        expectedException.expectMessage("expected a field indicating the comparison value or comparison quantifier");
+
+        factory.parseCondition("_id", parser);
+    }
+}

watcher/src/test/java/org/elasticsearch/watcher/watch/WatchTests.java

@@ -42,6 +42,9 @@ import org.elasticsearch.watcher.condition.compare.CompareCondition;
 import org.elasticsearch.watcher.condition.compare.CompareCondition.Op;
 import org.elasticsearch.watcher.condition.compare.CompareConditionFactory;
 import org.elasticsearch.watcher.condition.compare.ExecutableCompareCondition;
+import org.elasticsearch.watcher.condition.compare.array.ArrayCompareCondition;
+import org.elasticsearch.watcher.condition.compare.array.ArrayCompareConditionFactory;
+import org.elasticsearch.watcher.condition.compare.array.ExecutableArrayCompareCondition;
 import org.elasticsearch.watcher.condition.script.ExecutableScriptCondition;
 import org.elasticsearch.watcher.condition.script.ScriptCondition;
 import org.elasticsearch.watcher.condition.script.ScriptConditionFactory;
@@ -328,12 +331,14 @@ public class WatchTests extends ESTestCase {
     }
 
     private ExecutableCondition randomCondition() {
-        String type = randomFrom(ScriptCondition.TYPE, AlwaysCondition.TYPE, CompareCondition.TYPE);
+        String type = randomFrom(ScriptCondition.TYPE, AlwaysCondition.TYPE, CompareCondition.TYPE, ArrayCompareCondition.TYPE);
         switch (type) {
             case ScriptCondition.TYPE:
                 return new ExecutableScriptCondition(new ScriptCondition(Script.inline("_script").build()), logger, scriptService);
             case CompareCondition.TYPE:
                 return new ExecutableCompareCondition(new CompareCondition("_path", randomFrom(Op.values()), randomFrom(5, "3")), logger, SystemClock.INSTANCE);
+            case ArrayCompareCondition.TYPE:
+                return new ExecutableArrayCompareCondition(new ArrayCompareCondition("_array_path", "_path", randomFrom(ArrayCompareCondition.Op.values()), randomFrom(5, "3"), ArrayCompareCondition.Quantifier.SOME), logger, SystemClock.INSTANCE);
             default:
                 return new ExecutableAlwaysCondition(logger);
         }
@@ -348,6 +353,9 @@ public class WatchTests extends ESTestCase {
             case CompareCondition.TYPE:
                 parsers.put(CompareCondition.TYPE, new CompareConditionFactory(settings, SystemClock.INSTANCE));
                 return new ConditionRegistry(parsers.build());
+            case ArrayCompareCondition.TYPE:
+                parsers.put(ArrayCompareCondition.TYPE, new ArrayCompareConditionFactory(settings, SystemClock.INSTANCE));
+                return new ConditionRegistry(parsers.build());
             default:
                 parsers.put(AlwaysCondition.TYPE, new AlwaysConditionFactory(settings));
                 return new ConditionRegistry(parsers.build());