fix merge conflicts

Author: herin049

collector/receiver/telemetryapireceiver/receiver.go

@@ -28,6 +28,8 @@ import (
 	"time"
 
 	"github.com/golang-collections/go-datastructures/queue"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/telemetryapi"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -37,8 +39,6 @@ import (
 	"go.opentelemetry.io/collector/receiver"
 	semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
 	"go.uber.org/zap"
-
-	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/telemetryapi"
 )
 
 const (
@@ -48,7 +48,7 @@ const (
 	telemetryFailureStatus              = "failure"
 	telemetryErrorStatus                = "error"
 	telemetryTimeoutStatus              = "timeout"
-	platformReportLogFmt                = "REPORT RequestId: %s Duration: %.2f ms Billed Duration: %.0f ms Memory Size: %.0f MB Max Memory Used: %.0f MB"
+	platformReportLogFmt                = "REPORT RequestId: %s"
 	platformStartLogFmt                 = "START RequestId: %s Version: %s"
 	platformRuntimeDoneLogFmt           = "END RequestId: %s Version: %s"
 	platformInitStartLogFmt             = "INIT_START Runtime Version: %s Runtime Version ARN: %s"
@@ -80,6 +80,7 @@ type telemetryAPIReceiver struct {
 	faasName                string
 	faaSMetricBuilders      *FaaSMetricBuilders
 	currentFaasInvocationID string
+	lambdaInitType          lambdalifecycle.InitType
 	logReport               bool
 }
 
@@ -148,8 +149,10 @@ func (r *telemetryAPIReceiver) httpHandler(w http.ResponseWriter, req *http.Requ
 		switch el.Type {
 		// Function initialization started.
 		case string(telemetryapi.PlatformInitStart):
-			r.logger.Info(fmt.Sprintf("Init start: %s", r.lastPlatformStartTime), zap.Any("event", el))
-			r.lastPlatformStartTime = el.Time
+			if el.Time != "" {
+				r.lastPlatformStartTime = el.Time
+				r.logger.Info(fmt.Sprintf("Init start: %s", r.lastPlatformStartTime), zap.Any("event", el))
+			}
 
 			if record, ok := el.Record.(map[string]any); ok {
 				functionName, _ := record["functionName"].(string)
@@ -159,8 +162,10 @@ func (r *telemetryAPIReceiver) httpHandler(w http.ResponseWriter, req *http.Requ
 			}
 		// Function initialization completed.
 		case string(telemetryapi.PlatformInitRuntimeDone):
-			r.logger.Info(fmt.Sprintf("Init end: %s", r.lastPlatformEndTime), zap.Any("event", el))
-			r.lastPlatformEndTime = el.Time
+			if r.lastPlatformStartTime != "" && el.Time != "" {
+				r.lastPlatformEndTime = el.Time
+				r.logger.Info(fmt.Sprintf("Init end: %s", r.lastPlatformEndTime), zap.Any("event", el))
+			}
 
 			if len(r.lastPlatformStartTime) > 0 && len(r.lastPlatformEndTime) > 0 {
 				if record, ok := el.Record.(map[string]any); ok {
@@ -175,6 +180,11 @@ func (r *telemetryAPIReceiver) httpHandler(w http.ResponseWriter, req *http.Requ
 					}
 				}
 			}
+		case string(telemetryapi.PlatformInitReport):
+			if r.lastPlatformStartTime != "" && el.Time != "" {
+				r.lastPlatformEndTime = el.Time
+				r.logger.Info(fmt.Sprintf("Init end: %s", r.lastPlatformEndTime), zap.Any("event", el))
+			}
 		}
 		// TODO: add support for additional events, see https://docs.aws.amazon.com/lambda/latest/dg/telemetry-api.html
 		// A report of function initialization.
@@ -225,14 +235,27 @@ func (r *telemetryAPIReceiver) httpHandler(w http.ResponseWriter, req *http.Requ
 }
 
 func (r *telemetryAPIReceiver) getRecordRequestId(record map[string]interface{}) string {
-	if requestId, ok := record["requestId"].(string); ok {
-		return requestId
-	} else if r.currentFaasInvocationID != "" {
+	if record != nil {
+		if requestId, ok := record["requestId"].(string); ok {
+			return requestId
+		}
+	}
+	return ""
+}
+
+func (r *telemetryAPIReceiver) getCurrentRequestId() string {
+	if r.lambdaInitType != lambdalifecycle.LambdaManagedInstances {
 		return r.currentFaasInvocationID
 	}
 	return ""
 }
 
+func (r *telemetryAPIReceiver) updateCurrentRequestId(requestId string) {
+	if r.lambdaInitType != lambdalifecycle.LambdaManagedInstances {
+		r.currentFaasInvocationID = requestId
+	}
+}
+
 func (r *telemetryAPIReceiver) createMetrics(slice []event) (pmetric.Metrics, error) {
 	metric := pmetric.NewMetrics()
 	resourceMetric := metric.ResourceMetrics().AppendEmpty()
@@ -340,6 +363,17 @@ func (r *telemetryAPIReceiver) createLogs(slice []event) (plog.Logs, error) {
 		}
 		if record, ok := el.Record.(map[string]interface{}); ok {
 			requestId := r.getRecordRequestId(record)
+
+			// If this is the first event in the invocation with a request id (i.e. the "platform.start" event),
+			// set the current invocation id to this request id.
+			if requestId != "" && el.Type == string(telemetryapi.PlatformStart) {
+				r.updateCurrentRequestId(requestId)
+			}
+
+			if requestId == "" {
+				requestId = r.getCurrentRequestId()
+			}
+
 			if requestId != "" {
 				logRecord.Attributes().PutStr(string(semconv.FaaSInvocationIDKey), requestId)
 
@@ -370,6 +404,10 @@ func (r *telemetryAPIReceiver) createLogs(slice []event) (plog.Logs, error) {
 					if functionVersion != "" {
 						r.faasFunctionVersion = functionVersion
 					}
+				} else if el.Type == string(telemetryapi.PlatformStart) {
+					if version, _ := record["version"].(string); version != "" {
+						r.faasFunctionVersion = version
+					}
 				}
 
 				message := createPlatformMessage(requestId, r.faasFunctionVersion, el.Type, record)
@@ -380,16 +418,22 @@ func (r *telemetryAPIReceiver) createLogs(slice []event) (plog.Logs, error) {
 				logRecord.Body().SetStr(line)
 			}
 		} else {
-			if r.currentFaasInvocationID != "" {
-				logRecord.Attributes().PutStr(string(semconv.FaaSInvocationIDKey), r.currentFaasInvocationID)
+			requestId := r.getRecordRequestId(nil)
+			if requestId == "" {
+				requestId = r.getCurrentRequestId()
 			}
+
+			if requestId != "" {
+				logRecord.Attributes().PutStr(string(semconv.FaaSInvocationIDKey), requestId)
+			}
+
 			// 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)
 			}
 		}
 		if el.Type == string(telemetryapi.PlatformRuntimeDone) {
-			r.currentFaasInvocationID = ""
+			r.updateCurrentRequestId("")
 		}
 	}
 	return log, nil
@@ -480,44 +524,34 @@ func createPlatformMessage(requestId string, functionVersion string, eventType s
 	}
 	return ""
 }
-
 func createPlatformReportMessage(requestId string, record map[string]interface{}) string {
-	// gathering metrics
-	metrics, ok := record["metrics"].(map[string]interface{})
-	if !ok {
+	if requestId == "" {
 		return ""
 	}
-	var durationMs, billedDurationMs, memorySizeMB, maxMemoryUsedMB float64
-	if durationMs, ok = metrics[string(telemetryapi.MetricDurationMs)].(float64); !ok {
-		return ""
+
+	message := fmt.Sprintf(platformReportLogFmt, requestId)
+	metrics, ok := record["metrics"].(map[string]interface{})
+	if !ok {
+		return message
 	}
-	if billedDurationMs, ok = metrics[string(telemetryapi.MetricBilledDurationMs)].(float64); !ok {
-		return ""
+
+	if durationMs, ok := metrics["durationMs"].(float64); ok {
+		message += fmt.Sprintf(" Duration: %.2f ms", durationMs)
 	}
-	if memorySizeMB, ok = metrics[string(telemetryapi.MetricMemorySizeMB)].(float64); !ok {
-		return ""
+
+	if billedDurationMs, ok := metrics["billedDurationMs"].(float64); ok {
+		message += fmt.Sprintf(" Billed Duration: %.0f ms", billedDurationMs)
 	}
-	if maxMemoryUsedMB, ok = metrics[string(telemetryapi.MetricMaxMemoryUsedMB)].(float64); !ok {
-		return ""
+
+	if memorySizeMB, ok := metrics["memorySizeMB"].(float64); ok {
+		message += fmt.Sprintf(" Memory Size: %.0f MB", memorySizeMB)
 	}
 
-	// optionally gather information about cold start time
-	var initDurationMs float64
-	if initDurationMsVal, exists := metrics[string(telemetryapi.MetricInitDurationMs)]; exists {
-		if val, ok := initDurationMsVal.(float64); ok {
-			initDurationMs = val
-		}
+	if maxMemoryUsedMB, ok := metrics["maxMemoryUsedMB"].(float64); ok {
+		message += fmt.Sprintf(" Max Memory Used: %.0f MB", maxMemoryUsedMB)
 	}
 
-	message := fmt.Sprintf(
-		platformReportLogFmt,
-		requestId,
-		durationMs,
-		billedDurationMs,
-		memorySizeMB,
-		maxMemoryUsedMB,
-	)
-	if initDurationMs > 0 {
+	if initDurationMs, ok := metrics["initDurationMs"].(float64); ok {
 		message += fmt.Sprintf(" Init Duration: %.2f ms", initDurationMs)
 	}
 
@@ -666,6 +700,8 @@ func newTelemetryAPIReceiver(
 		}
 	}
 
+	lambdaInitType := lambdalifecycle.InitTypeFromEnv(lambdalifecycle.InitTypeEnvVar)
+
 	return &telemetryAPIReceiver{
 		logger:             set.Logger,
 		queue:              queue.New(initialQueueSize),
@@ -674,6 +710,7 @@ func newTelemetryAPIReceiver(
 		types:              subscribedTypes,
 		resource:           r,
 		faaSMetricBuilders: NewFaaSMetricBuilders(pcommon.NewTimestampFromTime(time.Now()), getMetricsTemporality(cfg)),
+		lambdaInitType:     lambdaInitType,
 		logReport:          cfg.LogReport,
 	}, nil
 }