Merge pull request #2526 from weaveworks/mike/k8s/add-daemonsets

Add daemonsets

app/api_topologies.go

@@ -29,6 +29,7 @@ const (
 	podsID                 = "pods"
 	replicaSetsID          = "replica-sets"
 	deploymentsID          = "deployments"
+	daemonsetsID           = "daemonsets"
 	servicesID             = "services"
 	hostsID                = "hosts"
 	weaveID                = "weave"
@@ -118,7 +119,7 @@ func updateKubeFilters(rpt report.Report, topologies []APITopologyDesc) []APITop
 	sort.Strings(ns)
 	topologies = append([]APITopologyDesc{}, topologies...) // Make a copy so we can make changes safely
 	for i, t := range topologies {
-		if t.id == containersID || t.id == podsID || t.id == servicesID || t.id == deploymentsID || t.id == replicaSetsID {
+		if t.id == containersID || t.id == podsID || t.id == servicesID || t.id == deploymentsID || t.id == replicaSetsID || t.id == daemonsetsID {
 			topologies[i] = mergeTopologyFilters(t, []APITopologyOptionGroup{
 				namespaceFilters(ns, "All Namespaces"),
 			})
@@ -258,6 +259,14 @@ func MakeRegistry() *Registry {
 			Options:     []APITopologyOptionGroup{unmanagedFilter},
 			HideIfEmpty: true,
 		},
+		APITopologyDesc{
+			id:          daemonsetsID,
+			parent:      podsID,
+			renderer:    render.DaemonSetRenderer,
+			Name:        "daemonsets",
+			Options:     []APITopologyOptionGroup{unmanagedFilter},
+			HideIfEmpty: true,
+		},
 		APITopologyDesc{
 			id:          servicesID,
 			parent:      podsID,

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,52 @@
+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, error)
+	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, error) {
+	selector, err := unversioned.LabelSelectorAsSelector(d.Spec.Selector)
+	if err != nil {
+		return nil, err
+	}
+	return selector, nil
+}
+
+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/deployment.go

@@ -21,7 +21,7 @@ const (
 // Deployment represents a Kubernetes deployment
 type Deployment interface {
 	Meta
-	Selector() labels.Selector
+	Selector() (labels.Selector, error)
 	GetNode(probeID string) report.Node
 }
 
@@ -36,13 +36,12 @@ func NewDeployment(d *extensions.Deployment) Deployment {
 	return &deployment{Deployment: d, Meta: meta{d.ObjectMeta}}
 }
 
-func (d *deployment) Selector() labels.Selector {
+func (d *deployment) Selector() (labels.Selector, error) {
 	selector, err := unversioned.LabelSelectorAsSelector(d.Spec.Selector)
 	if err != nil {
-		// TODO(paulbellamy): Remove the panic!
-		panic(err)
+		return nil, err
 	}
-	return selector
+	return selector, nil
 }
 
 func (d *deployment) GetNode(probeID string) report.Node {

probe/kubernetes/replica_set.go

@@ -18,7 +18,7 @@ const (
 // ReplicaSet represents a Kubernetes replica set
 type ReplicaSet interface {
 	Meta
-	Selector() labels.Selector
+	Selector() (labels.Selector, error)
 	AddParent(topology, id string)
 	GetNode(probeID string) report.Node
 }
@@ -39,13 +39,12 @@ func NewReplicaSet(r *extensions.ReplicaSet) ReplicaSet {
 	}
 }
 
-func (r *replicaSet) Selector() labels.Selector {
+func (r *replicaSet) Selector() (labels.Selector, error) {
 	selector, err := unversioned.LabelSelectorAsSelector(r.Spec.Selector)
 	if err != nil {
-		// TODO(paulbellamy): Remove the panic!
-		panic(err)
+		return nil, err
 	}
-	return selector
+	return selector, nil
 }
 
 func (r *replicaSet) AddParent(topology, id string) {

probe/kubernetes/replication_controller.go

@@ -11,11 +11,12 @@ import (
 // ReplicationController represents a Kubernetes replication controller
 type ReplicationController interface {
 	Meta
-	Selector() labels.Selector
+	Selector() (labels.Selector, error)
 	AddParent(topology, id string)
 	GetNode(probeID string) report.Node
 }
 
+// replicationController implements both ReplicationController and ReplicaSet
 type replicationController struct {
 	*api.ReplicationController
 	Meta
@@ -32,11 +33,11 @@ func NewReplicationController(r *api.ReplicationController) ReplicationControlle
 	}
 }
 
-func (r *replicationController) Selector() labels.Selector {
+func (r *replicationController) Selector() (labels.Selector, error) {
 	if r.Spec.Selector == nil {
-		return labels.Nothing()
+		return labels.Nothing(), nil
 	}
-	return labels.SelectorFromSet(labels.Set(r.Spec.Selector))
+	return labels.SelectorFromSet(labels.Set(r.Spec.Selector)), nil
 }
 
 func (r *replicationController) AddParent(topology, id string) {

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().
@@ -289,9 +317,14 @@ func (r *Reporter) replicaSetTopology(probeID string, deployments []Deployment)
 	result.Controls.AddControls(ScalingControls)
 
 	for _, deployment := range deployments {
+		selector, err := deployment.Selector()
+		if err != nil {
+			return result, replicaSets, err
+		}
+
 		selectors = append(selectors, match(
 			deployment.Namespace(),
-			deployment.Selector(),
+			selector,
 			report.Deployment,
 			report.MakeDeploymentNodeID(deployment.UID()),
 		))
@@ -335,7 +368,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).
@@ -364,13 +397,29 @@ func (r *Reporter) podTopology(services []Service, replicaSets []ReplicaSet) (re
 		))
 	}
 	for _, replicaSet := range replicaSets {
+		selector, err := replicaSet.Selector()
+		if err != nil {
+			return pods, err
+		}
 		selectors = append(selectors, match(
 			replicaSet.Namespace(),
-			replicaSet.Selector(),
+			selector,
 			report.ReplicaSet,
 			report.MakeReplicaSetNodeID(replicaSet.UID()),
 		))
 	}
+	for _, daemonSet := range daemonSets {
+		selector, err := daemonSet.Selector()
+		if err != nil {
+			return pods, err
+		}
+		selectors = append(selectors, match(
+			daemonSet.Namespace(),
+			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
 }

render/detailed/parents.go

@@ -25,6 +25,7 @@ var (
 		report.Pod:            kubernetesParentLabel,
 		report.ReplicaSet:     kubernetesParentLabel,
 		report.Deployment:     kubernetesParentLabel,
+		report.DaemonSet:      kubernetesParentLabel,
 		report.Service:        kubernetesParentLabel,
 		report.ECSTask:        latestLookup(awsecs.TaskFamily),
 		report.ECSService:     ecsServiceParentLabel,

render/detailed/summary.go

@@ -68,6 +68,7 @@ var renderers = map[string]func(NodeSummary, report.Node) (NodeSummary, bool){
 	report.Pod:            podNodeSummary,
 	report.Service:        podGroupNodeSummary,
 	report.Deployment:     podGroupNodeSummary,
+	report.DaemonSet:      podGroupNodeSummary,
 	report.ReplicaSet:     podGroupNodeSummary,
 	report.ECSTask:        ecsTaskNodeSummary,
 	report.ECSService:     ecsServiceNodeSummary,
@@ -91,6 +92,7 @@ var primaryAPITopology = map[string]string{
 	report.Pod:            "pods",
 	report.ReplicaSet:     "replica-sets",
 	report.Deployment:     "deployments",
+	report.DaemonSet:      "daemonsets",
 	report.Service:        "services",
 	report.ECSTask:        "ecs-tasks",
 	report.ECSService:     "ecs-services",

render/pod.go

@@ -96,6 +96,21 @@ var ReplicaSetRenderer = ConditionalRenderer(renderKubernetesTopologies,
 	),
 )
 
+// DaemonSetRenderer is a Renderer which produces a renderable kubernetes daemonsets
+// graph by merging the pods graph and the daemonsets topology.
+var DaemonSetRenderer = ConditionalRenderer(renderKubernetesTopologies,
+	MakeMap(
+		PropagateSingleMetrics(report.Pod),
+		MakeReduce(
+			MakeMap(
+				Map2Parent(report.DaemonSet, "", nil),
+				PodRenderer,
+			),
+			SelectDaemonSet,
+		),
+	),
+)
+
 func mapPodCounts(parent, original report.Node) report.Node {
 	// When mapping ReplicaSets to Deployments, we want to propagate the Pods counter
 	if count, ok := original.Counters.Lookup(report.Pod); ok {

render/selectors.go

@@ -30,6 +30,7 @@ var (
 	SelectPod            = TopologySelector(report.Pod)
 	SelectService        = TopologySelector(report.Service)
 	SelectDeployment     = TopologySelector(report.Deployment)
+	SelectDaemonSet      = TopologySelector(report.DaemonSet)
 	SelectReplicaSet     = TopologySelector(report.ReplicaSet)
 	SelectECSTask        = TopologySelector(report.ECSTask)
 	SelectECSService     = TopologySelector(report.ECSService)

report/id.go

@@ -125,6 +125,12 @@ var (
 	// ParseReplicaSetNodeID parses a replica set node ID
 	ParseReplicaSetNodeID = parseSingleComponentID("replica_set")
 
+	// MakeDaemonSetNodeID produces a replica set node ID from its composite parts.
+	MakeDaemonSetNodeID = makeSingleComponentID("daemonset")
+
+	// ParseDaemonSetNodeID parses a daemon set node ID
+	ParseDaemonSetNodeID = parseSingleComponentID("daemonset")
+
 	// MakeECSTaskNodeID produces a replica set node ID from its composite parts.
 	MakeECSTaskNodeID = makeSingleComponentID("ecs_task")
 

report/report.go

@@ -19,6 +19,7 @@ const (
 	Service        = "service"
 	Deployment     = "deployment"
 	ReplicaSet     = "replica_set"
+	DaemonSet      = "daemon_set"
 	ContainerImage = "container_image"
 	Host           = "host"
 	Overlay        = "overlay"
@@ -74,6 +75,11 @@ type Report struct {
 	// present.
 	ReplicaSet Topology
 
+	// DaemonSet nodes represent all Kubernetes DaemonSets running on hosts running probes.
+	// Metadata includes things like DaemonSet id, name etc. Edges are not
+	// present.
+	DaemonSet Topology
+
 	// ContainerImages nodes represent all Docker containers images on
 	// hosts running probes. Metadata includes things like image id, name etc.
 	// Edges are not present.
@@ -164,6 +170,10 @@ func MakeReport() Report {
 			WithShape(Heptagon).
 			WithLabel("replica set", "replica sets"),
 
+		DaemonSet: MakeTopology().
+			WithShape(Heptagon).
+			WithLabel("daemonset", "daemonsets"),
+
 		Overlay: MakeTopology().
 			WithShape(Circle).
 			WithLabel("peer", "peers"),
@@ -198,6 +208,7 @@ func (r *Report) TopologyMap() map[string]*Topology {
 		Service:        &r.Service,
 		Deployment:     &r.Deployment,
 		ReplicaSet:     &r.ReplicaSet,
+		DaemonSet:      &r.DaemonSet,
 		Host:           &r.Host,
 		Overlay:        &r.Overlay,
 		ECSTask:        &r.ECSTask,
@@ -260,6 +271,7 @@ func (r *Report) WalkPairedTopologies(o *Report, f func(*Topology, *Topology)) {
 	f(&r.Service, &o.Service)
 	f(&r.Deployment, &o.Deployment)
 	f(&r.ReplicaSet, &o.ReplicaSet)
+	f(&r.DaemonSet, &o.DaemonSet)
 	f(&r.Host, &o.Host)
 	f(&r.Overlay, &o.Overlay)
 	f(&r.ECSTask, &o.ECSTask)