master
go 1,091 lines 32.9 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 package k8s_apiserver
4
5 import (
6 "fmt"
7 "math"
8 "sort"
9 "strings"
10
11 "github.com/prometheus/common/model"
12
13 "github.com/netdata/netdata/go/plugins/pkg/prometheus"
14 "github.com/netdata/netdata/go/plugins/pkg/stm"
15 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
16 mtx "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/oldmetrix"
17 )
18
19 const (
20 precision = 1000 // for counters (e.g., CPU seconds)
21
22 // For latency: multiply seconds by 1e6 to get microseconds, then chart Div: 1000 gives milliseconds
23 latencyPrecision = 1000000
24
25 // Default cardinality limits to prevent unbounded memory growth
26 // Once these limits are reached, new dimensions are silently ignored
27 defaultMaxResources = 500
28 defaultMaxWorkqueues = 100
29 defaultMaxAdmCtrl = 100
30 defaultMaxAdmWebhooks = 50
31
32 // Cleanup: dimensions not seen for this many cycles are removed
33 staleThresholdCycles = 300 // ~5 minutes at 1s update interval
34 )
35
36 // K8s admission histogram bucket bounds (seconds)
37 // Ref: https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/endpoints/metrics/metrics.go
38 var admissionBucketBounds = []float64{0.005, 0.025, 0.1, 0.5, 1.0, 2.5}
39
40 // idReplacer sanitizes label values for use in chart/dimension IDs
41 // Replaces characters that could cause issues in Netdata with underscores
42 var idReplacer = strings.NewReplacer(
43 ".", "_",
44 " ", "_",
45 "/", "_",
46 ":", "_",
47 )
48
49 // cleanID sanitizes a string for use in chart or dimension IDs
50 func cleanID(s string) string {
51 return strings.ToLower(idReplacer.Replace(s))
52 }
53
54 func (c *Collector) collect() (map[string]int64, error) {
55 mfs, err := c.prom.Scrape()
56 if err != nil {
57 return nil, err
58 }
59
60 c.collectCycle++
61
62 mx := newMetrics()
63
64 c.collectRequests(mfs, mx)
65 c.collectInflight(mfs, mx)
66 c.collectRESTClient(mfs, mx)
67 c.collectAdmission(mfs, mx)
68 c.collectEtcd(mfs, mx)
69 c.collectWorkqueues(mfs, mx)
70 c.collectProcess(mfs, mx)
71 c.collectAudit(mfs, mx)
72 c.collectAuth(mfs, mx)
73
74 // Periodically cleanup stale dimensions (every 100 cycles to avoid overhead)
75 if c.collectCycle%100 == 0 {
76 c.cleanupStaleDimensions()
77 }
78
79 return stm.ToMap(mx), nil
80 }
81
82 // collectRequests collects apiserver_request_total, apiserver_request_duration_seconds, etc.
83 func (c *Collector) collectRequests(mfs prometheus.MetricFamilies, mx *metrics) {
84 // Total requests and by verb/code/resource
85 // Prefer apiserver_request_total (newer), fall back to apiserver_request_count (legacy)
86 mf := mfs.Get("apiserver_request_total")
87 if mf == nil {
88 mf = mfs.Get("apiserver_request_count")
89 }
90
91 if mf != nil {
92 for _, m := range mf.Metrics() {
93 value := metricValue(mf, m)
94 if math.IsNaN(value) {
95 continue
96 }
97
98 verb := m.Labels().Get("verb")
99 code := m.Labels().Get("code")
100 resource := m.Labels().Get("resource")
101
102 mx.Request.Total.Add(value)
103
104 // By verb
105 if verb != "" {
106 c.addVerbDimension(verb)
107 verbID := cleanID(verb)
108 mx.Request.ByVerb[verbID] = mtx.Gauge(mx.Request.ByVerb[verbID].Value() + value)
109 }
110
111 // By code
112 if code != "" {
113 c.addCodeDimension(code)
114 codeID := cleanID(code)
115 mx.Request.ByCode[codeID] = mtx.Gauge(mx.Request.ByCode[codeID].Value() + value)
116 }
117
118 // By resource (with cardinality limit)
119 if resource != "" {
120 resourceID := cleanID(resource)
121 _, seen := c.collectedResources[resourceID]
122 if seen || len(c.collectedResources) < defaultMaxResources {
123 c.addResourceDimension(resource)
124 c.collectedResources[resourceID] = c.collectCycle
125 mx.Request.ByResource[resourceID] = mtx.Gauge(mx.Request.ByResource[resourceID].Value() + value)
126 }
127 }
128 }
129 }
130
131 // Dropped/rejected requests - sum both legacy and APF (flow control) metrics
132 // Both can be present and meaningful on different K8s versions
133 var dropped float64
134 if mf := mfs.Get("apiserver_dropped_requests_total"); mf != nil {
135 for _, m := range mf.Metrics() {
136 if v := metricValue(mf, m); !math.IsNaN(v) {
137 dropped += v
138 }
139 }
140 }
141 if mf := mfs.Get("apiserver_flowcontrol_rejected_requests_total"); mf != nil {
142 for _, m := range mf.Metrics() {
143 if v := metricValue(mf, m); !math.IsNaN(v) {
144 dropped += v
145 }
146 }
147 }
148 mx.Request.Dropped.Set(dropped)
149
150 // Request latency and response size (histograms)
151 c.collectRequestLatency(mfs, mx)
152 c.collectResponseSize(mfs, mx)
153 }
154
155 func (c *Collector) collectRequestLatency(mfs prometheus.MetricFamilies, mx *metrics) {
156 mf := mfs.Get("apiserver_request_duration_seconds")
157 if mf == nil || mf.Type() != model.MetricTypeHistogram {
158 return
159 }
160
161 hd := collectHistogramBucketsFromMF(mf, nil)
162 if len(hd.buckets) > 0 {
163 if p50 := histogramPercentile(hd, 0.5); !math.IsNaN(p50) {
164 mx.Request.Latency.P50.Set(p50 * latencyPrecision)
165 }
166 if p90 := histogramPercentile(hd, 0.9); !math.IsNaN(p90) {
167 mx.Request.Latency.P90.Set(p90 * latencyPrecision)
168 }
169 if p99 := histogramPercentile(hd, 0.99); !math.IsNaN(p99) {
170 mx.Request.Latency.P99.Set(p99 * latencyPrecision)
171 }
172 }
173 }
174
175 func (c *Collector) collectResponseSize(mfs prometheus.MetricFamilies, mx *metrics) {
176 mf := mfs.Get("apiserver_response_sizes")
177 if mf == nil || mf.Type() != model.MetricTypeHistogram {
178 return
179 }
180
181 hd := collectHistogramBucketsFromMF(mf, nil)
182 if len(hd.buckets) > 0 {
183 if p50 := histogramPercentile(hd, 0.5); !math.IsNaN(p50) {
184 mx.Request.ResponseSize.P50.Set(p50)
185 }
186 if p90 := histogramPercentile(hd, 0.9); !math.IsNaN(p90) {
187 mx.Request.ResponseSize.P90.Set(p90)
188 }
189 if p99 := histogramPercentile(hd, 0.99); !math.IsNaN(p99) {
190 mx.Request.ResponseSize.P99.Set(p99)
191 }
192 }
193 }
194
195 // collectInflight collects apiserver_current_inflight_requests and apiserver_longrunning_requests
196 func (c *Collector) collectInflight(mfs prometheus.MetricFamilies, mx *metrics) {
197 if mf := mfs.Get("apiserver_current_inflight_requests"); mf != nil {
198 for _, m := range mf.Metrics() {
199 value := metricValue(mf, m)
200 if math.IsNaN(value) {
201 continue
202 }
203 kind := m.Labels().Get("request_kind")
204 switch kind {
205 case "mutating":
206 mx.Inflight.Mutating.Set(value)
207 case "readOnly":
208 mx.Inflight.ReadOnly.Set(value)
209 }
210 }
211 }
212
213 // Sum all long-running requests across all label combinations
214 if mf := mfs.Get("apiserver_longrunning_requests"); mf != nil {
215 var total float64
216 for _, m := range mf.Metrics() {
217 if v := metricValue(mf, m); !math.IsNaN(v) {
218 total += v
219 }
220 }
221 mx.Inflight.Longrunning.Set(total)
222 }
223 }
224
225 // collectRESTClient collects rest_client_requests_total and rest_client_request_duration_seconds
226 func (c *Collector) collectRESTClient(mfs prometheus.MetricFamilies, mx *metrics) {
227 codeChart := c.charts.Get("rest_client_requests_by_code")
228 methodChart := c.charts.Get("rest_client_requests_by_method")
229
230 if mf := mfs.Get("rest_client_requests_total"); mf != nil {
231 for _, m := range mf.Metrics() {
232 value := metricValue(mf, m)
233 if math.IsNaN(value) {
234 continue
235 }
236
237 code := m.Labels().Get("code")
238 method := m.Labels().Get("method")
239
240 // By code (track for cleanup)
241 if code != "" {
242 codeID := cleanID(code)
243 _, seen := c.collectedRESTCodes[codeID]
244 c.collectedRESTCodes[codeID] = c.collectCycle
245 if !seen {
246 dimID := "rest_client_by_code_" + codeID
247 if codeChart != nil && !codeChart.HasDim(dimID) {
248 if err := codeChart.AddDim(&Dim{ID: dimID, Name: code, Algo: collectorapi.Incremental}); err != nil {
249 c.Warningf("failed to add REST client code dimension %s: %v", code, err)
250 } else {
251 codeChart.MarkNotCreated()
252 }
253 }
254 }
255 mx.RESTClient.ByCode[codeID] = mtx.Gauge(mx.RESTClient.ByCode[codeID].Value() + value)
256 }
257
258 // By method (track for cleanup)
259 if method != "" {
260 methodID := cleanID(method)
261 _, seen := c.collectedRESTMethods[methodID]
262 c.collectedRESTMethods[methodID] = c.collectCycle
263 if !seen {
264 dimID := "rest_client_by_method_" + methodID
265 if methodChart != nil && !methodChart.HasDim(dimID) {
266 if err := methodChart.AddDim(&Dim{ID: dimID, Name: method, Algo: collectorapi.Incremental}); err != nil {
267 c.Warningf("failed to add REST client method dimension %s: %v", method, err)
268 } else {
269 methodChart.MarkNotCreated()
270 }
271 }
272 }
273 mx.RESTClient.ByMethod[methodID] = mtx.Gauge(mx.RESTClient.ByMethod[methodID].Value() + value)
274 }
275 }
276 }
277
278 // REST client latency
279 if mf := mfs.Get("rest_client_request_duration_seconds"); mf != nil && mf.Type() == model.MetricTypeHistogram {
280 hd := collectHistogramBucketsFromMF(mf, nil)
281 if len(hd.buckets) > 0 {
282 if p50 := histogramPercentile(hd, 0.5); !math.IsNaN(p50) {
283 mx.RESTClient.Latency.P50.Set(p50 * latencyPrecision)
284 }
285 if p90 := histogramPercentile(hd, 0.9); !math.IsNaN(p90) {
286 mx.RESTClient.Latency.P90.Set(p90 * latencyPrecision)
287 }
288 if p99 := histogramPercentile(hd, 0.99); !math.IsNaN(p99) {
289 mx.RESTClient.Latency.P99.Set(p99 * latencyPrecision)
290 }
291 }
292 }
293 }
294
295 // collectAdmission collects admission controller and webhook metrics
296 func (c *Collector) collectAdmission(mfs prometheus.MetricFamilies, mx *metrics) {
297 // Admission step latency - aggregate buckets across all operation/rejected label combinations
298 if mf := mfs.Get("apiserver_admission_step_admission_duration_seconds"); mf != nil && mf.Type() == model.MetricTypeHistogram {
299 // Use map[type][le] -> count for proper aggregation
300 stepBuckets := make(map[string]map[float64]float64)
301 stepTotals := make(map[string]float64)
302
303 for _, m := range mf.Metrics() {
304 if m.Histogram() == nil {
305 continue
306 }
307 stepType := m.Labels().Get("type")
308 if stepType == "" {
309 continue
310 }
311
312 if stepBuckets[stepType] == nil {
313 stepBuckets[stepType] = make(map[float64]float64)
314 }
315
316 for _, b := range m.Histogram().Buckets() {
317 if math.IsInf(b.UpperBound(), 0) {
318 stepTotals[stepType] += b.CumulativeCount()
319 continue
320 }
321 // Aggregate by summing counts for same le across all label combinations
322 stepBuckets[stepType][b.UpperBound()] += b.CumulativeCount()
323 }
324 }
325
326 // Convert aggregated buckets to histogramData for percentile calculation
327 for stepType, bucketMap := range stepBuckets {
328 hd := histogramData{total: stepTotals[stepType]}
329 for le, count := range bucketMap {
330 hd.buckets = append(hd.buckets, histogramBucket{le: le, count: count})
331 }
332 sortBuckets(hd.buckets)
333
334 if p50 := histogramPercentile(hd, 0.5); !math.IsNaN(p50) {
335 switch stepType {
336 case "validate":
337 mx.Admission.StepLatency.Validate.Set(p50 * latencyPrecision)
338 case "admit":
339 mx.Admission.StepLatency.Admit.Set(p50 * latencyPrecision)
340 }
341 }
342 }
343 }
344
345 // Admission controller latency (dynamic charts)
346 c.collectAdmissionControllerLatency(mfs, mx)
347 c.collectAdmissionWebhookLatency(mfs, mx)
348 }
349
350 func (c *Collector) collectAdmissionControllerLatency(mfs prometheus.MetricFamilies, mx *metrics) {
351 mf := mfs.Get("apiserver_admission_controller_admission_duration_seconds")
352 if mf == nil || mf.Type() != model.MetricTypeHistogram {
353 return
354 }
355
356 // Collect as heatmap with non-cumulative bucket counts
357 // Must aggregate by controller name, then by le bucket
358 controllerBucketMaps := make(map[string]map[float64]float64) // name -> le -> cumulative_count
359
360 for _, m := range mf.Metrics() {
361 if m.Histogram() == nil {
362 continue
363 }
364 name := m.Labels().Get("name")
365 if name == "" {
366 continue
367 }
368
369 if controllerBucketMaps[name] == nil {
370 controllerBucketMaps[name] = make(map[float64]float64)
371 }
372
373 for _, b := range m.Histogram().Buckets() {
374 controllerBucketMaps[name][b.UpperBound()] += float64(b.CumulativeCount())
375 }
376 }
377
378 // Collect names into slice first to avoid modifying maps during iteration
379 controllerNames := make([]string, 0, len(controllerBucketMaps))
380 for name := range controllerBucketMaps {
381 controllerNames = append(controllerNames, name)
382 }
383
384 for _, name := range controllerNames {
385 bucketMap := controllerBucketMaps[name]
386
387 // Sanitize name for use in chart/dimension IDs
388 cleanName := cleanID(name)
389
390 // Cardinality limit: only accept new items if under limit
391 _, seen := c.collectedAdmissionCtrl[cleanName]
392 if !seen && len(c.collectedAdmissionCtrl) >= defaultMaxAdmCtrl {
393 continue
394 }
395
396 // Track last-seen cycle
397 c.collectedAdmissionCtrl[cleanName] = c.collectCycle
398
399 if !seen {
400 if err := c.charts.Add(newAdmissionControllerLatencyChart(cleanName)); err != nil {
401 c.Warningf("failed to add admission controller chart %s: %v", name, err)
402 }
403 }
404
405 if mx.Admission.Controllers[cleanName] == nil {
406 mx.Admission.Controllers[cleanName] = &admissionControllerMetrics{}
407 }
408
409 // Use cumulative bucket values directly - chart uses Incremental algorithm
410 // This avoids negative values when Prometheus counters reset
411 c.setAdmissionBucketMetrics(bucketMap, mx.Admission.Controllers[cleanName])
412 }
413 }
414
415 func (c *Collector) collectAdmissionWebhookLatency(mfs prometheus.MetricFamilies, mx *metrics) {
416 mf := mfs.Get("apiserver_admission_webhook_admission_duration_seconds")
417 if mf == nil || mf.Type() != model.MetricTypeHistogram {
418 return
419 }
420
421 // Collect as heatmap with non-cumulative bucket counts
422 webhookBucketMaps := make(map[string]map[float64]float64) // name -> le -> cumulative_count
423
424 for _, m := range mf.Metrics() {
425 if m.Histogram() == nil {
426 continue
427 }
428 name := m.Labels().Get("name")
429 if name == "" {
430 continue
431 }
432
433 if webhookBucketMaps[name] == nil {
434 webhookBucketMaps[name] = make(map[float64]float64)
435 }
436
437 for _, b := range m.Histogram().Buckets() {
438 webhookBucketMaps[name][b.UpperBound()] += float64(b.CumulativeCount())
439 }
440 }
441
442 // Collect names into slice first to avoid modifying maps during iteration
443 webhookNames := make([]string, 0, len(webhookBucketMaps))
444 for name := range webhookBucketMaps {
445 webhookNames = append(webhookNames, name)
446 }
447
448 for _, name := range webhookNames {
449 bucketMap := webhookBucketMaps[name]
450
451 // Sanitize name for use in chart/dimension IDs
452 cleanName := cleanID(name)
453
454 // Cardinality limit: only accept new items if under limit
455 _, seen := c.collectedAdmissionWH[cleanName]
456 if !seen && len(c.collectedAdmissionWH) >= defaultMaxAdmWebhooks {
457 continue
458 }
459
460 // Track last-seen cycle
461 c.collectedAdmissionWH[cleanName] = c.collectCycle
462
463 if !seen {
464 if err := c.charts.Add(newAdmissionWebhookLatencyChart(cleanName)); err != nil {
465 c.Warningf("failed to add admission webhook chart %s: %v", name, err)
466 }
467 }
468
469 if mx.Admission.Webhooks[cleanName] == nil {
470 mx.Admission.Webhooks[cleanName] = &admissionWebhookMetrics{}
471 }
472
473 // Use cumulative bucket values directly - chart uses Incremental algorithm
474 // This avoids negative values when Prometheus counters reset
475 c.setAdmissionBucketMetrics(bucketMap, mx.Admission.Webhooks[cleanName])
476 }
477 }
478
479 // collectEtcd collects etcd/storage object count metrics
480 func (c *Collector) collectEtcd(mfs prometheus.MetricFamilies, mx *metrics) {
481 chart := c.charts.Get("etcd_object_counts")
482 if chart == nil {
483 chart = newEtcdObjectCountsChart()
484 if err := c.charts.Add(chart); err != nil {
485 c.Warningf("failed to add etcd object counts chart: %v", err)
486 }
487 }
488
489 // Try apiserver_storage_objects first (newer k8s), then etcd_object_counts (older)
490 mf := mfs.Get("apiserver_storage_objects")
491 if mf == nil {
492 mf = mfs.Get("etcd_object_counts")
493 }
494 if mf == nil {
495 return
496 }
497
498 for _, m := range mf.Metrics() {
499 value := metricValue(mf, m)
500 if math.IsNaN(value) {
501 continue
502 }
503
504 resource := m.Labels().Get("resource")
505 if resource == "" {
506 continue
507 }
508
509 resourceName := simplifyResourceName(resource)
510 resourceID := cleanID(resourceName)
511 dimID := "etcd_objects_" + resourceID
512
513 if chart != nil && !chart.HasDim(dimID) {
514 if err := chart.AddDim(&Dim{ID: dimID, Name: resourceName}); err != nil {
515 c.Debugf("failed to add etcd object dimension %s: %v", resourceName, err)
516 } else {
517 chart.MarkNotCreated()
518 }
519 }
520 mx.Etcd.ObjectCounts[resourceID] = mtx.Gauge(value)
521 }
522 }
523
524 // collectWorkqueues collects controller work queue metrics
525 func (c *Collector) collectWorkqueues(mfs prometheus.MetricFamilies, mx *metrics) {
526 // Pre-index metrics by queue name
527 depthByName := make(map[string]float64)
528 addsByName := make(map[string]float64)
529 retriesByName := make(map[string]float64)
530
531 if mf := mfs.Get("workqueue_depth"); mf != nil {
532 for _, m := range mf.Metrics() {
533 if name := m.Labels().Get("name"); name != "" {
534 if v := metricValue(mf, m); !math.IsNaN(v) {
535 depthByName[name] = v
536 }
537 }
538 }
539 }
540 if mf := mfs.Get("workqueue_adds_total"); mf != nil {
541 for _, m := range mf.Metrics() {
542 if name := m.Labels().Get("name"); name != "" {
543 if v := metricValue(mf, m); !math.IsNaN(v) {
544 addsByName[name] = v
545 }
546 }
547 }
548 }
549 if mf := mfs.Get("workqueue_retries_total"); mf != nil {
550 for _, m := range mf.Metrics() {
551 if name := m.Labels().Get("name"); name != "" {
552 if v := metricValue(mf, m); !math.IsNaN(v) {
553 retriesByName[name] = v
554 }
555 }
556 }
557 }
558
559 queueLatencyMF := mfs.Get("workqueue_queue_duration_seconds")
560 workDurationMF := mfs.Get("workqueue_work_duration_seconds")
561
562 // Merge all queue names from depth, adds, and retries to avoid missing workqueues
563 // that may not have depth metric but have other metrics
564 allQueueNames := make(map[string]struct{})
565 for name := range depthByName {
566 allQueueNames[name] = struct{}{}
567 }
568 for name := range addsByName {
569 allQueueNames[name] = struct{}{}
570 }
571 for name := range retriesByName {
572 allQueueNames[name] = struct{}{}
573 }
574
575 // Collect names into slice to avoid modifying maps during iteration
576 queueNames := make([]string, 0, len(allQueueNames))
577 for name := range allQueueNames {
578 queueNames = append(queueNames, name)
579 }
580
581 for _, queueName := range queueNames {
582 // Sanitize name for use in chart/dimension IDs
583 cleanName := cleanID(queueName)
584
585 // Cardinality limit: only accept new items if under limit
586 _, seen := c.collectedWorkqueues[cleanName]
587 if !seen && len(c.collectedWorkqueues) >= defaultMaxWorkqueues {
588 continue
589 }
590
591 // Track last-seen cycle
592 c.collectedWorkqueues[cleanName] = c.collectCycle
593
594 if !seen {
595 if err := c.charts.Add(newWorkqueueDepthChart(cleanName)); err != nil {
596 c.Warningf("failed to add workqueue depth chart %s: %v", queueName, err)
597 }
598 if err := c.charts.Add(newWorkqueueLatencyChart(cleanName)); err != nil {
599 c.Warningf("failed to add workqueue latency chart %s: %v", queueName, err)
600 }
601 if err := c.charts.Add(newWorkqueueAddsChart(cleanName)); err != nil {
602 c.Warningf("failed to add workqueue adds chart %s: %v", queueName, err)
603 }
604 if err := c.charts.Add(newWorkqueueDurationChart(cleanName)); err != nil {
605 c.Warningf("failed to add workqueue duration chart %s: %v", queueName, err)
606 }
607 }
608
609 if mx.Workqueue.Controllers[cleanName] == nil {
610 mx.Workqueue.Controllers[cleanName] = &workqueueMetrics{}
611 }
612
613 wq := mx.Workqueue.Controllers[cleanName]
614 wq.Depth.Set(depthByName[queueName])
615 wq.Adds.Set(addsByName[queueName])
616 wq.Retries.Set(retriesByName[queueName])
617
618 // Queue latency (histogram)
619 if queueLatencyMF != nil && queueLatencyMF.Type() == model.MetricTypeHistogram {
620 hd := collectHistogramBucketsFromMF(queueLatencyMF, func(m prometheus.Metric) bool {
621 return m.Labels().Get("name") == queueName
622 })
623 if len(hd.buckets) > 0 {
624 if p50 := histogramPercentile(hd, 0.5); !math.IsNaN(p50) {
625 wq.LatencyP50.Set(p50 * latencyPrecision)
626 }
627 if p90 := histogramPercentile(hd, 0.9); !math.IsNaN(p90) {
628 wq.LatencyP90.Set(p90 * latencyPrecision)
629 }
630 if p99 := histogramPercentile(hd, 0.99); !math.IsNaN(p99) {
631 wq.LatencyP99.Set(p99 * latencyPrecision)
632 }
633 }
634 }
635
636 // Work duration (histogram)
637 if workDurationMF != nil && workDurationMF.Type() == model.MetricTypeHistogram {
638 hd := collectHistogramBucketsFromMF(workDurationMF, func(m prometheus.Metric) bool {
639 return m.Labels().Get("name") == queueName
640 })
641 if len(hd.buckets) > 0 {
642 if p50 := histogramPercentile(hd, 0.5); !math.IsNaN(p50) {
643 wq.DurationP50.Set(p50 * latencyPrecision)
644 }
645 if p90 := histogramPercentile(hd, 0.9); !math.IsNaN(p90) {
646 wq.DurationP90.Set(p90 * latencyPrecision)
647 }
648 if p99 := histogramPercentile(hd, 0.99); !math.IsNaN(p99) {
649 wq.DurationP99.Set(p99 * latencyPrecision)
650 }
651 }
652 }
653 }
654 }
655
656 // collectProcess collects Go runtime and process metrics
657 func (c *Collector) collectProcess(mfs prometheus.MetricFamilies, mx *metrics) {
658 mx.Process.Goroutines.Set(getMaxValue(mfs, "go_goroutines"))
659 mx.Process.Threads.Set(getMaxValue(mfs, "go_threads"))
660 mx.Process.CPUSeconds.Set(getMaxValue(mfs, "process_cpu_seconds_total") * precision)
661 mx.Process.ResidentMemory.Set(getMaxValue(mfs, "process_resident_memory_bytes"))
662 mx.Process.VirtualMemory.Set(getMaxValue(mfs, "process_virtual_memory_bytes"))
663 mx.Process.OpenFDs.Set(getMaxValue(mfs, "process_open_fds"))
664 mx.Process.MaxFDs.Set(getMaxValue(mfs, "process_max_fds"))
665 mx.Process.HeapAlloc.Set(getMaxValue(mfs, "go_memstats_heap_alloc_bytes"))
666 mx.Process.HeapInuse.Set(getMaxValue(mfs, "go_memstats_heap_inuse_bytes"))
667 mx.Process.StackInuse.Set(getMaxValue(mfs, "go_memstats_stack_inuse_bytes"))
668
669 // GC duration (summary with quantile labels)
670 if mf := mfs.Get("go_gc_duration_seconds"); mf != nil && mf.Type() == model.MetricTypeSummary {
671 for _, m := range mf.Metrics() {
672 if m.Summary() == nil {
673 continue
674 }
675 for _, q := range m.Summary().Quantiles() {
676 if math.IsNaN(q.Value()) {
677 continue
678 }
679 switch q.Quantile() {
680 case 0:
681 mx.Process.GCDurationMin.Set(q.Value() * latencyPrecision)
682 case 0.25:
683 mx.Process.GCDurationP25.Set(q.Value() * latencyPrecision)
684 case 0.5:
685 mx.Process.GCDurationP50.Set(q.Value() * latencyPrecision)
686 case 0.75:
687 mx.Process.GCDurationP75.Set(q.Value() * latencyPrecision)
688 case 1:
689 mx.Process.GCDurationMax.Set(q.Value() * latencyPrecision)
690 }
691 }
692 }
693 }
694 }
695
696 // collectAudit collects audit event metrics
697 func (c *Collector) collectAudit(mfs prometheus.MetricFamilies, mx *metrics) {
698 mx.Audit.EventsTotal.Set(getMaxValue(mfs, "apiserver_audit_event_total"))
699 mx.Audit.RejectedTotal.Set(getMaxValue(mfs, "apiserver_audit_requests_rejected_total"))
700 }
701
702 // collectAuth collects authentication metrics
703 func (c *Collector) collectAuth(mfs prometheus.MetricFamilies, mx *metrics) {
704 // Sum across all usernames
705 if mf := mfs.Get("authenticated_user_requests"); mf != nil {
706 for _, m := range mf.Metrics() {
707 if v := metricValue(mf, m); !math.IsNaN(v) {
708 mx.Auth.AuthenticatedRequests.Add(v)
709 }
710 }
711 }
712
713 // Client certificate expiration - histogram, track count of certs expiring within 24h
714 if mf := mfs.Get("apiserver_client_certificate_expiration_seconds"); mf != nil && mf.Type() == model.MetricTypeHistogram {
715 for _, m := range mf.Metrics() {
716 if m.Histogram() == nil {
717 continue
718 }
719 for _, b := range m.Histogram().Buckets() {
720 if b.UpperBound() == 86400 { // 1 day bucket
721 mx.Auth.CertExpirationSeconds.Set(float64(b.CumulativeCount()))
722 break
723 }
724 }
725 }
726 }
727 }
728
729 // Helper functions for dynamic dimension creation
730
731 func (c *Collector) addVerbDimension(verb string) {
732 _, seen := c.collectedVerbs[verb]
733 c.collectedVerbs[verb] = c.collectCycle
734
735 if seen {
736 return
737 }
738
739 chart := c.charts.Get("requests_by_verb")
740 if chart == nil {
741 c.Warningf("chart 'requests_by_verb' not found, cannot add dimension for verb: %s", verb)
742 return
743 }
744 dimID := "request_by_verb_" + cleanID(verb)
745 if !chart.HasDim(dimID) {
746 if err := chart.AddDim(&Dim{ID: dimID, Name: verb, Algo: collectorapi.Incremental}); err != nil {
747 c.Warningf("failed to add verb dimension %s: %v", verb, err)
748 } else {
749 chart.MarkNotCreated()
750 }
751 }
752 }
753
754 func (c *Collector) addCodeDimension(code string) {
755 _, seen := c.collectedCodes[code]
756 c.collectedCodes[code] = c.collectCycle
757
758 if seen {
759 return
760 }
761
762 chart := c.charts.Get("requests_by_code")
763 if chart == nil {
764 c.Warningf("chart 'requests_by_code' not found, cannot add dimension for code: %s", code)
765 return
766 }
767 dimID := "request_by_code_" + cleanID(code)
768 if !chart.HasDim(dimID) {
769 if err := chart.AddDim(&Dim{ID: dimID, Name: code, Algo: collectorapi.Incremental}); err != nil {
770 c.Warningf("failed to add code dimension %s: %v", code, err)
771 } else {
772 chart.MarkNotCreated()
773 }
774 }
775 }
776
777 func (c *Collector) addResourceDimension(resource string) {
778 _, seen := c.collectedResources[resource]
779 // Note: cycle tracking for resources is done in collectRequests due to cardinality limit check
780 if seen {
781 return
782 }
783
784 chart := c.charts.Get("requests_by_resource")
785 if chart == nil {
786 c.Warningf("chart 'requests_by_resource' not found, cannot add dimension for resource: %s", resource)
787 return
788 }
789 dimID := "request_by_resource_" + cleanID(resource)
790 if !chart.HasDim(dimID) {
791 if err := chart.AddDim(&Dim{ID: dimID, Name: resource, Algo: collectorapi.Incremental}); err != nil {
792 c.Warningf("failed to add resource dimension %s: %v", resource, err)
793 } else {
794 chart.MarkNotCreated()
795 }
796 }
797 }
798
799 // simplifyResourceName removes API group suffix from resource names
800 func simplifyResourceName(resource string) string {
801 parts := strings.SplitN(resource, ".", 2)
802 return parts[0]
803 }
804
805 // metricValue extracts the value from a metric based on its type
806 func metricValue(mf *prometheus.MetricFamily, m prometheus.Metric) float64 {
807 switch mf.Type() {
808 case model.MetricTypeGauge:
809 if m.Gauge() != nil {
810 return m.Gauge().Value()
811 }
812 case model.MetricTypeCounter:
813 if m.Counter() != nil {
814 return m.Counter().Value()
815 }
816 case model.MetricTypeUnknown:
817 if m.Gauge() != nil {
818 return m.Gauge().Value()
819 }
820 if m.Counter() != nil {
821 return m.Counter().Value()
822 }
823 }
824 return math.NaN()
825 }
826
827 // getMaxValue gets the maximum value across all metrics in a family
828 func getMaxValue(mfs prometheus.MetricFamilies, name string) float64 {
829 mf := mfs.Get(name)
830 if mf == nil {
831 return 0
832 }
833
834 var maxVal float64
835 for _, m := range mf.Metrics() {
836 if v := metricValue(mf, m); !math.IsNaN(v) && v > maxVal {
837 maxVal = v
838 }
839 }
840 return maxVal
841 }
842
843 // Histogram percentile calculation utilities
844
845 type histogramBucket struct {
846 le float64
847 count float64
848 }
849
850 type histogramData struct {
851 buckets []histogramBucket
852 total float64 // from +Inf bucket
853 }
854
855 // collectHistogramBucketsFromMF collects histogram buckets from a metric family
856 func collectHistogramBucketsFromMF(mf *prometheus.MetricFamily, filter func(prometheus.Metric) bool) histogramData {
857 bucketMap := make(map[float64]float64)
858 var total float64
859
860 for _, m := range mf.Metrics() {
861 if filter != nil && !filter(m) {
862 continue
863 }
864 if m.Histogram() == nil {
865 continue
866 }
867
868 for _, b := range m.Histogram().Buckets() {
869 if math.IsInf(b.UpperBound(), 0) {
870 total += b.CumulativeCount()
871 continue
872 }
873 bucketMap[b.UpperBound()] += b.CumulativeCount()
874 }
875 }
876
877 buckets := make([]histogramBucket, 0, len(bucketMap))
878 for le, count := range bucketMap {
879 buckets = append(buckets, histogramBucket{le: le, count: count})
880 }
881
882 sortBuckets(buckets)
883 return histogramData{buckets: buckets, total: total}
884 }
885
886 func sortBuckets(buckets []histogramBucket) {
887 sort.Slice(buckets, func(i, j int) bool {
888 return buckets[i].le < buckets[j].le
889 })
890 }
891
892 // histogramPercentile estimates the percentile value from histogram buckets
893 // total should be from the +Inf bucket which contains the true total count
894 func histogramPercentile(hd histogramData, percentile float64) float64 {
895 if len(hd.buckets) == 0 || hd.total == 0 {
896 return math.NaN()
897 }
898
899 target := percentile * hd.total
900
901 var prevBound, prevCount float64
902 for _, b := range hd.buckets {
903 if b.count >= target {
904 bucketWidth := b.le - prevBound
905 bucketCount := b.count - prevCount
906 if bucketCount == 0 {
907 return b.le
908 }
909 fraction := (target - prevCount) / bucketCount
910 return prevBound + fraction*bucketWidth
911 }
912 prevBound = b.le
913 prevCount = b.count
914 }
915
916 return hd.buckets[len(hd.buckets)-1].le
917 }
918
919 // bucketSetter is implemented by admission controller and webhook metrics
920 type bucketSetter interface {
921 setBuckets(b5ms, b25ms, b100ms, b500ms, b1s, b2500ms, bInf float64)
922 }
923
924 func (m *admissionControllerMetrics) setBuckets(b5ms, b25ms, b100ms, b500ms, b1s, b2500ms, bInf float64) {
925 m.Bucket5ms.Set(b5ms)
926 m.Bucket25ms.Set(b25ms)
927 m.Bucket100ms.Set(b100ms)
928 m.Bucket500ms.Set(b500ms)
929 m.Bucket1s.Set(b1s)
930 m.Bucket2500ms.Set(b2500ms)
931 m.BucketInf.Set(bInf)
932 }
933
934 func (m *admissionWebhookMetrics) setBuckets(b5ms, b25ms, b100ms, b500ms, b1s, b2500ms, bInf float64) {
935 m.Bucket5ms.Set(b5ms)
936 m.Bucket25ms.Set(b25ms)
937 m.Bucket100ms.Set(b100ms)
938 m.Bucket500ms.Set(b500ms)
939 m.Bucket1s.Set(b1s)
940 m.Bucket2500ms.Set(b2500ms)
941 m.BucketInf.Set(bInf)
942 }
943
944 // setAdmissionBucketMetrics extracts bucket values and sets them on the metrics
945 // Converts Prometheus cumulative buckets to non-cumulative for heatmap display
946 // Uses max(0, diff) to protect against negative values from counter resets
947 func (c *Collector) setAdmissionBucketMetrics(bucketMap map[float64]float64, m bucketSetter) {
948 // Validate bucket presence and log warnings for missing buckets
949 var missingBuckets []string
950 for _, bound := range admissionBucketBounds {
951 if _, ok := bucketMap[bound]; !ok {
952 missingBuckets = append(missingBuckets, formatBucketBound(bound))
953 }
954 }
955 if _, ok := bucketMap[math.Inf(1)]; !ok {
956 missingBuckets = append(missingBuckets, "+Inf")
957 }
958 if len(missingBuckets) > 0 {
959 c.Debugf("missing histogram buckets: %v", missingBuckets)
960 }
961
962 // Extract cumulative bucket values (0 if missing)
963 b5ms := bucketMap[admissionBucketBounds[0]]
964 b25ms := bucketMap[admissionBucketBounds[1]]
965 b100ms := bucketMap[admissionBucketBounds[2]]
966 b500ms := bucketMap[admissionBucketBounds[3]]
967 b1s := bucketMap[admissionBucketBounds[4]]
968 b2500ms := bucketMap[admissionBucketBounds[5]]
969 bInf := bucketMap[math.Inf(1)]
970
971 // Convert cumulative to non-cumulative (differential) bucket counts
972 // Use max(0, diff) to handle Prometheus counter resets gracefully
973 // When a counter resets, the cumulative value decreases, causing negative diffs
974 m.setBuckets(
975 b5ms,
976 math.Max(0, b25ms-b5ms),
977 math.Max(0, b100ms-b25ms),
978 math.Max(0, b500ms-b100ms),
979 math.Max(0, b1s-b500ms),
980 math.Max(0, b2500ms-b1s),
981 math.Max(0, bInf-b2500ms),
982 )
983 }
984
985 // formatBucketBound formats a bucket bound for logging
986 func formatBucketBound(bound float64) string {
987 if bound < 1 {
988 return fmt.Sprintf("%.0fms", bound*1000)
989 }
990 return fmt.Sprintf("%.1fs", bound)
991 }
992
993 // cleanupStaleDimensions removes dimensions that haven't been seen for staleThresholdCycles
994 func (c *Collector) cleanupStaleDimensions() {
995 threshold := c.collectCycle - staleThresholdCycles
996
997 // Cleanup resources
998 for name, lastSeen := range c.collectedResources {
999 if lastSeen < threshold {
1000 delete(c.collectedResources, name)
1001 if chart := c.charts.Get("requests_by_resource"); chart != nil {
1002 dimID := "request_by_resource_" + cleanID(name)
1003 _ = chart.RemoveDim(dimID)
1004 chart.MarkNotCreated()
1005 }
1006 c.Debugf("removed stale resource dimension: %s", name)
1007 }
1008 }
1009
1010 // Cleanup verbs
1011 for name, lastSeen := range c.collectedVerbs {
1012 if lastSeen < threshold {
1013 delete(c.collectedVerbs, name)
1014 if chart := c.charts.Get("requests_by_verb"); chart != nil {
1015 dimID := "request_by_verb_" + cleanID(name)
1016 _ = chart.RemoveDim(dimID)
1017 chart.MarkNotCreated()
1018 }
1019 c.Debugf("removed stale verb dimension: %s", name)
1020 }
1021 }
1022
1023 // Cleanup codes
1024 for name, lastSeen := range c.collectedCodes {
1025 if lastSeen < threshold {
1026 delete(c.collectedCodes, name)
1027 if chart := c.charts.Get("requests_by_code"); chart != nil {
1028 dimID := "request_by_code_" + cleanID(name)
1029 _ = chart.RemoveDim(dimID)
1030 chart.MarkNotCreated()
1031 }
1032 c.Debugf("removed stale code dimension: %s", name)
1033 }
1034 }
1035
1036 // Cleanup REST client codes
1037 for name, lastSeen := range c.collectedRESTCodes {
1038 if lastSeen < threshold {
1039 delete(c.collectedRESTCodes, name)
1040 if chart := c.charts.Get("rest_client_requests_by_code"); chart != nil {
1041 dimID := "rest_client_by_code_" + cleanID(name)
1042 _ = chart.RemoveDim(dimID)
1043 chart.MarkNotCreated()
1044 }
1045 c.Debugf("removed stale REST client code dimension: %s", name)
1046 }
1047 }
1048
1049 // Cleanup REST client methods
1050 for name, lastSeen := range c.collectedRESTMethods {
1051 if lastSeen < threshold {
1052 delete(c.collectedRESTMethods, name)
1053 if chart := c.charts.Get("rest_client_requests_by_method"); chart != nil {
1054 dimID := "rest_client_by_method_" + cleanID(name)
1055 _ = chart.RemoveDim(dimID)
1056 chart.MarkNotCreated()
1057 }
1058 c.Debugf("removed stale REST client method dimension: %s", name)
1059 }
1060 }
1061
1062 // Cleanup workqueues (remove charts)
1063 for name, lastSeen := range c.collectedWorkqueues {
1064 if lastSeen < threshold {
1065 delete(c.collectedWorkqueues, name)
1066 _ = c.charts.Remove("workqueue_depth_" + cleanID(name))
1067 _ = c.charts.Remove("workqueue_latency_" + cleanID(name))
1068 _ = c.charts.Remove("workqueue_adds_" + cleanID(name))
1069 _ = c.charts.Remove("workqueue_duration_" + cleanID(name))
1070 c.Debugf("removed stale workqueue charts: %s", name)
1071 }
1072 }
1073
1074 // Cleanup admission controllers (remove charts)
1075 for name, lastSeen := range c.collectedAdmissionCtrl {
1076 if lastSeen < threshold {
1077 delete(c.collectedAdmissionCtrl, name)
1078 _ = c.charts.Remove("admission_controller_latency_" + cleanID(name))
1079 c.Debugf("removed stale admission controller chart: %s", name)
1080 }
1081 }
1082
1083 // Cleanup admission webhooks (remove charts)
1084 for name, lastSeen := range c.collectedAdmissionWH {
1085 if lastSeen < threshold {
1086 delete(c.collectedAdmissionWH, name)
1087 _ = c.charts.Remove("admission_webhook_latency_" + cleanID(name))
1088 c.Debugf("removed stale admission webhook chart: %s", name)
1089 }
1090 }
1091 }