Small refactor

Author: gregfurman

cmd/localstack/main.go

@@ -14,10 +14,12 @@ import (
 	"sync"
 
 	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/localstack/lambda-runtime-init/internal/aws/lambda"
 	"github.com/localstack/lambda-runtime-init/internal/aws/xray"
 	"github.com/localstack/lambda-runtime-init/internal/bootstrap"
 	"github.com/localstack/lambda-runtime-init/internal/events"
 	"github.com/localstack/lambda-runtime-init/internal/hotreloading"
+	"github.com/localstack/lambda-runtime-init/internal/localstack"
 	"github.com/localstack/lambda-runtime-init/internal/logging"
 	"github.com/localstack/lambda-runtime-init/internal/server"
 
@@ -31,8 +33,8 @@ import (
 	supv "go.amzn.com/lambda/supervisor"
 )
 
-func InitLsOpts() *server.LsOpts {
-	return &server.LsOpts{
+func InitLsOpts() *localstack.Config {
+	return &localstack.Config{
 		// required
 		RuntimeEndpoint: utils.MustGetEnv("LOCALSTACK_RUNTIME_ENDPOINT"),
 		RuntimeId:       utils.MustGetEnv("LOCALSTACK_RUNTIME_ID"),
@@ -55,8 +57,8 @@ func InitLsOpts() *server.LsOpts {
 	}
 }
 
-func InitFunctionConfig() server.FunctionConfig {
-	return server.FunctionConfig{
+func InitFunctionConfig() lambda.FunctionConfig {
+	return lambda.FunctionConfig{
 		FunctionName:         utils.GetEnvWithDefault("AWS_LAMBDA_FUNCTION_NAME", "test_function"),
 		FunctionVersion:      utils.GetEnvWithDefault("AWS_LAMBDA_FUNCTION_VERSION", "$LATEST"),
 		FunctionTimeoutSec:   utils.GetEnvWithDefault("AWS_LAMBDA_FUNCTION_TIMEOUT", "30"),
@@ -189,14 +191,14 @@ func main() {
 
 	// Custom Interop Server
 	defaultServer := rapidcore.NewServer()
-	lsAdapter := server.NewLocalStackAdapter(lsOpts.RuntimeEndpoint, lsOpts.RuntimeId)
-	interopServer := server.NewInteropServer(defaultServer, lsAdapter)
+	lsClient := localstack.NewLocalStackClient(lsOpts.RuntimeEndpoint, lsOpts.RuntimeId)
+	interopServer := server.NewInteropServer(defaultServer, lsClient)
 
 	// Services required for Sandbox environment
 	logCollector := logging.NewLogCollector()
 	localStackLogsEgressApi := logging.NewLocalStackLogsEgressAPI(logCollector)
 	tracer := tracing.NewLocalStackTracer()
-	eventsListener := events.NewEventsListener(lsAdapter)
+	eventsListener := events.NewLocalStackEventsAPI(lsClient)
 
 	defaultSupv := supv.NewLocalSupervisor()
 	wrappedSupv := supervisor.NewLocalStackSupervisor(ctx, defaultSupv, eventsListener, interopServer.InternalState)
@@ -246,7 +248,7 @@ func main() {
 	interopServer.SetSandboxContext(sandboxContext)
 	interopServer.SetInternalStateGetter(internalStateFn)
 
-	localStackService := server.NewLocalStackService(interopServer, logCollector, lsAdapter, xrayConfig.Endpoint, lsOpts, functionConf, awsEnvConf)
+	localStackService := server.NewLocalStackService(interopServer, logCollector, lsClient, xrayConfig.Endpoint, lsOpts, functionConf, awsEnvConf)
 
 	// start runtime init. It is important to start `InitHandler` synchronously because we need to ensure the
 	// notification channels and status fields are properly initialized before `AwaitInitialized`
@@ -285,7 +287,7 @@ func main() {
 		// return
 	} else {
 		log.Debugln("Completed initialization of runtime init. Sending status ready to LocalStack.")
-		if err := localStackService.SendStatus(server.Ready, []byte{}); err != nil {
+		if err := localStackService.SendStatus(localstack.Ready, []byte{}); err != nil {
 			log.Fatalln("Failed to send status ready to LocalStack " + err.Error() + ". Exiting.")
 		}
 	}

go.mod

@@ -33,6 +33,7 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	golang.org/x/net v0.38.0 // indirect
+	golang.org/x/sync v0.15.0 // indirect
 	golang.org/x/text v0.23.0 // indirect
 	gopkg.in/yaml.v2 v2.2.8 // indirect
 )

go.sum

@@ -78,6 +78,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
 golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sync v0.15.0 h1:KWH3jNZsfyT6xfAfKiz6MRNmd46ByHDYaZ7KSkCtdW8=
+golang.org/x/sync v0.15.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

internal/aws/lambda/config.go

@@ -0,0 +1,12 @@
+package lambda
+
+type FunctionConfig struct {
+	FunctionName         string // AWS_LAMBDA_FUNCTION_NAME
+	FunctionMemorySizeMb string // AWS_LAMBDA_FUNCTION_MEMORY_SIZE
+	FunctionVersion      string // AWS_LAMBDA_FUNCTION_VERSION
+	FunctionTimeoutSec   string // AWS_LAMBDA_FUNCTION_TIMEOUT
+	InitializationType   string // AWS_LAMBDA_INITIALIZATION_TYPE
+	LogGroupName         string // AWS_LAMBDA_LOG_GROUP_NAME
+	LogStreamName        string // AWS_LAMBDA_LOG_STREAM_NAME
+	FunctionHandler      string //  AWS_LAMBDA_FUNCTION_HANDLER || _HANDLER
+}

internal/events/events.go

@@ -4,28 +4,31 @@ import (
 	"encoding/json"
 	"fmt"
 
-	"github.com/localstack/lambda-runtime-init/internal/server"
+	"github.com/localstack/lambda-runtime-init/internal/localstack"
 
 	"go.amzn.com/lambda/interop"
-	"go.amzn.com/lambda/telemetry"
+	"go.amzn.com/lambda/rapidcore/standalone/telemetry"
 )
 
-type EventsListener struct {
+// LocalStackEventsAPI handles internally emitted rapid events.
+// TODO: Logs should all be collected here
+type LocalStackEventsAPI struct {
 	interop.EventsAPI
-	adapter *server.LocalStackAdapter
+	adapter *localstack.LocalStackClient
 }
 
-func NewEventsListener(adapter *server.LocalStackAdapter) *EventsListener {
-	return &EventsListener{
-		adapter: adapter,
-		// For now just use the no-ops API to satisfy the EventsAPI interface
-		EventsAPI: &telemetry.NoOpEventsAPI{},
+func NewLocalStackEventsAPI(adapter *localstack.LocalStackClient) *LocalStackEventsAPI {
+	return &LocalStackEventsAPI{
+		adapter:   adapter,
+		EventsAPI: new(telemetry.StandaloneEventsAPI),
 	}
-
 }
 
-func (ev *EventsListener) SendFault(data interop.FaultData) error {
-	resp := server.ErrorResponse{
+func (ev *LocalStackEventsAPI) SendFault(data interop.FaultData) error {
+	// We can ignore whatever errors are returned here
+	_ = ev.EventsAPI.SendFault(data)
+
+	resp := localstack.ErrorResponse{
 		ErrorMessage: fmt.Sprintf("RequestId: %s Error: %s", data.RequestID, data.ErrorMessage),
 		ErrorType:    string(data.ErrorType),
 	}
@@ -35,5 +38,5 @@ func (ev *EventsListener) SendFault(data interop.FaultData) error {
 		return err
 	}
 
-	return ev.adapter.SendStatus(server.Error, payload)
+	return ev.adapter.SendStatus(localstack.Error, payload)
 }

internal/localstack/client.go

@@ -0,0 +1,43 @@
+package localstack
+
+import (
+	"bytes"
+	"net/http"
+	"net/url"
+
+	"github.com/sirupsen/logrus"
+)
+
+type LocalStackClient struct {
+	UpstreamEndpoint string
+	RuntimeId        string
+}
+
+func NewLocalStackClient(endpoint, runtimeId string) *LocalStackClient {
+	return &LocalStackClient{
+		UpstreamEndpoint: endpoint,
+		RuntimeId:        runtimeId,
+	}
+}
+
+func (ls *LocalStackClient) sendCallback(namespace, invokeId, dest string, payload []byte) error {
+	endpoint, err := url.JoinPath(ls.UpstreamEndpoint, namespace, invokeId, dest)
+	if err != nil {
+		return err
+	}
+	logrus.WithField("url", endpoint).WithField("invoke-id", invokeId).Debugf("Sending callback.")
+
+	if _, err := http.Post(endpoint, "application/json", bytes.NewReader(payload)); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (ls *LocalStackClient) SendStatus(status LocalStackStatus, payload []byte) error {
+	return ls.sendCallback("/status", ls.RuntimeId, string(status), payload)
+}
+
+func (ls *LocalStackClient) SendInvocation(invokeId, dest string, payload []byte) error {
+	return ls.sendCallback("invocations", invokeId, dest, payload)
+}

internal/localstack/config.go

@@ -0,0 +1,41 @@
+package localstack
+
+// Original implementation: lambda/rapidcore/server.go includes Server struct with state
+// Server interface between Runtime API and this init: lambda/interop/model.go:Server
+
+type Config struct {
+	InteropPort         string
+	RuntimeEndpoint     string
+	RuntimeId           string
+	AccountId           string
+	InitTracingPort     string
+	User                string
+	CodeArchives        string
+	HotReloadingPaths   []string
+	FileWatcherStrategy string
+	ChmodPaths          string
+	LocalstackIP        string
+	InitLogLevel        string
+	EdgePort            string
+	EnableXRayTelemetry string
+	PostInvokeWaitMS    string
+	MaxPayloadSize      string
+}
+
+// The InvokeRequest is sent by LocalStack to trigger an invocation
+type InvokeRequest struct {
+	InvokeId           string `json:"request-id"`
+	InvokedFunctionArn string `json:"invoked-function-arn"`
+	Payload            string `json:"payload"`
+	TraceId            string `json:"trace-id"`
+	ClientContext      string `json:"client-context"`
+}
+
+// The ErrorResponse is sent TO LocalStack when encountering an error
+type ErrorResponse struct {
+	ErrorMessage string   `json:"errorMessage"`
+	ErrorType    string   `json:"errorType,omitempty"`
+	RequestId    *string  `json:"requestId,omitempty"`
+	StackTrace   []string `json:"stackTrace,omitempty"`
+	Trace        []string `json:"trace,omitempty"`
+}

internal/localstack/model.go

@@ -0,0 +1,8 @@
+package localstack
+
+type LocalStackStatus string
+
+const (
+	Ready LocalStackStatus = "ready"
+	Error LocalStackStatus = "error"
+)

internal/server/handler.go

@@ -9,6 +9,7 @@ import (
 	"reflect"
 
 	"github.com/go-chi/chi"
+	"github.com/localstack/lambda-runtime-init/internal/localstack"
 	log "github.com/sirupsen/logrus"
 	"go.amzn.com/lambda/fatalerror"
 	"go.amzn.com/lambda/rapidcore"
@@ -35,7 +36,7 @@ func InvokeHandler(api *LocalStackService) http.HandlerFunc {
 
 		// TODO: We shouldn't be using a custom request body here,
 		// instead we should be forwarding the entire boto3/AWS request
-		var req InvokeRequest
+		var req localstack.InvokeRequest
 		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 			log.WithError(err).Error("Failed to decode invoke request")
 		}
@@ -46,7 +47,7 @@ func InvokeHandler(api *LocalStackService) http.HandlerFunc {
 		// we can actually just continue here, error message is sent below
 		case errors.Is(err, rapidcore.ErrInvokeTimeout):
 			log.Debugf("Got invoke timeout")
-			errorResponse := ErrorResponse{
+			errorResponse := localstack.ErrorResponse{
 				ErrorMessage: fmt.Sprintf(
 					"RequestId: %s Error: Task timed out after %s.00 seconds",
 					req.InvokeId,
@@ -70,7 +71,7 @@ func InvokeHandler(api *LocalStackService) http.HandlerFunc {
 
 		// If the response cannot be decoded into an ErrorResponse type,
 		// we assume no error is present.
-		var errorResponse ErrorResponse
+		var errorResponse localstack.ErrorResponse
 		if err := invokeRespDecoder.Decode(&errorResponse); err != nil || reflect.ValueOf(errorResponse).IsZero() {
 			api.SendResponse(req.InvokeId, response)
 			return