Commit f86b1192 authored by David Oppenheimer's avatar David Oppenheimer

Merge pull request #14550 from mesosphere/sttts-fix-daemonset-tests

Fix bugs in daemonset controller and e2e tests
parents 7fa09abe 2d8b0049
...@@ -29,6 +29,7 @@ import ( ...@@ -29,6 +29,7 @@ import (
clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api" clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
"k8s.io/kubernetes/pkg/cloudprovider" "k8s.io/kubernetes/pkg/cloudprovider"
"k8s.io/kubernetes/pkg/cloudprovider/providers/mesos" "k8s.io/kubernetes/pkg/cloudprovider/providers/mesos"
"k8s.io/kubernetes/pkg/controller/daemon"
kendpoint "k8s.io/kubernetes/pkg/controller/endpoint" kendpoint "k8s.io/kubernetes/pkg/controller/endpoint"
"k8s.io/kubernetes/pkg/controller/namespace" "k8s.io/kubernetes/pkg/controller/namespace"
"k8s.io/kubernetes/pkg/controller/node" "k8s.io/kubernetes/pkg/controller/node"
...@@ -47,7 +48,6 @@ import ( ...@@ -47,7 +48,6 @@ import (
"github.com/golang/glog" "github.com/golang/glog"
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
"github.com/spf13/pflag" "github.com/spf13/pflag"
"k8s.io/kubernetes/pkg/controller/daemon"
) )
// CMServer is the main context object for the controller manager. // CMServer is the main context object for the controller manager.
......
...@@ -291,12 +291,12 @@ func (r RealPodControl) createPods(nodeName, namespace string, template *api.Pod ...@@ -291,12 +291,12 @@ func (r RealPodControl) createPods(nodeName, namespace string, template *api.Pod
GenerateName: prefix, GenerateName: prefix,
}, },
} }
if len(nodeName) != 0 {
pod.Spec.NodeName = nodeName
}
if err := api.Scheme.Convert(&template.Spec, &pod.Spec); err != nil { if err := api.Scheme.Convert(&template.Spec, &pod.Spec); err != nil {
return fmt.Errorf("unable to convert pod template: %v", err) return fmt.Errorf("unable to convert pod template: %v", err)
} }
if len(nodeName) != 0 {
pod.Spec.NodeName = nodeName
}
if labels.Set(pod.Labels).AsSelector().Empty() { if labels.Set(pod.Labels).AsSelector().Empty() {
return fmt.Errorf("unable to create pods, no labels") return fmt.Errorf("unable to create pods, no labels")
} }
......
...@@ -18,6 +18,8 @@ package e2e ...@@ -18,6 +18,8 @@ package e2e
import ( import (
"fmt" "fmt"
"reflect"
"strings"
"time" "time"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
...@@ -35,8 +37,11 @@ import ( ...@@ -35,8 +37,11 @@ import (
) )
const ( const (
updateRetryPeriod = 5 * time.Second updateRetryPeriod = 5 * time.Second
updateRetryTimeout = 30 * time.Second updateRetryTimeout = 30 * time.Second
daemonsetLabelPrefix = "daemonset-"
daemonsetNameLabel = daemonsetLabelPrefix + "name"
daemonsetColorLabel = daemonsetLabelPrefix + "color"
) )
var _ = Describe("Daemon set", func() { var _ = Describe("Daemon set", func() {
...@@ -44,12 +49,12 @@ var _ = Describe("Daemon set", func() { ...@@ -44,12 +49,12 @@ var _ = Describe("Daemon set", func() {
BeforeEach(func() { BeforeEach(func() {
f.beforeEach() f.beforeEach()
err := clearNodeLabels(f.Client) err := clearDaemonSetNodeLabels(f.Client)
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
}) })
AfterEach(func() { AfterEach(func() {
err := clearNodeLabels(f.Client) err := clearDaemonSetNodeLabels(f.Client)
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
f.afterEach() f.afterEach()
}) })
...@@ -59,37 +64,74 @@ var _ = Describe("Daemon set", func() { ...@@ -59,37 +64,74 @@ var _ = Describe("Daemon set", func() {
}) })
}) })
func clearNodeLabels(c *client.Client) error { func separateDaemonSetNodeLabels(labels map[string]string) (map[string]string, map[string]string) {
daemonSetLabels := map[string]string{}
otherLabels := map[string]string{}
for k, v := range labels {
if strings.HasPrefix(k, daemonsetLabelPrefix) {
daemonSetLabels[k] = v
} else {
otherLabels[k] = v
}
}
return daemonSetLabels, otherLabels
}
func clearDaemonSetNodeLabels(c *client.Client) error {
nodeClient := c.Nodes() nodeClient := c.Nodes()
nodeList, err := nodeClient.List(labels.Everything(), fields.Everything()) nodeList, err := nodeClient.List(labels.Everything(), fields.Everything())
if err != nil { if err != nil {
return err return err
} }
for _, node := range nodeList.Items { for _, node := range nodeList.Items {
if len(node.Labels) != 0 { _, err := setDaemonSetNodeLabels(c, node.Name, map[string]string{})
node.Labels = map[string]string{} if err != nil {
var newNode *api.Node return err
err = wait.Poll(updateRetryPeriod, updateRetryTimeout, func() (bool, error) {
newNode, err = nodeClient.Update(&node)
if err == nil {
return true, err
}
if se, ok := err.(*apierrs.StatusError); ok && se.ErrStatus.Reason == unversioned.StatusReasonConflict {
Logf("failed to update node due to resource version conflict")
return false, nil
}
return false, err
})
if err != nil {
return err
} else if len(newNode.Labels) != 0 {
return fmt.Errorf("Could not make node labels nil.")
}
} }
} }
return nil return nil
} }
func setDaemonSetNodeLabels(c *client.Client, nodeName string, labels map[string]string) (*api.Node, error) {
nodeClient := c.Nodes()
var newNode *api.Node
var newLabels map[string]string
err := wait.Poll(updateRetryPeriod, updateRetryTimeout, func() (bool, error) {
node, err := nodeClient.Get(nodeName)
if err != nil {
return false, err
}
// remove all labels this test is creating
daemonSetLabels, otherLabels := separateDaemonSetNodeLabels(node.Labels)
if reflect.DeepEqual(daemonSetLabels, labels) {
newNode = node
return true, nil
}
node.Labels = otherLabels
for k, v := range labels {
node.Labels[k] = v
}
newNode, err = nodeClient.Update(node)
if err == nil {
newLabels, _ = separateDaemonSetNodeLabels(newNode.Labels)
return true, err
}
if se, ok := err.(*apierrs.StatusError); ok && se.ErrStatus.Reason == unversioned.StatusReasonConflict {
Logf("failed to update node due to resource version conflict")
return false, nil
}
return false, err
})
if err != nil {
return nil, err
} else if len(newLabels) != len(labels) {
return nil, fmt.Errorf("Could not set daemon set test labels as expected.")
}
return newNode, nil
}
func checkDaemonPodOnNodes(f *Framework, selector map[string]string, nodeNames []string) func() (bool, error) { func checkDaemonPodOnNodes(f *Framework, selector map[string]string, nodeNames []string) func() (bool, error) {
return func() (bool, error) { return func() (bool, error) {
podList, err := f.Client.Pods(f.Namespace.Name).List(labels.Set(selector).AsSelector(), fields.Everything()) podList, err := f.Client.Pods(f.Namespace.Name).List(labels.Set(selector).AsSelector(), fields.Everything())
...@@ -140,7 +182,7 @@ func testDaemonSets(f *Framework) { ...@@ -140,7 +182,7 @@ func testDaemonSets(f *Framework) {
c := f.Client c := f.Client
simpleDSName := "simple-daemon-set" simpleDSName := "simple-daemon-set"
image := "gcr.io/google_containers/serve_hostname:1.1" image := "gcr.io/google_containers/serve_hostname:1.1"
label := map[string]string{"name": simpleDSName} label := map[string]string{daemonsetNameLabel: simpleDSName}
retryTimeout := 1 * time.Minute retryTimeout := 1 * time.Minute
retryInterval := 5 * time.Second retryInterval := 5 * time.Second
...@@ -195,8 +237,8 @@ func testDaemonSets(f *Framework) { ...@@ -195,8 +237,8 @@ func testDaemonSets(f *Framework) {
Expect(err).NotTo(HaveOccurred(), "error waiting for daemon pod to revive") Expect(err).NotTo(HaveOccurred(), "error waiting for daemon pod to revive")
complexDSName := "complex-daemon-set" complexDSName := "complex-daemon-set"
complexLabel := map[string]string{"name": complexDSName} complexLabel := map[string]string{daemonsetNameLabel: complexDSName}
nodeSelector := map[string]string{"color": "blue"} nodeSelector := map[string]string{daemonsetColorLabel: "blue"}
Logf("Creating daemon with a node selector %s", complexDSName) Logf("Creating daemon with a node selector %s", complexDSName)
_, err = c.DaemonSets(ns).Create(&experimental.DaemonSet{ _, err = c.DaemonSets(ns).Create(&experimental.DaemonSet{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
...@@ -231,40 +273,16 @@ func testDaemonSets(f *Framework) { ...@@ -231,40 +273,16 @@ func testDaemonSets(f *Framework) {
nodeClient := c.Nodes() nodeClient := c.Nodes()
nodeList, err := nodeClient.List(labels.Everything(), fields.Everything()) nodeList, err := nodeClient.List(labels.Everything(), fields.Everything())
Expect(len(nodeList.Items)).To(BeNumerically(">", 0)) Expect(len(nodeList.Items)).To(BeNumerically(">", 0))
nodeList.Items[0].Labels = nodeSelector newNode, err := setDaemonSetNodeLabels(c, nodeList.Items[0].Name, nodeSelector)
var newNode *api.Node Expect(err).NotTo(HaveOccurred(), "error setting labels on node")
err = wait.Poll(updateRetryPeriod, updateRetryTimeout, func() (bool, error) { daemonSetLabels, _ := separateDaemonSetNodeLabels(newNode.Labels)
newNode, err = nodeClient.Update(&nodeList.Items[0]) Expect(len(daemonSetLabels)).To(Equal(1))
if err == nil {
return true, err
}
if se, ok := err.(*apierrs.StatusError); ok && se.ErrStatus.Reason == unversioned.StatusReasonConflict {
Logf("failed to update node due to resource version conflict")
return false, nil
}
return false, err
})
Expect(err).NotTo(HaveOccurred())
Expect(len(newNode.Labels)).To(Equal(1))
err = wait.Poll(retryInterval, retryTimeout, checkDaemonPodOnNodes(f, complexLabel, []string{newNode.Name})) err = wait.Poll(retryInterval, retryTimeout, checkDaemonPodOnNodes(f, complexLabel, []string{newNode.Name}))
Expect(err).NotTo(HaveOccurred(), "error waiting for daemon pods to be running on new nodes") Expect(err).NotTo(HaveOccurred(), "error waiting for daemon pods to be running on new nodes")
By("remove the node selector and wait for daemons to be unscheduled") By("remove the node selector and wait for daemons to be unscheduled")
newNode, err = nodeClient.Get(newNode.Name) _, err = setDaemonSetNodeLabels(c, nodeList.Items[0].Name, map[string]string{})
Expect(err).NotTo(HaveOccurred(), "error getting node") Expect(err).NotTo(HaveOccurred(), "error removing labels on node")
newNode.Labels = map[string]string{}
err = wait.Poll(updateRetryPeriod, updateRetryTimeout, func() (bool, error) {
newNode, err = nodeClient.Update(newNode)
if err == nil {
return true, err
}
if se, ok := err.(*apierrs.StatusError); ok && se.ErrStatus.Reason == unversioned.StatusReasonConflict {
Logf("failed to update node due to resource version conflict")
return false, nil
}
return false, err
})
Expect(err).NotTo(HaveOccurred())
Expect(wait.Poll(retryInterval, retryTimeout, checkRunningOnNoNodes(f, complexLabel))). Expect(wait.Poll(retryInterval, retryTimeout, checkRunningOnNoNodes(f, complexLabel))).
NotTo(HaveOccurred(), "error waiting for daemon pod to not be running on nodes") NotTo(HaveOccurred(), "error waiting for daemon pod to not be running on nodes")
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment