Remove ecs option on user_agent processor (#116077)

This removes the `ecs` option on the `user_agent` ingest processor, which is deprecated (way back in 6.7) and ignored. It will no longer be possible to create instances with this option, and the option will be removed from instances persisted in the cluster state on startup.

The mechanism to do this upgrade on startup is designed to be reusable for other upgrades either to ingest processors or more generally to any custom metadata. It is inspired by the existing mechanism to upgrade index templates.

Author: PeteGillinElastic

docs/changelog/116077.yaml

@@ -0,0 +1,14 @@
+pr: 116077
+summary: Remove `ecs` option on `user_agent` processor
+area: Ingest Node
+type: breaking
+issues: []
+breaking:
+  title: Remove `ecs` option on `user_agent` processor
+  area: Ingest
+  details: >-
+    The `user_agent` ingest processor no longer accepts the `ecs` option. (It was previously deprecated and ignored.)
+  impact: >-
+    Users should stop using the `ecs` option when creating instances of the `user_agent` ingest processor.
+    The option will be removed from existing processors stored in the cluster state on upgrade.
+  notable: false

modules/ingest-user-agent/src/main/java/module-info.java

@@ -10,4 +10,5 @@
 module org.elasticsearch.ingest.useragent {
     requires org.elasticsearch.server;
     requires org.elasticsearch.xcontent;
+    requires org.elasticsearch.base;
 }

modules/ingest-user-agent/src/main/java/org/elasticsearch/ingest/useragent/IngestUserAgentPlugin.java

@@ -9,7 +9,9 @@
 
 package org.elasticsearch.ingest.useragent;
 
+import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.ingest.IngestMetadata;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.plugins.IngestPlugin;
 import org.elasticsearch.plugins.Plugin;
@@ -23,6 +25,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.UnaryOperator;
 import java.util.stream.Stream;
 
 public class IngestUserAgentPlugin extends Plugin implements IngestPlugin {
@@ -97,4 +100,15 @@ static Map<String, UserAgentParser> createUserAgentParsers(Path userAgentConfigD
     public List<Setting<?>> getSettings() {
         return List.of(CACHE_SIZE_SETTING);
     }
+
+    @Override
+    public Map<String, UnaryOperator<Metadata.Custom>> getCustomMetadataUpgraders() {
+        return Map.of(
+            IngestMetadata.TYPE,
+            ingestMetadata -> ((IngestMetadata) ingestMetadata).maybeUpgradeProcessors(
+                UserAgentProcessor.TYPE,
+                UserAgentProcessor::maybeUpgradeConfig
+            )
+        );
+    }
 }

modules/ingest-user-agent/src/main/java/org/elasticsearch/ingest/useragent/UserAgentProcessor.java

@@ -9,9 +9,8 @@
 
 package org.elasticsearch.ingest.useragent;
 
-import org.elasticsearch.common.logging.DeprecationCategory;
-import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.core.UpdateForV10;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
@@ -32,8 +31,6 @@
 
 public class UserAgentProcessor extends AbstractProcessor {
 
-    private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(UserAgentProcessor.class);
-
     public static final String TYPE = "user_agent";
 
     private final String field;
@@ -198,21 +195,13 @@ public UserAgentProcessor create(
             String processorTag,
             String description,
             Map<String, Object> config
-        ) throws Exception {
+        ) {
             String field = readStringProperty(TYPE, processorTag, config, "field");
             String targetField = readStringProperty(TYPE, processorTag, config, "target_field", "user_agent");
             String regexFilename = readStringProperty(TYPE, processorTag, config, "regex_file", IngestUserAgentPlugin.DEFAULT_PARSER_NAME);
             List<String> propertyNames = readOptionalList(TYPE, processorTag, config, "properties");
             boolean extractDeviceType = readBooleanProperty(TYPE, processorTag, config, "extract_device_type", false);
             boolean ignoreMissing = readBooleanProperty(TYPE, processorTag, config, "ignore_missing", false);
-            Object ecsValue = config.remove("ecs");
-            if (ecsValue != null) {
-                deprecationLogger.warn(
-                    DeprecationCategory.SETTINGS,
-                    "ingest_useragent_ecs_settings",
-                    "setting [ecs] is deprecated as ECS format is the default and only option"
-                );
-            }
 
             UserAgentParser parser = userAgentParsers.get(regexFilename);
             if (parser == null) {
@@ -272,4 +261,14 @@ public static Property parseProperty(String propertyName) {
             }
         }
     }
+
+    @UpdateForV10(owner = UpdateForV10.Owner.DATA_MANAGEMENT)
+    // This can be removed in V10. It's not possible to create an instance with the ecs property in V9, and all instances created by V8 or
+    // earlier will have been fixed when upgraded to V9.
+    static boolean maybeUpgradeConfig(Map<String, Object> config) {
+        // Instances created using ES 8.x (or earlier) may have the 'ecs' config entry.
+        // This was ignored in 8.x and is unsupported in 9.0.
+        // In 9.x, we should remove it from any existing processors on startup.
+        return config.remove("ecs") != null;
+    }
 }

modules/ingest-user-agent/src/test/java/org/elasticsearch/ingest/useragent/UserAgentProcessorTests.java

@@ -331,4 +331,18 @@ public void testExtractDeviceTypeDisabled() {
         device.put("name", "Other");
         assertThat(target.get("device"), is(device));
     }
+
+    public void testMaybeUpgradeConfig_removesEcsIfPresent() {
+        Map<String, Object> config = new HashMap<>(Map.of("field", "user-agent", "ecs", "whatever"));
+        boolean changed = UserAgentProcessor.maybeUpgradeConfig(config);
+        assertThat(changed, is(true));
+        assertThat(config, is(Map.of("field", "user-agent")));
+    }
+
+    public void testMaybeUpgradeConfig_doesNothingIfEcsAbsent() {
+        Map<String, Object> config = new HashMap<>(Map.of("field", "user-agent"));
+        boolean changed = UserAgentProcessor.maybeUpgradeConfig(config);
+        assertThat(changed, is(false));
+        assertThat(config, is(Map.of("field", "user-agent")));
+    }
 }

server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java

@@ -300,18 +300,31 @@ static Metadata upgradeMetadata(Metadata metadata, IndexMetadataVerifier indexMe
             upgradedMetadata.put(newMetadata, false);
         }
         // upgrade current templates
-        if (applyPluginUpgraders(
+        if (applyPluginTemplateUpgraders(
             metadata.getTemplates(),
             metadataUpgrader.indexTemplateMetadataUpgraders,
             upgradedMetadata::removeTemplate,
             (s, indexTemplateMetadata) -> upgradedMetadata.put(indexTemplateMetadata)
         )) {
             changed = true;
         }
+        // upgrade custom metadata
+        for (Map.Entry<String, UnaryOperator<Metadata.Custom>> entry : metadataUpgrader.customMetadataUpgraders.entrySet()) {
+            String type = entry.getKey();
+            Function<Metadata.Custom, Metadata.Custom> upgrader = entry.getValue();
+            Metadata.Custom original = metadata.custom(type);
+            if (original != null) {
+                Metadata.Custom upgraded = upgrader.apply(original);
+                if (upgraded.equals(original) == false) {
+                    upgradedMetadata.putCustom(type, upgraded);
+                    changed = true;
+                }
+            }
+        }
         return changed ? upgradedMetadata.build() : metadata;
     }
 
-    private static boolean applyPluginUpgraders(
+    private static boolean applyPluginTemplateUpgraders(
         Map<String, IndexTemplateMetadata> existingData,
         UnaryOperator<Map<String, IndexTemplateMetadata>> upgrader,
         Consumer<String> removeData,

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

@@ -169,4 +169,39 @@ public boolean equals(Object o) {
     public int hashCode() {
         return pipelines.hashCode();
     }
+
+    /**
+     * Returns a copy of this object with processor upgrades applied, if necessary. Otherwise, returns this object.
+     *
+     * <p>The given upgrader is applied to the config map for any processor of the given type.
+     */
+    public IngestMetadata maybeUpgradeProcessors(String processorType, ProcessorConfigUpgrader processorConfigUpgrader) {
+        Map<String, PipelineConfiguration> newPipelines = null; // as an optimization, we will lazily copy the map only if needed
+        for (Map.Entry<String, PipelineConfiguration> entry : pipelines.entrySet()) {
+            String pipelineId = entry.getKey();
+            PipelineConfiguration originalPipeline = entry.getValue();
+            PipelineConfiguration upgradedPipeline = originalPipeline.maybeUpgradeProcessors(processorType, processorConfigUpgrader);
+            if (upgradedPipeline.equals(originalPipeline) == false) {
+                if (newPipelines == null) {
+                    newPipelines = new HashMap<>(pipelines);
+                }
+                newPipelines.put(pipelineId, upgradedPipeline);
+            }
+        }
+        return newPipelines != null ? new IngestMetadata(newPipelines) : this;
+    }
+
+    /**
+     * Functional interface for upgrading processor configs. An implementation of this will be associated with a specific processor type.
+     */
+    public interface ProcessorConfigUpgrader {
+
+        /**
+         * Upgrades the config for an individual processor of the appropriate type, if necessary.
+         *
+         * @param processorConfig The config to upgrade, which will be mutated if required
+         * @return Whether an upgrade was required
+         */
+        boolean maybeUpgrade(Map<String, Object> processorConfig);
+    }
 }

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.Map;
 import java.util.Objects;
 
@@ -156,4 +157,36 @@ public int hashCode() {
         result = 31 * result + getConfigAsMap().hashCode();
         return result;
     }
+
+    /**
+     * Returns a copy of this object with processor upgrades applied, if necessary. Otherwise, returns this object.
+     *
+     * <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();
+        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.
+        if (mutableConfigMap.get(Pipeline.PROCESSORS_KEY) instanceof Iterable<?> processors) {
+            for (Object processor : processors) {
+                if (processor instanceof Map<?, ?> processorMap && processorMap.get(type) instanceof Map<?, ?> targetProcessor) {
+                    @SuppressWarnings("unchecked") // All XContent maps will be <String, Object>
+                    Map<String, Object> processorConfigMap = (Map<String, Object>) targetProcessor;
+                    if (upgrader.maybeUpgrade(processorConfigMap)) {
+                        changed = true;
+                    }
+                }
+            }
+        }
+        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);
+            }
+        } else {
+            return this;
+        }
+    }
 }

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -47,6 +47,7 @@
 import org.elasticsearch.cluster.metadata.DataStreamGlobalRetentionSettings;
 import org.elasticsearch.cluster.metadata.IndexMetadataVerifier;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.MetadataCreateDataStreamService;
 import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
 import org.elasticsearch.cluster.metadata.MetadataDataStreamsService;
@@ -232,6 +233,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -970,7 +972,9 @@ private void construct(
         );
 
         var indexTemplateMetadataUpgraders = pluginsService.map(Plugin::getIndexTemplateMetadataUpgrader).toList();
-        modules.bindToInstance(MetadataUpgrader.class, new MetadataUpgrader(indexTemplateMetadataUpgraders));
+        List<Map<String, UnaryOperator<Metadata.Custom>>> customMetadataUpgraders = pluginsService.map(Plugin::getCustomMetadataUpgraders)
+            .toList();
+        modules.bindToInstance(MetadataUpgrader.class, new MetadataUpgrader(indexTemplateMetadataUpgraders, customMetadataUpgraders));
 
         final IndexMetadataVerifier indexMetadataVerifier = new IndexMetadataVerifier(
             settings,
@@ -1463,6 +1467,7 @@ private CircuitBreakerService createCircuitBreakerService(
 
     /**
      * Wrap a group of reloadable plugins into a single reloadable plugin interface
+     *
      * @param reloadablePlugins A list of reloadable plugins
      * @return A single ReloadablePlugin that, upon reload, reloads the plugins it wraps
      */

server/src/main/java/org/elasticsearch/plugins/MetadataUpgrader.java

@@ -14,22 +14,56 @@
 
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.function.UnaryOperator;
 
+import static java.util.stream.Collectors.collectingAndThen;
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.mapping;
+import static java.util.stream.Collectors.toList;
+
 /**
  * Upgrades {@link Metadata} on startup on behalf of installed {@link Plugin}s
  */
 public class MetadataUpgrader {
     public final UnaryOperator<Map<String, IndexTemplateMetadata>> indexTemplateMetadataUpgraders;
+    public final Map<String, UnaryOperator<Metadata.Custom>> customMetadataUpgraders;
 
-    public MetadataUpgrader(Collection<UnaryOperator<Map<String, IndexTemplateMetadata>>> indexTemplateMetadataUpgraders) {
+    public MetadataUpgrader(
+        Collection<UnaryOperator<Map<String, IndexTemplateMetadata>>> indexTemplateMetadataUpgraders,
+        Collection<Map<String, UnaryOperator<Metadata.Custom>>> customMetadataUpgraders
+    ) {
         this.indexTemplateMetadataUpgraders = templates -> {
             Map<String, IndexTemplateMetadata> upgradedTemplates = new HashMap<>(templates);
             for (UnaryOperator<Map<String, IndexTemplateMetadata>> upgrader : indexTemplateMetadataUpgraders) {
                 upgradedTemplates = upgrader.apply(upgradedTemplates);
             }
             return upgradedTemplates;
         };
+        this.customMetadataUpgraders = customMetadataUpgraders.stream()
+            // Flatten the stream of maps into a stream of entries
+            .flatMap(map -> map.entrySet().stream())
+            .collect(
+                groupingBy(
+                    // Group by the type of custom metadata to be upgraded (the entry key)
+                    Map.Entry::getKey,
+                    // For each type, extract the operators (the entry values), collect to a list, and make an operator which combines them
+                    collectingAndThen(mapping(Map.Entry::getValue, toList()), CombiningCustomUpgrader::new)
+                )
+            );
+    }
+
+    private record CombiningCustomUpgrader(List<UnaryOperator<Metadata.Custom>> upgraders) implements UnaryOperator<Metadata.Custom> {
+
+        @Override
+        public Metadata.Custom apply(Metadata.Custom custom) {
+            Metadata.Custom upgraded = custom;
+            for (UnaryOperator<Metadata.Custom> upgrader : upgraders) {
+                upgraded = upgrader.apply(upgraded);
+            }
+            return upgraded;
+        }
     }
+
 }