update unit tests

Author: herin049

collector/internal/telemetryapi/listener_test.go

@@ -45,11 +45,6 @@ func setupListener(t *testing.T) (*Listener, string) {
 
 	address, err := listener.Start()
 	require.NoError(t, err)
-
-	t.Cleanup(func() {
-		listener.Shutdown()
-	})
-
 	return listener, address
 }
 
@@ -183,6 +178,7 @@ func TestListenOnAddress(t *testing.T) {
 
 func TestListener_StartAndShutdown(t *testing.T) {
 	listener, address := setupListener(t)
+	defer listener.Shutdown()
 	require.NotEqual(t, address, "", "Start() should not return an empty address")
 	require.True(t, strings.HasPrefix(address, "http://"), "Address should start with http://")
 	require.NotNil(t, listener.httpServer, "httpServer should not be nil")
@@ -241,6 +237,7 @@ func TestListener_httpHandler(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 			listener, address := setupListener(t)
+			defer listener.Shutdown()
 			submitEvents(t, address, test.events)
 			require.EventuallyWithT(t, func(c *assert.CollectT) {
 				require.Equal(c, test.expectedCount, listener.queue.Len())
@@ -302,6 +299,7 @@ func TestListener_Wait_Success(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 			listener, address := setupListener(t)
+			defer listener.Shutdown()
 
 			waitDone := make(chan error, 1)
 			go func() {

collector/receiver/telemetryapireceiver/config.go

@@ -23,7 +23,7 @@ type Config struct {
 	extensionID string
 	Port        int      `mapstructure:"port"`
 	Types       []string `mapstructure:"types"`
-	LogReport   bool     `mapstructure:"log_report"`
+	LogReport   *bool    `mapstructure:"log_report"`
 }
 
 // Validate validates the configuration by checking for missing or invalid fields

collector/receiver/telemetryapireceiver/receiver.go

@@ -205,6 +205,9 @@ func (r *telemetryAPIReceiver) createLogs(slice []event) (plog.Logs, error) {
 	scopeLog.Scope().SetName(scopeName)
 	for _, el := range slice {
 		r.logger.Debug(fmt.Sprintf("Event: %s", el.Type), zap.Any("event", el))
+		if !r.logReport && el.Type == string(telemetryapi.PlatformReport) {
+			continue
+		}
 		logRecord := scopeLog.LogRecords().AppendEmpty()
 		logRecord.Attributes().PutStr("type", el.Type)
 		if t, err := time.Parse(time.RFC3339, el.Time); err == nil {
@@ -215,6 +218,17 @@ func (r *telemetryAPIReceiver) createLogs(slice []event) (plog.Logs, error) {
 			return plog.Logs{}, err
 		}
 		if record, ok := el.Record.(map[string]interface{}); ok {
+			requestId := r.getRecordRequestId(record)
+			if requestId != "" {
+				logRecord.Attributes().PutStr(semconv.AttributeFaaSInvocationID, requestId)
+
+				// 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 el.Type == string(telemetryapi.PlatformStart) {
+					r.currentFaasInvocationID = requestId
+				}
+			}
+
 			// in JSON format https://docs.aws.amazon.com/lambda/latest/dg/telemetry-schema-reference.html#telemetry-api-function
 			if timestamp, ok := record["timestamp"].(string); ok {
 				if t, err := time.Parse(time.RFC3339, timestamp); err == nil {
@@ -229,24 +243,13 @@ func (r *telemetryAPIReceiver) createLogs(slice []event) (plog.Logs, error) {
 				logRecord.SetSeverityText(logRecord.SeverityNumber().String())
 			}
 
-			requestId := r.getRecordRequestId(record)
-			if requestId != "" {
-				logRecord.Attributes().PutStr(semconv.AttributeFaaSInvocationID, requestId)
-
-				// 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 el.Type == string(telemetryapi.PlatformStart) {
-					r.currentFaasInvocationID = requestId
-				}
-			}
-
-			if line, ok := record["message"].(string); ok {
-				logRecord.Body().SetStr(line)
-			} else if el.Type == string(telemetryapi.PlatformReport) {
+			if el.Type == string(telemetryapi.PlatformReport) {
 				platformReportMessage := createPlatformReportMessage(requestId, record)
 				if platformReportMessage != "" {
 					logRecord.Body().SetStr(platformReportMessage)
 				}
+			} else if line, ok := record["message"].(string); ok {
+				logRecord.Body().SetStr(line)
 			}
 		} else {
 			if r.currentFaasInvocationID != "" {
@@ -406,7 +409,7 @@ func newTelemetryAPIReceiver(
 		}
 	}
 
-	subscribedTypes := []telemetryapi.EventType{}
+	var subscribedTypes []telemetryapi.EventType
 	for _, val := range cfg.Types {
 		switch val {
 		case "platform":
@@ -418,14 +421,19 @@ func newTelemetryAPIReceiver(
 		}
 	}
 
+	logReport := true
+	if cfg.LogReport != nil {
+		logReport = *cfg.LogReport
+	}
+
 	return &telemetryAPIReceiver{
 		logger:      set.Logger,
 		queue:       queue.New(initialQueueSize),
 		extensionID: cfg.extensionID,
 		port:        cfg.Port,
 		types:       subscribedTypes,
 		resource:    r,
-		logReport:   cfg.LogReport,
+		logReport:   logReport,
 	}, nil
 }