probe: merge controls from different probes

This change creates a host topology from cluster-wide probe using
kubernetes nodes with cordon/un-cordon control. This also adds the
logic to merge the controls coming from different probes for host
topology.

Author: goku321

examples/k8s/cluster-role.yaml

@@ -15,18 +15,13 @@ rules:
   - pods/log
   - replicationcontrollers
   - services
-  - nodes
   - namespaces
   - persistentvolumes
   - persistentvolumeclaims
   verbs:
   - get
   - list
   - watch
-  - create
-  - update
-  - patch
-  - delete
 - apiGroups:
   - ""
   resources:
@@ -99,3 +94,13 @@ rules:
   verbs:
   - list
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - nodes
+  verbs:
+  - get
+  - list
+  - watch
+  - update
+  - patch

probe/kubernetes/client.go

@@ -67,6 +67,8 @@ type Client interface {
 	ScaleDown(namespaceID, id string) error
 	// Cordon or Uncordon a node based on whether `desired` is true or false respectively.
 	CordonNode(name string, desired bool) error
+	// Returns a list of kubernetes nodes.
+	GetNodes() ([]apiv1.Node, error)
 }
 
 // ResourceMap is the mapping of resource and their GroupKind
@@ -647,3 +649,12 @@ func (c *client) CordonNode(name string, desired bool) error {
 	}
 	return nil
 }
+
+func (c *client) GetNodes() ([]apiv1.Node, error) {
+	l, err := c.client.CoreV1().Nodes().List(metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	return l.Items, nil
+}

probe/kubernetes/reporter.go

@@ -302,10 +302,6 @@ func (r *Tagger) Tag(rpt report.Report) (report.Report, error) {
 
 		rpt.Container.Nodes[id] = n.WithParent(report.Pod, report.MakePodNodeID(uid))
 	}
-	for id, n := range rpt.Host.Nodes {
-		controls := append(n.ActiveControls(), CordonNode, UncordonNode)
-		rpt.Host.Nodes[id] = n.WithLatestActiveControls(controls...)
-	}
 	return rpt, nil
 }
 
@@ -364,6 +360,10 @@ func (r *Reporter) Report() (report.Report, error) {
 	if err != nil {
 		return result, err
 	}
+	hostTopology, err := r.hostTopology()
+	if err != nil {
+		return result, err
+	}
 
 	result.Pod = result.Pod.Merge(podTopology)
 	result.Service = result.Service.Merge(serviceTopology)
@@ -378,12 +378,7 @@ func (r *Reporter) Report() (report.Report, error) {
 	result.VolumeSnapshot = result.VolumeSnapshot.Merge(volumeSnapshotTopology)
 	result.VolumeSnapshotData = result.VolumeSnapshotData.Merge(volumeSnapshotDataTopology)
 	result.Job = result.Job.Merge(jobTopology)
-
-	// Add buttons for Host view, with the ID of the Kubernetes probe
-	for _, control := range CordonControl {
-		control.ProbeID = r.probeID
-		result.Host.Controls.AddControl(control)
-	}
+	result.Host = result.Host.Merge(hostTopology)
 
 	return result, nil
 }
@@ -704,3 +699,32 @@ func (r *Reporter) namespaceTopology() (report.Topology, error) {
 	})
 	return result, err
 }
+
+func (r *Reporter) hostTopology() (report.Topology, error) {
+	result := report.MakeTopology()
+	// Add buttons for Host view, with the ID of the Kubernetes probe
+	for _, control := range CordonControl {
+		control.ProbeID = r.probeID
+		result.Controls.AddControl(control)
+	}
+
+	nodes, err := r.client.GetNodes()
+	if err != nil {
+		return result, err
+	}
+
+	for _, n := range nodes {
+		var activeControl string
+		if n.Spec.Unschedulable {
+			activeControl = UncordonNode
+		} else {
+			activeControl = CordonNode
+		}
+		result.AddNode(
+			report.MakeNode(report.MakeHostNodeID(n.Name)).
+				WithTopology("host").
+				WithLatestActiveControls(activeControl),
+		)
+	}
+	return result, nil
+}

probe/kubernetes/reporter_test.go

@@ -209,6 +209,10 @@ func (c *mockClient) CordonNode(name string, desired bool) error {
 	return nil
 }
 
+func (c *mockClient) GetNodes() ([]apiv1.Node, error) {
+	return nil, nil
+}
+
 type mockPipeClient map[string]xfer.Pipe
 
 func (c mockPipeClient) PipeConnection(appID, id string, pipe xfer.Pipe) error {

report/node.go

@@ -173,7 +173,7 @@ func (n Node) WithChild(child Node) Node {
 	return n
 }
 
-// Merge mergses the individual components of a node and returns a
+// Merge merges the individual components of a node and returns a
 // fresh node.
 func (n Node) Merge(other Node) Node {
 	id := n.ID
@@ -186,7 +186,8 @@ func (n Node) Merge(other Node) Node {
 	} else if other.Topology != "" && topology != other.Topology {
 		panic("Cannot merge nodes with different topology types: " + topology + " != " + other.Topology)
 	}
-	return Node{
+
+	newNode := Node{
 		ID:        id,
 		Topology:  topology,
 		Sets:      n.Sets.Merge(other.Sets),
@@ -196,6 +197,39 @@ func (n Node) Merge(other Node) Node {
 		Parents:   n.Parents.Merge(other.Parents),
 		Children:  n.Children.Merge(other.Children),
 	}
+
+	// Special case to merge controls from two different probes.
+	if topology == "host" {
+		controlString, _ := newNode.Latest.Lookup(NodeActiveControls)
+		activeControls := strings.Split(controlString, ScopeDelim)
+		// Host topology should have only two active controls.
+		// Length equals to 2 means controls were merged properly.
+		if len(activeControls) != 2 {
+			// Check if host_exec control is missing.
+			if !StringSet(activeControls).Contains("host_exec") {
+				activeControls = append(activeControls, "host_exec")
+				return newNode.
+					WithLatestActiveControls(activeControls...)
+			}
+			// Else control kubernetes_cordon_node or kubernetes_uncordon_node is missing.
+			// Merge controls from current node.
+			c, _ := n.Latest.Lookup(NodeActiveControls)
+			x := strings.Split(c, ScopeDelim)
+			activeControls = append(activeControls, x...)
+
+			// Merge controls from incoming node.
+			c, _ = other.Latest.Lookup(NodeActiveControls)
+			x = strings.Split(c, ScopeDelim)
+			activeControls = append(activeControls, x...)
+
+			// Use a StringSet to avoid duplicate controls from current and incoming node.
+			var final StringSet
+			return newNode.
+				WithLatestActiveControls(final.Add(activeControls...)...)
+		}
+	}
+
+	return newNode
 }
 
 // UnsafeUnMerge removes data from n that would be added by merging other,