From a4f00ec6259ea6e4f6978b62b585040d44ae89ee Mon Sep 17 00:00:00 2001 From: Nina Bongartz Date: Wed, 4 Nov 2026 17:38:53 +0100 Subject: [PATCH] fix: revoke readiness after repeated periodic validation failures (#162) In continuous mode the validation agent only ever set `ready = true` and never revoked it, so a model that started failing re-validation after the initial success kept serving traffic indefinitely. A tampered or swapped model was detected and logged, but the pod stayed in the Service endpoints. Add a `failureThreshold` field to `continuousValidation` that controls how many consecutive periodic failures are tolerated before the agent marks itself not-ready, which removes the pod from the Service until a later validation succeeds again. The webhook translates the field into an internal `--max-consecutive-failures` flag on the validation agent; users only ever interact with the CRD field. - `failureThreshold: 0` keeps the previous fail-open behavior (never revoke). - `failureThreshold: 1` (default) is fail-closed: a single failure removes the pod from rotation. - Higher values tolerate transient blips before revoking. The periodic loop is extracted from main() into `continuousValidationLoop` plus a pure `handleValidationResult` helper so the readiness logic is unit tested (the package had no tests before). BREAKING: the default of 1 means an existing continuous-validation deployment whose model starts failing re-validation will now be removed from rotation, where before it stayed ready. Set `failureThreshold: 0` to keep the old behavior. Fixes #162 Signed-off-by: Nina Bongartz --- README.md | 13 +- api/v1alpha1/modelvalidation_types.go | 20 +++ cmd/validation-agent/main.go | 102 ++++++++--- cmd/validation-agent/main_test.go | 169 ++++++++++++++++++ .../ml.sigstore.dev_modelvalidations.yaml | 18 ++ examples/continuous-validation.yaml | 5 + internal/webhooks/pod_webhook.go | 23 ++- internal/webhooks/pod_webhook_legacy_test.go | 52 +++++- 8 files changed, 370 insertions(+), 32 deletions(-) create mode 100644 cmd/validation-agent/main_test.go diff --git a/README.md b/README.md index 6a2f725c..8c716233 100644 --- a/README.md +++ b/README.md @@ -175,7 +175,10 @@ When continuous validation is enabled: 1. The validation container runs as a native sidecar (not just an init container) 2. After the initial validation succeeds, the container becomes ready 3. The validation repeats at the specified interval -4. On validation failure, the error is logged but the container continues running +4. On validation failure, the error is logged and, once the configured + `failureThreshold` of consecutive failures is reached, the container is marked + not-ready so Kubernetes removes the pod from the Service endpoints. A later + successful validation restores readiness. 5. The readiness probe reflects the validation state #### Configuration @@ -198,8 +201,16 @@ spec: continuousValidation: enabled: true interval: "10m" # Supports s, m, h units (e.g., "30s", "5m", "1h") + failureThreshold: 1 # Consecutive failed re-validations before the pod is + # removed from Service endpoints. Default 1 (fail-closed); + # 0 never revokes readiness (legacy fail-open behavior). ``` +> **Upgrade note:** `failureThreshold` defaults to `1`, so a pod whose model +> starts failing re-validation is now removed from rotation by default. To keep +> the previous behavior where a pod stays ready regardless of later failures, set +> `failureThreshold: 0`. Use a higher value to tolerate transient failures. + #### Requirements - Kubernetes 1.28 or later (for native sidecar support with `restartPolicy: Always`) diff --git a/api/v1alpha1/modelvalidation_types.go b/api/v1alpha1/modelvalidation_types.go index d40d47dd..c7ef02f9 100644 --- a/api/v1alpha1/modelvalidation_types.go +++ b/api/v1alpha1/modelvalidation_types.go @@ -17,6 +17,7 @@ package v1alpha1 import ( "crypto/sha256" "fmt" + "strconv" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -135,6 +136,23 @@ 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"` + + // FailureThreshold is the number of consecutive failed periodic re-validations + // after which the pod is marked not-ready and removed from the Service + // endpoints, until a later periodic validation succeeds again. + // Only used when Enabled is true. + // + // A value of 0 disables readiness revocation (fail-open): once the pod has + // become ready, it stays ready regardless of later periodic failures. This + // matches the pre-1.0 behavior. + // + // The default is 1 (fail-closed): a single failed periodic validation removes + // the pod from rotation. Set a higher value to tolerate transient failures + // (e.g. 3 only revokes readiness after three consecutive failures). + // +kubebuilder:default=1 + // +kubebuilder:validation:Minimum=0 + // +kubebuilder:validation:Optional + FailureThreshold int32 `json:"failureThreshold"` } // ModelValidationSpec defines the desired state of ModelValidation @@ -259,6 +277,8 @@ func (mv *ModelValidation) GetConfigHash() string { if mv.Spec.ContinuousValidation.Enabled { hasher.Write([]byte("continuous-enabled")) hasher.Write([]byte(mv.Spec.ContinuousValidation.Interval)) + hasher.Write([]byte("failure-threshold:" + + strconv.Itoa(int(mv.Spec.ContinuousValidation.FailureThreshold)))) } else { hasher.Write([]byte("continuous-disabled")) } diff --git a/cmd/validation-agent/main.go b/cmd/validation-agent/main.go index 813477d9..d68e2210 100644 --- a/cmd/validation-agent/main.go +++ b/cmd/validation-agent/main.go @@ -34,17 +34,25 @@ func main() { var interval time.Duration var healthPort int var skipInitial bool + var maxConsecutiveFailures int 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.IntVar(&maxConsecutiveFailures, "max-consecutive-failures", 0, + "Number of consecutive periodic validation failures after which readiness is revoked. "+ + "0 means never revoke readiness once ready (fail-open).") 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, + "maxConsecutiveFailures", maxConsecutiveFailures) // Setup signal handling ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) @@ -96,28 +104,17 @@ func main() { // Continuous validation loop if interval > 0 { - logger.Info("Starting continuous validation loop") + logger.Info("Starting continuous validation loop", "maxConsecutiveFailures", maxConsecutiveFailures) ticker := time.NewTicker(interval) defer ticker.Stop() - 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 - } - case <-ctx.Done(): - logger.Info("Shutting down gracefully") - <-healthServerDone - return - } - } + continuousValidationLoop(ctx, ticker.C, maxConsecutiveFailures, func() error { + return runValidation(ctx, validationArgs, logger, "periodic") + }, logger) + + logger.Info("Shutting down gracefully") + <-healthServerDone + return } // One-shot mode: validation complete, exit immediately (for init containers) logger.Info("One-shot validation complete, exiting") @@ -126,6 +123,54 @@ func main() { // Exit code 0 (success already handled, failures exit earlier at line 52) } +// continuousValidationLoop runs validate on every tick until ctx is cancelled. +// maxConsecutiveFailures controls when readiness is revoked after repeated +// failures; see handleValidationResult for the exact semantics. +func continuousValidationLoop( + ctx context.Context, tick <-chan time.Time, maxConsecutiveFailures int, + validate func() error, logger logr.Logger, +) { + consecutiveFailures := 0 + for { + select { + case <-tick: + logger.Info("Running periodic validation") + err := validate() + consecutiveFailures = handleValidationResult(err, consecutiveFailures, maxConsecutiveFailures) + if err != nil { + logger.Error(err, "Validation failed", + "consecutiveFailures", consecutiveFailures, + "maxConsecutiveFailures", maxConsecutiveFailures, + "ready", isReady()) + } else { + logger.Info("Validation successful") + } + case <-ctx.Done(): + return + } + } +} + +// handleValidationResult updates readiness based on a validation outcome and +// returns the new consecutive-failure count. +// +// On success the counter resets to 0 and the agent is marked ready. On failure +// the counter increments; when maxConsecutiveFailures > 0 and the counter +// reaches it, readiness is revoked (fail-closed). A maxConsecutiveFailures of 0 +// never revokes readiness (fail-open), preserving the legacy behavior. +func handleValidationResult(err error, consecutiveFailures, maxConsecutiveFailures int) int { + if err == nil { + markReady() + return 0 + } + + consecutiveFailures++ + if maxConsecutiveFailures > 0 && consecutiveFailures >= maxConsecutiveFailures { + markNotReady() + } + return consecutiveFailures +} + func runValidation(ctx context.Context, args []string, logger logr.Logger, validationType string) error { tracer := otel.Tracer("validation-agent") ctx, span := tracer.Start(ctx, "model.validate", @@ -164,10 +209,7 @@ func startHealthServer(ctx context.Context, port int, logger logr.Logger) { }) mux.HandleFunc("/ready", func(w http.ResponseWriter, _ *http.Request) { - readyMutex.RLock() - defer readyMutex.RUnlock() - - if ready { + if isReady() { w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("ready")) } else { @@ -210,3 +252,15 @@ func markReady() { defer readyMutex.Unlock() ready = true } + +func markNotReady() { + readyMutex.Lock() + defer readyMutex.Unlock() + ready = false +} + +func isReady() bool { + readyMutex.RLock() + defer readyMutex.RUnlock() + return ready +} diff --git a/cmd/validation-agent/main_test.go b/cmd/validation-agent/main_test.go new file mode 100644 index 00000000..d0acf83d --- /dev/null +++ b/cmd/validation-agent/main_test.go @@ -0,0 +1,169 @@ +// Copyright 2025 The Sigstore Authors. +// +// 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 main + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// resetReady puts the shared readiness state into a known value for a test. +// The agent uses a package-level ready flag, so tests must set it explicitly. +func resetReady(t *testing.T, value bool) { + t.Helper() + if value { + markReady() + } else { + markNotReady() + } +} + +func TestHandleValidationResult_SuccessMarksReadyAndResets(t *testing.T) { + resetReady(t, false) + + // A success after prior failures resets the counter and marks ready. + count := handleValidationResult(nil, 5, 3) + + assert.Equal(t, 0, count, "success must reset the consecutive-failure count") + assert.True(t, isReady(), "success must mark the agent ready") +} + +func TestHandleValidationResult_FailOpenNeverRevokes(t *testing.T) { + resetReady(t, true) + + // maxConsecutiveFailures == 0 means fail-open: readiness is never revoked. + count := 0 + for i := 0; i < 10; i++ { + count = handleValidationResult(errors.New("boom"), count, 0) + require.True(t, isReady(), "fail-open must keep the agent ready after failure %d", i+1) + } + assert.Equal(t, 10, count, "the counter still increments while fail-open") +} + +func TestHandleValidationResult_ThresholdOneRevokesImmediately(t *testing.T) { + resetReady(t, true) + + count := handleValidationResult(errors.New("boom"), 0, 1) + + assert.Equal(t, 1, count) + assert.False(t, isReady(), "threshold of 1 must revoke readiness on the first failure") +} + +func TestHandleValidationResult_RevokesOnlyAtThreshold(t *testing.T) { + resetReady(t, true) + + count := 0 + + count = handleValidationResult(errors.New("boom"), count, 3) + assert.Equal(t, 1, count) + assert.True(t, isReady(), "must stay ready before reaching the threshold") + + count = handleValidationResult(errors.New("boom"), count, 3) + assert.Equal(t, 2, count) + assert.True(t, isReady(), "must stay ready before reaching the threshold") + + count = handleValidationResult(errors.New("boom"), count, 3) + assert.Equal(t, 3, count) + assert.False(t, isReady(), "must revoke readiness once the threshold is reached") +} + +func TestHandleValidationResult_SuccessResetsBeforeThreshold(t *testing.T) { + resetReady(t, true) + + count := 0 + // Two failures, then a success: the counter resets so a later streak starts over. + count = handleValidationResult(errors.New("boom"), count, 3) + count = handleValidationResult(errors.New("boom"), count, 3) + require.Equal(t, 2, count) + require.True(t, isReady()) + + count = handleValidationResult(nil, count, 3) + require.Equal(t, 0, count) + require.True(t, isReady()) + + // Two more failures must not revoke readiness: the streak is only 2, not 3. + count = handleValidationResult(errors.New("boom"), count, 3) + count = handleValidationResult(errors.New("boom"), count, 3) + assert.Equal(t, 2, count) + assert.True(t, isReady(), "a success must prevent non-consecutive failures from revoking readiness") +} + +func TestHandleValidationResult_RecoversAfterRevoke(t *testing.T) { + resetReady(t, true) + + // Revoke via threshold of 1, then recover on the next success. + count := handleValidationResult(errors.New("boom"), 0, 1) + require.False(t, isReady()) + + count = handleValidationResult(nil, count, 1) + assert.Equal(t, 0, count) + assert.True(t, isReady(), "a successful validation must restore readiness") +} + +func TestContinuousValidationLoop_RevokesReadinessOnTick(t *testing.T) { + resetReady(t, true) + + tick := make(chan time.Time) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan struct{}) + go func() { + continuousValidationLoop(ctx, tick, 1, func() error { + return errors.New("boom") + }, logr.Discard()) + close(done) + }() + + tick <- time.Now() + + require.Eventually(t, func() bool { + return !isReady() + }, time.Second, 5*time.Millisecond, "loop must revoke readiness after a failing tick") + + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("loop did not return after context cancellation") + } +} + +func TestContinuousValidationLoop_StopsOnContextCancel(t *testing.T) { + resetReady(t, true) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + continuousValidationLoop(ctx, make(chan time.Time), 1, func() error { + t.Error("validate must not be called when no tick fires") + return nil + }, logr.Discard()) + close(done) + }() + + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("loop did not return after context cancellation") + } +} diff --git a/config/crd/bases/ml.sigstore.dev_modelvalidations.yaml b/config/crd/bases/ml.sigstore.dev_modelvalidations.yaml index d251fd54..c1e222f4 100644 --- a/config/crd/bases/ml.sigstore.dev_modelvalidations.yaml +++ b/config/crd/bases/ml.sigstore.dev_modelvalidations.yaml @@ -138,6 +138,24 @@ spec: When true, the validation container periodically re-validates the model. When false (default), the validation container runs as a standard init container (one-shot). type: boolean + failureThreshold: + default: 1 + description: |- + FailureThreshold is the number of consecutive failed periodic re-validations + after which the pod is marked not-ready and removed from the Service + endpoints, until a later periodic validation succeeds again. + Only used when Enabled is true. + + A value of 0 disables readiness revocation (fail-open): once the pod has + become ready, it stays ready regardless of later periodic failures. This + matches the pre-1.0 behavior. + + The default is 1 (fail-closed): a single failed periodic validation removes + the pod from rotation. Set a higher value to tolerate transient failures + (e.g. 3 only revokes readiness after three consecutive failures). + format: int32 + minimum: 0 + type: integer interval: default: 5m description: |- diff --git a/examples/continuous-validation.yaml b/examples/continuous-validation.yaml index 30f3d03e..bc3f3de4 100644 --- a/examples/continuous-validation.yaml +++ b/examples/continuous-validation.yaml @@ -13,6 +13,11 @@ spec: continuousValidation: enabled: true interval: "5m" + # Consecutive failed re-validations before the pod is removed from the + # Service endpoints. Defaults to 1 (fail-closed). Set to 0 to never revoke + # readiness once ready (legacy fail-open behavior), or higher to tolerate + # transient failures. + failureThreshold: 1 --- apiVersion: v1 kind: Pod diff --git a/internal/webhooks/pod_webhook.go b/internal/webhooks/pod_webhook.go index 33f0087a..d5f15533 100644 --- a/internal/webhooks/pod_webhook.go +++ b/internal/webhooks/pod_webhook.go @@ -220,8 +220,11 @@ 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 interval and failure-threshold flags to args + container.Args = append([]string{ + "--interval=" + interval, + "--max-consecutive-failures=" + strconv.Itoa(int(mv.Spec.ContinuousValidation.FailureThreshold)), + }, args...) // Add readiness probe (ready after first successful validation) container.ReadinessProbe = &corev1.Probe{ @@ -296,14 +299,22 @@ func buildLegacySidecarContainer( } interval := "5m" - if mv.Spec.ContinuousValidation != nil && mv.Spec.ContinuousValidation.Interval != "" { - interval = mv.Spec.ContinuousValidation.Interval + failureThreshold := int32(0) + if mv.Spec.ContinuousValidation != nil { + if mv.Spec.ContinuousValidation.Interval != "" { + interval = mv.Spec.ContinuousValidation.Interval + } + failureThreshold = mv.Spec.ContinuousValidation.FailureThreshold } - // Prepend interval flag and --skip-initial to args. + // Prepend interval, failure-threshold and --skip-initial flags 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...) + sidecarArgs := append([]string{ + "--interval=" + interval, + "--skip-initial", + "--max-consecutive-failures=" + strconv.Itoa(int(failureThreshold)), + }, args...) container := corev1.Container{ Name: constants.ModelValidationSidecarContainerName, diff --git a/internal/webhooks/pod_webhook_legacy_test.go b/internal/webhooks/pod_webhook_legacy_test.go index 1f6cc02e..cf14d880 100644 --- a/internal/webhooks/pod_webhook_legacy_test.go +++ b/internal/webhooks/pod_webhook_legacy_test.go @@ -83,9 +83,34 @@ var _ = Describe("Legacy sidecar support", func() { Expect(container.RestartPolicy).NotTo(BeNil(), "restartPolicy should be set for native sidecar") Expect(*container.RestartPolicy).To(Equal(corev1.ContainerRestartPolicyAlways)) Expect(container.Args[0]).To(Equal("--interval=10m")) + Expect(container.Args[1]).To(Equal("--max-consecutive-failures=0")) Expect(container.ReadinessProbe).NotTo(BeNil()) Expect(container.LivenessProbe).NotTo(BeNil()) }) + + It("Should pass the configured failureThreshold to the native sidecar", func() { + mv := testutil.CreateTestModelValidation(testutil.TestModelValidationOptions{ + Name: "test-mv", + Namespace: "default", + ConfigType: "sigstore", + }) + mv.Spec.ContinuousValidation = &v1alpha1.ContinuousValidation{ + Enabled: true, + Interval: "10m", + FailureThreshold: 3, + } + + pp := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: make(map[string]string), + }, + } + + args := []string{"verify", "sigstore", "--signature=/sig", "/model"} + container := buildValidationContainer(mv, args, nil, pp, nil, true) + + Expect(container.Args[1]).To(Equal("--max-consecutive-failures=3")) + }) }) Context("buildLegacySidecarContainer", func() { @@ -113,7 +138,8 @@ var _ = Describe("Legacy sidecar support", func() { Expect(container.RestartPolicy).To(BeNil(), "regular containers don't use restartPolicy") Expect(container.Args[0]).To(Equal("--interval=15m")) Expect(container.Args[1]).To(Equal("--skip-initial")) - Expect(container.Args[2:]).To(Equal(args)) + Expect(container.Args[2]).To(Equal("--max-consecutive-failures=0")) + Expect(container.Args[3:]).To(Equal(args)) Expect(container.ReadinessProbe).NotTo(BeNil()) Expect(container.LivenessProbe).NotTo(BeNil()) Expect(container.ReadinessProbe.HTTPGet.Path).To(Equal("/ready")) @@ -141,6 +167,30 @@ var _ = Describe("Legacy sidecar support", func() { Expect(container.Args[0]).To(Equal("--interval=5m")) }) + + It("Should pass the configured failureThreshold to the legacy sidecar", func() { + mv := testutil.CreateTestModelValidation(testutil.TestModelValidationOptions{ + Name: "test-mv", + Namespace: "default", + ConfigType: "sigstore", + }) + mv.Spec.ContinuousValidation = &v1alpha1.ContinuousValidation{ + Enabled: true, + Interval: "15m", + FailureThreshold: 5, + } + + pp := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: make(map[string]string), + }, + } + + args := []string{"verify", "sigstore", "--signature=/sig", "/model"} + container := buildLegacySidecarContainer(mv, args, nil, pp) + + Expect(container.Args[2]).To(Equal("--max-consecutive-failures=5")) + }) }) Context("Legacy sidecar webhook integration", func() {