Add adjacencies for kubernetes storage components

- Kubernetes storage components such as PV and PVC are connected based on two
parameters Persistent volume claim name and Persistent Volume name.
- PVC contains the volume name which is, PV name itself. Hence, we can
show edge for PVC and PV.
- This will bring higher level visibility for kubernetes storage components.

Signed-off-by: Satyam Zode <satyam.zode@openebs.io>

render/expected/expected.go

@@ -3,6 +3,7 @@ package expected
 import (
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/probe/host"
+	"github.com/weaveworks/scope/probe/kubernetes"
 	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/render"
 	"github.com/weaveworks/scope/report"
@@ -16,6 +17,7 @@ var (
 	heptagon = "heptagon"
 	hexagon  = "hexagon"
 	cloud    = "cloud"
+	cylinder = "cylinder"
 
 	// Helper to make a report.node with some common options
 	node = func(topology string) func(id string, adjacent ...string) report.Node {
@@ -37,6 +39,8 @@ var (
 	pod                   = node(report.Pod)
 	service               = node(report.Service)
 	hostNode              = node(report.Host)
+	persistentVolume      = node(report.PersistentVolume)
+	persistentVolumeClaim = node(report.PersistentVolumeClaim)
 
 	UnknownPseudoNode1ID = render.MakePseudoNodeID(fixture.UnknownClient1IP)
 	UnknownPseudoNode2ID = render.MakePseudoNodeID(fixture.UnknownClient3IP)
@@ -323,6 +327,28 @@ var (
 		render.IncomingInternetID: theIncomingInternetNode(fixture.ServerHostNodeID),
 		render.OutgoingInternetID: theOutgoingInternetNode,
 	}
+
+	RenderedPersistentVolume = report.Nodes{
+		fixture.PersistentVolumeClaimNodeID: persistentVolumeClaim(fixture.PersistentVolumeClaimNodeID, fixture.PersistentVolumeNodeID).
+			WithLatests(map[string]string{
+				kubernetes.Name:             "pvc-6124",
+				kubernetes.Namespace:        "ping",
+				kubernetes.Status:           "bound",
+				kubernetes.VolumeName:       "pongvolume",
+				kubernetes.AccessModes:      "ReadWriteOnce",
+				kubernetes.StorageClassName: "standard",
+			}).WithChild(report.MakeNode(fixture.PersistentVolumeNodeID).WithTopology(report.PersistentVolume)),
+
+		fixture.PersistentVolumeNodeID: persistentVolume(fixture.PersistentVolumeNodeID).
+			WithLatests(map[string]string{
+				kubernetes.Name:             "pongvolume",
+				kubernetes.Namespace:        "ping",
+				kubernetes.Status:           "bound",
+				kubernetes.VolumeClaim:      "pvc-6124",
+				kubernetes.AccessModes:      "ReadWriteOnce",
+				kubernetes.StorageClassName: "standard",
+			}),
+	}
 )
 
 func newu64(value uint64) *uint64 { return &value }

render/filters.go

@@ -80,7 +80,10 @@ func (f FilterFunc) Transform(nodes Nodes) Nodes {
 				newAdjacency = newAdjacency.Add(dstID)
 			}
 		}
-		node.Adjacency = newAdjacency
+		claimName, ok := node.Latest.Lookup(kubernetes.VolumeClaim)
+		if claimName == "" || !ok {
+			node.Adjacency = newAdjacency
+		}
 		output[id] = node
 	}
 

render/persistentvolume.go

@@ -0,0 +1,155 @@
+package render
+
+import (
+	"github.com/weaveworks/scope/probe/kubernetes"
+	"github.com/weaveworks/scope/report"
+)
+
+// PersistentVolumeRenderer is the common renderer for all the storage components.
+var PersistentVolumeRenderer = Memoise(
+	MakeReduce(
+		ConnectionStorageJoin(
+			Map2PVName,
+			report.PersistentVolumeClaim,
+		),
+		MakeFilter(
+			func(n report.Node) bool {
+				claimName, ok := n.Latest.Lookup(kubernetes.VolumeClaim)
+				if claimName == "" {
+					return !ok
+				}
+				return ok
+			},
+			MakeReduce(
+				PropagateSingleMetrics(report.Container,
+					MakeMap(
+						Map3Parent([]string{report.Pod}),
+						MakeFilter(
+							ComposeFilterFuncs(
+								IsRunning,
+								Complement(isPauseContainer),
+							),
+							ContainerWithImageNameRenderer,
+						),
+					),
+				),
+				ConnectionStorageJoin(
+					Map2PVCName,
+					report.Pod,
+				),
+			),
+		),
+		MapStorageEndpoints(
+			Map2PVNode,
+			report.PersistentVolume,
+		),
+	),
+)
+
+// Map3Parent returns a MapFunc which maps Nodes to some parent grouping.
+func Map3Parent(
+	// The topology IDs to look for parents in
+	topologies []string,
+) MapFunc {
+	return func(n report.Node) report.Nodes {
+		result := report.Nodes{}
+		for _, topology := range topologies {
+			if groupIDs, ok := n.Parents.Lookup(topology); ok {
+				for _, id := range groupIDs {
+					node := NewDerivedNode(id, n).WithTopology(topology)
+					node.Counters = node.Counters.Add(n.Topology, 1)
+					result[id] = node
+				}
+			}
+		}
+		return result
+	}
+}
+
+// ConnectionStorageJoin returns connectionStorageJoin object
+func ConnectionStorageJoin(toPV func(report.Node) string, topology string) Renderer {
+	return connectionStorageJoin{toPV: toPV, topology: topology}
+}
+
+// connectionStorageJoin holds the information about mapping of storage components
+// along with TopologySelector
+type connectionStorageJoin struct {
+	toPV     func(report.Node) string
+	topology string
+}
+
+func (c connectionStorageJoin) Render(rpt report.Report) Nodes {
+	inputNodes := TopologySelector(c.topology).Render(rpt).Nodes
+
+	var pvcNodes = map[string]string{}
+	for _, n := range inputNodes {
+		pvName := c.toPV(n)
+		pvcNodes[pvName] = n.ID
+	}
+
+	return MapStorageEndpoints(
+		func(m report.Node) string {
+			pvName, ok := m.Latest.Lookup(kubernetes.Name)
+			if !ok {
+				return ""
+			}
+			id := pvcNodes[pvName]
+			return id
+		}, c.topology).Render(rpt)
+}
+
+// Map2PVName accepts PV Node and returns Volume name associated with PV Node.
+func Map2PVName(m report.Node) string {
+	pvName, ok := m.Latest.Lookup(kubernetes.VolumeName)
+	if !ok {
+		pvName = ""
+	}
+	return pvName
+}
+
+// Map2PVCName returns pvc name
+func Map2PVCName(m report.Node) string {
+	pvcName, ok := m.Latest.Lookup(kubernetes.VolumeClaim)
+	if !ok {
+		pvcName = ""
+	}
+	return pvcName
+}
+
+// Map2PVNode returns pv node ID
+func Map2PVNode(n report.Node) string {
+	if pvNodeID, ok := n.Latest.Lookup(report.MakePersistentVolumeNodeID(n.ID)); ok {
+		return pvNodeID
+	}
+	return ""
+}
+
+// mapStorageEndpoints is the Renderer for rendering storage components together.
+type mapStorageEndpoints struct {
+	f        endpointMapFunc
+	topology string
+}
+
+// MapStorageEndpoints instantiates mapStorageEndpoints and returns same
+func MapStorageEndpoints(f endpointMapFunc, topology string) Renderer {
+	return mapStorageEndpoints{f: f, topology: topology}
+}
+
+func (e mapStorageEndpoints) Render(rpt report.Report) Nodes {
+
+	var endpoints Nodes
+	if e.topology == "persistent_volume_claim" {
+		endpoints = SelectPersistentVolume.Render(rpt)
+	}
+	if e.topology == "pod" {
+		endpoints = SelectPersistentVolumeClaim.Render(rpt)
+	}
+	ret := newJoinResults(TopologySelector(e.topology).Render(rpt).Nodes)
+
+	for _, n := range endpoints.Nodes {
+		if id := e.f(n); id != "" {
+			ret.addChild(n, id, e.topology)
+		}
+	}
+	return ret.storageResult(endpoints)
+}

render/persistentvolume_test.go

@@ -0,0 +1,20 @@
+package render_test
+
+import (
+	"testing"
+
+	"github.com/weaveworks/common/test"
+	"github.com/weaveworks/scope/render"
+	"github.com/weaveworks/scope/render/expected"
+	"github.com/weaveworks/scope/test/fixture"
+	"github.com/weaveworks/scope/test/reflect"
+	"github.com/weaveworks/scope/test/utils"
+)
+
+func TestPersistentVolumeRenderer(t *testing.T) {
+	have := utils.Prune(render.PersistentVolumeRenderer.Render(fixture.Report).Nodes)
+	want := utils.Prune(expected.RenderedPersistentVolume)
+	if !reflect.DeepEqual(want, have) {
+		t.Error(test.Diff(want, have))
+	}
+}

render/render.go

@@ -251,6 +251,26 @@ func (ret *joinResults) rewriteAdjacency(outID string, adjacency report.IDList)
 	ret.nodes[outID] = out
 }
 
+// storageAdjacency sets adjacency for the given node ID
+func (ret *joinResults) storageAdjacency(outID string, adjacency string) {
+	out := ret.nodes[outID]
+	out.Adjacency = out.Adjacency.Add(adjacency)
+	ret.nodes[outID] = out
+}
+
+// storageResult returns Nodes for after adding adjacencies
+func (ret *joinResults) storageResult(input Nodes) Nodes {
+	for _, n := range input.Nodes {
+		outID, ok := ret.mapped[n.ID]
+		if !ok {
+			continue
+		}
+		// Since PV and PVC will have only single adjacency
+		ret.storageAdjacency(outID, n.ID)
+	}
+	return Nodes{Nodes: ret.nodes}
+}
+
 // ResetCache blows away the rendered node cache, and known service
 // cache.
 func ResetCache() {