otlp: Add experimental support to receive logs

Adds support to receive OTLP logs on the GRPC receiver. This feature is _beta_ and should be used sparsely until the OTel spec declares it as stable. Currently, it's challenging to test this in the system tests since the `opentelemetry-go` SDK doesn't expose any APIs for logs. Signed-off-by: Marc Lopez Rubio <[email protected]>
elastic · Dec 2, 2021 · 017039d · 017039d
1 parent add9721
commit 017039d
Show file tree

Hide file tree

Showing 20 changed files with 398 additions and 30 deletions.
diff --git a/beater/otlp/grpc.go b/beater/otlp/grpc.go
@@ -44,18 +44,22 @@ var (
 	gRPCMetricsMonitoringMap = request.MonitoringMapForRegistry(gRPCMetricsRegistry, monitoringKeys)
 	gRPCTracesRegistry       = monitoring.Default.NewRegistry("apm-server.otlp.grpc.traces")
 	gRPCTracesMonitoringMap  = request.MonitoringMapForRegistry(gRPCTracesRegistry, monitoringKeys)
+	gRPCLogsRegistry         = monitoring.Default.NewRegistry("apm-server.otlp.grpc.logs")
+	gRPCLogsMonitoringMap    = request.MonitoringMapForRegistry(gRPCLogsRegistry, monitoringKeys)
 
 	// RegistryMonitoringMaps provides mappings from the fully qualified gRPC
 	// method name to its respective monitoring map.
 	RegistryMonitoringMaps = map[string]map[request.ResultID]*monitoring.Int{
 		metricsFullMethod: gRPCMetricsMonitoringMap,
 		tracesFullMethod:  gRPCTracesMonitoringMap,
+		logsFullMethod:    gRPCLogsMonitoringMap,
 	}
 )
 
 const (
 	metricsFullMethod = "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export"
 	tracesFullMethod  = "/opentelemetry.proto.collector.trace.v1.TraceService/Export"
+	logsFullMethod    = "/opentelemetry.proto.collector.logs.v1.LogsService/Export"
 )
 
 func init() {
@@ -68,6 +72,7 @@ func MethodAuthenticators(authenticator *auth.Authenticator) map[string]intercep
 	return map[string]interceptors.MethodAuthenticator{
 		metricsFullMethod: metadataMethodAuthenticator,
 		tracesFullMethod:  metadataMethodAuthenticator,
+		logsFullMethod:    metadataMethodAuthenticator,
 	}
 }
 
@@ -86,6 +91,9 @@ func RegisterGRPCServices(grpcServer *grpc.Server, processor model.BatchProcesso
 	if err := otlpreceiver.RegisterMetricsReceiver(context.Background(), consumer, grpcServer); err != nil {
 		return errors.Wrap(err, "failed to register OTLP metrics receiver")
 	}
+	if err := otlpreceiver.RegisterLogsReceiver(context.Background(), consumer, grpcServer); err != nil {
+		return errors.Wrap(err, "failed to register OTLP logs receiver")
+	}
 	return nil
 }
 

diff --git a/beater/otlp/grpc_test.go b/beater/otlp/grpc_test.go
@@ -133,6 +133,53 @@ func TestConsumeMetrics(t *testing.T) {
 	}, actual)
 }
 
+func TestConsumeLogs(t *testing.T) {
+	var batches []model.Batch
+	var reportError error
+	var batchProcessor model.ProcessBatchFunc = func(ctx context.Context, batch *model.Batch) error {
+		batches = append(batches, *batch)
+		return reportError
+	}
+
+	conn := newServer(t, batchProcessor)
+	client := otlpgrpc.NewLogsClient(conn)
+
+	// Send a minimal log to verify that everything is connected properly.
+	//
+	// We intentionally do not check the published event contents; those are
+	// tested in processor/otel.
+	logs := pdata.NewLogs()
+	log := logs.ResourceLogs().AppendEmpty().InstrumentationLibraryLogs().AppendEmpty().Logs().AppendEmpty()
+	log.SetName("log_name")
+
+	_, err := client.Export(context.Background(), logs)
+	assert.NoError(t, err)
+	require.Len(t, batches, 1)
+
+	reportError = errors.New("failed to publish events")
+	_, err = client.Export(context.Background(), logs)
+	assert.Error(t, err)
+	errStatus := status.Convert(err)
+	assert.Equal(t, "failed to publish events", errStatus.Message())
+	require.Len(t, batches, 2)
+	assert.Len(t, batches[0], 1)
+	assert.Len(t, batches[1], 1)
+
+	actual := map[string]interface{}{}
+	monitoring.GetRegistry("apm-server.otlp.grpc.logs").Do(monitoring.Full, func(key string, value interface{}) {
+		actual[key] = value
+	})
+	assert.Equal(t, map[string]interface{}{
+		"request.count":                int64(2),
+		"response.count":               int64(2),
+		"response.errors.count":        int64(1),
+		"response.valid.count":         int64(1),
+		"response.errors.ratelimit":    int64(0),
+		"response.errors.timeout":      int64(0),
+		"response.errors.unauthorized": int64(0),
+	}, actual)
+}
+
 func newServer(t *testing.T, batchProcessor model.BatchProcessor) *grpc.ClientConn {
 	lis, err := net.Listen("tcp", "localhost:0")
 	require.NoError(t, err)

diff --git a/model/apmevent.go b/model/apmevent.go
@@ -30,9 +30,6 @@ import (
 // Exactly one of the event fields should be non-nil.
 type APMEvent struct {
 	// DataStream optionally holds data stream identifiers.
-	//
-	// This will have the zero value when APM Server is run
-	// in standalone mode.
 	DataStream DataStream
 
 	ECSVersion  string
@@ -59,6 +56,7 @@ type APMEvent struct {
 	Child       Child
 	HTTP        HTTP
 	FAAS        FAAS
+	Log         Log
 
 	// Timestamp holds the event timestamp.
 	//
@@ -152,5 +150,6 @@ func (e *APMEvent) BeatEvent(ctx context.Context) beat.Event {
 	fields.maybeSetString("message", e.Message)
 	fields.maybeSetMapStr("http", e.HTTP.fields())
 	fields.maybeSetMapStr("faas", e.FAAS.fields())
+	fields.maybeSetMapStr("log", e.Log.fields())
 	return event
 }
diff --git a/model/error.go b/model/error.go
@@ -38,7 +38,7 @@ type Error struct {
 	Custom      common.MapStr
 
 	Exception *Exception
-	Log       *Log
+	Log       *ErrorLog
 }
 
 type Exception struct {
@@ -52,7 +52,7 @@ type Exception struct {
 	Cause      []Exception
 }
 
-type Log struct {
+type ErrorLog struct {
 	Message      string
 	Level        string
 	ParamMessage string

diff --git a/model/error_test.go b/model/error_test.go
@@ -36,8 +36,8 @@ func (e *Exception) withCode(code string) *Exception {
 	return e
 }
 
-func baseLog() *Log {
-	return &Log{Message: "error log message"}
+func baseLog() *ErrorLog {
+	return &ErrorLog{Message: "error log message"}
 }
 
 func TestHandleExceptionTree(t *testing.T) {
@@ -132,7 +132,7 @@ func TestEventFields(t *testing.T) {
 	loggerName := "logger"
 	logMsg := "error log message"
 	paramMsg := "param message"
-	log := Log{
+	log := ErrorLog{
 		Level:        level,
 		Message:      logMsg,
 		ParamMessage: paramMsg,

diff --git a/model/event.go b/model/event.go
@@ -36,10 +36,20 @@ type Event struct {
 
 	// Outcome holds the event outcome: "success", "failure", or "unknown".
 	Outcome string
+
+	// Severity holds the numeric severity of the event for log events.
+	Severity int64
+
+	// Severity holds the action captured by the event for log events.
+	Action string
 }
 
 func (e *Event) fields() common.MapStr {
 	var fields mapStr
 	fields.maybeSetString("outcome", e.Outcome)
+	fields.maybeSetString("action", e.Action)
+	if e.Severity > 0 {
+		fields.set("severity", e.Severity)
+	}
 	return common.MapStr(fields)
 }
diff --git a/model/log.go b/model/log.go
@@ -17,6 +17,8 @@
 
 package model
 
+import "github.com/elastic/beats/v7/libbeat/common"
+
 const (
 	AppLogsDataset = "apm.app"
 )
@@ -25,3 +27,17 @@ var (
 	// LogProcessor is the Processor value that should be assigned to log events.
 	LogProcessor = Processor{Name: "log", Event: "log"}
 )
+
+// Log holds information about a log, as defined by ECS.
+//
+// https://www.elastic.co/guide/en/ecs/current/ecs-log.html
+type Log struct {
+	// Level holds the log level of the log event.
+	Level string
+}
+
+func (e Log) fields() common.MapStr {
+	var fields mapStr
+	fields.maybeSetString("level", e.Level)
+	return common.MapStr(fields)
+}
diff --git a/model/modeldecoder/rumv3/decoder.go b/model/modeldecoder/rumv3/decoder.go
@@ -210,7 +210,7 @@ func mapToErrorModel(from *errorEvent, event *model.APMEvent) {
 		out.ID = from.ID.Val
 	}
 	if from.Log.IsSet() {
-		log := model.Log{}
+		log := model.ErrorLog{}
 		if from.Log.Level.IsSet() {
 			log.Level = from.Log.Level.Val
 		}

diff --git a/model/modeldecoder/rumv3/metadata_test.go b/model/modeldecoder/rumv3/metadata_test.go
@@ -82,6 +82,7 @@ func metadataExceptions(keys ...string) func(key string) bool {
 		"Session",
 		"Trace",
 		"URL",
+		"Log",
 
 		// Dedicated test for it.
 		"NumericLabels",

diff --git a/model/modeldecoder/v2/decoder.go b/model/modeldecoder/v2/decoder.go
@@ -362,7 +362,7 @@ func mapToErrorModel(from *errorEvent, event *model.APMEvent) {
 		out.ID = from.ID.Val
 	}
 	if from.Log.IsSet() {
-		log := model.Log{}
+		log := model.ErrorLog{}
 		if from.Log.Level.IsSet() {
 			log.Level = from.Log.Level.Val
 		}

diff --git a/model/modeldecoder/v2/metadata_test.go b/model/modeldecoder/v2/metadata_test.go
@@ -113,6 +113,10 @@ func isUnmappedMetadataField(key string) bool {
 		"Event",
 		"Event.Duration",
 		"Event.Outcome",
+		"Event.Severity",
+		"Event.Action",
+		"Log",
+		"Log.Level",
 		"Service.Origin",
 		"Service.Origin.ID",
 		"Service.Origin.Name",

diff --git a/model/modelprocessor/culprit_test.go b/model/modelprocessor/culprit_test.go
@@ -40,23 +40,23 @@ func TestSetCulprit(t *testing.T) {
 	}, {
 		input: model.Error{
 			Culprit: "already_set",
-			Log: &model.Log{
+			Log: &model.ErrorLog{
 				Stacktrace: model.Stacktrace{{SourcemapUpdated: false, Filename: "foo.go"}},
 			},
 		},
 		culprit: "already_set",
 	}, {
 		input: model.Error{
 			Culprit: "already_set",
-			Log: &model.Log{
+			Log: &model.ErrorLog{
 				Stacktrace: model.Stacktrace{{SourcemapUpdated: true, LibraryFrame: true, Filename: "foo.go"}},
 			},
 		},
 		culprit: "already_set",
 	}, {
 		input: model.Error{
 			Culprit: "already_set",
-			Log: &model.Log{
+			Log: &model.ErrorLog{
 				Stacktrace: model.Stacktrace{
 					{SourcemapUpdated: true, LibraryFrame: true, Filename: "foo.go"},
 					{SourcemapUpdated: true, LibraryFrame: false, Filename: "foo2.go"},
@@ -67,7 +67,7 @@ func TestSetCulprit(t *testing.T) {
 	}, {
 		input: model.Error{
 			Culprit: "already_set",
-			Log: &model.Log{
+			Log: &model.ErrorLog{
 				Stacktrace: model.Stacktrace{{SourcemapUpdated: true, LibraryFrame: true, Filename: "foo.go"}},
 			},
 			Exception: &model.Exception{
@@ -77,7 +77,7 @@ func TestSetCulprit(t *testing.T) {
 		culprit: "foo2.go",
 	}, {
 		input: model.Error{
-			Log: &model.Log{
+			Log: &model.ErrorLog{
 				Stacktrace: model.Stacktrace{
 					{SourcemapUpdated: true, Classname: "AbstractFactoryManagerBean", Function: "toString"},
 				},

diff --git a/model/modelprocessor/errormessage_test.go b/model/modelprocessor/errormessage_test.go
@@ -35,20 +35,20 @@ func TestSetErrorMessage(t *testing.T) {
 		input:   model.Error{},
 		message: "",
 	}, {
-		input:   model.Error{Log: &model.Log{Message: "log_message"}},
+		input:   model.Error{Log: &model.ErrorLog{Message: "log_message"}},
 		message: "log_message",
 	}, {
 		input:   model.Error{Exception: &model.Exception{Message: "exception_message"}},
 		message: "exception_message",
 	}, {
 		input: model.Error{
-			Log:       &model.Log{},
+			Log:       &model.ErrorLog{},
 			Exception: &model.Exception{Message: "exception_message"},
 		},
 		message: "exception_message",
 	}, {
 		input: model.Error{
-			Log:       &model.Log{Message: "log_message"},
+			Log:       &model.ErrorLog{Message: "log_message"},
 			Exception: &model.Exception{Message: "exception_message"},
 		},
 		message: "log_message",

diff --git a/model/modelprocessor/excludefromgrouping_test.go b/model/modelprocessor/excludefromgrouping_test.go
@@ -60,7 +60,7 @@ func TestSetExcludeFromGrouping(t *testing.T) {
 	}, {
 		input: model.Batch{{
 			Error: &model.Error{
-				Log: &model.Log{
+				Log: &model.ErrorLog{
 					Stacktrace: model.Stacktrace{
 						{Filename: "foo.go"},
 					},
@@ -82,7 +82,7 @@ func TestSetExcludeFromGrouping(t *testing.T) {
 		}},
 		output: model.Batch{{
 			Error: &model.Error{
-				Log: &model.Log{
+				Log: &model.ErrorLog{
 					Stacktrace: model.Stacktrace{
 						{ExcludeFromGrouping: true, Filename: "foo.go"},
 					},

diff --git a/model/modelprocessor/groupingkey_test.go b/model/modelprocessor/groupingkey_test.go
@@ -43,7 +43,7 @@ func TestSetGroupingKey(t *testing.T) {
 				Exception: &model.Exception{
 					Type: "exception_type",
 				},
-				Log: &model.Log{
+				Log: &model.ErrorLog{
 					ParamMessage: "log_parammessage",
 				},
 			},
@@ -72,15 +72,15 @@ func TestSetGroupingKey(t *testing.T) {
 						},
 					}},
 				},
-				Log: &model.Log{Stacktrace: model.Stacktrace{{Filename: "abc"}}}, // ignored
+				Log: &model.ErrorLog{Stacktrace: model.Stacktrace{{Filename: "abc"}}}, // ignored
 			},
 			groupingKey: hashStrings(
 				"module", "func_1", "filename", "func_2", "classname", "func_4", "func_5", "func_6",
 			),
 		},
 		"log_stacktrace": {
 			input: model.Error{
-				Log: &model.Log{
+				Log: &model.ErrorLog{
 					Stacktrace: model.Stacktrace{{Function: "function"}},
 				},
 			},
@@ -99,13 +99,13 @@ func TestSetGroupingKey(t *testing.T) {
 						Message: "message_4",
 					}},
 				},
-				Log: &model.Log{Message: "log_message"}, // ignored
+				Log: &model.ErrorLog{Message: "log_message"}, // ignored
 			},
 			groupingKey: hashStrings("message_1", "message_2", "message_3", "message_4"),
 		},
 		"log_message": {
 			input: model.Error{
-				Log: &model.Log{Message: "log_message"}, // ignored
+				Log: &model.ErrorLog{Message: "log_message"}, // ignored
 			},
 			groupingKey: hashStrings("log_message"),
 		},

diff --git a/model/modelprocessor/libraryframe_test.go b/model/modelprocessor/libraryframe_test.go
@@ -62,7 +62,7 @@ func TestSetLibraryFrames(t *testing.T) {
 	}, {
 		input: model.Batch{{
 			Error: &model.Error{
-				Log: &model.Log{
+				Log: &model.ErrorLog{
 					Stacktrace: model.Stacktrace{
 						{LibraryFrame: true, Filename: "foo.go"},
 						{LibraryFrame: false, AbsPath: "foobar.go"},
@@ -87,7 +87,7 @@ func TestSetLibraryFrames(t *testing.T) {
 		}},
 		output: model.Batch{{
 			Error: &model.Error{
-				Log: &model.Log{
+				Log: &model.ErrorLog{
 					Stacktrace: model.Stacktrace{
 						{LibraryFrame: true, Filename: "foo.go", Original: model.Original{LibraryFrame: true}},
 						{LibraryFrame: true, AbsPath: "foobar.go", Original: model.Original{LibraryFrame: false}},