Implement Pod Affinity and AntiAffinity for DaemonSets

pkg/controller/daemon/daemoncontroller.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"reflect"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -382,7 +383,7 @@ func (dsc *DaemonSetsController) addNode(obj interface{}) {
 	node := obj.(*v1.Node)
 	for i := range dsList.Items {
 		ds := &dsList.Items[i]
-		shouldEnqueue := dsc.nodeShouldRunDaemonPod(node, ds)
+		shouldEnqueue := dsc.nodeShouldRunDaemonPod(node, ds, nil)
 		if shouldEnqueue {
 			dsc.enqueueDaemonSet(ds)
 		}
@@ -403,7 +404,8 @@ func (dsc *DaemonSetsController) updateNode(old, cur interface{}) {
 	}
 	for i := range dsList.Items {
 		ds := &dsList.Items[i]
-		shouldEnqueue := (dsc.nodeShouldRunDaemonPod(oldNode, ds) != dsc.nodeShouldRunDaemonPod(curNode, ds))
+		// TBD: fix false positives (don't pass nil as nodeToDaemonPods)
+		shouldEnqueue := (dsc.nodeShouldRunDaemonPod(oldNode, ds, nil) != dsc.nodeShouldRunDaemonPod(curNode, ds, nil))
 		if shouldEnqueue {
 			dsc.enqueueDaemonSet(ds)
 		}
@@ -446,7 +448,7 @@ func (dsc *DaemonSetsController) manage(ds *extensions.DaemonSet) error {
 	}
 	var nodesNeedingDaemonPods, podsToDelete []string
 	for _, node := range nodeList.Items {
-		shouldRun := dsc.nodeShouldRunDaemonPod(&node, ds)
+		shouldRun := dsc.nodeShouldRunDaemonPod(&node, ds, nodeToDaemonPods)
 
 		daemonPods, isRunning := nodeToDaemonPods[node.Name]
 
@@ -574,7 +576,7 @@ func (dsc *DaemonSetsController) updateDaemonSetStatus(ds *extensions.DaemonSet)
 
 	var desiredNumberScheduled, currentNumberScheduled, numberMisscheduled, numberReady int
 	for _, node := range nodeList.Items {
-		shouldRun := dsc.nodeShouldRunDaemonPod(&node, ds)
+		shouldRun := dsc.nodeShouldRunDaemonPod(&node, ds, nodeToDaemonPods)
 
 		scheduled := len(nodeToDaemonPods[node.Name]) > 0
 
@@ -644,7 +646,7 @@ func (dsc *DaemonSetsController) syncDaemonSet(key string) error {
 	return dsc.updateDaemonSetStatus(ds)
 }
 
-func (dsc *DaemonSetsController) nodeShouldRunDaemonPod(node *v1.Node, ds *extensions.DaemonSet) bool {
+func (dsc *DaemonSetsController) nodeShouldRunDaemonPod(node *v1.Node, ds *extensions.DaemonSet, nodeToDaemonPods map[string][]*v1.Pod) bool {
 	// If the daemon set specifies a node name, check that it matches with node.Name.
 	if !(ds.Spec.Template.Spec.NodeName == "" || ds.Spec.Template.Spec.NodeName == node.Name) {
 		return false
@@ -696,6 +698,29 @@ func (dsc *DaemonSetsController) nodeShouldRunDaemonPod(node *v1.Node, ds *exten
 			}
 		}
 	}
+	if !fit {
+		return false
+	}
+
+	if nodeToDaemonPods != nil && len(nodeToDaemonPods[node.Name]) != 0 {
+		// pod (anti)affinity is currently schedule-only,
+		// so let's not delete the pod if it's already
+		// being run
+		return true
+	}
+
+	podAffinityChecker := predicates.NewPodAffinityPredicate(
+		&predicates.CachedNodeInfo{StoreToNodeLister: dsc.nodeStore},
+		dsc.podStore,
+		// problem: that should be obtained from --failure-domains flag
+		strings.Split(v1.DefaultFailureDomains, ","))
+	fit, reasons, err = podAffinityChecker(newPod, nil, nodeInfo)
+	if err != nil {
+		glog.Warningf("Pod affinity checker failed on pod %s due to unexpected error: %v", newPod.Name, err)
+	}
+	for _, r := range reasons {
+		glog.V(2).Infof("Pod affinity checker failed on pod %s for reason: %v", newPod.Name, r.GetReason())
+	}
 	return fit
 }
 

pkg/controller/daemon/daemoncontroller_test.go

@@ -41,6 +41,31 @@ var (
 	simpleNodeLabel       = map[string]string{"color": "blue", "speed": "fast"}
 	simpleNodeLabel2      = map[string]string{"color": "red", "speed": "fast"}
 	alwaysReady           = func() bool { return true }
+	podAffinity           = map[string]string{
+		v1.AffinityAnnotationKey: `
+			{"podAffinity": { "requiredDuringSchedulingIgnoredDuringExecution": [{
+				"labelSelector": {
+					"matchLabels": {
+						"foobar": "foo"
+					}
+				},
+				"topologyKey": "kubernetes.io/hostname",
+				"namespaces": []
+		}]}}`,
+	}
+	podAntiAffinity = map[string]string{
+		// make the pod "antiaffine" with itself, too
+		v1.AffinityAnnotationKey: `
+			{"podAntiAffinity": { "requiredDuringSchedulingIgnoredDuringExecution": [{
+				"labelSelector": {
+					"matchLabels": {
+						"type": "production"
+					}
+				},
+				"topologyKey": "kubernetes.io/hostname",
+				"namespaces": []
+		}]}}`,
+	}
 )
 
 func getKey(ds *extensions.DaemonSet, t *testing.T) string {
@@ -595,3 +620,55 @@ func TestNumberReadyStatus(t *testing.T) {
 		t.Errorf("Wrong daemon %s status: %v", daemon.Name, daemon.Status)
 	}
 }
+
+func testAffinityUsingSinglePod(t *testing.T, annotations map[string]string, expectedCreates int) {
+	manager, podControl := newTestController()
+	addNodes(manager.nodeStore.Store, 0, 1, map[string]string{
+		"kubernetes.io/hostname": "node-0",
+	})
+	daemon := newDaemonSet("foo")
+	daemon.Spec.Template.ObjectMeta.Annotations = annotations
+	manager.dsStore.Add(daemon)
+	syncAndValidateDaemonSets(t, manager, daemon, podControl, expectedCreates, 0)
+}
+
+func testAffinityUsingTwoPods(t *testing.T, annotations map[string]string, expectedCreates int) {
+	manager, podControl := newTestController()
+	addNodes(manager.nodeStore.Store, 0, 1, map[string]string{
+		"kubernetes.io/hostname": "node-0",
+	})
+	manager.podStore.Indexer.Add(&v1.Pod{
+		TypeMeta: metav1.TypeMeta{APIVersion: registered.GroupOrDie(v1.GroupName).GroupVersion.String()},
+		ObjectMeta: v1.ObjectMeta{
+			Name: "samplepod",
+			Labels: map[string]string{
+				"type":   "production",
+				"foobar": "foo",
+			},
+			Namespace: v1.NamespaceDefault,
+		},
+		Spec: v1.PodSpec{
+			NodeName: "default/node-0",
+		},
+	})
+	daemon := newDaemonSet("foo")
+	daemon.Spec.Template.ObjectMeta.Annotations = annotations
+	manager.dsStore.Add(daemon)
+	syncAndValidateDaemonSets(t, manager, daemon, podControl, expectedCreates, 0)
+}
+
+func TestPodAffinityDaemonDoesntLaunchNonMatchingPods(t *testing.T) {
+	testAffinityUsingSinglePod(t, podAffinity, 0)
+}
+
+func TestPodAffinityDaemonLaunchesMatchingPods(t *testing.T) {
+	testAffinityUsingTwoPods(t, podAffinity, 1)
+}
+
+func TestPodAntiAffinityDaemonLaunchesMatchingPods(t *testing.T) {
+	testAffinityUsingSinglePod(t, podAntiAffinity, 1)
+}
+
+func TestPodAntiAffinityDaemonDoesntLaunchNonMatchingPods(t *testing.T) {
+	testAffinityUsingTwoPods(t, podAntiAffinity, 0)
+}