Add enrich processor and enrich source meta field mapper.

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/build.gradle

@@ -20,5 +20,6 @@ run {
     plugin xpackModule('core')
 }
 
-// No tests yet:
-integTest.enabled = false
+integTestCluster {
+    plugin xpackModule('core')
+}

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

@@ -6,22 +6,32 @@
 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;
+import org.elasticsearch.index.mapper.MetadataFieldMapper;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.plugins.IngestPlugin;
+import org.elasticsearch.plugins.MapperPlugin;
 import org.elasticsearch.plugins.Plugin;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
-public class EnrichPlugin extends Plugin implements IngestPlugin {
+public class EnrichPlugin extends Plugin implements IngestPlugin, MapperPlugin {
 
     @Override
-    public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
-        return Collections.emptyMap();
+    public Map<String, MetadataFieldMapper.TypeParser> getMetadataMappers() {
+        return Collections.singletonMap(EnrichSourceFieldMapper.NAME, new EnrichSourceFieldMapper.TypeParser());
+    }
+
+    @Override
+    public Map<String, Processor.Factory> getProcessors(final Processor.Parameters parameters) {
+        final ClusterService clusterService = parameters.ingestService.getClusterService();
+        return Collections.singletonMap(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,72 @@
+/*
+ * 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 key = ConfigurationUtils.readStringProperty(TYPE, tag, config, "key", policy.getEnrichKey());
+        boolean ignoreKeyMissing = ConfigurationUtils.readBooleanProperty(TYPE, tag, config, "key_ignore_missing", false);
+
+        final List<EnrichSpecification> specifications;
+        final List<Map<?, ?>> specificationConfig = ConfigurationUtils.readList(TYPE, tag, config, "values");
+        specifications = specificationConfig.stream()
+            .map(entry -> new EnrichSpecification((String) entry.get("source"), (String) entry.get("target")))
+            .collect(Collectors.toList());
+
+        switch (policy.getType()) {
+            case EnrichPolicy.EXACT_MATCH_TYPE:
+                return new ExactMatchProcessor(tag, policyLookup, searchProvider, policyName, key, 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;
+        }
+    }
+
+}