Add enrich processor

The enrich processor performs a lookup in a locally allocated
enrich index shard using a field value from the document being enriched.
If there is a match then the _source of the enrich document is fetched.
The document being enriched then gets the decorate values from the
enrich document based on the configured decorate fields in the pipeline.

Note that the usage of the _source field is temporary until the enrich
source field that is part of elastic#41521 is merged into the enrich branch.
Using the _source field involves significant decompression which not
desired for enrich use cases.

The policy contains the information what field in the enrich index
to query and what fields are available to decorate a document being
enriched with.

The enrich processor has the following configuration options:
* `policy_name` - the name of the policy this processor should use
* `enrich_key` - the field in the document being enriched that holds to lookup value
* `enrich_key_ignore_missing` - Whether to allow the key field to be missing
* `enrich_values` - a list of fields to decorate the document being enriched with.
                    Each entry holds a source field and a target field.
                    The source field indicates what decorate field to use that is available in the policy.
                    The target field controls the field name to use in the document being enriched.
                    The source and target fields can be the same.

Example pipeline config:

```
{
   "processors": [
      {
         "policy_name": "my_policy",
         "enrich_key": "host_name",
         "enrich_values": [
            {
              "source": "globalRank",
              "target": "global_rank"
            }
         ]
      }
   ]
}
```

In the above example documents are being enriched with a global rank value.
For each document that has match in the enrich index based on its host_name field,
the document gets an global rank field value, which is fetched from the `globalRank`
field in the enrich index and saved as `global_rank` in the document being enriched.

This is PR is part one of elastic#41521

Author: martijnvg

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

@@ -28,7 +28,7 @@
  */
 public final class EnrichPolicy implements Writeable, ToXContentFragment {
 
-    static final String EXACT_MATCH_TYPE = "exact_match";
+    public static final String EXACT_MATCH_TYPE = "exact_match";
     public static final String[] SUPPORTED_POLICY_TYPES = new String[]{EXACT_MATCH_TYPE};
 
     static final ParseField TYPE = new ParseField("type");

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.enrich;
 
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
@@ -21,7 +22,8 @@ public class EnrichPlugin extends Plugin implements IngestPlugin {
 
     @Override
     public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
-        return Collections.emptyMap();
+        final ClusterService clusterService = parameters.ingestService.getClusterService();
+        return Map.of(EnrichProcessorFactory.TYPE, new EnrichProcessorFactory(clusterService::state, parameters.localShardSearcher));
     }
 
     @Override

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

@@ -0,0 +1,79 @@
+/*
+ * 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.xpack.enrich;
+
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.ingest.ConfigurationUtils;
+import org.elasticsearch.ingest.Processor;
+import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+final class EnrichProcessorFactory implements Processor.Factory {
+
+    static final String TYPE = "enrich";
+
+    private final Function<String, EnrichPolicy> policyLookup;
+    private final Function<String, Engine.Searcher> searchProvider;
+
+    EnrichProcessorFactory(Supplier<ClusterState> clusterStateSupplier,
+                           Function<String, Engine.Searcher> searchProvider) {
+        this.policyLookup = policyName -> {
+            ClusterState clusterState = clusterStateSupplier.get();
+            return EnrichStore.getPolicy(policyName, clusterState);
+        };
+        this.searchProvider = searchProvider;
+    }
+
+    @Override
+    public Processor create(Map<String, Processor.Factory> processorFactories, String tag, Map<String, Object> config) throws Exception {
+        String policyName = ConfigurationUtils.readStringProperty(TYPE, tag, config, "policy_name");
+        EnrichPolicy policy = policyLookup.apply(policyName);
+        if (policy == null) {
+            throw new IllegalArgumentException("policy [" + policyName + "] does not exists");
+        }
+
+        String enrichKey = ConfigurationUtils.readStringProperty(TYPE, tag, config, "enrich_key", policy.getEnrichKey());
+        boolean ignoreKeyMissing = ConfigurationUtils.readBooleanProperty(TYPE, tag, config, "enrich_key_ignore_missing", false);
+
+        final List<EnrichSpecification> specifications;
+        final List<Map<?, ?>> specificationConfig = ConfigurationUtils.readList(TYPE, tag, config, "enrich_values");
+        specifications = specificationConfig.stream()
+            .map(entry -> new EnrichSpecification((String) entry.get("source"), (String) entry.get("target")))
+            .collect(Collectors.toList());
+
+        for (EnrichSpecification specification : specifications) {
+            if (policy.getEnrichValues().contains(specification.sourceField) == false) {
+                throw new IllegalArgumentException("source field [" + specification.sourceField + "] does not exist in policy [" +
+                    policyName + "]");
+            }
+        }
+
+        switch (policy.getType()) {
+            case EnrichPolicy.EXACT_MATCH_TYPE:
+                return new ExactMatchProcessor(tag, policyLookup, searchProvider, policyName, enrichKey, ignoreKeyMissing, specifications);
+            default:
+                throw new IllegalArgumentException("unsupported policy type [" + policy.getType() + "]");
+        }
+    }
+
+    static final class EnrichSpecification {
+
+        final String sourceField;
+        final String targetField;
+
+        EnrichSpecification(String sourceField, String targetField) {
+            this.sourceField = sourceField;
+            this.targetField = targetField;
+        }
+    }
+
+}

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

@@ -0,0 +1,124 @@
+/*
+ * 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.xpack.enrich;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.PostingsEnum;
+import org.apache.lucene.index.Terms;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
+import org.elasticsearch.ingest.AbstractProcessor;
+import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+final class ExactMatchProcessor extends AbstractProcessor {
+
+    private final Function<String, EnrichPolicy> policyLookup;
+    private final Function<String, Engine.Searcher> searchProvider;
+
+    private final String policyName;
+    private final String enrichKey;
+    private final boolean ignoreEnrichKeyMissing;
+    private final List<EnrichProcessorFactory.EnrichSpecification> specifications;
+
+    ExactMatchProcessor(String tag,
+                        Function<String, EnrichPolicy> policyLookup,
+                        Function<String, Engine.Searcher> searchProvider, String policyName,
+                        String enrichKey,
+                        boolean ignoreEnrichKeyMissing,
+                        List<EnrichProcessorFactory.EnrichSpecification> specifications) {
+        super(tag);
+        this.policyLookup = policyLookup;
+        this.searchProvider = searchProvider;
+        this.policyName = policyName;
+        this.enrichKey = enrichKey;
+        this.ignoreEnrichKeyMissing = ignoreEnrichKeyMissing;
+        this.specifications = specifications;
+    }
+
+    @Override
+    public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
+        final EnrichPolicy policy = policyLookup.apply(policyName);
+        if (policy == null) {
+            throw new IllegalArgumentException("policy [" + policyName + "] does not exists");
+        }
+
+        final String value = ingestDocument.getFieldValue(enrichKey, String.class, ignoreEnrichKeyMissing);
+        if (value == null) {
+            return ingestDocument;
+        }
+
+        // TODO: re-use the engine searcher between enriching documents from the same write request
+        try (Engine.Searcher engineSearcher = searchProvider.apply(policy.getIndexPattern())) {
+            if (engineSearcher.getDirectoryReader().leaves().size() == 0) {
+                return ingestDocument;
+            } else if (engineSearcher.getDirectoryReader().leaves().size() != 1) {
+                throw new IllegalStateException("enrich index must have exactly a single segment");
+            }
+
+            final LeafReader leafReader = engineSearcher.getDirectoryReader().leaves().get(0).reader();
+            final Terms terms = leafReader.terms(policy.getEnrichKey());
+            if (terms == null) {
+                throw new IllegalStateException("enrich key field [" + policy.getEnrichKey() + "] does not exist");
+            }
+
+            final TermsEnum tenum = terms.iterator();
+            if (tenum.seekExact(new BytesRef(value))) {
+                PostingsEnum penum = tenum.postings(null, PostingsEnum.NONE);
+                final int docId = penum.nextDoc();
+                assert docId != PostingsEnum.NO_MORE_DOCS : "no matching doc id for [" + enrichKey + "]";
+                assert penum.nextDoc() == PostingsEnum.NO_MORE_DOCS : "more than one doc id matching for [" + enrichKey + "]";
+
+                // TODO: The use of _source is temporarily until enrich source field mapper has been added (see PR #41521)
+                Document document = leafReader.document(docId, Set.of(SourceFieldMapper.NAME));
+                BytesRef source = document.getBinaryValue(SourceFieldMapper.NAME);
+                assert source != null;
+
+                final BytesReference encoded = new BytesArray(source);
+                final Map<String, Object> decoded =
+                    XContentHelper.convertToMap(encoded, false, XContentType.SMILE).v2();
+                for (EnrichProcessorFactory.EnrichSpecification specification : specifications) {
+                    Object enrichValue = decoded.get(specification.sourceField);
+                    ingestDocument.setFieldValue(specification.targetField, enrichValue);
+                }
+            }
+        }
+        return ingestDocument;
+    }
+
+    @Override
+    public String getType() {
+        return EnrichProcessorFactory.TYPE;
+    }
+
+    String getPolicyName() {
+        return policyName;
+    }
+
+    String getEnrichKey() {
+        return enrichKey;
+    }
+
+    boolean isIgnoreEnrichKeyMissing() {
+        return ignoreEnrichKeyMissing;
+    }
+
+    List<EnrichProcessorFactory.EnrichSpecification> getSpecifications() {
+        return specifications;
+    }
+}