Merge pull request #793 from apprenda/node-labels

Add node labels option in the plan file to label nodes

Author: dkoshkin

ansible/_label-nodes.yaml

@@ -0,0 +1,13 @@
+---
+  - hosts: master:worker:ingress:storage
+    any_errors_fatal: true
+    name: Label Kubernetes Nodes 
+    serial: "{{ serial_count | default('100%') }}"
+    become: yes
+    vars_files:
+      - group_vars/all.yaml
+      
+    tasks:
+      - name: label nodes
+        command: kubectl label --overwrite nodes --selector kismatic/host={{ inventory_hostname }} --kubeconfig {{ kubernetes_kubeconfig_path }} {{ node_labels[inventory_hostname] | join(" ") }}
+        when: node_labels[inventory_hostname] is defined and node_labels[inventory_hostname]|length > 0

ansible/kubernetes-worker.yaml

@@ -9,6 +9,7 @@
   - include: _docker.yaml
   - include: _kubelet.yaml
   - include: _kube-proxy.yaml
+  - include: _label-nodes.yaml
   - include: _cni.yaml
     when: cni.enabled|bool == true
   - include: _calico.yaml

ansible/kubernetes.yaml

@@ -24,6 +24,7 @@
   # kubelet does not have an API yet to retrieve the status of a DS pod
   # after installing kube-proxy, there is a dependecy on the API server to validate the static pod
   - include: _kube-proxy.yaml
+  - include: _label-nodes.yaml
   - include: _cni.yaml
     when: cni.enabled|bool == true
   - include: _calico.yaml

ansible/roles/kubelet/templates/kubelet.service

@@ -21,7 +21,7 @@ ExecStart=/usr/bin/kubelet \
   --hostname-override={{ inventory_hostname }} \
   --require-kubeconfig=true \
   --kubeconfig={{ kubernetes_kubeconfig.kubelet }} \
-  --node-labels=kismatic/host={{ inventory_hostname }},kismatic/cni-provider={{ cni.provider|default("")}}{% if 'ingress' in group_names%},kismatic/ingress=true{% endif %}{% if 'storage' in group_names%},kismatic/storage=true{% endif %}{% if inventory_hostname in groups['master'] %},node-role.kubernetes.io/master={% endif %} \
+  --node-labels=kismatic/host={{ inventory_hostname }},kismatic/cni-provider={{ cni.provider|default("")}}{% if 'ingress' in group_names%},kismatic/ingress=true{% endif %}{% if 'storage' in group_names%},kismatic/storage=true{% endif %}{% if 'master' in group_names %},node-role.kubernetes.io/master={% endif %} \
   --node-ip={{ internal_ipv4 }} \
   --pod-infra-container-image={{ pause_img }} \
   --pod-manifest-path={{ kubelet_pod_manifests_dir }} \

ansible/roles/validate-pod/tasks/validate-pod.yaml

@@ -39,7 +39,7 @@
     register: docker_logs
     when: containerID is defined and containerID|success and containerID.stdout is defined and containerID.stdout != ""
 
-  - name: fail if pod '{{ name }}'
+  - name: fail if pod '{{ name }}' is not running
     fail:
       msg: |
         Waited for pod '{{ name }}' to be running, but it did not start up in time.

ansible/upgrade-nodes.yaml

@@ -33,6 +33,7 @@
   - include: _validate-control-plane-node.yaml serial_count="1" upgrading=true
   - include: _kube-proxy-stop.yaml play_name="Upgrade Kubernetes Proxy" upgrading=true
   - include: _kube-proxy.yaml play_name="Upgrade Kubernetes Proxy" upgrading=true
+  - include: _label-nodes.yaml
   - include: _cni.yaml play_name="Upgrade Kubernetes CNI" upgrading=true
     when: cni.enabled|bool == true
   - include: _calico.yaml play_name="Upgrade Calico Cluster Network" upgrading=true

cmd/gen-kismatic-ref-docs/main.go

@@ -133,6 +133,13 @@ func docForType(typeName string, allTypes []*godoc.Type, parentFieldName string)
 						if isStruct(typeName) {
 							docs = append(docs, docForType(typeName, allTypes, fieldName)...)
 						}
+					case *ast.MapType:
+						typeName = fmt.Sprintf("map[%s]%s", x.Key.(*ast.Ident).Name, x.Value.(*ast.Ident).Name)
+						d, err := parseDoc(fieldName, typeName, f.Doc.Text())
+						if err != nil {
+							panic(err)
+						}
+						docs = append(docs, d)
 					default:
 						panic(fmt.Sprintf("unhandled typespec type: %q", reflect.TypeOf(x).Name()))
 					}
@@ -198,7 +205,8 @@ func isStruct(s string) bool {
 
 // not a comprehensive list, but works for now...
 var basicTypes = map[string]bool{
-	"bool":   true,
-	"int":    true,
-	"string": true,
+	"bool":              true,
+	"int":               true,
+	"string":            true,
+	"map[string]string": true,
 }

docs/plan-file-reference.md

@@ -76,6 +76,7 @@
     * [host](#etcdnodeshost)
     * [ip](#etcdnodesip)
     * [internalip](#etcdnodesinternalip)
+    * [labels](#etcdnodeslabels)
 * [master](#master)
   * [expected_count](#masterexpected_count)
   * [load_balanced_fqdn](#masterload_balanced_fqdn)
@@ -84,24 +85,28 @@
     * [host](#masternodeshost)
     * [ip](#masternodesip)
     * [internalip](#masternodesinternalip)
+    * [labels](#masternodeslabels)
 * [worker](#worker)
   * [expected_count](#workerexpected_count)
   * [nodes](#workernodes)
     * [host](#workernodeshost)
     * [ip](#workernodesip)
     * [internalip](#workernodesinternalip)
+    * [labels](#workernodeslabels)
 * [ingress](#ingress)
   * [expected_count](#ingressexpected_count)
   * [nodes](#ingressnodes)
     * [host](#ingressnodeshost)
     * [ip](#ingressnodesip)
     * [internalip](#ingressnodesinternalip)
+    * [labels](#ingressnodeslabels)
 * [storage](#storage)
   * [expected_count](#storageexpected_count)
   * [nodes](#storagenodes)
     * [host](#storagenodeshost)
     * [ip](#storagenodesip)
     * [internalip](#storagenodesinternalip)
+    * [labels](#storagenodeslabels)
 * [nfs](#nfs)
   * [nfs_volume](#nfsnfs_volume)
     * [nfs_host](#nfsnfs_volumenfs_host)
@@ -717,6 +722,16 @@
 | **Required** |  No |
 | **Default** | ` ` | 
 
+###  etcd.nodes.labels
+
+ Labels to add when installing the node in the cluster. If a node is defined under multiple roles, the labels for that node will be merged. If a label is repeated for the same node, only one will be used in this order: etcd,master,worker,ingress,storage roles where 'storage' has the highest precedence. It is recommended to use reverse-DNS notation to avoid collision with other labels. 
+
+| | |
+|----------|-----------------|
+| **Kind** |  map[string]string |
+| **Required** |  No |
+| **Default** | ` ` | 
+
 ##  master
 
  Master nodes of the cluster 
@@ -785,6 +800,16 @@
 | **Required** |  No |
 | **Default** | ` ` | 
 
+###  master.nodes.labels
+
+ Labels to add when installing the node in the cluster. If a node is defined under multiple roles, the labels for that node will be merged. If a label is repeated for the same node, only one will be used in this order: etcd,master,worker,ingress,storage roles where 'storage' has the highest precedence. It is recommended to use reverse-DNS notation to avoid collision with other labels. 
+
+| | |
+|----------|-----------------|
+| **Kind** |  map[string]string |
+| **Required** |  No |
+| **Default** | ` ` | 
+
 ##  worker
 
  Worker nodes of the cluster 
@@ -833,6 +858,16 @@
 | **Required** |  No |
 | **Default** | ` ` | 
 
+###  worker.nodes.labels
+
+ Labels to add when installing the node in the cluster. If a node is defined under multiple roles, the labels for that node will be merged. If a label is repeated for the same node, only one will be used in this order: etcd,master,worker,ingress,storage roles where 'storage' has the highest precedence. It is recommended to use reverse-DNS notation to avoid collision with other labels. 
+
+| | |
+|----------|-----------------|
+| **Kind** |  map[string]string |
+| **Required** |  No |
+| **Default** | ` ` | 
+
 ##  ingress
 
  Ingress nodes of the cluster 
@@ -881,6 +916,16 @@
 | **Required** |  No |
 | **Default** | ` ` | 
 
+###  ingress.nodes.labels
+
+ Labels to add when installing the node in the cluster. If a node is defined under multiple roles, the labels for that node will be merged. If a label is repeated for the same node, only one will be used in this order: etcd,master,worker,ingress,storage roles where 'storage' has the highest precedence. It is recommended to use reverse-DNS notation to avoid collision with other labels. 
+
+| | |
+|----------|-----------------|
+| **Kind** |  map[string]string |
+| **Required** |  No |
+| **Default** | ` ` | 
+
 ##  storage
 
  Storage nodes of the cluster. 
@@ -929,6 +974,16 @@
 | **Required** |  No |
 | **Default** | ` ` | 
 
+###  storage.nodes.labels
+
+ Labels to add when installing the node in the cluster. If a node is defined under multiple roles, the labels for that node will be merged. If a label is repeated for the same node, only one will be used in this order: etcd,master,worker,ingress,storage roles where 'storage' has the highest precedence. It is recommended to use reverse-DNS notation to avoid collision with other labels. 
+
+| | |
+|----------|-----------------|
+| **Kind** |  map[string]string |
+| **Required** |  No |
+| **Default** | ` ` | 
+
 ##  nfs
 
  NFS volumes of the cluster. 

integration/add_worker.go

@@ -9,9 +9,12 @@ import (
 	. "github.com/onsi/ginkgo"
 )
 
-func addWorkerToCluster(newWorker NodeDeets, sshKey string) error {
+func addWorkerToCluster(newWorker NodeDeets, sshKey string, labels []string) error {
 	By("Adding new worker")
 	cmd := exec.Command("./kismatic", "install", "add-worker", "-f", "kismatic-testing.yaml", newWorker.Hostname, newWorker.PublicIP, newWorker.PrivateIP)
+	if len(labels) > 0 {
+		cmd.Args = append(cmd.Args, "--labels", strings.Join(labels, ","))
+	}
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 	if err := cmd.Run(); err != nil {

integration/hosts_file_test.go

@@ -51,7 +51,7 @@ var _ = Describe("hosts file modification feature", func() {
 				FailIfError(err)
 
 				By("Adding a worker with a bogus hostname that is added to hosts files")
-				err = addWorkerToCluster(nodes.worker[3], sshKey)
+				err = addWorkerToCluster(nodes.worker[3], sshKey, []string{})
 				FailIfError(err)
 			})
 		})

integration/install_test.go

@@ -234,7 +234,7 @@ var _ = Describe("kismatic", func() {
 
 						sub.It("should allow adding a worker node", func() error {
 							newWorker := allWorkers[len(allWorkers)-1]
-							return addWorkerToCluster(newWorker, sshKey)
+							return addWorkerToCluster(newWorker, sshKey, []string{"com.integrationtest/worker=true"})
 						})
 
 						sub.It("should be able to deploy a workload with ingress", func() error {
@@ -257,6 +257,10 @@ var _ = Describe("kismatic", func() {
 						sub.It("should have tiller running", func() error {
 							return verifyTiller(nodes.master[0], sshKey)
 						})
+
+						sub.It("nodes should contain expected labels", func() error {
+							return containsLabels(nodes, sshKey)
+						})
 					})
 				})
 			})
@@ -284,7 +288,7 @@ var _ = Describe("kismatic", func() {
 
 						sub.It("should allow adding a worker node", func() error {
 							newWorker := allWorkers[len(allWorkers)-1]
-							return addWorkerToCluster(newWorker, sshKey)
+							return addWorkerToCluster(newWorker, sshKey, []string{"com.integrationtest/worker=true"})
 						})
 
 						sub.It("should be able to deploy a workload with ingress", func() error {
@@ -307,6 +311,10 @@ var _ = Describe("kismatic", func() {
 						sub.It("should have tiller running", func() error {
 							return verifyTiller(nodes.master[0], sshKey)
 						})
+
+						sub.It("nodes should contain expected labels", func() error {
+							return containsLabels(nodes, sshKey)
+						})
 					})
 				})
 			})
@@ -334,7 +342,7 @@ var _ = Describe("kismatic", func() {
 
 		// 				sub.It("should allow adding a worker node", func() error {
 		// 					newWorker := allWorkers[len(allWorkers)-1]
-		// 					return addWorkerToCluster(newWorker, sshKey)
+		// 					return addWorkerToCluster(newWorker, sshKey, []string{})
 		// 				})
 
 		// 				// This test is flaky with contiv

integration/labels.go

@@ -0,0 +1,40 @@
+package integration
+
+import (
+	"fmt"
+	"time"
+)
+
+func containsLabels(nodes provisionedNodes, sshKey string) error {
+	// labels were hardcoded in th plan pattern
+	tests := []struct {
+		nodes []NodeDeets
+		label string
+	}{
+		{
+			nodes: nodes.master,
+			label: "com.integrationtest/master:true",
+		},
+		{
+			nodes: nodes.worker,
+			label: "com.integrationtest/worker:true",
+		},
+		{
+			nodes: nodes.ingress,
+			label: "com.integrationtest/ingress:true",
+		},
+		{
+			nodes: nodes.storage,
+			label: "com.integrationtest/storage:true",
+		},
+	}
+	for _, role := range tests {
+		for _, n := range role.nodes {
+			if err := runViaSSH([]string{fmt.Sprintf("sudo kubectl get nodes %s -o jsonpath='{.metadata.labels}' | grep %q", n.Hostname, role.label)}, []NodeDeets{nodes.master[0]}, sshKey, 1*time.Minute); err != nil {
+				return fmt.Errorf("error validating node %q label: %v", n.Hostname, err)
+			}
+		}
+	}
+
+	return nil
+}

integration/plan_patterns.go

@@ -108,27 +108,35 @@ master:
   nodes:{{range .Master}}
   - host: {{.Hostname}}
     ip: {{.PublicIP}}
-    internalip: {{.PrivateIP}}{{end}}
+    internalip: {{.PrivateIP}}
+    labels:
+      com.integrationtest/master: true{{end}}
   load_balanced_fqdn: {{.MasterNodeFQDN}}
   load_balanced_short_name: {{.MasterNodeShortName}}
 worker:
   expected_count: {{len .Worker}}
   nodes:{{range .Worker}}
   - host: {{.Hostname}}
     ip: {{.PublicIP}}
-    internalip: {{.PrivateIP}}{{end}}
+    internalip: {{.PrivateIP}}
+    labels:
+      com.integrationtest/worker: true{{end}}
 ingress:
   expected_count: {{len .Ingress}}
   nodes:{{range .Ingress}}
   - host: {{.Hostname}}
     ip: {{.PublicIP}}
-    internalip: {{.PrivateIP}}{{end}}
+    internalip: {{.PrivateIP}}
+    labels:
+      com.integrationtest/ingress: true{{end}}
 storage:
   expected_count: {{len .Storage}}
   nodes:{{range .Storage}}
   - host: {{.Hostname}}
     ip: {{.PublicIP}}
-    internalip: {{.PrivateIP}}{{end}}
+    internalip: {{.PrivateIP}}
+    labels:
+      com.integrationtest/storage: true{{end}}
 nfs:
   nfs_volume:{{range .NFSVolume}}
   - nfs_host: {{.Host}}

integration/upgrade_test.go

@@ -64,7 +64,7 @@ var _ = Describe("Upgrade", func() {
 
 						sub.It("should allow adding a worker node", func() error {
 							newWorker := allWorkers[len(allWorkers)-1]
-							return addWorkerToCluster(newWorker, sshKey)
+							return addWorkerToCluster(newWorker, sshKey, []string{})
 						})
 
 						sub.It("should be able to deploy a workload with ingress", func() error {
@@ -248,7 +248,7 @@ var _ = Describe("Upgrade", func() {
 
 						sub.It("should allow adding a worker node", func() error {
 							newWorker := allWorkers[len(allWorkers)-1]
-							return addWorkerToCluster(newWorker, sshKey)
+							return addWorkerToCluster(newWorker, sshKey, []string{})
 						})
 
 						sub.It("should be able to deploy a workload with ingress", func() error {

pkg/ansible/clustercatalog.go

@@ -121,6 +121,8 @@ type ClusterCatalog struct {
 	HTTPProxy  string `yaml:"http_proxy"`
 	HTTPSProxy string `yaml:"https_proxy"`
 	NoProxy    string `yaml:"no_proxy"`
+
+	NodeLabels map[string][]string `yaml:"node_labels"`
 }
 
 type NFSVolume struct {