Optimize PipelineConfiguration-checking ClusterStateListeners (#117038)

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/ingest/PipelineConfigurationBridge.java

@@ -28,8 +28,12 @@ public String getId() {
         return delegate.getId();
     }
 
-    public Map<String, Object> getConfigAsMap() {
-        return delegate.getConfigAsMap();
+    public Map<String, Object> getConfig() {
+        return delegate.getConfig();
+    }
+
+    public Map<String, Object> getConfig(final boolean unmodifiable) {
+        return delegate.getConfig(unmodifiable);
     }
 
     @Override

modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTaskExecutor.java

@@ -268,7 +268,7 @@ private static Set<String> pipelinesWithGeoIpProcessor(ClusterState clusterState
         Set<String> ids = new HashSet<>();
         // note: this loop is unrolled rather than streaming-style because it's hot enough to show up in a flamegraph
         for (PipelineConfiguration configuration : configurations) {
-            List<Map<String, Object>> processors = (List<Map<String, Object>>) configuration.getConfigAsMap().get(Pipeline.PROCESSORS_KEY);
+            List<Map<String, Object>> processors = (List<Map<String, Object>>) configuration.getConfig().get(Pipeline.PROCESSORS_KEY);
             if (hasAtLeastOneGeoipProcessor(processors, downloadDatabaseOnPipelineCreation)) {
                 ids.add(configuration.getId());
             }

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -203,6 +203,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INDEX_STATS_ADDITIONAL_FIELDS_REVERT = def(8_794_00_0);
     public static final TransportVersion FAST_REFRESH_RCO_2 = def(8_795_00_0);
     public static final TransportVersion ESQL_ENRICH_RUNTIME_WARNINGS = def(8_796_00_0);
+    public static final TransportVersion INGEST_PIPELINE_CONFIGURATION_AS_MAP = def(8_797_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/ingest/GetPipelineResponse.java

@@ -80,7 +80,7 @@ public RestStatus status() {
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         for (PipelineConfiguration pipeline : pipelines) {
-            builder.field(pipeline.getId(), summary ? Map.of() : pipeline.getConfigAsMap());
+            builder.field(pipeline.getId(), summary ? Map.of() : pipeline.getConfig());
         }
         builder.endObject();
         return builder;

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateService.java

@@ -782,7 +782,7 @@ private void validateUseOfDeprecatedIngestPipelines(String name, IngestMetadata
     private void emitWarningIfPipelineIsDeprecated(String name, Map<String, PipelineConfiguration> pipelines, String pipelineName) {
         Optional.ofNullable(pipelineName)
             .map(pipelines::get)
-            .filter(p -> Boolean.TRUE.equals(p.getConfigAsMap().get("deprecated")))
+            .filter(p -> Boolean.TRUE.equals(p.getConfig().get("deprecated")))
             .ifPresent(
                 p -> deprecationLogger.warn(
                     DeprecationCategory.TEMPLATES,

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -519,7 +519,7 @@ public static boolean isNoOpPipelineUpdate(ClusterState state, PutPipelineReques
             && currentIngestMetadata.getPipelines().containsKey(request.getId())) {
             var pipelineConfig = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
             var currentPipeline = currentIngestMetadata.getPipelines().get(request.getId());
-            if (currentPipeline.getConfigAsMap().equals(pipelineConfig)) {
+            if (currentPipeline.getConfig().equals(pipelineConfig)) {
                 return true;
             }
         }
@@ -1292,7 +1292,7 @@ synchronized void innerUpdatePipelines(IngestMetadata newIngestMetadata) {
             try {
                 Pipeline newPipeline = Pipeline.create(
                     newConfiguration.getId(),
-                    newConfiguration.getConfigAsMap(),
+                    newConfiguration.getConfig(false),
                     processorFactories,
                     scriptService
                 );
@@ -1416,7 +1416,7 @@ public <P extends Processor> Collection<String> getPipelineWithProcessorType(Cla
 
     public synchronized void reloadPipeline(String id) throws Exception {
         PipelineHolder holder = pipelines.get(id);
-        Pipeline updatedPipeline = Pipeline.create(id, holder.configuration.getConfigAsMap(), processorFactories, scriptService);
+        Pipeline updatedPipeline = Pipeline.create(id, holder.configuration.getConfig(false), processorFactories, scriptService);
         Map<String, PipelineHolder> updatedPipelines = new HashMap<>(this.pipelines);
         updatedPipelines.put(id, new PipelineHolder(holder.configuration, updatedPipeline));
         this.pipelines = Map.copyOf(updatedPipelines);

server/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java

@@ -9,39 +9,47 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.xcontent.ContextParser;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xcontent.json.JsonXContent;
 
 import java.io.IOException;
-import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
 /**
- * Encapsulates a pipeline's id and configuration as a blob
+ * Encapsulates a pipeline's id and configuration as a loosely typed map -- see {@link Pipeline} for the
+ * parsed and processed object(s) that a pipeline configuration will become. This class is used for things
+ * like keeping track of pipelines in the cluster state (where a pipeline is 'just some json') whereas the
+ * {@link Pipeline} class is used in the actual processing of ingest documents through pipelines in the
+ * {@link IngestService}.
  */
 public final class PipelineConfiguration implements SimpleDiffable<PipelineConfiguration>, ToXContentObject {
 
     private static final ObjectParser<Builder, Void> PARSER = new ObjectParser<>("pipeline_config", true, Builder::new);
     static {
         PARSER.declareString(Builder::setId, new ParseField("id"));
-        PARSER.declareField((parser, builder, aVoid) -> {
-            XContentBuilder contentBuilder = XContentBuilder.builder(parser.contentType().xContent());
-            contentBuilder.generator().copyCurrentStructure(parser);
-            builder.setConfig(BytesReference.bytes(contentBuilder), contentBuilder.contentType());
-        }, new ParseField("config"), ObjectParser.ValueType.OBJECT);
-
+        PARSER.declareField(
+            (parser, builder, aVoid) -> builder.setConfig(parser.map()),
+            new ParseField("config"),
+            ObjectParser.ValueType.OBJECT
+        );
     }
 
     public static ContextParser<Void, PipelineConfiguration> getParser() {
@@ -51,56 +59,94 @@ public static ContextParser<Void, PipelineConfiguration> getParser() {
     private static class Builder {
 
         private String id;
-        private BytesReference config;
-        private XContentType xContentType;
+        private Map<String, Object> config;
 
         void setId(String id) {
             this.id = id;
         }
 
-        void setConfig(BytesReference config, XContentType xContentType) {
+        void setConfig(Map<String, Object> config) {
             this.config = config;
-            this.xContentType = xContentType;
         }
 
         PipelineConfiguration build() {
-            return new PipelineConfiguration(id, config, xContentType);
+            return new PipelineConfiguration(id, config);
         }
     }
 
     private final String id;
-    // Store config as bytes reference, because the config is only used when the pipeline store reads the cluster state
-    // and the way the map of maps config is read requires a deep copy (it removes instead of gets entries to check for unused options)
-    // also the get pipeline api just directly returns this to the caller
-    private final BytesReference config;
-    private final XContentType xContentType;
+    private final Map<String, Object> config;
 
-    public PipelineConfiguration(String id, BytesReference config, XContentType xContentType) {
+    public PipelineConfiguration(String id, Map<String, Object> config) {
         this.id = Objects.requireNonNull(id);
-        this.config = Objects.requireNonNull(config);
-        this.xContentType = Objects.requireNonNull(xContentType);
+        this.config = deepCopy(config, true); // defensive deep copy
+    }
+
+    /**
+     * A convenience constructor that parses some bytes as a map representing a pipeline's config and then delegates to the
+     * conventional {@link #PipelineConfiguration(String, Map)} constructor.
+     *
+     * @param id the id of the pipeline
+     * @param config a parse-able bytes reference that will return a pipeline configuration
+     * @param xContentType the content-type to use while parsing the pipeline configuration
+     */
+    public PipelineConfiguration(String id, BytesReference config, XContentType xContentType) {
+        this(id, XContentHelper.convertToMap(config, true, xContentType).v2());
     }
 
     public String getId() {
         return id;
     }
 
-    public Map<String, Object> getConfigAsMap() {
-        return XContentHelper.convertToMap(config, true, xContentType).v2();
+    /**
+     * @return a reference to the unmodifiable configuration map for this pipeline
+     */
+    public Map<String, Object> getConfig() {
+        return getConfig(true);
     }
 
-    // pkg-private for tests
-    XContentType getXContentType() {
-        return xContentType;
+    /**
+     * @param unmodifiable whether the returned map should be unmodifiable or not
+     * @return a reference to the unmodifiable config map (if unmodifiable is true) or
+     * a reference to a freshly-created mutable deep copy of the config map (if unmodifiable is false)
+     */
+    public Map<String, Object> getConfig(boolean unmodifiable) {
+        if (unmodifiable) {
+            return config; // already unmodifiable
+        } else {
+            return deepCopy(config, false);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T deepCopy(final T value, final boolean unmodifiable) {
+        return (T) innerDeepCopy(value, unmodifiable);
     }
 
-    // pkg-private for tests
-    BytesReference getConfig() {
-        return config;
+    private static Object innerDeepCopy(final Object value, final boolean unmodifiable) {
+        if (value instanceof Map<?, ?> mapValue) {
+            final Map<Object, Object> copy = Maps.newLinkedHashMapWithExpectedSize(mapValue.size()); // n.b. maintain ordering
+            for (Map.Entry<?, ?> entry : mapValue.entrySet()) {
+                copy.put(innerDeepCopy(entry.getKey(), unmodifiable), innerDeepCopy(entry.getValue(), unmodifiable));
+            }
+            return unmodifiable ? Collections.unmodifiableMap(copy) : copy;
+        } else if (value instanceof List<?> listValue) {
+            final List<Object> copy = new ArrayList<>(listValue.size());
+            for (Object itemValue : listValue) {
+                copy.add(innerDeepCopy(itemValue, unmodifiable));
+            }
+            return unmodifiable ? Collections.unmodifiableList(copy) : copy;
+        } else {
+            // if this list of expected value types ends up not being exhaustive, then we want to learn about that
+            // at development time, but it's probably better to err on the side of passing through the value at runtime
+            assert (value == null || value instanceof String || value instanceof Number || value instanceof Boolean)
+                : "unexpected value type [" + value.getClass() + "]";
+            return value;
+        }
     }
 
     public Integer getVersion() {
-        Object o = getConfigAsMap().get("version");
+        Object o = config.get("version");
         if (o == null) {
             return null;
         } else if (o instanceof Number number) {
@@ -114,13 +160,22 @@ public Integer getVersion() {
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         builder.field("id", id);
-        builder.field("config", getConfigAsMap());
+        builder.field("config", config);
         builder.endObject();
         return builder;
     }
 
     public static PipelineConfiguration readFrom(StreamInput in) throws IOException {
-        return new PipelineConfiguration(in.readString(), in.readBytesReference(), in.readEnum(XContentType.class));
+        final String id = in.readString();
+        final Map<String, Object> config;
+        if (in.getTransportVersion().onOrAfter(TransportVersions.INGEST_PIPELINE_CONFIGURATION_AS_MAP)) {
+            config = in.readGenericMap();
+        } else {
+            final BytesReference bytes = in.readSlicedBytesReference();
+            final XContentType type = in.readEnum(XContentType.class);
+            config = XContentHelper.convertToMap(bytes, true, type).v2();
+        }
+        return new PipelineConfiguration(id, config);
     }
 
     public static Diff<PipelineConfiguration> readDiffFrom(StreamInput in) throws IOException {
@@ -135,8 +190,14 @@ public String toString() {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(id);
-        out.writeBytesReference(config);
-        XContentHelper.writeTo(out, xContentType);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.INGEST_PIPELINE_CONFIGURATION_AS_MAP)) {
+            out.writeGenericMap(config);
+        } else {
+            XContentBuilder builder = XContentBuilder.builder(JsonXContent.jsonXContent).prettyPrint();
+            builder.map(config);
+            out.writeBytesReference(BytesReference.bytes(builder));
+            XContentHelper.writeTo(out, XContentType.JSON);
+        }
     }
 
     @Override
@@ -147,14 +208,14 @@ public boolean equals(Object o) {
         PipelineConfiguration that = (PipelineConfiguration) o;
 
         if (id.equals(that.id) == false) return false;
-        return getConfigAsMap().equals(that.getConfigAsMap());
+        return config.equals(that.config);
 
     }
 
     @Override
     public int hashCode() {
         int result = id.hashCode();
-        result = 31 * result + getConfigAsMap().hashCode();
+        result = 31 * result + config.hashCode();
         return result;
     }
 
@@ -164,7 +225,7 @@ public int hashCode() {
      * <p>The given upgrader is applied to the config map for any processor of the given type.
      */
     PipelineConfiguration maybeUpgradeProcessors(String type, IngestMetadata.ProcessorConfigUpgrader upgrader) {
-        Map<String, Object> mutableConfigMap = getConfigAsMap();
+        Map<String, Object> mutableConfigMap = getConfig(false);
         boolean changed = false;
         // This should be a List of Maps, where the keys are processor types and the values are config maps.
         // But we'll skip upgrading rather than fail if not.
@@ -180,11 +241,7 @@ PipelineConfiguration maybeUpgradeProcessors(String type, IngestMetadata.Process
             }
         }
         if (changed) {
-            try (XContentBuilder builder = XContentBuilder.builder(xContentType.xContent())) {
-                return new PipelineConfiguration(id, BytesReference.bytes(builder.map(mutableConfigMap)), xContentType);
-            } catch (IOException e) {
-                throw new UncheckedIOException(e);
-            }
+            return new PipelineConfiguration(id, mutableConfigMap);
         } else {
             return this;
         }

server/src/test/java/org/elasticsearch/action/ingest/GetPipelineResponseTests.java

@@ -79,7 +79,7 @@ public void testXContentDeserialization() throws IOException {
         assertEquals(actualPipelines.size(), parsedPipelines.size());
         for (PipelineConfiguration pipeline : parsedPipelines) {
             assertTrue(pipelinesMap.containsKey(pipeline.getId()));
-            assertEquals(pipelinesMap.get(pipeline.getId()).getConfigAsMap(), pipeline.getConfigAsMap());
+            assertEquals(pipelinesMap.get(pipeline.getId()).getConfig(), pipeline.getConfig());
         }
     }
 

server/src/test/java/org/elasticsearch/ingest/IngestMetadataTests.java

@@ -56,8 +56,8 @@ public void testFromXContent() throws IOException {
             assertEquals(2, custom.getPipelines().size());
             assertEquals("1", custom.getPipelines().get("1").getId());
             assertEquals("2", custom.getPipelines().get("2").getId());
-            assertEquals(pipeline.getConfigAsMap(), custom.getPipelines().get("1").getConfigAsMap());
-            assertEquals(pipeline2.getConfigAsMap(), custom.getPipelines().get("2").getConfigAsMap());
+            assertEquals(pipeline.getConfig(), custom.getPipelines().get("1").getConfig());
+            assertEquals(pipeline2.getConfig(), custom.getPipelines().get("2").getConfig());
         }
     }