feat: add filesystem watcher for event-driven model re-validation

api/v1alpha1/modelvalidation_types.go

@@ -133,8 +133,21 @@
 	// Minimum interval is 1m to prevent excessive CPU usage.
 	// +kubebuilder:default="5m"
 	// +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 @@
 		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"))
 		}

cmd/validation-agent/main.go

@@ -16,6 +16,7 @@
 
 	"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 @@
 	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 @@
 		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")

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

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)
+}