Merge pull request elastic/elasticsearch#574 from jasontedor/feature/array-compare-condition

Add compare condition to handle arrays

Original commit: elastic/x-pack-elasticsearch@c548b84b76

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 <<condition-script, script>>, <<condition-compare, compare>>  and <<condition-array-compare, array-compare>>
-the payload to determine whether or not  the necessary conditions have been met. 
+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> {
+public class ExecutableCompareCondition extends AbstractExecutableCompareCondition<CompareCondition, CompareCondition.Result> {
-
-    static final Pattern DATE_MATH_PATTERN = Pattern.compile("<\\{(.+)\\}>");
-    static final Pattern PATH_PATTERN = Pattern.compile("\\{\\{(.+)\\}\\}");
-
-
-    private final Clock clock;
-
     public ExecutableCompareCondition(CompareCondition condition, ESLogger logger, Clock clock) {
-        super(condition, logger);
+        super(condition, logger, clock);
-        this.clock = clock;
     }
 
     @Override
-    public CompareCondition.Result execute(WatchExecutionContext ctx) {
+    protected CompareCondition.Result doExecute(Map<String, Object> model, Map<String, Object> resolvedValues) throws Exception {
-        Map<String, Object> resolvedValues = new HashMap<>();
+        Object configuredValue = resolveConfiguredValue(resolvedValues, model, condition.getValue());
-        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);
-                }
-            }
-        }
 
         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());