add invocation ids for plain text logs

Author: adamantal

collector/internal/telemetryapi/types.go

@@ -24,6 +24,10 @@ const (
 	PlatformInitStart EventType = Platform + ".initStart"
 	// PlatformInitRuntimeDone is used when function initialization ended.
 	PlatformInitRuntimeDone EventType = Platform + ".initRuntimeDone"
+	// Function invocation started.
+	PlatformStart EventType = Platform + ".start"
+	// The runtime finished processing an event with either success or failure.
+	PlatformRuntimeDone EventType = Platform + ".runtimeDone"
 	// Function is used to receive log events emitted by the function
 	Function EventType = "function"
 	// Extension is used is to receive log events emitted by the extension

collector/receiver/telemetryapireceiver/receiver.go

@@ -45,17 +45,18 @@ const initialQueueSize = 5
 const scopeName = "github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapi"
 
 type telemetryAPIReceiver struct {
-	httpServer            *http.Server
-	logger                *zap.Logger
-	queue                 *queue.Queue // queue is a synchronous queue and is used to put the received log events to be dispatched later
-	nextTraces            consumer.Traces
-	nextLogs              consumer.Logs
-	lastPlatformStartTime string
-	lastPlatformEndTime   string
-	extensionID           string
-	port                  int
-	types                 []telemetryapi.EventType
-	resource              pcommon.Resource
+	httpServer              *http.Server
+	logger                  *zap.Logger
+	queue                   *queue.Queue // queue is a synchronous queue and is used to put the received log events to be dispatched later
+	nextTraces              consumer.Traces
+	nextLogs                consumer.Logs
+	lastPlatformStartTime   string
+	lastPlatformEndTime     string
+	extensionID             string
+	port                    int
+	types                   []telemetryapi.EventType
+	resource                pcommon.Resource
+	currentFaasInvocationID string
 }
 
 func (r *telemetryAPIReceiver) Start(ctx context.Context, host component.Host) error {
@@ -217,16 +218,31 @@ func (r *telemetryAPIReceiver) createLogs(slice []event) (plog.Logs, error) {
 				}
 				if requestId, ok := record["requestId"].(string); ok {
 					logRecord.Attributes().PutStr(semconv.AttributeFaaSInvocationID, requestId)
+				} else if r.currentFaasInvocationID != "" {
+					logRecord.Attributes().PutStr(semconv.AttributeFaaSInvocationID, r.currentFaasInvocationID)
 				}
 				if line, ok := record["message"].(string); ok {
 					logRecord.Body().SetStr(line)
 				}
 			} else {
+				if r.currentFaasInvocationID != "" {
+					logRecord.Attributes().PutStr(semconv.AttributeFaaSInvocationID, r.currentFaasInvocationID)
+				}
 				// in plain text https://docs.aws.amazon.com/lambda/latest/dg/telemetry-schema-reference.html#telemetry-api-function
 				if line, ok := el.Record.(string); ok {
 					logRecord.Body().SetStr(line)
 				}
 			}
+		} else { // platform events, if subscribed to
+			if el.Type == string(telemetryapi.PlatformStart) {
+				if record, ok := el.Record.(map[string]interface{}); ok {
+					if requestId, ok := record["requestId"].(string); ok {
+						r.currentFaasInvocationID = requestId
+					}
+				}
+			} else if el.Type == string(telemetryapi.PlatformRuntimeDone) {
+				r.currentFaasInvocationID = ""
+			}
 		}
 	}
 	return log, nil

collector/receiver/telemetryapireceiver/receiver_test.go

@@ -221,6 +221,37 @@ func TestCreateLogs(t *testing.T) {
 			expectedSeverityNumber:    plog.SeverityNumberUnspecified,
 			expectError:               false,
 		},
+		{
+			desc: "function text with requestId",
+			slice: []event{
+				{
+					Time: "2022-10-12T00:03:50.000Z",
+					Type: "platform.start",
+					Record: map[string]any{
+						"requestId": "34472c47-5ff0-4df5-a9ad-03776afa5473",
+					},
+				},
+				{
+					Time:   "2022-10-12T00:03:50.000Z",
+					Type:   "function",
+					Record: "[INFO] Hello world, I am an extension!",
+				},
+				{
+					Time:   "2022-10-12T00:03:50.000Z",
+					Type:   "platform.runtimeDone",
+					Record: map[string]any{},
+				},
+			},
+			expectedLogRecords:        1,
+			expectedType:              "function",
+			expectedTimestamp:         "2022-10-12T00:03:50.000Z",
+			expectedBody:              "[INFO] Hello world, I am an extension!",
+			expectedContainsRequestId: true,
+			expectedRequestId:         "34472c47-5ff0-4df5-a9ad-03776afa5473",
+			expectedSeverityText:      "",
+			expectedSeverityNumber:    plog.SeverityNumberUnspecified,
+			expectError:               false,
+		},
 		{
 			desc: "function json",
 			slice: []event{
@@ -351,9 +382,11 @@ func TestCreateLogs(t *testing.T) {
 			desc: "platform.start anything",
 			slice: []event{
 				{
-					Time:   "2022-10-12T00:03:50.000Z",
-					Type:   "platform.start",
-					Record: map[string]any{},
+					Time: "2022-10-12T00:03:50.000Z",
+					Type: "platform.start",
+					Record: map[string]any{
+						"requestId": "test-id",
+					},
 				},
 			},
 			expectedLogRecords: 0,