diff --git a/.gitignore b/.gitignore index 911ec73d6..e8a9f21ce 100644 --- a/.gitignore +++ b/.gitignore @@ -55,7 +55,7 @@ escape_analysis.txt *.prof # Compiled test wasm processors -pkg/plugin/processor/standalone/test/wasm_processors/*/processor.wasm +**/test/**/*.wasm # Test data **/test/*.txt diff --git a/pkg/plugin/processor/standalone/command/builder.go b/pkg/plugin/processor/standalone/command/builder.go new file mode 100644 index 000000000..a14de114c --- /dev/null +++ b/pkg/plugin/processor/standalone/command/builder.go @@ -0,0 +1,66 @@ +// Copyright © 2025 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package command + +import ( + "context" + + "github.com/conduitio/conduit-processor-sdk/pprocutils" + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/stealthrocket/wazergo" + "github.com/tetratelabs/wazero" +) + +type Builder struct { + logger log.CtxLogger + + runtime wazero.Runtime + // hostModule is the conduit host module that exposes Conduit host functions + // to the WASM module. The host module is compiled once and instantiated + // multiple times, once for each WASM module. + hostModule *wazergo.CompiledModule[*HostModuleInstance] + schemaService pprocutils.SchemaService +} + +func NewBuilder( + ctx context.Context, + logger log.CtxLogger, + runtime wazero.Runtime, + schemaService pprocutils.SchemaService, +) (*Builder, error) { + logger = logger.WithComponentFromType(Builder{}) + + // init host module + compiledHostModule, err := wazergo.Compile(ctx, runtime, HostModule) + if err != nil { + return nil, cerrors.Errorf("failed to compile host module: %w", err) + } + + return &Builder{ + logger: logger, + runtime: runtime, + hostModule: compiledHostModule, + schemaService: schemaService, + }, nil +} + +func (b *Builder) Build( + ctx context.Context, + processorModule wazero.CompiledModule, + id string, +) (*WasmProcessor, error) { + return NewWasmProcessor(ctx, b.runtime, processorModule, b.hostModule, b.schemaService, id, b.logger) +} diff --git a/pkg/plugin/processor/standalone/command/host_module.go b/pkg/plugin/processor/standalone/command/host_module.go new file mode 100644 index 000000000..63f2229bc --- /dev/null +++ b/pkg/plugin/processor/standalone/command/host_module.go @@ -0,0 +1,158 @@ +// Copyright © 2025 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package command + +import ( + "context" + + "github.com/conduitio/conduit-processor-sdk/pprocutils" + processorv1 "github.com/conduitio/conduit-processor-sdk/proto/processor/v1" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/common" + "github.com/stealthrocket/wazergo" + "github.com/stealthrocket/wazergo/types" + "google.golang.org/protobuf/proto" +) + +// HostModule declares the host module that is exported to the WASM module. The +// host module is used to communicate between the WASM module (processor) and Conduit. +var HostModule wazergo.HostModule[*HostModuleInstance] = HostModuleFunctions{ + "command_request": wazergo.F1((*HostModuleInstance).CommandRequest), + "command_response": wazergo.F1((*HostModuleInstance).CommandResponse), + "create_schema": wazergo.F1((*HostModuleInstance).CreateSchema), + "get_schema": wazergo.F1((*HostModuleInstance).GetSchema), +} + +// HostModuleFunctions type implements HostModule, providing the module name, +// map of exported functions, and the ability to create instances of the module +// type. +type HostModuleFunctions wazergo.Functions[*HostModuleInstance] + +// Name returns the name of the module. +func (f HostModuleFunctions) Name() string { + return "conduit" +} + +// Functions is a helper that returns the exported functions of the module. +func (f HostModuleFunctions) Functions() wazergo.Functions[*HostModuleInstance] { + return (wazergo.Functions[*HostModuleInstance])(f) +} + +// Instantiate creates a new instance of the module. This is called by the +// runtime when a new instance of the module is created. +func (f HostModuleFunctions) Instantiate(_ context.Context, opts ...HostModuleOption) (*HostModuleInstance, error) { + mod := NewHostModuleInstance() + wazergo.Configure(mod, opts...) + return mod, nil +} + +type HostModuleOption = wazergo.Option[*HostModuleInstance] + +func HostModuleOptions( + logger log.CtxLogger, + schemaService pprocutils.SchemaService, + requests <-chan *processorv1.CommandRequest, + responses chan<- tuple[*processorv1.CommandResponse, error], +) HostModuleOption { + return wazergo.OptionFunc(func(m *HostModuleInstance) { + wazergo.Configure(m.HostModuleInstance, common.HostModuleOptions(logger, schemaService)) + m.logger = logger + m.commandRequests = requests + m.commandResponses = responses + }) +} + +// HostModuleInstance is used to maintain the state of our module instance. +type HostModuleInstance struct { + *common.HostModuleInstance + + logger log.CtxLogger + commandRequests <-chan *processorv1.CommandRequest + commandResponses chan<- tuple[*processorv1.CommandResponse, error] + parkedCommandRequest *processorv1.CommandRequest +} + +func NewHostModuleInstance() *HostModuleInstance { + return &HostModuleInstance{ + HostModuleInstance: common.NewHostModuleInstance(), + } +} + +func (m *HostModuleInstance) CreateSchema(ctx context.Context, buf types.Bytes) types.Uint32 { + return m.HostModuleInstance.CreateSchema(ctx, buf) +} + +func (m *HostModuleInstance) GetSchema(ctx context.Context, buf types.Bytes) types.Uint32 { + return m.HostModuleInstance.GetSchema(ctx, buf) +} + +// CommandRequest is the exported function that is called by the WASM module to +// get the next command request. It returns the size of the command request +// message. If the buffer is too small, it returns the size of the command +// request message and parks the command request. The next call to this function +// will return the same command request. +func (m *HostModuleInstance) CommandRequest(ctx context.Context, buf types.Bytes) types.Uint32 { + m.logger.Trace(ctx).Msg("executing command_request") + + if m.parkedCommandRequest == nil { + // No parked command, so we need to wait for the next one. If the command + // channel is closed, then we return an error. + var ok bool + m.parkedCommandRequest, ok = <-m.commandRequests + if !ok { + return pprocutils.ErrorCodeNoMoreCommands + } + } + + // If the buffer is too small, we park the command and return the size of the + // command. The next call to nextCommand will return the same command. + if size := proto.Size(m.parkedCommandRequest); len(buf) < size { + m.logger.Warn(ctx). + Int("command_bytes", size). + Int("allocated_bytes", len(buf)). + Msgf("insufficient memory, command will be parked until next call to command_request") + return types.Uint32(size) //nolint:gosec // no risk of overflow + } + + // If the buffer is large enough, we marshal the command into the buffer and + // return the size of the command. The next call to nextCommand will return + // the next command. + out, err := proto.MarshalOptions{}.MarshalAppend(buf[:0], m.parkedCommandRequest) + if err != nil { + m.logger.Err(ctx, err).Msg("failed marshalling protobuf command request") + return pprocutils.ErrorCodeUnknownCommandRequest + } + m.parkedCommandRequest = nil + + m.logger.Trace(ctx).Msg("returning next command") + return types.Uint32(len(out)) //nolint:gosec // no risk of overflow +} + +// CommandResponse is the exported function that is called by the WASM module to +// send a command response. It returns 0 on success, or an error code on error. +func (m *HostModuleInstance) CommandResponse(ctx context.Context, buf types.Bytes) types.Uint32 { + m.logger.Trace(ctx).Msg("executing command_response") + + var resp processorv1.CommandResponse + err := proto.Unmarshal(buf, &resp) + if err != nil { + m.logger.Err(ctx, err).Msg("failed unmarshalling protobuf command response") + m.commandResponses <- tuple[*processorv1.CommandResponse, error]{nil, err} + return pprocutils.ErrorCodeUnknownCommandResponse + } + + m.commandResponses <- tuple[*processorv1.CommandResponse, error]{&resp, nil} + return 0 +} diff --git a/pkg/plugin/processor/standalone/processor.go b/pkg/plugin/processor/standalone/command/processor.go similarity index 87% rename from pkg/plugin/processor/standalone/processor.go rename to pkg/plugin/processor/standalone/command/processor.go index aba22c5cc..3f793df0b 100644 --- a/pkg/plugin/processor/standalone/processor.go +++ b/pkg/plugin/processor/standalone/command/processor.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package standalone +package command import ( "context" @@ -27,6 +27,8 @@ import ( "github.com/conduitio/conduit/pkg/foundation/cerrors" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/conduitio/conduit/pkg/plugin" + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/common" + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/common/fromproto" "github.com/stealthrocket/wazergo" "github.com/tetratelabs/wazero" "github.com/tetratelabs/wazero/api" @@ -45,9 +47,8 @@ const ( conduitLogLevelKey = "CONDUIT_LOG_LEVEL" ) -type wasmProcessor struct { +type WasmProcessor struct { sdk.UnimplementedProcessor - protoconv protoConverter id string logger log.CtxLogger @@ -59,7 +60,7 @@ type wasmProcessor struct { // WASM module) commandRequests chan *processorv1.CommandRequest // commandResponses is used to communicate replies between the actual - // processor (the WASM module) and wasmProcessor + // processor (the WASM module) and WasmProcessor commandResponses chan tuple[*processorv1.CommandResponse, error] // moduleStopped is used to know when the module stopped running @@ -73,20 +74,20 @@ type tuple[T1, T2 any] struct { V2 T2 } -func newWASMProcessor( +func NewWasmProcessor( ctx context.Context, runtime wazero.Runtime, processorModule wazero.CompiledModule, - hostModule *wazergo.CompiledModule[*hostModuleInstance], + hostModule *wazergo.CompiledModule[*HostModuleInstance], schemaService pprocutils.SchemaService, id string, logger log.CtxLogger, -) (*wasmProcessor, error) { - logger = logger.WithComponent("standalone.wasmProcessor") +) (*WasmProcessor, error) { + logger = logger.WithComponentFromType(WasmProcessor{}) logger.Logger = logger.With().Str(log.ProcessorIDField, id).Logger() - wasmLogger := newWasmLogWriter(logger, processorModule) + wasmLogger := common.NewWasmLogWriter(logger, processorModule) commandRequests := make(chan *processorv1.CommandRequest) commandResponses := make(chan tuple[*processorv1.CommandResponse, error]) @@ -96,11 +97,11 @@ func newWASMProcessor( logger.Debug(ctx).Msg("instantiating conduit host module") ins, err := hostModule.Instantiate( ctx, - hostModuleOptions( + HostModuleOptions( logger, + schemaService, commandRequests, commandResponses, - schemaService, ), ) if err != nil { @@ -136,7 +137,7 @@ func newWASMProcessor( return nil, fmt.Errorf("failed to instantiate processor module: %w", err) } - p := &wasmProcessor{ + p := &WasmProcessor{ id: id, logger: logger, module: mod, @@ -154,7 +155,7 @@ func newWASMProcessor( // run is the main loop of the WASM module. It runs in a goroutine and blocks // until the module is closed. -func (p *wasmProcessor) run(ctx context.Context) { +func (p *WasmProcessor) run(ctx context.Context) { defer close(p.moduleStopped) _, err := p.module.ExportedFunction("_start").Call(ctx) @@ -175,7 +176,7 @@ func (p *wasmProcessor) run(ctx context.Context) { p.logger.Err(ctx, err).Msg("WASM module stopped") } -func (p *wasmProcessor) Specification() (sdk.Specification, error) { +func (p *WasmProcessor) Specification() (sdk.Specification, error) { req := &processorv1.CommandRequest{ Request: &processorv1.CommandRequest_Specify{ Specify: &processorv1.Specify_Request{}, @@ -193,13 +194,13 @@ func (p *wasmProcessor) Specification() (sdk.Specification, error) { switch specResp := resp.Response.(type) { case *processorv1.CommandResponse_Specify: - return p.protoconv.specification(specResp.Specify) + return fromproto.Specification(specResp.Specify) default: return sdk.Specification{}, fmt.Errorf("unexpected response type: %T", resp) } } -func (p *wasmProcessor) Configure(ctx context.Context, config config.Config) error { +func (p *WasmProcessor) Configure(ctx context.Context, config config.Config) error { req := &processorv1.CommandRequest{ Request: &processorv1.CommandRequest_Configure{ Configure: &processorv1.Configure_Request{ @@ -221,7 +222,7 @@ func (p *wasmProcessor) Configure(ctx context.Context, config config.Config) err } } -func (p *wasmProcessor) Open(ctx context.Context) error { +func (p *WasmProcessor) Open(ctx context.Context) error { req := &processorv1.CommandRequest{ Request: &processorv1.CommandRequest_Open{ Open: &processorv1.Open_Request{}, @@ -241,8 +242,8 @@ func (p *wasmProcessor) Open(ctx context.Context) error { } } -func (p *wasmProcessor) Process(ctx context.Context, records []opencdc.Record) []sdk.ProcessedRecord { - protoRecords, err := p.protoconv.records(records) +func (p *WasmProcessor) Process(ctx context.Context, records []opencdc.Record) []sdk.ProcessedRecord { + protoRecords, err := fromproto.Records(records) if err != nil { p.logger.Err(ctx, err).Msg("failed to convert records to proto") return []sdk.ProcessedRecord{sdk.ErrorRecord{Error: err}} @@ -263,7 +264,7 @@ func (p *wasmProcessor) Process(ctx context.Context, records []opencdc.Record) [ switch procResp := resp.Response.(type) { case *processorv1.CommandResponse_Process: - processedRecords, err := p.protoconv.processedRecords(procResp.Process.Records) + processedRecords, err := fromproto.ProcessedRecords(procResp.Process.Records) if err != nil { p.logger.Err(ctx, err).Msg("failed to convert processed records from proto") return []sdk.ProcessedRecord{sdk.ErrorRecord{Error: err}} @@ -275,7 +276,7 @@ func (p *wasmProcessor) Process(ctx context.Context, records []opencdc.Record) [ } } -func (p *wasmProcessor) Teardown(ctx context.Context) error { +func (p *WasmProcessor) Teardown(ctx context.Context) error { // TODO: we should probably have a timeout for the teardown command in case // the plugin is stuck teardownErr := p.executeTeardownCommand(ctx) @@ -285,7 +286,7 @@ func (p *wasmProcessor) Teardown(ctx context.Context) error { return cerrors.Join(teardownErr, stopErr, p.moduleError) } -func (p *wasmProcessor) executeTeardownCommand(ctx context.Context) error { +func (p *WasmProcessor) executeTeardownCommand(ctx context.Context) error { req := &processorv1.CommandRequest{ Request: &processorv1.CommandRequest_Teardown{ Teardown: &processorv1.Teardown_Request{}, @@ -304,7 +305,7 @@ func (p *wasmProcessor) executeTeardownCommand(ctx context.Context) error { } } -func (p *wasmProcessor) closeModule(ctx context.Context) error { +func (p *WasmProcessor) closeModule(ctx context.Context) error { // Closing the command channel will send an error code to the WASM module // signaling it to exit. close(p.commandRequests) @@ -326,7 +327,7 @@ func (p *wasmProcessor) closeModule(ctx context.Context) error { // executeCommand sends a command request to the WASM module and waits for the // response. It returns the response, or an error if the response is an error. // If the context is canceled, it returns ctx.Err(). -func (p *wasmProcessor) executeCommand(ctx context.Context, req *processorv1.CommandRequest) (*processorv1.CommandResponse, error) { +func (p *WasmProcessor) executeCommand(ctx context.Context, req *processorv1.CommandRequest) (*processorv1.CommandResponse, error) { select { case <-ctx.Done(): return nil, ctx.Err() @@ -355,7 +356,7 @@ func (p *wasmProcessor) executeCommand(ctx context.Context, req *processorv1.Com // check if the response is an error if errResp, ok := resp.Response.(*processorv1.CommandResponse_Error); ok { - return nil, p.protoconv.error(errResp.Error) + return nil, fromproto.Error(errResp.Error) } return resp, nil diff --git a/pkg/plugin/processor/standalone/processor_test.go b/pkg/plugin/processor/standalone/command/processor_test.go similarity index 89% rename from pkg/plugin/processor/standalone/processor_test.go rename to pkg/plugin/processor/standalone/command/processor_test.go index fad3c499e..f5cad58a4 100644 --- a/pkg/plugin/processor/standalone/processor_test.go +++ b/pkg/plugin/processor/standalone/command/processor_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package standalone +package command import ( "context" @@ -34,7 +34,7 @@ func TestWASMProcessor_Specification_Success(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) gotSpec, err := underTest.Specification() @@ -51,7 +51,7 @@ func TestWASMProcessor_Specification_Error(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, SpecifyErrorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, SpecifyErrorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) _, err = underTest.Specification() @@ -66,7 +66,7 @@ func TestWASMProcessor_Configure_Success(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) err = underTest.Configure(ctx, nil) @@ -80,7 +80,7 @@ func TestWASMProcessor_Configure_Error(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) err = underTest.Configure(ctx, map[string]string{"configure": "error"}) @@ -95,7 +95,7 @@ func TestWASMProcessor_Configure_Panic(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) err = underTest.Configure(ctx, map[string]string{"configure": "panic"}) @@ -111,7 +111,7 @@ func TestWASMProcessor_Open_Success(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) err = underTest.Open(ctx) @@ -125,7 +125,7 @@ func TestWASMProcessor_Open_Error(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) err = underTest.Configure(ctx, map[string]string{"open": "error"}) @@ -143,7 +143,7 @@ func TestWASMProcessor_Open_Panic(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) err = underTest.Configure(ctx, map[string]string{"open": "panic"}) @@ -162,7 +162,7 @@ func TestWASMProcessor_Process_Success(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) is.NoErr(underTest.Configure(ctx, map[string]string{"process.prefix": "hello!\n\n"})) @@ -199,7 +199,7 @@ func TestWASMProcessor_Process_Error(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) is.NoErr(underTest.Configure(ctx, map[string]string{"process": "error"})) @@ -220,7 +220,7 @@ func TestWASMProcessor_Process_Panic(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) is.NoErr(underTest.Configure(ctx, map[string]string{"process": "panic"})) @@ -242,7 +242,7 @@ func TestWASMProcessor_Configure_Schema_Success(t *testing.T) { ctx := context.Background() logger := log.Test(t) - underTest, err := newWASMProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) + underTest, err := NewWasmProcessor(ctx, TestRuntime, ChaosProcessorModule, CompiledHostModule, schema.NewInMemoryService(), "test-processor", logger) is.NoErr(err) err = underTest.Configure(ctx, map[string]string{"configure": "create_and_get_schema"}) diff --git a/pkg/plugin/processor/standalone/command/standalone_test.go b/pkg/plugin/processor/standalone/command/standalone_test.go new file mode 100644 index 000000000..c6c098022 --- /dev/null +++ b/pkg/plugin/processor/standalone/command/standalone_test.go @@ -0,0 +1,169 @@ +// Copyright © 2023 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package command + +import ( + "context" + "fmt" + "maps" + "os" + "os/exec" + "testing" + "time" + + "github.com/conduitio/conduit-commons/config" + sdk "github.com/conduitio/conduit-processor-sdk" + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/stealthrocket/wazergo" + "github.com/tetratelabs/wazero" + "github.com/tetratelabs/wazero/imports/wasi_snapshot_preview1" +) + +const ( + testPluginDir = "./test/wasm_processors/" + + testPluginChaosDir = testPluginDir + "chaos/" + testPluginMalformedDir = testPluginDir + "malformed/" + testPluginSpecifyErrorDir = testPluginDir + "specify_error/" +) + +var ( + // TestRuntime can be reused in tests to avoid recompiling the test modules + TestRuntime wazero.Runtime + CompiledHostModule *wazergo.CompiledModule[*HostModuleInstance] + + ChaosProcessorBinary []byte + MalformedProcessorBinary []byte + SpecifyErrorBinary []byte + + ChaosProcessorModule wazero.CompiledModule + SpecifyErrorModule wazero.CompiledModule + + testProcessorPaths = map[string]tuple[*[]byte, *wazero.CompiledModule]{ + testPluginChaosDir + "processor.wasm": {&ChaosProcessorBinary, &ChaosProcessorModule}, + testPluginMalformedDir + "processor.txt": {&MalformedProcessorBinary, nil}, + testPluginSpecifyErrorDir + "processor.wasm": {&SpecifyErrorBinary, &SpecifyErrorModule}, + } +) + +func TestMain(m *testing.M) { + exitOnError := func(err error, msg string) { + if err != nil { + _, _ = fmt.Fprintf(os.Stderr, "%v: %v", msg, err) + os.Exit(1) + } + } + + cmd := exec.Command("bash", "./test/build-test-processors.sh") + + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + + { + fmt.Printf("Building test processors (%s)...\n", cmd.String()) + start := time.Now() + + err := cmd.Run() + exitOnError(err, "error executing bash script") + + fmt.Printf("Built test processors in %v\n", time.Since(start)) + } + + // instantiate shared test runtime + ctx := context.Background() + + // use interpreter runtime as it's faster for tests + TestRuntime = func(ctx context.Context) wazero.Runtime { + cfg := wazero.NewRuntimeConfigInterpreter() + return wazero.NewRuntimeWithConfig(ctx, cfg) + }(ctx) + + _, err := wasi_snapshot_preview1.Instantiate(ctx, TestRuntime) + exitOnError(err, "error instantiating WASI") + + CompiledHostModule, err = wazergo.Compile(ctx, TestRuntime, HostModule) + exitOnError(err, "error compiling host module") + + // load test processors + for path, t := range testProcessorPaths { + *t.V1, err = os.ReadFile(path) + exitOnError(err, "error reading file "+path) + + if t.V2 == nil { + continue + } + + fmt.Printf("Compiling module %s...\n", path) + start := time.Now() + + // note that modules can't be compiled in parallel, because the runtime + // is not thread-safe + *t.V2, err = TestRuntime.CompileModule(ctx, *t.V1) + exitOnError(err, "error compiling module "+path) + + fmt.Printf("Compiled module %s in %v\n", path, time.Since(start)) + } + + // run tests + code := m.Run() + + err = TestRuntime.Close(ctx) + exitOnError(err, "error closing wasm runtime") + + os.Exit(code) +} + +func ChaosProcessorSpecifications() sdk.Specification { + param := config.Parameter{ + Default: "success", + Type: config.ParameterTypeString, + Description: "prefix", + Validations: []config.Validation{ + config.ValidationInclusion{List: []string{"success", "error", "panic"}}, + }, + } + + dummyProcessor := sdk.NewProcessorFunc(sdk.Specification{}, nil) + spec, err := sdk.ProcessorWithMiddleware(dummyProcessor, sdk.DefaultProcessorMiddleware()...).Specification() + if err != nil { + panic(cerrors.Errorf("failed to get specifications for middleware: %w", err)) + } + + chaosParams := map[string]config.Parameter{ + "configure": param, + "open": param, + "process.prefix": { + Default: "", + Type: config.ParameterTypeString, + Description: "prefix to be added to the payload's after", + Validations: []config.Validation{ + config.ValidationRequired{}, + }, + }, + "process": param, + "teardown": param, + } + // add parameters from middleware + maps.Copy(chaosParams, spec.Parameters) + + return sdk.Specification{ + Name: "chaos-processor", + Summary: "chaos processor summary", + Description: "chaos processor description", + Version: "v1.3.5", + Author: "Meroxa, Inc.", + Parameters: chaosParams, + } +} diff --git a/pkg/plugin/processor/standalone/command/test/build-test-processors.sh b/pkg/plugin/processor/standalone/command/test/build-test-processors.sh new file mode 100755 index 000000000..14c363590 --- /dev/null +++ b/pkg/plugin/processor/standalone/command/test/build-test-processors.sh @@ -0,0 +1,16 @@ +#!/bin/bash + +SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" + +WASM_PROCESSORS_DIR="$SCRIPT_DIR/wasm_processors" + +for dir in "$WASM_PROCESSORS_DIR"/*/; do + # Check if the directory contains a .go file + if [ -e "${dir}processor.go" ]; then + cd "$dir" || exit + + GOOS=wasip1 GOARCH=wasm go build -o processor.wasm processor.go + + cd "$WASM_PROCESSORS_DIR" || exit + fi +done diff --git a/pkg/plugin/processor/standalone/command/test/wasm_processors/chaos/processor.go b/pkg/plugin/processor/standalone/command/test/wasm_processors/chaos/processor.go new file mode 100644 index 000000000..15c8eea79 --- /dev/null +++ b/pkg/plugin/processor/standalone/command/test/wasm_processors/chaos/processor.go @@ -0,0 +1,164 @@ +// Copyright © 2023 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build wasm + +package main + +import ( + "bytes" + "context" + "errors" + "fmt" + + "github.com/conduitio/conduit-commons/config" + "github.com/conduitio/conduit-commons/opencdc" + "github.com/conduitio/conduit-commons/schema" + sdk "github.com/conduitio/conduit-processor-sdk" + procschema "github.com/conduitio/conduit-processor-sdk/schema" +) + +func main() { + sdk.Run(&chaosProcessor{}) +} + +type chaosProcessor struct { + sdk.UnimplementedProcessor + cfg map[string]string +} + +func (p *chaosProcessor) Specification() (sdk.Specification, error) { + param := config.Parameter{ + Default: "success", + Type: config.ParameterTypeString, + Description: "prefix", + Validations: []config.Validation{ + config.ValidationInclusion{List: []string{"success", "error", "panic"}}, + }, + } + return sdk.Specification{ + Name: "chaos-processor", + Summary: "chaos processor summary", + Description: "chaos processor description", + Version: "v1.3.5", + Author: "Meroxa, Inc.", + Parameters: map[string]config.Parameter{ + "configure": param, + "open": param, + "process.prefix": { + Default: "", + Type: config.ParameterTypeString, + Description: "prefix to be added to the payload's after", + Validations: []config.Validation{ + config.ValidationRequired{}, + }, + }, + "process": param, + "teardown": param, + }, + }, nil +} + +func (p *chaosProcessor) Configure(ctx context.Context, cfg config.Config) error { + p.cfg = cfg + + return p.methodBehavior(ctx, "configure") +} + +func (p *chaosProcessor) Open(ctx context.Context) error { + return p.methodBehavior(ctx, "open") +} + +func (p *chaosProcessor) Process(ctx context.Context, records []opencdc.Record) []sdk.ProcessedRecord { + err := p.methodBehavior(ctx, "process") + if err != nil { + // on error we return a single record with the error + return []sdk.ProcessedRecord{sdk.ErrorRecord{Error: err}} + } + + _, ok := p.cfg["process.prefix"] + if !ok { + return []sdk.ProcessedRecord{sdk.ErrorRecord{Error: errors.New("missing prefix")}} + } + + out := make([]sdk.ProcessedRecord, len(records)) + for i, record := range records { + original := record.Payload.After.(opencdc.RawData) + record.Payload.After = opencdc.RawData(p.cfg["process.prefix"] + string(original.Bytes())) + + out[i] = sdk.SingleRecord(record) + } + + return out +} + +func (p *chaosProcessor) Teardown(ctx context.Context) error { + return p.methodBehavior(ctx, "teardown") +} + +func (p *chaosProcessor) methodBehavior(ctx context.Context, name string) error { + switch p.cfg[name] { + case "error": + return errors.New("boom") + case "panic": + panic(name + " panic") + case "create_and_get_schema": + want := schema.Schema{ + Subject: "chaosProcessor", + Version: 1, + Type: schema.TypeAvro, + Bytes: []byte("int"), + } + + sch1, err := procschema.Create(ctx, want.Type, want.Subject, want.Bytes) + if err != nil { + return fmt.Errorf("failed to create schema: %w", err) + } + switch { + case sch1.ID == 0: + return fmt.Errorf("id is 0") + case sch1.Subject != want.Subject: + return fmt.Errorf("subjects do not match: %v != %v", sch1.Subject, want.Subject) + case sch1.Version != want.Version: + return fmt.Errorf("versions do not match: %v != %v", sch1.Version, want.Version) + case sch1.Type != want.Type: + return fmt.Errorf("types do not match: %v != %v", sch1.Type, want.Type) + case !bytes.Equal(sch1.Bytes, want.Bytes): + return fmt.Errorf("schemas do not match: %s != %s", sch1.Bytes, want.Bytes) + } + + sch2, err := procschema.Get(ctx, sch1.Subject, sch1.Version) + if err != nil { + return fmt.Errorf("failed to get schema: %w", err) + } + switch { + case sch1.ID != sch2.ID: + return fmt.Errorf("ids do not match: %v != %v", sch1.ID, sch2.ID) + case sch1.Subject != sch2.Subject: + return fmt.Errorf("subjects do not match: %v != %v", sch1.Subject, sch2.Subject) + case sch1.Version != sch2.Version: + return fmt.Errorf("versions do not match: %v != %v", sch1.Version, sch2.Version) + case sch1.Type != sch2.Type: + return fmt.Errorf("types do not match: %v != %v", sch1.Type, sch2.Type) + case !bytes.Equal(sch1.Bytes, sch2.Bytes): + return fmt.Errorf("schemas do not match: %s != %s", sch1.Bytes, sch2.Bytes) + } + + return nil + case "", "success": + return nil + default: + panic("unknown mode: " + p.cfg[name]) + } +} diff --git a/pkg/plugin/processor/standalone/command/test/wasm_processors/malformed/processor.txt b/pkg/plugin/processor/standalone/command/test/wasm_processors/malformed/processor.txt new file mode 100644 index 000000000..f13acb5c1 --- /dev/null +++ b/pkg/plugin/processor/standalone/command/test/wasm_processors/malformed/processor.txt @@ -0,0 +1 @@ +this is not a valid wasm binary \ No newline at end of file diff --git a/pkg/plugin/processor/standalone/command/test/wasm_processors/specify_error/processor.go b/pkg/plugin/processor/standalone/command/test/wasm_processors/specify_error/processor.go new file mode 100644 index 000000000..37f54aac9 --- /dev/null +++ b/pkg/plugin/processor/standalone/command/test/wasm_processors/specify_error/processor.go @@ -0,0 +1,41 @@ +// Copyright © 2023 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build wasm + +package main + +import ( + "context" + "errors" + + "github.com/conduitio/conduit-commons/opencdc" + sdk "github.com/conduitio/conduit-processor-sdk" +) + +func main() { + sdk.Run(&testProcessor{}) +} + +type testProcessor struct { + sdk.UnimplementedProcessor +} + +func (p *testProcessor) Specification() (sdk.Specification, error) { + return sdk.Specification{}, errors.New("boom") +} + +func (p *testProcessor) Process(context.Context, []opencdc.Record) []sdk.ProcessedRecord { + panic("shouldn't be called") +} diff --git a/pkg/plugin/processor/standalone/proto.go b/pkg/plugin/processor/standalone/common/fromproto/converter.go similarity index 68% rename from pkg/plugin/processor/standalone/proto.go rename to pkg/plugin/processor/standalone/common/fromproto/converter.go index 9a02259ea..871156d74 100644 --- a/pkg/plugin/processor/standalone/proto.go +++ b/pkg/plugin/processor/standalone/common/fromproto/converter.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package standalone +package fromproto import ( "fmt" @@ -25,10 +25,7 @@ import ( processorv1 "github.com/conduitio/conduit-processor-sdk/proto/processor/v1" ) -// protoConverter converts between the SDK and protobuf types. -type protoConverter struct{} - -func (c protoConverter) specification(resp *processorv1.Specify_Response) (sdk.Specification, error) { +func Specification(resp *processorv1.Specify_Response) (sdk.Specification, error) { params := make(config.Parameters, len(resp.Parameters)) err := params.FromProto(resp.Parameters) if err != nil { @@ -45,7 +42,7 @@ func (c protoConverter) specification(resp *processorv1.Specify_Response) (sdk.S }, nil } -func (c protoConverter) records(in []opencdc.Record) ([]*opencdcv1.Record, error) { +func Records(in []opencdc.Record) ([]*opencdcv1.Record, error) { if in == nil { return nil, nil } @@ -62,7 +59,7 @@ func (c protoConverter) records(in []opencdc.Record) ([]*opencdcv1.Record, error return out, nil } -func (c protoConverter) processedRecords(in []*processorv1.Process_ProcessedRecord) ([]sdk.ProcessedRecord, error) { +func ProcessedRecords(in []*processorv1.Process_ProcessedRecord) ([]sdk.ProcessedRecord, error) { if in == nil { return nil, nil } @@ -70,7 +67,7 @@ func (c protoConverter) processedRecords(in []*processorv1.Process_ProcessedReco out := make([]sdk.ProcessedRecord, len(in)) var err error for i, r := range in { - out[i], err = c.processedRecord(r) + out[i], err = ProcessedRecord(r) if err != nil { return nil, err } @@ -79,24 +76,24 @@ func (c protoConverter) processedRecords(in []*processorv1.Process_ProcessedReco return out, nil } -func (c protoConverter) processedRecord(in *processorv1.Process_ProcessedRecord) (sdk.ProcessedRecord, error) { +func ProcessedRecord(in *processorv1.Process_ProcessedRecord) (sdk.ProcessedRecord, error) { if in == nil || in.Record == nil { return nil, nil } switch v := in.Record.(type) { case *processorv1.Process_ProcessedRecord_SingleRecord: - return c.singleRecord(v) + return SingleRecord(v) case *processorv1.Process_ProcessedRecord_FilterRecord: - return c.filterRecord(v) + return FilterRecord(v) case *processorv1.Process_ProcessedRecord_ErrorRecord: - return c.errorRecord(v) + return ErrorRecord(v) default: return nil, fmt.Errorf("unknown processed record type: %T", in.Record) } } -func (c protoConverter) singleRecord(in *processorv1.Process_ProcessedRecord_SingleRecord) (sdk.SingleRecord, error) { +func SingleRecord(in *processorv1.Process_ProcessedRecord_SingleRecord) (sdk.SingleRecord, error) { if in == nil { return sdk.SingleRecord{}, nil } @@ -110,17 +107,17 @@ func (c protoConverter) singleRecord(in *processorv1.Process_ProcessedRecord_Sin return sdk.SingleRecord(rec), nil } -func (c protoConverter) filterRecord(_ *processorv1.Process_ProcessedRecord_FilterRecord) (sdk.FilterRecord, error) { +func FilterRecord(_ *processorv1.Process_ProcessedRecord_FilterRecord) (sdk.FilterRecord, error) { return sdk.FilterRecord{}, nil } -func (c protoConverter) errorRecord(in *processorv1.Process_ProcessedRecord_ErrorRecord) (sdk.ErrorRecord, error) { +func ErrorRecord(in *processorv1.Process_ProcessedRecord_ErrorRecord) (sdk.ErrorRecord, error) { if in == nil || in.ErrorRecord == nil || in.ErrorRecord.Error == nil { return sdk.ErrorRecord{}, nil } - return sdk.ErrorRecord{Error: c.error(in.ErrorRecord.Error)}, nil + return sdk.ErrorRecord{Error: Error(in.ErrorRecord.Error)}, nil } -func (c protoConverter) error(e *processorv1.Error) error { +func Error(e *processorv1.Error) error { return pprocutils.NewError(e.Code, e.Message) } diff --git a/pkg/plugin/processor/standalone/host_module.go b/pkg/plugin/processor/standalone/common/host_module.go similarity index 51% rename from pkg/plugin/processor/standalone/host_module.go rename to pkg/plugin/processor/standalone/common/host_module.go index d013fc33b..11691c03d 100644 --- a/pkg/plugin/processor/standalone/host_module.go +++ b/pkg/plugin/processor/standalone/common/host_module.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package standalone +package common import ( "bytes" @@ -21,7 +21,6 @@ import ( "github.com/conduitio/conduit-processor-sdk/pprocutils" "github.com/conduitio/conduit-processor-sdk/pprocutils/v1/fromproto" "github.com/conduitio/conduit-processor-sdk/pprocutils/v1/toproto" - processorv1 "github.com/conduitio/conduit-processor-sdk/proto/processor/v1" procutilsv1 "github.com/conduitio/conduit-processor-sdk/proto/procutils/v1" "github.com/conduitio/conduit/pkg/foundation/cerrors" "github.com/conduitio/conduit/pkg/foundation/log" @@ -30,141 +29,71 @@ import ( "google.golang.org/protobuf/proto" ) -// hostModule declares the host module that is exported to the WASM module. The +// HostModule declares the host module that is exported to the WASM module. The // host module is used to communicate between the WASM module (processor) and Conduit. -var hostModule wazergo.HostModule[*hostModuleInstance] = hostModuleFunctions{ - "command_request": wazergo.F1((*hostModuleInstance).commandRequest), - "command_response": wazergo.F1((*hostModuleInstance).commandResponse), - "create_schema": wazergo.F1((*hostModuleInstance).createSchema), - "get_schema": wazergo.F1((*hostModuleInstance).getSchema), +var HostModule wazergo.HostModule[*HostModuleInstance] = HostModuleFunctions{ + "create_schema": wazergo.F1((*HostModuleInstance).CreateSchema), + "get_schema": wazergo.F1((*HostModuleInstance).GetSchema), } -// hostModuleFunctions type implements HostModule, providing the module name, +// HostModuleFunctions type implements HostModule, providing the module name, // map of exported functions, and the ability to create instances of the module // type. -type hostModuleFunctions wazergo.Functions[*hostModuleInstance] +type HostModuleFunctions wazergo.Functions[*HostModuleInstance] // Name returns the name of the module. -func (f hostModuleFunctions) Name() string { +func (f HostModuleFunctions) Name() string { return "conduit" } // Functions is a helper that returns the exported functions of the module. -func (f hostModuleFunctions) Functions() wazergo.Functions[*hostModuleInstance] { - return (wazergo.Functions[*hostModuleInstance])(f) +func (f HostModuleFunctions) Functions() wazergo.Functions[*HostModuleInstance] { + return (wazergo.Functions[*HostModuleInstance])(f) } // Instantiate creates a new instance of the module. This is called by the // runtime when a new instance of the module is created. -func (f hostModuleFunctions) Instantiate(_ context.Context, opts ...hostModuleOption) (*hostModuleInstance, error) { - mod := &hostModuleInstance{ - parkedResponses: make(map[string]proto.Message), - lastRequestBytes: make(map[string][]byte), - } +func (f HostModuleFunctions) Instantiate(_ context.Context, opts ...HostModuleOption) (*HostModuleInstance, error) { + mod := NewHostModuleInstance() wazergo.Configure(mod, opts...) - if mod.commandRequests == nil { - return nil, cerrors.New("missing command requests channel") - } - if mod.commandResponses == nil { - return nil, cerrors.New("missing command responses channel") - } return mod, nil } -type hostModuleOption = wazergo.Option[*hostModuleInstance] +type HostModuleOption = wazergo.Option[*HostModuleInstance] -func hostModuleOptions( +func HostModuleOptions( logger log.CtxLogger, - requests <-chan *processorv1.CommandRequest, - responses chan<- tuple[*processorv1.CommandResponse, error], schemaService pprocutils.SchemaService, -) hostModuleOption { - return wazergo.OptionFunc(func(m *hostModuleInstance) { +) HostModuleOption { + return wazergo.OptionFunc(func(m *HostModuleInstance) { m.logger = logger - m.commandRequests = requests - m.commandResponses = responses m.schemaService = schemaService }) } -// hostModuleInstance is used to maintain the state of our module instance. -type hostModuleInstance struct { - logger log.CtxLogger - commandRequests <-chan *processorv1.CommandRequest - commandResponses chan<- tuple[*processorv1.CommandResponse, error] - schemaService pprocutils.SchemaService - - parkedCommandRequest *processorv1.CommandRequest - parkedResponses map[string]proto.Message - lastRequestBytes map[string][]byte +// HostModuleInstance is used to maintain the state of our module instance. +type HostModuleInstance struct { + logger log.CtxLogger + schemaService pprocutils.SchemaService + + parkedResponses map[string]proto.Message + lastRequestBytes map[string][]byte } -func (*hostModuleInstance) Close(context.Context) error { return nil } - -// commandRequest is the exported function that is called by the WASM module to -// get the next command request. It returns the size of the command request -// message. If the buffer is too small, it returns the size of the command -// request message and parks the command request. The next call to this function -// will return the same command request. -func (m *hostModuleInstance) commandRequest(ctx context.Context, buf types.Bytes) types.Uint32 { - m.logger.Trace(ctx).Msg("executing command_request") - - if m.parkedCommandRequest == nil { - // No parked command, so we need to wait for the next one. If the command - // channel is closed, then we return an error. - var ok bool - m.parkedCommandRequest, ok = <-m.commandRequests - if !ok { - return pprocutils.ErrorCodeNoMoreCommands - } - } - - // If the buffer is too small, we park the command and return the size of the - // command. The next call to nextCommand will return the same command. - if size := proto.Size(m.parkedCommandRequest); len(buf) < size { - m.logger.Warn(ctx). - Int("command_bytes", size). - Int("allocated_bytes", len(buf)). - Msgf("insufficient memory, command will be parked until next call to command_request") - return types.Uint32(size) //nolint:gosec // no risk of overflow - } - - // If the buffer is large enough, we marshal the command into the buffer and - // return the size of the command. The next call to nextCommand will return - // the next command. - out, err := proto.MarshalOptions{}.MarshalAppend(buf[:0], m.parkedCommandRequest) - if err != nil { - m.logger.Err(ctx, err).Msg("failed marshalling protobuf command request") - return pprocutils.ErrorCodeUnknownCommandRequest +func NewHostModuleInstance() *HostModuleInstance { + return &HostModuleInstance{ + parkedResponses: make(map[string]proto.Message), + lastRequestBytes: make(map[string][]byte), } - m.parkedCommandRequest = nil - - m.logger.Trace(ctx).Msg("returning next command") - return types.Uint32(len(out)) //nolint:gosec // no risk of overflow } -// commandResponse is the exported function that is called by the WASM module to -// send a command response. It returns 0 on success, or an error code on error. -func (m *hostModuleInstance) commandResponse(ctx context.Context, buf types.Bytes) types.Uint32 { - m.logger.Trace(ctx).Msg("executing command_response") +func (*HostModuleInstance) Close(context.Context) error { return nil } - var resp processorv1.CommandResponse - err := proto.Unmarshal(buf, &resp) - if err != nil { - m.logger.Err(ctx, err).Msg("failed unmarshalling protobuf command response") - m.commandResponses <- tuple[*processorv1.CommandResponse, error]{nil, err} - return pprocutils.ErrorCodeUnknownCommandResponse - } - - m.commandResponses <- tuple[*processorv1.CommandResponse, error]{&resp, nil} - return 0 -} - -// handleWasmRequest is a helper function that handles WASM requests. It +// HandleWasmRequest is a helper function that handles WASM requests. It // unmarshalls the request, calls the service function, and marshals the response. // If the buffer is too small, it parks the response and returns the size of the // response. The next call to this method will return the same response. -func (m *hostModuleInstance) handleWasmRequest( +func (m *HostModuleInstance) HandleWasmRequest( ctx context.Context, buf types.Bytes, serviceMethod string, @@ -217,10 +146,10 @@ func (m *hostModuleInstance) handleWasmRequest( return types.Uint32(len(out)) //nolint:gosec // no risk of overflow } -// createSchema is the exported function that is called by the WASM module to +// CreateSchema is the exported function that is called by the WASM module to // create a schema and set the buffer bytes to the marshalled response. // It returns the response size on success, or an error code on error. -func (m *hostModuleInstance) createSchema(ctx context.Context, buf types.Bytes) types.Uint32 { +func (m *HostModuleInstance) CreateSchema(ctx context.Context, buf types.Bytes) types.Uint32 { m.logger.Trace(ctx).Msg("executing create_schema") serviceFunc := func(ctx context.Context, buf types.Bytes) (proto.Message, error) { @@ -237,13 +166,13 @@ func (m *hostModuleInstance) createSchema(ctx context.Context, buf types.Bytes) return toproto.CreateSchemaResponse(schemaResp), nil } - return m.handleWasmRequest(ctx, buf, "create_schema", serviceFunc) + return m.HandleWasmRequest(ctx, buf, "create_schema", serviceFunc) } -// getSchema is the exported function that is called by the WASM module to +// GetSchema is the exported function that is called by the WASM module to // get a schema and set the buffer bytes to the marshalled response. -// It returns the response size on success, or an error code on error. -func (m *hostModuleInstance) getSchema(ctx context.Context, buf types.Bytes) types.Uint32 { +// It returns the response size on success or an error code on error. +func (m *HostModuleInstance) GetSchema(ctx context.Context, buf types.Bytes) types.Uint32 { m.logger.Trace(ctx).Msg("executing get_schema") serviceFunc := func(ctx context.Context, buf types.Bytes) (proto.Message, error) { @@ -260,5 +189,5 @@ func (m *hostModuleInstance) getSchema(ctx context.Context, buf types.Bytes) typ return toproto.GetSchemaResponse(schemaResp), nil } - return m.handleWasmRequest(ctx, buf, "get_schema", serviceFunc) + return m.HandleWasmRequest(ctx, buf, "get_schema", serviceFunc) } diff --git a/pkg/plugin/processor/standalone/logger.go b/pkg/plugin/processor/standalone/common/logger.go similarity index 83% rename from pkg/plugin/processor/standalone/logger.go rename to pkg/plugin/processor/standalone/common/logger.go index a6fed12c4..f97eed965 100644 --- a/pkg/plugin/processor/standalone/logger.go +++ b/pkg/plugin/processor/standalone/common/logger.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package standalone +package common import ( "io" @@ -23,27 +23,27 @@ import ( "github.com/tetratelabs/wazero" ) -// wasmLogWriter is a logger adapter for the WASM stderr and stdout streams. +// WasmLogWriter is a logger adapter for the WASM stderr and stdout streams. // It parses the JSON log events and emits them as structured logs. It expects // the log events to be in the default format produced by zerolog. If the // parsing fails, it falls back to writing the raw bytes as-is. -type wasmLogWriter struct { +type WasmLogWriter struct { logger zerolog.Logger } -var _ io.Writer = (*wasmLogWriter)(nil) +var _ io.Writer = (*WasmLogWriter)(nil) -func newWasmLogWriter(logger log.CtxLogger, module wazero.CompiledModule) wasmLogWriter { +func NewWasmLogWriter(logger log.CtxLogger, module wazero.CompiledModule) WasmLogWriter { name := module.Name() if name == "" { // no module name, use the component name instead name = logger.Component() + ".module" } logger = logger.WithComponent(name) - return wasmLogWriter{logger: logger.ZerologWithComponent()} + return WasmLogWriter{logger: logger.ZerologWithComponent()} } -func (l wasmLogWriter) Write(p []byte) (int, error) { +func (l WasmLogWriter) Write(p []byte) (int, error) { err := l.emitJSONEvent(p) if err != nil { // fallback to writing the bytes as-is @@ -52,7 +52,7 @@ func (l wasmLogWriter) Write(p []byte) (int, error) { return len(p), nil } -func (l wasmLogWriter) emitJSONEvent(p []byte) error { +func (l WasmLogWriter) emitJSONEvent(p []byte) error { var raw map[string]any err := json.Unmarshal(p, &raw) if err != nil { diff --git a/pkg/plugin/processor/standalone/reactor/builder.go b/pkg/plugin/processor/standalone/reactor/builder.go new file mode 100644 index 000000000..3243c7bc3 --- /dev/null +++ b/pkg/plugin/processor/standalone/reactor/builder.go @@ -0,0 +1,66 @@ +// Copyright © 2025 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package reactor + +import ( + "context" + + "github.com/conduitio/conduit-processor-sdk/pprocutils" + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/stealthrocket/wazergo" + "github.com/tetratelabs/wazero" +) + +type Builder struct { + logger log.CtxLogger + + runtime wazero.Runtime + // hostModule is the conduit host module that exposes Conduit host functions + // to the WASM module. The host module is compiled once and instantiated + // multiple times, once for each WASM module. + hostModule *wazergo.CompiledModule[*HostModuleInstance] + schemaService pprocutils.SchemaService +} + +func NewBuilder( + ctx context.Context, + logger log.CtxLogger, + runtime wazero.Runtime, + schemaService pprocutils.SchemaService, +) (*Builder, error) { + logger = logger.WithComponentFromType(Builder{}) + + // init host module + compiledHostModule, err := wazergo.Compile(ctx, runtime, HostModule) + if err != nil { + return nil, cerrors.Errorf("failed to compile host module: %w", err) + } + + return &Builder{ + logger: logger, + runtime: runtime, + hostModule: compiledHostModule, + schemaService: schemaService, + }, nil +} + +func (b *Builder) Build( + ctx context.Context, + processorModule wazero.CompiledModule, + id string, +) (*WasmProcessor, error) { + return NewWasmProcessor(ctx, b.runtime, processorModule, b.hostModule, b.schemaService, id, b.logger) +} diff --git a/pkg/plugin/processor/standalone/reactor/exported.go b/pkg/plugin/processor/standalone/reactor/exported.go new file mode 100644 index 000000000..0445f3cc1 --- /dev/null +++ b/pkg/plugin/processor/standalone/reactor/exported.go @@ -0,0 +1,170 @@ +// Copyright © 2025 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package reactor + +import ( + "fmt" + "strings" + + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/tetratelabs/wazero/api" +) + +type functionDefinition struct { + name string + paramTypes []api.ValueType // parameter types + resultTypes []api.ValueType // result types +} + +var ( + mallocFunctionDefinition = functionDefinition{ + name: "conduit.processor.v1.malloc", + paramTypes: []api.ValueType{api.ValueTypeI32}, // i32 (size of the buffer) + resultTypes: []api.ValueType{api.ValueTypeI32}, // i32 (pointer to the allocated buffer) + } + specificationFunctionDefinition = functionDefinition{ + name: "conduit.processor.v1.specification", + paramTypes: []api.ValueType{ + api.ValueTypeI32, // i32 (pointer to buffer) + api.ValueTypeI32, // i32 (size of the buffer) + }, + resultTypes: []api.ValueType{api.ValueTypeI64}, // i64 (pointer and size of the buffer packed in a single i64) + } + configureFunctionDefinition = functionDefinition{ + name: "conduit.processor.v1.configure", + paramTypes: []api.ValueType{ + api.ValueTypeI32, // i32 (pointer to buffer) + api.ValueTypeI32, // i32 (size of the buffer) + }, + resultTypes: []api.ValueType{api.ValueTypeI64}, // i64 (pointer and size of the buffer packed in a single i64) + } + openFunctionDefinition = functionDefinition{ + name: "conduit.processor.v1.open", + paramTypes: []api.ValueType{ + api.ValueTypeI32, // i32 (pointer to buffer) + api.ValueTypeI32, // i32 (size of the buffer) + }, + resultTypes: []api.ValueType{api.ValueTypeI64}, // i64 (pointer and size of the buffer packed in a single i64) + } + processFunctionDefinition = functionDefinition{ + name: "conduit.processor.v1.process", + paramTypes: []api.ValueType{ + api.ValueTypeI32, // i32 (pointer to buffer) + api.ValueTypeI32, // i32 (size of the buffer) + }, + resultTypes: []api.ValueType{api.ValueTypeI64}, // i64 (pointer and size of the buffer packed in a single i64) + } + teardownFunctionDefinition = functionDefinition{ + name: "conduit.processor.v1.teardown", + paramTypes: []api.ValueType{ + api.ValueTypeI32, // i32 (pointer to buffer) + api.ValueTypeI32, // i32 (size of the buffer) + }, + resultTypes: []api.ValueType{api.ValueTypeI64}, // i64 (pointer and size of the buffer packed in a single i64) + } +) + +// getExportedFunction retrieves an exported function from the given module +// and checks if it matches the expected function definition. It returns the +// function if it exists and matches the expected definition, or an error if it +// does not exist or does not match the expected definition. +func getExportedFunction(module api.Module, wantFn functionDefinition) (fn api.Function, err error) { + fn = module.ExportedFunction(wantFn.name) + var def api.FunctionDefinition + + func() { + // Recover from panic that occurs if the function does not exist. + defer func() { + if recover() != nil { + fn = nil + err = cerrors.Errorf("exported function %q does not exist", wantFn.name) + } + }() + def = fn.Definition() + }() + + if !isValidFunctionDefinition(wantFn, def) { + return nil, newFunctionDefinitionError(wantFn, def.ParamTypes(), def.ResultTypes()) + } + + return fn, nil +} + +func isValidFunctionDefinition(want functionDefinition, got api.FunctionDefinition) bool { + if len(got.ParamTypes()) != len(want.paramTypes) || + len(got.ResultTypes()) != len(want.resultTypes) { + return false + } + for i, typ := range got.ParamTypes() { + if want.paramTypes[i] != typ { + return false + } + } + for i, typ := range got.ResultTypes() { + if want.resultTypes[i] != typ { + return false + } + } + + return true +} + +type functionDefinitionError struct { + expected functionDefinition + gotParamTypes []api.ValueType // parameter types + gotResultTypes []api.ValueType // result types +} + +func newFunctionDefinitionError(expected functionDefinition, gotParamTypes, gotResultTypes []api.ValueType) *functionDefinitionError { + return &functionDefinitionError{ + expected: expected, + gotParamTypes: gotParamTypes, + gotResultTypes: gotResultTypes, + } +} + +func (e *functionDefinitionError) Error() string { + return fmt.Sprintf( + "exported Wasm function definition mismatch, expected %s, got %s", + e.formatFunctionDefinition(e.expected.paramTypes, e.expected.resultTypes), + e.formatFunctionDefinition(e.gotParamTypes, e.gotResultTypes), + ) +} + +func (e *functionDefinitionError) formatFunctionDefinition(params []api.ValueType, results []api.ValueType) string { + var out strings.Builder + out.WriteString(e.expected.name + "(") + out.WriteString(e.formatValueTypes(params)) + out.WriteString(")") + if len(results) > 0 { + out.WriteString(" -> (") + out.WriteString(e.formatValueTypes(results)) + out.WriteString(")") + } + return out.String() +} + +func (e *functionDefinitionError) formatValueTypes(types []api.ValueType) string { + if len(types) == 0 { + return "" + } + + var out string + for _, typ := range types { + out += api.ValueTypeName(typ) + ", " + } + out = out[:len(out)-2] // Remove the trailing comma and space + return out +} diff --git a/pkg/plugin/processor/standalone/reactor/host_module.go b/pkg/plugin/processor/standalone/reactor/host_module.go new file mode 100644 index 000000000..0cb9f430d --- /dev/null +++ b/pkg/plugin/processor/standalone/reactor/host_module.go @@ -0,0 +1,27 @@ +// Copyright © 2025 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package reactor + +import ( + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/common" +) + +var HostModule = common.HostModule + +type HostModuleFunctions = common.HostModuleFunctions +type HostModuleOption = common.HostModuleOption +type HostModuleInstance = common.HostModuleInstance + +var HostModuleOptions = common.HostModuleOptions diff --git a/pkg/plugin/processor/standalone/reactor/processor.go b/pkg/plugin/processor/standalone/reactor/processor.go new file mode 100644 index 000000000..06c3e6b32 --- /dev/null +++ b/pkg/plugin/processor/standalone/reactor/processor.go @@ -0,0 +1,358 @@ +// Copyright © 2025 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package reactor + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/conduitio/conduit-commons/config" + "github.com/conduitio/conduit-commons/opencdc" + sdk "github.com/conduitio/conduit-processor-sdk" + "github.com/conduitio/conduit-processor-sdk/pprocutils" + processorv1 "github.com/conduitio/conduit-processor-sdk/proto/processor/v1" + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/common" + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/common/fromproto" + "github.com/stealthrocket/wazergo" + "github.com/tetratelabs/wazero" + "github.com/tetratelabs/wazero/api" + "google.golang.org/protobuf/proto" +) + +const ( + // magicCookieKey and value are used as a very basic verification + // that a plugin is intended to be launched. This is not a security + // measure, just a UX feature. If the magic cookie doesn't match, + // we show human-friendly output. + magicCookieKey = "CONDUIT_MAGIC_COOKIE" + magicCookieValue = "3stnegqd0x02axggy0vrc4izjeq2zik6g7somyb3ye4vy5iivvjm5s1edppl5oja" + + conduitProcessorIDKey = "CONDUIT_PROCESSOR_ID" + conduitLogLevelKey = "CONDUIT_LOG_LEVEL" +) + +type WasmProcessor struct { + sdk.UnimplementedProcessor + + id string + logger log.CtxLogger + + // module is the WASM module that implements the processor + module api.Module + + mallocFn api.Function + + specificationFn api.Function + configureFn api.Function + openFn api.Function + processFn api.Function + teardownFn api.Function + + m sync.Mutex + // buf is the buffer used to communicate with the WASM module. + buf []byte + // lastMemorySize is the size of the memory when we last allocated the + // buffer in the WASM module. This is used to detect if the memory has grown, + // and the pointer might be invalidated. + lastMemorySize uint32 + // modulePointer is the pointer to the buffer in the WASM module. It is used + // to write data to the WASM module. + modulePointer uint32 +} + +func NewWasmProcessor( + ctx context.Context, + + runtime wazero.Runtime, + processorModule wazero.CompiledModule, + hostModule *wazergo.CompiledModule[*HostModuleInstance], + schemaService pprocutils.SchemaService, + + id string, + logger log.CtxLogger, +) (*WasmProcessor, error) { + logger = logger.WithComponentFromType(WasmProcessor{}) + logger.Logger = logger.With().Str(log.ProcessorIDField, id).Logger() + wasmLogger := common.NewWasmLogWriter(logger, processorModule) + + // instantiate conduit host module and inject it into the context + logger.Debug(ctx).Msg("instantiating conduit host module") + ins, err := hostModule.Instantiate( + ctx, + HostModuleOptions( + logger, + schemaService, + ), + ) + if err != nil { + return nil, fmt.Errorf("failed to instantiate conduit host module: %w", err) + } + ctx = wazergo.WithModuleInstance(ctx, ins) + + logger.Debug(ctx).Msg("instantiating processor module") + mod, err := runtime.InstantiateModule( + ctx, + processorModule, + wazero.NewModuleConfig(). + WithName(id). // ensure unique module name + WithEnv(magicCookieKey, magicCookieValue). + WithEnv(conduitProcessorIDKey, id). + WithEnv(conduitLogLevelKey, logger.GetLevel().String()). + + // set up logging + WithStdout(wasmLogger). + WithStderr(wasmLogger). + + // enable time.Now to include correct wall time + WithSysWalltime(). + // enable time.Now to include correct monotonic time + WithSysNanotime(). + // enable time.Sleep to sleep for the correct amount of time + WithSysNanosleep(). + + // don't start right away + WithStartFunctions(), + ) + if err != nil { + return nil, fmt.Errorf("failed to instantiate processor module: %w", err) + } + + p := &WasmProcessor{ + id: id, + logger: logger, + module: mod, + } + + err = p.init(ctx) + if err != nil { + _ = p.module.Close(ctx) // close the module if initialization failed + return nil, fmt.Errorf("failed to initialize processor module: %w", err) + } + + return p, nil +} + +// run is the main loop of the WASM module. It runs in a goroutine and blocks +// until the module is closed. +func (p *WasmProcessor) init(ctx context.Context) error { + _, err := p.module.ExportedFunction("_initialize").Call(ctx) + if err != nil { + return cerrors.Errorf("failed to initialize processor Wasm module: %w", err) + } + + if p.mallocFn, err = getExportedFunction(p.module, mallocFunctionDefinition); err != nil { + return err + } + if p.specificationFn, err = getExportedFunction(p.module, specificationFunctionDefinition); err != nil { + return err + } + if p.configureFn, err = getExportedFunction(p.module, configureFunctionDefinition); err != nil { + return err + } + if p.openFn, err = getExportedFunction(p.module, openFunctionDefinition); err != nil { + return err + } + if p.processFn, err = getExportedFunction(p.module, processFunctionDefinition); err != nil { + return err + } + if p.teardownFn, err = getExportedFunction(p.module, teardownFunctionDefinition); err != nil { + return err + } + + return nil +} + +func (p *WasmProcessor) Specification() (sdk.Specification, error) { + var req processorv1.Specify_Request + var resp processorv1.Specify_Response + + // the function has no context parameter, so we need to set a timeout + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + defer cancel() + + err := p.executeCall(ctx, &req, &resp, p.specificationFn) + if err != nil { + return sdk.Specification{}, fmt.Errorf("failed to execute specification command: %w", err) + } + return fromproto.Specification(&resp) +} + +func (p *WasmProcessor) Configure(ctx context.Context, config config.Config) error { + var req processorv1.Configure_Request + var resp processorv1.Configure_Response + + req.Parameters = config + + err := p.executeCall(ctx, &req, &resp, p.configureFn) + if err != nil { + return fmt.Errorf("failed to execute configure command: %w", err) + } + return nil +} + +func (p *WasmProcessor) Open(ctx context.Context) error { + var req processorv1.Open_Request + var resp processorv1.Open_Response + + err := p.executeCall(ctx, &req, &resp, p.openFn) + if err != nil { + return fmt.Errorf("failed to execute open command: %w", err) + } + return nil +} + +func (p *WasmProcessor) Process(ctx context.Context, records []opencdc.Record) []sdk.ProcessedRecord { + var req processorv1.Process_Request + var resp processorv1.Process_Response + + // Convert the records to the protobuf format expected by the WASM module. + protoRecords, err := fromproto.Records(records) + if err != nil { + p.logger.Err(ctx, err).Msg("failed to convert records to proto") + return []sdk.ProcessedRecord{sdk.ErrorRecord{Error: err}} + } + + req.Records = protoRecords + + err = p.executeCall(ctx, &req, &resp, p.processFn) + if err != nil { + p.logger.Err(ctx, err).Msg("failed to execute process command") + return []sdk.ProcessedRecord{sdk.ErrorRecord{Error: err}} + } + + // Convert the processed records back to the SDK format. + processedRecords, err := fromproto.ProcessedRecords(resp.Records) + if err != nil { + p.logger.Err(ctx, err).Msg("failed to convert processed records from proto") + return []sdk.ProcessedRecord{sdk.ErrorRecord{Error: err}} + } + return processedRecords +} + +func (p *WasmProcessor) Teardown(ctx context.Context) error { + // TODO: we should probably have a timeout for the teardown command in case + // the plugin is stuck + teardownErr := p.executeTeardownCommand(ctx) + + // close module regardless of teardown error + closeErr := p.module.CloseWithExitCode(ctx, 0) + + return cerrors.Join(teardownErr, closeErr) +} + +func (p *WasmProcessor) executeTeardownCommand(ctx context.Context) error { + var req processorv1.Teardown_Request + var resp processorv1.Teardown_Response + + // Execute the teardown command in the WASM module. + err := p.executeCall(ctx, &req, &resp, p.teardownFn) + if err != nil { + p.logger.Err(ctx, err).Msg("failed to execute teardown command") + return fmt.Errorf("failed to execute teardown command: %w", err) + } + return nil +} + +// executeCall executes a function call to the WASM module, ensuring it allocates +// enough memory and collects the response. It returns the response, or an error +// if the response is an error. +func (p *WasmProcessor) executeCall( + ctx context.Context, + req proto.Message, + resp proto.Message, + fn api.Function, +) error { + p.m.Lock() + defer p.m.Unlock() + + logger := p.logger.With().Ctx(ctx).Str("function", fn.Definition().Name()).Logger() + + // Step 1: Allocate memory in the Wasm module if needed. + if msgSize := proto.Size(req); cap(p.buf) < msgSize || + p.lastMemorySize != p.module.Memory().Size() { + logger.Debug().Msg("memory buffer is too small or memory size has changed, reallocating using malloc function") + + results, err := p.mallocFn.Call(ctx, api.EncodeI32(int32(msgSize))) + if err != nil { + logger.Err(err).Msg("failed to call Wasm function") + return cerrors.Errorf("failed to call Wasm function %q: %w", p.mallocFn.Definition().Name(), err) + } + p.modulePointer = api.DecodeU32(results[0]) + p.lastMemorySize = p.module.Memory().Size() + + if cap(p.buf) < msgSize { + p.buf = make([]byte, msgSize) + } + } + + // Step 2: Marshal the request into the buffer. + reqBytes, err := proto.MarshalOptions{}.MarshalAppend(p.buf[:0], req) + if err != nil { + logger.Err(err).Msg("failed marshalling protobuf command request") + return cerrors.Errorf("failed to marshal protobuf command request: %w", err) + } + + // Step 3: Write the request to the Wasm module's memory. + if !p.module.Memory().Write(p.modulePointer, reqBytes) { + logger.Error(). + Uint32("ptr", p.modulePointer). + Int("size", len(reqBytes)). + Msg("failed to write to Wasm module memory") + return cerrors.Errorf("failed to write to Wasm module memory at pointer %d with size %d", p.modulePointer, len(reqBytes)) + } + + // Step 4: Call the Wasm function with the pointer and size of the buffer. + results, err := fn.Call( + ctx, + api.EncodeU32(p.modulePointer), + api.EncodeU32(uint32(len(reqBytes))), + ) + if err != nil { + logger.Err(err).Msg("failed to call Wasm function") + return cerrors.Errorf("failed to call Wasm function %q: %w", fn.Definition().Name(), err) + } + + // Step 5: Check the results of the function call. + ptrSize := results[0] + ptr := uint32(ptrSize >> 32) + size := uint32(ptrSize) + + // Read the byte slice from the module's memory. + respBytes, ok := p.module.Memory().Read(ptr, size) + if !ok { + logger.Error(). + Uint32("ptr", ptr). + Uint32("size", size). + Msg("failed to read from Wasm module memory") + return cerrors.Errorf("failed to read from Wasm module memory at pointer %d with size %d", ptr, size) + } + + if err := proto.Unmarshal(respBytes, resp); err != nil { + logger.Err(err).Msg("failed to unmarshal protobuf command response") + return cerrors.Errorf("failed to unmarshal protobuf command response: %w", err) + } + + // Check if the response is an error + // TODO + // if errResp, ok := resp.Response.(*processorv1.CommandResponse_Error); ok { + // return nil, fromproto.Error(errResp.Error) + // } + + return nil +} diff --git a/pkg/plugin/processor/standalone/registry.go b/pkg/plugin/processor/standalone/registry.go index 46148f927..74a2a3952 100644 --- a/pkg/plugin/processor/standalone/registry.go +++ b/pkg/plugin/processor/standalone/registry.go @@ -27,7 +27,8 @@ import ( "github.com/conduitio/conduit/pkg/foundation/cerrors" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/conduitio/conduit/pkg/plugin" - "github.com/stealthrocket/wazergo" + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/command" + "github.com/conduitio/conduit/pkg/plugin/processor/standalone/reactor" "github.com/tetratelabs/wazero" "github.com/tetratelabs/wazero/imports/wasi_snapshot_preview1" ) @@ -46,12 +47,14 @@ type Registry struct { pluginDir string runtime wazero.Runtime - // hostModule is the conduit host module that exposes Conduit host functions - // to the WASM module. The host module is compiled once and instantiated - // multiple times, once for each WASM module. - hostModule *wazergo.CompiledModule[*hostModuleInstance] + // buildCommandProcessor is a builder function for command processors, which + // run their own main loop and call into Conduit host functions. This is the + // old way of running processors, which is still supported for backwards + // compatibility. + buildCommandProcessor processorBuilderFunc[sdk.Processor] - schemaService pprocutils.SchemaService + // TODO comment + buildReactorProcessor processorBuilderFunc[sdk.Processor] // plugins stores plugin blueprints in a 2D map, first key is the plugin // name, the second key is the plugin version @@ -65,10 +68,26 @@ type blueprint struct { specification sdk.Specification path string module wazero.CompiledModule + buildFunc processorBuilderFunc[sdk.Processor] // TODO store hash of plugin binary and compare before running the binary to // ensure someone can't switch the plugin after we registered it } +// processorBuilderFunc is a function type that can build any processor +type processorBuilderFunc[T sdk.Processor] func( + ctx context.Context, + processorModule wazero.CompiledModule, + id string, +) (T, error) + +// wrapProcessorBuilderFunc takes a function that matches the signature of +// processorBuilderFunc and returns a generic processorBuilderFunc[sdk.Processor]. +func wrapProcessorBuilderFunc[T sdk.Processor](build processorBuilderFunc[T]) processorBuilderFunc[sdk.Processor] { + return func(ctx context.Context, processorModule wazero.CompiledModule, id string) (sdk.Processor, error) { + return build(ctx, processorModule, id) + } +} + func NewRegistry(logger log.CtxLogger, pluginDir string, schemaService pprocutils.SchemaService) (*Registry, error) { // context is only used for logging, it's not used for long running operations ctx := context.Background() @@ -95,19 +114,27 @@ func NewRegistry(logger log.CtxLogger, pluginDir string, schemaService pprocutil return nil, cerrors.Errorf("failed to instantiate WASI: %w", err) } - // init host module - compiledHostModule, err := wazergo.Compile(ctx, runtime, hostModule) + commandBuilder, err := command.NewBuilder(ctx, logger, runtime, schemaService) + if err != nil { + _ = runtime.Close(ctx) + return nil, cerrors.Errorf("failed to create command processor builder: %w", err) + } + + reactorBuilder, err := reactor.NewBuilder(ctx, logger, runtime, schemaService) if err != nil { _ = runtime.Close(ctx) - return nil, cerrors.Errorf("failed to compile host module: %w", err) + return nil, cerrors.Errorf("failed to create reactor processor builder: %w", err) } r := &Registry{ - logger: logger, - runtime: runtime, - hostModule: compiledHostModule, - pluginDir: pluginDir, - schemaService: schemaService, + logger: logger, + pluginDir: pluginDir, + runtime: runtime, + + buildCommandProcessor: wrapProcessorBuilderFunc(commandBuilder.Build), + buildReactorProcessor: wrapProcessorBuilderFunc(reactorBuilder.Build), + + plugins: map[string]map[string]blueprint{}, } r.reloadPlugins() @@ -136,9 +163,9 @@ func (r *Registry) NewProcessor(ctx context.Context, fullName plugin.FullName, i return nil, cerrors.Errorf("could not find standalone processor plugin, only found versions %v: %w", availableVersions, plugin.ErrPluginNotFound) } - p, err := newWASMProcessor(ctx, r.runtime, bp.module, r.hostModule, r.schemaService, id, r.logger) + p, err := bp.buildFunc(ctx, bp.module, id) if err != nil { - return nil, cerrors.Errorf("failed to create a new WASM processor: %w", err) + return nil, cerrors.Errorf("failed to create a new Wasm processor: %w", err) } return p, nil @@ -241,16 +268,16 @@ func (r *Registry) loadPlugins(ctx context.Context, pluginDir string) map[string func (r *Registry) loadBlueprint(ctx context.Context, path string) (blueprint, error) { wasmBytes, err := os.ReadFile(path) if err != nil { - return blueprint{}, fmt.Errorf("failed to read WASM file %q: %w", path, err) + return blueprint{}, fmt.Errorf("failed to read Wasm file %q: %w", path, err) } r.logger.Debug(ctx). Str("path", path). - Msg("compiling WASM module") + Msg("compiling Wasm module") module, err := r.runtime.CompileModule(ctx, wasmBytes) if err != nil { - return blueprint{}, fmt.Errorf("failed to compile WASM module: %w", err) + return blueprint{}, fmt.Errorf("failed to compile Wasm module: %w", err) } defer func() { if err != nil { @@ -258,10 +285,14 @@ func (r *Registry) loadBlueprint(ctx context.Context, path string) (blueprint, e } }() - p, err := newWASMProcessor(ctx, r.runtime, module, r.hostModule, r.schemaService, "init-processor", r.logger) + // TODO: first try to instantiate it as a reactor processor and then fall back + // to a command processor if we fail. + buildFn := r.buildReactorProcessor + p, err := buildFn(ctx, module, "init-processor") if err != nil { - return blueprint{}, fmt.Errorf("failed to create a new WASM processor: %w", err) + return blueprint{}, fmt.Errorf("failed to create a new command Wasm processor: %w", err) } + defer func() { err := p.Teardown(ctx) if err != nil { @@ -279,6 +310,7 @@ func (r *Registry) loadBlueprint(ctx context.Context, path string) (blueprint, e specification: specs, path: path, module: module, + buildFunc: buildFn, }, nil } diff --git a/pkg/plugin/processor/standalone/standalone_test.go b/pkg/plugin/processor/standalone/standalone_test.go index db47a728a..203912ec5 100644 --- a/pkg/plugin/processor/standalone/standalone_test.go +++ b/pkg/plugin/processor/standalone/standalone_test.go @@ -26,13 +26,12 @@ import ( "github.com/conduitio/conduit-commons/config" sdk "github.com/conduitio/conduit-processor-sdk" "github.com/conduitio/conduit/pkg/foundation/cerrors" - "github.com/stealthrocket/wazergo" "github.com/tetratelabs/wazero" "github.com/tetratelabs/wazero/imports/wasi_snapshot_preview1" ) const ( - testPluginDir = "./test/wasm_processors/" + testPluginDir = "./command/test/wasm_processors/" testPluginChaosDir = testPluginDir + "chaos/" testPluginMalformedDir = testPluginDir + "malformed/" @@ -41,8 +40,7 @@ const ( var ( // TestRuntime can be reused in tests to avoid recompiling the test modules - TestRuntime wazero.Runtime - CompiledHostModule *wazergo.CompiledModule[*hostModuleInstance] + TestRuntime wazero.Runtime ChaosProcessorBinary []byte MalformedProcessorBinary []byte @@ -58,6 +56,11 @@ var ( } ) +type tuple[T1, T2 any] struct { + V1 T1 + V2 T2 +} + func TestMain(m *testing.M) { exitOnError := func(err error, msg string) { if err != nil { @@ -95,7 +98,6 @@ func TestMain(m *testing.M) { _, err := wasi_snapshot_preview1.Instantiate(ctx, TestRuntime) exitOnError(err, "error instantiating WASI") - CompiledHostModule, err = wazergo.Compile(ctx, TestRuntime, hostModule) exitOnError(err, "error compiling host module") // load test processors