improvement(go.d/k8sstate): collect deployments (#19657)
* improvement(go.d/k8sstate): collect replicasets * replicasets -> deployments
Ilya Mashchenko committed
Feb 17, 2025 at 16:23 UTC
1e31593eaa7e1ff46ff6b946071c3475a8049ecd
10 files changed
+352
-19
src/go/plugin/go.d/collector/k8s_state/charts.go
+77
-12
@@ -55,6 +55,11 @@ const (
55
prioPodContainerTerminatedStateReason
56
)
57
58
+const (
59
+ prioDeploymentReplicas = 50500 + iota
60
+ prioDeploymentAge
61
+)
62
+
63
const (
64
labelKeyPrefix = "k8s_"
65
//labelKeyLabelPrefix = labelKeyPrefix + "label_"
@@ -71,6 +76,7 @@ const (
76
labelKeyContainerName = labelKeyPrefix + "container_name"
77
labelKeyContainerID = labelKeyPrefix + "container_id"
78
labelKeyQoSClass = labelKeyPrefix + "qos_class"
79
+ labelKeyDeploymentName = labelKeyPrefix + "deployment_name"
80
)
81
82
var baseCharts = module.Charts{
@@ -122,6 +128,11 @@ var containerChartsTmpl = module.Charts{
128
containersStateTerminatedChartTmpl.Copy(),
129
}
130
131
+var deploymentChartsTmpl = module.Charts{
132
+ deploymentReplicasChartTmpl.Copy(),
133
+ deploymentAgeChartTmpl.Copy(),
134
+}
135
+
136
var (
137
// CPU resource
138
nodeAllocatableCPURequestsUtilChartTmpl = module.Chart{
@@ -429,12 +440,7 @@ func (c *Collector) addNodeCharts(ns *nodeState) {
440
441
func (c *Collector) removeNodeCharts(ns *nodeState) {
442
prefix := fmt.Sprintf("node_%s", replaceDots(ns.id()))
432
- for _, c := range *c.Charts() {
433
- if strings.HasPrefix(c.ID, prefix) {
434
- c.MarkRemove()
435
- c.MarkNotCreated()
436
- }
437
- }
443
+ c.removeCharts(prefix)
444
}
445
446
var (
@@ -645,12 +651,7 @@ func updateNodeLabel(c *module.Chart, nodeName string) {
651
652
func (c *Collector) removePodCharts(ps *podState) {
653
prefix := fmt.Sprintf("pod_%s", replaceDots(ps.id()))
648
- for _, c := range *c.Charts() {
649
- if strings.HasPrefix(c.ID, prefix) {
650
- c.MarkRemove()
651
- c.MarkNotCreated()
652
- }
653
- }
654
+ c.removeCharts(prefix)
655
}
656
657
var (
@@ -757,6 +758,70 @@ func (c *Collector) addContainerCharts(ps *podState, cs *containerState) {
758
}
759
}
760
761
+var (
762
+ deploymentReplicasChartTmpl = module.Chart{
763
+ IDSep: true,
764
+ ID: "deployment_%s.replicas",
765
+ Title: "Deployment Replicas",
766
+ Units: "replicas",
767
+ Fam: "deployment replicas",
768
+ Ctx: "k8s_state.deployment_replicas",
769
+ Priority: prioDeploymentReplicas,
770
+ Dims: module.Dims{
771
+ {ID: "deploy_%s_desired_replicas", Name: "desired"},
772
+ {ID: "deploy_%s_current_replicas", Name: "current"},
773
+ {ID: "deploy_%s_ready_replicas", Name: "ready"},
774
+ },
775
+ }
776
+ deploymentAgeChartTmpl = module.Chart{
777
+ IDSep: true,
778
+ ID: "deployment_%s.age",
779
+ Title: "Deployment Age",
780
+ Units: "seconds",
781
+ Fam: "deployment age",
782
+ Ctx: "k8s_state.deployment_age",
783
+ Priority: prioDeploymentAge,
784
+ Dims: module.Dims{
785
+ {ID: "deploy_%s_age", Name: "age"},
786
+ },
787
+ }
788
+)
789
+
790
+func (c *Collector) addDeploymentCharts(rs *deploymentState) {
791
+ charts := deploymentChartsTmpl.Copy()
792
+
793
+ for _, chart := range *charts {
794
+ chart.ID = fmt.Sprintf(chart.ID, replaceDots(rs.id()))
795
+ chart.Labels = []module.Label{
796
+ {Key: labelKeyClusterID, Value: c.kubeClusterID, Source: module.LabelSourceK8s},
797
+ {Key: labelKeyClusterName, Value: c.kubeClusterName, Source: module.LabelSourceK8s},
798
+ {Key: labelKeyDeploymentName, Value: rs.name, Source: module.LabelSourceK8s},
799
+ {Key: labelKeyNamespace, Value: rs.namespace, Source: module.LabelSourceK8s},
800
+ }
801
+ for _, d := range chart.Dims {
802
+ d.ID = fmt.Sprintf(d.ID, rs.id())
803
+ }
804
+ }
805
+
806
+ if err := c.Charts().Add(*charts...); err != nil {
807
+ c.Warning(err)
808
+ }
809
+}
810
+
811
+func (c *Collector) removeDeploymentCharts(rs *deploymentState) {
812
+ prefix := fmt.Sprintf("deployment_%s", replaceDots(rs.id()))
813
+ c.removeCharts(prefix)
814
+}
815
+
816
+func (c *Collector) removeCharts(prefix string) {
817
+ for _, c := range *c.Charts() {
818
+ if strings.HasPrefix(c.ID, prefix) {
819
+ c.MarkRemove()
820
+ c.MarkNotCreated()
821
+ }
822
+ }
823
+}
824
+
825
var discoveryStatusChart = module.Chart{
826
ID: "discovery_discoverers_state",
827
Title: "Running discoverers state",
src/go/plugin/go.d/collector/k8s_state/collect.go
+27
@@ -5,6 +5,7 @@ package k8s_state
5
import (
6
"errors"
7
"fmt"
8
+ "maps"
9
"slices"
10
"time"
11
@@ -107,6 +108,7 @@ func (c *Collector) collectKubeState(mx map[string]int64) {
108
}
109
c.collectPodsState(mx)
110
c.collectNodesState(mx)
111
+ c.collectDeploymentState(mx)
112
}
113
114
func (c *Collector) collectPodsState(mx map[string]int64) {
@@ -308,6 +310,31 @@ func (c *Collector) collectNodesState(mx map[string]int64) {
310
}
311
}
312
313
+func (c *Collector) collectDeploymentState(mx map[string]int64) {
314
+ now := time.Now()
315
+
316
+ maps.DeleteFunc(c.state.deployments, func(s string, ds *deploymentState) bool {
317
+ if ds.deleted {
318
+ c.removeDeploymentCharts(ds)
319
+ return true
320
+ }
321
+
322
+ if ds.new {
323
+ ds.new = false
324
+ c.addDeploymentCharts(ds)
325
+ }
326
+
327
+ px := fmt.Sprintf("deploy_%s_", ds.id())
328
+
329
+ mx[px+"age"] = int64(now.Sub(ds.creationTime).Seconds())
330
+ mx[px+"desired_replicas"] = ds.replicas
331
+ mx[px+"current_replicas"] = ds.availableReplicas
332
+ mx[px+"ready_replicas"] = ds.readyReplicas
333
+
334
+ return false
335
+ })
336
+}
337
+
338
func condStatusToInt(cs corev1.ConditionStatus) int64 {
339
switch cs {
340
case corev1.ConditionFalse:
src/go/plugin/go.d/collector/k8s_state/collector_test.go
+28
-2
@@ -15,6 +15,7 @@ import (
15
16
"github.com/stretchr/testify/assert"
17
"github.com/stretchr/testify/require"
18
+ appsv1 "k8s.io/api/apps/v1"
19
corev1 "k8s.io/api/core/v1"
20
apiresource "k8s.io/apimachinery/pkg/api/resource"
21
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -356,13 +357,15 @@ func TestCollector_Collect(t *testing.T) {
357
}
358
},
359
},
359
- "Nodes and Pods": {
360
+ "Nodes and Pods and Deployment": {
361
create: func(t *testing.T) testCase {
362
node := newNode("node01")
363
pod := newPod(node.Name, "pod01")
364
+ deploy := newDeployment("replicaset01")
365
client := fake.NewClientset(
366
node,
367
pod,
368
+ deploy,
369
)
370
371
step1 := func(t *testing.T, collr *Collector) {
@@ -477,13 +480,21 @@ func TestCollector_Collect(t *testing.T) {
480
"pod_default_pod01_status_reason_Other": 0,
481
"pod_default_pod01_status_reason_Shutdown": 0,
482
"pod_default_pod01_status_reason_UnexpectedAdmissionError": 0,
483
+ "deploy_default_replicaset01_age": 3,
484
+ "deploy_default_replicaset01_current_replicas": 1,
485
+ "deploy_default_replicaset01_desired_replicas": 2,
486
+ "deploy_default_replicaset01_ready_replicas": 3,
487
}
488
489
copyAge(expected, mx)
490
491
assert.Equal(t, expected, mx)
492
assert.Equal(t,
486
- len(nodeChartsTmpl)+len(podChartsTmpl)+len(containerChartsTmpl)*len(pod.Spec.Containers)+len(baseCharts),
493
+ len(nodeChartsTmpl)+
494
+ len(podChartsTmpl)+
495
+ len(containerChartsTmpl)*len(pod.Spec.Containers)+
496
+ len(deploymentChartsTmpl)+
497
+ len(baseCharts),
498
len(*collr.Charts()),
499
)
500
module.TestMetricsHasAllChartsDims(t, collr.Charts(), mx)
@@ -967,6 +978,21 @@ func newPod(nodeName, name string) *corev1.Pod {
978
}
979
}
980
981
+func newDeployment(name string) *appsv1.Deployment {
982
+ return &appsv1.Deployment{
983
+ ObjectMeta: metav1.ObjectMeta{
984
+ Name: name,
985
+ Namespace: corev1.NamespaceDefault,
986
+ CreationTimestamp: metav1.Time{Time: time.Now()},
987
+ },
988
+ Status: appsv1.DeploymentStatus{
989
+ AvailableReplicas: 1,
990
+ Replicas: 2,
991
+ ReadyReplicas: 3,
992
+ },
993
+ }
994
+}
995
+
996
type brokenInfoKubeClient struct {
997
kubernetes.Interface
998
}
src/go/plugin/go.d/collector/k8s_state/discover_deployment.go
new
+107
@@ -0,0 +1,107 @@
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 newDeploymentDiscoverer(si cache.SharedInformer, l *logger.Logger) *deploymentDiscoverer {
15
+ if si == nil {
16
+ panic("nil deployment& shared informer")
17
+ }
18
+
19
+ queue := workqueue.NewTypedWithConfig(workqueue.TypedQueueConfig[any]{Name: "replicaset"})
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 &deploymentDiscoverer{
28
+ Logger: l,
29
+ informer: si,
30
+ queue: queue,
31
+ readyCh: make(chan struct{}),
32
+ stopCh: make(chan struct{}),
33
+ }
34
+}
35
+
36
+type deployResource struct {
37
+ src string
38
+ val any
39
+}
40
+
41
+func (r deployResource) source() string { return r.src }
42
+func (r deployResource) kind() kubeResourceKind { return kubeResourceDeployment }
43
+func (r deployResource) value() any { return r.val }
44
+
45
+type deploymentDiscoverer struct {
46
+ *logger.Logger
47
+ informer cache.SharedInformer
48
+ queue *workqueue.Typed[any]
49
+ readyCh chan struct{}
50
+ stopCh chan struct{}
51
+}
52
+
53
+func (d *deploymentDiscoverer) run(ctx context.Context, in chan<- resource) {
54
+ d.Info("deployment_discoverer is started")
55
+ defer func() { close(d.stopCh); d.Info("deployment_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 *deploymentDiscoverer) ready() bool { return isChanClosed(d.readyCh) }
73
+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()
78
+ if shutdown {
79
+ return
80
+ }
81
+
82
+ func() {
83
+ defer d.queue.Done(item)
84
+
85
+ key := item.(string)
86
+ ns, name, err := cache.SplitMetaNamespaceKey(key)
87
+ if err != nil {
88
+ return
89
+ }
90
+
91
+ item, exists, err := d.informer.GetStore().GetByKey(key)
92
+ if err != nil {
93
+ return
94
+ }
95
+
96
+ r := &deployResource{src: deploymentSource(ns, name)}
97
+ if exists {
98
+ r.val = item
99
+ }
100
+ send(ctx, in, r)
101
+ }()
102
+ }
103
+}
104
+
105
+func deploymentSource(namespace, name string) string {
106
+ return "k8s/rs/" + namespace + "/" + name
107
+}
src/go/plugin/go.d/collector/k8s_state/discover_kubernetes.go
+8
@@ -10,6 +10,7 @@ import (
10
11
"github.com/netdata/netdata/go/plugins/logger"
12
13
+ appsv1 "k8s.io/api/apps/v1"
14
corev1 "k8s.io/api/core/v1"
15
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
16
"k8s.io/apimachinery/pkg/runtime"
@@ -132,9 +133,16 @@ func (d *kubeDiscovery) setupDiscoverers(ctx context.Context) []discoverer {
133
},
134
}
135
136
+ deploy := d.client.AppsV1().Deployments(corev1.NamespaceAll)
137
+ deployWatcher := &cache.ListWatch{
138
+ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return deploy.List(ctx, options) },
139
+ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return deploy.Watch(ctx, options) },
140
+ }
141
+
142
return []discoverer{
143
newNodeDiscoverer(cache.NewSharedInformer(nodeWatcher, &corev1.Node{}, resyncPeriod), d.Logger),
144
newPodDiscoverer(cache.NewSharedInformer(podWatcher, &corev1.Pod{}, resyncPeriod), d.Logger),
145
+ newDeploymentDiscoverer(cache.NewSharedInformer(deployWatcher, &appsv1.Deployment{}, resyncPeriod), d.Logger),
146
}
147
}
148
src/go/plugin/go.d/collector/k8s_state/metadata.yaml
+26
@@ -213,6 +213,32 @@ modules:
213
chart_type: line
214
dimensions:
215
- name: age
216
+ - name: deployment
217
+ description: These metrics refer to Deployments.
218
+ labels:
219
+ - name: k8s_cluster_id
220
+ description: Cluster ID. This is equal to the kube-system namespace UID.
221
+ - name: k8s_cluster_name
222
+ description: Cluster name. Cluster name discovery only works in GKE.
223
+ - name: k8s_deployment_name
224
+ description: Deployment name.
225
+ - name: k8s_namespace
226
+ description: Namespace.
227
+ metrics:
228
+ - name: k8s_state.deployment_replicas
229
+ description: Deployment Replicas
230
+ unit: 'replicas'
231
+ chart_type: line
232
+ dimensions:
233
+ - name: desired
234
+ - name: current
235
+ - name: ready
236
+ - name: k8s_state.deployment_age
237
+ description: Deployment Age
238
+ unit: 'seconds'
239
+ chart_type: line
240
+ dimensions:
241
+ - name: age
242
- name: pod
243
description: These metrics refer to the Pod.
244
labels:
src/go/plugin/go.d/collector/k8s_state/resource.go
+13
@@ -5,6 +5,7 @@ package k8s_state
5
import (
6
"fmt"
7
8
+ appsv1 "k8s.io/api/apps/v1"
9
corev1 "k8s.io/api/core/v1"
10
)
11
@@ -19,6 +20,7 @@ type kubeResourceKind uint8
20
const (
21
kubeResourceNode kubeResourceKind = iota + 1
22
kubeResourcePod
23
+ kubeResourceDeployment
24
)
25
26
func toNode(i any) (*corev1.Node, error) {
@@ -42,3 +44,14 @@ func toPod(i any) (*corev1.Pod, error) {
44
return nil, fmt.Errorf("unexpected type: %T (expected %T or %T)", v, &corev1.Pod{}, resource(nil))
45
}
46
}
47
+
48
+func toDeployment(i any) (*appsv1.Deployment, error) {
49
+ switch v := i.(type) {
50
+ case *appsv1.Deployment:
51
+ return v, nil
52
+ case resource:
53
+ return toDeployment(v.value())
54
+ default:
55
+ return nil, fmt.Errorf("unexpected type: %T (expected %T or %T)", v, &appsv1.Deployment{}, resource(nil))
56
+ }
57
+}
src/go/plugin/go.d/collector/k8s_state/state.go
+29
-5
@@ -11,9 +11,10 @@ import (
11
12
func newKubeState() *kubeState {
13
return &kubeState{
14
- Mutex: &sync.Mutex{},
15
- nodes: make(map[string]*nodeState),
16
- pods: make(map[string]*podState),
14
+ Mutex: &sync.Mutex{},
15
+ nodes: make(map[string]*nodeState),
16
+ pods: make(map[string]*podState),
17
+ deployments: make(map[string]*deploymentState),
18
}
19
}
20
@@ -39,10 +40,17 @@ func newContainerState() *containerState {
40
}
41
}
42
43
+func newDeploymentState() *deploymentState {
44
+ return &deploymentState{
45
+ new: true,
46
+ }
47
+}
48
+
49
type kubeState struct {
50
*sync.Mutex
44
- nodes map[string]*nodeState
45
- pods map[string]*podState
51
+ nodes map[string]*nodeState
52
+ pods map[string]*podState
53
+ deployments map[string]*deploymentState
54
}
55
56
type (
@@ -149,3 +157,19 @@ type containerState struct {
157
waitingReason string
158
terminatedReason string
159
}
160
+
161
+type deploymentState struct {
162
+ new bool
163
+ deleted bool
164
+
165
+ uid string
166
+ name string
167
+ namespace string
168
+
169
+ creationTime time.Time
170
+ replicas int64 // desired
171
+ availableReplicas int64 // current
172
+ readyReplicas int64
173
+}
174
+
175
+func (ds deploymentState) id() string { return ds.namespace + "_" + ds.name }
src/go/plugin/go.d/collector/k8s_state/update_deployment_state.go
new
+35
@@ -0,0 +1,35 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+package k8s_state
4
+
5
+func (c *Collector) updateDeploymentState(r resource) {
6
+ if r.value() == nil {
7
+ if rs, ok := c.state.deployments[r.source()]; ok {
8
+ rs.deleted = true
9
+ }
10
+ return
11
+ }
12
+
13
+ deploy, err := toDeployment(r)
14
+ if err != nil {
15
+ c.Warning(err)
16
+ return
17
+ }
18
+
19
+ ds, ok := c.state.deployments[r.source()]
20
+ if !ok {
21
+ ds = newDeploymentState()
22
+ c.state.deployments[r.source()] = ds
23
+ }
24
+
25
+ if !ok {
26
+ ds.name = deploy.Name
27
+ ds.namespace = deploy.Namespace
28
+ ds.uid = string(deploy.UID)
29
+ ds.creationTime = deploy.CreationTimestamp.Time
30
+ }
31
+
32
+ ds.replicas = int64(deploy.Status.Replicas)
33
+ ds.availableReplicas = int64(deploy.Status.AvailableReplicas)
34
+ ds.readyReplicas = int64(deploy.Status.ReadyReplicas)
35
+}
src/go/plugin/go.d/collector/k8s_state/update_state.go
+2
@@ -14,6 +14,8 @@ func (c *Collector) runUpdateState(in <-chan resource) {
14
c.updateNodeState(r)
15
case kubeResourcePod:
16
c.updatePodState(r)
17
+ case kubeResourceDeployment:
18
+ c.updateDeploymentState(r)
19
}
20
c.state.Unlock()
21
}