feat: implement the Telemetry API

internal/lambda/rapidcore/standalone/telemetry/events_api.go

@@ -61,10 +61,30 @@ type tailLogs struct {
 	Events []SandboxEvent `json:"events,omitempty"`
 }
 
+// EventDispatcher receives every sandbox event as it is produced, so that
+// subscribers to the Telemetry API can be sent the same records the event log
+// records.
+type EventDispatcher interface {
+	Dispatch(event SandboxEvent)
+}
+
 type StandaloneEventsAPI struct {
-	lock      sync.Mutex
-	requestID interop.RequestID
-	eventLog  EventLog
+	lock       sync.Mutex
+	requestID  interop.RequestID
+	eventLog   EventLog
+	dispatcher EventDispatcher
+	// quiet suppresses the per-event log line. The emulator already reports the
+	// lifecycle in its own output, so repeating every event there is noise.
+	quiet bool
+}
+
+// SetDispatcher attaches a dispatcher, and stops each event being logged
+// individually since the dispatcher is now responsible for reporting them.
+func (s *StandaloneEventsAPI) SetDispatcher(dispatcher EventDispatcher) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.dispatcher = dispatcher
+	s.quiet = true
 }
 
 func (s *StandaloneEventsAPI) LogTrace(entry TracingEvent) {
@@ -275,11 +295,22 @@ func (s *StandaloneEventsAPI) sendLogEvent(eventType, logMessage string) error {
 
 func (s *StandaloneEventsAPI) appendEvent(event SandboxEvent) {
 	s.lock.Lock()
-	defer s.lock.Unlock()
 	s.eventLog.Events = append(s.eventLog.Events, event)
+	dispatcher := s.dispatcher
+	s.lock.Unlock()
+
+	if dispatcher != nil {
+		dispatcher.Dispatch(event)
+	}
 }
 
 func (s *StandaloneEventsAPI) logEvent(e SandboxEvent) {
+	s.lock.Lock()
+	quiet := s.quiet
+	s.lock.Unlock()
+	if quiet {
+		return
+	}
 	log.WithField("event", e).Info("sandbox event")
 }
 

internal/lambda/rie/handlers.go

@@ -24,6 +24,8 @@ import (
 	"github.com/google/uuid"
 
 	log "github.com/sirupsen/logrus"
+
+	standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry"
 )
 
 type Sandbox interface {
@@ -46,6 +48,16 @@ type InteropServer interface {
 
 var initDone bool
 
+// telemetryEvents, when set, receives the end-of-invoke events. The emulator
+// prints END and REPORT but has never reported them as telemetry, so extensions
+// that measure billed duration or memory had nothing to read.
+var telemetryEvents *TelemetrySubscriptionService
+
+// SetTelemetryEvents attaches the service that end-of-invoke events are reported to.
+func SetTelemetryEvents(service *TelemetrySubscriptionService) {
+	telemetryEvents = service
+}
+
 func GetenvWithDefault(key string, defaultValue string) string {
 	envValue := os.Getenv(key)
 
@@ -56,12 +68,54 @@ func GetenvWithDefault(key string, defaultValue string) string {
 	return envValue
 }
 
-func printEndReports(invokeId string, initDuration string, memorySize string, invokeStart time.Time, timeoutDuration time.Duration) {
+// reportRecord builds the platform.report record. The metrics are numbers, as the
+// Telemetry API defines them: a consumer decoding into a typed struct rejects the
+// strings these values arrive as.
+func reportRecord(invokeId string, status string, invokeDuration float64, memorySize string, initDurationMs float64) map[string]interface{} {
+	// The emulator cannot measure memory actually used, so it reports the
+	// configured size for both, as the printed REPORT line does.
+	memorySizeMB, err := strconv.Atoi(memorySize)
+	if err != nil {
+		log.Warnf("AWS_LAMBDA_FUNCTION_MEMORY_SIZE is %q, which is not a number", memorySize)
+	}
+
+	metrics := map[string]interface{}{
+		"durationMs":       invokeDuration,
+		"billedDurationMs": math.Ceil(invokeDuration),
+		"memorySizeMB":     memorySizeMB,
+		"maxMemoryUsedMB":  memorySizeMB,
+	}
+	// Present only on the report for an invocation that initialized the
+	// environment, as it is in telemetry from a real function.
+	if initDurationMs > 0 {
+		metrics["initDurationMs"] = initDurationMs
+	}
+
+	return map[string]interface{}{
+		"requestId": invokeId,
+		"status":    status,
+		"metrics":   metrics,
+	}
+}
+
+// printEndReports reports the end of an invocation. status is what the Telemetry
+// API calls it: "success", or "timeout" when the invocation ran out of time.
+func printEndReports(invokeId string, initDuration string, memorySize string, invokeStart time.Time, timeoutDuration time.Duration, status string, initDurationMs float64) {
 	// Calcuation invoke duration
 	invokeDuration := math.Min(float64(time.Now().Sub(invokeStart).Nanoseconds()),
 		float64(timeoutDuration.Nanoseconds())) / float64(time.Millisecond)
 
 	fmt.Println("END RequestId: " + invokeId)
+
+	if telemetryEvents != nil {
+		// platform.end is deliberately not emitted: it is absent from telemetry
+		// captured off a real function under the current schema version.
+		telemetryEvents.Dispatch(standalonetelemetry.SandboxEvent{
+			Time:          time.Now().Format(time.RFC3339),
+			Type:          "platform.report",
+			PlatformEvent: reportRecord(invokeId, status, invokeDuration, memorySize, initDurationMs),
+		})
+	}
 	// We set the Max Memory Used and Memory Size to be the same (whatever it is set to) since there is
 	// not a clean way to get this information from rapidcore
 	fmt.Printf(
@@ -91,6 +145,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i
 	}
 
 	initDuration := ""
+	initDurationMs := float64(0)
 	inv := GetenvWithDefault("AWS_LAMBDA_FUNCTION_TIMEOUT", "300")
 	timeoutDuration, _ := time.ParseDuration(inv + "s")
 	// Default
@@ -111,6 +166,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i
 			float64(timeoutDuration.Nanoseconds())) / float64(time.Millisecond)
 
 		initDuration = fmt.Sprintf("Init Duration: %.2f ms\t", initTimeMS)
+		initDurationMs = initTimeMS
 
 		// Set initDone so next invokes do not try to Init the function again
 		initDone = true
@@ -197,7 +253,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i
 			w.WriteHeader(http.StatusGatewayTimeout)
 			return
 		case rapidcore.ErrInvokeTimeout:
-			printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration)
+			printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration, "timeout", initDurationMs)
 
 			w.Write([]byte(fmt.Sprintf("Task timed out after %d.00 seconds", timeout)))
 			time.Sleep(100 * time.Millisecond)
@@ -206,7 +262,7 @@ func InvokeHandler(w http.ResponseWriter, r *http.Request, sandbox Sandbox, bs i
 		}
 	}
 
-	printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration)
+	printEndReports(invokePayload.ID, initDuration, memorySize, invokeStart, timeoutDuration, "success", initDurationMs)
 
 	if invokeResp.StatusCode != 0 {
 		w.WriteHeader(invokeResp.StatusCode)

internal/lambda/rie/handlers_report_test.go

@@ -0,0 +1,66 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package rie
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The Telemetry API defines these metrics as numbers. A consumer decoding into a
+// typed struct rejects the strings the memory size arrives as, so the record has
+// to carry numbers however the value reached us.
+func TestReportRecordMetricsAreNumbers(t *testing.T) {
+	encoded, err := json.Marshal(reportRecord("abc", "success", 12.5, "512", 0))
+	require.NoError(t, err)
+
+	var decoded struct {
+		RequestID string `json:"requestId"`
+		Status    string `json:"status"`
+		Metrics   struct {
+			DurationMs       float64 `json:"durationMs"`
+			BilledDurationMs float64 `json:"billedDurationMs"`
+			MemorySizeMB     int     `json:"memorySizeMB"`
+			MaxMemoryUsedMB  int     `json:"maxMemoryUsedMB"`
+		} `json:"metrics"`
+	}
+	require.NoError(t, json.Unmarshal(encoded, &decoded), "a typed consumer must be able to decode this")
+
+	assert.Equal(t, "abc", decoded.RequestID)
+	assert.Equal(t, "success", decoded.Status)
+	assert.Equal(t, 12.5, decoded.Metrics.DurationMs)
+	assert.Equal(t, float64(13), decoded.Metrics.BilledDurationMs)
+	assert.Equal(t, 512, decoded.Metrics.MemorySizeMB)
+}
+
+// An invocation that ran out of time is not a successful one. Extensions alarm on
+// this field, so reporting a timeout as a success would hide exactly the failure
+// they are watching for.
+func TestReportRecordCarriesTheInvocationStatus(t *testing.T) {
+	for _, status := range []string{"success", "timeout"} {
+		t.Run(status, func(t *testing.T) {
+			assert.Equal(t, status, reportRecord("abc", status, 1, "128", 0)["status"])
+		})
+	}
+}
+
+// A memory size that isn't a number should not make the record undecodable.
+func TestReportRecordSurvivesAnUnparseableMemorySize(t *testing.T) {
+	record := reportRecord("abc", "success", 1, "not-a-number", 0)
+	metrics := record["metrics"].(map[string]interface{})
+	assert.Equal(t, 0, metrics["memorySizeMB"])
+}
+
+// A cold start reports how long initialization took; a warm invocation has no
+// such figure and must not report a zero one.
+func TestReportRecordIncludesInitDurationOnlyOnColdStart(t *testing.T) {
+	cold := reportRecord("abc", "success", 1, "128", 120.5)["metrics"].(map[string]interface{})
+	assert.Equal(t, 120.5, cold["initDurationMs"])
+
+	warm := reportRecord("abc", "success", 1, "128", 0)["metrics"].(map[string]interface{})
+	assert.NotContains(t, warm, "initDurationMs")
+}

internal/lambda/rie/http.go

@@ -6,9 +6,9 @@ package rie
 import (
 	"net/http"
 
-	log "github.com/sirupsen/logrus"
 	"github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop"
 	"github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore"
+	log "github.com/sirupsen/logrus"
 )
 
 func startHTTPServer(ipport string, sandbox *rapidcore.SandboxBuilder, bs interop.Bootstrap) {

internal/lambda/rie/run.go

@@ -10,10 +10,11 @@ import (
 	"os"
 	"runtime/debug"
 
-	"github.com/jessevdk/go-flags"
 	"github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop"
 	"github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore"
+	"github.com/jessevdk/go-flags"
 
+	standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -64,11 +65,23 @@ func Run() {
 	}
 
 	bootstrap, handler := getBootstrap(args, opts)
+
+	// Serve the Telemetry API rather than the stub. Events are produced by the
+	// same code that reports the sandbox lifecycle, so subscribers receive the
+	// documented record types; log lines still reach the console as well.
+	eventsAPI := &standalonetelemetry.StandaloneEventsAPI{}
+	telemetrySubscriptions := NewTelemetrySubscriptionService()
+	eventsAPI.SetDispatcher(telemetrySubscriptions)
+	SetTelemetryEvents(telemetrySubscriptions)
+
 	sandbox := rapidcore.
 		NewSandboxBuilder().
 		AddShutdownFunc(context.CancelFunc(func() { os.Exit(0) })).
 		SetExtensionsFlag(true).
-		SetInitCachingFlag(opts.InitCachingEnabled)
+		SetInitCachingFlag(opts.InitCachingEnabled).
+		SetEventsAPI(eventsAPI).
+		SetLogsEgressAPI(newTeeLogsEgressAPI(eventsAPI)).
+		SetTelemetrySubscription(telemetrySubscriptions, telemetrySubscriptions)
 
 	if len(handler) > 0 {
 		sandbox.SetHandler(handler)

internal/lambda/rie/telemetry_logs.go

@@ -0,0 +1,46 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package rie
+
+import (
+	"io"
+	"os"
+
+	standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry"
+	"github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/telemetry"
+)
+
+// teeLogsEgressAPI sends the runtime's and extensions' output to the console, as
+// the emulator has always done, and additionally reports each line as a telemetry
+// event so that subscribers can receive it.
+//
+// The two sources are reported separately, as the Telemetry API defines them:
+// runtime output is "function" telemetry and an extension's own output is
+// "extension" telemetry. Keeping them distinct is what lets an extension
+// subscribe to function logs without being sent its own.
+type teeLogsEgressAPI struct {
+	eventsAPI *standalonetelemetry.StandaloneEventsAPI
+}
+
+func newTeeLogsEgressAPI(eventsAPI *standalonetelemetry.StandaloneEventsAPI) *teeLogsEgressAPI {
+	return &teeLogsEgressAPI{eventsAPI: eventsAPI}
+}
+
+func (t *teeLogsEgressAPI) GetExtensionSockets() (io.Writer, io.Writer, error) {
+	writer := t.writerFor("extension")
+	return writer, writer, nil
+}
+
+func (t *teeLogsEgressAPI) GetRuntimeSockets() (io.Writer, io.Writer, error) {
+	writer := t.writerFor("function")
+	return writer, writer, nil
+}
+
+// writerFor tees to stdout so console output is unchanged. os.Stderr is not used
+// for either source, because stderr carries the emulator's own logging.
+func (t *teeLogsEgressAPI) writerFor(source string) io.Writer {
+	return io.MultiWriter(os.Stdout, standalonetelemetry.NewSandboxAgentWriter(t.eventsAPI, source))
+}
+
+var _ telemetry.StdLogsEgressAPI = (*teeLogsEgressAPI)(nil)