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: 20 * time.Second, // default 20s
			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 20s.
	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.

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 time.Duration // flush cadence. Default 20s.
	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.

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.