lambda/interop/sandbox_model.go

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0

package interop

import (
	"bytes"
	"io"
	"net/http"
	"time"

	"go.amzn.com/lambda/fatalerror"
	"go.amzn.com/lambda/rapidcore/env"
)

// Init represents an init message
// In Rapid Shim, this is a START GirD message
// In Rapid Daemon, this is an INIT GirP message
type Init struct {
	InvokeID          string
	Handler           string
	AccountID         string
	AwsKey            string
	AwsSecret         string
	AwsSession        string
	CredentialsExpiry time.Time
	SuppressInit      bool
	InvokeTimeoutMs   int64  // timeout duration of whole invoke
	InitTimeoutMs     int64  // timeout duration for init only
	XRayDaemonAddress string // only in standalone
	FunctionName      string // only in standalone
	FunctionVersion   string // only in standalone
	// In standalone mode, these env vars come from test/init but from environment otherwise.
	CustomerEnvironmentVariables map[string]string
	SandboxType                  SandboxType
	LogStreamName                string
	InstanceMaxMemory            uint64
	OperatorDomainExtraConfig    DynamicDomainConfig
	RuntimeDomainExtraConfig     DynamicDomainConfig
	RuntimeInfo                  RuntimeInfo
	Bootstrap                    Bootstrap
	EnvironmentVariables         *env.Environment // contains env vars for agents and runtime procs
}

// InitSuccess indicates that runtime/extensions initialization completed successfully
// In Rapid Shim, this translates to a DONE GirD message to Slicer
// In Rapid Daemon, this is followed by a DONEDONE GirP message to MM
type InitSuccess struct {
	NumActiveExtensions int    // indicates number of active extensions
	ExtensionNames      string // file names of extensions in /opt/extensions
	RuntimeRelease      string
	LogsAPIMetrics      TelemetrySubscriptionMetrics // used if telemetry API enabled
	Ack                 chan struct{}                // used by the sending goroutine to wait until ipc message has been sent
}

// InitFailure indicates that runtime/extensions initialization failed due to process exit or /error calls
// In Rapid Shim, this translates to either a DONE or a DONEFAIL GirD message to Slicer (depending on extensions mode)
// However, even on failure, the next invoke is expected to work with a suppressed init - i.e. we init again as aprt of the invoke
type InitFailure struct {
	ResetReceived       bool // indicates if failure happened due to a reset received
	RequestReset        bool // Indicates whether reset should be requested on init failure
	ErrorType           fatalerror.ErrorType
	ErrorMessage        error
	NumActiveExtensions int
	RuntimeRelease      string // value of the User Agent HTTP header provided by runtime
	LogsAPIMetrics      TelemetrySubscriptionMetrics
	Ack                 chan struct{} // used by the sending goroutine to wait until ipc message has been sent
}

// ErrorInvokeResponse represents a buffered response received via Runtime API
// for error responses. When body (Payload) is not provided, e.g.
// not retrievable, error type and error message headers will be
// used by the platform to construct a response json, e.g:
//
// default error response produced by the Slicer:
// '{"errorMessage":"Unknown application error occurred"}',
//
// when error type is provided, error response becomes:
// '{"errorMessage":"Unknown application error occurred","errorType":"ErrorType"}'
type ErrorInvokeResponse struct {
	Headers       InvokeResponseHeaders
	Payload       []byte
	FunctionError FunctionError
}

// StreamableInvokeResponse represents a response received via Runtime API that can be streamed
type StreamableInvokeResponse struct {
	Headers  map[string]string
	Payload  io.Reader
	Trailers http.Header
	Request  *CancellableRequest // streaming request may need to gracefully terminate request streams
}

// InvokeResponseHeaders contains the headers received via Runtime API /invocation/response
type InvokeResponseHeaders struct {
	ContentType          string
	FunctionResponseMode string
}

// FunctionError represents information about function errors or 'user errors'
// These are not platform errors and hence are returned as 200 by Lambda
// In the absence of a response payload, the Function Error is serialized and sent
type FunctionError struct {
	// Type of error is derived from the Lambda-Runtime-Function-Error-Type set by the Runtime
	// This is customer data, so RAPID scrubs this error type to contain only allowlisted values
	Type fatalerror.ErrorType `json:"errorType,omitempty"`
	// ErrorMessage is generated by RAPID and can never be specified by runtime
	Message string `json:"errorMessage,omitempty"`
}

type InvokeResponseSender interface {
	// SendResponse sends invocation response received from Runtime to platform
	// This is response may be streamed based on function and invoke response mode
	SendResponse(invokeID string, response *StreamableInvokeResponse) error
	// SendErrorResponse sends error response in the case of function errors, which are always buffered
	SendErrorResponse(invokeID string, response *ErrorInvokeResponse) error
}

// ResponseMetrics groups metrics related to the response stream
type ResponseMetrics struct {
	RuntimeOutboundThroughputBps int64
	RuntimeProducedBytes         int64
	RuntimeResponseLatencyMs     float64
	RuntimeTimeThrottledMs       int64
}

// InvokeMetrics groups metrics related to the invoke phase
type InvokeMetrics struct {
	InvokeRequestReadTimeNs int64
	InvokeRequestSizeBytes  int64
	RuntimeReadyTime        int64
}

// InvokeSuccess is the success response to invoke phase end
type InvokeSuccess struct {
	RuntimeRelease         string // value of the User Agent HTTP header provided by runtime
	NumActiveExtensions    int
	ExtensionNames         string
	InvokeCompletionTimeNs int64
	InvokeReceivedTime     int64
	LogsAPIMetrics         TelemetrySubscriptionMetrics
	ResponseMetrics        ResponseMetrics
	InvokeMetrics          InvokeMetrics
	InvokeResponseMode     InvokeResponseMode
}

// InvokeFailure is the failure response to invoke phase end
type InvokeFailure struct {
	ResetReceived        bool // indicates if failure happened due to a reset received
	RequestReset         bool // indicates if reset must be requested after the failure
	ErrorType            fatalerror.ErrorType
	ErrorMessage         error
	RuntimeRelease       string // value of the User Agent HTTP header provided by runtime
	NumActiveExtensions  int
	InvokeReceivedTime   int64
	LogsAPIMetrics       TelemetrySubscriptionMetrics
	ResponseMetrics      ResponseMetrics
	InvokeMetrics        InvokeMetrics
	ExtensionNames       string
	DefaultErrorResponse *ErrorInvokeResponse // error resp constructed by platform during fn errors
	InvokeResponseMode   InvokeResponseMode
}

// ResetSuccess is the success response to reset request
type ResetSuccess struct {
	ExtensionsResetMs  int64
	ErrorType          fatalerror.ErrorType
	ResponseMetrics    ResponseMetrics
	InvokeResponseMode InvokeResponseMode
}

// ResetFailure is the failure response to reset request
type ResetFailure struct {
	ExtensionsResetMs  int64
	ErrorType          fatalerror.ErrorType
	ResponseMetrics    ResponseMetrics
	InvokeResponseMode InvokeResponseMode
}

// ShutdownSuccess is the response to a shutdown request
type ShutdownSuccess struct {
	ErrorType fatalerror.ErrorType
}

// SandboxInfoFromInit captures data from init request that
// is required during invoke (e.g. for suppressed init)
type SandboxInfoFromInit struct {
	EnvironmentVariables *env.Environment // contains agent env vars (creds, customer, platform)
	SandboxType          SandboxType      // indicating Pre-Warmed, On-Demand etc
	RuntimeBootstrap     Bootstrap        // contains the runtime bootstrap binary path, Cwd, Args, Env, Cmd
}

// RestoreResult represents the result of `HandleRestore` function
// in RapidCore
type RestoreResult struct {
	RestoreMs int64
}

// RapidContext expose methods for functionality of the Rapid Core library
type RapidContext interface {
	HandleInit(i *Init, success chan<- InitSuccess, failure chan<- InitFailure)
	HandleInvoke(i *Invoke, sbMetadata SandboxInfoFromInit, requestBuf *bytes.Buffer, responseSender InvokeResponseSender) (InvokeSuccess, *InvokeFailure)
	HandleReset(reset *Reset) (ResetSuccess, *ResetFailure)
	HandleShutdown(shutdown *Shutdown) ShutdownSuccess
	HandleRestore(restore *Restore) (RestoreResult, error)
	Clear()

	SetRuntimeStartedTime(runtimeStartedTime int64)
	SetInvokeResponseMetrics(metrics *InvokeResponseMetrics)

	SetEventsAPI(eventsAPI EventsAPI)
}

// SandboxContext represents the sandbox lifecycle context
type SandboxContext interface {
	Init(i *Init, timeoutMs int64) InitContext
	Reset(reset *Reset) (ResetSuccess, *ResetFailure)
	Shutdown(shutdown *Shutdown) ShutdownSuccess
	Restore(restore *Restore) (RestoreResult, error)

	// TODO: refactor this
	// runtimeStartedTime and InvokeResponseMetrics are needed to compute the runtimeDone metrics
	// in case of a Reset during an invoke (reset.reason=failure or reset.reason=timeout).
	// Ideally:
	// - the InvokeContext will have a Reset method to deal with Reset during an invoke and will hold runtimeStartedTime and InvokeResponseMetrics
	// - the SandboxContext will have its own Reset/Spindown method
	SetRuntimeStartedTime(invokeReceivedTime int64)
	SetInvokeResponseMetrics(metrics *InvokeResponseMetrics)
}

// InitContext represents the lifecycle of a sandbox initialization
type InitContext interface {
	Wait() (InitSuccess, *InitFailure)
	Reserve() InvokeContext
}

// InvokeContext represents the lifecycle of a sandbox reservation
type InvokeContext interface {
	SendRequest(i *Invoke, r InvokeResponseSender)
	Wait() (InvokeSuccess, *InvokeFailure)
}

// LifecyclePhase represents enum for possible Sandbox lifecycle phases, like init, invoke, etc.
type LifecyclePhase int

const (
	LifecyclePhaseInit LifecyclePhase = iota + 1
	LifecyclePhaseInvoke
)