internal

Documentation

Overview

Package internal provides types and functionality used to aggregate and cycle the state of metric measurements made by the SDK. These types and functionality are meant only for internal SDK use.

Example

//go:build go1.18
// +build go1.18

package main

import (
	"context"
	"fmt"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric/instrument"
	"go.opentelemetry.io/otel/metric/instrument/syncint64"
	"go.opentelemetry.io/otel/sdk/metric/aggregation"
	"go.opentelemetry.io/otel/sdk/metric/metricdata"
)

type meter struct {
	// When a reader initiates a collection, the meter would collect
	// aggregations from each of these functions.
	aggregations []metricdata.Aggregation
}

func (m *meter) SyncInt64() syncint64.InstrumentProvider {
	// The same would be done for all the other instrument providers.
	return (*syncInt64Provider)(m)
}

type syncInt64Provider meter

func (p *syncInt64Provider) Counter(string, ...instrument.Option) (syncint64.Counter, error) {
	// This is an example of how a synchronous int64 provider would create an
	// aggregator for a new counter. At this point the provider would
	// determine the aggregation and temporality to used based on the Reader
	// and View configuration. Assume here these are determined to be a
	// cumulative sum.

	aggregator := NewCumulativeSum[int64](true)
	count := inst{aggregateFunc: aggregator.Aggregate}

	p.aggregations = append(p.aggregations, aggregator.Aggregation())

	fmt.Printf("using %T aggregator for counter\n", aggregator)

	return count, nil
}

func (p *syncInt64Provider) UpDownCounter(string, ...instrument.Option) (syncint64.UpDownCounter, error) {
	// This is an example of how a synchronous int64 provider would create an
	// aggregator for a new up-down counter. At this point the provider would
	// determine the aggregation and temporality to used based on the Reader
	// and View configuration. Assume here these are determined to be a
	// last-value aggregation (the temporality does not affect the produced
	// aggregations).

	aggregator := NewLastValue[int64]()
	upDownCount := inst{aggregateFunc: aggregator.Aggregate}

	p.aggregations = append(p.aggregations, aggregator.Aggregation())

	fmt.Printf("using %T aggregator for up-down counter\n", aggregator)

	return upDownCount, nil
}

func (p *syncInt64Provider) Histogram(string, ...instrument.Option) (syncint64.Histogram, error) {
	// This is an example of how a synchronous int64 provider would create an
	// aggregator for a new histogram. At this point the provider would
	// determine the aggregation and temporality to used based on the Reader
	// and View configuration. Assume here these are determined to be a delta
	// explicit-bucket histogram.

	aggregator := NewDeltaHistogram[int64](aggregation.ExplicitBucketHistogram{
		Boundaries: []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 1000},
		NoMinMax:   false,
	})
	hist := inst{aggregateFunc: aggregator.Aggregate}

	p.aggregations = append(p.aggregations, aggregator.Aggregation())

	fmt.Printf("using %T aggregator for histogram\n", aggregator)

	return hist, nil
}

// inst is a generalized int64 synchronous counter, up-down counter, and
// histogram used for demonstration purposes only.
type inst struct {
	instrument.Synchronous

	aggregateFunc func(int64, attribute.Set)
}

func (inst) Add(context.Context, int64, ...attribute.KeyValue)    {}
func (inst) Record(context.Context, int64, ...attribute.KeyValue) {}

func main() {
	m := meter{}
	provider := m.SyncInt64()

	_, _ = provider.Counter("counter example")
	_, _ = provider.UpDownCounter("up-down counter example")
	_, _ = provider.Histogram("histogram example")

}

Output:

using *internal.cumulativeSum[int64] aggregator for counter
using *internal.lastValue[int64] aggregator for up-down counter
using *internal.deltaHistogram[int64] aggregator for histogram

Index

• type Aggregator
  • func NewCumulativeHistogram[N int64 | float64](cfg aggregation.ExplicitBucketHistogram) Aggregator[N]
  • func NewCumulativeSum[N int64 | float64](monotonic bool) Aggregator[N]
  • func NewDeltaHistogram[N int64 | float64](cfg aggregation.ExplicitBucketHistogram) Aggregator[N]
  • func NewDeltaSum[N int64 | float64](monotonic bool) Aggregator[N]
  • func NewFilter[N int64 | float64](agg Aggregator[N], fn func(attribute.Set) attribute.Set) Aggregator[N]
  • func NewLastValue[N int64 | float64]() Aggregator[N]

Examples

• Package

Types

type Aggregator

type Aggregator[N int64 | float64] interface {
	// Aggregate records the measurement, scoped by attr, and aggregates it
	// into an aggregation.
	Aggregate(measurement N, attr attribute.Set)

	// Aggregation returns an Aggregation, for all the aggregated
	// measurements made and ends an aggregation cycle.
	Aggregation() metricdata.Aggregation
}

Aggregator forms an aggregation from a collection of recorded measurements.

func NewCumulativeHistogram

func NewCumulativeHistogram[N int64 | float64](cfg aggregation.ExplicitBucketHistogram) Aggregator[N]

NewCumulativeHistogram returns an Aggregator that summarizes a set of measurements as an histogram. Each histogram is scoped by attributes.

Each aggregation cycle builds from the previous, the histogram counts are the bucketed counts of all values aggregated since the returned Aggregator was created.

func NewCumulativeSum

func NewCumulativeSum[N int64 | float64](monotonic bool) Aggregator[N]

NewCumulativeSum returns an Aggregator that summarizes a set of measurements as their arithmetic sum. Each sum is scoped by attributes and the aggregation cycle the measurements were made in.

The monotonic value is used to communicate the produced Aggregation is monotonic or not. The returned Aggregator does not make any guarantees this value is accurate. It is up to the caller to ensure it.

Each aggregation cycle is treated independently. When the returned Aggregator's Aggregation method is called it will reset all sums to zero.

func NewDeltaHistogram

func NewDeltaHistogram[N int64 | float64](cfg aggregation.ExplicitBucketHistogram) Aggregator[N]

NewDeltaHistogram returns an Aggregator that summarizes a set of measurements as an histogram. Each histogram is scoped by attributes and the aggregation cycle the measurements were made in.

Each aggregation cycle is treated independently. When the returned Aggregator's Aggregations method is called it will reset all histogram counts to zero.

func NewDeltaSum

func NewDeltaSum[N int64 | float64](monotonic bool) Aggregator[N]

NewDeltaSum returns an Aggregator that summarizes a set of measurements as their arithmetic sum. Each sum is scoped by attributes and the aggregation cycle the measurements were made in.

The monotonic value is used to communicate the produced Aggregation is monotonic or not. The returned Aggregator does not make any guarantees this value is accurate. It is up to the caller to ensure it.

Each aggregation cycle is treated independently. When the returned Aggregator's Aggregation method is called it will reset all sums to zero.

func NewFilter

func NewFilter[N int64 | float64](agg Aggregator[N], fn func(attribute.Set) attribute.Set) Aggregator[N]

NewFilter wraps an Aggregator with an attribute filtering function.

func NewLastValue

func NewLastValue[N int64 | float64]() Aggregator[N]

NewLastValue returns an Aggregator that summarizes a set of measurements as the last one made.