Refactor metadata enrichment and fix bug in v3 Intake (elastic#3943)

Decode request metadata into a struct instead of a map and pass the metadata struct
into a decoder function. The refactoring automatically fixes a bug where the `client.ip`
and `user_agent` were not set for v3 RUM events.

implements elastic#3885
fixes elastic#3942

beater/api/intake/handler.go

@@ -30,6 +30,7 @@ import (
 	"github.com/elastic/apm-server/beater/headers"
 	"github.com/elastic/apm-server/beater/request"
 	"github.com/elastic/apm-server/decoder"
+	"github.com/elastic/apm-server/model"
 	"github.com/elastic/apm-server/processor/stream"
 	"github.com/elastic/apm-server/publish"
 )
@@ -63,7 +64,11 @@ func Handler(processor *stream.Processor, report publish.Reporter) request.Handl
 			return
 		}
 
-		res := processor.HandleStream(c.Request.Context(), c.RateLimiter, c.RequestMetadata, reader, report)
+		metadata := model.Metadata{
+			UserAgent: model.UserAgent{Original: c.RequestMetadata.UserAgent},
+			Client:    model.Client{IP: c.RequestMetadata.ClientIP},
+			System:    model.System{IP: c.RequestMetadata.SystemIP}}
+		res := processor.HandleStream(c.Request.Context(), c.RateLimiter, &metadata, reader, report)
 		sendResponse(c, res)
 	}
 }

beater/api/profile/handler.go

@@ -35,7 +35,6 @@ import (
 	"github.com/elastic/apm-server/model/modeldecoder"
 	"github.com/elastic/apm-server/publish"
 	"github.com/elastic/apm-server/transform"
-	"github.com/elastic/apm-server/utility"
 	"github.com/elastic/apm-server/validation"
 )
 
@@ -123,11 +122,11 @@ func Handler(
 						err: errors.Wrap(err, "failed to decode metadata JSON"),
 					}
 				}
-				for k, v := range c.RequestMetadata {
-					utility.InsertInMap(raw, k, v.(map[string]interface{}))
-				}
-				metadata, err := modeldecoder.DecodeMetadata(raw, false)
-				if err != nil {
+				metadata := model.Metadata{
+					UserAgent: model.UserAgent{Original: c.RequestMetadata.UserAgent},
+					Client:    model.Client{IP: c.RequestMetadata.ClientIP},
+					System:    model.System{IP: c.RequestMetadata.SystemIP}}
+				if err := modeldecoder.DecodeMetadata(raw, false, &metadata); err != nil {
 					var ve *validation.Error
 					if errors.As(err, &ve) {
 						return nil, requestError{
@@ -140,7 +139,7 @@ func Handler(
 						err: errors.Wrap(err, "failed to decode metadata"),
 					}
 				}
-				profileMetadata = *metadata
+				profileMetadata = metadata
 
 			case "profile":
 				params, err := validateContentType(http.Header(part.Header), pprofMediaType)

beater/middleware/request_metadata_middleware.go

@@ -28,13 +28,8 @@ func UserMetadataMiddleware() Middleware {
 	return func(h request.Handler) (request.Handler, error) {
 		return func(c *request.Context) {
 			dec := utility.ManualDecoder{}
-			user := map[string]interface{}{
-				"user-agent": dec.UserAgentHeader(c.Request.Header),
-			}
-			if ip := utility.ExtractIP(c.Request); ip != nil {
-				user["ip"] = ip.String()
-			}
-			c.RequestMetadata["user"] = user
+			c.RequestMetadata.UserAgent = dec.UserAgentHeader(c.Request.Header)
+			c.RequestMetadata.ClientIP = utility.ExtractIP(c.Request)
 			h(c)
 		}, nil
 	}
@@ -45,10 +40,7 @@ func UserMetadataMiddleware() Middleware {
 func SystemMetadataMiddleware() Middleware {
 	return func(h request.Handler) (request.Handler, error) {
 		return func(c *request.Context) {
-			if ip := utility.ExtractIP(c.Request); ip != nil {
-				system := map[string]interface{}{"ip": ip.String()}
-				c.RequestMetadata["system"] = system
-			}
+			c.RequestMetadata.SystemIP = utility.ExtractIP(c.Request)
 			h(c)
 		}, nil
 	}

beater/middleware/request_metadata_middleware_test.go

@@ -19,6 +19,7 @@ package middleware
 
 import (
 	"fmt"
+	"net"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -28,17 +29,17 @@ import (
 
 func TestUserMetadataMiddleware(t *testing.T) {
 	type test struct {
-		remoteAddr      string
-		userAgent       []string
-		expectIP        string
-		expectUserAgent string
+		remoteAddr        string
+		userAgent         []string
+		expectedIP        net.IP
+		expectedUserAgent string
 	}
 
 	ua1 := "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.103 Safari/537.36"
 	ua2 := "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.14; rv:67.0) Gecko/20100101 Firefox/67.0"
 	tests := []test{
-		{remoteAddr: "1.2.3.4:1234", expectIP: "1.2.3.4", userAgent: []string{ua1, ua2}, expectUserAgent: fmt.Sprintf("%s, %s", ua1, ua2)},
-		{remoteAddr: "not-an-ip:1234", userAgent: []string{ua1}, expectUserAgent: ua1},
+		{remoteAddr: "1.2.3.4:1234", expectedIP: net.ParseIP("1.2.3.4"), userAgent: []string{ua1, ua2}, expectedUserAgent: fmt.Sprintf("%s, %s", ua1, ua2)},
+		{remoteAddr: "not-an-ip:1234", userAgent: []string{ua1}, expectedUserAgent: ua1},
 		{remoteAddr: ""},
 	}
 
@@ -50,23 +51,18 @@ func TestUserMetadataMiddleware(t *testing.T) {
 		}
 
 		Apply(UserMetadataMiddleware(), beatertest.HandlerIdle)(c)
-		expect := map[string]interface{}{
-			"user-agent": test.expectUserAgent, // even if empty value
-		}
-		if test.expectIP != "" {
-			expect["ip"] = test.expectIP
-		}
-		assert.Equal(t, map[string]interface{}{"user": expect}, c.RequestMetadata)
+		assert.Equal(t, test.expectedUserAgent, c.RequestMetadata.UserAgent)
+		assert.Equal(t, test.expectedIP, c.RequestMetadata.ClientIP)
 	}
 }
 
 func TestSystemMetadataMiddleware(t *testing.T) {
 	type test struct {
 		remoteAddr string
-		expectIP   string
+		expectedIP net.IP
 	}
 	tests := []test{
-		{remoteAddr: "1.2.3.4:1234", expectIP: "1.2.3.4"},
+		{remoteAddr: "1.2.3.4:1234", expectedIP: net.ParseIP("1.2.3.4")},
 		{remoteAddr: "not-an-ip:1234"},
 		{remoteAddr: ""},
 	}
@@ -76,14 +72,6 @@ func TestSystemMetadataMiddleware(t *testing.T) {
 		c.Request.RemoteAddr = test.remoteAddr
 
 		Apply(SystemMetadataMiddleware(), beatertest.HandlerIdle)(c)
-		if test.expectIP == "" {
-			assert.Empty(t, c.RequestMetadata)
-		} else {
-			assert.Equal(t, map[string]interface{}{
-				"system": map[string]interface{}{
-					"ip": test.expectIP,
-				},
-			}, c.RequestMetadata)
-		}
+		assert.Equal(t, test.expectedIP, c.RequestMetadata.SystemIP)
 	}
 }

beater/request/context.go

@@ -19,6 +19,7 @@ package request
 
 import (
 	"encoding/json"
+	"net"
 	"net/http"
 	"strings"
 
@@ -42,23 +43,29 @@ var (
 
 // Context abstracts request and response information for http requests
 type Context struct {
-	Request       *http.Request
-	Logger        *logp.Logger
-	RateLimiter   *rate.Limiter
-	Authorization authorization.Authorization
-	IsRum         bool
-	Result        Result
-
-	// RequestMetadata contains metadata extracted from the request
-	// by middleware, and should be merged into event metadata.
-	RequestMetadata map[string]interface{}
+	Request         *http.Request
+	Logger          *logp.Logger
+	RateLimiter     *rate.Limiter
+	Authorization   authorization.Authorization
+	IsRum           bool
+	Result          Result
+	RequestMetadata Metadata
 
 	w             http.ResponseWriter
 	writeAttempts int
 }
 
+// Metadata contains metadata extracted from the request by middleware,
+// and should be merged into the event metadata.
+type Metadata struct {
+	ClientIP  net.IP
+	SystemIP  net.IP
+	UserAgent string
+}
+
+// NewContext creates an empty Context struct
 func NewContext() *Context {
-	return &Context{RequestMetadata: make(map[string]interface{})}
+	return &Context{}
 }
 
 // Reset allows to reuse a context by removing all request specific information
@@ -69,14 +76,19 @@ func (c *Context) Reset(w http.ResponseWriter, r *http.Request) {
 	c.Authorization = &authorization.AllowAuth{}
 	c.IsRum = false
 	c.Result.Reset()
-	for k := range c.RequestMetadata {
-		delete(c.RequestMetadata, k)
-	}
+	c.RequestMetadata.Reset()
 
 	c.w = w
 	c.writeAttempts = 0
 }
 
+// Reset sets all attribtues of the Metadata instance to it's zero value
+func (m *Metadata) Reset() {
+	m.ClientIP = nil
+	m.SystemIP = nil
+	m.UserAgent = ""
+}
+
 // Header returns the http.Header of the context's writer
 func (c *Context) Header() http.Header {
 	return c.w.Header()

beater/request/context_test.go

@@ -65,6 +65,8 @@ func TestContext_Reset(t *testing.T) {
 			assert.Equal(t, 0, c.writeAttempts)
 		case "Result":
 			assertResultIsEmpty(t, cVal.Field(i).Interface().(Result))
+		case "RequestMetadata":
+			assert.Equal(t, Metadata{}, cVal.Field(i).Interface().(Metadata))
 		default:
 			assert.Empty(t, cVal.Field(i).Interface(), cType.Field(i).Name)
 		}

beater/test_approved_es_documents/TestPublishIntegrationMetricsets.approved.json

@@ -23,6 +23,9 @@
                 "version": "1.5.0"
             },
             "float_gauge": 9.16,
+            "host": {
+                "ip": "127.0.0.1"
+            },
             "integer_gauge": 42767,
             "labels": {
                 "code": 200,
@@ -129,6 +132,9 @@
                     }
                 }
             },
+            "host": {
+                "ip": "127.0.0.1"
+            },
             "labels": {
                 "tag1": "one",
                 "tag2": 2

docs/data/elasticsearch/generated/metricsets.json

@@ -20,6 +20,9 @@
                 }
             }
         },
+        "host": {
+            "ip": "127.0.0.1"
+        },
         "labels": {
             "tag1": "one",
             "tag2": 2
@@ -74,6 +77,9 @@
             "ingested": "2020-04-22T14:55:05.368308Z"
         },
         "float_gauge": 9.16,
+        "host": {
+            "ip": "127.0.0.1"
+        },
         "integer_gauge": 42767,
         "labels": {
             "code": 200,

model/modeldecoder/metadata.go

@@ -33,21 +33,13 @@ var (
 )
 
 // DecodeRUMV3Metadata decodes v3 RUM metadata.
-func DecodeRUMV3Metadata(input interface{}, hasShortFieldNames bool) (*model.Metadata, error) {
-	var out model.Metadata
-	if err := decodeMetadata(input, hasShortFieldNames, rumV3MetadataSchema, &out); err != nil {
-		return nil, err
-	}
-	return &out, nil
+func DecodeRUMV3Metadata(input interface{}, hasShortFieldNames bool, out *model.Metadata) error {
+	return decodeMetadata(input, hasShortFieldNames, rumV3MetadataSchema, out)
 }
 
 // DecodeMetadata decodes v2 metadata.
-func DecodeMetadata(input interface{}, hasShortFieldNames bool) (*model.Metadata, error) {
-	var out model.Metadata
-	if err := decodeMetadata(input, hasShortFieldNames, metadataSchema, &out); err != nil {
-		return nil, err
-	}
-	return &out, nil
+func DecodeMetadata(input interface{}, hasShortFieldNames bool, out *model.Metadata) error {
+	return decodeMetadata(input, hasShortFieldNames, metadataSchema, out)
 }
 
 func decodeMetadata(input interface{}, hasShortFieldNames bool, schema *jsonschema.Schema, out *model.Metadata) error {
@@ -61,8 +53,6 @@ func decodeMetadata(input interface{}, hasShortFieldNames bool, schema *jsonsche
 	decodeProcess(getObject(raw, "process"), &out.Process)
 	if userObj := getObject(raw, fieldName("user")); userObj != nil {
 		decodeUser(userObj, hasShortFieldNames, &out.User, &out.Client)
-		// TODO(axw) stop decoding user.user-agent here (see #3885)
-		decodeString(userObj, "user-agent", &out.UserAgent.Original)
 	}
 	decodeCloud(getObject(raw, "cloud"), &out.Cloud)
 	decodeLabels(getObject(raw, fieldName("labels")), &out.Labels)