2 Copyright 2016 The Kubernetes Authors.
4 Licensed under the Apache License, Version 2.0 (the "License");
5 you may not use this file except in compliance with the License.
6 You may obtain a copy of the License at
8 http://www.apache.org/licenses/LICENSE-2.0
10 Unless required by applicable law or agreed to in writing, software
11 distributed under the License is distributed on an "AS IS" BASIS,
12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 See the License for the specific language governing permissions and
14 limitations under the License.
23 "k8s.io/apimachinery/pkg/util/clock"
26 // This file provides abstractions for setting the provider (e.g., prometheus)
29 type queueMetrics interface {
33 updateUnfinishedWork()
36 // GaugeMetric represents a single numerical value that can arbitrarily go up
38 type GaugeMetric interface {
43 // SettableGaugeMetric represents a single numerical value that can arbitrarily go up
44 // and down. (Separate from GaugeMetric to preserve backwards compatibility.)
45 type SettableGaugeMetric interface {
49 // CounterMetric represents a single numerical value that only ever
51 type CounterMetric interface {
55 // SummaryMetric captures individual observations.
56 type SummaryMetric interface {
60 type noopMetric struct{}
62 func (noopMetric) Inc() {}
63 func (noopMetric) Dec() {}
64 func (noopMetric) Set(float64) {}
65 func (noopMetric) Observe(float64) {}
67 // defaultQueueMetrics expects the caller to lock before setting any metrics.
68 type defaultQueueMetrics struct {
71 // current depth of a workqueue
73 // total number of adds handled by a workqueue
75 // how long an item stays in a workqueue
77 // how long processing an item from a workqueue takes
78 workDuration SummaryMetric
79 addTimes map[t]time.Time
80 processingStartTimes map[t]time.Time
82 // how long have current threads been working?
83 unfinishedWorkSeconds SettableGaugeMetric
84 longestRunningProcessor SettableGaugeMetric
87 func (m *defaultQueueMetrics) add(item t) {
94 if _, exists := m.addTimes[item]; !exists {
95 m.addTimes[item] = m.clock.Now()
99 func (m *defaultQueueMetrics) get(item t) {
105 m.processingStartTimes[item] = m.clock.Now()
106 if startTime, exists := m.addTimes[item]; exists {
107 m.latency.Observe(m.sinceInMicroseconds(startTime))
108 delete(m.addTimes, item)
112 func (m *defaultQueueMetrics) done(item t) {
117 if startTime, exists := m.processingStartTimes[item]; exists {
118 m.workDuration.Observe(m.sinceInMicroseconds(startTime))
119 delete(m.processingStartTimes, item)
123 func (m *defaultQueueMetrics) updateUnfinishedWork() {
124 // Note that a summary metric would be better for this, but prometheus
125 // doesn't seem to have non-hacky ways to reset the summary metrics.
128 for _, t := range m.processingStartTimes {
129 age := m.sinceInMicroseconds(t)
135 // Convert to seconds; microseconds is unhelpfully granular for this.
137 m.unfinishedWorkSeconds.Set(total)
138 m.longestRunningProcessor.Set(oldest) // in microseconds.
141 type noMetrics struct{}
143 func (noMetrics) add(item t) {}
144 func (noMetrics) get(item t) {}
145 func (noMetrics) done(item t) {}
146 func (noMetrics) updateUnfinishedWork() {}
148 // Gets the time since the specified start in microseconds.
149 func (m *defaultQueueMetrics) sinceInMicroseconds(start time.Time) float64 {
150 return float64(m.clock.Since(start).Nanoseconds() / time.Microsecond.Nanoseconds())
153 type retryMetrics interface {
157 type defaultRetryMetrics struct {
158 retries CounterMetric
161 func (m *defaultRetryMetrics) retry() {
169 // MetricsProvider generates various metrics used by the queue.
170 type MetricsProvider interface {
171 NewDepthMetric(name string) GaugeMetric
172 NewAddsMetric(name string) CounterMetric
173 NewLatencyMetric(name string) SummaryMetric
174 NewWorkDurationMetric(name string) SummaryMetric
175 NewUnfinishedWorkSecondsMetric(name string) SettableGaugeMetric
176 NewLongestRunningProcessorMicrosecondsMetric(name string) SettableGaugeMetric
177 NewRetriesMetric(name string) CounterMetric
180 type noopMetricsProvider struct{}
182 func (_ noopMetricsProvider) NewDepthMetric(name string) GaugeMetric {
186 func (_ noopMetricsProvider) NewAddsMetric(name string) CounterMetric {
190 func (_ noopMetricsProvider) NewLatencyMetric(name string) SummaryMetric {
194 func (_ noopMetricsProvider) NewWorkDurationMetric(name string) SummaryMetric {
198 func (_ noopMetricsProvider) NewUnfinishedWorkSecondsMetric(name string) SettableGaugeMetric {
202 func (_ noopMetricsProvider) NewLongestRunningProcessorMicrosecondsMetric(name string) SettableGaugeMetric {
206 func (_ noopMetricsProvider) NewRetriesMetric(name string) CounterMetric {
210 var globalMetricsFactory = queueMetricsFactory{
211 metricsProvider: noopMetricsProvider{},
214 type queueMetricsFactory struct {
215 metricsProvider MetricsProvider
220 func (f *queueMetricsFactory) setProvider(mp MetricsProvider) {
221 f.onlyOnce.Do(func() {
222 f.metricsProvider = mp
226 func (f *queueMetricsFactory) newQueueMetrics(name string, clock clock.Clock) queueMetrics {
227 mp := f.metricsProvider
228 if len(name) == 0 || mp == (noopMetricsProvider{}) {
231 return &defaultQueueMetrics{
233 depth: mp.NewDepthMetric(name),
234 adds: mp.NewAddsMetric(name),
235 latency: mp.NewLatencyMetric(name),
236 workDuration: mp.NewWorkDurationMetric(name),
237 unfinishedWorkSeconds: mp.NewUnfinishedWorkSecondsMetric(name),
238 longestRunningProcessor: mp.NewLongestRunningProcessorMicrosecondsMetric(name),
239 addTimes: map[t]time.Time{},
240 processingStartTimes: map[t]time.Time{},
244 func newRetryMetrics(name string) retryMetrics {
245 var ret *defaultRetryMetrics
249 return &defaultRetryMetrics{
250 retries: globalMetricsFactory.metricsProvider.NewRetriesMetric(name),
254 // SetProvider sets the metrics provider for all subsequently created work
255 // queues. Only the first call has an effect.
256 func SetProvider(metricsProvider MetricsProvider) {
257 globalMetricsFactory.setProvider(metricsProvider)