apiserver/pkg/util/flowcontrol/metrics/sample_and_watermark.go

/*
Copyright 2019 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package metrics

import (
	"sync"
	"time"

	"k8s.io/apimachinery/pkg/util/clock"
	compbasemetrics "k8s.io/component-base/metrics"
	"k8s.io/klog/v2"
)

const (
	labelNameMark       = "mark"
	labelValueLo        = "low"
	labelValueHi        = "high"
	labelNamePhase      = "phase"
	labelValueWaiting   = "waiting"
	labelValueExecuting = "executing"
)

// SampleAndWaterMarkPairGenerator makes pairs of TimedObservers that
// track samples and watermarks.
type SampleAndWaterMarkPairGenerator struct {
	urGenerator SampleAndWaterMarkObserverGenerator
}

var _ TimedObserverPairGenerator = SampleAndWaterMarkPairGenerator{}

// NewSampleAndWaterMarkHistogramsPairGenerator makes a new pair generator
func NewSampleAndWaterMarkHistogramsPairGenerator(clock clock.PassiveClock, samplePeriod time.Duration, sampleOpts, waterMarkOpts *compbasemetrics.HistogramOpts, labelNames []string) SampleAndWaterMarkPairGenerator {
	return SampleAndWaterMarkPairGenerator{
		urGenerator: NewSampleAndWaterMarkHistogramsGenerator(clock, samplePeriod, sampleOpts, waterMarkOpts, append([]string{labelNamePhase}, labelNames...)),
	}
}

// Generate makes a new pair
func (spg SampleAndWaterMarkPairGenerator) Generate(waiting1, executing1 float64, labelValues []string) TimedObserverPair {
	return TimedObserverPair{
		RequestsWaiting:   spg.urGenerator.Generate(0, waiting1, append([]string{labelValueWaiting}, labelValues...)),
		RequestsExecuting: spg.urGenerator.Generate(0, executing1, append([]string{labelValueExecuting}, labelValues...)),
	}
}

func (spg SampleAndWaterMarkPairGenerator) metrics() Registerables {
	return spg.urGenerator.metrics()
}

// SampleAndWaterMarkObserverGenerator creates TimedObservers that
// populate histograms of samples and low- and high-water-marks.  The
// generator has a samplePeriod, and the histograms get an observation
// every samplePeriod.
type SampleAndWaterMarkObserverGenerator struct {
	*sampleAndWaterMarkObserverGenerator
}

type sampleAndWaterMarkObserverGenerator struct {
	clock        clock.PassiveClock
	samplePeriod time.Duration
	samples      *compbasemetrics.HistogramVec
	waterMarks   *compbasemetrics.HistogramVec
}

var _ TimedObserverGenerator = (*sampleAndWaterMarkObserverGenerator)(nil)

// NewSampleAndWaterMarkHistogramsGenerator makes a new one
func NewSampleAndWaterMarkHistogramsGenerator(clock clock.PassiveClock, samplePeriod time.Duration, sampleOpts, waterMarkOpts *compbasemetrics.HistogramOpts, labelNames []string) SampleAndWaterMarkObserverGenerator {
	return SampleAndWaterMarkObserverGenerator{
		&sampleAndWaterMarkObserverGenerator{
			clock:        clock,
			samplePeriod: samplePeriod,
			samples:      compbasemetrics.NewHistogramVec(sampleOpts, labelNames),
			waterMarks:   compbasemetrics.NewHistogramVec(waterMarkOpts, append([]string{labelNameMark}, labelNames...)),
		}}
}

func (swg *sampleAndWaterMarkObserverGenerator) quantize(when time.Time) int64 {
	return when.UnixNano() / int64(swg.samplePeriod)
}

// Generate makes a new TimedObserver
func (swg *sampleAndWaterMarkObserverGenerator) Generate(x, x1 float64, labelValues []string) TimedObserver {
	relX := x / x1
	when := swg.clock.Now()
	return &sampleAndWaterMarkHistograms{
		sampleAndWaterMarkObserverGenerator: swg,
		labelValues:                         labelValues,
		loLabelValues:                       append([]string{labelValueLo}, labelValues...),
		hiLabelValues:                       append([]string{labelValueHi}, labelValues...),
		x1:                                  x1,
		sampleAndWaterMarkAccumulator: sampleAndWaterMarkAccumulator{
			lastSet:    when,
			lastSetInt: swg.quantize(when),
			x:          x,
			relX:       relX,
			loRelX:     relX,
			hiRelX:     relX,
		}}
}

func (swg *sampleAndWaterMarkObserverGenerator) metrics() Registerables {
	return Registerables{swg.samples, swg.waterMarks}
}

type sampleAndWaterMarkHistograms struct {
	*sampleAndWaterMarkObserverGenerator
	labelValues                  []string
	loLabelValues, hiLabelValues []string

	sync.Mutex
	x1 float64
	sampleAndWaterMarkAccumulator
}

type sampleAndWaterMarkAccumulator struct {
	lastSet        time.Time
	lastSetInt     int64 // lastSet / samplePeriod
	x              float64
	relX           float64 // x / x1
	loRelX, hiRelX float64
}

var _ TimedObserver = (*sampleAndWaterMarkHistograms)(nil)

func (saw *sampleAndWaterMarkHistograms) Add(deltaX float64) {
	saw.innerSet(func() {
		saw.x += deltaX
	})
}

func (saw *sampleAndWaterMarkHistograms) Set(x float64) {
	saw.innerSet(func() {
		saw.x = x
	})
}

func (saw *sampleAndWaterMarkHistograms) SetX1(x1 float64) {
	saw.innerSet(func() {
		saw.x1 = x1
	})
}

func (saw *sampleAndWaterMarkHistograms) innerSet(updateXOrX1 func()) {
	saw.Lock()
	when := saw.clock.Now()
	whenInt := saw.quantize(when)
	acc := saw.sampleAndWaterMarkAccumulator
	wellOrdered := !when.Before(acc.lastSet)
	if wellOrdered {
		updateXOrX1()
		saw.relX = saw.x / saw.x1
		if acc.lastSetInt < whenInt {
			saw.loRelX, saw.hiRelX = acc.relX, acc.relX
			saw.lastSetInt = whenInt
		}
		if saw.relX < saw.loRelX {
			saw.loRelX = saw.relX
		} else if saw.relX > saw.hiRelX {
			saw.hiRelX = saw.relX
		}
		saw.lastSet = when
	}
	saw.Unlock()
	if !wellOrdered {
		lastSetS := acc.lastSet.Format(time.RFC3339Nano)
		whenS := when.Format(time.RFC3339Nano)
		klog.Fatalf("Time went backwards from %s to %s for labelValues=%#+v", lastSetS, whenS, saw.labelValues)
		panic(append([]string{lastSetS, whenS}, saw.labelValues...))
	}
	for acc.lastSetInt < whenInt {
		saw.samples.WithLabelValues(saw.labelValues...).Observe(acc.relX)
		saw.waterMarks.WithLabelValues(saw.loLabelValues...).Observe(acc.loRelX)
		saw.waterMarks.WithLabelValues(saw.hiLabelValues...).Observe(acc.hiRelX)
		acc.lastSetInt++
		acc.loRelX, acc.hiRelX = acc.relX, acc.relX
	}
}