Merge pull request #12943 from mesosphere/node-selector-support

Add NodeSelector support to Mesos scheduler

contrib/mesos/pkg/scheduler/podtask/pod_task.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/metrics"
 	mresource "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/resource"
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/labels"
 
 	log "github.com/golang/glog"
 	mesos "github.com/mesos/mesos-go/mesosproto"
@@ -220,6 +221,20 @@ func (t *T) AcceptOffer(offer *mesos.Offer) bool {
 		return false
 	}
 
+	// check the NodeSelector
+	if len(t.Pod.Spec.NodeSelector) > 0 {
+		slaveLabels := map[string]string{}
+		for _, a := range offer.Attributes {
+			if a.GetType() == mesos.Value_TEXT {
+				slaveLabels[a.GetName()] = a.GetText().GetValue()
+			}
+		}
+		selector := labels.SelectorFromSet(t.Pod.Spec.NodeSelector)
+		if !selector.Matches(labels.Set(slaveLabels)) {
+			return false
+		}
+	}
+
 	// check ports
 	if _, err := t.mapper.Generate(t, offer); err != nil {
 		log.V(3).Info(err)

contrib/mesos/pkg/scheduler/podtask/pod_task_test.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/resource"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -265,3 +266,55 @@ func TestGeneratePodName(t *testing.T) {
 		t.Fatalf("expected %q instead of %q", expected, name)
 	}
 }
+
+func TestNodeSelector(t *testing.T) {
+	t.Parallel()
+
+	sel1 := map[string]string{"rack": "a"}
+	sel2 := map[string]string{"rack": "a", "gen": "2014"}
+
+	tests := []struct {
+		selector map[string]string
+		attrs    []*mesos.Attribute
+		ok       bool
+	}{
+		{sel1, []*mesos.Attribute{newTextAttribute("rack", "a")}, true},
+		{sel1, []*mesos.Attribute{newTextAttribute("rack", "b")}, false},
+		{sel1, []*mesos.Attribute{newTextAttribute("rack", "a"), newTextAttribute("gen", "2014")}, true},
+		{sel1, []*mesos.Attribute{newTextAttribute("rack", "a"), newScalarAttribute("num", 42.0)}, true},
+		{sel1, []*mesos.Attribute{newScalarAttribute("rack", 42.0)}, false},
+		{sel2, []*mesos.Attribute{newTextAttribute("rack", "a"), newTextAttribute("gen", "2014")}, true},
+		{sel2, []*mesos.Attribute{newTextAttribute("rack", "a"), newTextAttribute("gen", "2015")}, false},
+	}
+
+	for _, ts := range tests {
+		task, _ := fakePodTask("foo")
+		task.Pod.Spec.NodeSelector = ts.selector
+		offer := &mesos.Offer{
+			Resources: []*mesos.Resource{
+				mutil.NewScalarResource("cpus", t_min_cpu),
+				mutil.NewScalarResource("mem", t_min_mem),
+			},
+			Attributes: ts.attrs,
+		}
+		if got, want := task.AcceptOffer(offer), ts.ok; got != want {
+			t.Fatalf("expected acceptance of offer %v for selector %v to be %v, got %v:", want, got, ts.attrs, ts.selector)
+		}
+	}
+}
+
+func newTextAttribute(name string, val string) *mesos.Attribute {
+	return &mesos.Attribute{
+		Name: proto.String(name),
+		Type: mesos.Value_TEXT.Enum(),
+		Text: &mesos.Value_Text{Value: &val},
+	}
+}
+
+func newScalarAttribute(name string, val float64) *mesos.Attribute {
+	return &mesos.Attribute{
+		Name:   proto.String(name),
+		Type:   mesos.Value_SCALAR.Enum(),
+		Scalar: &mesos.Value_Scalar{Value: proto.Float64(val)},
+	}
+}