From c4f64aa464c17888586257acf67ea8b8ca04aef5 Mon Sep 17 00:00:00 2001 From: Arsolitt Date: Sat, 30 May 2026 09:38:57 +0300 Subject: [PATCH 1/2] fix(operator): guard watch error before wrapping with watch.Filter NewRedisFailoverRetriever's WatchFunc wrapped the result of WatchRedisFailovers with watch.Filter unconditionally. On a watch error the typed client returns a nil watch.Interface, and watch.Filter spawns a goroutine whose loop dereferences the source watcher's ResultChan, crashing the operator with SIGSEGV instead of letting the reflector retry the watch. This reliably takes the operator into CrashLoopBackOff on clusters where the initial watch is flaky (e.g. a freshly bootstrapped control plane), so no RedisFailover is ever reconciled. Return (nil, err) before wrapping, mirroring the existing ListFunc guard, and add a regression test that reproduces the panic and covers the happy path. Signed-off-by: Arsolitt --- operator/redisfailover/factory.go | 12 +++-- operator/redisfailover/factory_test.go | 75 ++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 3 deletions(-) create mode 100644 operator/redisfailover/factory_test.go diff --git a/operator/redisfailover/factory.go b/operator/redisfailover/factory.go index da80503a2..00e06ffab 100644 --- a/operator/redisfailover/factory.go +++ b/operator/redisfailover/factory.go @@ -90,14 +90,20 @@ func NewRedisFailoverRetriever(cfg Config, cli k8s.Services) controller.Retrieve }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { watcher, err := cli.WatchRedisFailovers(context.Background(), "", options) - watcher = watch.Filter(watcher, func(event watch.Event) (watch.Event, bool) { + if err != nil { + // Do not wrap a nil watcher: watch.Filter starts a goroutine + // that dereferences the source watcher's ResultChan, so passing + // the nil returned alongside an error panics the operator. The + // reflector retries the watch on a returned error instead. + return nil, err + } + return watch.Filter(watcher, func(event watch.Event) (watch.Event, bool) { rf, ok := event.Object.(*redisfailoverv1.RedisFailover) if !ok { return event, false } return event, isNamespaceSupported(*rf) - }) - return watcher, err + }), nil }, }) } diff --git a/operator/redisfailover/factory_test.go b/operator/redisfailover/factory_test.go new file mode 100644 index 000000000..072e372f6 --- /dev/null +++ b/operator/redisfailover/factory_test.go @@ -0,0 +1,75 @@ +package redisfailover_test + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + + mK8SService "github.com/spotahome/redis-operator/mocks/service/k8s" + rfOperator "github.com/spotahome/redis-operator/operator/redisfailover" +) + +// TestRetrieverWatchPropagatesErrorWithoutWrapping is a regression test for the +// operator crashing with a nil pointer dereference when the RedisFailover watch +// could not be established. +// +// The retriever's WatchFunc used to call watch.Filter unconditionally on the +// result of WatchRedisFailovers. On a watch error the typed client returns a +// nil watch.Interface, and watch.Filter immediately spawns a goroutine whose +// loop dereferences the source watcher's ResultChan, panicking the whole +// operator process (SIGSEGV) instead of letting the reflector retry the watch. +// +// The fix returns (nil, err) before wrapping. This test asserts the error is +// propagated and that no filtered watcher is returned (which is what would +// otherwise carry the panicking goroutine). +func TestRetrieverWatchPropagatesErrorWithoutWrapping(t *testing.T) { + assert := assert.New(t) + + watchErr := errors.New("the server could not establish the watch") + ms := &mK8SService.Services{} + ms.On("WatchRedisFailovers", mock.Anything, mock.Anything, mock.Anything). + Return(nil, watchErr) + + retriever := rfOperator.NewRedisFailoverRetriever( + rfOperator.Config{SupportedNamespacesRegex: ".*"}, + ms, + ) + + var ( + w watch.Interface + err error + ) + assert.NotPanics(func() { + w, err = retriever.Watch(context.Background(), metav1.ListOptions{}) + }) + assert.Equal(watchErr, err) + assert.Nil(w, "a nil watcher must not be wrapped by watch.Filter") + ms.AssertExpectations(t) +} + +// TestRetrieverWatchWrapsWatcherOnSuccess verifies the happy path still wraps +// the underlying watcher (so namespace filtering stays in effect). +func TestRetrieverWatchWrapsWatcherOnSuccess(t *testing.T) { + assert := assert.New(t) + + fake := watch.NewFake() + defer fake.Stop() + ms := &mK8SService.Services{} + ms.On("WatchRedisFailovers", mock.Anything, mock.Anything, mock.Anything). + Return(fake, nil) + + retriever := rfOperator.NewRedisFailoverRetriever( + rfOperator.Config{SupportedNamespacesRegex: ".*"}, + ms, + ) + + w, err := retriever.Watch(context.Background(), metav1.ListOptions{}) + assert.NoError(err) + assert.NotNil(w) + ms.AssertExpectations(t) +} From 93fdbf30a6caa9d0f2a75d237e86bcc340dd8b34 Mon Sep 17 00:00:00 2001 From: Arsolitt Date: Sat, 30 May 2026 10:36:04 +0300 Subject: [PATCH 2/2] fix(operator): propagate watch.Error events through namespace filter watch.Error events carry a *metav1.Status object rather than a *RedisFailover, so the namespace filter's type assertion silently dropped them. The reflector depends on these events to detect a failed watch (such as an expired resource version) and restart it promptly instead of waiting for a connection timeout. Pass watch.Error events through unchanged and add a regression test covering the propagation. Signed-off-by: Arsolitt --- operator/redisfailover/factory.go | 9 ++++++ operator/redisfailover/factory_test.go | 38 ++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/operator/redisfailover/factory.go b/operator/redisfailover/factory.go index 00e06ffab..13c7382fc 100644 --- a/operator/redisfailover/factory.go +++ b/operator/redisfailover/factory.go @@ -98,6 +98,15 @@ func NewRedisFailoverRetriever(cfg Config, cli k8s.Services) controller.Retrieve return nil, err } return watch.Filter(watcher, func(event watch.Event) (watch.Event, bool) { + // Always propagate watch.Error events. Their Object is a + // *metav1.Status, not a *RedisFailover, so they would otherwise + // be dropped by the type assertion below. The reflector relies + // on these events to learn the watch has failed (e.g. an expired + // resource version) and to restart it promptly instead of + // waiting for a connection timeout. + if event.Type == watch.Error { + return event, true + } rf, ok := event.Object.(*redisfailoverv1.RedisFailover) if !ok { return event, false diff --git a/operator/redisfailover/factory_test.go b/operator/redisfailover/factory_test.go index 072e372f6..5ca58a393 100644 --- a/operator/redisfailover/factory_test.go +++ b/operator/redisfailover/factory_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -73,3 +74,40 @@ func TestRetrieverWatchWrapsWatcherOnSuccess(t *testing.T) { assert.NotNil(w) ms.AssertExpectations(t) } + +// TestRetrieverWatchPropagatesWatchErrorEvents verifies that watch.Error events +// are not swallowed by the namespace filter. Their Object is a *metav1.Status +// (not a *RedisFailover), so without an explicit passthrough the filter would +// drop them and the reflector would never learn the watch must be restarted. +func TestRetrieverWatchPropagatesWatchErrorEvents(t *testing.T) { + assert := assert.New(t) + + fake := watch.NewFake() + defer fake.Stop() + ms := &mK8SService.Services{} + ms.On("WatchRedisFailovers", mock.Anything, mock.Anything, mock.Anything). + Return(fake, nil) + + retriever := rfOperator.NewRedisFailoverRetriever( + rfOperator.Config{SupportedNamespacesRegex: ".*"}, + ms, + ) + + w, err := retriever.Watch(context.Background(), metav1.ListOptions{}) + assert.NoError(err) + assert.NotNil(w) + + errStatus := &metav1.Status{Status: metav1.StatusFailure, Reason: metav1.StatusReasonExpired} + go fake.Error(errStatus) + + select { + case event, ok := <-w.ResultChan(): + assert.True(ok, "result channel must stay open for watch.Error events") + assert.Equal(watch.Error, event.Type, "watch.Error events must pass through the filter") + assert.Equal(errStatus, event.Object) + case <-time.After(time.Second): + assert.Fail("timed out waiting for the watch.Error event to be propagated") + } + + ms.AssertExpectations(t) +}