k8s probe: Collect info on daemonsets for new DaemonSet topology

probe/kubernetes/client.go

@@ -26,6 +26,7 @@ type Client interface {
 	WalkServices(f func(Service) error) error
 	WalkDeployments(f func(Deployment) error) error
 	WalkReplicaSets(f func(ReplicaSet) error) error
+	WalkDaemonSets(f func(DaemonSet) error) error
 	WalkReplicationControllers(f func(ReplicationController) error) error
 	WalkNodes(f func(*api.Node) error) error
 
@@ -46,6 +47,7 @@ type client struct {
 	serviceStore               *cache.StoreToServiceLister
 	deploymentStore            *cache.StoreToDeploymentLister
 	replicaSetStore            *cache.StoreToReplicaSetLister
+	daemonSetStore             *cache.StoreToDaemonSetLister
 	replicationControllerStore *cache.StoreToReplicationControllerLister
 	nodeStore                  *cache.StoreToNodeLister
 
@@ -155,6 +157,7 @@ func NewClient(config ClientConfig) (Client, error) {
 	} else {
 		result.deploymentStore = &cache.StoreToDeploymentLister{Store: result.setupStore(ec, "deployments", &extensions.Deployment{}, nil)}
 		result.replicaSetStore = &cache.StoreToReplicaSetLister{Store: result.setupStore(ec, "replicasets", &extensions.ReplicaSet{}, nil)}
+		result.daemonSetStore = &cache.StoreToDaemonSetLister{Store: result.setupStore(ec, "daemonsets", &extensions.DaemonSet{}, nil)}
 	}
 
 	return result, nil
@@ -258,6 +261,20 @@ func (c *client) WalkReplicationControllers(f func(ReplicationController) error)
 	return nil
 }
 
+// WalkDaemonSets calls f for each daemonset
+func (c *client) WalkDaemonSets(f func(DaemonSet) error) error {
+	list, err := c.daemonSetStore.List()
+	if err != nil {
+		return err
+	}
+	for i := range list.Items {
+		if err := f(NewDaemonSet(&(list.Items[i]))); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (c *client) WalkNodes(f func(*api.Node) error) error {
 	list, err := c.nodeStore.List()
 	if err != nil {

probe/kubernetes/daemonset.go

@@ -0,0 +1,53 @@
+package kubernetes
+
+import (
+	"fmt"
+
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/apis/extensions"
+	"k8s.io/kubernetes/pkg/labels"
+
+	"github.com/weaveworks/scope/report"
+)
+
+// These constants are keys used in node metadata
+const (
+	MisscheduledReplicas = "kubernetes_misscheduled_replicas"
+)
+
+// DaemonSet represents a Kubernetes daemonset
+type DaemonSet interface {
+	Meta
+	Selector() labels.Selector
+	GetNode() report.Node
+}
+
+type daemonSet struct {
+	*extensions.DaemonSet
+	Meta
+}
+
+// NewDaemonSet creates a new daemonset
+func NewDaemonSet(d *extensions.DaemonSet) DaemonSet {
+	return &daemonSet{
+		DaemonSet: d,
+		Meta:      meta{d.ObjectMeta},
+	}
+}
+
+func (d *daemonSet) Selector() labels.Selector {
+	selector, err := unversioned.LabelSelectorAsSelector(d.Spec.Selector)
+	if err != nil {
+		// TODO(paulbellamy): Remove the panic!
+		panic(err)
+	}
+	return selector
+}
+
+func (d *daemonSet) GetNode() report.Node {
+	return d.MetaNode(report.MakeDaemonSetNodeID(d.UID())).WithLatests(map[string]string{
+		DesiredReplicas:      fmt.Sprint(d.Status.DesiredNumberScheduled),
+		Replicas:             fmt.Sprint(d.Status.CurrentNumberScheduled),
+		MisscheduledReplicas: fmt.Sprint(d.Status.NumberMisscheduled),
+	})
+}

probe/kubernetes/reporter.go

@@ -66,6 +66,15 @@ var (
 
 	ReplicaSetMetricTemplates = PodMetricTemplates
 
+	DaemonSetMetadataTemplates = report.MetadataTemplates{
+		Namespace:       {ID: Namespace, Label: "Namespace", From: report.FromLatest, Priority: 2},
+		Created:         {ID: Created, Label: "Created", From: report.FromLatest, Datatype: "datetime", Priority: 3},
+		DesiredReplicas: {ID: DesiredReplicas, Label: "Desired Replicas", From: report.FromLatest, Datatype: "number", Priority: 4},
+		report.Pod:      {ID: report.Pod, Label: "# Pods", From: report.FromCounters, Datatype: "number", Priority: 5},
+	}
+
+	DaemonSetMetricTemplates = PodMetricTemplates
+
 	TableTemplates = report.TableTemplates{
 		LabelPrefix: {
 			ID:     LabelPrefix,
@@ -204,6 +213,10 @@ func (r *Reporter) Report() (report.Report, error) {
 	if err != nil {
 		return result, err
 	}
+	daemonSetTopology, daemonSets, err := r.daemonSetTopology()
+	if err != nil {
+		return result, err
+	}
 	deploymentTopology, deployments, err := r.deploymentTopology(r.probeID)
 	if err != nil {
 		return result, err
@@ -212,13 +225,14 @@ func (r *Reporter) Report() (report.Report, error) {
 	if err != nil {
 		return result, err
 	}
-	podTopology, err := r.podTopology(services, replicaSets)
+	podTopology, err := r.podTopology(services, replicaSets, daemonSets)
 	if err != nil {
 		return result, err
 	}
 	result.Pod = result.Pod.Merge(podTopology)
 	result.Service = result.Service.Merge(serviceTopology)
 	result.Host = result.Host.Merge(hostTopology)
+	result.DaemonSet = result.DaemonSet.Merge(daemonSetTopology)
 	result.Deployment = result.Deployment.Merge(deploymentTopology)
 	result.ReplicaSet = result.ReplicaSet.Merge(replicaSetTopology)
 	return result, nil
@@ -277,6 +291,20 @@ func (r *Reporter) deploymentTopology(probeID string) (report.Topology, []Deploy
 	return result, deployments, err
 }
 
+func (r *Reporter) daemonSetTopology() (report.Topology, []DaemonSet, error) {
+	daemonSets := []DaemonSet{}
+	result := report.MakeTopology().
+		WithMetadataTemplates(DaemonSetMetadataTemplates).
+		WithMetricTemplates(DaemonSetMetricTemplates).
+		WithTableTemplates(TableTemplates)
+	err := r.client.WalkDaemonSets(func(d DaemonSet) error {
+		result = result.AddNode(d.GetNode())
+		daemonSets = append(daemonSets, d)
+		return nil
+	})
+	return result, daemonSets, err
+}
+
 func (r *Reporter) replicaSetTopology(probeID string, deployments []Deployment) (report.Topology, []ReplicaSet, error) {
 	var (
 		result = report.MakeTopology().
@@ -335,7 +363,7 @@ func match(namespace string, selector labels.Selector, topology, id string) func
 	}
 }
 
-func (r *Reporter) podTopology(services []Service, replicaSets []ReplicaSet) (report.Topology, error) {
+func (r *Reporter) podTopology(services []Service, replicaSets []ReplicaSet, daemonSets []DaemonSet) (report.Topology, error) {
 	var (
 		pods = report.MakeTopology().
 			WithMetadataTemplates(PodMetadataTemplates).
@@ -371,6 +399,14 @@ func (r *Reporter) podTopology(services []Service, replicaSets []ReplicaSet) (re
 			report.MakeReplicaSetNodeID(replicaSet.UID()),
 		))
 	}
+	for _, daemonSet := range daemonSets {
+		selectors = append(selectors, match(
+			daemonSet.Namespace(),
+			daemonSet.Selector(),
+			report.DaemonSet,
+			report.MakeDaemonSetNodeID(daemonSet.UID()),
+		))
+	}
 
 	// Obtain the local pods from kubelet since we only want to report those
 	// for performance reasons.

probe/kubernetes/reporter_test.go

@@ -136,6 +136,9 @@ func (c *mockClient) WalkServices(f func(kubernetes.Service) error) error {
 	}
 	return nil
 }
+func (c *mockClient) WalkDaemonSets(f func(kubernetes.DaemonSet) error) error {
+	return nil
+}
 func (c *mockClient) WalkDeployments(f func(kubernetes.Deployment) error) error {
 	return nil
 }