Adding geolocation attrs to app_logs

apmpackage/apm/changelog.yml

@@ -1,5 +1,8 @@
 - version: generated
   changes:
+    - description: Add geoip processing to add client.geo.* fields for app_logs
+      type: enhancement
+      link: https://github.com/elastic/apm-server/pull/11699
     - description: Define data retentions to support DLM
       type: enhancement
       link: https://github.com/elastic/apm-server/pull/11539

apmpackage/apm/data_stream/app_logs/elasticsearch/ingest_pipeline/default.yml

@@ -9,3 +9,5 @@ processors:
       name: observer_ids
   - pipeline:
       name: remove_ecs_version
+  - pipeline:
+      name: client_geoip

changelogs/head.asciidoc

@@ -35,3 +35,5 @@ https://github.com/elastic/apm-server/compare/8.10\...main[View commits]
 	- grpc.server.response.errors.count
 	- grpc.server.errors.timeout
 	- grpc.server.errors.ratelimit
+- Add geoip processing to app_logs ingest pipeline on `client.ip` {pull}11699[11699]
+

systemtest/approvals/TestOTLPGRPCLogsClientIP.approved.json

@@ -0,0 +1,67 @@
+{
+  "events": [
+    {
+      "@timestamp": "1970-01-01T00:00:01.000Z",
+      "agent": {
+        "name": "android/java",
+        "version": "unknown"
+      },
+      "client": {
+        "geo": {
+          "continent_name": "Europe",
+          "region_iso_code": "ES-M",
+          "city_name": "Madrid",
+          "country_iso_code": "ES",
+          "country_name": "Spain",
+          "region_name": "Madrid",
+          "location": {
+            "lon": -3.6934,
+            "lat": 40.4163
+          }
+        },
+        "ip": "195.55.79.118"
+      },
+      "data_stream": {
+        "dataset": "apm.app.unknown",
+        "namespace": "default",
+        "type": "logs"
+      },
+      "event": {
+        "severity": 9
+      },
+      "labels": {
+        "key": "value"
+      },
+      "log": {
+        "level": "Info"
+      },
+      "message": "a log message",
+      "numeric_labels": {
+        "numeric_key": 1234
+      },
+      "observer": {
+        "hostname": "dynamic",
+        "type": "apm-server",
+        "version": "dynamic"
+      },
+      "service": {
+        "language": {
+          "name": "java"
+        },
+        "name": "unknown"
+      },
+      "source": {
+        "nat": {
+          "ip": "127.0.0.1"
+        },
+        "ip": "195.55.79.118"
+      },
+      "span": {
+        "id": "0200000000000000"
+      },
+      "trace": {
+        "id": "01000000000000000000000000000000"
+      }
+    }
+  ]
+}

systemtest/otlp_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/tidwall/gjson"
+	"google.golang.org/grpc/metadata"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -453,6 +454,63 @@ func TestOTLPRateLimit(t *testing.T) {
 	assert.Equal(t, "traces export: rpc error: code = ResourceExhausted desc = rate limit exceeded", errStatus.Message())
 }
 
+func TestOTLPGRPCLogsClientIP(t *testing.T) {
+	systemtest.CleanupElasticsearch(t)
+	srv := apmservertest.NewUnstartedServerTB(t)
+	err := srv.Start()
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Override local IP address to be found in "GeoLite2-City.mmdb".
+	md := metadata.New(map[string]string{"X-Forwarded-For": "195.55.79.118"})
+	ctx = metadata.NewOutgoingContext(ctx, md)
+
+	conn, err := grpc.Dial(serverAddr(srv), grpc.WithInsecure(), grpc.WithBlock(), grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip")))
+	require.NoError(t, err)
+	defer conn.Close()
+
+	logsClient := plogotlp.NewGRPCClient(conn)
+
+	logs := newMobileLogs("a log message")
+	_, err = logsClient.Export(ctx, plogotlp.NewExportRequestFromLogs(logs))
+	require.NoError(t, err)
+
+	result := estest.ExpectDocs(t, systemtest.Elasticsearch, "logs-apm*", nil)
+	approvaltest.ApproveEvents(t, t.Name(), result.Hits.Hits)
+}
+
+func newMobileLogs(body interface{}) plog.Logs {
+	logs := plog.NewLogs()
+	resourceLogs := logs.ResourceLogs().AppendEmpty()
+	resourceAttrs := logs.ResourceLogs().At(0).Resource().Attributes()
+	resourceAttrs.PutStr(semconv.AttributeTelemetrySDKLanguage, "java")
+	resourceAttrs.PutStr(semconv.AttributeTelemetrySDKName, "android")
+
+	scopeLogs := resourceLogs.ScopeLogs().AppendEmpty()
+	otelLog := scopeLogs.LogRecords().AppendEmpty()
+	otelLog.SetTraceID(pcommon.TraceID{1})
+	otelLog.SetSpanID(pcommon.SpanID{2})
+	otelLog.SetSeverityNumber(plog.SeverityNumberInfo)
+	otelLog.SetSeverityText("Info")
+	otelLog.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(1, 0)))
+	otelLog.Attributes().PutStr("key", "value")
+	otelLog.Attributes().PutDouble("numeric_key", 1234)
+
+	switch b := body.(type) {
+	case string:
+		otelLog.Body().SetStr(b)
+	case int:
+		otelLog.Body().SetInt(int64(b))
+	case float64:
+		otelLog.Body().SetDouble(float64(b))
+	case bool:
+		otelLog.Body().SetBool(b)
+	}
+	return logs
+}
+
 func newOTLPTraceExporter(t testing.TB, srv *apmservertest.Server, options ...otlptracegrpc.Option) *otlptrace.Exporter {
 	options = append(options, otlptracegrpc.WithEndpoint(serverAddr(srv)), otlptracegrpc.WithInsecure())
 	exporter, err := otlptracegrpc.New(context.Background(), options...)