Files
zeekayandHanzo Dev e1280967b6 Give the last two payloads a wire to state
A prometheus Data held its labels in a map, and control.ExecArgs held the
live kernel objects an exec runs against -- a mount namespace, an fd
table, a limit set. Neither has offsets, so neither payload could state
a layout.

Labels become a list, which also retires the O(n^2) walk that existed to
read a map in sorted order. The kernel objects move to control.Env, which
is what the sentry supplies and what a call arriving on the wire cannot:
a namespace is a reference, not a value.

Every urpc payload now states its own wire.


Co-authored-by: Hanzo Dev <[email protected]>
2026-09-22 23:10:33 -07:00

1675 lines
58 KiB
Go

// Copyright 2018 The gVisor 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 metric provides primitives for collecting metrics.
package metric
import (
"errors"
"fmt"
"math"
re "regexp"
"sort"
"strings"
"time"
"google.golang.org/protobuf/types/known/timestamppb"
"gvisor.dev/gvisor/pkg/atomicbitops"
"gvisor.dev/gvisor/pkg/eventchannel"
"gvisor.dev/gvisor/pkg/log"
pb "gvisor.dev/gvisor/pkg/metric/metric_go_proto"
"gvisor.dev/gvisor/pkg/prometheus"
"gvisor.dev/gvisor/pkg/sync"
)
var (
// ErrNameInUse indicates that another metric is already defined for
// the given name.
ErrNameInUse = errors.New("metric name already in use")
// ErrInitializationDone indicates that the caller tried to create a
// new metric after initialization.
ErrInitializationDone = errors.New("metric cannot be created after initialization is complete")
// ErrFieldValueContainsIllegalChar indicates that the value of a metric
// field had an invalid character in it.
ErrFieldValueContainsIllegalChar = errors.New("metric field value contains illegal character")
// ErrFieldHasNoAllowedValues indicates that the field needs to define some
// allowed values to be a valid and useful field.
ErrFieldHasNoAllowedValues = errors.New("metric field does not define any allowed values")
// ErrTooManyFieldCombinations indicates that the number of unique
// combinations of fields is too large to support.
ErrTooManyFieldCombinations = errors.New("metric has too many combinations of allowed field values")
)
// Weirdness metric type constants.
var (
WeirdnessTypeTimeFallback = FieldValue{"time_fallback"}
WeirdnessTypePartialResult = FieldValue{"partial_result"}
WeirdnessTypeVsyscallCount = FieldValue{"vsyscall_count"}
WeirdnessTypeWatchdogStuckStartup = FieldValue{"watchdog_stuck_startup"}
WeirdnessTypeWatchdogStuckTasks = FieldValue{"watchdog_stuck_tasks"}
)
// Suspicious operations metric type constants.
var (
SuspiciousOperationsTypeOpenedWriteExecuteFile = FieldValue{"opened_write_execute_file"}
)
// List of global metrics that are used in multiple places.
var (
// WeirdnessMetric is a metric with fields created to track the number
// of weird occurrences such as time fallback, partial_result, vsyscall
// count, watchdog startup timeouts and stuck tasks.
WeirdnessMetric = MustCreateNewUint64Metric(
"/weirdness",
Uint64Metadata{
Cumulative: true,
Sync: true,
Description: "Increment for weird occurrences of problems such as time fallback, partial result, vsyscalls invoked in the sandbox, watchdog startup timeouts and stuck tasks.",
Fields: []Field{
NewField("weirdness_type",
&WeirdnessTypeTimeFallback,
&WeirdnessTypePartialResult,
&WeirdnessTypeVsyscallCount,
&WeirdnessTypeWatchdogStuckStartup,
&WeirdnessTypeWatchdogStuckTasks,
),
},
})
// SuspiciousOperationsMetric is a metric with fields created to detect
// operations such as opening an executable file to write from a gofer.
SuspiciousOperationsMetric = MustCreateNewUint64Metric(
"/suspicious_operations",
Uint64Metadata{
Cumulative: true,
Sync: true,
Description: "Increment for suspicious operations such as opening an executable file to write from a gofer.",
Fields: []Field{
NewField("operation_type",
&SuspiciousOperationsTypeOpenedWriteExecuteFile,
),
},
})
SentryEntryPointMetric = MustCreateNewUint64Metric(
"/sentry_entry_point",
Uint64Metadata{
Cumulative: true,
Sync: true,
Description: "The method that was used to create the sentry process.",
Fields: NewFieldsEntryPointType(),
},
)
)
// InitStage is the name of a Sentry initialization stage.
type InitStage string
// List of all Sentry initialization stages.
var (
InitRestoreConfig InitStage = "restore_config"
InitExecConfig InitStage = "exec_config"
InitRestore InitStage = "restore"
InitCreateProcess InitStage = "create_process"
InitTaskStart InitStage = "task_start"
// allStages is the list of allowed stages.
allStages = []InitStage{
InitRestoreConfig,
InitExecConfig,
InitRestore,
InitCreateProcess,
InitTaskStart,
}
)
// Uint64Metric encapsulates a uint64 that represents some kind of metric to be
// monitored.
//
// Metrics are not saved across save/restore and thus reset to zero on restore.
type Uint64Metric struct {
name string
// fields is the map of field-value combination index keys to Uint64 counters.
fields []atomicbitops.Uint64
// fieldMapper is used to generate index keys for the fields array (above)
// based on field value combinations, and vice-versa.
fieldMapper fieldMapper
}
var (
// initialized indicates that all metrics are registered. allMetrics is
// immutable once initialized is true.
initialized atomicbitops.Bool
// allMetrics are the registered metrics.
allMetrics = makeMetricSet()
)
// Initialize sends a metric registration event over the event channel.
//
// Precondition:
// - All metrics are registered.
// - Initialize/Disable has not been called.
func Initialize() error {
if initialized.Load() {
return errors.New("metric.Initialize called after metric.Initialize or metric.Disable")
}
m := pb.MetricRegistration{}
for _, v := range allMetrics.uint64Metrics {
m.Metrics = append(m.Metrics, v.metadata)
}
for _, v := range allMetrics.distributionMetrics {
m.Metrics = append(m.Metrics, v.metadata)
}
m.Stages = make([]string, 0, len(allStages))
for _, s := range allStages {
m.Stages = append(m.Stages, string(s))
}
allMetrics.registration = &m
if err := eventchannel.Emit(&m); err != nil {
return fmt.Errorf("unable to emit metric initialize event: %w", err)
}
if initialized.Swap(true) {
return errors.New("raced with another call to metric.Initialize or metric.Disable")
}
return nil
}
// ErrNotYetInitialized is returned by GetMetricRegistration if metrics are not yet initialized.
var ErrNotYetInitialized = errors.New("metrics are not yet initialized")
// GetMetricRegistration returns the metric registration data for all registered metrics.
// Must be called after Initialize().
// Returns ErrNotYetInitialized if metrics are not yet initialized.
func GetMetricRegistration() (*pb.MetricRegistration, error) {
if !initialized.Load() {
return nil, ErrNotYetInitialized
}
if allMetrics.registration == nil {
return nil, errors.New("metrics are disabled")
}
return allMetrics.registration, nil
}
// Disable sends an empty metric registration event over the event channel,
// disabling metric collection.
//
// Precondition:
// - All metrics are registered.
// - Initialize/Disable has not been called.
func Disable() error {
if initialized.Load() {
return errors.New("metric.Disable called after metric.Initialize or metric.Disable")
}
m := pb.MetricRegistration{}
if err := eventchannel.Emit(&m); err != nil {
return fmt.Errorf("unable to emit empty metric registration event (metrics disabled): %w", err)
}
if initialized.Swap(true) {
return errors.New("raced with another call to metric.Initialize or metric.Disable")
}
return nil
}
// Uint64Metadata is the metadata for a uint64 metric.
type Uint64Metadata struct {
Cumulative bool
Sync bool
Unit pb.MetricMetadata_Units
Description string
Fields []Field
}
type customUint64Metric struct {
// metadata describes the metric. It is immutable.
metadata *pb.MetricMetadata
// prometheusMetric describes the metric in Prometheus format. It is immutable.
prometheusMetric *prometheus.Metric
// fields is the set of fields of the metric.
fields []Field
// fieldMapper is used to generate index keys for the fields array (above)
// based on field value combinations, and vice-versa.
fieldMapper fieldMapper
// value returns the current value of the metric for the given set of
// fields. It takes a variadic number of field values as argument.
value func(fieldValues ...*FieldValue) uint64
// forEachNonZero calls the given function on each possible field value of
// the metric where the metric's value is non-zero.
// The passed-in function should not allocate new memory, and may not save
// or modify `fields` directly, as the slice memory is reused across calls.
// `forEachNonZero` does not guarantee that it will be called on a
// consistent snapshot of this metric's values.
// `forEachNonZero` may be nil.
forEachNonZero func(f func(fields []*FieldValue, val uint64))
}
// FieldValue is a string that can be used as a value for a Field.
// It must be referred to by address when the Field is created and when its
// metric value is modified. This ensures that the same FieldValue reference
// is used, which in turn enables the metric code to use the address of a
// FieldValue as comparison operator, rather than doing string comparisons.
type FieldValue struct {
Value string
}
// fieldMapperMapThreshold is the number of field values after which we switch
// to using map lookups when looking up field values.
// This value was determined using benchmarks to see which is fastest.
const fieldMapperMapThreshold = 48
// Field contains the field name and allowed values for the metric which is
// used in registration of the metric.
type Field struct {
// name is the metric field name.
name string
// values is the list of values for the field.
// `values` is always populated but not always used for lookup. It depends
// on the number of allowed field values. `values` is used for lookups on
// fields with small numbers of field values.
values []*FieldValue
// valuesPtrMap is a map version of `values`. For each item in `values`,
// its pointer is mapped to its index within `values`.
// `valuesPtrMap` is used for fields with large numbers of possible values.
// For fields with small numbers of field values, it is nil.
// This map allows doing faster string matching than a normal string map,
// as it avoids the string hashing step that normal string maps need to do.
valuesPtrMap map[*FieldValue]int
}
// toProto returns the proto definition of this field, for use in metric
// metadata.
func (f Field) toProto() *pb.MetricMetadata_Field {
allowedValues := make([]string, len(f.values))
for i, v := range f.values {
allowedValues[i] = v.Value
}
return &pb.MetricMetadata_Field{
FieldName: f.name,
AllowedValues: allowedValues,
}
}
// NewField defines a new Field that can be used to break down a metric.
// The set of allowedValues must be unique strings wrapped with `FieldValue`.
// The *same* `FieldValue` pointers must be used during metric modifications.
// In practice, in most cases, this means you should declare these
// `FieldValue`s as package-level `var`s, and always use the address of these
// package-level `var`s during metric modifications.
func NewField(name string, allowedValues ...*FieldValue) Field {
// Verify that all string values have a unique value.
strMap := make(map[string]bool, len(allowedValues))
for _, v := range allowedValues {
if strMap[v.Value] {
panic(fmt.Sprintf("found duplicate field value: %q", v))
}
strMap[v.Value] = true
}
if useMap := len(allowedValues) > fieldMapperMapThreshold; !useMap {
return Field{
name: name,
values: allowedValues,
}
}
valuesPtrMap := make(map[*FieldValue]int, len(allowedValues))
for i, v := range allowedValues {
valuesPtrMap[v] = i
}
return Field{
name: name,
values: allowedValues,
valuesPtrMap: valuesPtrMap,
}
}
// fieldMapper provides multi-dimensional fields to a single unique integer key
type fieldMapper struct {
// fields is a list of Field objects, which importantly include individual
// Field names which are used to perform the keyToMultiField function; and
// allowedValues for each field type which are used to perform the lookup
// function.
fields []Field
// numFieldCombinations is the number of unique keys for all possible field
// combinations.
numFieldCombinations int
}
// newFieldMapper returns a new fieldMapper for the given set of fields.
func newFieldMapper(fields ...Field) (fieldMapper, error) {
numFieldCombinations := 1
for _, f := range fields {
// Disallow fields with no possible values. We could also ignore them
// instead, but passing in a no-allowed-values field is probably a mistake.
if len(f.values) == 0 {
return fieldMapper{nil, 0}, ErrFieldHasNoAllowedValues
}
numFieldCombinations *= len(f.values)
// Sanity check, could be useful in case someone dynamically generates too
// many fields accidentally.
if numFieldCombinations > math.MaxUint32 || numFieldCombinations < 0 {
return fieldMapper{nil, 0}, ErrTooManyFieldCombinations
}
}
return fieldMapper{
fields: fields,
numFieldCombinations: numFieldCombinations,
}, nil
}
// lookupSingle looks up a single key for a single field within fieldMapper.
// It is used internally within lookupConcat.
// It returns the updated `idx` and `remainingCombinationBucket` values.
// +checkescape:all
//
//go:nosplit
func (m *fieldMapper) lookupSingle(fieldIndex int, fieldValue *FieldValue, idx, remainingCombinationBucket int) (int, int) {
field := m.fields[fieldIndex]
numValues := len(field.values)
// Are we doing a linear search?
if field.valuesPtrMap == nil {
// We scan by pointers only. This means the caller must pass the same
// FieldValue pointer as the one used in `NewField`.
for valIdx, allowedVal := range field.values {
if fieldValue == allowedVal {
remainingCombinationBucket /= numValues
idx += remainingCombinationBucket * valIdx
return idx, remainingCombinationBucket
}
}
panic("invalid field value or did not reuse the same FieldValue pointer as passed in NewField")
}
// Match using FieldValue pointer.
// This avoids the string hashing step that string maps otherwise do.
valIdx, found := field.valuesPtrMap[fieldValue]
if found {
remainingCombinationBucket /= numValues
idx += remainingCombinationBucket * valIdx
return idx, remainingCombinationBucket
}
panic("invalid field value or did not reuse the same FieldValue pointer as passed in NewField")
}
// lookupConcat looks up a key within the fieldMapper where the fields are
// the concatenation of two list of fields.
// The returned key is an index that can be used to access to map created by
// makeMap().
// This *must* be called with the correct number of fields, or it will panic.
// +checkescape:all
//
//go:nosplit
func (m *fieldMapper) lookupConcat(fields1, fields2 []*FieldValue) int {
if (len(fields1) + len(fields2)) != len(m.fields) {
panic("invalid field lookup depth")
}
idx := 0
remainingCombinationBucket := m.numFieldCombinations
for i, val := range fields1 {
idx, remainingCombinationBucket = m.lookupSingle(i, val, idx, remainingCombinationBucket)
}
numFields1 := len(fields1)
for i, val := range fields2 {
idx, remainingCombinationBucket = m.lookupSingle(i+numFields1, val, idx, remainingCombinationBucket)
}
return idx
}
// lookup looks up a key within the fieldMapper.
// The returned key is an index that can be used to access to map created by
// makeMap().
// This *must* be called with the correct number of fields, or it will panic.
// +checkescape:all
//
//go:nosplit
func (m *fieldMapper) lookup(fields ...*FieldValue) int {
return m.lookupConcat(fields, nil)
}
// numKeys returns the total number of key-to-field-combinations mappings
// defined by the fieldMapper.
//
//go:nosplit
func (m *fieldMapper) numKeys() int {
return m.numFieldCombinations
}
// makeDistributionSampleMap creates a two dimensional array, where:
// - The first level corresponds to unique field value combinations and is
// accessed using index "keys" made by fieldMapper.
// - The second level corresponds to buckets within a metric. The number of
// buckets is specified by numBuckets.
func (m *fieldMapper) makeDistributionSampleMap(numBuckets int) [][]atomicbitops.Uint64 {
samples := make([][]atomicbitops.Uint64, m.numKeys())
for i := range samples {
samples[i] = make([]atomicbitops.Uint64, numBuckets)
}
return samples
}
// keyToMultiField is the reverse of lookup/lookupConcat. The returned list of
// field values corresponds to the same order of fields that were passed in to
// newFieldMapper.
func (m *fieldMapper) keyToMultiField(key int) []string {
depth := len(m.fields)
if depth == 0 && key == 0 {
return nil
}
fieldValues := make([]string, depth)
remainingCombinationBucket := m.numFieldCombinations
for i := 0; i < depth; i++ {
remainingCombinationBucket /= len(m.fields[i].values)
fieldValues[i] = m.fields[i].values[key/remainingCombinationBucket].Value
key = key % remainingCombinationBucket
}
return fieldValues
}
// keyToMultiFieldInPlace does the operation described in `keyToMultiField`
// but modifies `fieldValues` in-place. It must already be of size
// `len(m.fields)`.
//
//go:nosplit
func (m *fieldMapper) keyToMultiFieldInPlace(key int, fieldValues []*FieldValue) {
if len(m.fields) == 0 {
return
}
depth := len(m.fields)
remainingCombinationBucket := m.numFieldCombinations
for i := 0; i < depth; i++ {
remainingCombinationBucket /= len(m.fields[i].values)
fieldValues[i] = m.fields[i].values[key/remainingCombinationBucket]
key = key % remainingCombinationBucket
}
}
// nameToPrometheusName transforms a path-style metric name (/foo/bar) into a Prometheus-style
// metric name (foo_bar).
func nameToPrometheusName(name string) string {
return strings.ReplaceAll(strings.TrimPrefix(name, "/"), "/", "_")
}
var validMetricNameRegexp = re.MustCompile(`^(?:/[_\w]+)+$`)
// verifyName verifies that the given metric name is a valid path-style metric
// name.
func verifyName(name string) error {
if !strings.HasPrefix(name, "/") {
return fmt.Errorf("metric name must start with a '/': %q", name)
}
if !validMetricNameRegexp.MatchString(name) {
return fmt.Errorf("invalid metric name: %q", name)
}
return nil
}
// RegisterCustomUint64Metric registers a metric with the given name.
//
// Register must only be called at init and will return and error if called
// after Initialized.
//
// Preconditions:
// - name must be globally unique.
// - Initialize/Disable have not been called.
// - value is expected to accept exactly len(fields) arguments.
func RegisterCustomUint64Metric(name string, metadata Uint64Metadata, value func(...*FieldValue) uint64) error {
if initialized.Load() {
return ErrInitializationDone
}
if _, ok := allMetrics.uint64Metrics[name]; ok {
return ErrNameInUse
}
if _, ok := allMetrics.distributionMetrics[name]; ok {
return ErrNameInUse
}
promType := prometheus.TypeGauge
if metadata.Cumulative {
promType = prometheus.TypeCounter
}
fm, err := newFieldMapper(metadata.Fields...)
if err != nil {
return fmt.Errorf("invalid fields: %w", err)
}
allMetrics.uint64Metrics[name] = customUint64Metric{
metadata: &pb.MetricMetadata{
Name: name,
PrometheusName: nameToPrometheusName(name),
Description: metadata.Description,
Cumulative: metadata.Cumulative,
Sync: metadata.Sync,
Type: pb.MetricMetadata_TYPE_UINT64,
Units: metadata.Unit,
},
prometheusMetric: &prometheus.Metric{
Name: nameToPrometheusName(name),
Help: metadata.Description,
Type: promType,
},
fields: metadata.Fields,
fieldMapper: fm,
value: value,
}
for _, field := range metadata.Fields {
allMetrics.uint64Metrics[name].metadata.Fields = append(allMetrics.uint64Metrics[name].metadata.Fields, field.toProto())
}
return nil
}
// MustRegisterCustomUint64Metric calls RegisterCustomUint64Metric for metrics
// without fields and panics if it returns an error.
func MustRegisterCustomUint64Metric(name string, metadata Uint64Metadata, value func(...*FieldValue) uint64) {
if err := RegisterCustomUint64Metric(name, metadata, value); err != nil {
panic(fmt.Sprintf("Unable to register metric %q: %s", name, err))
}
}
// NewUint64Metric creates and registers a new cumulative metric with the given
// name.
//
// Metrics must be statically defined (i.e., at init).
func NewUint64Metric(name string, metadata Uint64Metadata) (*Uint64Metric, error) {
if err := verifyName(name); err != nil {
return nil, err
}
m := Uint64Metric{name: name}
if err := RegisterCustomUint64Metric(name, metadata, m.Value); err != nil {
return nil, err
}
cm := allMetrics.uint64Metrics[name]
cm.forEachNonZero = m.forEachNonZero
m.fieldMapper = cm.fieldMapper
m.fields = make([]atomicbitops.Uint64, cm.fieldMapper.numKeys())
allMetrics.uint64Metrics[name] = cm
return &m, nil
}
// MustCreateNewUint64Metric calls NewUint64Metric and panics if it returns
// an error.
func MustCreateNewUint64Metric(name string, metadata Uint64Metadata) *Uint64Metric {
m, err := NewUint64Metric(name, metadata)
if err != nil {
panic(fmt.Sprintf("Unable to create metric %q: %s", name, err))
}
return m
}
// Value returns the current value of the metric for the given set of fields.
// This must be called with the correct number of field values or it will panic.
//
//go:nosplit
func (m *Uint64Metric) Value(fieldValues ...*FieldValue) uint64 {
key := m.fieldMapper.lookupConcat(fieldValues, nil)
return m.fields[key].Load()
}
// forEachNonZero iterates over each field combination and calls the given
// function whenever this metric's value is not zero.
func (m *Uint64Metric) forEachNonZero(f func(fieldValues []*FieldValue, value uint64)) {
numCombinations := m.fieldMapper.numKeys()
if len(m.fieldMapper.fields) == 0 {
// Special-case the "there are no fields" case for speed and to avoid
// allocating a slice.
if val := m.fields[0].Load(); val != 0 {
f(nil, val)
}
return
}
var fieldValues []*FieldValue
for k := 0; k < numCombinations; k++ {
val := m.fields[k].Load()
if val == 0 {
continue
}
if fieldValues == nil {
fieldValues = make([]*FieldValue, len(m.fieldMapper.fields))
}
m.fieldMapper.keyToMultiFieldInPlace(k, fieldValues)
f(fieldValues, val)
}
}
// Increment increments the metric by 1.
// This must be called with the correct number of field values or it will panic.
//
//go:nosplit
func (m *Uint64Metric) Increment(fieldValues ...*FieldValue) {
m.IncrementBy(1, fieldValues...)
}
// Decrement decrements the metric by 1.
// This must be called with the correct number of field values or it will panic.
//
//go:nosplit
func (m *Uint64Metric) Decrement(fieldValues ...*FieldValue) {
m.IncrementBy(0xFFFFFFFFFFFFFFFF, fieldValues...)
}
// IncrementBy increments the metric by v.
// It is also possible to use this function to decrement the metric by using
// a two's-complement int64 representation of the negative number to add.
// This must be called with the correct number of field values or it will panic.
//
//go:nosplit
func (m *Uint64Metric) IncrementBy(v uint64, fieldValues ...*FieldValue) {
key := m.fieldMapper.lookupConcat(fieldValues, nil)
m.fields[key].Add(v)
}
// Set sets the metric to v.
// This must be called with the correct number of field values or it will panic.
//
//go:nosplit
func (m *Uint64Metric) Set(v uint64, fieldValues ...*FieldValue) {
key := m.fieldMapper.lookupConcat(fieldValues, nil)
m.fields[key].Store(v)
}
// Bucketer is an interface to bucket values into finite, distinct buckets.
type Bucketer interface {
// NumFiniteBuckets is the number of finite buckets in the distribution.
// This is only called once and never expected to return a different value.
NumFiniteBuckets() int
// LowerBound takes the index of a bucket (within [0, NumBuckets()]) and
// returns the inclusive lower bound of that bucket.
// In other words, the lowest value of `x` for which `BucketIndex(x) == i`
// should be `x = LowerBound(i)`.
// The upper bound of a bucket is the lower bound of the next bucket.
// The last bucket (with `bucketIndex == NumFiniteBuckets()`) is infinite,
// i.e. it has no upper bound (but it still has a lower bound).
LowerBound(bucketIndex int) int64
// BucketIndex takes a sample and returns the index of the bucket that the
// sample should fall into.
// Must return either:
// - A value within [0, NumBuckets() -1] if the sample falls within a
// finite bucket
// - NumBuckets() if the sample falls within the last (infinite) bucket
// - '-1' if the sample is lower than what any bucket can represent, i.e.
// the sample should be in the implicit "underflow" bucket.
// This function must be go:nosplit-compatible and have no escapes.
// +checkescape:all
BucketIndex(sample int64) int
}
// ExponentialBucketer implements Bucketer, with the first bucket starting
// with 0 as lowest bound with `Width` width, and each subsequent bucket being
// wider by a scaled exponentially-growing series, until `NumFiniteBuckets`
// buckets exist.
type ExponentialBucketer struct {
// numFinitebuckets is the total number of finite buckets in the scheme.
numFiniteBuckets int
// width is the size of the first (0-th) finite bucket.
width float64
// scale is a factor applied uniformly to the exponential growth portion
// of the bucket size.
scale float64
// growth is the exponential growth factor for finite buckets.
// The n-th bucket is `growth` times wider than the (n-1)-th bucket.
// Bucket sizes are floored, so `width` and `growth` must be large enough
// such that the second bucket is actually wider than the first after
// flooring (unless, of course, fixed-width buckets are what's desired).
growth float64
// growthLog is math.Log(growth).
growthLog float64
// maxSample is the max sample value which can be represented in a finite
// bucket.
maxSample int64
// lowerbounds is a precomputed set of lower bounds of the buckets.
// The "underflow" bucket has no lower bound, so it is not included here.
// lowerBounds[0] is the lower bound of the first finite bucket, which is
// also the upper bound of the underflow bucket.
// lowerBounds[numFiniteBuckets] is the lower bound of the overflow bucket.
lowerBounds []int64
}
// Minimum/maximum finite buckets for exponential bucketers.
const (
exponentialMinBuckets = 1
exponentialMaxBuckets = 100
)
// NewExponentialBucketer returns a new Bucketer with exponential buckets.
func NewExponentialBucketer(numFiniteBuckets int, width uint64, scale, growth float64) *ExponentialBucketer {
if numFiniteBuckets < exponentialMinBuckets || numFiniteBuckets > exponentialMaxBuckets {
panic(fmt.Sprintf("number of finite buckets must be in [%d, %d]", exponentialMinBuckets, exponentialMaxBuckets))
}
if scale < 0 || growth < 0 {
panic(fmt.Sprintf("scale and growth for exponential buckets must be >0, got scale=%f and growth=%f", scale, growth))
}
b := &ExponentialBucketer{
numFiniteBuckets: numFiniteBuckets,
width: float64(width),
scale: scale,
growth: growth,
growthLog: math.Log(growth),
lowerBounds: make([]int64, numFiniteBuckets+1),
}
b.lowerBounds[0] = 0
for i := 1; i <= numFiniteBuckets; i++ {
b.lowerBounds[i] = int64(b.width*float64(i) + b.scale*math.Pow(b.growth, float64(i-1)))
if b.lowerBounds[i] < 0 {
panic(fmt.Sprintf("encountered bucket width overflow at bucket %d", i))
}
}
b.maxSample = b.lowerBounds[numFiniteBuckets] - 1
return b
}
// NumFiniteBuckets implements Bucketer.NumFiniteBuckets.
func (b *ExponentialBucketer) NumFiniteBuckets() int {
return int(b.numFiniteBuckets)
}
// LowerBound implements Bucketer.LowerBound.
func (b *ExponentialBucketer) LowerBound(bucketIndex int) int64 {
return b.lowerBounds[bucketIndex]
}
// BucketIndex implements Bucketer.BucketIndex.
// +checkescape:all
//
//go:nosplit
func (b *ExponentialBucketer) BucketIndex(sample int64) int {
if sample < 0 {
return -1
}
if sample == 0 {
return 0
}
if sample > b.maxSample {
return b.numFiniteBuckets
}
// Do a binary search. For the number of buckets we expect to deal with in
// this code (a few dozen at most), this may be faster than computing a
// logarithm. We can't use recursion because this would violate go:nosplit.
lowIndex := 0
highIndex := b.numFiniteBuckets
for {
pivotIndex := (highIndex + lowIndex) >> 1
lowerBound := b.lowerBounds[pivotIndex]
if sample < lowerBound {
highIndex = pivotIndex
continue
}
upperBound := b.lowerBounds[pivotIndex+1]
if sample >= upperBound {
lowIndex = pivotIndex
continue
}
return pivotIndex
}
}
// Verify that ExponentialBucketer implements Bucketer.
var _ = (Bucketer)((*ExponentialBucketer)(nil))
// DistributionMetric represents a distribution of values in finite buckets.
// It also separately keeps track of min/max in order to ascertain whether the
// buckets can faithfully represent the range of values encountered in the
// distribution.
type DistributionMetric struct {
// exponentialBucketer is the bucketing scheme used for this metric.
// Because we need DistributionMetric.AddSample to be go:nosplit-compatible,
// we cannot use an interface reference here, as we would not be able to call
// it in AddSample. Instead, we need one field per Bucketer implementation,
// and we call whichever one is in use in AddSample.
exponentialBucketer *ExponentialBucketer
// metadata is the metadata about this metric. It is immutable.
metadata *pb.MetricMetadata
// prometheusMetric describes the metric in Prometheus format. It is immutable.
prometheusMetric *prometheus.Metric
// fieldsToKey converts a multi-dimensional fields to a single string to use
// as key for `samples`.
fieldsToKey fieldMapper
// samples is the number of samples that fell within each bucket.
// It is mapped by the concatenation of the fields using `fieldsToKey`.
// The value is a list of bucket sample counts, with the 0-th being the
// "underflow bucket", i.e. the bucket of samples which cannot fall into
// any bucket that the bucketer supports.
// The i-th value is the number of samples that fell into the bucketer's
// (i-1)-th finite bucket.
// The last value is the number of samples that fell into the bucketer's
// last (i.e. infinite) bucket.
samples [][]atomicbitops.Uint64
// statistics is a set of statistics about each distribution.
// It is mapped by the concatenation of the fields using `fieldsToKey`.
statistics []distributionStatistics
}
// NewDistributionMetric creates and registers a new distribution metric.
func NewDistributionMetric(name string, sync bool, bucketer Bucketer, unit pb.MetricMetadata_Units, description string, fields ...Field) (*DistributionMetric, error) {
if err := verifyName(name); err != nil {
return nil, err
}
if initialized.Load() {
return nil, ErrInitializationDone
}
if _, ok := allMetrics.uint64Metrics[name]; ok {
return nil, ErrNameInUse
}
if _, ok := allMetrics.distributionMetrics[name]; ok {
return nil, ErrNameInUse
}
var exponentialBucketer *ExponentialBucketer
if expBucketer, ok := bucketer.(*ExponentialBucketer); ok {
exponentialBucketer = expBucketer
} else {
return nil, fmt.Errorf("unsupported bucketer implementation: %T", bucketer)
}
fieldsToKey, err := newFieldMapper(fields...)
if err != nil {
return nil, err
}
numFiniteBuckets := bucketer.NumFiniteBuckets()
samples := fieldsToKey.makeDistributionSampleMap(numFiniteBuckets + 2)
protoFields := make([]*pb.MetricMetadata_Field, len(fields))
for i, f := range fields {
protoFields[i] = f.toProto()
}
lowerBounds := make([]int64, numFiniteBuckets+1)
for i := 0; i <= numFiniteBuckets; i++ {
lowerBounds[i] = bucketer.LowerBound(i)
}
allMetrics.distributionMetrics[name] = &DistributionMetric{
exponentialBucketer: exponentialBucketer,
fieldsToKey: fieldsToKey,
samples: samples,
statistics: make([]distributionStatistics, fieldsToKey.numKeys()),
metadata: &pb.MetricMetadata{
Name: name,
PrometheusName: nameToPrometheusName(name),
Description: description,
Cumulative: false,
Sync: sync,
Type: pb.MetricMetadata_TYPE_DISTRIBUTION,
Units: unit,
Fields: protoFields,
DistributionBucketLowerBounds: lowerBounds,
},
prometheusMetric: &prometheus.Metric{
Name: nameToPrometheusName(name),
Type: prometheus.TypeHistogram,
Help: description,
},
}
return allMetrics.distributionMetrics[name], nil
}
// MustCreateNewDistributionMetric creates and registers a distribution metric.
// If an error occurs, it panics.
func MustCreateNewDistributionMetric(name string, sync bool, bucketer Bucketer, unit pb.MetricMetadata_Units, description string, fields ...Field) *DistributionMetric {
distrib, err := NewDistributionMetric(name, sync, bucketer, unit, description, fields...)
if err != nil {
panic(err)
}
return distrib
}
// distributionStatistics is a set of useful statistics for a distribution.
// As metric update operations must be non-blocking, this uses a bunch of
// atomic numbers rather than a mutex.
type distributionStatistics struct {
// sampleCount is the total number of samples.
sampleCount atomicbitops.Uint64
// sampleSum is the sum of samples.
sampleSum atomicbitops.Int64
// sumOfSquaredDeviations is the running sum of squared deviations from the
// mean of each sample.
// This quantity is useful as part of Welford's online algorithm:
// https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm
sumOfSquaredDeviations atomicbitops.Float64
// min and max are the minimum and maximum samples ever recorded.
min, max atomicbitops.Int64
}
// Update updates the distribution statistics with the given sample.
// This function must be non-blocking, i.e. no mutexes.
// As a result, it is not entirely accurate when it races with itself,
// though the imprecision should be fairly small and should not practically
// matter for distributions with more than a handful of records.
func (s *distributionStatistics) Update(sample int64) {
newSampleCount := s.sampleCount.Add(1)
newSampleSum := s.sampleSum.Add(sample)
if newSampleCount > 1 {
// Not the first sample of the distribution.
floatSample := float64(sample)
oldMean := float64(newSampleSum-sample) / float64(newSampleCount-1)
newMean := float64(newSampleSum) / float64(newSampleCount)
devSquared := (floatSample - oldMean) * (floatSample - newMean)
s.sumOfSquaredDeviations.Add(devSquared)
// Update min and max.
// We optimistically load racily here in the hope that it passes the CaS
// operation. If it doesn't, we'll load it atomically, so this is not a
// race.
sync.RaceDisable()
for oldMin := s.min.RacyLoad(); sample < oldMin && !s.min.CompareAndSwap(oldMin, sample); oldMin = s.min.Load() {
}
for oldMax := s.max.RacyLoad(); sample > oldMax && !s.max.CompareAndSwap(oldMax, sample); oldMax = s.max.Load() {
}
sync.RaceEnable()
} else {
// We are the first sample, so set the min and max to the current sample.
// See above for why disabling race detection is safe here as well.
sync.RaceDisable()
if !s.min.CompareAndSwap(0, sample) {
for oldMin := s.min.RacyLoad(); sample < oldMin && !s.min.CompareAndSwap(oldMin, sample); oldMin = s.min.Load() {
}
}
if !s.max.CompareAndSwap(0, sample) {
for oldMax := s.max.RacyLoad(); sample > oldMax && !s.max.CompareAndSwap(oldMax, sample); oldMax = s.max.Load() {
}
}
sync.RaceEnable()
}
}
// distributionStatisticsSnapshot an atomically-loaded snapshot of
// distributionStatistics.
type distributionStatisticsSnapshot struct {
// sampleCount is the total number of samples.
sampleCount uint64
// sampleSum is the sum of samples.
sampleSum int64
// sumOfSquaredDeviations is the running sum of squared deviations from the
// mean of each sample.
// This quantity is useful as part of Welford's online algorithm:
// https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm
sumOfSquaredDeviations float64
// min and max are the minimum and maximum samples ever recorded.
min, max int64
}
// Load generates a consistent snapshot of the distribution statistics.
func (s *distributionStatistics) Load() distributionStatisticsSnapshot {
// We start out reading things racily, but will verify each of them
// atomically later in this function, so this is OK. Disable the race
// checker for this part of the function.
sync.RaceDisable()
snapshot := distributionStatisticsSnapshot{
sampleCount: s.sampleCount.RacyLoad(),
sampleSum: s.sampleSum.RacyLoad(),
sumOfSquaredDeviations: s.sumOfSquaredDeviations.RacyLoad(),
min: s.min.RacyLoad(),
max: s.max.RacyLoad(),
}
sync.RaceEnable()
// Now verify that we loaded an atomic snapshot of the statistics.
// This relies on the fact that each update should at least change the
// count statistic, so we should be able to tell if anything changed based
// on whether we have an exact match with the currently-loaded values.
// If not, we reload that value and try again until all is consistent.
retry:
if sampleCount := s.sampleCount.Load(); sampleCount != snapshot.sampleCount {
snapshot.sampleCount = sampleCount
goto retry
}
if sampleSum := s.sampleSum.Load(); sampleSum != snapshot.sampleSum {
snapshot.sampleSum = sampleSum
goto retry
}
if ssd := s.sumOfSquaredDeviations.Load(); ssd != snapshot.sumOfSquaredDeviations {
snapshot.sumOfSquaredDeviations = ssd
goto retry
}
if min := s.min.Load(); min != snapshot.min {
snapshot.min = min
goto retry
}
if max := s.max.Load(); max != snapshot.max {
snapshot.max = max
goto retry
}
return snapshot
}
// AddSample adds a sample to the distribution.
// This *must* be called with the correct number of fields, or it will panic.
// +checkescape:all
//
//go:nosplit
func (d *DistributionMetric) AddSample(sample int64, fields ...*FieldValue) {
d.addSampleByKey(sample, d.fieldsToKey.lookup(fields...))
}
// addSampleByKey works like AddSample, with the field key already known.
// +checkescape:all
//
//go:nosplit
func (d *DistributionMetric) addSampleByKey(sample int64, key int) {
bucket := d.exponentialBucketer.BucketIndex(sample)
d.samples[key][bucket+1].Add(1)
d.statistics[key].Update(sample)
}
// Minimum number of buckets for NewDurationBucket.
const durationMinBuckets = 3
// NewDurationBucketer returns a Bucketer well-suited for measuring durations in
// nanoseconds. Useful for NewTimerMetric.
// minDuration and maxDuration are conservative estimates of the minimum and
// maximum durations expected to be accurately measured by the Bucketer.
func NewDurationBucketer(numFiniteBuckets int, minDuration, maxDuration time.Duration) Bucketer {
if numFiniteBuckets < durationMinBuckets {
panic(fmt.Sprintf("duration bucketer must have at least %d buckets, got %d", durationMinBuckets, numFiniteBuckets))
}
minNs := minDuration.Nanoseconds()
exponentCoversNs := float64(maxDuration.Nanoseconds()-int64(numFiniteBuckets-durationMinBuckets)*minNs) / float64(minNs)
exponent := math.Log(exponentCoversNs) / math.Log(float64(numFiniteBuckets-durationMinBuckets))
minNs = int64(float64(minNs) / exponent)
return NewExponentialBucketer(numFiniteBuckets, uint64(minNs), float64(minNs), exponent)
}
// TimerMetric wraps a distribution metric with convenience functions for
// latency measurements, which is a popular specialization of distribution
// metrics.
type TimerMetric struct {
DistributionMetric
}
// NewTimerMetric provides a convenient way to measure latencies.
// The arguments are the same as `NewDistributionMetric`, except:
// - `nanoBucketer`: Same as `NewDistribution`'s `bucketer`, expected to hold
// durations in nanoseconds. Adjust parameters accordingly.
// NewDurationBucketer may be helpful here.
func NewTimerMetric(name string, nanoBucketer Bucketer, description string, fields ...Field) (*TimerMetric, error) {
distrib, err := NewDistributionMetric(name, false, nanoBucketer, pb.MetricMetadata_UNITS_NANOSECONDS, description, fields...)
if err != nil {
return nil, err
}
return &TimerMetric{
DistributionMetric: *distrib,
}, nil
}
// MustCreateNewTimerMetric creates and registers a timer metric.
// If an error occurs, it panics.
func MustCreateNewTimerMetric(name string, nanoBucketer Bucketer, description string, fields ...Field) *TimerMetric {
timer, err := NewTimerMetric(name, nanoBucketer, description, fields...)
if err != nil {
panic(err)
}
return timer
}
// TimedOperation is used by TimerMetric to keep track of the time elapsed
// between an operation starting and stopping.
type TimedOperation struct {
// metric is a reference to the timer metric for the operation.
metric *TimerMetric
// partialFields is a prefix of the fields used in this operation.
// The rest of the fields is provided in TimedOperation.Finish.
partialFields []*FieldValue
// startedNs is the number of nanoseconds measured in TimerMetric.Start().
startedNs int64
}
// Start starts a timer measurement for the given combination of fields.
// It returns a TimedOperation which can be passed around as necessary to
// measure the duration of the operation.
// Once the operation is finished, call Finish on the TimedOperation.
// The fields passed to Start may be partially specified; if so, the remaining
// fields must be passed to TimedOperation.Finish. This is useful for cases
// where which path an operation took is only known after it happens. This
// path can be part of the fields passed to Finish.
// +checkescape:all
//
//go:nosplit
func (t *TimerMetric) Start(fields ...*FieldValue) TimedOperation {
return TimedOperation{
metric: t,
partialFields: fields,
startedNs: CheapNowNano(),
}
}
// Finish marks an operation as finished and records its duration.
// `extraFields` is the rest of the fields appended to the fields passed to
// `TimerMetric.Start`. The concatenation of these two must be the exact
// number of fields that the underlying metric has.
// +checkescape:all
//
//go:nosplit
func (o TimedOperation) Finish(extraFields ...*FieldValue) {
ended := CheapNowNano()
fieldKey := o.metric.fieldsToKey.lookupConcat(o.partialFields, extraFields)
o.metric.addSampleByKey(ended-o.startedNs, fieldKey)
}
// stageTiming contains timing data for an initialization stage.
type stageTiming struct {
stage InitStage
started time.Time
// ended is the zero time when the stage has not ended yet.
ended time.Time
}
// inProgress returns whether this stage hasn't ended yet.
func (s stageTiming) inProgress() bool {
return !s.started.IsZero() && s.ended.IsZero()
}
// metricSet holds metric data.
type metricSet struct {
// Metric registration data for all the metrics below.
registration *pb.MetricRegistration
// Map of uint64 metrics.
uint64Metrics map[string]customUint64Metric
// Map of distribution metrics.
distributionMetrics map[string]*DistributionMetric
// mu protects the fields below.
mu sync.RWMutex
// Information about the stages reached by the Sentry. Only appended to, so
// reading a shallow copy of the slice header concurrently is safe.
finished []stageTiming
// The current stage in progress.
currentStage stageTiming
}
// makeMetricSet returns a new metricSet.
func makeMetricSet() *metricSet {
return &metricSet{
uint64Metrics: make(map[string]customUint64Metric),
distributionMetrics: make(map[string]*DistributionMetric),
finished: make([]stageTiming, 0, len(allStages)),
}
}
// Values returns a snapshot of all values in m.
func (m *metricSet) Values() metricValues {
m.mu.Lock()
stages := m.finished[:]
m.mu.Unlock()
vals := metricValues{
uint64Metrics: make(map[string]any, len(m.uint64Metrics)),
distributionMetrics: make(map[string][][]uint64, len(m.distributionMetrics)),
distributionTotalSamples: make(map[string][]uint64, len(m.distributionMetrics)),
distributionStatistics: make(map[string][]distributionStatisticsSnapshot, len(m.distributionMetrics)),
stages: stages,
}
var tmpFieldValues []*FieldValue
for k, v := range m.uint64Metrics {
fields := v.fields
switch len(fields) {
case 0:
vals.uint64Metrics[k] = v.value()
default:
numFieldCombinations := v.fieldMapper.numKeys()
perFieldKeyVals := make([]uint64, numFieldCombinations)
if v.forEachNonZero != nil {
v.forEachNonZero(func(fieldValues []*FieldValue, val uint64) {
perFieldKeyVals[v.fieldMapper.lookup(fieldValues...)] = val
})
} else {
if len(tmpFieldValues) < len(v.fields) {
tmpFieldValues = make([]*FieldValue, len(v.fields))
}
fieldValues := tmpFieldValues[:len(v.fields)]
for fieldKey := 0; fieldKey < numFieldCombinations; fieldKey++ {
v.fieldMapper.keyToMultiFieldInPlace(fieldKey, fieldValues)
perFieldKeyVals[fieldKey] = v.value(fieldValues...)
}
}
vals.uint64Metrics[k] = perFieldKeyVals
}
}
for name, metric := range m.distributionMetrics {
fieldKeysToValues := make([][]uint64, len(metric.samples))
fieldKeysToTotalSamples := make([]uint64, len(metric.samples))
fieldKeysToStatistics := make([]distributionStatisticsSnapshot, len(metric.samples))
for fieldKey, samples := range metric.samples {
samplesSnapshot := snapshotDistribution(samples)
totalSamples := uint64(0)
for _, bucket := range samplesSnapshot {
totalSamples += bucket
}
if totalSamples == 0 {
// No samples recorded for this combination of field, so leave
// the maps for this fieldKey as nil. This lessens the memory cost
// of distributions with unused field combinations.
fieldKeysToTotalSamples[fieldKey] = 0
fieldKeysToStatistics[fieldKey] = distributionStatisticsSnapshot{}
fieldKeysToValues[fieldKey] = nil
} else {
fieldKeysToTotalSamples[fieldKey] = totalSamples
fieldKeysToStatistics[fieldKey] = metric.statistics[fieldKey].Load()
fieldKeysToValues[fieldKey] = samplesSnapshot
}
}
vals.distributionMetrics[name] = fieldKeysToValues
vals.distributionTotalSamples[name] = fieldKeysToTotalSamples
vals.distributionStatistics[name] = fieldKeysToStatistics
}
return vals
}
// metricValues contains a copy of the values of all metrics.
type metricValues struct {
// uint64Metrics is a map of uint64 metrics,
// with key as metric name. Value can be either uint64, or []uint64
// to support metrics with fields. The index corresponds to the field
// value combination key as managed by `fieldMapper`.
uint64Metrics map[string]any
// distributionMetrics is a map of distribution metrics.
// The first key level is the metric name.
// The second key level is an index ID corresponding to the combination of
// field values. The index is decoded to field strings using keyToMultiField.
// The slice value is the number of samples in each bucket of the
// distribution, with the first (0-th) element being the underflow bucket
// and the last element being the "infinite" (overflow) bucket.
// The slice value may also be nil for field combinations with no samples.
// This saves memory by avoiding storing anything for unused field
// combinations.
distributionMetrics map[string][][]uint64
// distributionTotalSamples is the total number of samples for each
// distribution metric and field values.
// It allows performing a quick diff between snapshots without having to
// iterate over all the buckets individually, so that distributions with
// no new samples are not retransmitted.
distributionTotalSamples map[string][]uint64
// distributionStatistics is a set of statistics about the samples.
distributionStatistics map[string][]distributionStatisticsSnapshot
// Information on when initialization stages were reached. Does not include
// the currently-ongoing stage, if any.
stages []stageTiming
}
var (
// emitMu protects metricsAtLastEmit and ensures that all emitted
// metrics are strongly ordered (older metrics are never emitted after
// newer metrics).
emitMu sync.Mutex
// metricsAtLastEmit contains the state of the metrics at the last emit event.
metricsAtLastEmit metricValues
)
// EmitMetricUpdate emits a MetricUpdate over the event channel.
//
// Only metrics that have changed since the last call are emitted.
//
// EmitMetricUpdate is thread-safe.
//
// Preconditions:
// - Initialize has been called.
func EmitMetricUpdate() {
emitMu.Lock()
defer emitMu.Unlock()
snapshot := allMetrics.Values()
m := pb.MetricUpdate{}
// On the first call metricsAtLastEmit will be empty. Include all
// metrics then.
for k, v := range snapshot.uint64Metrics {
prev, ok := metricsAtLastEmit.uint64Metrics[k]
switch t := v.(type) {
case uint64:
// Metric exists and value did not change.
if ok && prev.(uint64) == t {
continue
}
m.Metrics = append(m.Metrics, &pb.MetricValue{
Name: k,
Value: &pb.MetricValue_Uint64Value{Uint64Value: t},
})
case []uint64:
for fieldKey, metricValue := range t {
// Emit data on the first call only if the field
// value has been incremented. For all other
// calls, emit data if the field value has been
// changed from the previous emit.
if (!ok && metricValue == 0) || (ok && prev.([]uint64)[fieldKey] == metricValue) {
continue
}
uint64M := allMetrics.uint64Metrics[k]
m.Metrics = append(m.Metrics, &pb.MetricValue{
Name: k,
FieldValues: uint64M.fieldMapper.keyToMultiField(fieldKey),
Value: &pb.MetricValue_Uint64Value{Uint64Value: metricValue},
})
}
default:
panic(fmt.Sprintf("unsupported type in uint64Metrics: %T (%v)", v, v))
}
}
for name, dist := range snapshot.distributionTotalSamples {
prev, ok := metricsAtLastEmit.distributionTotalSamples[name]
for fieldKey, currentTotal := range dist {
if currentTotal == 0 {
continue
}
if ok {
if prevTotal := prev[fieldKey]; prevTotal == currentTotal {
continue
}
}
oldSamples := metricsAtLastEmit.distributionMetrics[name]
var newSamples []uint64
if oldSamples != nil && oldSamples[fieldKey] != nil {
currentSamples := snapshot.distributionMetrics[name][fieldKey]
numBuckets := len(currentSamples)
newSamples = make([]uint64, numBuckets)
for i := 0; i < numBuckets; i++ {
newSamples[i] = currentSamples[i] - oldSamples[fieldKey][i]
}
} else {
// oldSamples == nil means that the previous snapshot has no samples.
// This means the delta is the current number of samples, no need for
// a copy.
newSamples = snapshot.distributionMetrics[name][fieldKey]
}
m.Metrics = append(m.Metrics, &pb.MetricValue{
Name: name,
FieldValues: allMetrics.distributionMetrics[name].fieldsToKey.keyToMultiField(fieldKey),
Value: &pb.MetricValue_DistributionValue{
DistributionValue: &pb.Samples{
NewSamples: newSamples,
},
},
})
}
}
for s := len(metricsAtLastEmit.stages); s < len(snapshot.stages); s++ {
newStage := snapshot.stages[s]
m.StageTiming = append(m.StageTiming, &pb.StageTiming{
Stage: string(newStage.stage),
Started: &timestamppb.Timestamp{
Seconds: newStage.started.Unix(),
Nanos: int32(newStage.started.Nanosecond()),
},
Ended: &timestamppb.Timestamp{
Seconds: newStage.ended.Unix(),
Nanos: int32(newStage.ended.Nanosecond()),
},
})
}
metricsAtLastEmit = snapshot
if len(m.Metrics) == 0 && len(m.StageTiming) == 0 {
return
}
if log.IsLogging(log.Debug) {
sort.Slice(m.Metrics, func(i, j int) bool {
return m.Metrics[i].GetName() < m.Metrics[j].GetName()
})
log.Debugf("Emitting metrics:")
for _, metric := range m.Metrics {
var valueStr string
switch metric.GetValue().(type) {
case *pb.MetricValue_Uint64Value:
valueStr = fmt.Sprintf("%d", metric.GetUint64Value())
case *pb.MetricValue_DistributionValue:
valueStr = fmt.Sprintf("new distribution samples: %+v", metric.GetDistributionValue())
default:
valueStr = "unsupported type"
}
if len(metric.GetFieldValues()) > 0 {
var foundMetadata *pb.MetricMetadata
if metricObj, found := allMetrics.uint64Metrics[metric.GetName()]; found {
foundMetadata = metricObj.metadata
} else if metricObj, found := allMetrics.distributionMetrics[metric.GetName()]; found {
foundMetadata = metricObj.metadata
}
if foundMetadata == nil || len(foundMetadata.GetFields()) != len(metric.GetFieldValues()) {
// This should never happen, but if it somehow does, we don't want to crash here, as
// this is debug output that may already be printed in the context of panic.
log.Debugf("%s%v (cannot find metric definition!): %s", metric.GetName(), metric.GetFieldValues(), valueStr)
continue
}
var sb strings.Builder
for i, fieldValue := range metric.GetFieldValues() {
if i > 0 {
sb.WriteRune(',')
}
sb.WriteString(foundMetadata.GetFields()[i].GetFieldName())
sb.WriteRune('=')
sb.WriteString(fieldValue)
}
log.Debugf(" Metric %s[%s]: %s", metric.GetName(), sb.String(), valueStr)
} else {
log.Debugf(" Metric %s: %s", metric.GetName(), valueStr)
}
}
for _, stage := range m.StageTiming {
duration := time.Duration(stage.Ended.Seconds-stage.Started.Seconds)*time.Second + time.Duration(stage.Ended.Nanos-stage.Started.Nanos)*time.Nanosecond
log.Debugf("Stage %s took %v", stage.GetStage(), duration)
}
}
if err := eventchannel.Emit(&m); err != nil {
log.Warningf("Unable to emit metrics: %s", err)
}
}
// EmitMetricUpdateWithTimeout is equivalent to EmitMetricUpdate, but only
// waits for timeout before returning. Metrics may be emitted after
// EmitMetricUpdateWithTimeout returns.
func EmitMetricUpdateWithTimeout(timeout time.Duration) {
ch := make(chan struct{})
go func() {
EmitMetricUpdate()
close(ch)
}()
select {
case <-ch:
case <-time.After(timeout):
}
}
// SnapshotOptions controls how snapshots are exported in GetSnapshot.
type SnapshotOptions struct {
// Filter, if set, should return true for metrics that should be written to
// the snapshot. If unset, all metrics are written to the snapshot.
Filter func(*prometheus.Metric) bool
}
// GetSnapshot returns a Prometheus snapshot of the metric data.
// Returns ErrNotYetInitialized if metrics have not yet been initialized.
func GetSnapshot(options SnapshotOptions) (*prometheus.Snapshot, error) {
if !initialized.Load() {
return nil, ErrNotYetInitialized
}
values := allMetrics.Values()
snapshot := prometheus.NewSnapshot()
for k, v := range values.uint64Metrics {
m := allMetrics.uint64Metrics[k]
if options.Filter != nil && !options.Filter(m.prometheusMetric) {
continue
}
switch t := v.(type) {
case uint64:
if m.metadata.GetCumulative() && t == 0 {
// Zero-valued counter, ignore.
continue
}
snapshot.Add(prometheus.NewIntData(m.prometheusMetric, int64(t)))
case []uint64:
for fieldKey, metricValue := range t {
if m.metadata.GetCumulative() && metricValue == 0 {
// Zero-valued counter, ignore.
continue
}
fieldValues := m.fieldMapper.keyToMultiField(fieldKey)
fieldKeysAndValues := make(map[string]string, len(fieldValues))
for i, field := range m.fields {
fieldKeysAndValues[field.name] = fieldValues[i]
}
snapshot.Add(prometheus.LabeledIntData(m.prometheusMetric, fieldKeysAndValues, int64(metricValue)))
}
default:
panic(fmt.Sprintf("unsupported type in uint64Metrics: %T (%v)", v, v))
}
}
for k, dists := range values.distributionTotalSamples {
m := allMetrics.distributionMetrics[k]
if options.Filter != nil && !options.Filter(m.prometheusMetric) {
continue
}
distributionSamples := values.distributionMetrics[k]
numFiniteBuckets := m.exponentialBucketer.NumFiniteBuckets()
statistics := values.distributionStatistics[k]
for fieldKey := range dists {
var labels map[string]string
if numFields := m.fieldsToKey.numKeys(); numFields > 0 {
labels = make(map[string]string, numFields)
for fieldIndex, field := range m.fieldsToKey.keyToMultiField(fieldKey) {
labels[m.metadata.Fields[fieldIndex].GetFieldName()] = field
}
}
currentSamples := distributionSamples[fieldKey]
buckets := make([]prometheus.Bucket, numFiniteBuckets+2)
samplesForFieldKey := uint64(0)
for b := 0; b < numFiniteBuckets+2; b++ {
var upperBound prometheus.Number
if b == numFiniteBuckets+1 {
upperBound = prometheus.Number{Float: math.Inf(1)} // Overflow bucket.
} else {
upperBound = prometheus.Number{Int: m.exponentialBucketer.LowerBound(b)}
}
samples := uint64(0)
if currentSamples != nil {
samples = currentSamples[b]
samplesForFieldKey += samples
}
buckets[b] = prometheus.Bucket{
Samples: samples,
UpperBound: upperBound,
}
}
if samplesForFieldKey == 0 {
// Zero-valued distribution (no samples in any bucket for this field
// combination). Ignore.
continue
}
snapshot.Add(&prometheus.Data{
Metric: m.prometheusMetric,
Labels: prometheus.Labels(labels),
HistogramValue: &prometheus.Histogram{
Total: prometheus.Number{Int: statistics[fieldKey].sampleSum},
SumOfSquaredDeviations: prometheus.Number{Float: statistics[fieldKey].sumOfSquaredDeviations},
Min: prometheus.Number{Int: statistics[fieldKey].min},
Max: prometheus.Number{Int: statistics[fieldKey].max},
Buckets: buckets,
},
})
}
}
return snapshot, nil
}
// StartStage should be called when an initialization stage is started.
// It returns a function that must be called to indicate that the stage ended.
// Alternatively, future calls to StartStage will implicitly indicate that the
// previous stage ended.
// Stage information will be emitted in the next call to EmitMetricUpdate after
// a stage has ended.
//
// This function may (and is expected to) be called prior to final
// initialization of this metric library, as it has to capture early stages
// of Sentry initialization.
func StartStage(stage InitStage) func() {
now := time.Now()
allMetrics.mu.Lock()
defer allMetrics.mu.Unlock()
if allMetrics.currentStage.inProgress() {
endStage(now)
}
allMetrics.currentStage.stage = stage
allMetrics.currentStage.started = now
return func() {
now := time.Now()
allMetrics.mu.Lock()
defer allMetrics.mu.Unlock()
// The current stage may have been ended by another call to StartStage, so
// double-check prior to clearing the current stage.
if allMetrics.currentStage.inProgress() && allMetrics.currentStage.stage == stage {
endStage(now)
}
}
}
// endStage marks allMetrics.currentStage as ended, adding it to the list of
// finished stages. It assumes allMetrics.mu is locked.
func endStage(when time.Time) {
allMetrics.currentStage.ended = when
allMetrics.finished = append(allMetrics.finished, allMetrics.currentStage)
allMetrics.currentStage = stageTiming{}
}