opencensus

README

opencensus-go-metrics-optimized

Generic, sharded pre-aggregation for OpenCensus Go metrics.

Calling stats.Record on every event builds a tag.Map and enqueues a recordReq to OpenCensus's single global worker goroutine — under high throughput that worker becomes a bottleneck and a point of lock contention. This library accumulates values per label key across N in-memory shards and flushes them to OpenCensus in bursts on a configurable interval, cutting the number of stats.Record calls (and their allocations) by orders of magnitude without changing the semantics of the underlying views.

Features

• Generic over your label key. Define any comparable struct K as your label set and a Schema[K] that knows how to project it onto OpenCensus — no map[string]interface{} or reflection involved.
• Generic over the measure value type N. Each config takes a second type parameter N (~int64 | ~float64), so a variant can be backed by either a *stats.Float64Measure or a *stats.Int64Measure.
• Four aggregator variants behind one interface (Aggregator[K, N]): Count, Sum, Distribution (with optional bounded reservoir sampling), and LastValue (for gauges).
• Sharded, lock-striped storage. Writes only lock the shard for their key, not a single global mutex.
• Allocation-free hot path. After a key has been seen once, Add does not allocate; the per-key context.Context (with tags already applied) is memoized and reused across flushes.
• Non-blocking flush. Each shard's map is swapped out under its own lock so writers are never blocked by a flush in progress.

Installation

go get github.com/mresti/opencensus-go-metrics-optimized

Requires Go 1.26+ (uses generics and sync.WaitGroup.Go).

Quick start

1. Define your label key and a Schema[K] for it:

package main

import "go.opencensus.io/tag"

type HTTPLabels struct {
	User   string
	Route  string
	Status string
}

type HTTPSchema struct {
	KeyUser   tag.Key
	KeyRoute  tag.Key
	KeyStatus tag.Key
}

// Hash is called on every Add: keep it cheap and non-allocating.
func (HTTPSchema) Hash(k HTTPLabels) uint64 {
	return hashStrings(k.User, k.Route, k.Status)
}

// Mutators is called only on flush, once per distinct combination.
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),
	}
}

1. Create an aggregator and record events against it instead of calling stats.Record directly:

package main

import (
	"time"

	oc "github.com/mresti/opencensus-go-metrics-optimized"
	"go.opencensus.io/stats"
)

var requestCount = stats.Float64("myapp/requests", "HTTP requests", stats.UnitDimensionless)

func main() {
	schema := HTTPSchema{ /* register tag.Key's via tag.NewKey */ }

	agg := oc.NewCountAggregator(oc.CountConfig[HTTPLabels, float64]{
		Config: oc.Config[HTTPLabels]{
			Shards:   16,               // default 16, rounded up to a power of 2
			Interval: 10 * time.Second, // default 10s
			Schema:   schema,
		},
		CountMeasure: requestCount,
	})
	defer agg.Stop() // flushes any remaining data before returning

	agg.Add(HTTPLabels{User: "alice", Route: "/orders", Status: "200"}, 1)
}

Stop performs a final flush before returning, so no data is lost on shutdown.

Aggregator variants

All variants implement the same interface:

type Aggregator[K comparable, N Number] interface {
	Add(k K, value N)
	Stop()
}

Constructor	Config	Use case
NewCountAggregator[K, N]	CountConfig[K, N]	Counting occurrences per key (equivalent to view.Count()).
NewSumAggregator[K, N]	SumConfig[K, N]	Summing a numeric value per key (equivalent to view.Sum()).
NewDistributionAggregator[K, N]	DistributionConfig[K, N]	Latency/size histograms; supports MaxSamplesPerKey reservoir sampling to bound memory under high/unbounded cardinality per key.
NewLastValueAggregator[K, N]	LastValueConfig[K, N]	Gauges; the measure's view must use view.LastValue().

The value type N is inferred from the config literal, e.g. SumConfig[HTTPLabels, float64] (Float64 measure) or SumConfig[HTTPLabels, int64] (Int64 measure).

Configuration

Config[K] is embedded in every variant's config struct:

type Config[K comparable] struct {
	Shards   int           // rounded up to a power of 2. Default 16.
	Interval time.Duration // flush cadence. Default 10s.
	Schema   Schema[K]     // key -> OpenCensus projection strategy
}

• Shards trades memory/lock granularity for contention: more shards means less contention under concurrent Add calls from many goroutines.
• Interval trades staleness for the volume of stats.Record traffic sent to the OpenCensus worker (default 10s). Set it to an exact divisor of the exporter reporting period (view.SetReportingPeriod) and at most half of it — e.g. 10s or 15s for a 30s period; 10s also divides the common 60s period. A non-divisor interval causes sawtooth in delta/rate charts. Aggregators apply a random startup delay in [0, Interval) so instances created together don't all flush on the same tick and burst the OpenCensus worker.

Design notes

• The per-key context.Context cache (ctxCache) grows with the number of distinct keys observed. It is stable under bounded label cardinality; for unbounded/high-cardinality label sets, keep K's cardinality bounded upstream (e.g. bucket free-form values) or clear the aggregator periodically.
• Distribution's MaxSamplesPerKey (when > 0) uses reservoir sampling so memory per key is bounded regardless of how many samples arrive between flushes, trading exactness for a bounded, uniformly-sampled subset.
• Concurrency-sensitive internals (sharded store, contract behavior across the three variants) are covered by a shared contract test suite and fuzz tests; see Testing.

Testing

The project uses make for all common workflows — run make help to list targets. The most relevant ones:

make test           # go test -race -cover ./...
make test-bench     # benchmarks across the aggregator variants
make test-fuzz       # replay the committed Fuzz* seed corpus (fast, CI-safe)
make test-fuzz-one FUZZ=FuzzCountAggregator_Add FUZZTIME=1m  # fuzz one target
make lint            # golangci-lint (run `make tools` once to install it)
make ci              # vet + lint + test — what CI runs

Contributing

Issues and pull requests are welcome. Please run make ci locally before submitting a PR — the same target gates CI on GitHub Actions.

License

Licensed under either of Apache License, Version 2.0 (LICENSE or http://www.apache.org/licenses/LICENSE-2.0)

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). 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
  • func NewCountAggregator[K comparable, N Number](cfg CountConfig[K, N]) *CountAggregator[K, N]
  • func (a *CountAggregator[K, N]) Add(k K, _ N)
  • func (a *CountAggregator[K, N]) Stop()
• type CountConfig
• type DistributionAggregator
  • func NewDistributionAggregator[K comparable, N Number](cfg DistributionConfig[K, N]) *DistributionAggregator[K, N]
  • func (a *DistributionAggregator[K, N]) Add(k K, value N)
  • func (a *DistributionAggregator[K, N]) Stop()
• type DistributionConfig
• type LastValueAggregator
  • func NewLastValueAggregator[K comparable, N Number](cfg LastValueConfig[K, N]) *LastValueAggregator[K, N]
  • func (a *LastValueAggregator[K, N]) Add(k K, value N)
  • func (a *LastValueAggregator[K, N]) Stop()
• type LastValueConfig
• type Measure
• type Number
• type Schema
• type SumAggregator
  • func NewSumAggregator[K comparable, N Number](cfg SumConfig[K, N]) *SumAggregator[K, N]
  • func (a *SumAggregator[K, N]) Add(k K, value N)
  • func (a *SumAggregator[K, N]) Stop()
• type SumConfig

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

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 Number

type Number interface{ ~int64 | ~float64 }

Number are the value types supported by OpenCensus measures.

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.

type SumConfig

type SumConfig[K comparable, N Number] struct {
	Config[K]
	SumMeasure Measure[N]
}

SumConfig configures a SumAggregator, pairing the shared Config with the measures used to record the accumulated sum and count.