-
Notifications
You must be signed in to change notification settings - Fork 847
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #2650 from redpanda-data/lambda-serverless
Add lambda binaries back into releases
- Loading branch information
Showing
7 changed files
with
264 additions
and
22 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "github.com/redpanda-data/connect/v4/internal/impl/aws" | ||
|
|
||
| // Import all plugins defined within the repo. | ||
| _ "github.com/redpanda-data/connect/v4/public/components/all" | ||
| ) | ||
|
|
||
| func main() { | ||
| aws.RunLambda() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| package aws | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "os" | ||
| "time" | ||
|
|
||
| "github.com/aws/aws-lambda-go/lambda" | ||
|
|
||
| "github.com/redpanda-data/connect/v4/internal/serverless" | ||
| ) | ||
|
|
||
| var handler *serverless.Handler | ||
|
|
||
| // RunLambda executes Benthos as an AWS Lambda function. Configuration can be | ||
| // stored within the environment variable CONNECT_CONFIG. | ||
| func RunLambda() { | ||
| // A list of default config paths to check for if not explicitly defined | ||
| defaultPaths := []string{ | ||
| "./redpanda-connect.yaml", | ||
| "/redpanda-connect.yaml", | ||
| "/etc/redpanda-connect/config.yaml", | ||
| "/etc/redpanda-connect.yaml", | ||
|
|
||
| "./connect.yaml", | ||
| "/connect.yaml", | ||
| "/etc/connect/config.yaml", | ||
| "/etc/connect.yaml", | ||
|
|
||
| "./benthos.yaml", | ||
| "./config.yaml", | ||
| "/benthos.yaml", | ||
| "/etc/benthos/config.yaml", | ||
| "/etc/benthos.yaml", | ||
| } | ||
| if path := os.Getenv("BENTHOS_CONFIG_PATH"); path != "" { | ||
| defaultPaths = append([]string{path}, defaultPaths...) | ||
| } | ||
| if path := os.Getenv("CONNECT_CONFIG_PATH"); path != "" { | ||
| defaultPaths = append([]string{path}, defaultPaths...) | ||
| } | ||
|
|
||
| confStr := os.Getenv("BENTHOS_CONFIG") | ||
| if confStr == "" { | ||
| confStr = os.Getenv("CONNECT_CONFIG") | ||
| } | ||
|
|
||
| if confStr == "" { | ||
| // Iterate default config paths | ||
| for _, path := range defaultPaths { | ||
| if confBytes, err := os.ReadFile(path); err == nil { | ||
| confStr = string(confBytes) | ||
| break | ||
| } | ||
| } | ||
| } | ||
|
|
||
| var err error | ||
| if handler, err = serverless.NewHandler(confStr); err != nil { | ||
| fmt.Fprintf(os.Stderr, "Initialisation error: %v\n", err) | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| lambda.Start(handler.Handle) | ||
|
|
||
| ctx, done := context.WithTimeout(context.Background(), time.Second*30) | ||
| defer done() | ||
|
|
||
| if err = handler.Close(ctx); err != nil { | ||
| fmt.Fprintf(os.Stderr, "Shut down error: %v\n", err) | ||
| os.Exit(1) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| package serverless | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/redpanda-data/benthos/v4/public/service" | ||
| ) | ||
|
|
||
| // Handler provides a mechanism for controlling the lifetime of a serverless | ||
| // handler runtime of Redpanda Connect. | ||
| type Handler struct { | ||
| prodFn service.MessageHandlerFunc | ||
| strm *service.Stream | ||
| } | ||
|
|
||
| // NewHandler creates a new serverless stream handler, where the provided config | ||
| // is used in order to determine the behaviour of the pipeline. | ||
| func NewHandler(confYAML string) (*Handler, error) { | ||
| env := service.GlobalEnvironment() | ||
| schema := env.FullConfigSchema("", "") | ||
| schema.SetFieldDefault(map[string]any{ | ||
| "none": map[string]any{}, | ||
| }, "metrics") | ||
| schema.SetFieldDefault("json", "logger", "format") | ||
| schema.SetFieldDefault(map[string]any{ | ||
| "inproc": "____ignored", | ||
| }, "input") | ||
| schema.SetFieldDefault(map[string]any{ | ||
| "switch": map[string]any{ | ||
| "retry_until_success": false, | ||
| "cases": []any{ | ||
| map[string]any{ | ||
| "check": "errored()", | ||
| "output": map[string]any{ | ||
| "reject": "processing failed due to: ${! error() }", | ||
| }, | ||
| }, | ||
| map[string]any{ | ||
| "output": map[string]any{ | ||
| "sync_response": map[string]any{}, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, "output") | ||
|
|
||
| strmBuilder := env.NewStreamBuilder() | ||
| strmBuilder.SetSchema(schema) | ||
|
|
||
| if err := strmBuilder.SetYAML(confYAML); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| prod, err := strmBuilder.AddProducerFunc() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| strm, err := strmBuilder.Build() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| go func() { | ||
| _ = strm.Run(context.Background()) | ||
| }() | ||
|
|
||
| return &Handler{ | ||
| prodFn: prod, | ||
| strm: strm, | ||
| }, nil | ||
| } | ||
|
|
||
| // Close shuts down the underlying pipeline. | ||
| func (h *Handler) Close(ctx context.Context) error { | ||
| return h.strm.Stop(ctx) | ||
| } | ||
|
|
||
| // Handle is a request/response func that injects a payload into the underlying | ||
| // Benthos pipeline and returns a result. | ||
| func (h *Handler) Handle(ctx context.Context, v any) (any, error) { | ||
| msg := service.NewMessage(nil) | ||
| msg.SetStructured(v) | ||
|
|
||
| msg, store := msg.WithSyncResponseStore() | ||
|
|
||
| if err := h.prodFn(ctx, msg); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| resultBatches := store.Read() | ||
|
|
||
| anyResults := make([][]any, len(resultBatches)) | ||
| for i, batch := range resultBatches { | ||
| batchResults := make([]any, len(batch)) | ||
| for j, p := range batch { | ||
| var merr error | ||
| if batchResults[j], merr = p.AsStructured(); merr != nil { | ||
| return nil, fmt.Errorf("failed to process result batch '%v': failed to marshal json response: %v", i, merr) | ||
| } | ||
| } | ||
| anyResults[i] = batchResults | ||
| } | ||
|
|
||
| if len(anyResults) == 1 { | ||
| if len(anyResults[0]) == 1 { | ||
| return anyResults[0][0], nil | ||
| } | ||
| return anyResults[0], nil | ||
| } | ||
|
|
||
| genBatchOfBatches := make([]any, len(anyResults)) | ||
| for i, b := range anyResults { | ||
| genBatchOfBatches[i] = b | ||
| } | ||
| return genBatchOfBatches, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| package serverless_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/redpanda-data/connect/v4/internal/serverless" | ||
|
|
||
| _ "github.com/redpanda-data/connect/v4/public/components/pure" | ||
| ) | ||
|
|
||
| func TestServerlessHandlerDefaults(t *testing.T) { | ||
| h, err := serverless.NewHandler(` | ||
| pipeline: | ||
| processors: | ||
| - mapping: 'root = content().uppercase()' | ||
| logger: | ||
| level: NONE | ||
| `) | ||
| require.NoError(t, err) | ||
|
|
||
| ctx, done := context.WithTimeout(context.Background(), time.Second*5) | ||
| defer done() | ||
|
|
||
| res, err := h.Handle(ctx, "hello world") | ||
| require.NoError(t, err) | ||
|
|
||
| assert.Equal(t, "HELLO WORLD", res) | ||
|
|
||
| require.NoError(t, h.Close(ctx)) | ||
| } |