diff --git a/operator/redisfailover/factory.go b/operator/redisfailover/factory.go index da80503a2..13c7382fc 100644 --- a/operator/redisfailover/factory.go +++ b/operator/redisfailover/factory.go @@ -90,14 +90,29 @@ 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) { + // 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 } 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..5ca58a393 --- /dev/null +++ b/operator/redisfailover/factory_test.go @@ -0,0 +1,113 @@ +package redisfailover_test + +import ( + "context" + "errors" + "testing" + "time" + + "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) +} + +// 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) +}