@cryptotaxi247 / netdata-1 / commits / 4f16097ec

improvement(go.d/k8s_state): collect cronjobs (#19793)

* collect cronjobs * completed

Ilya Mashchenko committed Mar 7, 2025 at 16:30 UTC 4f16097ec87fa308999ea465c15192e320691c7c
16 files changed +698 -24
src/go/plugin/go.d/collector/k8s_state/charts.go
+70 -2
@@ -61,6 +61,11 @@ const (
61 prioDeploymentAge
62 )
63
64 +const (
65 + prioCronJobJobsCountByStatus = 50700 + iota
66 + prioCronJobAge
67 +)
68 +
69 const (
70 labelKeyPrefix = "k8s_"
71 //labelKeyLabelPrefix = labelKeyPrefix + "label_"
@@ -78,6 +83,7 @@ const (
83 labelKeyContainerID = labelKeyPrefix + "container_id"
84 labelKeyQoSClass = labelKeyPrefix + "qos_class"
85 labelKeyDeploymentName = labelKeyPrefix + "deployment_name"
86 + labelKeyCronJobName = labelKeyPrefix + "cronjob_name"
87 )
88
89 var baseCharts = module.Charts{
@@ -135,6 +141,11 @@ var deploymentChartsTmpl = module.Charts{
141 deploymentAgeChartTmpl.Copy(),
142 }
143
144 +var cronJobChartsTmpl = module.Charts{
145 + cronJobJobsCountByStatusChartTmpl.Copy(),
146 + cronJobAgeChartTmpl.Copy(),
147 +}
148 +
149 var (
150 // CPU resource
151 nodeAllocatableCPURequestsUtilChartTmpl = module.Chart{
@@ -824,8 +835,65 @@ func (c *Collector) addDeploymentCharts(rs *deploymentState) {
835 }
836 }
837
827 -func (c *Collector) removeDeploymentCharts(rs *deploymentState) {
828 - prefix := fmt.Sprintf("deployment_%s", replaceDots(rs.id()))
838 +func (c *Collector) removeDeploymentCharts(st *deploymentState) {
839 + prefix := fmt.Sprintf("deployment_%s", replaceDots(st.id()))
840 + c.removeCharts(prefix)
841 +}
842 +
843 +var (
844 + cronJobJobsCountByStatusChartTmpl = module.Chart{
845 + IDSep: true,
846 + ID: "cronjob_%s.jobs_count_by_status",
847 + Title: "CronJob Jobs Count by Status",
848 + Units: "jobs",
849 + Fam: "cronjob jobs",
850 + Ctx: "k8s_state.cronjob_jobs_count_by_status",
851 + Priority: prioCronJobJobsCountByStatus,
852 + Type: module.Stacked,
853 + Dims: module.Dims{
854 + {ID: "cronjob_%s_complete_jobs", Name: "completed"},
855 + {ID: "cronjob_%s_failed_jobs", Name: "failed"},
856 + {ID: "cronjob_%s_running_jobs", Name: "running"},
857 + {ID: "cronjob_%s_suspended_jobs", Name: "suspended"},
858 + },
859 + }
860 + cronJobAgeChartTmpl = module.Chart{
861 + IDSep: true,
862 + ID: "cronjob_%s.age",
863 + Title: "CronJob Age",
864 + Units: "seconds",
865 + Fam: "cronjob age",
866 + Ctx: "k8s_state.cronjob_age",
867 + Priority: prioCronJobAge,
868 + Dims: module.Dims{
869 + {ID: "cronjob_%s_age", Name: "age"},
870 + },
871 + }
872 +)
873 +
874 +func (c *Collector) addCronJobCharts(st *cronJobState) {
875 + charts := cronJobChartsTmpl.Copy()
876 +
877 + for _, chart := range *charts {
878 + chart.ID = fmt.Sprintf(chart.ID, replaceDots(st.id()))
879 + chart.Labels = []module.Label{
880 + {Key: labelKeyClusterID, Value: c.kubeClusterID, Source: module.LabelSourceK8s},
881 + {Key: labelKeyClusterName, Value: c.kubeClusterName, Source: module.LabelSourceK8s},
882 + {Key: labelKeyCronJobName, Value: st.name, Source: module.LabelSourceK8s},
883 + {Key: labelKeyNamespace, Value: st.namespace, Source: module.LabelSourceK8s},
884 + }
885 + for _, d := range chart.Dims {
886 + d.ID = fmt.Sprintf(d.ID, st.id())
887 + }
888 + }
889 +
890 + if err := c.Charts().Add(*charts...); err != nil {
891 + c.Warning(err)
892 + }
893 +}
894 +
895 +func (c *Collector) removeCronJobCharts(st *cronJobState) {
896 + prefix := fmt.Sprintf("cronjob_%s", replaceDots(st.id()))
897 c.removeCharts(prefix)
898 }
899
src/go/plugin/go.d/collector/k8s_state/collect.go
+55
@@ -10,6 +10,7 @@ import (
10 "time"
11
12 appsv1 "k8s.io/api/apps/v1"
13 + batchv1 "k8s.io/api/batch/v1"
14 corev1 "k8s.io/api/core/v1"
15
16 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
@@ -110,6 +111,7 @@ func (c *Collector) collectKubeState(mx map[string]int64) {
111 c.collectPodsState(mx)
112 c.collectNodesState(mx)
113 c.collectDeploymentState(mx)
114 + c.collectCronJobState(mx)
115 }
116
117 func (c *Collector) collectPodsState(mx map[string]int64) {
@@ -353,6 +355,59 @@ func (c *Collector) collectDeploymentState(mx map[string]int64) {
355 })
356 }
357
358 +func (c *Collector) collectCronJobState(mx map[string]int64) {
359 + now := time.Now()
360 +
361 + maps.DeleteFunc(c.state.jobs, func(s string, st *jobState) bool {
362 + return st.deleted
363 + })
364 +
365 + maps.DeleteFunc(c.state.cronJobs, func(s string, st *cronJobState) bool {
366 + if st.deleted {
367 + c.removeCronJobCharts(st)
368 + return true
369 + }
370 + if st.new {
371 + c.addCronJobCharts(st)
372 + st.new = false
373 + }
374 +
375 + px := fmt.Sprintf("cronjob_%s_", st.id())
376 +
377 + mx[px+"age"] = int64(now.Sub(st.creationTime).Seconds())
378 +
379 + mx[px+"running_jobs"] = 0
380 + mx[px+"failed_jobs"] = 0
381 + mx[px+"complete_jobs"] = 0
382 + mx[px+"suspended_jobs"] = 0
383 +
384 + for _, job := range c.state.jobs {
385 + switch {
386 + case job.controller.kind != "CronJob", job.controller.uid != st.uid, job.startTime == nil:
387 + continue
388 + case job.active > 0:
389 + mx[px+"running_jobs"]++
390 + case len(job.conditions) > 0:
391 + for _, cond := range job.conditions {
392 + if cond.Status != corev1.ConditionTrue {
393 + continue
394 + }
395 + switch cond.Type {
396 + case batchv1.JobFailed:
397 + mx[px+"failed_jobs"]++
398 + case batchv1.JobComplete:
399 + mx[px+"complete_jobs"]++
400 + case batchv1.JobSuspended:
401 + mx[px+"suspended_jobs"]++
402 + }
403 + }
404 + }
405 + }
406 +
407 + return false
408 + })
409 +}
410 +
411 func condStatusToInt(cs corev1.ConditionStatus) int64 {
412 return metrix.Bool(cs == corev1.ConditionTrue)
413 }
src/go/plugin/go.d/collector/k8s_state/collector.go
+1 -1
@@ -31,7 +31,7 @@ func init() {
31
32 func New() *Collector {
33 return &Collector{
34 - initDelay: time.Second * 3,
34 + initDelay: time.Second * 10,
35 newKubeClient: newKubeClient,
36 charts: baseCharts.Copy(),
37 once: &sync.Once{},
src/go/plugin/go.d/collector/k8s_state/collector_test.go
+118
@@ -11,6 +11,9 @@ import (
11 "testing"
12 "time"
13
14 + batchv1 "k8s.io/api/batch/v1"
15 + "k8s.io/apimachinery/pkg/types"
16 +
17 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
18
19 "github.com/stretchr/testify/assert"
@@ -509,6 +512,53 @@ func TestCollector_Collect(t *testing.T) {
512 }
513 },
514 },
515 + "CronJobs": {
516 + create: func(t *testing.T) testCase {
517 + cj := prepareCronJob("cronjob01")
518 + jobNotStarted := prepareCronJobNotStartedJob("job-not-started", cj)
519 + jobComplete := prepareCronJobCompleteJob("job-complete", cj)
520 + jobFailed := prepareCronJobFailedJob("job-failed", cj)
521 + jobRunning := prepareCronJobRunningJob("job-running", cj)
522 + jobSuspended := prepareCronJobSuspendedJob("job-suspended", cj)
523 +
524 + client := fake.NewClientset(
525 + cj,
526 + jobNotStarted,
527 + jobComplete,
528 + jobFailed,
529 + jobRunning,
530 + jobSuspended,
531 + )
532 +
533 + step1 := func(t *testing.T, collr *Collector) {
534 + mx := collr.Collect(context.Background())
535 + expected := map[string]int64{
536 + "cronjob_default_cronjob01_age": 10,
537 + "cronjob_default_cronjob01_complete_jobs": 1,
538 + "cronjob_default_cronjob01_failed_jobs": 1,
539 + "cronjob_default_cronjob01_running_jobs": 1,
540 + "cronjob_default_cronjob01_suspended_jobs": 1,
541 + "discovery_node_discoverer_state": 1,
542 + "discovery_pod_discoverer_state": 1,
543 + }
544 +
545 + copyAge(expected, mx)
546 +
547 + assert.Equal(t, expected, mx)
548 + assert.Equal(t,
549 + len(cronJobChartsTmpl)+
550 + len(baseCharts),
551 + len(*collr.Charts()),
552 + )
553 + module.TestMetricsHasAllChartsDims(t, collr.Charts(), mx)
554 + }
555 +
556 + return testCase{
557 + client: client,
558 + steps: []testCaseStep{step1},
559 + }
560 + },
561 + },
562 "delete a Pod in runtime": {
563 create: func(t *testing.T) testCase {
564 ctx := context.Background()
@@ -1010,6 +1060,74 @@ func newDeployment(name string) *appsv1.Deployment {
1060 }
1061 }
1062
1063 +func prepareCronJob(name string) *batchv1.CronJob {
1064 + return &batchv1.CronJob{
1065 + ObjectMeta: metav1.ObjectMeta{
1066 + Name: name,
1067 + Namespace: corev1.NamespaceDefault,
1068 + UID: types.UID(name),
1069 + CreationTimestamp: metav1.Time{Time: time.Now()},
1070 + },
1071 + }
1072 +}
1073 +
1074 +func prepareCronJobNotStartedJob(name string, cj *batchv1.CronJob) *batchv1.Job {
1075 + return prepareCronJobJob(name, cj)
1076 +}
1077 +
1078 +func prepareCronJobRunningJob(name string, cj *batchv1.CronJob) *batchv1.Job {
1079 + job := prepareCronJobJob(name, cj)
1080 + job.Status.StartTime = &metav1.Time{Time: time.Now()}
1081 + job.Status.Active = 1
1082 + return job
1083 +}
1084 +
1085 +func prepareCronJobCompleteJob(name string, cj *batchv1.CronJob) *batchv1.Job {
1086 + job := prepareCronJobJob(name, cj)
1087 + job.Status.StartTime = &metav1.Time{Time: time.Now()}
1088 + job.Status.Conditions = []batchv1.JobCondition{
1089 + {Type: batchv1.JobComplete, Status: corev1.ConditionTrue},
1090 + }
1091 + return job
1092 +}
1093 +
1094 +func prepareCronJobFailedJob(name string, cj *batchv1.CronJob) *batchv1.Job {
1095 + job := prepareCronJobJob(name, cj)
1096 + job.Status.StartTime = &metav1.Time{Time: time.Now()}
1097 + job.Status.Conditions = []batchv1.JobCondition{
1098 + {Type: batchv1.JobFailed, Status: corev1.ConditionTrue},
1099 + }
1100 + return job
1101 +}
1102 +
1103 +func prepareCronJobSuspendedJob(name string, cj *batchv1.CronJob) *batchv1.Job {
1104 + job := prepareCronJobJob(name, cj)
1105 + job.Status.StartTime = &metav1.Time{Time: time.Now()}
1106 + job.Status.Conditions = []batchv1.JobCondition{
1107 + {Type: batchv1.JobSuspended, Status: corev1.ConditionTrue},
1108 + }
1109 + return job
1110 +}
1111 +
1112 +func prepareCronJobJob(name string, cj *batchv1.CronJob) *batchv1.Job {
1113 + return &batchv1.Job{
1114 + ObjectMeta: metav1.ObjectMeta{
1115 + Name: name,
1116 + Namespace: corev1.NamespaceDefault,
1117 + UID: types.UID(name),
1118 + CreationTimestamp: metav1.Time{Time: time.Now()},
1119 + OwnerReferences: []metav1.OwnerReference{
1120 + {
1121 + Controller: ptr(true),
1122 + Kind: "CronJob",
1123 + UID: cj.UID,
1124 + Name: cj.Name,
1125 + },
1126 + },
1127 + },
1128 + }
1129 +}
1130 +
1131 type brokenInfoKubeClient struct {
1132 kubernetes.Interface
1133 }
src/go/plugin/go.d/collector/k8s_state/discover_cronjob.go new
+106
@@ -0,0 +1,106 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package k8s_state
4 +
5 +import (
6 + "context"
7 +
8 + "k8s.io/client-go/tools/cache"
9 + "k8s.io/client-go/util/workqueue"
10 +
11 + "github.com/netdata/netdata/go/plugins/logger"
12 +)
13 +
14 +func newCronJobDiscoverer(si cache.SharedInformer, l *logger.Logger) *cronJobDiscoverer {
15 + if si == nil {
16 + panic("nil cronjob shared informer")
17 + }
18 +
19 + queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[string]{Name: "cronjob"})
20 +
21 + _, _ = si.AddEventHandler(cache.ResourceEventHandlerFuncs{
22 + AddFunc: func(obj any) { enqueue(queue, obj) },
23 + UpdateFunc: func(_, obj any) { enqueue(queue, obj) },
24 + DeleteFunc: func(obj any) { enqueue(queue, obj) },
25 + })
26 +
27 + return &cronJobDiscoverer{
28 + Logger: l,
29 + informer: si,
30 + queue: queue,
31 + readyCh: make(chan struct{}),
32 + stopCh: make(chan struct{}),
33 + }
34 +}
35 +
36 +type cronjobResource struct {
37 + src string
38 + val any
39 +}
40 +
41 +func (r cronjobResource) source() string { return r.src }
42 +func (r cronjobResource) kind() kubeResourceKind { return kubeResourceCronJob }
43 +func (r cronjobResource) value() any { return r.val }
44 +
45 +type cronJobDiscoverer struct {
46 + *logger.Logger
47 + informer cache.SharedInformer
48 + queue *workqueue.Typed[string]
49 + readyCh chan struct{}
50 + stopCh chan struct{}
51 +}
52 +
53 +func (d *cronJobDiscoverer) run(ctx context.Context, in chan<- resource) {
54 + d.Info("cronjob_discoverer is started")
55 + defer func() { close(d.stopCh); d.Info("cronjob_discoverer is stopped") }()
56 +
57 + defer d.queue.ShutDown()
58 +
59 + go d.informer.Run(ctx.Done())
60 +
61 + if !cache.WaitForCacheSync(ctx.Done(), d.informer.HasSynced) {
62 + return
63 + }
64 +
65 + go d.runDiscover(ctx, in)
66 +
67 + close(d.readyCh)
68 +
69 + <-ctx.Done()
70 +}
71 +
72 +func (d *cronJobDiscoverer) runDiscover(ctx context.Context, in chan<- resource) {
73 + for {
74 + key, shutdown := d.queue.Get()
75 + if shutdown {
76 + return
77 + }
78 +
79 + func() {
80 + defer d.queue.Done(key)
81 +
82 + ns, name, err := cache.SplitMetaNamespaceKey(key)
83 + if err != nil {
84 + return
85 + }
86 +
87 + item, exists, err := d.informer.GetStore().GetByKey(key)
88 + if err != nil {
89 + return
90 + }
91 +
92 + r := &cronjobResource{src: cronjobSource(ns, name)}
93 + if exists {
94 + r.val = item
95 + }
96 + send(ctx, in, r)
97 + }()
98 + }
99 +}
100 +
101 +func cronjobSource(namespace, name string) string {
102 + return "k8s/cj/" + namespace + "/" + name
103 +}
104 +
105 +func (d *cronJobDiscoverer) ready() bool { return isChanClosed(d.readyCh) }
106 +func (d *cronJobDiscoverer) stopped() bool { return isChanClosed(d.stopCh) }
src/go/plugin/go.d/collector/k8s_state/discover_deployment.go
+4 -5
@@ -16,7 +16,7 @@ func newDeploymentDiscoverer(si cache.SharedInformer, l *logger.Logger) *deploym
16 panic("nil deployment& shared informer")
17 }
18
19 - queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[any]{Name: "replicaset"})
19 + queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[string]{Name: "replicaset"})
20
21 _, _ = si.AddEventHandler(cache.ResourceEventHandlerFuncs{
22 AddFunc: func(obj any) { enqueue(queue, obj) },
@@ -45,7 +45,7 @@ func (r deployResource) value() any { return r.val }
45 type deploymentDiscoverer struct {
46 *logger.Logger
47 informer cache.SharedInformer
48 - queue *workqueue.Typed[any]
48 + queue *workqueue.Typed[string]
49 readyCh chan struct{}
50 stopCh chan struct{}
51 }
@@ -74,15 +74,14 @@ func (d *deploymentDiscoverer) stopped() bool { return isChanClosed(d.stopCh) }
74
75 func (d *deploymentDiscoverer) runDiscover(ctx context.Context, in chan<- resource) {
76 for {
77 - item, shutdown := d.queue.Get()
77 + key, shutdown := d.queue.Get()
78 if shutdown {
79 return
80 }
81
82 func() {
83 - defer d.queue.Done(item)
83 + defer d.queue.Done(key)
84
85 - key := item.(string)
85 ns, name, err := cache.SplitMetaNamespaceKey(key)
86 if err != nil {
87 return
src/go/plugin/go.d/collector/k8s_state/discover_job.go new
+106
@@ -0,0 +1,106 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package k8s_state
4 +
5 +import (
6 + "context"
7 +
8 + "k8s.io/client-go/tools/cache"
9 + "k8s.io/client-go/util/workqueue"
10 +
11 + "github.com/netdata/netdata/go/plugins/logger"
12 +)
13 +
14 +func newJobDiscoverer(si cache.SharedInformer, l *logger.Logger) *jobDiscoverer {
15 + if si == nil {
16 + panic("nil job shared informer")
17 + }
18 +
19 + queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[string]{Name: "job"})
20 +
21 + _, _ = si.AddEventHandler(cache.ResourceEventHandlerFuncs{
22 + AddFunc: func(obj any) { enqueue(queue, obj) },
23 + UpdateFunc: func(_, obj any) { enqueue(queue, obj) },
24 + DeleteFunc: func(obj any) { enqueue(queue, obj) },
25 + })
26 +
27 + return &jobDiscoverer{
28 + Logger: l,
29 + informer: si,
30 + queue: queue,
31 + readyCh: make(chan struct{}),
32 + stopCh: make(chan struct{}),
33 + }
34 +}
35 +
36 +type jobResource struct {
37 + src string
38 + val any
39 +}
40 +
41 +func (r jobResource) source() string { return r.src }
42 +func (r jobResource) kind() kubeResourceKind { return kubeResourceJob }
43 +func (r jobResource) value() any { return r.val }
44 +
45 +type jobDiscoverer struct {
46 + *logger.Logger
47 + informer cache.SharedInformer
48 + queue *workqueue.Typed[string]
49 + readyCh chan struct{}
50 + stopCh chan struct{}
51 +}
52 +
53 +func (d *jobDiscoverer) run(ctx context.Context, in chan<- resource) {
54 + d.Info("job_discoverer is started")
55 + defer func() { close(d.stopCh); d.Info("job_discoverer is stopped") }()
56 +
57 + defer d.queue.ShutDown()
58 +
59 + go d.informer.Run(ctx.Done())
60 +
61 + if !cache.WaitForCacheSync(ctx.Done(), d.informer.HasSynced) {
62 + return
63 + }
64 +
65 + go d.runDiscover(ctx, in)
66 +
67 + close(d.readyCh)
68 +
69 + <-ctx.Done()
70 +}
71 +
72 +func (d *jobDiscoverer) runDiscover(ctx context.Context, in chan<- resource) {
73 + for {
74 + key, shutdown := d.queue.Get()
75 + if shutdown {
76 + return
77 + }
78 +
79 + func() {
80 + defer d.queue.Done(key)
81 +
82 + ns, name, err := cache.SplitMetaNamespaceKey(key)
83 + if err != nil {
84 + return
85 + }
86 +
87 + item, exists, err := d.informer.GetStore().GetByKey(key)
88 + if err != nil {
89 + return
90 + }
91 +
92 + r := &jobResource{src: jobSource(ns, name)}
93 + if exists {
94 + r.val = item
95 + }
96 + send(ctx, in, r)
97 + }()
98 + }
99 +}
100 +
101 +func jobSource(namespace, name string) string {
102 + return "k8s/job/" + namespace + "/" + name
103 +}
104 +
105 +func (d *jobDiscoverer) ready() bool { return isChanClosed(d.readyCh) }
106 +func (d *jobDiscoverer) stopped() bool { return isChanClosed(d.stopCh) }
src/go/plugin/go.d/collector/k8s_state/discover_kubernetes.go
+16 -1
@@ -11,6 +11,7 @@ import (
11 "github.com/netdata/netdata/go/plugins/logger"
12
13 appsv1 "k8s.io/api/apps/v1"
14 + batchv1 "k8s.io/api/batch/v1"
15 corev1 "k8s.io/api/core/v1"
16 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
17 "k8s.io/apimachinery/pkg/runtime"
@@ -139,14 +140,28 @@ func (d *kubeDiscovery) setupDiscoverers(ctx context.Context) []discoverer {
140 WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return deploy.Watch(ctx, options) },
141 }
142
143 + cj := d.client.BatchV1().CronJobs(corev1.NamespaceAll)
144 + cjWatcher := &cache.ListWatch{
145 + ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return cj.List(ctx, options) },
146 + WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return cj.Watch(ctx, options) },
147 + }
148 +
149 + jobs := d.client.BatchV1().Jobs(corev1.NamespaceAll)
150 + jobsWatcher := &cache.ListWatch{
151 + ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return jobs.List(ctx, options) },
152 + WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return jobs.Watch(ctx, options) },
153 + }
154 +
155 return []discoverer{
156 newNodeDiscoverer(cache.NewSharedInformer(nodeWatcher, &corev1.Node{}, resyncPeriod), d.Logger),
157 newPodDiscoverer(cache.NewSharedInformer(podWatcher, &corev1.Pod{}, resyncPeriod), d.Logger),
158 newDeploymentDiscoverer(cache.NewSharedInformer(deployWatcher, &appsv1.Deployment{}, resyncPeriod), d.Logger),
159 + newCronJobDiscoverer(cache.NewSharedInformer(cjWatcher, &batchv1.CronJob{}, resyncPeriod), d.Logger),
160 + newJobDiscoverer(cache.NewSharedInformer(jobsWatcher, &batchv1.Job{}, resyncPeriod), d.Logger),
161 }
162 }
163
149 -func enqueue(queue *workqueue.Typed[any], obj any) {
164 +func enqueue(queue *workqueue.Typed[string], obj any) {
165 key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
166 if err != nil {
167 return
src/go/plugin/go.d/collector/k8s_state/discover_node.go
+4 -5
@@ -16,7 +16,7 @@ func newNodeDiscoverer(si cache.SharedInformer, l *logger.Logger) *nodeDiscovere
16 panic("nil node shared informer")
17 }
18
19 - queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[any]{Name: "node"})
19 + queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[string]{Name: "node"})
20
21 _, _ = si.AddEventHandler(cache.ResourceEventHandlerFuncs{
22 AddFunc: func(obj any) { enqueue(queue, obj) },
@@ -45,7 +45,7 @@ func (r nodeResource) value() any { return r.val }
45 type nodeDiscoverer struct {
46 *logger.Logger
47 informer cache.SharedInformer
48 - queue *workqueue.Typed[any]
48 + queue *workqueue.Typed[string]
49 readyCh chan struct{}
50 stopCh chan struct{}
51 }
@@ -73,15 +73,14 @@ func (d *nodeDiscoverer) stopped() bool { return isChanClosed(d.stopCh) }
73
74 func (d *nodeDiscoverer) runDiscover(ctx context.Context, in chan<- resource) {
75 for {
76 - item, shutdown := d.queue.Get()
76 + key, shutdown := d.queue.Get()
77 if shutdown {
78 return
79 }
80
81 func() {
82 - defer d.queue.Done(item)
82 + defer d.queue.Done(key)
83
84 - key := item.(string)
84 _, name, err := cache.SplitMetaNamespaceKey(key)
85 if err != nil {
86 return
src/go/plugin/go.d/collector/k8s_state/discover_pod.go
+4 -5
@@ -19,7 +19,7 @@ func newPodDiscoverer(si cache.SharedInformer, l *logger.Logger) *podDiscoverer
19 panic("nil pod shared informer")
20 }
21
22 - queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[any]{Name: "pod"})
22 + queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[string]{Name: "pod"})
23
24 _, _ = si.AddEventHandler(cache.ResourceEventHandlerFuncs{
25 AddFunc: func(obj any) { enqueue(queue, obj) },
@@ -48,7 +48,7 @@ func (r podResource) value() any { return r.val }
48 type podDiscoverer struct {
49 *logger.Logger
50 informer cache.SharedInformer
51 - queue *workqueue.Typed[any]
51 + queue *workqueue.Typed[string]
52 readyCh chan struct{}
53 stopCh chan struct{}
54 }
@@ -76,15 +76,14 @@ func (d *podDiscoverer) stopped() bool { return isChanClosed(d.stopCh) }
76
77 func (d *podDiscoverer) runDiscover(ctx context.Context, in chan<- resource) {
78 for {
79 - item, shutdown := d.queue.Get()
79 + key, shutdown := d.queue.Get()
80 if shutdown {
81 return
82 }
83
84 func() {
85 - defer d.queue.Done(item)
85 + defer d.queue.Done(key)
86
87 - key := item.(string)
87 ns, name, err := cache.SplitMetaNamespaceKey(key)
88 if err != nil {
89 return
src/go/plugin/go.d/collector/k8s_state/metadata.yaml
+27
@@ -247,6 +247,33 @@ modules:
247 chart_type: line
248 dimensions:
249 - name: age
250 + - name: cronjob
251 + description: These metrics refer to CronJobs.
252 + labels:
253 + - name: k8s_cluster_id
254 + description: Cluster ID. This is equal to the kube-system namespace UID.
255 + - name: k8s_cluster_name
256 + description: Cluster name. Cluster name discovery only works in GKE.
257 + - name: k8s_cronjob_name
258 + description: CronJob name.
259 + - name: k8s_namespace
260 + description: Namespace.
261 + metrics:
262 + - name: k8s_state.cronjob_jobs_count_by_status
263 + description: CronJob Jobs Count by Status
264 + unit: 'jobs'
265 + chart_type: stacked
266 + dimensions:
267 + - name: completed
268 + - name: failed
269 + - name: running
270 + - name: suspended
271 + - name: k8s_state.cronjob_age
272 + description: CronJob Age
273 + unit: 'seconds'
274 + chart_type: line
275 + dimensions:
276 + - name: age
277 - name: pod
278 description: These metrics refer to the Pod.
279 labels:
src/go/plugin/go.d/collector/k8s_state/resource.go
+25
@@ -6,6 +6,7 @@ import (
6 "fmt"
7
8 appsv1 "k8s.io/api/apps/v1"
9 + batchv1 "k8s.io/api/batch/v1"
10 corev1 "k8s.io/api/core/v1"
11 )
12
@@ -21,6 +22,8 @@ const (
22 kubeResourceNode kubeResourceKind = iota + 1
23 kubeResourcePod
24 kubeResourceDeployment
25 + kubeResourceCronJob
26 + kubeResourceJob
27 )
28
29 func toNode(i any) (*corev1.Node, error) {
@@ -55,3 +58,25 @@ func toDeployment(i any) (*appsv1.Deployment, error) {
58 return nil, fmt.Errorf("unexpected type: %T (expected %T or %T)", v, &appsv1.Deployment{}, resource(nil))
59 }
60 }
61 +
62 +func toCronJob(i any) (*batchv1.CronJob, error) {
63 + switch v := i.(type) {
64 + case *batchv1.CronJob:
65 + return v, nil
66 + case resource:
67 + return toCronJob(v.value())
68 + default:
69 + return nil, fmt.Errorf("unexpected type: %T (expected %T or %T)", v, &batchv1.CronJob{}, resource(nil))
70 + }
71 +}
72 +
73 +func toJob(i any) (*batchv1.Job, error) {
74 + switch v := i.(type) {
75 + case *batchv1.Job:
76 + return v, nil
77 + case resource:
78 + return toJob(v.value())
79 + default:
80 + return nil, fmt.Errorf("unexpected type: %T (expected %T or %T)", v, &batchv1.Job{}, resource(nil))
81 + }
82 +}
src/go/plugin/go.d/collector/k8s_state/state.go
+54
@@ -7,6 +7,7 @@ import (
7 "time"
8
9 appsv1 "k8s.io/api/apps/v1"
10 + batchv1 "k8s.io/api/batch/v1"
11 corev1 "k8s.io/api/core/v1"
12 )
13
@@ -16,6 +17,8 @@ func newKubeState() *kubeState {
17 nodes: make(map[string]*nodeState),
18 pods: make(map[string]*podState),
19 deployments: make(map[string]*deploymentState),
20 + cronJobs: make(map[string]*cronJobState),
21 + jobs: make(map[string]*jobState),
22 }
23 }
24
@@ -47,11 +50,25 @@ func newDeploymentState() *deploymentState {
50 }
51 }
52
53 +func newCronJobState() *cronJobState {
54 + return &cronJobState{
55 + new: true,
56 + }
57 +}
58 +
59 +func newJobState() *jobState {
60 + return &jobState{
61 + new: true,
62 + }
63 +}
64 +
65 type kubeState struct {
66 *sync.Mutex
67 nodes map[string]*nodeState
68 pods map[string]*podState
69 deployments map[string]*deploymentState
70 + cronJobs map[string]*cronJobState
71 + jobs map[string]*jobState
72 }
73
74 type (
@@ -176,3 +193,40 @@ type deploymentState struct {
193 }
194
195 func (ds deploymentState) id() string { return ds.namespace + "_" + ds.name }
196 +
197 +type cronJobState struct {
198 + new bool
199 + deleted bool
200 +
201 + uid string
202 + name string
203 + namespace string
204 + creationTime time.Time
205 +}
206 +
207 +func (cs cronJobState) id() string { return cs.namespace + "_" + cs.name }
208 +
209 +type jobState struct {
210 + new bool
211 + deleted bool
212 +
213 + uid string
214 + name string
215 + namespace string
216 + creationTime time.Time
217 +
218 + controller struct {
219 + kind string
220 + name string
221 + uid string
222 + }
223 +
224 + conditions []batchv1.JobCondition
225 +
226 + startTime *time.Time
227 + active int32
228 + uncountedTerminatedPods struct {
229 + succeeded int
230 + failed int
231 + }
232 +}
src/go/plugin/go.d/collector/k8s_state/update_cronjob_state.go new
+29
@@ -0,0 +1,29 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package k8s_state
4 +
5 +func (c *Collector) updateCronJobState(r resource) {
6 + if r.value() == nil {
7 + if rs, ok := c.state.cronJobs[r.source()]; ok {
8 + rs.deleted = true
9 + }
10 + return
11 + }
12 +
13 + cj, err := toCronJob(r)
14 + if err != nil {
15 + c.Warning(err)
16 + return
17 + }
18 +
19 + _, ok := c.state.cronJobs[r.source()]
20 + if !ok {
21 + st := newCronJobState()
22 + c.state.cronJobs[r.source()] = st
23 +
24 + st.uid = string(cj.UID)
25 + st.name = cj.Name
26 + st.namespace = cj.Namespace
27 + st.creationTime = cj.CreationTimestamp.Time
28 + }
29 +}
src/go/plugin/go.d/collector/k8s_state/update_job_state.go new
+66
@@ -0,0 +1,66 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package k8s_state
4 +
5 +import (
6 + "slices"
7 +
8 + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
9 +)
10 +
11 +func (c *Collector) updateJobState(r resource) {
12 + if r.value() == nil {
13 + if rs, ok := c.state.jobs[r.source()]; ok {
14 + rs.deleted = true
15 + }
16 + return
17 + }
18 +
19 + job, err := toJob(r)
20 + if err != nil {
21 + c.Warning(err)
22 + return
23 + }
24 +
25 + if !slices.ContainsFunc(job.OwnerReferences, func(ref metav1.OwnerReference) bool {
26 + return ref.Controller != nil && *ref.Controller && ref.Kind == "CronJob"
27 + }) {
28 + return
29 + }
30 +
31 + st, ok := c.state.jobs[r.source()]
32 + if !ok {
33 + st = newJobState()
34 + c.state.jobs[r.source()] = st
35 +
36 + st.uid = string(job.UID)
37 + st.name = job.Name
38 + st.namespace = job.Namespace
39 + st.creationTime = job.CreationTimestamp.Time
40 +
41 + for _, ref := range job.OwnerReferences {
42 + if ref.Controller != nil && *ref.Controller {
43 + st.controller.kind = ref.Kind
44 + st.controller.name = ref.Name
45 + st.controller.uid = string(ref.UID)
46 + }
47 + }
48 + }
49 +
50 + if job.Status.StartTime != nil {
51 + st.startTime = ptr(job.Status.StartTime.Time)
52 + }
53 +
54 + st.active = job.Status.Active
55 +
56 + st.conditions = job.Status.Conditions
57 +
58 + st.uncountedTerminatedPods.succeeded = 0
59 + st.uncountedTerminatedPods.failed = 0
60 +
61 + if v := job.Status.UncountedTerminatedPods; v != nil {
62 + st.uncountedTerminatedPods.succeeded = len(v.Succeeded)
63 + st.uncountedTerminatedPods.failed = len(v.Failed)
64 + }
65 +
66 +}
src/go/plugin/go.d/collector/k8s_state/update_state.go
+13 -5
@@ -7,15 +7,19 @@ func (c *Collector) runUpdateState(in <-chan resource) {
7 select {
8 case <-c.ctx.Done():
9 return
10 - case r := <-in:
10 + case res := <-in:
11 c.state.Lock()
12 - switch r.kind() {
12 + switch res.kind() {
13 case kubeResourceNode:
14 - c.updateNodeState(r)
14 + c.updateNodeState(res)
15 case kubeResourcePod:
16 - c.updatePodState(r)
16 + c.updatePodState(res)
17 case kubeResourceDeployment:
18 - c.updateDeploymentState(r)
18 + c.updateDeploymentState(res)
19 + case kubeResourceCronJob:
20 + c.updateCronJobState(res)
21 + case kubeResourceJob:
22 + c.updateJobState(res)
23 }
24 c.state.Unlock()
25 }
@@ -27,3 +31,7 @@ func copyLabels(dst, src map[string]string) {
31 dst[k] = v
32 }
33 }
34 +
35 +func ptr[T any](v T) *T {
36 + return &v
37 +}