Merge pull request #17265 from gambol99/rj/node_labels

cmd/kubelet/app/server.go

@@ -116,6 +116,8 @@ type KubeletServer struct {
 	MinimumGCAge                   time.Duration
 	NetworkPluginDir               string
 	NetworkPluginName              string
+	NodeLabels                     []string
+	NodeLabelsFile                 string
 	NodeStatusUpdateFrequency      time.Duration
 	OOMScoreAdj                    int
 	PodCIDR                        string
@@ -202,6 +204,8 @@ func NewKubeletServer() *KubeletServer {
 		MinimumGCAge:                1 * time.Minute,
 		NetworkPluginDir:            "/usr/libexec/kubernetes/kubelet-plugins/net/exec/",
 		NetworkPluginName:           "",
+		NodeLabels:                  []string{},
+		NodeLabelsFile:              "",
 		NodeStatusUpdateFrequency:   10 * time.Second,
 		OOMScoreAdj:                 qos.KubeletOOMScoreAdj,
 		PodInfraContainerImage:      dockertools.PodInfraContainerImage,
@@ -303,6 +307,8 @@ func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.MasterServiceNamespace, "master-service-namespace", s.MasterServiceNamespace, "The namespace from which the kubernetes master services should be injected into pods")
 	fs.IPVar(&s.ClusterDNS, "cluster-dns", s.ClusterDNS, "IP address for a cluster DNS server.  If set, kubelet will configure all containers to use this for DNS resolution in addition to the host's DNS servers")
 	fs.DurationVar(&s.StreamingConnectionIdleTimeout, "streaming-connection-idle-timeout", s.StreamingConnectionIdleTimeout, "Maximum time a streaming connection can be idle before the connection is automatically closed.  Example: '5m'")
+	fs.StringSliceVar(&s.NodeLabels, "node-label", []string{}, "add labels when registering the node in the cluster, the flag can be used multiple times (key=value)")
+	fs.StringVar(&s.NodeLabelsFile, "node-labels-file", "", "the path to a yaml or json file containing a series of key pair labels to apply on node registration")
 	fs.DurationVar(&s.NodeStatusUpdateFrequency, "node-status-update-frequency", s.NodeStatusUpdateFrequency, "Specifies how often kubelet posts node status to master. Note: be cautious when changing the constant, it must work with nodeMonitorGracePeriod in nodecontroller. Default: 10s")
 	fs.IntVar(&s.ImageGCHighThresholdPercent, "image-gc-high-threshold", s.ImageGCHighThresholdPercent, "The percent of disk usage after which image garbage collection is always run. Default: 90%")
 	fs.IntVar(&s.ImageGCLowThresholdPercent, "image-gc-low-threshold", s.ImageGCLowThresholdPercent, "The percent of disk usage before which image garbage collection is never run. Lowest disk usage to garbage collect to. Default: 80%")
@@ -442,6 +448,8 @@ func (s *KubeletServer) UnsecuredKubeletConfig() (*KubeletConfig, error) {
 		ChmodRunner:               chmodRunner,
 		NetworkPluginName:         s.NetworkPluginName,
 		NetworkPlugins:            ProbeNetworkPlugins(s.NetworkPluginDir),
+		NodeLabels:                s.NodeLabels,
+		NodeLabelsFile:            s.NodeLabelsFile,
 		NodeStatusUpdateFrequency: s.NodeStatusUpdateFrequency,
 		OOMAdjuster:               oom.NewOOMAdjuster(),
 		OSInterface:               kubecontainer.RealOS{},
@@ -908,6 +916,8 @@ type KubeletConfig struct {
 	NetworkPluginName              string
 	NetworkPlugins                 []network.NetworkPlugin
 	NodeName                       string
+	NodeLabels                     []string
+	NodeLabelsFile                 string
 	NodeStatusUpdateFrequency      time.Duration
 	OOMAdjuster                    *oom.OOMAdjuster
 	OSInterface                    kubecontainer.OSInterface
@@ -992,6 +1002,8 @@ func CreateAndInitKubelet(kc *KubeletConfig) (k KubeletBootstrap, pc *config.Pod
 		kc.ImageGCPolicy,
 		kc.DiskSpacePolicy,
 		kc.Cloud,
+		kc.NodeLabels,
+		kc.NodeLabelsFile,
 		kc.NodeStatusUpdateFrequency,
 		kc.ResourceContainer,
 		kc.OSInterface,

docs/admin/kubelet.md

@@ -111,6 +111,8 @@ kubelet
       --minimum-container-ttl-duration=1m0s: Minimum age for a finished container before it is garbage collected.  Examples: '300ms', '10s' or '2h45m'
       --network-plugin="": <Warning: Alpha feature> The name of the network plugin to be invoked for various events in kubelet/pod lifecycle
       --network-plugin-dir="/usr/libexec/kubernetes/kubelet-plugins/net/exec/": <Warning: Alpha feature> The full path of the directory in which to search for network plugins
+      --node-label=[]: add labels when registering the node in the cluster, the flag can be used multiple times (key=value)
+      --node-labels-file="": the path to a yaml or json file containing a series of key pair labels to apply on node registration
       --node-status-update-frequency=10s: Specifies how often kubelet posts node status to master. Note: be cautious when changing the constant, it must work with nodeMonitorGracePeriod in nodecontroller. Default: 10s
       --oom-score-adj=-999: The oom-score-adj value for kubelet process. Values must be within the range [-1000, 1000]
       --pod-cidr="": The CIDR to use for pod IP addresses, only used in standalone mode.  In cluster mode, this is obtained from the master.
@@ -137,7 +139,7 @@ kubelet
       --tls-private-key-file="": File containing x509 private key matching --tls-cert-file.
 ```
 
-###### Auto generated by spf13/cobra on 11-Nov-2015
+###### Auto generated by spf13/cobra on 18-Nov-2015
 
 
 <!-- BEGIN MUNGE: GENERATED_ANALYTICS -->

hack/verify-flags/known-flags.txt

@@ -207,6 +207,8 @@ network-plugin-dir
 node-instance-group
 node-monitor-grace-period
 node-monitor-period
+node-label
+node-labels-file
 node-startup-grace-period
 node-status-update-frequency
 node-sync-period

pkg/kubelet/kubelet.go

@@ -20,6 +20,7 @@ package kubelet
 // contrib/mesos/pkg/executor/.
 
 import (
+	"bufio"
 	"bytes"
 	"errors"
 	"fmt"
@@ -80,6 +81,7 @@ import (
 	"k8s.io/kubernetes/pkg/util/procfs"
 	"k8s.io/kubernetes/pkg/util/selinux"
 	"k8s.io/kubernetes/pkg/util/sets"
+	"k8s.io/kubernetes/pkg/util/yaml"
 	"k8s.io/kubernetes/pkg/version"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/watch"
@@ -189,6 +191,8 @@ func NewMainKubelet(
 	imageGCPolicy ImageGCPolicy,
 	diskSpacePolicy DiskSpacePolicy,
 	cloud cloudprovider.Interface,
+	nodeLabels []string,
+	nodeLabelsFile string,
 	nodeStatusUpdateFrequency time.Duration,
 	resourceContainer string,
 	osInterface kubecontainer.OSInterface,
@@ -304,6 +308,8 @@ func NewMainKubelet(
 		volumeManager:                  volumeManager,
 		cloud:                          cloud,
 		nodeRef:                        nodeRef,
+		nodeLabels:                     nodeLabels,
+		nodeLabelsFile:                 nodeLabelsFile,
 		nodeStatusUpdateFrequency:      nodeStatusUpdateFrequency,
 		resourceContainer:              resourceContainer,
 		os:                             osInterface,
@@ -510,6 +516,12 @@ type Kubelet struct {
 	serviceLister          serviceLister
 	nodeLister             nodeLister
 
+	// a list of node labels to register
+	nodeLabels []string
+
+	// the path to a yaml or json file container series of node labels
+	nodeLabelsFile string
+
 	// Last timestamp when runtime responded on ping.
 	// Mutex is used to protect this value.
 	runtimeState *runtimeState
@@ -903,6 +915,19 @@ func (kl *Kubelet) initialNodeStatus() (*api.Node, error) {
 			Unschedulable: !kl.registerSchedulable,
 		},
 	}
+
+	labels, err := kl.getNodeLabels()
+	if err != nil {
+		return nil, err
+	}
+	// @question: should this be place after the call to the cloud provider? which also applies labels
+	for k, v := range labels {
+		if cv, found := node.ObjectMeta.Labels[k]; found {
+			glog.Warningf("the node label %s=%s will overwrite default setting %s", k, v, cv)
+		}
+		node.ObjectMeta.Labels[k] = v
+	}
+
 	if kl.cloud != nil {
 		instances, ok := kl.cloud.Instances()
 		if !ok {
@@ -951,6 +976,80 @@ func (kl *Kubelet) initialNodeStatus() (*api.Node, error) {
 	return node, nil
 }
 
+// getNodeLabels is just a wrapper method for the two below, not to duplicate above
+func (kl *Kubelet) getNodeLabels() (map[string]string, error) {
+	var err error
+	labels := make(map[string]string, 0)
+
+	if kl.nodeLabelsFile != "" {
+		labels, err = kl.retrieveNodeLabelsFile(kl.nodeLabelsFile)
+		if err != nil {
+			return labels, err
+		}
+	}
+	// step: apply the command line label - permitted to override those from file
+	if len(kl.nodeLabels) > 0 {
+		nl, err := kl.retrieveNodeLabels(kl.nodeLabels)
+		if err != nil {
+			return labels, err
+		}
+		for k, v := range nl {
+			if vl, found := labels[k]; found {
+				glog.Warningf("the --node-label %s=%s option will overwrite %s from node-labels-file", k, v, vl)
+			}
+			labels[k] = v
+		}
+	}
+
+	return labels, nil
+}
+
+// retrieveNodeLabels extracts the node labels specified on the command line
+func (kl *Kubelet) retrieveNodeLabels(labels []string) (map[string]string, error) {
+	nodeLabels := make(map[string]string, 0)
+
+	for _, label := range labels {
+		items := strings.Split(label, "=")
+		if len(items) != 2 {
+			return nodeLabels, fmt.Errorf("--node-label %s, should be in the form key=pair", label)
+		}
+		nodeLabels[strings.TrimSpace(items[0])] = strings.TrimSpace(items[1])
+	}
+
+	return nodeLabels, nil
+}
+
+// retrieveNodeLabelsFile reads in and parses the yaml or json node labels file
+func (kl *Kubelet) retrieveNodeLabelsFile(path string) (map[string]string, error) {
+	labels := make(map[string]string, 0)
+	kps := make(map[string]interface{}, 0)
+
+	fd, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer fd.Close()
+
+	err = yaml.NewYAMLOrJSONDecoder(bufio.NewReader(fd), 12).Decode(&kps)
+	if err != nil {
+		return nil, fmt.Errorf("the --node-labels-file %s content is invalid, %s", path, err)
+	}
+
+	for k, v := range kps {
+		// we ONLY accept key=value pairs, no complex types
+		switch v.(type) {
+		case string:
+			labels[k] = v.(string)
+		case float64:
+			labels[k] = fmt.Sprintf("%d", v.(float64))
+		default:
+			return nil, fmt.Errorf("--node-labels-file only supports key:string, not complex values e.g arrays, maps")
+		}
+	}
+
+	return labels, nil
+}
+
 // registerWithApiserver registers the node with the cluster master. It is safe
 // to call multiple times, but not concurrently (kl.registrationCompleted is
 // not locked).

pkg/kubelet/kubelet_test.go

@@ -3460,6 +3460,176 @@ func TestRegisterExistingNodeWithApiserver(t *testing.T) {
 	}
 }
 
+func TestGetNodeLabels(t *testing.T) {
+	kubelet := newTestKubelet(t).kubelet
+
+	testCases := []struct {
+		Expecting    map[string]string
+		LabelOptions []string
+		FileContent  string
+		Ok           bool
+	}{
+		{
+			Ok:           true,
+			Expecting:    map[string]string{"key1": "pair1", "key2": "pair2", "key3": "pair3", "key4": "pair4", "key5": "pair5"},
+			LabelOptions: []string{"key5=pair5"},
+			FileContent: `---
+key1: pair1
+key2: pair2
+key3: pair3
+key4: pair4
+`,
+		}, {
+			Ok:           true,
+			Expecting:    map[string]string{"key1": "pair1", "key2": "override"},
+			LabelOptions: []string{"key2=override"},
+			FileContent: `---
+key1: pair1
+key2: pair2
+`,
+		},
+	}
+
+	for i, test := range testCases {
+		fd := createTestNodeLabelFile(t, test.FileContent)
+		defer func(f *os.File) {
+			os.Remove(f.Name())
+		}(fd)
+
+		kubelet.nodeLabels = test.LabelOptions
+		kubelet.nodeLabelsFile = fd.Name()
+
+		list, err := kubelet.getNodeLabels()
+		if test.Ok && err != nil {
+			t.Errorf("test case %d should not have failed, error: %s", i, err)
+		}
+		if !reflect.DeepEqual(test.Expecting, list) {
+			t.Errorf("test case %d are not the same, %v ~ %v", i, list, test.Expecting)
+		}
+	}
+}
+
+func TestRetrieveNodeLabels(t *testing.T) {
+	kubelet := newTestKubelet(t).kubelet
+
+	testCases := []struct {
+		Expecting    map[string]string
+		LabelOptions []string
+		Ok           bool
+	}{
+		{
+			Expecting:    map[string]string{"key1": "pair1", "key2": "pair2", "key3": "pair3", "key4": "pair4"},
+			LabelOptions: []string{"key1=pair1", "key2=pair2", "key3=pair3", "key4=pair4"},
+			Ok:           true,
+		},
+		{
+			Expecting:    map[string]string{"key1": "pair1"},
+			LabelOptions: []string{"key1=pair1", "key2paiwdsr2"},
+		},
+	}
+
+	for i, test := range testCases {
+		list, err := kubelet.retrieveNodeLabels(test.LabelOptions)
+		if test.Ok && err != nil {
+			t.Errorf("test case %d should not have failed, error: %s", i, err)
+		}
+		if !reflect.DeepEqual(test.Expecting, list) {
+			t.Errorf("test case %d are not the same, %v ~ %v", i, list, test.Expecting)
+		}
+	}
+}
+
+func TestRetrieveNodeLabelsFile(t *testing.T) {
+	kubelet := newTestKubelet(t).kubelet
+
+	testCases := []struct {
+		Expecting   map[string]string
+		Ok          bool
+		FileContent string
+	}{
+		{
+			Expecting: map[string]string{"key1": "pair1", "key2": "pair2", "key3": "pair3", "key4": "pair4"},
+			Ok:        true,
+			FileContent: `---
+key1: pair1
+key2: pair2
+key3: pair3
+key4: pair4`,
+		}, {
+			FileContent: `---
+key1: pair1
+hash_map:
+  key2: pair2
+`,
+		}, {
+			Expecting: map[string]string{"key1": "pair1", "key2": "pair2"},
+			Ok:        true,
+			FileContent: `
+
+key1: pair1
+key2: pair2
+`,
+		}, {
+			FileContent: `---
+key1: pair1
+bad_key_pair
+`,
+		}, {
+			Expecting: nil,
+			FileContent: `{
+	"key1": "pair1",
+	"key2": "pair2",
+	"key3": "pair3",
+	"key4": {
+		"some_key": "some_value"
+	}
+}`,
+		}, {
+			FileContent: "",
+		}, {
+			Expecting: map[string]string{"key1": "pair1", "key2": "pair2", "key3": "pair3", "key4": "pair4"},
+			Ok:        true,
+			FileContent: `---
+key1: pair1
+key2: pair2
+key3: pair3
+key4: pair4
+`,
+		},
+	}
+
+	for i, test := range testCases {
+		fd := createTestNodeLabelFile(t, test.FileContent)
+		defer func(f *os.File) {
+			os.Remove(f.Name())
+		}(fd)
+
+		labels, err := kubelet.retrieveNodeLabelsFile(fd.Name())
+		if test.Ok && err != nil {
+			t.Errorf("test case %d should not have returned an error, %s", i, err)
+			continue
+		}
+
+		if test.Expecting != nil && !reflect.DeepEqual(test.Expecting, labels) {
+			t.Errorf("test case %d not as expected, got: %#v, expecting: %#v", i, labels, test.Expecting)
+		}
+	}
+}
+
+func createTestNodeLabelFile(t *testing.T, content string) *os.File {
+	f, err := ioutil.TempFile("", "node_label_file")
+	if err != nil {
+		t.Fatalf("unexpected error creating node_label_file: %v", err)
+	}
+	f.Close()
+
+	if err := ioutil.WriteFile(f.Name(), []byte(content), 0700); err != nil {
+		t.Fatalf("unexpected error writing node label file: %v", err)
+	}
+
+	return f
+}
+
 func TestMakePortMappings(t *testing.T) {
 	tests := []struct {
 		container            *api.Container