Add support for kubernetes jobs

influxdata · Jul 11, 2020 · 7a16e9d · 7a16e9d
1 parent d259515
commit 7a16e9d
Show file tree

Hide file tree

Showing 5 changed files with 498 additions and 8 deletions.
diff --git a/plugins/inputs/kube_inventory/README.md b/plugins/inputs/kube_inventory/README.md
@@ -6,6 +6,7 @@ This plugin generates metrics derived from the state of the following Kubernetes
 - deployments
 - endpoints
 - ingress
+- jobs
 - nodes
 - persistentvolumes
 - persistentvolumeclaims
@@ -55,7 +56,7 @@ avoid cardinality issues:
 
   ## Optional Resources to exclude from gathering
   ## Leave them with blank with try to gather everything available.
-  ## Values can be - "daemonsets", deployments", "endpoints", "ingress", "nodes",
+  ## Values can be - "daemonsets", deployments", "endpoints", "ingress", "jobs", "nodes",
   ## "persistentvolumes", "persistentvolumeclaims", "pods", "services", "statefulsets"
   # resource_exclude = [ "deployments", "nodes", "statefulsets" ]
 
@@ -186,7 +187,19 @@ subjects:
     - backend_service_port
     - tls
 
-- kubernetes_node
+- kubernetes_job
+  - tags:
+    - job_name
+    - namespace
+    - selector (*varies)
+  - fields:
+    - active
+    - completed
+    - failed
+    - started
+    - succeeded
+
+* kubernetes_node
   - tags:
     - node_name
   - fields:
@@ -197,15 +210,15 @@ subjects:
     - allocatable_memory_bytes
     - allocatable_pods
 
-* kubernetes_persistentvolume
+- kubernetes_persistentvolume
   - tags:
     - pv_name
     - phase
     - storageclass
   - fields:
     - phase_type (int, [see below](#pv-phase_type))
 
-- kubernetes_persistentvolumeclaim
+* kubernetes_persistentvolumeclaim
   - tags:
     - pvc_name
     - namespace
@@ -215,7 +228,7 @@ subjects:
   - fields:
     - phase_type (int, [see below](#pvc-phase_type))
 
-* kubernetes_pod_container
+- kubernetes_pod_container
   - tags:
     - container_name
     - namespace
@@ -234,7 +247,7 @@ subjects:
     - resource_limits_cpu_units
     - resource_limits_memory_bytes
 
-- kubernetes_service
+* kubernetes_service
   - tags:
     - service_name
     - namespace
@@ -249,7 +262,7 @@ subjects:
     - port
     - target_port
 
-* kubernetes_statefulset
+- kubernetes_statefulset
   - tags:
     - statefulset_name
     - namespace

diff --git a/plugins/inputs/kube_inventory/client.go b/plugins/inputs/kube_inventory/client.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/ericchiang/k8s"
 	v1APPS "github.com/ericchiang/k8s/apis/apps/v1"
+	v1BATCH "github.com/ericchiang/k8s/apis/batch/v1"
 	v1 "github.com/ericchiang/k8s/apis/core/v1"
 	v1beta1EXT "github.com/ericchiang/k8s/apis/extensions/v1beta1"
 
@@ -75,6 +76,13 @@ func (c *client) getIngress(ctx context.Context) (*v1beta1EXT.IngressList, error
 	return list, c.List(ctx, c.namespace, list)
 }
 
+func (c *client) getJobs(ctx context.Context) (*v1BATCH.JobList, error) {
+	list := &v1BATCH.JobList{}
+	ctx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+	return list, c.List(ctx, c.namespace, list)
+}
+
 func (c *client) getNodes(ctx context.Context) (*v1.NodeList, error) {
 	list := new(v1.NodeList)
 	ctx, cancel := context.WithTimeout(ctx, c.timeout)

diff --git a/plugins/inputs/kube_inventory/job.go b/plugins/inputs/kube_inventory/job.go
@@ -0,0 +1,46 @@
+package kube_inventory
+
+import (
+	"context"
+	"time"
+
+	v1 "github.com/ericchiang/k8s/apis/batch/v1"
+	"github.com/influxdata/telegraf"
+)
+
+func collectJobs(ctx context.Context, acc telegraf.Accumulator, ki *KubernetesInventory) {
+	list, err := ki.client.getJobs(ctx)
+	if err != nil {
+		acc.AddError(err)
+		return
+	}
+	for _, d := range list.Items {
+		if err = ki.gatherJob(*d, acc); err != nil {
+			acc.AddError(err)
+			return
+		}
+	}
+}
+
+func (ki *KubernetesInventory) gatherJob(d v1.Job, acc telegraf.Accumulator) error {
+	fields := map[string]interface{}{
+		"active":    d.Status.GetActive(),
+		"completed": time.Unix(d.Status.CompletionTime.GetSeconds(), int64(d.Status.CompletionTime.GetNanos())).UnixNano(),
+		"failed":    d.Status.GetFailed(),
+		"started":   time.Unix(d.Status.StartTime.GetSeconds(), int64(d.Status.StartTime.GetNanos())).UnixNano(),
+		"succeeded": d.Status.GetSucceeded(),
+	}
+	tags := map[string]string{
+		"job_name":  d.Metadata.GetName(),
+		"namespace": d.Metadata.GetNamespace(),
+	}
+	for key, val := range d.GetSpec().GetSelector().GetMatchLabels() {
+		if ki.selectorFilter.Match(key) {
+			tags["selector_"+key] = val
+		}
+	}
+
+	acc.AddFields(jobMeasurement, fields, tags)
+
+	return nil
+}