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.
• Four aggregator variants behind one interface (Aggregator[K]): 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]{
		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] interface {
	Add(k K, value float64)
	Stop()
}

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

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-APACHE or http://www.apache.org/licenses/LICENSE-2.0)
• MIT license (LICENSE-MIT or http://opensource.org/licenses/MIT)

at your option.

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

Types

type Aggregator

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

Aggregator is the common interface, generic over the labels key K.

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] 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](cfg CountConfig[K]) *CountAggregator[K]

NewCountAggregator builds a CountAggregator from cfg, applying defaults and starting the background flusher.

func (*CountAggregator[K]) Add

func (a *CountAggregator[K]) Add(k K, _ float64)

Add adds value to the running increments its count.

func (*CountAggregator[K]) Stop

func (a *CountAggregator[K]) Stop()

Stop halts the background flusher.

type CountConfig

type CountConfig[K comparable] struct {
	Config[K]
	CountMeasure *stats.Float64Measure
}

CountConfig configures a SumCountAggregator, pairing the shared Config with the measures used to record the accumulated count.

type DistributionAggregator

type DistributionAggregator[K comparable] 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](cfg DistributionConfig[K]) *DistributionAggregator[K]

NewDistributionAggregator builds a DistributionAggregator from cfg, applying defaults and starting the background flusher.

func (*DistributionAggregator[K]) Add

func (a *DistributionAggregator[K]) Add(k K, value float64)

Add records value as a sample for k, using reservoir sampling once the per-key sample cap is reached.

func (*DistributionAggregator[K]) Stop

func (a *DistributionAggregator[K]) Stop()

Stop halts the background flusher.

type DistributionConfig

type DistributionConfig[K comparable] struct {
	Config[K]
	Measure          *stats.Float64Measure
	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] 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](cfg LastValueConfig[K]) *LastValueAggregator[K]

NewLastValueAggregator builds a LastValueAggregator from cfg, applying defaults and starting the background flusher.

func (*LastValueAggregator[K]) Add

func (a *LastValueAggregator[K]) Add(k K, value float64)

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]) Stop

func (a *LastValueAggregator[K]) Stop()

Stop halts the background flusher.

type LastValueConfig

type LastValueConfig[K comparable] struct {
	Config[K]
	// Measure must be backed by a view of type view.LastValue().
	Measure *stats.Float64Measure
}

LastValueConfig configures a LastValueAggregator: it embeds the shared Config and the measure whose view must be of type view.LastValue().

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] 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](cfg SumConfig[K]) *SumAggregator[K]

NewSumAggregator builds a SumCountAggregator from cfg, applying defaults and starting the background flusher.

func (*SumAggregator[K]) Add

func (a *SumAggregator[K]) Add(k K, value float64)

Add adds value to the running sum for k.

func (*SumAggregator[K]) Stop

func (a *SumAggregator[K]) Stop()

Stop halts the background flusher.

type SumConfig

type SumConfig[K comparable] struct {
	Config[K]
	SumMeasure *stats.Float64Measure
}

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