Move setting of destination.ip to ingest pipeline

apmpackage/apm/data_stream/app_metrics/elasticsearch/ingest_pipeline/apm_convert_destination_address.json

@@ -0,0 +1,14 @@
+{
+  "description": "Convert destination.address to an IP, storing in destination.ip",
+  "processors": [
+    {
+      "convert": {
+        "field": "destination.address",
+        "target_field": "destination.ip",
+        "type": "ip",
+        "ignore_missing": true,
+        "ignore_failure": true
+      }
+    }
+  ]
+}

apmpackage/apm/data_stream/app_metrics/elasticsearch/ingest_pipeline/default.json

@@ -21,6 +21,11 @@
         "name": "metrics-apm.app-0.4.0-apm_remove_span_metadata"
       }
     },
+    {
+      "pipeline": {
+        "name": "metrics-apm.app-0.4.0-apm_convert_destination_address"
+      }
+    },
     {
       "pipeline": {
         "name": "metrics-apm.app-0.4.0-apm_error_grouping_name",

apmpackage/apm/data_stream/error_logs/elasticsearch/ingest_pipeline/apm_convert_destination_address.json

@@ -0,0 +1,14 @@
+{
+  "description": "Convert destination.address to an IP, storing in destination.ip",
+  "processors": [
+    {
+      "convert": {
+        "field": "destination.address",
+        "target_field": "destination.ip",
+        "type": "ip",
+        "ignore_missing": true,
+        "ignore_failure": true
+      }
+    }
+  ]
+}

apmpackage/apm/data_stream/error_logs/elasticsearch/ingest_pipeline/default.json

@@ -21,6 +21,11 @@
         "name": "logs-apm.error-0.4.0-apm_remove_span_metadata"
       }
     },
+    {
+      "pipeline": {
+        "name": "logs-apm.error-0.4.0-apm_convert_destination_address"
+      }
+    },
     {
       "pipeline": {
         "name": "logs-apm.error-0.4.0-apm_error_grouping_name",

apmpackage/apm/data_stream/internal_metrics/elasticsearch/ingest_pipeline/apm_convert_destination_address.json

@@ -0,0 +1,14 @@
+{
+  "description": "Convert destination.address to an IP, storing in destination.ip",
+  "processors": [
+    {
+      "convert": {
+        "field": "destination.address",
+        "target_field": "destination.ip",
+        "type": "ip",
+        "ignore_missing": true,
+        "ignore_failure": true
+      }
+    }
+  ]
+}

apmpackage/apm/data_stream/internal_metrics/elasticsearch/ingest_pipeline/default.json

@@ -21,6 +21,11 @@
         "name": "metrics-apm.internal-0.4.0-apm_remove_span_metadata"
       }
     },
+    {
+      "pipeline": {
+        "name": "metrics-apm.internal-0.4.0-apm_convert_destination_address"
+      }
+    },
     {
       "pipeline": {
         "name": "metrics-apm.internal-0.4.0-apm_error_grouping_name",

apmpackage/apm/data_stream/profile_metrics/elasticsearch/ingest_pipeline/apm_convert_destination_address.json

@@ -0,0 +1,14 @@
+{
+  "description": "Convert destination.address to an IP, storing in destination.ip",
+  "processors": [
+    {
+      "convert": {
+        "field": "destination.address",
+        "target_field": "destination.ip",
+        "type": "ip",
+        "ignore_missing": true,
+        "ignore_failure": true
+      }
+    }
+  ]
+}

apmpackage/apm/data_stream/profile_metrics/elasticsearch/ingest_pipeline/default.json

@@ -21,6 +21,11 @@
         "name": "metrics-apm.profiling-0.4.0-apm_remove_span_metadata"
       }
     },
+    {
+      "pipeline": {
+        "name": "metrics-apm.profiling-0.4.0-apm_convert_destination_address"
+      }
+    },
     {
       "pipeline": {
         "name": "metrics-apm.profiling-0.4.0-apm_error_grouping_name",

apmpackage/apm/data_stream/traces/elasticsearch/ingest_pipeline/apm_convert_destination_address.json

@@ -0,0 +1,14 @@
+{
+  "description": "Convert destination.address to an IP, storing in destination.ip",
+  "processors": [
+    {
+      "convert": {
+        "field": "destination.address",
+        "target_field": "destination.ip",
+        "type": "ip",
+        "ignore_missing": true,
+        "ignore_failure": true
+      }
+    }
+  ]
+}

apmpackage/apm/data_stream/traces/elasticsearch/ingest_pipeline/default.json

@@ -21,6 +21,11 @@
         "name": "traces-apm-0.4.0-apm_remove_span_metadata"
       }
     },
+    {
+      "pipeline": {
+        "name": "traces-apm-0.4.0-apm_convert_destination_address"
+      }
+    },
     {
       "pipeline": {
         "name": "traces-apm-0.4.0-apm_error_grouping_name",

beater/test_approved_es_documents/TestPublishIntegrationSpans.approved.json

@@ -567,7 +567,6 @@
             },
             "destination": {
                 "address": "0:0::0:1",
-                "ip": "0:0::0:1",
                 "port": 5432
             },
             "ecs": {
@@ -787,8 +786,7 @@
                 "id": "container-id"
             },
             "destination": {
-                "address": "0:0::0:1",
-                "ip": "0:0::0:1"
+                "address": "0:0::0:1"
             },
             "ecs": {
                 "version": "1.8.0"

ingest/pipeline/definition.json

@@ -24,6 +24,11 @@
             "name": "apm_remove_span_metadata"
           }
         },
+        {
+          "pipeline": {
+            "name": "apm_convert_destination_address"
+          }
+        },
         {
           "pipeline": {
             "name": "apm_error_grouping_name",
@@ -95,6 +100,23 @@
       ]
     }
   },
+  {
+    "id": "apm_convert_destination_address",
+    "body": {
+      "description": "Convert destination.address to an IP, storing in destination.ip",
+      "processors": [
+        {
+          "convert": {
+            "field": "destination.address",
+            "target_field": "destination.ip",
+            "type": "ip",
+            "ignore_missing": true,
+            "ignore_failure": true
+          }
+        }
+      ]
+    }
+  },
   {
     "id": "apm_remove_span_metadata",
     "body": {

ingest/pipeline/definition.yml

@@ -12,6 +12,8 @@ apm:
       name: apm_user_geo
   - pipeline:
       name: apm_remove_span_metadata
+  - pipeline:
+      name: apm_convert_destination_address
   - pipeline:
       name: apm_error_grouping_name
       if: ctx.processor?.event == 'error'
@@ -50,6 +52,16 @@ apm_ingest_timestamp:
       field: event.ingested
       value: "{{_ingest.timestamp}}"
 
+apm_convert_destination_address:
+  description: Convert destination.address to an IP, storing in destination.ip
+  processors:
+  - convert:
+      field: destination.address
+      target_field: destination.ip
+      type: ip
+      ignore_missing: true
+      ignore_failure: true
+
 apm_remove_span_metadata:
   description: Removes metadata fields available already on the parent transaction, to save storage
   processors:

model/span.go

@@ -19,7 +19,6 @@ package model
 
 import (
 	"context"
-	"net"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -168,9 +167,6 @@ func (d *Destination) fields() common.MapStr {
 	var fields mapStr
 	if d.Address != "" {
 		fields.set("address", d.Address)
-		if ip := net.ParseIP(d.Address); ip != nil {
-			fields.set("ip", d.Address)
-		}
 	}
 	if d.Port > 0 {
 		fields.set("port", d.Port)

model/span_test.go

@@ -164,7 +164,7 @@ func TestSpanTransform(t *testing.T) {
 				"timestamp":   common.MapStr{"us": timestampUs},
 				"trace":       common.MapStr{"id": traceID},
 				"parent":      common.MapStr{"id": parentID},
-				"destination": common.MapStr{"address": address, "ip": address, "port": port},
+				"destination": common.MapStr{"address": address, "port": port},
 				"event":       common.MapStr{"outcome": "unknown"},
 				"http": common.MapStr{
 					"response":       common.MapStr{"status_code": statusCode},

processor/stream/test_approved_es_documents/testIntakeIntegrationSpans.approved.json

@@ -517,7 +517,6 @@
             "data_stream.type": "traces",
             "destination": {
                 "address": "0:0::0:1",
-                "ip": "0:0::0:1",
                 "port": 5432
             },
             "event": {
@@ -724,8 +723,7 @@
             "data_stream.dataset": "apm",
             "data_stream.type": "traces",
             "destination": {
-                "address": "0:0::0:1",
-                "ip": "0:0::0:1"
+                "address": "0:0::0:1"
             },
             "event": {
                 "outcome": "unknown"

systemtest/ingest_test.go

@@ -0,0 +1,55 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package systemtest_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/tidwall/gjson"
+
+	"github.com/elastic/apm-server/systemtest"
+	"github.com/elastic/apm-server/systemtest/apmservertest"
+	"github.com/elastic/apm-server/systemtest/estest"
+)
+
+func TestIngestPipeline(t *testing.T) {
+	systemtest.CleanupElasticsearch(t)
+	srv := apmservertest.NewServer(t)
+
+	tracer := srv.Tracer()
+	tx := tracer.StartTransaction("name", "type")
+	span := tx.StartSpan("name", "type", nil)
+	// If a destination address is recorded, and it is a valid
+	// IPv4 or IPv6 address, it will be copied to destination.ip.
+	span.Context.SetDestinationAddress("::1", 1234)
+	span.End()
+	tx.End()
+	tracer.Flush(nil)
+
+	result := systemtest.Elasticsearch.ExpectDocs(t, "apm-*", estest.TermQuery{
+		Field: "span.id",
+		Value: span.TraceContext().Span.String(),
+	})
+	require.Len(t, result.Hits.Hits, 1)
+
+	destinationIP := gjson.GetBytes(result.Hits.Hits[0].RawSource, "destination.ip")
+	assert.True(t, destinationIP.Exists())
+	assert.Equal(t, "::1", destinationIP.String())
+}