fix(manager): re-instate killing terminated steps

docs/EXAMPLES.md

@@ -232,15 +232,6 @@ This example creates errors randomly
 kubectl apply -f https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/301-erroring-pipeline.yaml
 ```
 
-### [Log sink](examples/302-log-sink-pipeline.yaml)
-
-This example uses a log sink to debug messages
-
-
-```
-kubectl apply -f https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/302-log-sink-pipeline.yaml
-```
-
 ### [Default Kafka config](examples/dataflow-kafka-default-secret.yaml)
 
 This is an example of providing a namespace named Kafka configuration.

manager/controllers/step_controller.go

@@ -144,24 +144,22 @@ func (r *StepReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		r.Recorder.Eventf(step, "Normal", eventReason(currentReplicas, targetReplicas), "Scaling from %d to %d", currentReplicas, targetReplicas)
 	}
 
-	deletedPods := false // we only want to delete one pod per sync
 	for _, pod := range pods.Items {
-		if i, _ := strconv.Atoi(pod.GetAnnotations()[dfv1.KeyReplica]); (i >= targetReplicas || hash != pod.GetAnnotations()[dfv1.KeyHash]) && !deletedPods {
+		if i, _ := strconv.Atoi(pod.GetAnnotations()[dfv1.KeyReplica]); i >= targetReplicas || hash != pod.GetAnnotations()[dfv1.KeyHash] {
 			log.Info("deleting excess pod", "podName", pod.Name)
 			if err := r.Client.Delete(ctx, &pod); client.IgnoreNotFound(err) != nil {
 				return ctrl.Result{}, fmt.Errorf("failed to delete excess pod %s: %w", pod.Name, err)
 			}
-			deletedPods = true
 		} else {
 			phase, message := inferPhase(pod)
-			log.Info("pod", "name", pod.Name, "phase", phase, "message", message)
 			x := dfv1.MinStepPhaseMessage(dfv1.NewStepPhaseMessage(newStatus.Phase, newStatus.Message), dfv1.NewStepPhaseMessage(phase, message))
 			newStatus.Phase, newStatus.Message = x.GetPhase(), x.GetMessage()
 			// if the main container has terminated, kill all sidecars
 			mainCtrTerminated := false
 			for _, s := range pod.Status.ContainerStatuses {
 				mainCtrTerminated = mainCtrTerminated || (s.Name == dfv1.CtrMain && s.State.Terminated != nil)
 			}
+			log.Info("pod", "name", pod.Name, "phase", phase, "message", message, "mainCtrTerminated", mainCtrTerminated)
 			if mainCtrTerminated {
 				for _, s := range pod.Status.ContainerStatuses {
 					if s.Name != dfv1.CtrMain {

manager/main.go

@@ -21,6 +21,8 @@ import (
 	"fmt"
 	"os"
 
+	"k8s.io/client-go/kubernetes"
+
 	"github.com/argoproj-labs/argo-dataflow/manager/controllers/bus"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -70,11 +72,13 @@ func main() {
 		panic(fmt.Errorf("unable to start manager: %w", err))
 	}
 
+	clientset := kubernetes.NewForConfigOrDie(restConfig)
+	containerKiller := containerkiller.New(clientset, restConfig)
 	if err = (&controllers.PipelineReconciler{
 		Client:          mgr.GetClient(),
 		Log:             ctrl.Log.WithName("controllers").WithName("Pipeline"),
 		Scheme:          mgr.GetScheme(),
-		ContainerKiller: containerkiller.Fake,
+		ContainerKiller: containerKiller,
 		Installer:       bus.NewInstaller(),
 	}).SetupWithManager(mgr); err != nil {
 		panic(fmt.Errorf("unable to create controller manager: %w", err))
@@ -85,7 +89,7 @@ func main() {
 		Log:             ctrl.Log.WithName("controllers").WithName("Step"),
 		Scheme:          mgr.GetScheme(),
 		Recorder:        mgr.GetEventRecorderFor("step-reconciler"),
-		ContainerKiller: containerkiller.Fake,
+		ContainerKiller: containerKiller,
 	}).SetupWithManager(mgr); err != nil {
 		panic(fmt.Errorf("unable to create controller manager: %w", err))
 	}

runner/sidecar/sidecar.go

@@ -495,7 +495,7 @@ func connectSink() (func([]byte) error, error) {
 				debug.Info("◷ → stan", "subject", s.Subject, "m", printable(m))
 				err := sc.Publish(s.Subject, m)
 				if err != nil {
-					withLock(func() {sinkStatues.IncErrors(sink.Name, replica, err)})
+					withLock(func() { sinkStatues.IncErrors(sink.Name, replica, err) })
 				}
 				return err
 			})