master
go 459 lines 14.3 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 package k8s_state
4
5 import (
6 "errors"
7 "fmt"
8 "maps"
9 "slices"
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/framework/collectorapi"
17 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/oldmetrix"
18 )
19
20 const precision = 1000
21
22 var (
23 podStatusReasons = []string{
24 "Evicted",
25 "NodeAffinity",
26 "NodeLost",
27 "Shutdown",
28 "UnexpectedAdmissionError",
29 "Other",
30 }
31
32 containerWaitingStateReasons = []string{
33 "ContainerCreating",
34 "CrashLoopBackOff",
35 "CreateContainerConfigError",
36 "CreateContainerError",
37 "ErrImagePull",
38 "ImagePullBackOff",
39 "InvalidImageName",
40 "PodInitializing",
41 "Other",
42 }
43 containerTerminatedStateReasons = []string{
44 "Completed",
45 "ContainerCannotRun",
46 "DeadlineExceeded",
47 "Error",
48 "Evicted",
49 "OOMKilled",
50 "Other",
51 }
52 )
53
54 var (
55 nodeConditionStatuses = []string{
56 "Ready",
57 "DiskPressure",
58 "MemoryPressure",
59 "NetworkUnavailable",
60 "PIDPressure",
61 }
62 )
63
64 func (c *Collector) collect() (map[string]int64, error) {
65 if c.discoverer == nil {
66 return nil, errors.New("nil discoverer")
67 }
68
69 c.once.Do(func() {
70 c.startTime = time.Now()
71 in := make(chan resource)
72
73 c.wg.Go(func() { c.runUpdateState(in) })
74
75 c.wg.Go(func() { c.discoverer.run(c.ctx, in) })
76
77 c.kubeClusterID = c.getKubeClusterID()
78 c.kubeClusterName = c.getKubeClusterName()
79
80 if chart := c.Charts().Get(discoveryStatusChart.ID); chart != nil {
81 chart.Labels = []collectorapi.Label{
82 {Key: labelKeyClusterID, Value: c.kubeClusterID, Source: collectorapi.LabelSourceK8s},
83 {Key: labelKeyClusterName, Value: c.kubeClusterName, Source: collectorapi.LabelSourceK8s},
84 }
85 }
86 })
87
88 mx := map[string]int64{
89 "discovery_node_discoverer_state": 1,
90 "discovery_pod_discoverer_state": 1,
91 }
92
93 if !c.discoverer.ready() || time.Since(c.startTime) < c.initDelay {
94 return mx, nil
95 }
96
97 c.state.Lock()
98 defer c.state.Unlock()
99
100 c.collectKubeState(mx)
101
102 return mx, nil
103 }
104
105 func (c *Collector) collectKubeState(mx map[string]int64) {
106 for _, ns := range c.state.nodes {
107 ns.resetStats()
108 }
109 c.collectPodsState(mx)
110 c.collectNodesState(mx)
111 c.collectDeploymentState(mx)
112 c.collectCronJobState(mx)
113 }
114
115 func (c *Collector) collectPodsState(mx map[string]int64) {
116 now := time.Now()
117 for _, ps := range c.state.pods {
118 // Skip cronjobs (each of them is a unique container because the name contains hash)
119 // to avoid overwhelming Netdata with high cardinality metrics.
120 // Related issue https://github.com/netdata/netdata/issues/16412
121 if ps.controllerKind == "Job" {
122 continue
123 }
124
125 if ps.deleted {
126 delete(c.state.pods, podSource(ps.namespace, ps.name))
127 c.removePodCharts(ps)
128 continue
129 }
130
131 if ps.new {
132 ps.new = false
133 c.addPodCharts(ps)
134 ps.unscheduled = ps.nodeName == ""
135 } else if ps.unscheduled && ps.nodeName != "" {
136 ps.unscheduled = false
137 c.updatePodChartsNodeLabel(ps)
138 }
139
140 ns := c.state.nodes[nodeSource(ps.nodeName)]
141 if ns != nil {
142 ns.stats.pods++
143 ns.stats.reqCPU += ps.reqCPU
144 ns.stats.limitCPU += ps.limitCPU
145 ns.stats.reqMem += ps.reqMem
146 ns.stats.limitMem += ps.limitMem
147 ns.stats.podsCondPodReady += condStatusToInt(ps.condPodReady)
148 ns.stats.podsCondPodScheduled += condStatusToInt(ps.condPodScheduled)
149 ns.stats.podsCondPodInitialized += condStatusToInt(ps.condPodInitialized)
150 ns.stats.podsCondContainersReady += condStatusToInt(ps.condContainersReady)
151 ns.stats.podsReadinessReady += oldmetrix.Bool(ps.condPodReady == corev1.ConditionTrue)
152 ns.stats.podsReadinessUnready += oldmetrix.Bool(ps.condPodReady != corev1.ConditionTrue)
153 ns.stats.podsPhasePending += oldmetrix.Bool(ps.phase == corev1.PodPending)
154 ns.stats.podsPhaseRunning += oldmetrix.Bool(ps.phase == corev1.PodRunning)
155 ns.stats.podsPhaseSucceeded += oldmetrix.Bool(ps.phase == corev1.PodSucceeded)
156 ns.stats.podsPhaseFailed += oldmetrix.Bool(ps.phase == corev1.PodFailed)
157
158 for _, cs := range ps.initContainers {
159 ns.stats.initContainers++
160 ns.stats.initContStateRunning += oldmetrix.Bool(cs.stateRunning)
161 ns.stats.initContStateWaiting += oldmetrix.Bool(cs.stateWaiting)
162 ns.stats.initContStateTerminated += oldmetrix.Bool(cs.stateTerminated)
163 }
164
165 for _, cs := range ps.containers {
166 ns.stats.containers++
167 ns.stats.contStateRunning += oldmetrix.Bool(cs.stateRunning)
168 ns.stats.contStateWaiting += oldmetrix.Bool(cs.stateWaiting)
169 ns.stats.contStateTerminated += oldmetrix.Bool(cs.stateTerminated)
170 }
171 }
172
173 px := fmt.Sprintf("pod_%s_", ps.id())
174
175 mx[px+"cond_podready"] = condStatusToInt(ps.condPodReady)
176 mx[px+"cond_podscheduled"] = condStatusToInt(ps.condPodScheduled)
177 mx[px+"cond_podinitialized"] = condStatusToInt(ps.condPodInitialized)
178 mx[px+"cond_containersready"] = condStatusToInt(ps.condContainersReady)
179 mx[px+"phase_running"] = oldmetrix.Bool(ps.phase == corev1.PodRunning)
180 mx[px+"phase_failed"] = oldmetrix.Bool(ps.phase == corev1.PodFailed)
181 mx[px+"phase_succeeded"] = oldmetrix.Bool(ps.phase == corev1.PodSucceeded)
182 mx[px+"phase_pending"] = oldmetrix.Bool(ps.phase == corev1.PodPending)
183 mx[px+"age"] = int64(now.Sub(ps.creationTime).Seconds())
184
185 for _, v := range podStatusReasons {
186 mx[px+"status_reason_"+v] = 0
187 }
188 if v := ps.statusReason; v != "" {
189 if !slices.Contains(podStatusReasons, v) {
190 v = "Other"
191 }
192 mx[px+"status_reason_"+v] = 1
193 }
194
195 mx[px+"cpu_requests_used"] = ps.reqCPU
196 mx[px+"cpu_limits_used"] = ps.limitCPU
197 mx[px+"mem_requests_used"] = ps.reqMem
198 mx[px+"mem_limits_used"] = ps.limitMem
199
200 mx[px+"init_containers"] = int64(len(ps.initContainers))
201 mx[px+"containers"] = int64(len(ps.containers))
202
203 mx[px+"init_containers_state_running"] = 0
204 mx[px+"init_containers_state_waiting"] = 0
205 mx[px+"init_containers_state_terminated"] = 0
206
207 for _, cs := range ps.initContainers {
208 mx[px+"init_containers_state_running"] += oldmetrix.Bool(cs.stateRunning)
209 mx[px+"init_containers_state_waiting"] += oldmetrix.Bool(cs.stateWaiting)
210 mx[px+"init_containers_state_terminated"] += oldmetrix.Bool(cs.stateTerminated)
211 }
212 mx[px+"containers_state_running"] = 0
213 mx[px+"containers_state_waiting"] = 0
214 mx[px+"containers_state_terminated"] = 0
215
216 for _, cs := range ps.containers {
217 if cs.new {
218 cs.new = false
219 c.addContainerCharts(ps, cs)
220 }
221 mx[px+"containers_state_running"] += oldmetrix.Bool(cs.stateRunning)
222 mx[px+"containers_state_waiting"] += oldmetrix.Bool(cs.stateWaiting)
223 mx[px+"containers_state_terminated"] += oldmetrix.Bool(cs.stateTerminated)
224
225 ppx := fmt.Sprintf("%scontainer_%s_", px, cs.name)
226 mx[ppx+"state_running"] = oldmetrix.Bool(cs.stateRunning)
227 mx[ppx+"state_waiting"] = oldmetrix.Bool(cs.stateWaiting)
228 mx[ppx+"state_terminated"] = oldmetrix.Bool(cs.stateTerminated)
229 mx[ppx+"readiness"] = oldmetrix.Bool(cs.ready)
230 mx[ppx+"restarts"] = cs.restarts
231
232 for _, v := range containerWaitingStateReasons {
233 mx[ppx+"state_waiting_reason_"+v] = 0
234 }
235 if v := cs.waitingReason; v != "" {
236 if !slices.Contains(containerWaitingStateReasons, v) {
237 v = "Other"
238 }
239 mx[ppx+"state_waiting_reason_"+v] = 1
240 }
241
242 for _, v := range containerTerminatedStateReasons {
243 mx[ppx+"state_terminated_reason_"+v] = 0
244 }
245 if v := cs.terminatedReason; v != "" {
246 if !slices.Contains(containerTerminatedStateReasons, v) {
247 v = "Other"
248 }
249 mx[ppx+"state_terminated_reason_"+v] = 1
250 }
251 }
252 }
253 }
254
255 func (c *Collector) collectNodesState(mx map[string]int64) {
256 now := time.Now()
257 for _, ns := range c.state.nodes {
258 if ns.deleted {
259 delete(c.state.nodes, nodeSource(ns.name))
260 c.removeNodeCharts(ns)
261 continue
262 }
263 if ns.new {
264 ns.new = false
265 c.addNodeCharts(ns)
266 }
267
268 px := fmt.Sprintf("node_%s_", ns.id())
269
270 for _, v := range nodeConditionStatuses {
271 mx[px+"cond_"+v] = 0
272 }
273 for _, v := range ns.conditions {
274 mx[px+"cond_"+string(v.Type)] = condStatusToInt(v.Status)
275 }
276
277 mx[px+"age"] = int64(now.Sub(ns.creationTime).Seconds())
278 mx[px+"alloc_pods_util"] = calcPercentage(ns.stats.pods, ns.allocatablePods)
279 mx[px+"pods_readiness_ready"] = ns.stats.podsReadinessReady
280 mx[px+"pods_readiness_unready"] = ns.stats.podsReadinessUnready
281 mx[px+"pods_readiness"] = calcPercentage(ns.stats.podsReadinessReady, ns.stats.pods)
282 mx[px+"pods_phase_running"] = ns.stats.podsPhaseRunning
283 mx[px+"pods_phase_failed"] = ns.stats.podsPhaseFailed
284 mx[px+"pods_phase_succeeded"] = ns.stats.podsPhaseSucceeded
285 mx[px+"pods_phase_pending"] = ns.stats.podsPhasePending
286 mx[px+"pods_cond_podready"] = ns.stats.podsCondPodReady
287 mx[px+"pods_cond_podscheduled"] = ns.stats.podsCondPodScheduled
288 mx[px+"pods_cond_podinitialized"] = ns.stats.podsCondPodInitialized
289 mx[px+"pods_cond_containersready"] = ns.stats.podsCondContainersReady
290 mx[px+"pods_cond_containersready"] = ns.stats.podsCondContainersReady
291 mx[px+"schedulability_schedulable"] = oldmetrix.Bool(!ns.unSchedulable)
292 mx[px+"schedulability_unschedulable"] = oldmetrix.Bool(ns.unSchedulable)
293 mx[px+"alloc_pods_available"] = ns.allocatablePods - ns.stats.pods
294 mx[px+"alloc_pods_allocated"] = ns.stats.pods
295 mx[px+"alloc_cpu_requests_util"] = calcPercentage(ns.stats.reqCPU, ns.allocatableCPU)
296 mx[px+"alloc_cpu_limits_util"] = calcPercentage(ns.stats.limitCPU, ns.allocatableCPU)
297 mx[px+"alloc_mem_requests_util"] = calcPercentage(ns.stats.reqMem, ns.allocatableMem)
298 mx[px+"alloc_mem_limits_util"] = calcPercentage(ns.stats.limitMem, ns.allocatableMem)
299 mx[px+"alloc_cpu_requests_used"] = ns.stats.reqCPU
300 mx[px+"alloc_cpu_limits_used"] = ns.stats.limitCPU
301 mx[px+"alloc_mem_requests_used"] = ns.stats.reqMem
302 mx[px+"alloc_mem_limits_used"] = ns.stats.limitMem
303 mx[px+"init_containers"] = ns.stats.initContainers
304 mx[px+"containers"] = ns.stats.containers
305 mx[px+"containers_state_running"] = ns.stats.contStateRunning
306 mx[px+"containers_state_waiting"] = ns.stats.contStateWaiting
307 mx[px+"containers_state_terminated"] = ns.stats.contStateTerminated
308 mx[px+"init_containers_state_running"] = ns.stats.initContStateRunning
309 mx[px+"init_containers_state_waiting"] = ns.stats.initContStateWaiting
310 mx[px+"init_containers_state_terminated"] = ns.stats.initContStateTerminated
311 }
312 }
313
314 func (c *Collector) collectDeploymentState(mx map[string]int64) {
315 now := time.Now()
316
317 maps.DeleteFunc(c.state.deployments, func(s string, ds *deploymentState) bool {
318 if ds.deleted {
319 c.removeDeploymentCharts(ds)
320 return true
321 }
322
323 if ds.new {
324 ds.new = false
325 c.addDeploymentCharts(ds)
326 }
327
328 px := fmt.Sprintf("deploy_%s_", ds.id())
329
330 mx[px+"age"] = int64(now.Sub(ds.creationTime).Seconds())
331 mx[px+"desired_replicas"] = ds.replicas
332 mx[px+"current_replicas"] = ds.availableReplicas
333 mx[px+"ready_replicas"] = ds.readyReplicas
334
335 mx[px+"condition_available"] = 0
336 mx[px+"condition_progressing"] = 0
337 mx[px+"condition_replica_failure"] = 0
338
339 for _, cond := range ds.conditions {
340 v := oldmetrix.Bool(cond.Status == corev1.ConditionTrue)
341 switch cond.Type {
342 case appsv1.DeploymentAvailable:
343 // https://github.com/kubernetes/kubernetes/blob/2b3da7dfc846fec7c4044a320f8f38b4a45367a3/pkg/controller/deployment/sync.go#L518-L525
344 mx[px+"condition_available"] = v
345 case appsv1.DeploymentProgressing:
346 mx[px+"condition_progressing"] = v
347 case appsv1.DeploymentReplicaFailure:
348 mx[px+"condition_replica_failure"] = v
349 }
350 }
351
352 return false
353 })
354 }
355
356 func (c *Collector) collectCronJobState(mx map[string]int64) {
357 now := time.Now()
358
359 maps.DeleteFunc(c.state.jobs, func(s string, st *jobState) bool {
360 return st.deleted
361 })
362
363 maps.DeleteFunc(c.state.cronJobs, func(s string, st *cronJobState) bool {
364 if st.deleted {
365 c.removeCronJobCharts(st)
366 return true
367 }
368 if st.new {
369 c.addCronJobCharts(st)
370 st.new = false
371 }
372
373 px := fmt.Sprintf("cronjob_%s_", st.id())
374
375 mx[px+"age"] = int64(now.Sub(st.creationTime).Seconds())
376 if st.lastScheduleTime != nil {
377 mx[px+"last_schedule_seconds_ago"] = int64(now.Sub(*st.lastScheduleTime).Seconds())
378 }
379 if st.lastSuccessfulTime != nil {
380 mx[px+"last_successful_seconds_ago"] = int64(now.Sub(*st.lastSuccessfulTime).Seconds())
381 }
382
383 mx[px+"running_jobs"] = 0
384 mx[px+"failed_jobs"] = 0
385 mx[px+"complete_jobs"] = 0
386 mx[px+"complete_jobs"] = 0
387 mx[px+"suspended_jobs"] = 0
388
389 mx[px+"suspend_status_enabled"] = oldmetrix.Bool(!st.suspend)
390 mx[px+"suspend_status_suspended"] = oldmetrix.Bool(st.suspend)
391
392 mx[px+"failed_jobs_reason_pod_failure_policy"] = 0
393 mx[px+"failed_jobs_reason_backoff_limit_exceeded"] = 0
394 mx[px+"failed_jobs_reason_deadline_exceeded"] = 0
395
396 mx[px+"last_execution_status_succeeded"] = 0
397 mx[px+"last_execution_status_failed"] = 0
398
399 var lastExecutedEndTime time.Time
400 var lastCompleteTime time.Time
401
402 for _, job := range c.state.jobs {
403 if job.controller.kind != "CronJob" || job.controller.uid != st.uid || job.startTime == nil {
404 continue
405 }
406 if job.active > 0 {
407 mx[px+"running_jobs"]++
408 continue
409 }
410
411 for _, cond := range job.conditions {
412 if cond.Status != corev1.ConditionTrue {
413 continue
414 }
415
416 switch cond.Type {
417 case batchv1.JobComplete:
418 mx[px+"complete_jobs"]++
419 if job.completionTime != nil {
420 if job.completionTime.After(lastExecutedEndTime) {
421 lastExecutedEndTime = *job.completionTime
422 mx[px+"last_execution_status_succeeded"] = 1
423 mx[px+"last_execution_status_failed"] = 0
424 }
425 if job.completionTime.After(lastCompleteTime) {
426 lastCompleteTime = *job.completionTime
427 mx[px+"last_completion_duration"] = int64(job.completionTime.Sub(*job.startTime).Seconds())
428 }
429 }
430 case batchv1.JobFailed:
431 mx[px+"failed_jobs"]++
432 mx[px+"failed_jobs_reason_pod_failure_policy"] += oldmetrix.Bool(cond.Reason == batchv1.JobReasonPodFailurePolicy)
433 mx[px+"failed_jobs_reason_backoff_limit_exceeded"] += oldmetrix.Bool(cond.Reason == batchv1.JobReasonBackoffLimitExceeded)
434 mx[px+"failed_jobs_reason_deadline_exceeded"] += oldmetrix.Bool(cond.Reason == batchv1.JobReasonDeadlineExceeded)
435 if cond.LastTransitionTime.Time.After(lastExecutedEndTime) {
436 lastExecutedEndTime = cond.LastTransitionTime.Time
437 mx[px+"last_execution_status_succeeded"] = 0
438 mx[px+"last_execution_status_failed"] = 1
439 }
440 case batchv1.JobSuspended:
441 mx[px+"suspended_jobs"]++
442 }
443 }
444 }
445
446 return false
447 })
448 }
449
450 func condStatusToInt(cs corev1.ConditionStatus) int64 {
451 return oldmetrix.Bool(cs == corev1.ConditionTrue)
452 }
453
454 func calcPercentage(value, total int64) int64 {
455 if total == 0 {
456 return 0
457 }
458 return int64(float64(value) / float64(total) * 100 * precision)
459 }