Dep. check for ECS changes to User Agent processor (elastic#38362)

The User Agent ingest processor is changing to align with ECS. Users
need to be made aware that any pipelines using this processor will
behave differently in 7.0.

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

@@ -335,7 +335,7 @@ public String getType() {
         return new Pipeline(id, description, null, new CompoundProcessor(failureProcessor));
     }
 
-    static ClusterState innerPut(PutPipelineRequest request, ClusterState currentState) {
+    public static ClusterState innerPut(PutPipelineRequest request, ClusterState currentState) {
         IngestMetadata currentIngestMetadata = currentState.metaData().custom(IngestMetadata.TYPE);
         Map<String, PipelineConfiguration> pipelines;
         if (currentIngestMetadata != null) {

x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/ClusterDeprecationChecks.java

@@ -8,8 +8,16 @@
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.ingest.ConfigurationUtils;
+import org.elasticsearch.ingest.IngestService;
+import org.elasticsearch.ingest.PipelineConfiguration;
 import org.elasticsearch.xpack.core.deprecation.DeprecationIssue;
 
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
 public class ClusterDeprecationChecks {
 
     static DeprecationIssue checkShardLimit(ClusterState state) {
@@ -40,4 +48,36 @@ static DeprecationIssue checkClusterName(ClusterState state) {
         }
         return null;
     }
+
+    @SuppressWarnings("unchecked")
+    static DeprecationIssue checkUserAgentPipelines(ClusterState state) {
+        List<PipelineConfiguration> pipelines = IngestService.getPipelines(state);
+
+        List<String> pipelinesWithDeprecatedEcsConfig = pipelines.stream()
+            .filter(Objects::nonNull)
+            .filter(pipeline -> {
+                Map<String, Object> pipelineConfig = pipeline.getConfigAsMap();
+
+                List<Map<String, Map<String, Object>>> processors =
+                    (List<Map<String, Map<String, Object>>>) pipelineConfig.get("processors");
+                return processors.stream()
+                    .filter(Objects::nonNull)
+                    .filter(processor -> processor.containsKey("user_agent"))
+                    .map(processor -> processor.get("user_agent"))
+                    .anyMatch(processorConfig ->
+                        false == ConfigurationUtils.readBooleanProperty(null, null, processorConfig, "ecs", false));
+            })
+            .map(PipelineConfiguration::getId)
+            .sorted() // Make the warning consistent for testing purposes
+            .collect(Collectors.toList());
+        if (pipelinesWithDeprecatedEcsConfig.isEmpty() == false) {
+            return new DeprecationIssue(DeprecationIssue.Level.WARNING,
+                "User-Agent ingest plugin will use ECS-formatted output",
+                "https://www.elastic.co/guide/en/elasticsearch/reference/master/breaking-changes-7.0.html" +
+                    "#ingest-user-agent-ecs-always",
+                "Ingest pipelines " + pipelinesWithDeprecatedEcsConfig + " will change to using ECS output format in 7.0");
+        }
+        return null;
+
+    }
 }

x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/DeprecationChecks.java

@@ -33,6 +33,7 @@ private DeprecationChecks() {
 
     static List<Function<ClusterState, DeprecationIssue>> CLUSTER_SETTINGS_CHECKS =
         Collections.unmodifiableList(Arrays.asList(
+            ClusterDeprecationChecks::checkUserAgentPipelines,
             ClusterDeprecationChecks::checkShardLimit,
             ClusterDeprecationChecks::checkClusterName
         ));

x-pack/plugin/deprecation/src/test/java/org/elasticsearch/xpack/deprecation/ClusterDeprecationChecksTests.java

@@ -6,14 +6,18 @@
 package org.elasticsearch.xpack.deprecation;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.deprecation.DeprecationIssue;
 
@@ -83,4 +87,57 @@ public void testCheckShardLimit() {
         issues = DeprecationChecks.filterChecks(CLUSTER_SETTINGS_CHECKS, c -> c.apply(goodState));
         assertTrue(issues.isEmpty());
     }
+
+    public void testUserAgentEcsCheck() {
+        PutPipelineRequest ecsFalseRequest = new PutPipelineRequest("ecs_false",
+            new BytesArray("{\n" +
+                "  \"description\" : \"This has ecs set to false\",\n" +
+                "  \"processors\" : [\n" +
+                "    {\n" +
+                "      \"user_agent\" : {\n" +
+                "        \"field\" : \"agent\",\n" +
+                "        \"ecs\" : false\n" +
+                "      }\n" +
+                "    }\n" +
+                "  ]\n" +
+                "}"), XContentType.JSON);
+        PutPipelineRequest ecsNullRequest = new PutPipelineRequest("ecs_null",
+            new BytesArray("{\n" +
+                "  \"description\" : \"This has ecs set to false\",\n" +
+                "  \"processors\" : [\n" +
+                "    {\n" +
+                "      \"user_agent\" : {\n" +
+                "        \"field\" : \"agent\"\n" +
+                "      }\n" +
+                "    }\n" +
+                "  ]\n" +
+                "}"), XContentType.JSON);
+        PutPipelineRequest ecsTrueRequest = new PutPipelineRequest("ecs_true",
+            new BytesArray("{\n" +
+                "  \"description\" : \"This has ecs set to false\",\n" +
+                "  \"processors\" : [\n" +
+                "    {\n" +
+                "      \"user_agent\" : {\n" +
+                "        \"field\" : \"agent\",\n" +
+                "        \"ecs\" : true\n" +
+                "      }\n" +
+                "    }\n" +
+                "  ]\n" +
+                "}"), XContentType.JSON);
+
+        ClusterState state = ClusterState.builder(new ClusterName("test")).build();
+        state = IngestService.innerPut(ecsTrueRequest, state);
+        state = IngestService.innerPut(ecsFalseRequest, state);
+        state = IngestService.innerPut(ecsNullRequest, state);
+
+        final ClusterState finalState = state;
+        List<DeprecationIssue> issues = DeprecationChecks.filterChecks(CLUSTER_SETTINGS_CHECKS, c -> c.apply(finalState));
+
+        DeprecationIssue expected = new DeprecationIssue(DeprecationIssue.Level.WARNING,
+            "User-Agent ingest plugin will use ECS-formatted output",
+            "https://www.elastic.co/guide/en/elasticsearch/reference/master/breaking-changes-7.0.html" +
+                "#ingest-user-agent-ecs-always",
+            "Ingest pipelines [ecs_false, ecs_null] will change to using ECS output format in 7.0");
+        assertEquals(singletonList(expected), issues);
+    }
 }