RFC: Handle affinity assistant deadlock for node maintenance

cmd/controller/main.go

@@ -27,6 +27,7 @@ import (
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	"github.com/tektoncd/pipeline/pkg/reconciler/customrun"
+	"github.com/tektoncd/pipeline/pkg/reconciler/node"
 	"github.com/tektoncd/pipeline/pkg/reconciler/pipelinerun"
 	"github.com/tektoncd/pipeline/pkg/reconciler/resolutionrequest"
 	"github.com/tektoncd/pipeline/pkg/reconciler/taskrun"
@@ -132,6 +133,7 @@ func main() {
 		pipelinerun.NewController(opts, clock.RealClock{}, tpPipelineRun),
 		resolutionrequest.NewController(clock.RealClock{}),
 		customrun.NewController(),
+		node.NewController(),
 	)
 
 	// Cleanly shutdown and flush telemetry when the application exits.

config/200-clusterrole.yaml

@@ -25,6 +25,9 @@ rules:
     # Controller needs to watch Pods created by TaskRuns to see them progress.
     resources: ["pods"]
     verbs: ["list", "watch"]
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["list", "watch"]
     # Controller needs cluster access to all of the CRDs that it is responsible for
     # managing.
   - apiGroups: ["tekton.dev"]

pkg/reconciler/node/controller.go

@@ -0,0 +1,41 @@
+package node
+
+import (
+	"context"
+
+	"github.com/tektoncd/pipeline/pkg/apis/config"
+
+	"github.com/tektoncd/pipeline/pkg/pipelinerunmetrics"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	nodeinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/node"
+	nodereconciler "knative.dev/pkg/client/injection/kube/reconciler/core/v1/node"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+)
+
+// NewController instantiates a new controller.Impl from knative.dev/pkg/controller
+func NewController() func(context.Context, configmap.Watcher) *controller.Impl {
+	return func(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+		logger := logging.FromContext(ctx)
+		kubeclientset := kubeclient.Get(ctx)
+		nodeInformer := nodeinformer.Get(ctx)
+		configStore := config.NewStore(logger.Named("config-store"), pipelinerunmetrics.MetricsOnStore(logger))
+		configStore.WatchConfigs(cmw)
+
+		c := &Reconciler{
+			KubeClientSet: kubeclientset,
+		}
+
+		impl := nodereconciler.NewImpl(ctx, c, func(impl *controller.Impl) controller.Options {
+			return controller.Options{
+				AgentName:         "pod",
+				SkipStatusUpdates: true,
+				ConfigStore:       configStore,
+			}
+		})
+
+		nodeInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+		return impl
+	}
+}

pkg/reconciler/node/node.go

@@ -0,0 +1,53 @@
+package node
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/tektoncd/pipeline/pkg/workspace"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	nodereconciler "knative.dev/pkg/client/injection/kube/reconciler/core/v1/node"
+	"knative.dev/pkg/logging"
+	pkgreconciler "knative.dev/pkg/reconciler"
+)
+
+// Reconciler implements controller.Reconciler for Configuration resources.
+type Reconciler struct {
+	KubeClientSet kubernetes.Interface
+}
+
+// Check that our Reconciler implements noderunreconciler.Interface
+var (
+	_ nodereconciler.Interface = (*Reconciler)(nil)
+)
+
+// ReconcileKind deletes any affinity assistant pods on unschedulable nodes
+func (c *Reconciler) ReconcileKind(ctx context.Context, n *corev1.Node) pkgreconciler.Event {
+	if !n.Spec.Unschedulable {
+		return nil
+	}
+	logger := logging.FromContext(ctx)
+	logger.Debugf("found unschedulable node %s", n.Name)
+
+	// Find all affinity assistant pods on the node
+	pods, err := c.KubeClientSet.CoreV1().Pods("").List(ctx, metav1.ListOptions{
+		LabelSelector: workspace.LabelComponent + "=" + workspace.ComponentNameAffinityAssistant,
+		FieldSelector: "spec.nodeName=" + n.Name,
+	})
+	if err != nil {
+		return fmt.Errorf("couldn't list affinity assistant pods on node %s: %s", n.Name, err)
+	}
+
+	// Delete affinity assistant pods, allowing their statefulset controllers to recreate them.
+	// Pods are deleted individually since API server rejects DeleteCollection
+	for _, pod := range pods.Items {
+		err = c.KubeClientSet.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})
+		if err != nil {
+			return fmt.Errorf("error deleting affinity assistant pod %s in ns %s: %s", pod.Name, pod.Namespace, err)
+		}
+	}
+
+	return nil
+}