k8s: added cronjob probe

Author: hunchback

etc/skydive.yml.default

@@ -368,6 +368,7 @@ k8s:
   probes:
   - cluster
   - container
+  - cronjob
   - deployment
   - ingress
   - job

statics/img/cronjob.png

@@ -27,6 +27,7 @@ var nodeImgMap = setupFixedImages({
   "bond": "port",
   "cluster": "cluster",
   "container": "container",
+  "cronjob": "cronjob",
   "daemonset": "daemonset",
   "deployment": "deployment",
   "ingress": "ingress",

statics/js/components/layout.js

@@ -0,0 +1,19 @@
+apiVersion: batch/v1beta1
+kind: CronJob
+metadata:
+  name: skydive-test-cronjob
+spec:
+  schedule: "*/1 * * * *"
+  jobTemplate:
+    spec:
+      template:
+        spec:
+          containers:
+          - name: hello
+            image: busybox
+            args:
+            - /bin/sh
+            - -c
+            - date; echo Hello from the Kubernetes cluster
+          restartPolicy: OnFailure
+

tests/k8s/cronjob.yaml

@@ -165,6 +165,10 @@ func TestK8sContainerNode(t *testing.T) {
 	testNodeCreationFromConfig(t, "container", objName+"-container", "Image", "Labels", "Pod")
 }
 
+func TestK8sCronJobNode(t *testing.T) {
+	testNodeCreationFromConfig(t, "cronjob", objName+"-cronjob")
+}
+
 func TestK8sDeploymentNode(t *testing.T) {
 	testNodeCreationFromConfig(t, "deployment", objName+"-deployment")
 }

tests/k8s_test.go

@@ -0,0 +1,109 @@
+/*
+ * Copyright (C) 2018 IBM, Inc.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package k8s
+
+import (
+	"fmt"
+
+	"github.com/skydive-project/skydive/logging"
+	"github.com/skydive-project/skydive/probe"
+	"github.com/skydive-project/skydive/topology/graph"
+
+	"k8s.io/api/batch/v1beta1"
+	"k8s.io/client-go/tools/cache"
+)
+
+type cronJobProbe struct {
+	defaultKubeCacheEventHandler
+	*kubeCache
+	graph *graph.Graph
+}
+
+func dumpCronJob(cj *v1beta1.CronJob) string {
+	return fmt.Sprintf("cronjob{Namespace: %s, Name: %s}", cj.Namespace, cj.Name)
+}
+
+func (p *cronJobProbe) newMetadata(cj *v1beta1.CronJob) graph.Metadata {
+	m := newMetadata("cronjob", cj.Namespace, cj.Name, cj)
+	m.SetField("Schedule", cj.Spec.Schedule)
+	m.SetField("Suspended", cj.Spec.Suspend != nil && *cj.Spec.Suspend)
+	return m
+}
+
+func cronJobUID(cj *v1beta1.CronJob) graph.Identifier {
+	return graph.Identifier(cj.GetUID())
+}
+
+func (p *cronJobProbe) OnAdd(obj interface{}) {
+	if cj, ok := obj.(*v1beta1.CronJob); ok {
+		p.graph.Lock()
+		defer p.graph.Unlock()
+
+		newNode(p.graph, cronJobUID(cj), p.newMetadata(cj))
+		logging.GetLogger().Debugf("Added %s", dumpCronJob(cj))
+	}
+}
+
+func (p *cronJobProbe) OnUpdate(oldObj, newObj interface{}) {
+	if cj, ok := newObj.(*v1beta1.CronJob); ok {
+		p.graph.Lock()
+		defer p.graph.Unlock()
+
+		if node := p.graph.GetNode(cronJobUID(cj)); node != nil {
+			addMetadata(p.graph, node, cj)
+			logging.GetLogger().Debugf("Updated %s", dumpCronJob(cj))
+		}
+	}
+}
+
+func (p *cronJobProbe) OnDelete(obj interface{}) {
+	if cj, ok := obj.(*v1beta1.CronJob); ok {
+		p.graph.Lock()
+		defer p.graph.Unlock()
+
+		if node := p.graph.GetNode(cronJobUID(cj)); node != nil {
+			p.graph.DelNode(node)
+			logging.GetLogger().Debugf("Deleted %s", dumpCronJob(cj))
+		}
+	}
+}
+
+func (p *cronJobProbe) Start() {
+	p.kubeCache.Start()
+}
+
+func (p *cronJobProbe) Stop() {
+	p.kubeCache.Stop()
+}
+
+func newCronJobKubeCache(handler cache.ResourceEventHandler) *kubeCache {
+	return newKubeCache(getClientset().BatchV1beta1().RESTClient(), &v1beta1.CronJob{}, "cronjobs", handler)
+}
+
+func newCronJobProbe(g *graph.Graph) probe.Probe {
+	p := &cronJobProbe{
+		graph: g,
+	}
+	p.kubeCache = newCronJobKubeCache(p)
+	return p
+}

topology/probes/k8s/cronjob.go

@@ -45,6 +45,7 @@ type namespaceProbe struct {
 
 func newObjectIndexer(g *graph.Graph) *graph.MetadataIndexer {
 	ownedByNamespaceFilter := filters.NewOrFilter(
+		filters.NewTermStringFilter("Type", "cronjob"),
 		filters.NewTermStringFilter("Type", "deployment"),
 		filters.NewTermStringFilter("Type", "daemonset"),
 		filters.NewTermStringFilter("Type", "ingress"),

topology/probes/k8s/namespace.go

@@ -51,6 +51,7 @@ func makeProbeBundle(g *graph.Graph) *probe.ProbeBundle {
 	name2ctor := map[string](func(*graph.Graph) probe.Probe){
 		"cluster":               newClusterProbe,
 		"container":             newContainerProbe,
+		"cronjob":               newCronJobProbe,
 		"daemonset":             newDaemonSetProbe,
 		"deployment":            newDeploymentProbe,
 		"ingress":               newIngressProbe,