React on pod events for readiness gates

We're relying on endpoints events to re-trigger reconciliations during
rollouts, and we're considering pod's containers status (eg. are all pod's
containers ContainerReady?) to either act upon, or swallow those events.

Both (pods changes, ep changes) can be out of sync. For instance: a starting
pod whose containers aren't all ContainerReady might have its addresses
registered to an ep subset's NotReadyAddresses, kicking a reconcile event
which won't propagate to the TargetGroups (+ conditions updates) since the pod
is evaluated as not ready. Further pod changes won't kick an endpoint change
(due to readiness gates, the pod's address will stay in NotReadyAddresses
until we do something). As probably seen in  #1205 .

In order to react reliably on pods changes, we have to hook in a pod watch.
Doing so is slightly expensive as we have to map pod -> [service ->]
endpoint -> ingress on pods events, though we limit the search to pod's ns
(services can only reference pods from their own ns, and ingress services from
their ns).

Author: bpineau

internal/ingress/controller/controller.go

@@ -115,6 +115,12 @@ func watchClusterEvents(c controller.Controller, cache cache.Cache, ingressChan
 	}); err != nil {
 		return err
 	}
+	if err := c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handlers.EnqueueRequestsForPodsEvent{
+		IngressClass: ingressClass,
+		Cache:        cache,
+	}); err != nil {
+		return err
+	}
 
 	return nil
 }

internal/ingress/controller/handlers/pod.go

@@ -0,0 +1,120 @@
+package handlers
+
+import (
+	"context"
+	"reflect"
+
+	"github.com/golang/glog"
+	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/alb/tg"
+	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/ingress/annotations/class"
+	corev1 "k8s.io/api/core/v1"
+	extensions "k8s.io/api/extensions/v1beta1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+var _ handler.EventHandler = (*EnqueueRequestsForPodsEvent)(nil)
+
+type EnqueueRequestsForPodsEvent struct {
+	IngressClass string
+	Cache        cache.Cache
+}
+
+// Create is called in response to an create event - e.g. Pod Creation.
+func (h *EnqueueRequestsForPodsEvent) Create(e event.CreateEvent, queue workqueue.RateLimitingInterface) {
+	h.enqueueImpactedIngresses(e.Object.(*corev1.Pod), queue)
+}
+
+// Update is called in response to an update event -  e.g. Pod Updated.
+func (h *EnqueueRequestsForPodsEvent) Update(e event.UpdateEvent, queue workqueue.RateLimitingInterface) {
+	podOld := e.ObjectOld.(*corev1.Pod)
+	podNew := e.ObjectNew.(*corev1.Pod)
+	if !reflect.DeepEqual(podOld, podNew) {
+		h.enqueueImpactedIngresses(podNew, queue)
+	}
+}
+
+// Delete is called in response to a delete event - e.g. Pod Deleted.
+func (h *EnqueueRequestsForPodsEvent) Delete(e event.DeleteEvent, queue workqueue.RateLimitingInterface) {
+}
+
+// Generic is called in response to an event of an unknown type or a synthetic event triggered as a cron or
+// external trigger request - e.g. reconcile Autoscaling, or a Webhook.
+func (h *EnqueueRequestsForPodsEvent) Generic(event.GenericEvent, workqueue.RateLimitingInterface) {
+}
+
+func (h *EnqueueRequestsForPodsEvent) enqueueImpactedIngresses(pod *corev1.Pod, queue workqueue.RateLimitingInterface) {
+	ingressList := &extensions.IngressList{}
+	if err := h.Cache.List(context.Background(), client.InNamespace(pod.Namespace), ingressList); err != nil {
+		glog.Errorf("failed to fetch ingresses impacted by pod %s due to %v", pod.GetName(), err)
+		return
+	}
+
+	if pod.Status.PodIP == "" {
+		return
+	}
+
+	for _, ingress := range ingressList.Items {
+		if !class.IsValidIngress(h.IngressClass, &ingress) {
+			continue
+		}
+
+		backends, _, err := tg.ExtractTargetGroupBackends(&ingress)
+		if err != nil {
+			glog.Errorf("failed to extract backend services from ingress %s/%s, reconciling the ingress. Error: %e",
+				ingress.Namespace, ingress.Name, err)
+			queue.Add(reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Namespace: ingress.Namespace,
+					Name:      ingress.Name,
+				},
+			})
+			break
+		}
+
+		for _, backend := range backends {
+			endpoint := &corev1.Endpoints{}
+			nspname := types.NamespacedName{
+				Namespace: ingress.Namespace,
+				Name:      backend.ServiceName,
+			}
+			if err = h.Cache.Get(context.Background(), nspname, endpoint); err != nil {
+				glog.Errorf("failed to fetch enpoint %s backing ingress %s/%s, ignoring",
+					backend.ServiceName, ingress.Namespace, ingress.Name)
+				continue
+			}
+
+			if h.isPodInEndpoint(pod, endpoint) {
+				queue.Add(reconcile.Request{
+					NamespacedName: types.NamespacedName{
+						Namespace: ingress.Namespace,
+						Name:      ingress.Name,
+					},
+				})
+			}
+		}
+	}
+}
+
+func (h *EnqueueRequestsForPodsEvent) isPodInEndpoint(pod *corev1.Pod, endpoint *corev1.Endpoints) bool {
+	for _, sub := range endpoint.Subsets {
+		for _, addr := range sub.Addresses {
+			if addr.IP == "" || addr.TargetRef == nil || addr.TargetRef.Kind != "Pod" || addr.TargetRef.Name != pod.Name {
+				continue
+			}
+			return true
+		}
+		for _, addr := range sub.NotReadyAddresses {
+			if addr.IP == "" || addr.TargetRef == nil || addr.TargetRef.Kind != "Pod" || addr.TargetRef.Name != pod.Name {
+				continue
+			}
+			return true
+		}
+	}
+	return false
+}