Skip to content

Add processor wasm validation #84

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ssicard merged 17 commits into from
Jun 17, 2025
Merged

Add processor wasm validation #84

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
bb4c295
adding PluginURL to allow for wasm processors
nstehr Jun 9, 2025
db25515
working code for validation
Jun 9, 2025
5d56539
renaming to processor url
nstehr Jun 9, 2025
155c7e1
add schemaregistry support
Jun 10, 2025
0b8f161
fix style
Jun 10, 2025
362d539
add unit tests for processor validation
Jun 10, 2025
5468c7e
add processor validation during webhook update
Jun 10, 2025
695d0c6
move file close, ensure not open when passing to register
Jun 11, 2025
66c9465
add tests for validation, mock out registry calls
Jun 12, 2025
24ab7fc
fix bodyclose lint err
Jun 12, 2025
075322b
fix filepath name
Jun 12, 2025
fcbbf18
fix webhook tests
Jun 12, 2025
ad2a909
fix code style
Jun 12, 2025
d0b8a53
fix validator init ordering
Jun 12, 2025
362b3ad
remove comment
Jun 12, 2025
a211942
Update internal/testutil/conduit.go
ssicard Jun 17, 2025
5b035c8
remove debug statements
Jun 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions api/v1alpha/conduit_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,11 +91,12 @@ type ConduitConnector struct {
}

type ConduitProcessor struct {
ID string `json:"id,omitempty"`
Name string `json:"name,omitempty"`
Plugin string `json:"plugin,omitempty"`
Condition string `json:"condition,omitempty"`
Workers int `json:"workers,omitempty"`
ID string `json:"id,omitempty"`
Name string `json:"name,omitempty"`
Plugin string `json:"plugin,omitempty"`
ProcessorURL string `json:"processorURL,omitempty"`
Condition string `json:"condition,omitempty"`
Workers int `json:"workers,omitempty"`

Settings []SettingsVar `json:"settings,omitempty"`
}
Expand Down
4 changes: 4 additions & 0 deletions charts/conduit-operator/templates/crd.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,8 @@ spec:
type: string
plugin:
type: string
processorURL:
type: string
settings:
items:
properties:
Expand Down Expand Up @@ -188,6 +190,8 @@ spec:
type: string
plugin:
type: string
processorURL:
type: string
settings:
items:
properties:
Expand Down
4 changes: 4 additions & 0 deletions config/crd/bases/operator.conduit.io_conduits.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@ spec:
type: string
plugin:
type: string
processorURL:
type: string
settings:
items:
properties:
Expand Down Expand Up @@ -177,6 +179,8 @@ spec:
type: string
plugin:
type: string
processorURL:
type: string
settings:
items:
properties:
Expand Down
108 changes: 103 additions & 5 deletions internal/testutil/conduit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@ func SetupSampleConduit(t *testing.T) *v1alpha.Conduit {
Name: "my-pipeline",
Version: "v0.13.2",
Description: "my-description",
Registry: &v1alpha.SchemaRegistry{
URL: "http://apicurio:8080/apis/ccompat/v7",
},
Connectors: []*v1alpha.ConduitConnector{
{
Name: "source-connector",
Expand Down Expand Up @@ -60,11 +63,12 @@ func SetupSampleConduit(t *testing.T) *v1alpha.Conduit {
},
Processors: []*v1alpha.ConduitProcessor{
{
ID: "proc1",
Name: "proc1",
Plugin: "builtin:base64.encode",
Workers: 2,
Condition: "{{ eq .Metadata.key \"pipeline\" }}",
ID: "proc1",
Name: "proc1",
Plugin: "builtin:base64.encode",
Workers: 2,
Condition: "{{ eq .Metadata.key \"pipeline\" }}",
ProcessorURL: "http://127.0.0.1:8090/api/files/processors/RECORD_ID/FILENAME",
Settings: []v1alpha.SettingsVar{
{
Name: "setting01",
Expand Down Expand Up @@ -103,6 +107,9 @@ func SetupBadNameConduit(t *testing.T) *v1alpha.Conduit {
Name: "my-pipeline",
Description: "my-description",
Version: "v0.13.2",
Registry: &v1alpha.SchemaRegistry{
URL: "http://apicurio:8080/apis/ccompat/v7",
},
Connectors: []*v1alpha.ConduitConnector{
{
Name: "source-connector",
Expand Down Expand Up @@ -157,6 +164,9 @@ func SetupBadValidationConduit(t *testing.T) *v1alpha.Conduit {
Image: "ghcr.io/conduitio/conduit",
Description: "my-description",
Version: "v0.13.2",
Registry: &v1alpha.SchemaRegistry{
URL: "http://apicurio:8080/apis/ccompat/v7",
},
Connectors: []*v1alpha.ConduitConnector{
{
Name: "source-connector",
Expand Down Expand Up @@ -208,6 +218,9 @@ func SetupSecretConduit(t *testing.T) *v1alpha.Conduit {
Name: "my-pipeline",
Version: "v0.13.2",
Description: "my-description",
Registry: &v1alpha.SchemaRegistry{
URL: "http://apicurio:8080/apis/ccompat/v7",
},
Connectors: []*v1alpha.ConduitConnector{
{
Name: "source-connector",
Expand Down Expand Up @@ -258,3 +271,88 @@ func SetupSecretConduit(t *testing.T) *v1alpha.Conduit {

return c
}

func SetupSourceProcConduit(t *testing.T) *v1alpha.Conduit {
t.Helper()
running := true

c := &v1alpha.Conduit{
ObjectMeta: metav1.ObjectMeta{
Name: "sample",
Namespace: "sample",
},
Spec: v1alpha.ConduitSpec{
Running: &running,
Name: "my-pipeline",
Version: "v0.13.2",
Description: "my-description",
Registry: &v1alpha.SchemaRegistry{
URL: "http://apicurio:8080/apis/ccompat/v7",
},
Connectors: []*v1alpha.ConduitConnector{
{
Name: "source-connector",
Type: "source",
Plugin: "builtin:generator",
PluginVersion: "latest",
PluginName: "builtin:generator",
Settings: []v1alpha.SettingsVar{
{
Name: "servers",
Value: "127.0.0.1",
},
{
Name: "topics",
Value: "input-topic",
},
},
Processors: []*v1alpha.ConduitProcessor{
{
ID: "proc1",
Name: "proc1",
Plugin: "standalone:foo.encode",
Workers: 2,
Condition: "{{ eq .Metadata.key \"pipeline\" }}",
ProcessorURL: "http://127.0.0.1:8090/api/files/processors/RECORD_ID/FILENAME",
Settings: []v1alpha.SettingsVar{
{
Name: "setting01",
SecretRef: &corev1.SecretKeySelector{
Key: "setting01-%p-key",
LocalObjectReference: corev1.LocalObjectReference{
Name: "setting01-secret-name",
},
},
},
{
Name: "setting02",
Value: "setting02-val",
},
},
},
},
},
{
Name: "destination-connector",
Type: "destination",
Plugin: "builtin:log",
PluginVersion: "latest",
PluginName: "builtin:log",
Settings: []v1alpha.SettingsVar{
{
Name: "servers",
Value: "127.0.0.1",
},
{
Name: "topic",
Value: "output-topic",
},
},
},
},
Processors: nil,
},
}

return c
}
66 changes: 56 additions & 10 deletions internal/webhook/v1alpha/conduit_webhook.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,12 @@ import (
"github.com/Masterminds/semver/v3"
v1alpha "github.com/conduitio/conduit-operator/api/v1alpha"
validation "github.com/conduitio/conduit-operator/pkg/conduit"
conduitLog "github.com/conduitio/conduit/pkg/foundation/log"
"github.com/conduitio/conduit/pkg/plugin/processor/procutils"
"github.com/conduitio/conduit/pkg/plugin/processor/standalone"
"github.com/conduitio/conduit/pkg/schemaregistry"
"github.com/rs/zerolog"
"github.com/twmb/franz-go/pkg/sr"
"sigs.k8s.io/controller-runtime/pkg/log"
)

Expand Down Expand Up @@ -175,21 +181,28 @@ func (v *ConduitCustomValidator) ValidateCreate(ctx context.Context, obj runtime
errs = append(errs, err)
}

if verrs := v.validateConnectors(ctx, conduit.Spec.Connectors); len(verrs) > 0 {
if err := v.validateRegistry(conduit.Spec.Registry); err != nil {
errs = append(errs, err)
}

r, err := registry(conduit.Spec.Registry, field.NewPath("schemaregistry"))
if err != nil {
errs = append(errs, err)
}

if verrs := v.validateConnectors(ctx, conduit.Spec.Connectors, r); len(verrs) > 0 {
errs = append(errs, verrs...)
}

if verrs := v.validateProcessors(
ctx,
conduit.Spec.Processors,
r,
field.NewPath("spec").Child("processors"),
); len(verrs) > 0 {
errs = append(errs, verrs...)
}

if err := v.validateRegistry(conduit.Spec.Registry); err != nil {
errs = append(errs, err)
}

if len(errs) > 0 {
return nil, apierrors.NewInvalid(v1alpha.GroupKind, conduit.Name, errs)
}
Expand All @@ -204,7 +217,16 @@ func (v *ConduitCustomValidator) ValidateUpdate(ctx context.Context, _, newObj r
return nil, fmt.Errorf("expected a Conduit object for the newObj but got %T", newObj)
}

if errs := v.validateConnectors(ctx, conduit.Spec.Connectors); len(errs) > 0 {
r, err := registry(conduit.Spec.Registry, field.NewPath("schemaregistry"))
if err != nil {
return nil, fmt.Errorf("failed creating schema registry %w", err)
}

if errs := v.validateConnectors(ctx, conduit.Spec.Connectors, r); len(errs) > 0 {
return nil, apierrors.NewInvalid(v1alpha.GroupKind, conduit.Name, errs)
}

if errs := v.validateProcessors(ctx, conduit.Spec.Processors, r, field.NewPath("spec").Child("processors")); len(errs) > 0 {
return nil, apierrors.NewInvalid(v1alpha.GroupKind, conduit.Name, errs)
}

Expand All @@ -222,7 +244,7 @@ func (v *ConduitCustomValidator) ValidateDelete(_ context.Context, obj runtime.O

// validateConnectors validates the attributes of connectors in the slice.
// Error is return when the validation fails.
func (v *ConduitCustomValidator) validateConnectors(ctx context.Context, cc []*v1alpha.ConduitConnector) field.ErrorList {
func (v *ConduitCustomValidator) validateConnectors(ctx context.Context, cc []*v1alpha.ConduitConnector, r validation.PluginRegistry) field.ErrorList {
var errs field.ErrorList

fp := field.NewPath("spec").Child("connectors")
Expand All @@ -231,7 +253,7 @@ func (v *ConduitCustomValidator) validateConnectors(ctx context.Context, cc []*v
errs = append(errs, err)
}

if procErrs := v.validateProcessors(c.Processors, fp); procErrs != nil {
if procErrs := v.validateProcessors(ctx, c.Processors, r, fp); procErrs != nil {
errs = append(errs, procErrs...)
}
}
Expand All @@ -243,11 +265,11 @@ func (v *ConduitCustomValidator) validateConnectors(ctx context.Context, cc []*v
return nil
}

func (v *ConduitCustomValidator) validateProcessors(pp []*v1alpha.ConduitProcessor, fp *field.Path) field.ErrorList {
func (v *ConduitCustomValidator) validateProcessors(ctx context.Context, pp []*v1alpha.ConduitProcessor, r validation.PluginRegistry, fp *field.Path) field.ErrorList {
var errs field.ErrorList

for _, p := range pp {
if err := v.ValidateProcessorPlugin(p, fp); err != nil {
if err := v.ValidateProcessor(ctx, p, r, fp); err != nil {
errs = append(errs, err)
}
}
Expand Down Expand Up @@ -277,6 +299,30 @@ func (*ConduitCustomValidator) validateConduitVersion(ver string) *field.Error {
return nil
}

var registryFactory = func(r *v1alpha.SchemaRegistry, fp *field.Path) (validation.PluginRegistry, *field.Error) {
Copy link
Contributor

@nstehr nstehr Jun 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just curious on your approach around creating the factory function as opposed to having one function that combines the factory implementation directly with the registry function implementation.

Copy link
Contributor Author

@ssicard ssicard Jun 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I purely did this to be able to mock out the registry call in the webhook unit tests.

cl, err := schemaregistry.NewClient(conduitLog.Nop(), sr.URLs(r.URL))
if err != nil {
return nil, field.Invalid(fp, sr.URLs(r.URL), fmt.Sprintf("failed to create schema registry: %s", err))
}

conduitLogger := conduitLog.InitLogger(zerolog.DebugLevel, conduitLog.FormatJSON)
procSchemaService := procutils.NewSchemaService(conduitLogger, cl)
reg, err := standalone.NewRegistry(conduitLogger, "/tmp", procSchemaService)
if err != nil {
return nil, field.InternalError(fp.Child("schema"), fmt.Errorf("failed to create standalone registry: %w", err))
}

return reg, nil
}

func registry(r *v1alpha.SchemaRegistry, fp *field.Path) (validation.PluginRegistry, *field.Error) {
if r == nil || r.URL == "" {
return nil, field.InternalError(fp, fmt.Errorf("registry must be set"))
}

return registryFactory(r, fp)
}

func (*ConduitCustomValidator) validateRegistry(sr *v1alpha.SchemaRegistry) *field.Error {
if sr == nil || sr.URL == "" {
return nil
Expand Down
Loading