Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -33,3 +33,32 @@ stats := statter.New(reporter, 10*time.Second).With("my-prefix")

stats.Counter("my-counter", tags.Str("tag", "value")).Inc(1)
```

### Scopes

A `Scope` is a sub-statter identified by an id, which tracks the metrics created
through it.

Requesting a scope with an existing id but different tags removes the previous
scope and every metric created through it. This keeps a metric unique for a set
of tags, such as a gauge carrying a revision that changes over time:

```go
stats.Scope("build-info", tags.Str("revision", rev)).Gauge("info").Set(1)
```

The tracked metrics can also be removed together as a batch:

```go
scope := stats.Scope("tenant:"+id, tags.Str("tenant", id))
scope.Counter("requests").Inc(1)
scope.Timing("latency").Observe(d)

scope.Delete() // Both metrics are removed.
```

Use `HasScope` to determine if a scope currently exists.

**Note:** Metrics are only removed from the backend if the reporter supports
removal, which the Prometheus and VictoriaMetrics reporters do. Otherwise
deletion only stops local aggregation.
45 changes: 45 additions & 0 deletions bench_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,51 @@ func BenchmarkStatter_Timing(b *testing.B) {
_ = s.Close()
}

func BenchmarkScope_Resolve(b *testing.B) {
s := statter.New(discardReporter{}, time.Second)
s.Scope("scope", tags.Str("rev", "abc"))

b.ReportAllocs()
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
s.Scope("scope", tags.Str("rev", "abc"))
}
})

b.StopTimer()
_ = s.Close()
}

func BenchmarkScope_Gauge(b *testing.B) {
s := statter.New(discardReporter{}, time.Second)
sc := s.Scope("scope", tags.Str("rev", "abc"))

b.ReportAllocs()
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
sc.Gauge("test", tags.Str("test", "test")).Set(1)
}
})

b.StopTimer()
_ = s.Close()
}

func BenchmarkScope_Rotate(b *testing.B) {
s := statter.New(discardReporter{}, time.Second)

b.ReportAllocs()
b.ResetTimer()
for i := 0; b.Loop(); i++ {
s.Scope("scope", tags.Int("rev", i)).Gauge("test").Set(1)
}

b.StopTimer()
_ = s.Close()
}

func BenchmarkStatter_PrometheusHistogram(b *testing.B) {
s := statter.New(prometheus.New("test"), time.Second)

Expand Down
11 changes: 10 additions & 1 deletion doc.go
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,16 @@
// with identical resolved prefix and tags are deduplicated and share the same
// instance.
//
// The [Statter.Scope] method returns a [Scope], a sub-statter identified by a
// caller supplied id. A Scope tracks the metrics created through it, so they
// can be removed together with [Scope.Delete]. Requesting a Scope with an
// existing id but different tags deletes the previous Scope and its metrics,
// which keeps metrics unique for a set of tags, such as a gauge carrying a
// revision that changes over time.
//
// The [Reporter] interface is the only contract that backend adapters must
// satisfy. Richer adapters may additionally implement [HistogramReporter],
// [TimingReporter], and the corresponding Removable* interfaces.
// [TimingReporter], and the corresponding Removable* interfaces. Metrics are
// only removed from the backend if the reporter implements the relevant
// Removable* interface, otherwise deletion only stops local aggregation.
package statter
19 changes: 19 additions & 0 deletions example_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,3 +24,22 @@ func ExampleHistogram_Observe() {

stat.Histogram("my_histo", tags.Str("label", "blah")).Observe(2.34)
}

func ExampleStatter_Scope() {
stat := statter.New(statter.DiscardReporter, time.Second)

// The gauge is unique for the given revision. When the revision changes,
// the gauge reported for the previous revision is removed.
stat.Scope("build-info", tags.Str("revision", "abc123")).Gauge("info").Set(1)
}

func ExampleScope_Delete() {
stat := statter.New(statter.DiscardReporter, time.Second)

scope := stat.Scope("tenant:1", tags.Str("tenant", "1"))
scope.Counter("requests").Inc(1)
scope.Timing("latency").Observe(time.Second)

// Both the counter and the timing are removed.
scope.Delete()
}
38 changes: 38 additions & 0 deletions registry.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@ type registry struct {
gauges hashtriemap.HashTrieMap[string, *Gauge]
histograms hashtriemap.HashTrieMap[string, *Histogram]
timings hashtriemap.HashTrieMap[string, *Timing]
scopes hashtriemap.HashTrieMap[string, *Scope]

reportMu sync.RWMutex

mu sync.RWMutex
root *Statter
Expand Down Expand Up @@ -76,6 +79,9 @@ func (r *registry) runReportLoop(d time.Duration) {
}

func (r *registry) report() {
r.reportMu.Lock()
defer r.reportMu.Unlock()

r.counters.Range(func(_ string, c *Counter) bool {
val := c.value()
if val == 0 {
Expand Down Expand Up @@ -181,6 +187,38 @@ func (r *registry) SubStatter(parent *Statter, prefix string, tags []Tag) *Statt
return s
}

// Scope returns the scope with the given id, creating it if needed. If a scope
// with the id already exists under a different parent or with different tags,
// it and all metrics created through it are deleted and replaced.
func (r *registry) Scope(parent *Statter, id string, tags []Tag) *Scope {
for {
cur, ok := r.scopes.Load(id)
if ok && cur.matches(parent, tags) {
return cur
}

next := newScope(r, parent, id, tags)

if !ok {
if _, loaded := r.scopes.LoadOrStore(id, next); !loaded {
return next
}
continue
}

if r.scopes.CompareAndSwap(id, cur, next) {
cur.delete(false)
return next
}
}
}

// HasScope determines if a scope with the given id exists.
func (r *registry) HasScope(id string) bool {
_, ok := r.scopes.Load(id)
return ok
}

// Close closes the registry if the caller is the root statter,
// otherwise an error is returned.
func (r *registry) Close(caller *Statter) error {
Expand Down
154 changes: 154 additions & 0 deletions scope.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
package statter

import (
"sync"
"sync/atomic"
)

// deletable is a metric that can remove itself from the registry and the
// reporter.
type deletable interface {
Delete()
}

// Scope is a sub-statter identified by a unique id.
//
// Metrics created through a Scope are tracked by it, and are removed together
// when the Scope is deleted or replaced. See [Statter.Scope].
type Scope struct {
reg *registry
id string
parent *Statter
tags []Tag
s *Statter

deleted atomic.Bool

mu sync.Mutex
metrics []deletable
}

func newScope(reg *registry, parent *Statter, id string, tags []Tag) *Scope {
// The tags are owned by the caller, so they must be copied before they are
// merged and sorted.
rawTags := make([]Tag, len(tags))
copy(rawTags, tags)

name, scopeTags := mergeDescriptors(parent.prefix, reg.cfg.separator, "", parent.tags, rawTags)

return &Scope{
reg: reg,
id: id,
parent: parent,
tags: rawTags,
// The scope statter is intentionally not registered with the registry
// statter cache, as that cache is never evicted.
s: &Statter{
reg: reg,
prefix: name,
tags: scopeTags,
},
}
}

// matches determines if the scope was created from the given parent and tags.
//
// The tags are compared as given, rather than as merged, as identical input on
// the same parent always resolves to the same scope. This keeps the comparison
// allocation free.
func (sc *Scope) matches(parent *Statter, tags []Tag) bool {
if sc.parent != parent || len(sc.tags) != len(tags) {
return false
}
for i, tag := range tags {
if sc.tags[i] != tag {
return false
}
}
return true
}
Comment on lines +59 to +69

// FullName returns the full name with prefix for the given name.
func (sc *Scope) FullName(name string) string {
return sc.s.FullName(name)
}

// Counter returns a counter for the given name and tags, tracking it against
// the scope.
func (sc *Scope) Counter(name string, tags ...Tag) *Counter {
c, created := sc.s.counter(name, tags)
if created {
sc.track(c)
}
return c
}

// Gauge returns a gauge for the given name and tags, tracking it against the
// scope.
func (sc *Scope) Gauge(name string, tags ...Tag) *Gauge {
g, created := sc.s.gauge(name, tags)
if created {
sc.track(g)
}
return g
}

// Histogram returns a histogram for the given name and tags, tracking it
// against the scope.
func (sc *Scope) Histogram(name string, tags ...Tag) *Histogram {
h, created := sc.s.histogram(name, tags)
if created {
sc.track(h)
}
return h
}

// Timing returns a timing for the given name and tags, tracking it against the
// scope.
func (sc *Scope) Timing(name string, tags ...Tag) *Timing {
t, created := sc.s.timing(name, tags)
if created {
sc.track(t)
}
return t
}

func (sc *Scope) track(m deletable) {
sc.mu.Lock()
if sc.deleted.Load() {
sc.mu.Unlock()
m.Delete()
return
}
sc.metrics = append(sc.metrics, m)
sc.mu.Unlock()
}

// Delete removes the scope and all metrics created through it.
//
// Delete is idempotent. Metrics requested from a deleted scope are returned to
// the caller but are not tracked, and are removed immediately.
func (sc *Scope) Delete() {
sc.delete(true)
}

func (sc *Scope) delete(unregister bool) {
if !sc.deleted.CompareAndSwap(false, true) {
return
}

if unregister {
_ = sc.reg.scopes.CompareAndDelete(sc.id, sc)
}

sc.mu.Lock()
metrics := sc.metrics
sc.metrics = nil
sc.mu.Unlock()

// The metrics take the registry report lock themselves, so it must not be
// held here. A sync.RWMutex is not reentrant.
for _, m := range metrics {
m.Delete()
}
}
Loading