Allow range types to be used for enrich matching

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/enrich/EnrichPolicy.java

@@ -38,9 +38,11 @@ public final class EnrichPolicy implements Writeable, ToXContentFragment {
 
     public static final String MATCH_TYPE = "match";
     public static final String GEO_MATCH_TYPE = "geo_match";
+    public static final String RANGE_TYPE = "range";
     public static final String[] SUPPORTED_POLICY_TYPES = new String[]{
         MATCH_TYPE,
-        GEO_MATCH_TYPE
+        GEO_MATCH_TYPE,
+        RANGE_TYPE
     };
 
     private static final ParseField QUERY = new ParseField("query");

x-pack/plugin/enrich/qa/common/src/main/java/org/elasticsearch/test/enrich/CommonEnrichRestTestCase.java

@@ -28,19 +28,22 @@
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.notNullValue;
 
 public abstract class CommonEnrichRestTestCase extends ESRestTestCase {
 
     @After
     public void deletePolicies() throws Exception {
         Map<String, Object> responseMap = toMap(adminClient().performRequest(new Request("GET", "/_enrich/policy")));
-        @SuppressWarnings("unchecked")
-        List<Map<?, ?>> policies = (List<Map<?, ?>>) responseMap.get("policies");
+        List<Map<?, ?>> policies = unsafeGetProperty(responseMap, "policies");
 
         for (Map<?, ?> entry : policies) {
-            client().performRequest(new Request("DELETE", "/_enrich/policy/" + XContentMapValues.extractValue("config.match.name", entry)));
+            Map<?, Map<String, ?>> config = unsafeGetProperty(entry, "config");
+            Map<String, ?> policy = config.values().iterator().next();
+            String endpoint = "/_enrich/policy/" + policy.get("name");
+            assertOK(client().performRequest(new Request("DELETE", endpoint)));
 
-            List<?> sourceIndices = (List<?>) XContentMapValues.extractValue("config.match.indices", entry);
+            List<?> sourceIndices = (List<?>) policy.get("indices");
             for (Object sourceIndex : sourceIndices) {
                 try {
                     client().performRequest(new Request("DELETE", "/" + sourceIndex));
@@ -51,17 +54,49 @@ public void deletePolicies() throws Exception {
         }
     }
 
-    private void setupGenericLifecycleTest(boolean deletePipeilne) throws Exception {
+    @SuppressWarnings("unchecked")
+    private <Property> Property unsafeGetProperty(Map<?, ?> map, String key) {
+        return (Property) map.get(key);
+    }
+
+    private void setupGenericLifecycleTest(boolean deletePipeilne, String field, String type, String value) throws Exception {
         // Create source index:
         createSourceIndex("my-source-index");
+
         // Create the policy:
         Request putPolicyRequest = new Request("PUT", "/_enrich/policy/my_policy");
-        putPolicyRequest.setJsonEntity(generatePolicySource("my-source-index"));
+        putPolicyRequest.setJsonEntity(generatePolicySource("my-source-index", field, type));
         assertOK(client().performRequest(putPolicyRequest));
 
         // Add entry to source index and then refresh:
         Request indexRequest = new Request("PUT", "/my-source-index/_doc/elastic.co");
-        indexRequest.setJsonEntity("{\"host\": \"elastic.co\",\"globalRank\": 25,\"tldRank\": 7,\"tld\": \"co\"}");
+        indexRequest.setJsonEntity("{" +
+            "\"host\": \"elastic.co\"," +
+            "\"globalRank\": 25," +
+            "\"tldRank\": 7," +
+            "\"tld\": \"co\", " +
+            "\"date\": {" +
+                "\"gte\" : \"2021-09-05\"," +
+                "\"lt\" : \"2021-09-07\"" +
+            "}, " +
+            "\"integer\": {" +
+                "\"gte\" : 40," +
+                "\"lt\" : 42" +
+            "}, " +
+            "\"long\": {" +
+                "\"gte\" : 8000000," +
+                "\"lt\" : 9000000" +
+            "}, " +
+            "\"double\": {" +
+                "\"gte\" : 10.10," +
+                "\"lt\" : 20.20" +
+            "}, " +
+            "\"float\": {" +
+                "\"gte\" : 10000.5," +
+                "\"lt\" : 10000.7" +
+            "}, " +
+            "\"ip\": \"100.0.0.0/4\"" +
+            "}");
         assertOK(client().performRequest(indexRequest));
         Request refreshRequest = new Request("POST", "/my-source-index/_refresh");
         assertOK(client().performRequest(refreshRequest));
@@ -73,34 +108,66 @@ private void setupGenericLifecycleTest(boolean deletePipeilne) throws Exception
         // Create pipeline
         Request putPipelineRequest = new Request("PUT", "/_ingest/pipeline/my_pipeline");
         putPipelineRequest.setJsonEntity(
-            "{\"processors\":[" + "{\"enrich\":{\"policy_name\":\"my_policy\",\"field\":\"host\",\"target_field\":\"entry\"}}" + "]}"
+            "{\"processors\":[" + "{\"enrich\":{\"policy_name\":\"my_policy\",\"field\":\""+field+"\",\"target_field\":\"entry\"}}" + "]}"
         );
         assertOK(client().performRequest(putPipelineRequest));
 
         // Index document using pipeline with enrich processor:
         indexRequest = new Request("PUT", "/my-index/_doc/1");
         indexRequest.addParameter("pipeline", "my_pipeline");
-        indexRequest.setJsonEntity("{\"host\": \"elastic.co\"}");
+        indexRequest.setJsonEntity("{\""+field+"\": \""+value+"\"}");
         assertOK(client().performRequest(indexRequest));
 
         // Check if document has been enriched
         Request getRequest = new Request("GET", "/my-index/_doc/1");
         Map<String, Object> response = toMap(client().performRequest(getRequest));
         Map<?, ?> entry = (Map<?, ?>) ((Map<?, ?>) response.get("_source")).get("entry");
         assertThat(entry.size(), equalTo(4));
-        assertThat(entry.get("host"), equalTo("elastic.co"));
+        assertThat(entry.get(field), notNullValue());
         assertThat(entry.get("tld"), equalTo("co"));
         assertThat(entry.get("globalRank"), equalTo(25));
         assertThat(entry.get("tldRank"), equalTo(7));
+        Object originalMatchValue = ((Map<?, ?>) response.get("_source")).get(field);
+        assertThat(originalMatchValue, equalTo(value));
 
         if (deletePipeilne) {
             // delete the pipeline so the policies can be deleted
             client().performRequest(new Request("DELETE", "/_ingest/pipeline/my_pipeline"));
         }
     }
 
-    public void testBasicFlow() throws Exception {
-        setupGenericLifecycleTest(true);
+    public void testBasicFlowKeyword() throws Exception {
+        setupGenericLifecycleTest(true, "host", "match", "elastic.co");
+        assertBusy(CommonEnrichRestTestCase::verifyEnrichMonitoring, 3, TimeUnit.MINUTES);
+    }
+
+    public void testBasicFlowDate() throws Exception {
+        setupGenericLifecycleTest(true, "date", "range", "2021-09-06");
+        assertBusy(CommonEnrichRestTestCase::verifyEnrichMonitoring, 3, TimeUnit.MINUTES);
+    }
+
+    public void testBasicFlowInteger() throws Exception {
+        setupGenericLifecycleTest(true, "integer", "range", "41");
+        assertBusy(CommonEnrichRestTestCase::verifyEnrichMonitoring, 3, TimeUnit.MINUTES);
+    }
+
+    public void testBasicFlowLong() throws Exception {
+        setupGenericLifecycleTest(true, "long", "range", "8411017");
+        assertBusy(CommonEnrichRestTestCase::verifyEnrichMonitoring, 3, TimeUnit.MINUTES);
+    }
+
+    public void testBasicFlowDouble() throws Exception {
+        setupGenericLifecycleTest(true, "double", "range", "15.15");
+        assertBusy(CommonEnrichRestTestCase::verifyEnrichMonitoring, 3, TimeUnit.MINUTES);
+    }
+
+    public void testBasicFlowFloat() throws Exception {
+        setupGenericLifecycleTest(true, "float", "range", "10000.66666");
+        assertBusy(CommonEnrichRestTestCase::verifyEnrichMonitoring, 3, TimeUnit.MINUTES);
+    }
+
+    public void testBasicFlowIp() throws Exception {
+        setupGenericLifecycleTest(true, "ip", "range", "100.120.140.160");
         assertBusy(CommonEnrichRestTestCase::verifyEnrichMonitoring, 3, TimeUnit.MINUTES);
     }
 
@@ -129,7 +196,7 @@ public void testDeleteIsCaseSensitive() throws Exception {
 
     public void testDeleteExistingPipeline() throws Exception {
         // lets not delete the pipeline at first, to test the failure
-        setupGenericLifecycleTest(false);
+        setupGenericLifecycleTest(false, "host", "match", "elastic.co");
 
         Request putPipelineRequest = new Request("PUT", "/_ingest/pipeline/another_pipeline");
         putPipelineRequest.setJsonEntity(
@@ -156,14 +223,18 @@ public void testDeleteExistingPipeline() throws Exception {
     }
 
     public static String generatePolicySource(String index) throws IOException {
-        XContentBuilder source = jsonBuilder().startObject().startObject("match");
+        return generatePolicySource(index, "host", "match");
+    }
+
+    public static String generatePolicySource(String index, String field, String type) throws IOException {
+        XContentBuilder source = jsonBuilder().startObject().startObject(type);
         {
             source.field("indices", index);
             if (randomBoolean()) {
                 source.field("query", QueryBuilders.matchAllQuery());
             }
-            source.field("match_field", "host");
-            source.field("enrich_fields", new String[] { "globalRank", "tldRank", "tld" });
+            source.field("match_field", field);
+            source.field("enrich_fields", new String[]{"globalRank", "tldRank", "tld"});
         }
         source.endObject().endObject();
         return Strings.toString(source);
@@ -179,7 +250,13 @@ public static String createSourceIndexMapping() {
             + "{\"host\": {\"type\":\"keyword\"},"
             + "\"globalRank\":{\"type\":\"keyword\"},"
             + "\"tldRank\":{\"type\":\"keyword\"},"
-            + "\"tld\":{\"type\":\"keyword\"}"
+            + "\"tld\":{\"type\":\"keyword\"},"
+            + "\"date\":{\"type\":\"date_range\", \"format\": \"yyyy-MM-dd\"},"
+            + "\"integer\":{\"type\":\"integer_range\"},"
+            + "\"long\":{\"type\":\"long_range\"},"
+            + "\"double\":{\"type\":\"double_range\"},"
+            + "\"float\":{\"type\":\"float_range\"},"
+            + "\"ip\":{\"type\":\"ip_range\"}"
             + "}";
     }
 
@@ -207,8 +284,8 @@ private static void verifyEnrichMonitoring() throws IOException {
         List<?> hits = (List<?>) XContentMapValues.extractValue("hits.hits", response);
         assertThat(hits.size(), greaterThanOrEqualTo(1));
 
-        for (int i = 0; i < hits.size(); i++) {
-            Map<?, ?> hit = (Map<?, ?>) hits.get(i);
+        for (Object o : hits) {
+            Map<?, ?> hit = (Map<?, ?>) o;
 
             int foundRemoteRequestsTotal = (int) XContentMapValues.extractValue(
                 "_source.enrich_coordinator_stats.remote_requests_total",