diff --git a/internal/lambda/rapidcore/standalone/telemetry/events_api.go b/internal/lambda/rapidcore/standalone/telemetry/events_api.go index 9c4410ea..adad0084 100644 --- a/internal/lambda/rapidcore/standalone/telemetry/events_api.go +++ b/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") } diff --git a/internal/lambda/rie/handlers.go b/internal/lambda/rie/handlers.go index 1abe1541..73a5c16d 100644 --- a/internal/lambda/rie/handlers.go +++ b/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) diff --git a/internal/lambda/rie/handlers_report_test.go b/internal/lambda/rie/handlers_report_test.go new file mode 100644 index 00000000..4613dad0 --- /dev/null +++ b/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") +} diff --git a/internal/lambda/rie/http.go b/internal/lambda/rie/http.go index 4bc2d5ae..deaaf59d 100644 --- a/internal/lambda/rie/http.go +++ b/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) { diff --git a/internal/lambda/rie/run.go b/internal/lambda/rie/run.go index 4b3c9eaa..73025870 100644 --- a/internal/lambda/rie/run.go +++ b/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) diff --git a/internal/lambda/rie/telemetry_logs.go b/internal/lambda/rie/telemetry_logs.go new file mode 100644 index 00000000..adc89b90 --- /dev/null +++ b/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) diff --git a/internal/lambda/rie/telemetry_subscription.go b/internal/lambda/rie/telemetry_subscription.go new file mode 100644 index 00000000..150e20f1 --- /dev/null +++ b/internal/lambda/rie/telemetry_subscription.go @@ -0,0 +1,398 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package rie + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" + + log "github.com/sirupsen/logrus" + + "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/interop" + 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" +) + +// TelemetrySubscriptionService implements the Telemetry API for the emulator: it +// accepts subscriptions from extensions and delivers the sandbox's events to +// them in the format the real service uses. +// +// Event records are produced by the same code that already reports platform +// events, so subscribers receive the documented types (platform.initStart, +// platform.start, platform.runtimeDone, platform.report, function, extension, +// ...) rather than a single undifferentiated stream. +type TelemetrySubscriptionService struct { + lock sync.Mutex + subscriptions map[string]*subscription + metrics map[string]int + off bool + + // Events produced before any extension has subscribed, replayed to each new + // subscriber. Initialization telemetry is emitted before extensions have had + // the chance to subscribe, and is exactly what an extension reporting cold + // start needs, so dropping it would lose the most valuable records. + earlyEvents []telemetryRecord +} + +// maxEarlyEvents bounds the pre-subscription buffer, so an environment whose +// extensions never subscribe doesn't accumulate records for its whole life. +const maxEarlyEvents = 1000 + +// Defaults from the Telemetry API documentation, applied when a subscription +// omits the corresponding buffering field. +const ( + defaultTimeoutMs = 1000 + defaultMaxItems = 1000 + defaultMaxBytes = 262144 +) + +type subscribeRequest struct { + SchemaVersion string `json:"schemaVersion"` + Types []string + Buffering struct { + TimeoutMs int `json:"timeoutMs"` + MaxItems int `json:"maxItems"` + MaxBytes int64 `json:"maxBytes"` + } + Destination struct { + Protocol string `json:"protocol"` + URI string `json:"URI"` + } +} + +// telemetryRecord is one delivered event, in the shape the Telemetry API defines. +// The record is an object for platform events and a string for log lines. +type telemetryRecord struct { + Time string `json:"time"` + Type string `json:"type"` + Record interface{} `json:"record"` +} + +type subscription struct { + agentName string + destination string + types map[string]bool + timeout time.Duration + maxItems int + maxBytes int64 + + lock sync.Mutex + pending []telemetryRecord + bytes int64 + timer *time.Timer +} + +func NewTelemetrySubscriptionService() *TelemetrySubscriptionService { + return &TelemetrySubscriptionService{ + subscriptions: map[string]*subscription{}, + metrics: map[string]int{}, + } +} + +// Subscribe registers an extension's interest in some subset of telemetry. +func (s *TelemetrySubscriptionService) Subscribe(agentName string, body io.Reader, headers map[string][]string, remoteAddr string) ([]byte, int, map[string][]string, error) { + s.lock.Lock() + off := s.off + s.lock.Unlock() + if off { + return nil, http.StatusBadRequest, nil, fmt.Errorf("%s", s.GetServiceClosedErrorMessage()) + } + + raw, err := io.ReadAll(body) + if err != nil { + return nil, http.StatusInternalServerError, nil, err + } + + var request subscribeRequest + if err := json.Unmarshal(raw, &request); err != nil { + return errorResponse("InvalidRequest", "Invalid subscription request"), http.StatusBadRequest, nil, nil + } + if !strings.EqualFold(request.Destination.Protocol, "HTTP") { + return errorResponse("InvalidRequest", "Only the HTTP destination protocol is supported"), http.StatusBadRequest, nil, nil + } + destination, err := resolveDestination(request.Destination.URI) + if err != nil { + return errorResponse("InvalidRequest", "Invalid destination URI"), http.StatusBadRequest, nil, nil + } + if len(request.Types) == 0 { + return errorResponse("InvalidRequest", "At least one telemetry type is required"), http.StatusBadRequest, nil, nil + } + + sub := &subscription{ + agentName: agentName, + destination: destination, + types: map[string]bool{}, + timeout: durationOrDefault(request.Buffering.TimeoutMs, defaultTimeoutMs), + maxItems: intOrDefault(request.Buffering.MaxItems, defaultMaxItems), + maxBytes: int64OrDefault(request.Buffering.MaxBytes, defaultMaxBytes), + } + for _, name := range request.Types { + sub.types[strings.ToLower(name)] = true + } + + s.lock.Lock() + s.subscriptions[agentName] = sub + early := make([]telemetryRecord, len(s.earlyEvents)) + copy(early, s.earlyEvents) + s.lock.Unlock() + + for _, record := range early { + if sub.wants(record.Type) { + sub.enqueue(record) + } + } + + log.Infof("Telemetry API: extension %s subscribed to %v, delivering to %s", agentName, request.Types, destination) + + // The real service reports each subscription as telemetry in its own right. + s.Dispatch(standalonetelemetry.SandboxEvent{ + Time: time.Now().Format(time.RFC3339), + Type: "platform.telemetrySubscription", + PlatformEvent: map[string]interface{}{ + "name": agentName, + "state": "Subscribed", + "types": request.Types, + }, + }) + + return []byte("OK"), http.StatusOK, map[string][]string{}, nil +} + +// Dispatch queues a sandbox event for every subscriber that asked for its type. +// +// Deliberately not gated on off: the sandbox closes the subscription window as +// soon as initialization finishes, which stops new subscriptions but must leave +// delivery to existing ones running for the life of the environment. +func (s *TelemetrySubscriptionService) Dispatch(event standalonetelemetry.SandboxEvent) { + record := telemetryRecord{Time: event.Time, Type: event.Type} + if event.PlatformEvent != nil { + record.Record = event.PlatformEvent + } else { + record.Record = event.LogMessage + } + + // Buffering and delivery are chosen under a single held lock. Deciding in one + // lock section and acting in another lets a Subscribe interleave: it would + // insert itself and replay a copy of the buffer that does not yet hold this + // event, and the event would then be buffered for nobody. + s.lock.Lock() + if len(s.subscriptions) == 0 { + if len(s.earlyEvents) < maxEarlyEvents { + s.earlyEvents = append(s.earlyEvents, record) + } + s.lock.Unlock() + return + } + subscriptions := make([]*subscription, 0, len(s.subscriptions)) + for _, sub := range s.subscriptions { + subscriptions = append(subscriptions, sub) + } + s.lock.Unlock() + + for _, sub := range subscriptions { + if sub.wants(event.Type) { + sub.enqueue(record) + } + } +} + +// wants reports whether a subscription covers an event type. Subscriptions name +// categories ("platform", "function", "extension"); platform events carry a +// dotted subtype, such as platform.runtimeDone. +func (sub *subscription) wants(eventType string) bool { + category := eventType + if index := strings.IndexRune(eventType, '.'); index != -1 { + category = eventType[:index] + } + return sub.types[category] +} + +// enqueue buffers a record, flushing when the subscription's limits are reached +// and otherwise scheduling a flush for its timeout. +func (sub *subscription) enqueue(record telemetryRecord) { + sub.lock.Lock() + + sub.pending = append(sub.pending, record) + if encoded, err := json.Marshal(record); err == nil { + sub.bytes += int64(len(encoded)) + } + + if len(sub.pending) >= sub.maxItems || sub.bytes >= sub.maxBytes { + batch := sub.take() + sub.lock.Unlock() + sub.deliver(batch) + return + } + + if sub.timer == nil { + sub.timer = time.AfterFunc(sub.timeout, func() { + sub.lock.Lock() + batch := sub.take() + sub.lock.Unlock() + sub.deliver(batch) + }) + } + sub.lock.Unlock() +} + +// take removes and returns everything buffered. Callers must hold the lock. +func (sub *subscription) take() []telemetryRecord { + batch := sub.pending + sub.pending = nil + sub.bytes = 0 + if sub.timer != nil { + sub.timer.Stop() + sub.timer = nil + } + return batch +} + +func (sub *subscription) deliver(batch []telemetryRecord) { + if len(batch) == 0 { + return + } + body, err := json.Marshal(batch) + if err != nil { + log.WithError(err).Warn("Telemetry API: could not encode a batch") + return + } + response, err := deliveryClient.Post(sub.destination, "application/json", bytes.NewReader(body)) + if err != nil { + log.WithError(err).Warnf("Telemetry API: could not deliver to %s", sub.destination) + return + } + defer response.Body.Close() + io.Copy(io.Discard, response.Body) + if response.StatusCode >= 300 { + log.Warnf("Telemetry API: %s answered %s", sub.destination, response.Status) + } +} + +// Flush delivers everything buffered, so that telemetry produced late in an +// invocation isn't stranded when the sandbox is about to be frozen. +func (s *TelemetrySubscriptionService) Flush() { + s.lock.Lock() + subscriptions := make([]*subscription, 0, len(s.subscriptions)) + for _, sub := range s.subscriptions { + subscriptions = append(subscriptions, sub) + } + s.lock.Unlock() + + for _, sub := range subscriptions { + sub.lock.Lock() + batch := sub.take() + sub.lock.Unlock() + sub.deliver(batch) + } +} + +func (s *TelemetrySubscriptionService) RecordCounterMetric(metricName string, count int) { + s.lock.Lock() + defer s.lock.Unlock() + s.metrics[metricName] += count +} + +func (s *TelemetrySubscriptionService) FlushMetrics() interop.TelemetrySubscriptionMetrics { + s.lock.Lock() + defer s.lock.Unlock() + metrics := interop.TelemetrySubscriptionMetrics{} + for name, count := range s.metrics { + metrics[name] = count + } + s.metrics = map[string]int{} + return metrics +} + +// Clear drops all subscriptions, which the sandbox does when an execution +// environment is reset and its extensions are gone. +func (s *TelemetrySubscriptionService) Clear() { + s.Flush() + s.lock.Lock() + defer s.lock.Unlock() + s.subscriptions = map[string]*subscription{} + s.earlyEvents = nil + s.off = false +} + +// TurnOff closes the subscription window. Existing subscriptions keep receiving +// telemetry; only further Subscribe calls are refused. +func (s *TelemetrySubscriptionService) TurnOff() { + s.lock.Lock() + defer s.lock.Unlock() + s.off = true +} + +func (s *TelemetrySubscriptionService) GetEndpointURL() string { return telemetryEndpointPath } + +func (s *TelemetrySubscriptionService) GetServiceClosedErrorMessage() string { + return "Telemetry API is no longer accepting subscriptions" +} + +func (s *TelemetrySubscriptionService) GetServiceClosedErrorType() string { + return "Telemetry.SubscriptionClosed" +} + +const telemetryEndpointPath = "/2022-07-01/telemetry" + +// Batches that reach a size limit are delivered on the goroutine producing the +// event, so a subscriber that accepts a connection and never answers would stall +// the sandbox's event pipeline. The default client has no timeout; this one gives +// up instead. +var deliveryClient = &http.Client{Timeout: 5 * time.Second} + +// resolveDestination rewrites the hostname extensions are told to use. Inside a +// real execution environment "sandbox" resolves to the host running the runtime; +// in the emulator everything shares one network namespace. +func resolveDestination(uri string) (string, error) { + parsed, err := url.Parse(uri) + if err != nil { + return "", err + } + if parsed.Scheme != "http" { + return "", fmt.Errorf("unsupported scheme %q", parsed.Scheme) + } + if parsed.Hostname() == "sandbox" { + host := "localhost" + if port := parsed.Port(); port != "" { + host += ":" + port + } + parsed.Host = host + } + return parsed.String(), nil +} + +func errorResponse(errorType, message string) []byte { + body, _ := json.Marshal(map[string]string{"errorType": errorType, "errorMessage": message}) + return body +} + +func durationOrDefault(value, fallback int) time.Duration { + if value <= 0 { + value = fallback + } + return time.Duration(value) * time.Millisecond +} + +func intOrDefault(value, fallback int) int { + if value <= 0 { + return fallback + } + return value +} + +func int64OrDefault(value, fallback int64) int64 { + if value <= 0 { + return fallback + } + return value +} + +var _ telemetry.SubscriptionAPI = (*TelemetrySubscriptionService)(nil) diff --git a/internal/lambda/rie/telemetry_subscription_test.go b/internal/lambda/rie/telemetry_subscription_test.go new file mode 100644 index 00000000..9709f742 --- /dev/null +++ b/internal/lambda/rie/telemetry_subscription_test.go @@ -0,0 +1,367 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package rie + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + standalonetelemetry "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda/rapidcore/standalone/telemetry" +) + +// receiver stands in for a subscribed extension. +type receiver struct { + server *httptest.Server + lock sync.Mutex + events []map[string]interface{} +} + +func newReceiver(t *testing.T) *receiver { + t.Helper() + r := &receiver{} + r.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + body, err := io.ReadAll(request.Body) + if err != nil { + t.Errorf("reading delivery: %v", err) + return + } + var batch []map[string]interface{} + if err := json.Unmarshal(body, &batch); err != nil { + t.Errorf("delivered body is not a JSON array: %v (%s)", err, body) + return + } + r.lock.Lock() + r.events = append(r.events, batch...) + r.lock.Unlock() + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(r.server.Close) + return r +} + +func (r *receiver) received() []map[string]interface{} { + r.lock.Lock() + defer r.lock.Unlock() + events := make([]map[string]interface{}, len(r.events)) + copy(events, r.events) + return events +} + +// waitFor polls until the receiver has at least count events, so tests don't +// depend on delivery timing. +func (r *receiver) waitFor(t *testing.T, count int) []map[string]interface{} { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if events := r.received(); len(events) >= count { + return events + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("only %d of %d events delivered", len(r.received()), count) + return nil +} + +func subscribeBody(destination string, types []string, timeoutMs int) string { + body, _ := json.Marshal(map[string]interface{}{ + "schemaVersion": "2022-12-13", + "types": types, + "buffering": map[string]int{"timeoutMs": timeoutMs, "maxItems": 1000, "maxBytes": 262144}, + "destination": map[string]string{"protocol": "HTTP", "URI": destination}, + }) + return string(body) +} + +func subscribe(t *testing.T, service *TelemetrySubscriptionService, agent, destination string, types []string) { + t.Helper() + response, status, _, err := service.Subscribe(agent, strings.NewReader(subscribeBody(destination, types, 10)), nil, "") + require.NoError(t, err) + require.Equal(t, http.StatusOK, status, "subscribe should succeed: %s", response) +} + +func platformEvent(eventType string, record map[string]interface{}) standalonetelemetry.SandboxEvent { + return standalonetelemetry.SandboxEvent{ + Time: "2026-07-28T12:00:00Z", + Type: eventType, + PlatformEvent: record, + } +} + +func logEvent(eventType, message string) standalonetelemetry.SandboxEvent { + return standalonetelemetry.SandboxEvent{ + Time: "2026-07-28T12:00:00Z", + Type: eventType, + LogMessage: message, + } +} + +// A subscriber receives platform events as objects and log lines as strings, +// which is the distinction the Telemetry API defines and that consumers branch on. +func TestDeliversRecordsInTheDocumentedShape(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform", "function"}) + + service.Dispatch(platformEvent("platform.start", map[string]interface{}{"requestId": "abc"})) + service.Dispatch(logEvent("function", "hello from the handler")) + + // The subscription itself is reported, so expect it alongside the two events. + events := extension.waitFor(t, 3) + + byType := map[string]map[string]interface{}{} + for _, event := range events { + byType[event["type"].(string)] = event + } + + start := byType["platform.start"] + require.NotNil(t, start, "platform.start should be delivered") + record, ok := start["record"].(map[string]interface{}) + require.True(t, ok, "a platform record must be an object, got %T", start["record"]) + assert.Equal(t, "abc", record["requestId"]) + + function := byType["function"] + require.NotNil(t, function, "function telemetry should be delivered") + assert.Equal(t, "hello from the handler", function["record"], "a log record must be the line itself") + + assert.NotNil(t, byType["platform.telemetrySubscription"], "the subscription should be reported") +} + +// Subscribing to one category must not deliver the others, which is what lets an +// extension take function logs without receiving its own output back. +func TestTypeFilteringIsRespected(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"function"}) + + service.Dispatch(logEvent("function", "wanted")) + service.Dispatch(logEvent("extension", "not wanted")) + service.Dispatch(platformEvent("platform.start", map[string]interface{}{"requestId": "abc"})) + + events := extension.waitFor(t, 1) + time.Sleep(100 * time.Millisecond) // let anything unwanted arrive if it's going to + + for _, event := range extension.received() { + assert.Equal(t, "function", event["type"], "only function telemetry was subscribed to") + } + assert.Len(t, events, 1) +} + +// Initialization telemetry is produced before an extension can subscribe. Real +// Lambda delivers it anyway, so events are held and replayed to a new subscriber. +func TestEventsBeforeSubscriptionAreReplayed(t *testing.T) { + service := NewTelemetrySubscriptionService() + + service.Dispatch(platformEvent("platform.initStart", map[string]interface{}{"phase": "init"})) + + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + + events := extension.waitFor(t, 2) + types := map[string]bool{} + for _, event := range events { + types[event["type"].(string)] = true + } + assert.True(t, types["platform.initStart"], "initialization telemetry should survive until someone subscribes") +} + +// The sandbox closes the subscription window once initialization finishes. +// Delivery to those already subscribed has to continue regardless. +func TestTurnOffStopsSubscriptionsNotDelivery(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + + service.TurnOff() + + service.Dispatch(platformEvent("platform.runtimeDone", map[string]interface{}{"requestId": "abc"})) + events := extension.waitFor(t, 2) + + var delivered bool + for _, event := range events { + if event["type"] == "platform.runtimeDone" { + delivered = true + } + } + assert.True(t, delivered, "telemetry must keep flowing after the subscription window closes") + + _, status, _, err := service.Subscribe("late", strings.NewReader(subscribeBody(extension.server.URL, []string{"platform"}, 10)), nil, "") + assert.Error(t, err, "a late subscription should be refused") + assert.Equal(t, http.StatusBadRequest, status) +} + +// A reset means a fresh execution environment, so nothing should carry over. +func TestClearDropsSubscriptions(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + extension.waitFor(t, 1) + + service.Clear() + before := len(extension.received()) + service.Dispatch(platformEvent("platform.start", map[string]interface{}{"requestId": "abc"})) + time.Sleep(100 * time.Millisecond) + + assert.Equal(t, before, len(extension.received()), "a cleared service has no subscribers to deliver to") +} + +// Buffering limits are honored: reaching maxItems delivers without waiting for +// the timeout, which is what keeps a busy function's telemetry moving. +func TestMaxItemsFlushesImmediately(t *testing.T) { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + + body, _ := json.Marshal(map[string]interface{}{ + "schemaVersion": "2022-12-13", + "types": []string{"function"}, + // A long timeout, so only reaching maxItems can cause delivery. + "buffering": map[string]int{"timeoutMs": 60000, "maxItems": 2, "maxBytes": 262144}, + "destination": map[string]string{"protocol": "HTTP", "URI": extension.server.URL}, + }) + _, status, _, err := service.Subscribe("ext", strings.NewReader(string(body)), nil, "") + require.NoError(t, err) + require.Equal(t, http.StatusOK, status) + + service.Dispatch(logEvent("function", "one")) + service.Dispatch(logEvent("function", "two")) + + events := extension.waitFor(t, 2) + assert.Len(t, events, 2) +} + +func TestSubscribeRejectsUnusableRequests(t *testing.T) { + extension := newReceiver(t) + + testCases := []struct { + name string + body string + }{ + {"not JSON", `{`}, + {"no types", subscribeBody(extension.server.URL, nil, 10)}, + {"unsupported protocol", `{"types":["function"],"destination":{"protocol":"TCP","URI":"http://localhost:1"}}`}, + {"unparseable destination", `{"types":["function"],"destination":{"protocol":"HTTP","URI":"http://[::1"}}`}, + {"non-http destination", `{"types":["function"],"destination":{"protocol":"HTTP","URI":"file:///tmp/x"}}`}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + service := NewTelemetrySubscriptionService() + _, status, _, err := service.Subscribe("ext", strings.NewReader(tc.body), nil, "") + assert.NoError(t, err, "a bad request is reported by status, not by error") + assert.Equal(t, http.StatusBadRequest, status) + }) + } +} + +// Extensions are told to deliver to the "sandbox" host, which only resolves +// inside a real execution environment. +func TestSandboxHostIsRewritten(t *testing.T) { + testCases := []struct { + in string + want string + }{ + {"http://sandbox:3000", "http://localhost:3000"}, + {"http://sandbox:3000/path", "http://localhost:3000/path"}, + {"http://sandbox", "http://localhost"}, + {"http://localhost:3000", "http://localhost:3000"}, + {"http://127.0.0.1:3000", "http://127.0.0.1:3000"}, + } + + for _, tc := range testCases { + t.Run(tc.in, func(t *testing.T) { + got, err := resolveDestination(tc.in) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +// Exercises Subscribe and Dispatch concurrently, for the race detector and to +// confirm the event still arrives. +// +// This does not reproduce the lost-event window that motivated deciding +// buffer-versus-deliver under a single lock: the two lock sections it needed to +// interleave between were adjacent, and 400 attempts never hit it. The +// correctness of that arrangement rests on reading the code, not on this test. +func TestConcurrentSubscribeAndDispatch(t *testing.T) { + for attempt := 0; attempt < 100; attempt++ { + service := NewTelemetrySubscriptionService() + extension := newReceiver(t) + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + service.Dispatch(platformEvent("platform.initStart", map[string]interface{}{"phase": "init"})) + }() + go func() { + defer wg.Done() + subscribe(t, service, "ext", extension.server.URL, []string{"platform"}) + }() + wg.Wait() + + // Whichever order they interleaved in, the event is either delivered or + // still buffered for the next subscriber. It must not be stranded. + deadline := time.Now().Add(2 * time.Second) + var seen bool + for time.Now().Before(deadline) && !seen { + for _, event := range extension.received() { + if event["type"] == "platform.initStart" { + seen = true + } + } + if !seen { + time.Sleep(10 * time.Millisecond) + } + } + if !seen { + t.Fatalf("attempt %d: initStart reached no subscriber", attempt) + } + } +} + +// A subscriber that accepts a connection and never answers must not stall the +// sandbox: delivery happens on the event-producing goroutine when a batch fills. +func TestDeliveryToAHungSubscriberDoesNotBlockForever(t *testing.T) { + blocked := make(chan struct{}) + + hung := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-blocked // never answers while the test runs + })) + // Close waits for handlers in flight, so the handler has to be released + // first. Defers run last-in-first-out, hence this order. + defer hung.Close() + defer close(blocked) + + service := NewTelemetrySubscriptionService() + body, _ := json.Marshal(map[string]interface{}{ + "types": []string{"function"}, + "buffering": map[string]int{"timeoutMs": 60000, "maxItems": 1, "maxBytes": 262144}, + "destination": map[string]string{"protocol": "HTTP", "URI": hung.URL}, + }) + _, status, _, err := service.Subscribe("ext", strings.NewReader(string(body)), nil, "") + require.NoError(t, err) + require.Equal(t, http.StatusOK, status) + + // maxItems of 1 means this dispatch delivers synchronously. + done := make(chan struct{}) + go func() { + service.Dispatch(logEvent("function", "hello")) + close(done) + }() + + select { + case <-done: + case <-time.After(deliveryClient.Timeout + 5*time.Second): + t.Fatal("Dispatch never returned; a hung subscriber can stall the event pipeline") + } +}