Add support for a more compact enrich values format (#45033)

In the case that source and target are the same in `enrich_values` then
a string array can be specified.

For example instead of this:

```
PUT /_ingest/pipeline/my-pipeline
{
    "processors": [
        {
            "enrich" : {
                "policy_name": "my-policy",
                "enrich_values": [
                    {
                        "source": "first_name",
                        "target": "first_name"
                    },
                    {
                        "source": "last_name",
                        "target": "last_name"
                    },
                    {
                        "source": "address",
                        "target": "address"
                    },
                    {
                        "source": "city",
                        "target": "city"
                    },
                    {
                        "source": "state",
                        "target": "state"
                    },
                    {
                        "source": "zip",
                        "target": "zip"
                    }
                ]
            }
        }
    ]
}
```
This more compact format can be specified:

```
PUT /_ingest/pipeline/my-pipeline
{
    "processors": [
        {
            "enrich" : {
                "policy_name": "my-policy",
                "targets": [
                   "first_name",
                   "last_name",
                   "address",
                   "city",
                   "state",
                   "zip"
                ]
            }
        }
    ]
}
```

And the `enrich_values` key has been renamed to `set_from`.

Relates to #32789

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

@@ -54,7 +54,7 @@ public abstract class CommonEnrichRestTestCase extends ESRestTestCase {
         // Create pipeline
         Request putPipelineRequest = new Request("PUT", "/_ingest/pipeline/my_pipeline");
         putPipelineRequest.setJsonEntity("{\"processors\":[" +
-            "{\"enrich\":{\"policy_name\":\"my_policy\",\"enrich_key\":\"host\",\"enrich_values\":[" +
+            "{\"enrich\":{\"policy_name\":\"my_policy\",\"enrich_key\":\"host\",\"set_from\":[" +
             "{\"source\":\"globalRank\",\"target\":\"global_rank\"}," +
             "{\"source\":\"tldRank\",\"target\":\"tld_rank\"}" +
             "]}}" +

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

@@ -39,11 +39,26 @@ final class EnrichProcessorFactory implements Processor.Factory, Consumer<Cluste
         boolean overrideEnabled = ConfigurationUtils.readBooleanProperty(TYPE, tag, config, "override", true);
 
         final List<EnrichSpecification> specifications;
-        final List<Map<?, ?>> specificationConfig = ConfigurationUtils.readList(TYPE, tag, config, "enrich_values");
-        specifications = specificationConfig.stream()
+        final List<Map<?, ?>> setFromConfig = ConfigurationUtils.readOptionalList(TYPE, tag, config, "set_from");
+        if (setFromConfig != null) {
+            if (setFromConfig.isEmpty()) {
+                throw new IllegalArgumentException("provided set_from is empty");
+            }
+
             // TODO: Add templating support in enrich_values source and target options
-            .map(entry -> new EnrichSpecification((String) entry.get("source"), (String) entry.get("target")))
-            .collect(Collectors.toList());
+            specifications = setFromConfig.stream()
+                .map(entry -> new EnrichSpecification((String) entry.get("source"), (String) entry.get("target")))
+                .collect(Collectors.toList());
+        } else {
+            final List<String> targetsConfig = ConfigurationUtils.readList(TYPE, tag, config, "targets");
+            if (targetsConfig.isEmpty()) {
+                throw new IllegalArgumentException("provided targets is empty");
+            }
+
+            specifications = targetsConfig.stream()
+                .map(value -> new EnrichSpecification(value, value))
+                .collect(Collectors.toList());
+        }
 
         for (EnrichSpecification specification : specifications) {
             if (policy.getEnrichValues().contains(specification.sourceField) == false) {

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/BasicEnrichTests.java

@@ -53,7 +53,7 @@ public class BasicEnrichTests extends ESSingleNodeTestCase {
 
         String pipelineName = "my-pipeline";
         String pipelineBody = "{\"processors\": [{\"enrich\": {\"policy_name\":\"" + policyName +
-            "\", \"enrich_values\": [{\"source\": \"" + DECORATE_FIELDS[0] + "\", \"target\": \"" + DECORATE_FIELDS[0] + "\"}," +
+            "\", \"set_from\": [{\"source\": \"" + DECORATE_FIELDS[0] + "\", \"target\": \"" + DECORATE_FIELDS[0] + "\"}," +
             "{\"source\": \"" + DECORATE_FIELDS[1] + "\", \"target\": \"" + DECORATE_FIELDS[1] + "\"}," +
             "{\"source\": \"" + DECORATE_FIELDS[2] + "\", \"target\": \"" + DECORATE_FIELDS[2] + "\"}" +
             "]}}]}";
@@ -100,7 +100,7 @@ public class BasicEnrichTests extends ESSingleNodeTestCase {
 
             String pipelineName = "pipeline" + i;
             String pipelineBody = "{\"processors\": [{\"enrich\": {\"policy_name\":\"" + policyName +
-                "\", \"enrich_values\": [{\"source\": \"value\", \"target\": \"value\"}" +
+                "\", \"set_from\": [{\"source\": \"value\", \"target\": \"value\"}" +
                 "]}}]}";
             PutPipelineRequest putPipelineRequest = new PutPipelineRequest(pipelineName, new BytesArray(pipelineBody), XContentType.JSON);
             client().admin().cluster().putPipeline(putPipelineRequest).actionGet();

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichMultiNodeIT.java

@@ -165,7 +165,7 @@ public class EnrichMultiNodeIT extends ESIntegTestCase {
 
     private static void createPipeline() {
         String pipelineBody = "{\"processors\": [{\"enrich\": {\"policy_name\":\"" + POLICY_NAME +
-            "\", \"enrich_values\": [{\"source\": \"" + DECORATE_FIELDS[0] + "\", \"target\": \"" + DECORATE_FIELDS[0] + "\"}," +
+            "\", \"set_from\": [{\"source\": \"" + DECORATE_FIELDS[0] + "\", \"target\": \"" + DECORATE_FIELDS[0] + "\"}," +
             "{\"source\": \"" + DECORATE_FIELDS[1] + "\", \"target\": \"" + DECORATE_FIELDS[1] + "\"}," +
             "{\"source\": \"" + DECORATE_FIELDS[2] + "\", \"target\": \"" + DECORATE_FIELDS[2] + "\"}" +
             "]}}]}";

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichPolicyUpdateTests.java

@@ -41,7 +41,7 @@ public class EnrichPolicyUpdateTests extends ESSingleNodeTestCase {
         assertAcked(client().execute(PutEnrichPolicyAction.INSTANCE, putPolicyRequest).actionGet());
         assertThat(enrichProcessorFactory.policies.get("my_policy"), equalTo(instance1));
 
-        String pipelineConfig = "{\"processors\":[{\"enrich\": {\"policy_name\": \"my_policy\", \"enrich_values\": []}}]}";
+        String pipelineConfig = "{\"processors\":[{\"enrich\": {\"policy_name\": \"my_policy\", \"targets\": [\"field1\"]}}]}";
         PutPipelineRequest putPipelineRequest = new PutPipelineRequest("1", new BytesArray(pipelineConfig), XContentType.JSON);
         assertAcked(client().admin().cluster().putPipeline(putPipelineRequest).actionGet());
         Pipeline pipelineInstance1 = ingestService.getPipeline("1");

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichProcessorFactoryTests.java

@@ -53,7 +53,7 @@ public class EnrichProcessorFactoryTests extends ESTestCase {
         for (Tuple<String, String> tuple : randomValues) {
             valuesConfig.add(Map.of("source", tuple.v1(), "target", tuple.v2()));
         }
-        config.put("enrich_values", valuesConfig);
+        config.put("set_from", valuesConfig);
 
         ExactMatchProcessor result = (ExactMatchProcessor) factory.create(Collections.emptyMap(), "_tag", config);
         assertThat(result, notNullValue());
@@ -96,7 +96,7 @@ public class EnrichProcessorFactoryTests extends ESTestCase {
         for (Tuple<String, String> tuple : randomValues) {
             valuesConfig.add(Map.of("source", tuple.v1(), "target", tuple.v2()));
         }
-        config.put("enrich_values", valuesConfig);
+        config.put("set_from", valuesConfig);
 
         Exception e = expectThrows(IllegalArgumentException.class, () -> factory.create(Collections.emptyMap(), "_tag", config));
         assertThat(e.getMessage(), equalTo("policy [majestic] does not exists"));
@@ -126,7 +126,7 @@ public class EnrichProcessorFactoryTests extends ESTestCase {
         for (Tuple<String, String> tuple : randomValues) {
             valuesConfig.add(Map.of("source", tuple.v1(), "target", tuple.v2()));
         }
-        config.put("enrich_values", valuesConfig);
+        config.put("set_from", valuesConfig);
 
         Exception e = expectThrows(ElasticsearchParseException.class, () -> factory.create(Collections.emptyMap(), "_tag", config));
         assertThat(e.getMessage(), equalTo("[policy_name] required property is missing"));
@@ -156,7 +156,7 @@ public class EnrichProcessorFactoryTests extends ESTestCase {
         for (Tuple<String, String> tuple : randomValues) {
             valuesConfig.add(Map.of("source", tuple.v1(), "target", tuple.v2()));
         }
-        config.put("enrich_values", valuesConfig);
+        config.put("set_from", valuesConfig);
 
         Exception e = expectThrows(IllegalArgumentException.class, () -> factory.create(Collections.emptyMap(), "_tag", config));
         assertThat(e.getMessage(), equalTo("unsupported policy type [unsupported]"));
@@ -173,10 +173,55 @@ public class EnrichProcessorFactoryTests extends ESTestCase {
         config.put("policy_name", "majestic");
         config.put("enrich_key", "host");
         List<Map<String, Object>> valuesConfig = List.of(Map.of("source", "rank", "target", "rank"));
-        config.put("enrich_values", valuesConfig);
+        config.put("set_from", valuesConfig);
 
         Exception e = expectThrows(IllegalArgumentException.class, () -> factory.create(Collections.emptyMap(), "_tag", config));
         assertThat(e.getMessage(), equalTo("source field [rank] does not exist in policy [majestic]"));
     }
 
+    public void testCompactEnrichValuesFormat() throws Exception {
+        List<String> enrichValues = List.of("globalRank", "tldRank", "tld");
+        EnrichPolicy policy = new EnrichPolicy(EnrichPolicy.EXACT_MATCH_TYPE, null, List.of("source_index"), "host",
+            enrichValues);
+        EnrichProcessorFactory factory = new EnrichProcessorFactory(null);
+        factory.policies = Map.of("majestic", policy);
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("policy_name", "majestic");
+        config.put("targets", enrichValues);
+
+        ExactMatchProcessor result = (ExactMatchProcessor) factory.create(Collections.emptyMap(), "_tag", config);
+        assertThat(result, notNullValue());
+        assertThat(result.getPolicyName(), equalTo("majestic"));
+        assertThat(result.getEnrichKey(), equalTo("host"));
+        assertThat(result.getSpecifications().size(), equalTo(enrichValues.size()));
+        for (int i = 0; i < enrichValues.size(); i++) {
+            EnrichSpecification actual = result.getSpecifications().get(i);
+            String expected = enrichValues.get(i);
+            assertThat(actual.sourceField, equalTo(expected));
+            assertThat(actual.targetField, equalTo(expected));
+        }
+    }
+
+    public void testNoEnrichValues() throws Exception {
+        List<String> enrichValues = List.of("globalRank", "tldRank", "tld");
+        EnrichPolicy policy = new EnrichPolicy(EnrichPolicy.EXACT_MATCH_TYPE, null, List.of("source_index"), "host",
+            enrichValues);
+        EnrichProcessorFactory factory = new EnrichProcessorFactory(null);
+        factory.policies = Map.of("majestic", policy);
+
+        Map<String, Object> config1 = new HashMap<>();
+        config1.put("policy_name", "majestic");
+        config1.put("set_from", List.of());
+
+        Exception e = expectThrows(IllegalArgumentException.class, () -> factory.create(Collections.emptyMap(), "_tag", config1));
+        assertThat(e.getMessage(), equalTo("provided set_from is empty"));
+
+        Map<String, Object> config2 = new HashMap<>();
+        config2.put("policy_name", "majestic");
+        config2.put("targets", List.of());
+        e = expectThrows(IllegalArgumentException.class, () -> factory.create(Collections.emptyMap(), "_tag", config2));
+        assertThat(e.getMessage(), equalTo("provided targets is empty"));
+    }
+
 }