Documentation
¶
Overview ¶
Package opencensus normalizes the use of OpenCensus through local, "sharded" aggregation that is GENERIC over the labels key K.
Instead of calling stats.Record per event (each call builds a tag.Map and sends a recordReq to the global worker), we accumulate per key K across N shards and emit in bursts every `interval`. The key K is any comparable struct you define; a Schema[K] (Strategy pattern) projects it onto OpenCensus.
Three variants behind the SAME Aggregator[K, N] interface (SumCount and Distribution here; LastValue in lastvalue.go), plus a multi-metric aggregator (multi.go) that shares one store, flusher and ctxCache across several Count/Sum/LastValue metrics on the same key K. The hot path (Add) does not allocate on the heap after a key is seen for the first time; the flush swaps the map to avoid blocking writers and reuses the per-key context via ctxCache.
Index ¶
- type Aggregator
- type Config
- type CountAggregator
- type CountConfig
- type DistributionAggregator
- type DistributionConfig
- type LastValueAggregator
- type LastValueConfig
- type Measure
- type MultiAggregator
- type MultiBuilder
- type MultiConfig
- type MultiHandle
- type Number
- type Schema
- type SumAggregator
- type SumConfig
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Aggregator ¶
type Aggregator[K comparable, N Number] interface { Add(k K, value N) Stop() }
Aggregator is the common interface, generic over the labels key K and the measure value type N.
type Config ¶
type Config[K comparable] struct { Shards int // rounded up to a power of 2. Default 16. // Interval is the flush cadence. Default 10s. // // Tuning rule: set Interval to an exact divisor of the exporter reporting // period (view.SetReportingPeriod) and at most half of it. For the common // 30s period use 10s or 15s; 10s also divides the common 60s period, hence // the default. Interval equal to the reporting period causes phase races // (export windows with no fresh flush); non-divisor intervals cause sawtooth // in delta/rate charts. // // For LastValue gauges, Interval is the maximum staleness at export time. // For distributions with MaxSamplesPerKey, the reservoir resets each flush, // so shorter intervals improve percentile fidelity. Interval time.Duration Schema Schema[K] // key projection strategy }
Config holds the settings shared by every aggregator variant: the shard count, the flush interval and the key projection Schema.
Cardinality is the dimension that drives cost. Flush is O(distinct keys): each flush does one ctxCache lookup plus one stats.Record per key, and every record is funneled to the single global OpenCensus worker goroutine. The ctxCache (see ctxcache.go) grows with the number of distinct keys ever seen and is never evicted, so with unbounded label cardinality you must cap or project keys via Schema. Rough guidance:
- ≲1k keys: Interval 5–10s is fine.
- 10k–100k+ keys: prefer Interval 10–15s and set MaxSamplesPerKey on distributions to bound memory.
- Do not go below ~5s at high cardinality: flush bursts can saturate the OpenCensus worker channel and block the flusher goroutine.
Shards affect writer contention (concurrent Add goroutines), not key count.
type CountAggregator ¶
type CountAggregator[K comparable, N Number] struct { // contains filtered or unexported fields }
CountAggregator accumulates the running count per key K across sharded stores and flushes both to OpenCensus on the configured interval.
func NewCountAggregator ¶
func NewCountAggregator[K comparable, N Number](cfg CountConfig[K, N]) *CountAggregator[K, N]
NewCountAggregator builds a CountAggregator from cfg, applying defaults and starting the background flusher.
func (*CountAggregator[K, N]) Add ¶
func (a *CountAggregator[K, N]) Add(k K, _ N)
Add adds value to the running increments its count.
func (*CountAggregator[K, N]) Stop ¶
func (a *CountAggregator[K, N]) Stop()
Stop halts the background flusher.
type CountConfig ¶
type CountConfig[K comparable, N Number] struct { Config[K] CountMeasure Measure[N] }
CountConfig configures a SumCountAggregator, pairing the shared Config with the measures used to record the accumulated count.
type DistributionAggregator ¶
type DistributionAggregator[K comparable, N Number] struct { // contains filtered or unexported fields }
DistributionAggregator collects per-key samples across sharded stores and flushes them to OpenCensus on the configured interval. When MaxSamplesPerKey is set it keeps a bounded reservoir sample per key.
func NewDistributionAggregator ¶
func NewDistributionAggregator[K comparable, N Number](cfg DistributionConfig[K, N]) *DistributionAggregator[K, N]
NewDistributionAggregator builds a DistributionAggregator from cfg, applying defaults and starting the background flusher.
func (*DistributionAggregator[K, N]) Add ¶
func (a *DistributionAggregator[K, N]) Add(k K, value N)
Add records value as a sample for k, using reservoir sampling once the per-key sample cap is reached.
func (*DistributionAggregator[K, N]) Stop ¶
func (a *DistributionAggregator[K, N]) Stop()
Stop halts the background flusher.
type DistributionConfig ¶
type DistributionConfig[K comparable, N Number] struct { Config[K] Measure Measure[N] MaxSamplesPerKey int // 0 = exact; >0 = reservoir sampling (bounded memory) }
DistributionConfig configures a DistributionAggregator with the shared Config, the measure to record samples against and the optional per-key sample cap.
type LastValueAggregator ¶
type LastValueAggregator[K comparable, N Number] struct { // contains filtered or unexported fields }
LastValueAggregator keeps the last value recorded per key K across sharded stores and flushes it to OpenCensus on the configured interval.
func NewLastValueAggregator ¶
func NewLastValueAggregator[K comparable, N Number](cfg LastValueConfig[K, N]) *LastValueAggregator[K, N]
NewLastValueAggregator builds a LastValueAggregator from cfg, applying defaults and starting the background flusher.
func (*LastValueAggregator[K, N]) Add ¶
func (a *LastValueAggregator[K, N]) Add(k K, value N)
Add overwrites the value of the key. The shard lock serializes the writes: "last-write-wins" is well defined by the acquisition order.
func (*LastValueAggregator[K, N]) Stop ¶
func (a *LastValueAggregator[K, N]) Stop()
Stop halts the background flusher.
type LastValueConfig ¶
type LastValueConfig[K comparable, N Number] struct { Config[K] // Measure must be backed by a view of type view.LastValue(). Measure Measure[N] }
LastValueConfig configures a LastValueAggregator: it embeds the shared Config and the measure whose view must be of type view.LastValue().
type Measure ¶ added in v1.1.0
type Measure[N Number] interface { M(v N) stats.Measurement }
Measure is satisfied by *stats.Float64Measure (N=float64) and *stats.Int64Measure (N=int64). A parametrized interface is needed because M has a different signature on each concrete measure, so a union constraint alone cannot expose it directly.
type MultiAggregator ¶ added in v1.3.0
type MultiAggregator[K comparable, N Number] struct { // contains filtered or unexported fields }
MultiAggregator accumulates several Count/Sum/LastValue metrics per key K in one sharded store and flushes them together to OpenCensus on the configured interval.
func (*MultiAggregator[K, N]) Stop ¶ added in v1.3.0
func (a *MultiAggregator[K, N]) Stop()
Stop halts the background flusher after a final flush.
type MultiBuilder ¶ added in v1.3.0
type MultiBuilder[K comparable, N Number] struct { // contains filtered or unexported fields }
MultiBuilder registers the metrics of a MultiAggregator and freezes them on Build. It is single-use: registering after Build or calling Build twice panics.
func NewMultiBuilder ¶ added in v1.3.0
func NewMultiBuilder[K comparable, N Number](cfg MultiConfig[K, N]) *MultiBuilder[K, N]
NewMultiBuilder starts a builder for a MultiAggregator over key K and value type N. Register metrics with Count/Sum/LastValue, then call Build.
Example ¶
ExampleNewMultiBuilder folds several metrics over the same HTTPLabels key into one aggregator: a single sharded store, flusher and ctxCache, and one stats.Record per key on flush. Each Count/Sum/LastValue registration returns a lightweight handle.
package main
// Example of a concrete Schema for the original case (user/route/status). It serves
// as a reference: for a different set of properties, define your own key K and your
// Schema[K] the same way (with the fields you need).
import (
"time"
"go.opencensus.io/stats"
"go.opencensus.io/tag"
)
// HTTPLabels is an example labels key. Any comparable struct works.
type HTTPLabels struct {
User string
Route string
Status string
}
// HTTPSchema implements Schema[HTTPLabels].
type HTTPSchema struct {
KeyUser tag.Key
KeyRoute tag.Key
KeyStatus tag.Key
}
// Hash computes the shard hash for k from its user, route and status fields.
func (HTTPSchema) Hash(k HTTPLabels) uint64 {
return hashStrings(k.User, k.Route, k.Status)
}
// Mutators returns the tag.Mutator values that upsert the user, route and status
// labels into an OpenCensus context.
func (s HTTPSchema) Mutators(k HTTPLabels) []tag.Mutator {
return []tag.Mutator{
tag.Upsert(s.KeyUser, k.User),
tag.Upsert(s.KeyRoute, k.Route),
tag.Upsert(s.KeyStatus, k.Status),
}
}
// Compile-time conformance of the three variants with the interface.
var (
_ Schema[HTTPLabels] = HTTPSchema{}
_ Aggregator[HTTPLabels, float64] = (*CountAggregator[HTTPLabels, float64])(nil)
_ Aggregator[HTTPLabels, float64] = (*SumAggregator[HTTPLabels, float64])(nil)
_ Aggregator[HTTPLabels, float64] = (*DistributionAggregator[HTTPLabels, float64])(nil)
_ Aggregator[HTTPLabels, float64] = (*LastValueAggregator[HTTPLabels, float64])(nil)
_ Aggregator[HTTPLabels, int64] = (*CountAggregator[HTTPLabels, int64])(nil)
_ Aggregator[HTTPLabels, int64] = (*SumAggregator[HTTPLabels, int64])(nil)
_ Aggregator[HTTPLabels, int64] = (*DistributionAggregator[HTTPLabels, int64])(nil)
_ Aggregator[HTTPLabels, int64] = (*LastValueAggregator[HTTPLabels, int64])(nil)
)
// ExampleNewMultiBuilder folds several metrics over the same HTTPLabels key into one
// aggregator: a single sharded store, flusher and ctxCache, and one stats.Record per
// key on flush. Each Count/Sum/LastValue registration returns a lightweight handle.
func main() {
schema := HTTPSchema{
KeyUser: tag.MustNewKey("user"),
KeyRoute: tag.MustNewKey("route"),
KeyStatus: tag.MustNewKey("status"),
}
requestMeasure := stats.Float64("myapp/requests", "HTTP requests", stats.UnitDimensionless)
bytesMeasure := stats.Float64("myapp/bytes_out", "response bytes", stats.UnitBytes)
inflightMeasure := stats.Float64("myapp/inflight", "in-flight requests", stats.UnitDimensionless)
b := NewMultiBuilder[HTTPLabels, float64](MultiConfig[HTTPLabels, float64]{
Config: Config[HTTPLabels]{
Shards: 16,
Interval: 10 * time.Second,
Schema: schema,
},
// SkipZeros: true, // omit Count/Sum slots still 0 at flush time
})
// Register every metric before Build; each call returns a handle to record against.
requests := b.Count(requestMeasure) // Count ignores the value: each Add is +1
bytesOut := b.Sum(bytesMeasure) // Sum accumulates the value
inflight := b.LastValue(inflightMeasure) // LastValue keeps the last write
agg := b.Build() // applies defaults, starts the background flusher
defer agg.Stop() // final flush before returning
k := HTTPLabels{User: "alice", Route: "/orders", Status: "200"}
requests.Add(k, 1)
bytesOut.Add(k, 2048)
inflight.Add(k, 3)
}
// hashStrings computes FNV-1a inline over the string fields, without allocating on
// the heap (the variadic does not escape, the backing array goes on the stack). The
// separator avoids collisions of the form ("ab","c") vs ("a","bc").
func hashStrings(parts ...string) uint64 {
const (
offset = uint64(14695981039346656037)
prime = uint64(1099511628211)
)
h := offset
for _, s := range parts {
for i := 0; i < len(s); i++ {
h ^= uint64(s[i])
h *= prime
}
h ^= '|'
h *= prime
}
return h
}
Output:
func (*MultiBuilder[K, N]) Build ¶ added in v1.3.0
func (b *MultiBuilder[K, N]) Build() *MultiAggregator[K, N]
Build freezes the registered metrics, applies Config defaults, starts the background flusher and returns the aggregator. It panics if called twice.
func (*MultiBuilder[K, N]) Count ¶ added in v1.3.0
func (b *MultiBuilder[K, N]) Count(m Measure[N]) MultiHandle[K, N]
Count registers a count metric and returns its handle. Each Add increments the slot by one regardless of the value passed.
func (*MultiBuilder[K, N]) LastValue ¶ added in v1.3.0
func (b *MultiBuilder[K, N]) LastValue(m Measure[N]) MultiHandle[K, N]
LastValue registers a gauge metric and returns its handle. Each Add overwrites the slot; the measure must be backed by a view of type view.LastValue().
func (*MultiBuilder[K, N]) Sum ¶ added in v1.3.0
func (b *MultiBuilder[K, N]) Sum(m Measure[N]) MultiHandle[K, N]
Sum registers a sum metric and returns its handle. Each Add accumulates the value into the slot.
type MultiConfig ¶ added in v1.3.0
type MultiConfig[K comparable, N Number] struct { Config[K] // SkipZeros, when true, omits Count/Sum slots whose value is 0 from the flush. // It never affects LastValue: a gauge of 0 is a legitimate reading and is // emitted whenever its slot was written this window (see multiAcc.touched). SkipZeros bool }
MultiConfig configures a MultiAggregator with the shared Config plus the zero-slot policy applied to Count/Sum on flush.
type MultiHandle ¶ added in v1.3.0
type MultiHandle[K comparable, N Number] struct { // contains filtered or unexported fields }
MultiHandle is the write endpoint for one registered metric. It is a lightweight value (a shared aggregator pointer plus the slot coordinates) meant to be copied freely; every copy writes the same underlying slot. It intentionally lacks Stop, so it does not satisfy Aggregator[K, N].
func (MultiHandle[K, N]) Add ¶ added in v1.3.0
func (h MultiHandle[K, N]) Add(k K, value N)
Add folds value into this metric's slot for key k under the shard lock: Count increments by one (value ignored), Sum accumulates, LastValue overwrites.
type Schema ¶
type Schema[K comparable] interface { Hash(k K) uint64 Mutators(k K) []tag.Mutator }
Schema is the strategy that projects a labels key K onto OpenCensus: Hash distributes the key across shards on the hot path, and Mutators builds the tag.Mutator values used to derive the recording context.
type SumAggregator ¶
type SumAggregator[K comparable, N Number] struct { // contains filtered or unexported fields }
SumAggregator accumulates the running sum per key K across sharded stores and flushes both to OpenCensus on the configured interval.
func NewSumAggregator ¶
func NewSumAggregator[K comparable, N Number](cfg SumConfig[K, N]) *SumAggregator[K, N]
NewSumAggregator builds a SumCountAggregator from cfg, applying defaults and starting the background flusher.
func (*SumAggregator[K, N]) Add ¶
func (a *SumAggregator[K, N]) Add(k K, value N)
Add adds value to the running sum for k.
func (*SumAggregator[K, N]) Stop ¶
func (a *SumAggregator[K, N]) Stop()
Stop halts the background flusher.