From 8681af73430677d1d019cf9d95344954d4df2ac6 Mon Sep 17 00:00:00 2001 From: Nina Bongartz Date: Fri, 7 Aug 2026 16:44:47 +0200 Subject: [PATCH] feat: add filesystem watcher for event-driven model re-validation Use fsnotify/inotify to watch the model path for file changes and trigger re-validation immediately, instead of waiting for the next polling interval. The interval-based ticker remains as a fallback for network filesystems (NFS, CIFS) where inotify cannot observe remote writes. Adds --watch flag to validation-agent, Watch field to ContinuousValidation CRD, and watcher package with recursive directory watching and debounce. Resolves: SECURESIGN-3615 Co-Authored-By: Claude Opus 4.6 Signed-off-by: Nina Bongartz --- api/v1alpha1/modelvalidation_types.go | 16 ++ cmd/validation-agent/main.go | 39 ++++- go.mod | 2 +- internal/watcher/watcher.go | 143 +++++++++++++++++ internal/watcher/watcher_test.go | 212 ++++++++++++++++++++++++++ internal/webhooks/pod_webhook.go | 14 +- 6 files changed, 418 insertions(+), 8 deletions(-) create mode 100644 internal/watcher/watcher.go create mode 100644 internal/watcher/watcher_test.go diff --git a/api/v1alpha1/modelvalidation_types.go b/api/v1alpha1/modelvalidation_types.go index d40d47dd..6adc46e7 100644 --- a/api/v1alpha1/modelvalidation_types.go +++ b/api/v1alpha1/modelvalidation_types.go @@ -135,6 +135,19 @@ type ContinuousValidation struct { // +kubebuilder:validation:Pattern=`^([0-9]+(\.[0-9]+)?(m|h))+$` // +kubebuilder:validation:XValidation:rule="self == '' || duration(self) >= duration('1m')", message="interval must be at least 1m" Interval string `json:"interval,omitempty"` + + // Watch enables filesystem event-based re-validation using inotify. + // When true, the agent watches the model path for file changes and + // triggers re-validation on create/write/remove events (with debounce). + // The interval-based polling still runs as a fallback. + // + // Supported: local and block-backed storage (NVMe, SSD, HDD, iSCSI, + // Ceph RBD, local PVs, emptyDir, hostPath). + // Not supported: network filesystems (NFS, CIFS/SMB, GlusterFS) where + // the kernel does not generate inotify events for remote writes. + // + // +kubebuilder:default=false + Watch bool `json:"watch,omitempty"` } // ModelValidationSpec defines the desired state of ModelValidation @@ -259,6 +272,9 @@ func (mv *ModelValidation) GetConfigHash() string { if mv.Spec.ContinuousValidation.Enabled { hasher.Write([]byte("continuous-enabled")) hasher.Write([]byte(mv.Spec.ContinuousValidation.Interval)) + if mv.Spec.ContinuousValidation.Watch { + hasher.Write([]byte("watch-enabled")) + } } else { hasher.Write([]byte("continuous-disabled")) } diff --git a/cmd/validation-agent/main.go b/cmd/validation-agent/main.go index 813477d9..e9c191e5 100644 --- a/cmd/validation-agent/main.go +++ b/cmd/validation-agent/main.go @@ -16,6 +16,7 @@ import ( "github.com/go-logr/logr" "github.com/sigstore/model-validation-operator/internal/validation" + "github.com/sigstore/model-validation-operator/internal/watcher" "github.com/sigstore/model-validation-operator/pkg/tracing" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -34,17 +35,22 @@ func main() { var interval time.Duration var healthPort int var skipInitial bool + var watch bool flag.DurationVar(&interval, "interval", 0, "Validation interval (e.g., 5m, 1h). If 0 or not set, runs once and exits.") flag.IntVar(&healthPort, "health-port", 8080, "Health check server port") flag.BoolVar(&skipInitial, "skip-initial", false, "Skip initial validation (used with legacy sidecar mode where init container already validated)") + flag.BoolVar(&watch, "watch", false, + "Watch model path for file changes using inotify and re-validate on change. "+ + "Works on local/block storage (NVMe, SSD, Ceph RBD, etc). "+ + "Not supported on network filesystems (NFS, CIFS) — use --interval as fallback.") flag.Parse() log.SetLogger(zap.New()) logger := log.Log.WithName("validation-agent") - logger.Info("Starting validation agent", "interval", interval, "healthPort", healthPort, "skipInitial", skipInitial) + logger.Info("Starting validation agent", "interval", interval, "healthPort", healthPort, "skipInitial", skipInitial, "watch", watch) // Setup signal handling ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) @@ -100,17 +106,42 @@ func main() { ticker := time.NewTicker(interval) defer ticker.Stop() + var watchCh <-chan struct{} + if watch { + cfg, parseErr := validation.ParseArgs(validationArgs) + if parseErr != nil { + logger.Error(parseErr, "Failed to parse args for file watcher, continuing without watch") + } else { + fw := watcher.New(cfg.ModelPath, logger) + var watchErr error + watchCh, watchErr = fw.Run(ctx) + if watchErr != nil { + logger.Error(watchErr, "Failed to start file watcher, continuing with interval-only") + } + } + } + for { select { case <-ticker.C: logger.Info("Running periodic validation") if err := runValidation(ctx, validationArgs, logger, "periodic"); err != nil { logger.Error(err, "Validation failed") - // Don't unmark ready - once ready, stay ready - // This allows the pod to continue running despite transient failures } else { logger.Info("Validation successful") - markReady() // Ensure ready state persists + markReady() + } + case _, ok := <-watchCh: + if !ok { + watchCh = nil + continue + } + logger.Info("File change detected, running validation") + if err := runValidation(ctx, validationArgs, logger, "file-change"); err != nil { + logger.Error(err, "Validation failed") + } else { + logger.Info("Validation successful") + markReady() } case <-ctx.Done(): logger.Info("Shutting down gracefully") diff --git a/go.mod b/go.mod index 0e957071..e641832c 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.25.7 godebug default=go1.25 require ( + github.com/fsnotify/fsnotify v1.9.0 github.com/go-logr/logr v1.4.3 github.com/onsi/ginkgo/v2 v2.29.0 github.com/onsi/gomega v1.40.0 @@ -61,7 +62,6 @@ require ( github.com/emicklei/go-restful/v3 v3.12.2 // indirect github.com/evanphx/json-patch/v5 v5.9.11 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect - github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go new file mode 100644 index 00000000..627735e9 --- /dev/null +++ b/internal/watcher/watcher.go @@ -0,0 +1,143 @@ +// Package watcher provides filesystem event-based model change detection using inotify. +// +// Supported storage backends (local/block-backed filesystems): +// - Local PVs, hostPath, emptyDir +// - Block storage: NVMe, SSD, HDD, iSCSI, Ceph RBD, AWS EBS, GCE PD, Azure Disk +// +// Not supported (network filesystems where inotify cannot observe remote writes): +// - NFS, CIFS/SMB, GlusterFS, CephFS (FUSE-mounted) +// +// For unsupported backends, the interval-based polling fallback in the agent +// ensures models are still re-validated periodically. +package watcher + +import ( + "context" + "os" + "path/filepath" + "sync" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/go-logr/logr" +) + +const defaultDebounce = 2 * time.Second + +type Watcher struct { + path string + debounce time.Duration + logger logr.Logger + + mu sync.Mutex + timer *time.Timer +} + +func New(path string, logger logr.Logger) *Watcher { + return &Watcher{ + path: path, + debounce: defaultDebounce, + logger: logger.WithName("file-watcher"), + } +} + +// Run watches path for filesystem events and sends to the returned channel +// on debounced changes. Blocks until ctx is cancelled. +func (w *Watcher) Run(ctx context.Context) (<-chan struct{}, error) { + fsw, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + + if err := w.addRecursive(fsw, w.path); err != nil { + fsw.Close() + return nil, err + } + + ch := make(chan struct{}, 1) + + go func() { + defer fsw.Close() + defer close(ch) + + for { + select { + case event, ok := <-fsw.Events: + if !ok { + return + } + if !isRelevant(event) { + continue + } + w.logger.V(1).Info("File event detected", "name", event.Name, "op", event.Op.String()) + + if event.Op.Has(fsnotify.Create) { + w.tryAddWatch(fsw, event.Name) + } + + w.debounceSend(ch) + + case err, ok := <-fsw.Errors: + if !ok { + return + } + w.logger.Error(err, "Filesystem watcher error") + + case <-ctx.Done(): + return + } + } + }() + + w.logger.Info("Watching for file changes", "path", w.path) + return ch, nil +} + +func (w *Watcher) debounceSend(ch chan<- struct{}) { + w.mu.Lock() + defer w.mu.Unlock() + + if w.timer != nil { + w.timer.Stop() + } + w.timer = time.AfterFunc(w.debounce, func() { + select { + case ch <- struct{}{}: + default: + } + }) +} + +func (w *Watcher) addRecursive(fsw *fsnotify.Watcher, root string) error { + return filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + if err := fsw.Add(path); err != nil { + w.logger.Error(err, "Failed to watch directory", "path", path) + return err + } + } + return nil + }) +} + +func (w *Watcher) tryAddWatch(fsw *fsnotify.Watcher, path string) { + info, err := os.Stat(path) + if err != nil { + return + } + if info.IsDir() { + if err := w.addRecursive(fsw, path); err != nil { + w.logger.Error(err, "Failed to watch new directory", "path", path) + } + } +} + +func isRelevant(event fsnotify.Event) bool { + return event.Op.Has(fsnotify.Create) || + event.Op.Has(fsnotify.Write) || + event.Op.Has(fsnotify.Remove) || + event.Op.Has(fsnotify.Rename) +} diff --git a/internal/watcher/watcher_test.go b/internal/watcher/watcher_test.go new file mode 100644 index 00000000..ae9833fc --- /dev/null +++ b/internal/watcher/watcher_test.go @@ -0,0 +1,212 @@ +package watcher + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "sigs.k8s.io/controller-runtime/pkg/log/zap" +) + +func TestWatcher_FileWrite(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "model.bin") + if err := os.WriteFile(f, []byte("v1"), 0644); err != nil { + t.Fatal(err) + } + + logger := zap.New() + w := New(dir, logger) + w.debounce = 100 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ch, err := w.Run(ctx) + if err != nil { + t.Fatal(err) + } + + if err := os.WriteFile(f, []byte("v2"), 0644); err != nil { + t.Fatal(err) + } + + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for file write event") + } +} + +func TestWatcher_FileCreate(t *testing.T) { + dir := t.TempDir() + + logger := zap.New() + w := New(dir, logger) + w.debounce = 100 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ch, err := w.Run(ctx) + if err != nil { + t.Fatal(err) + } + + if err := os.WriteFile(filepath.Join(dir, "new-model.bin"), []byte("data"), 0644); err != nil { + t.Fatal(err) + } + + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for file create event") + } +} + +func TestWatcher_FileRemove(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "model.bin") + if err := os.WriteFile(f, []byte("data"), 0644); err != nil { + t.Fatal(err) + } + + logger := zap.New() + w := New(dir, logger) + w.debounce = 100 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ch, err := w.Run(ctx) + if err != nil { + t.Fatal(err) + } + + if err := os.Remove(f); err != nil { + t.Fatal(err) + } + + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for file remove event") + } +} + +func TestWatcher_SubdirectoryCreate(t *testing.T) { + dir := t.TempDir() + + logger := zap.New() + w := New(dir, logger) + w.debounce = 100 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ch, err := w.Run(ctx) + if err != nil { + t.Fatal(err) + } + + subdir := filepath.Join(dir, "submodel") + if err := os.MkdirAll(subdir, 0755); err != nil { + t.Fatal(err) + } + + // Wait for subdirectory watch to be established, then write a file in it + time.Sleep(200 * time.Millisecond) + + // Drain any event from the mkdir itself + select { + case <-ch: + case <-time.After(500 * time.Millisecond): + } + + if err := os.WriteFile(filepath.Join(subdir, "weights.bin"), []byte("data"), 0644); err != nil { + t.Fatal(err) + } + + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for event in new subdirectory") + } +} + +func TestWatcher_Debounce(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "model.bin") + if err := os.WriteFile(f, []byte("v1"), 0644); err != nil { + t.Fatal(err) + } + + logger := zap.New() + w := New(dir, logger) + w.debounce = 300 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ch, err := w.Run(ctx) + if err != nil { + t.Fatal(err) + } + + // Rapid-fire writes should produce a single debounced event + for i := range 10 { + if err := os.WriteFile(f, []byte{byte(i)}, 0644); err != nil { + t.Fatal(err) + } + time.Sleep(10 * time.Millisecond) + } + + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for debounced event") + } + + // No second event should arrive (channel buffer is 1, debounce coalesces) + select { + case <-ch: + t.Fatal("unexpected second event after debounce") + case <-time.After(500 * time.Millisecond): + } +} + +func TestWatcher_ContextCancellation(t *testing.T) { + dir := t.TempDir() + + logger := zap.New() + w := New(dir, logger) + + ctx, cancel := context.WithCancel(context.Background()) + + ch, err := w.Run(ctx) + if err != nil { + t.Fatal(err) + } + + cancel() + + // Channel should be closed after context cancellation + select { + case _, ok := <-ch: + if ok { + // Might get a buffered event, drain it + select { + case _, ok := <-ch: + if ok { + t.Fatal("channel should be closed after context cancellation") + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for channel close") + } + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for channel close after cancel") + } +} diff --git a/internal/webhooks/pod_webhook.go b/internal/webhooks/pod_webhook.go index c71fd9fe..78695a73 100644 --- a/internal/webhooks/pod_webhook.go +++ b/internal/webhooks/pod_webhook.go @@ -203,8 +203,12 @@ func buildValidationContainer( // Make it a native sidecar with restartPolicy: Always container.RestartPolicy = ptr.To(corev1.ContainerRestartPolicyAlways) - // Prepend interval flag to args - container.Args = append([]string{"--interval=" + interval}, args...) + // Prepend flags to args + flags := []string{"--interval=" + interval} + if mv.Spec.ContinuousValidation.Watch { + flags = append(flags, "--watch") + } + container.Args = append(flags, args...) // Add readiness probe (ready after first successful validation) container.ReadinessProbe = &corev1.Probe{ @@ -284,7 +288,11 @@ func buildLegacySidecarContainer( // Prepend interval flag and --skip-initial to args. // --skip-initial tells the agent to skip initial validation since the // init container already performed it. - sidecarArgs := append([]string{"--interval=" + interval, "--skip-initial"}, args...) + sidecarFlags := []string{"--interval=" + interval, "--skip-initial"} + if mv.Spec.ContinuousValidation != nil && mv.Spec.ContinuousValidation.Watch { + sidecarFlags = append(sidecarFlags, "--watch") + } + sidecarArgs := append(sidecarFlags, args...) container := corev1.Container{ Name: constants.ModelValidationSidecarContainerName,