Ensure ECS compliant logging when enabled.

_meta/beat.yml

@@ -1063,8 +1063,8 @@ output.elasticsearch:
 
 # Set to true, to log messages with minimal required Elastic Common Schema (ECS)
 # information. Recommended to use in combination with `logging.json=true`
-# Defaults to false.
-#logging.ecs: false
+# Defaults to true.
+#logging.ecs: true
 
 #=============================== HTTP Endpoint ===============================
 

apm-server.docker.yml

@@ -1063,8 +1063,8 @@ output.elasticsearch:
 
 # Set to true, to log messages with minimal required Elastic Common Schema (ECS)
 # information. Recommended to use in combination with `logging.json=true`
-# Defaults to false.
-#logging.ecs: false
+# Defaults to true.
+#logging.ecs: true
 
 #=============================== HTTP Endpoint ===============================
 

apm-server.yml

@@ -1063,8 +1063,8 @@ output.elasticsearch:
 
 # Set to true, to log messages with minimal required Elastic Common Schema (ECS)
 # information. Recommended to use in combination with `logging.json=true`
-# Defaults to false.
-#logging.ecs: false
+# Defaults to true.
+#logging.ecs: true
 
 #=============================== HTTP Endpoint ===============================
 

beater/middleware/log_middleware.go

@@ -34,73 +34,76 @@ import (
 
 // LogMiddleware returns a middleware taking care of logging processing a request in the middleware and the request handler
 func LogMiddleware() Middleware {
-	logger := logp.NewLogger(logs.Request)
 	return func(h request.Handler) (request.Handler, error) {
-
 		return func(c *request.Context) {
-			var reqID, transactionID, traceID string
 			start := time.Now()
-			tx := apm.TransactionFromContext(c.Request.Context())
-			if tx != nil {
-				// This request is being traced, grab its IDs to add to logs.
-				traceContext := tx.TraceContext()
-				transactionID = traceContext.Span.String()
-				traceID = traceContext.Trace.String()
-				reqID = transactionID
-			} else {
-				uuid, err := uuid.NewV4()
-				if err != nil {
-					id := request.IDResponseErrorsInternal
-					logger.Errorw(request.MapResultIDToStatus[id].Keyword, "error", err)
-					c.Result.SetWithError(id, err)
-					c.Write()
-					return
-				}
-				reqID = uuid.String()
-			}
-
-			reqLogger := logger.With(
-				"request_id", reqID,
-				"method", c.Request.Method,
-				"URL", c.Request.URL,
-				"content_length", c.Request.ContentLength,
-				"remote_address", utility.RemoteAddr(c.Request),
-				"user-agent", c.Request.Header.Get(headers.UserAgent))
-
-			if traceID != "" {
-				reqLogger = reqLogger.With(
-					"trace.id", traceID,
-					"transaction.id", transactionID,
-				)
+			c.Logger = loggerWithRequestContext(c)
+			var err error
+			if c.Logger, err = loggerWithTraceContext(c); err != nil {
+				id := request.IDResponseErrorsInternal
+				c.Logger.Error(request.MapResultIDToStatus[id].Keyword, logp.Error(err))
+				c.Result.SetWithError(id, err)
+				c.Write()
+				return
 			}
-
-			c.Logger = reqLogger
 			h(c)
-			reqLogger = reqLogger.With("event.duration", time.Since(start))
-
+			c.Logger = c.Logger.With("event.duration", time.Since(start))
 			if c.MultipleWriteAttempts() {
-				reqLogger.Warn("multiple write attempts")
+				c.Logger.Warn("multiple write attempts")
 			}
-
 			keyword := c.Result.Keyword
 			if keyword == "" {
 				keyword = "handled request"
 			}
-
-			keysAndValues := []interface{}{"response_code", c.Result.StatusCode}
-			if c.Result.Err != nil {
-				keysAndValues = append(keysAndValues, "error", c.Result.Err.Error())
-			}
-			if c.Result.Stacktrace != "" {
-				keysAndValues = append(keysAndValues, "stacktrace", c.Result.Stacktrace)
-			}
-
+			c.Logger = loggerWithResult(c)
 			if c.Result.Failure() {
-				reqLogger.Errorw(keyword, keysAndValues...)
-			} else {
-				reqLogger.Infow(keyword, keysAndValues...)
+				c.Logger.Error(keyword)
+				return
 			}
-
+			c.Logger.Info(keyword)
 		}, nil
 	}
 }
+
+func loggerWithRequestContext(c *request.Context) *logp.Logger {
+	logger := logp.NewLogger(logs.Request).With(
+		"url.original", c.Request.URL.String(),
+		"http.request.method", c.Request.Method,
+		"user_agent.original", c.Request.Header.Get(headers.UserAgent),
+		"source.address", utility.RemoteAddr(c.Request))
+	if c.Request.ContentLength != -1 {
+		logger = logger.With("http.request.body.bytes", c.Request.ContentLength)
+	}
+	return logger
+}
+
+func loggerWithTraceContext(c *request.Context) (*logp.Logger, error) {
+	tx := apm.TransactionFromContext(c.Request.Context())
+	if tx == nil {
+		uuid, err := uuid.NewV4()
+		if err != nil {
+			return c.Logger, err
+		}
+		return c.Logger.With("http.request.id", uuid.String()), nil
+	}
+	// This request is being traced, grab its IDs to add to logs.
+	traceContext := tx.TraceContext()
+	transactionID := traceContext.Span.String()
+	return c.Logger.With(
+		"trace.id", traceContext.Trace.String(),
+		"transaction.id", transactionID,
+		"http.request.id", transactionID,
+	), nil
+}
+
+func loggerWithResult(c *request.Context) *logp.Logger {
+	logger := c.Logger.With(
+		"http.response.status_code", c.Result.StatusCode)
+	if c.Result.Err != nil {
+		logger = logger.With("error.message", c.Result.Err.Error())
+	}
+	if c.Result.Stacktrace != "" {
+		logger = logger.With("error.stack_trace", c.Result.Stacktrace)
+	}
+	return logger
+}

beater/middleware/log_middleware_test.go

@@ -21,15 +21,16 @@ import (
 	"net/http"
 	"testing"
 
-	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zapcore"
 
 	"go.elastic.co/apm"
 	"go.elastic.co/apm/apmtest"
 
+	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/logp"
+	"github.com/elastic/beats/v7/libbeat/logp/configure"
 
 	"github.com/elastic/apm-server/beater/beatertest"
 	"github.com/elastic/apm-server/beater/headers"
@@ -38,31 +39,30 @@ import (
 )
 
 func TestLogMiddleware(t *testing.T) {
-	err := logp.DevelopmentSetup(logp.ToObserverOutput())
-	require.NoError(t, err)
 
 	testCases := []struct {
 		name, message string
 		level         zapcore.Level
 		handler       request.Handler
 		code          int
-		error         error
-		stacktrace    bool
 		traced        bool
+		ecsKeys       []string
 	}{
 		{
 			name:    "Accepted",
 			message: "request accepted",
 			level:   zapcore.InfoLevel,
 			handler: beatertest.Handler202,
 			code:    http.StatusAccepted,
+			ecsKeys: []string{"url.original"},
 		},
 		{
 			name:    "Traced",
 			message: "request accepted",
 			level:   zapcore.InfoLevel,
 			handler: beatertest.Handler202,
 			code:    http.StatusAccepted,
+			ecsKeys: []string{"url.original", "trace.id", "transaction.id"},
 			traced:  true,
 		},
 		{
@@ -71,16 +71,15 @@ func TestLogMiddleware(t *testing.T) {
 			level:   zapcore.ErrorLevel,
 			handler: beatertest.Handler403,
 			code:    http.StatusForbidden,
-			error:   errors.New("forbidden request"),
+			ecsKeys: []string{"url.original", "error.message"},
 		},
 		{
-			name:       "Panic",
-			message:    "internal error",
-			level:      zapcore.ErrorLevel,
-			handler:    Apply(RecoverPanicMiddleware(), beatertest.HandlerPanic),
-			code:       http.StatusInternalServerError,
-			error:      errors.New("panic on Handle"),
-			stacktrace: true,
+			name:    "Panic",
+			message: "internal error",
+			level:   zapcore.ErrorLevel,
+			handler: Apply(RecoverPanicMiddleware(), beatertest.HandlerPanic),
+			code:    http.StatusInternalServerError,
+			ecsKeys: []string{"url.original", "error.message", "error.stack_trace"},
 		},
 		{
 			name:    "Error without keyword",
@@ -90,12 +89,19 @@ func TestLogMiddleware(t *testing.T) {
 				c.Result.StatusCode = http.StatusForbidden
 				c.Write()
 			},
-			code: http.StatusForbidden,
+			code:    http.StatusForbidden,
+			ecsKeys: []string{"url.original"},
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			// log setup
+			configure.Logging("APM Server test",
+				common.MustNewConfigFrom(`{"ecs":true}`))
+			require.NoError(t, logp.DevelopmentSetup(logp.ToObserverOutput()))
+
+			// prepare and record request
 			c, rec := beatertest.DefaultContextWithResponseRecorder()
 			c.Request.Header.Set(headers.UserAgent, tc.name)
 			if tc.traced {
@@ -105,39 +111,27 @@ func TestLogMiddleware(t *testing.T) {
 			}
 			Apply(LogMiddleware(), tc.handler)(c)
 
+			// check log lines
 			assert.Equal(t, tc.code, rec.Code)
-			for i, entry := range logp.ObserverLogs().TakeAll() {
-				// expect only one log entry per request
-				assert.Equal(t, i, 0)
-				assert.Equal(t, logs.Request, entry.LoggerName)
-				assert.Equal(t, tc.level, entry.Level)
-				assert.Equal(t, tc.message, entry.Message)
+			entries := logp.ObserverLogs().TakeAll()
+			require.Equal(t, 1, len(entries))
+			entry := entries[0]
+			assert.Equal(t, logs.Request, entry.LoggerName)
+			assert.Equal(t, tc.level, entry.Level)
+			assert.Equal(t, tc.message, entry.Message)
 
-				ec := entry.ContextMap()
-				t.Logf("context map: %v", ec)
-
-				assert.NotEmpty(t, ec["request_id"])
-				assert.NotEmpty(t, ec["method"])
-				assert.Equal(t, c.Request.URL.String(), ec["URL"])
-				assert.NotEmpty(t, ec["remote_address"])
-				assert.Contains(t, ec, "event.duration")
-				assert.Equal(t, c.Request.Header.Get(headers.UserAgent), ec["user-agent"])
-				// zap encoded type
-				assert.Equal(t, tc.code, int(ec["response_code"].(int64)))
-				if tc.error != nil {
-					assert.Equal(t, tc.error.Error(), ec["error"])
-				}
-				if tc.stacktrace {
-					assert.NotZero(t, ec["stacktrace"])
-				}
-				if tc.traced {
-					assert.NotEmpty(t, ec, "trace.id")
-					assert.NotEmpty(t, ec, "transaction.id")
-					assert.Equal(t, ec["request_id"], ec["transaction.id"])
-				} else {
-					assert.NotContains(t, ec, "trace.id")
-					assert.NotContains(t, ec, "transaction.id")
-				}
+			encoder := zapcore.NewMapObjectEncoder()
+			ec := common.MapStr{}
+			for _, f := range entry.Context {
+				f.AddTo(encoder)
+				ec.DeepUpdate(encoder.Fields)
+			}
+			keys := []string{"http.request.id", "http.request.method", "http.request.body.bytes",
+				"source.address", "user_agent.original", "http.response.status_code", "event.duration"}
+			keys = append(keys, tc.ecsKeys...)
+			for _, key := range keys {
+				ok, _ := ec.HasKey(key)
+				assert.True(t, ok, key)
 			}
 		})
 	}

changelogs/head.asciidoc

@@ -28,3 +28,4 @@ https://github.com/elastic/apm-server/compare/7.10\...master[View commits]
 * Label/custom/mark keys are now sanitized (rather than validated and rejected) by the server {pull}4465[4465]
 * Upgrade Go to 1.14.12 {pull}4478[4478]
 * Added apm-server.response_headers config {pull}4523[4523]
+* Switch logging format to be ECS compliant where possible {pull}3829[3829]

cmd/root.go

@@ -48,6 +48,7 @@ var libbeatConfigOverrides = []cfgfile.ConditionalOverride{{
 			"metrics": map[string]interface{}{
 				"enabled": false,
 			},
+			"ecs": true,
 		},
 	}),
 }}

docs/copied-from-beats/docs/loggingconfig.asciidoc

@@ -249,8 +249,7 @@ When true, logs messages in JSON format. The default is false.
 [float]
 ==== `logging.ecs`
 
-When true, logs messages with minimal required Elastic Common Schema (ECS)
-information.
+When true, logs messages in Elastic Common Schema (ECS) compliant format.
 
 ifndef::serverless[]
 [float]

systemtest/export_test.go

@@ -49,6 +49,7 @@ func TestExportConfigDefaults(t *testing.T) {
 
 	expectedConfig := strings.ReplaceAll(`
 logging:
+  ecs: true
   metrics:
     enabled: false
 path:
@@ -69,6 +70,7 @@ func TestExportConfigOverrideDefaults(t *testing.T) {
 
 	expectedConfig := strings.ReplaceAll(`
 logging:
+  ecs: true
   metrics:
     enabled: true
 path:

systemtest/logging_test.go

@@ -69,8 +69,8 @@ func TestAPMServerRequestLoggingValid(t *testing.T) {
 
 	srv.Close()
 	for _, entry := range srv.Logs.All() {
-		if entry.Logger == "request" && entry.Fields["URL"] == "/intake/v2/events" {
-			statusCode, _ := entry.Fields["response_code"].(float64)
+		if entry.Logger == "request" && entry.Fields["url.original"] == "/intake/v2/events" {
+			statusCode, _ := entry.Fields["http.response.status_code"].(float64)
 			logEntries = append(logEntries, entry)
 			requestEntries = append(requestEntries, requestEntry{
 				level:      entry.Level,
@@ -95,8 +95,8 @@ func TestAPMServerRequestLoggingValid(t *testing.T) {
 	}}, requestEntries)
 
 	assert.NotContains(t, logEntries[0].Fields, "error")
-	assert.Regexp(t, "validation error: 'transaction' required", logEntries[1].Fields["error"])
-	assert.Equal(t, "event exceeded the permitted size.", logEntries[2].Fields["error"])
+	assert.Regexp(t, "validation error: 'transaction' required", logEntries[1].Fields["error.message"])
+	assert.Equal(t, "event exceeded the permitted size.", logEntries[2].Fields["error.message"])
 }
 
 // validMetadataJSON returns valid JSON-encoded metadata,

tests/system/apmserver.py

@@ -337,8 +337,8 @@ def check_for_no_smap(self, doc):
     def logged_requests(self, url="/intake/v2/events"):
         for line in self.get_log_lines():
             jline = json.loads(line)
-            u = urlparse(jline.get("URL", ""))
-            if jline.get("logger") == "request" and u.path == url:
+            u = urlparse(jline.get("url.original", ""))
+            if jline.get("log.logger") == "request" and u.path == url:
                 yield jline
 
     def approve_docs(self, base_path, received):