fix(kubernetes): informer tombstones panic the deployment/statefulset watchers

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 {

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
 }

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

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

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
 }