From af2380572e8398892ba3ffde7ee655209ba35dc1 Mon Sep 17 00:00:00 2001 From: Alexis Couvreur Date: Sat, 11 Jul 2026 02:30:12 -0400 Subject: [PATCH] fix(kubernetes): handle informer tombstones in deployment/statefulset watchers The deployment and statefulset event handlers asserted event objects unchecked (obj.(*appsv1.Deployment)). Informers deliver cache.DeletedFinalStateUnknown tombstones to DeleteFunc when a watch disconnect made them miss the final delete; the assertion then panics inside the informer processor goroutine and kills the whole process. The CNPG watcher already unwrapped tombstones (clusterFromObject); the two older watchers never got the same fix. All handlers now go through a shared generic eventObject[T] helper that unwraps tombstones and refuses unexpected types, and clusterFromObject delegates to it. UpdateFunc additionally guarded: Spec.Replicas is a *int32 and was dereferenced without a nil check; replicasOf defaults it to 1 like the API server does. Reproduced by feeding tombstones through the real handlers (extracted as named methods for testability): interface conversion panic before, clean Removed+Stopped events after. --- pkg/provider/kubernetes/cnpgcluster_events.go | 10 +-- pkg/provider/kubernetes/deployment_events.go | 40 +++++++--- pkg/provider/kubernetes/event_object.go | 30 ++++++++ .../kubernetes/events_tombstone_test.go | 77 +++++++++++++++++++ pkg/provider/kubernetes/statefulset_events.go | 40 +++++++--- 5 files changed, 164 insertions(+), 33 deletions(-) create mode 100644 pkg/provider/kubernetes/event_object.go create mode 100644 pkg/provider/kubernetes/events_tombstone_test.go diff --git a/pkg/provider/kubernetes/cnpgcluster_events.go b/pkg/provider/kubernetes/cnpgcluster_events.go index 3b387b6a..e1c24ed3 100644 --- a/pkg/provider/kubernetes/cnpgcluster_events.go +++ b/pkg/provider/kubernetes/cnpgcluster_events.go @@ -52,15 +52,7 @@ func (p *Provider) clusterCRDInstalled(ctx context.Context) bool { // clusterFromObject extracts the *unstructured.Unstructured from an informer event, // unwrapping a tombstone when the final state was missed. func clusterFromObject(obj any) (*unstructured.Unstructured, bool) { - if u, ok := obj.(*unstructured.Unstructured); ok { - return u, true - } - tombstone, ok := obj.(cache.DeletedFinalStateUnknown) - if !ok { - return nil, false - } - u, ok := tombstone.Obj.(*unstructured.Unstructured) - return u, ok + return eventObject[*unstructured.Unstructured](obj) } func (p *Provider) watchClusters(ctx context.Context, instance chan<- sablier.InstanceEvent, wantStopped, wantStarted, wantCreated, wantRemoved bool) cache.SharedIndexInformer { diff --git a/pkg/provider/kubernetes/deployment_events.go b/pkg/provider/kubernetes/deployment_events.go index 5b55e8d9..6df27ccb 100644 --- a/pkg/provider/kubernetes/deployment_events.go +++ b/pkg/provider/kubernetes/deployment_events.go @@ -14,12 +14,24 @@ import ( ) func (p *Provider) watchDeployments(ctx context.Context, instance chan<- sablier.InstanceEvent, wantStopped, wantStarted, wantCreated, wantRemoved bool) cache.SharedIndexInformer { - handler := cache.ResourceEventHandlerFuncs{ + handler := p.deploymentEventHandler(ctx, instance, wantStopped, wantStarted, wantCreated, wantRemoved) + factory := informers.NewSharedInformerFactoryWithOptions(p.Client, 2*time.Second, informers.WithNamespace(corev1.NamespaceAll)) + informer := factory.Apps().V1().Deployments().Informer() + + _, _ = informer.AddEventHandler(handler) + return informer +} + +func (p *Provider) deploymentEventHandler(ctx context.Context, instance chan<- sablier.InstanceEvent, wantStopped, wantStarted, wantCreated, wantRemoved bool) cache.ResourceEventHandlerFuncs { + return cache.ResourceEventHandlerFuncs{ AddFunc: func(obj any) { if !wantCreated { return } - d := obj.(*appsv1.Deployment) + d, ok := eventObject[*appsv1.Deployment](obj) + if !ok { + return + } parsed := DeploymentName(d, ParseOptions{Delimiter: p.delimiter}) info, err := p.InstanceInspect(ctx, parsed.Original) if err != nil { @@ -30,15 +42,21 @@ func (p *Provider) watchDeployments(ctx context.Context, instance chan<- sablier instance <- sablier.InstanceEvent{Type: provider.InstanceEventCreated, Info: info} }, UpdateFunc: func(old, new any) { - newDeployment := new.(*appsv1.Deployment) - oldDeployment := old.(*appsv1.Deployment) + newDeployment, ok := eventObject[*appsv1.Deployment](new) + if !ok { + return + } + oldDeployment, ok := eventObject[*appsv1.Deployment](old) + if !ok { + return + } if newDeployment.ResourceVersion == oldDeployment.ResourceVersion { return } - oldReplicas := *oldDeployment.Spec.Replicas - newReplicas := *newDeployment.Spec.Replicas + oldReplicas := replicasOf(oldDeployment.Spec.Replicas) + newReplicas := replicasOf(newDeployment.Spec.Replicas) if wantStopped && oldReplicas != 0 && newReplicas == 0 { parsed := DeploymentName(newDeployment, ParseOptions{Delimiter: p.delimiter}) @@ -65,7 +83,10 @@ func (p *Provider) watchDeployments(ctx context.Context, instance chan<- sablier if !wantRemoved && !wantStopped { return } - d := obj.(*appsv1.Deployment) + d, ok := eventObject[*appsv1.Deployment](obj) + if !ok { + return + } parsed := DeploymentName(d, ParseOptions{Delimiter: p.delimiter}) // Deployment is gone; build InstanceInfo from the deleted object directly. var image string @@ -92,9 +113,4 @@ func (p *Provider) watchDeployments(ctx context.Context, instance chan<- sablier } }, } - factory := informers.NewSharedInformerFactoryWithOptions(p.Client, 2*time.Second, informers.WithNamespace(corev1.NamespaceAll)) - informer := factory.Apps().V1().Deployments().Informer() - - _, _ = informer.AddEventHandler(handler) - return informer } diff --git a/pkg/provider/kubernetes/event_object.go b/pkg/provider/kubernetes/event_object.go new file mode 100644 index 00000000..b1623a4a --- /dev/null +++ b/pkg/provider/kubernetes/event_object.go @@ -0,0 +1,30 @@ +package kubernetes + +import "k8s.io/client-go/tools/cache" + +// eventObject extracts a typed object from an informer event, unwrapping a +// cache.DeletedFinalStateUnknown tombstone when the watch missed the final +// delete (the informer delivers tombstones on relist after a disconnect). +// An unchecked type assertion in an event handler panics inside the informer's +// processor goroutine and takes the whole process down, so every handler must +// go through this helper instead of asserting directly. +func eventObject[T any](obj any) (T, bool) { + if t, ok := obj.(T); ok { + return t, true + } + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + t, ok := tombstone.Obj.(T) + return t, ok + } + var zero T + return zero, false +} + +// replicasOf dereferences an optional replica count, defaulting to 1 exactly +// like the API server does when the field is unset. +func replicasOf(replicas *int32) int32 { + if replicas == nil { + return 1 + } + return *replicas +} diff --git a/pkg/provider/kubernetes/events_tombstone_test.go b/pkg/provider/kubernetes/events_tombstone_test.go new file mode 100644 index 00000000..6987b652 --- /dev/null +++ b/pkg/provider/kubernetes/events_tombstone_test.go @@ -0,0 +1,77 @@ +package kubernetes + +import ( + "log/slog" + "testing" + + appsv1 "k8s.io/api/apps/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/cache" + + "github.com/sablierapp/sablier/pkg/provider" + "github.com/sablierapp/sablier/pkg/sablier" + "gotest.tools/v3/assert" +) + +// The informer machinery delivers cache.DeletedFinalStateUnknown tombstones to +// DeleteFunc when a watch disconnect made it miss the final delete (relist). +// These tests feed a tombstone through the real handlers: before the fix the +// unchecked type assertions panicked inside the informer's processor +// goroutine, killing the whole process. + +func tombstoneTestProvider() *Provider { + return &Provider{l: slog.New(slog.DiscardHandler), delimiter: "_"} +} + +func TestDeploymentDeleteHandlesTombstone(t *testing.T) { + t.Parallel() + + p := tombstoneTestProvider() + events := make(chan sablier.InstanceEvent, 4) + handler := p.deploymentEventHandler(t.Context(), events, true, true, true, true) + + deleted := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: "myapp", Namespace: "demo"}, + } + handler.OnDelete(cache.DeletedFinalStateUnknown{Key: "demo/myapp", Obj: deleted}) + + ev := <-events + assert.Equal(t, provider.InstanceEventRemoved, ev.Type) + assert.Equal(t, "deployment_demo_myapp_1", ev.Info.Name) + ev = <-events + assert.Equal(t, provider.InstanceEventStopped, ev.Type) +} + +func TestStatefulSetDeleteHandlesTombstone(t *testing.T) { + t.Parallel() + + p := tombstoneTestProvider() + events := make(chan sablier.InstanceEvent, 4) + handler := p.statefulSetEventHandler(t.Context(), events, true, true, true, true) + + deleted := &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "mydb", Namespace: "demo"}, + } + handler.OnDelete(cache.DeletedFinalStateUnknown{Key: "demo/mydb", Obj: deleted}) + + ev := <-events + assert.Equal(t, provider.InstanceEventRemoved, ev.Type) + assert.Equal(t, "statefulset_demo_mydb_1", ev.Info.Name) + ev = <-events + assert.Equal(t, provider.InstanceEventStopped, ev.Type) +} + +// TestUpdateHandlesNilReplicas pins that an update event with a nil +// Spec.Replicas (legal: the field is a *int32 defaulted server-side) does not +// panic the handler. +func TestUpdateHandlesNilReplicas(t *testing.T) { + t.Parallel() + + p := tombstoneTestProvider() + events := make(chan sablier.InstanceEvent, 4) + handler := p.deploymentEventHandler(t.Context(), events, false, false, false, false) + + oldD := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: "myapp", Namespace: "demo", ResourceVersion: "1"}} + newD := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: "myapp", Namespace: "demo", ResourceVersion: "2"}} + handler.OnUpdate(oldD, newD) // Spec.Replicas nil on both: must not panic +} diff --git a/pkg/provider/kubernetes/statefulset_events.go b/pkg/provider/kubernetes/statefulset_events.go index 0ea93891..dcae6f8d 100644 --- a/pkg/provider/kubernetes/statefulset_events.go +++ b/pkg/provider/kubernetes/statefulset_events.go @@ -14,12 +14,24 @@ import ( ) func (p *Provider) watchStatefulSets(ctx context.Context, instance chan<- sablier.InstanceEvent, wantStopped, wantStarted, wantCreated, wantRemoved bool) cache.SharedIndexInformer { - handler := cache.ResourceEventHandlerFuncs{ + handler := p.statefulSetEventHandler(ctx, instance, wantStopped, wantStarted, wantCreated, wantRemoved) + factory := informers.NewSharedInformerFactoryWithOptions(p.Client, 2*time.Second, informers.WithNamespace(core_v1.NamespaceAll)) + informer := factory.Apps().V1().StatefulSets().Informer() + + _, _ = informer.AddEventHandler(handler) + return informer +} + +func (p *Provider) statefulSetEventHandler(ctx context.Context, instance chan<- sablier.InstanceEvent, wantStopped, wantStarted, wantCreated, wantRemoved bool) cache.ResourceEventHandlerFuncs { + return cache.ResourceEventHandlerFuncs{ AddFunc: func(obj any) { if !wantCreated { return } - ss := obj.(*appsv1.StatefulSet) + ss, ok := eventObject[*appsv1.StatefulSet](obj) + if !ok { + return + } parsed := StatefulSetName(ss, ParseOptions{Delimiter: p.delimiter}) info, err := p.InstanceInspect(ctx, parsed.Original) if err != nil { @@ -30,15 +42,21 @@ func (p *Provider) watchStatefulSets(ctx context.Context, instance chan<- sablie instance <- sablier.InstanceEvent{Type: provider.InstanceEventCreated, Info: info} }, UpdateFunc: func(old, new any) { - newStatefulSet := new.(*appsv1.StatefulSet) - oldStatefulSet := old.(*appsv1.StatefulSet) + newStatefulSet, ok := eventObject[*appsv1.StatefulSet](new) + if !ok { + return + } + oldStatefulSet, ok := eventObject[*appsv1.StatefulSet](old) + if !ok { + return + } if newStatefulSet.ResourceVersion == oldStatefulSet.ResourceVersion { return } - oldReplicas := *oldStatefulSet.Spec.Replicas - newReplicas := *newStatefulSet.Spec.Replicas + oldReplicas := replicasOf(oldStatefulSet.Spec.Replicas) + newReplicas := replicasOf(newStatefulSet.Spec.Replicas) if wantStopped && oldReplicas != 0 && newReplicas == 0 { parsed := StatefulSetName(newStatefulSet, ParseOptions{Delimiter: p.delimiter}) @@ -66,7 +84,10 @@ func (p *Provider) watchStatefulSets(ctx context.Context, instance chan<- sablie if !wantRemoved && !wantStopped { return } - ss := obj.(*appsv1.StatefulSet) + ss, ok := eventObject[*appsv1.StatefulSet](obj) + if !ok { + return + } parsed := StatefulSetName(ss, ParseOptions{Delimiter: p.delimiter}) // StatefulSet is gone; build InstanceInfo from the deleted object directly. var image string @@ -93,9 +114,4 @@ func (p *Provider) watchStatefulSets(ctx context.Context, instance chan<- sablie } }, } - factory := informers.NewSharedInformerFactoryWithOptions(p.Client, 2*time.Second, informers.WithNamespace(core_v1.NamespaceAll)) - informer := factory.Apps().V1().StatefulSets().Informer() - - _, _ = informer.AddEventHandler(handler) - return informer }