[wip] feat(observability/metrics): add Emitter, options, buckets, and shared utils

observability/metrics/BUILD.bazel

@@ -0,0 +1,28 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "metrics",
+    srcs = [
+        "buckets.go",
+        "emitter.go",
+        "names.go",
+        "options.go",
+    ],
+    importpath = "github.com/uber/tango/observability/metrics",
+    visibility = ["//visibility:public"],
+    deps = ["@com_github_uber_go_tally//:tally"],
+)
+
+go_test(
+    name = "metrics_test",
+    srcs = [
+        "emitter_test.go",
+        "inflight_test.go",
+    ],
+    embed = [":metrics"],
+    deps = [
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally//:tally",
+    ],
+)

observability/metrics/buckets.go

@@ -0,0 +1,25 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"time"
+
+	"github.com/uber-go/tally"
+)
+
+var _defaultDurationBuckets = tally.MustMakeExponentialDurationBuckets(time.Millisecond, 2.0, 27)
+
+var _defaultCountBuckets = tally.MustMakeExponentialValueBuckets(1.0, 2.0, 24)

observability/metrics/emitter.go

@@ -0,0 +1,141 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package metrics is a thin wrapper over tally.Scope that pins the naming,
+// bucket, and tag conventions defined in the Tango Metrics Inventory.
+package metrics
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/uber-go/tally"
+)
+
+// Emitter is the interface for emitting metrics.
+type Emitter interface {
+	Inc(op, name string, opts ...Option)
+	Gauge(op, name string, v float64, opts ...Option)
+	RecordDur(op, name string, d time.Duration, opts ...Option)
+	RecordCount(op, name string, v int64, opts ...Option)
+	TrackInFlight(op string) func()
+	Tagged(tags map[string]string) Emitter
+}
+
+type defaultEmitter struct {
+	scope           tally.Scope
+	baseTags        map[string]string
+	durationBuckets tally.DurationBuckets
+	valueBuckets    tally.ValueBuckets
+	inFlight        *sync.Map
+}
+
+// New returns a tally-backed Emitter. A nil scope falls back to
+// tally.NoopScope so callers do not need to special-case unwired metrics.
+func New(scope tally.Scope) Emitter {
+	if scope == nil {
+		scope = tally.NoopScope
+	}
+	return &defaultEmitter{
+		scope:           scope,
+		durationBuckets: _defaultDurationBuckets,
+		valueBuckets:    _defaultCountBuckets,
+		inFlight:        &sync.Map{},
+	}
+}
+
+func (e *defaultEmitter) Tagged(tags map[string]string) Emitter {
+	if len(tags) == 0 {
+		return e
+	}
+	merged := make(map[string]string, len(e.baseTags)+len(tags))
+	for k, v := range e.baseTags {
+		merged[k] = v
+	}
+	for k, v := range tags {
+		merged[k] = v
+	}
+	return &defaultEmitter{
+		scope:           e.scope,
+		baseTags:        merged,
+		durationBuckets: e.durationBuckets,
+		valueBuckets:    e.valueBuckets,
+		inFlight:        e.inFlight,
+	}
+}
+
+func (e *defaultEmitter) Inc(op, name string, opts ...Option) {
+	o := e.applyOptions(opts)
+	e.subscope(op, o.tags).Counter(name).Inc(1)
+}
+
+func (e *defaultEmitter) Gauge(op, name string, v float64, opts ...Option) {
+	o := e.applyOptions(opts)
+	e.subscope(op, o.tags).Gauge(name).Update(v)
+}
+
+func (e *defaultEmitter) RecordDur(op, name string, d time.Duration, opts ...Option) {
+	o := e.applyOptions(opts)
+	buckets := o.durationBuckets
+	if buckets == nil {
+		buckets = e.durationBuckets
+	}
+	e.subscope(op, o.tags).Histogram(name, buckets).RecordDuration(d)
+}
+
+func (e *defaultEmitter) RecordCount(op, name string, v int64, opts ...Option) {
+	o := e.applyOptions(opts)
+	buckets := o.valueBuckets
+	if buckets == nil {
+		buckets = e.valueBuckets
+	}
+	e.subscope(op, o.tags).Histogram(name, buckets).RecordValue(float64(v))
+}
+
+func (e *defaultEmitter) subscope(op string, callTags map[string]string) tally.Scope {
+	s := e.scope.SubScope(op)
+	if len(e.baseTags) == 0 && len(callTags) == 0 {
+		return s
+	}
+	merged := make(map[string]string, len(e.baseTags)+len(callTags))
+	for k, v := range e.baseTags {
+		merged[k] = v
+	}
+	for k, v := range callTags {
+		merged[k] = v
+	}
+	return s.Tagged(merged)
+}
+
+func (e *defaultEmitter) applyOptions(opts []Option) emitOpts {
+	var o emitOpts
+	for _, opt := range opts {
+		opt.apply(&o)
+	}
+	return o
+}
+
+func (e *defaultEmitter) TrackInFlight(op string) func() {
+	v, _ := e.inFlight.LoadOrStore(op, new(int64))
+	counter := v.(*int64)
+	e.emitInFlight(op, atomic.AddInt64(counter, 1))
+	return func() {
+		e.emitInFlight(op, atomic.AddInt64(counter, -1))
+	}
+}
+
+func (e *defaultEmitter) emitInFlight(op string, n int64) {
+	e.scope.Tagged(map[string]string{TagOperation: op}).Gauge(InFlightRequests).Update(float64(n))
+}

observability/metrics/emitter_test.go

@@ -0,0 +1,182 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+)
+
+func counterValue(t *testing.T, s tally.TestScope, name string, tags map[string]string) int64 {
+	t.Helper()
+	for _, c := range s.Snapshot().Counters() {
+		if c.Name() == name && tagsEqual(c.Tags(), tags) {
+			return c.Value()
+		}
+	}
+	return 0
+}
+
+func gaugeValue(t *testing.T, s tally.TestScope, name string, tags map[string]string) (float64, bool) {
+	t.Helper()
+	for _, g := range s.Snapshot().Gauges() {
+		if g.Name() == name && tagsEqual(g.Tags(), tags) {
+			return g.Value(), true
+		}
+	}
+	return 0, false
+}
+
+func histogramDurationSamples(t *testing.T, s tally.TestScope, name string, tags map[string]string) int {
+	t.Helper()
+	total := 0
+	for _, h := range s.Snapshot().Histograms() {
+		if h.Name() != name || !tagsEqual(h.Tags(), tags) {
+			continue
+		}
+		for _, n := range h.Durations() {
+			total += int(n)
+		}
+	}
+	return total
+}
+
+func histogramValueSamples(t *testing.T, s tally.TestScope, name string, tags map[string]string) int {
+	t.Helper()
+	total := 0
+	for _, h := range s.Snapshot().Histograms() {
+		if h.Name() != name || !tagsEqual(h.Tags(), tags) {
+			continue
+		}
+		for _, n := range h.Values() {
+			total += int(n)
+		}
+	}
+	return total
+}
+
+func tagsEqual(a, b map[string]string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for k, v := range a {
+		if b[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
+func TestNewNilScopeIsNoop(t *testing.T) {
+	e := New(nil)
+	require.NotNil(t, e)
+	e.Inc("op", "n")
+	e.Gauge("op", "n", 1)
+	e.RecordDur("op", "n", time.Millisecond)
+	e.RecordCount("op", "n", 42)
+}
+
+func TestIncEmitsUnderOpSubscope(t *testing.T) {
+	s := tally.NewTestScope("root", nil)
+	e := New(s)
+	e.Inc("get_target_graph", "requests")
+	assert.Equal(t, int64(1), counterValue(t, s, "root.get_target_graph.requests", map[string]string{}))
+}
+
+func TestIncMergesTags(t *testing.T) {
+	s := tally.NewTestScope("", nil)
+	e := New(s).Tagged(map[string]string{TagEmitter: "service", TagRepo: "acme/monorepo"})
+	e.Inc("op", "requests", WithTags(map[string]string{TagResult: string(ResultSuccess)}))
+	assert.Equal(t, int64(1), counterValue(t, s, "op.requests", map[string]string{
+		TagEmitter: "service",
+		TagRepo:    "acme/monorepo",
+		TagResult:  string(ResultSuccess),
+	}))
+	e.Inc("op", "requests",
+		WithTags(map[string]string{TagResult: string(ResultSuccess)}),
+		WithTags(map[string]string{TagResult: string(ResultFail)}),
+	)
+	assert.Equal(t, int64(1), counterValue(t, s, "op.requests", map[string]string{
+		TagEmitter: "service",
+		TagRepo:    "acme/monorepo",
+		TagResult:  string(ResultFail),
+	}))
+}
+
+func TestTaggedDoesNotMutateParent(t *testing.T) {
+	s := tally.NewTestScope("", nil)
+	parent := New(s).Tagged(map[string]string{TagEmitter: "service"})
+	child := parent.Tagged(map[string]string{TagRepo: "acme"})
+
+	child.Inc("op", "requests")
+	parent.Inc("op", "requests")
+
+	assert.Equal(t, int64(1), counterValue(t, s, "op.requests", map[string]string{
+		TagEmitter: "service",
+		TagRepo:    "acme",
+	}))
+	assert.Equal(t, int64(1), counterValue(t, s, "op.requests", map[string]string{
+		TagEmitter: "service",
+	}))
+}
+
+func TestTaggedEmptyReturnsSameEmitter(t *testing.T) {
+	e := New(nil)
+	assert.Same(t, e, e.Tagged(nil))
+	assert.Same(t, e, e.Tagged(map[string]string{}))
+}
+
+func TestGaugeUpdatesValue(t *testing.T) {
+	s := tally.NewTestScope("", nil)
+	e := New(s)
+	e.Gauge("op", "in_flight_requests", 3, WithTags(map[string]string{TagOperation: "op"}))
+	v, ok := gaugeValue(t, s, "op.in_flight_requests", map[string]string{TagOperation: "op"})
+	require.True(t, ok)
+	assert.Equal(t, float64(3), v)
+}
+
+func TestRecordDurUsesDefaultBuckets(t *testing.T) {
+	s := tally.NewTestScope("", nil)
+	e := New(s)
+	e.RecordDur("op", "total_duration", 5*time.Millisecond)
+	assert.Equal(t, 1, histogramDurationSamples(t, s, "op.total_duration", map[string]string{}))
+}
+
+func TestRecordDurBucketsOverride(t *testing.T) {
+	custom := tally.MustMakeLinearDurationBuckets(time.Millisecond, time.Millisecond, 5)
+	s := tally.NewTestScope("", nil)
+	e := New(s)
+	e.RecordDur("op", "n", 2*time.Millisecond, WithDurationBuckets(custom))
+	assert.Equal(t, 1, histogramDurationSamples(t, s, "op.n", map[string]string{}))
+}
+
+func TestRecordCountUsesDefaultBuckets(t *testing.T) {
+	s := tally.NewTestScope("", nil)
+	e := New(s)
+	e.RecordCount("op", "targets_count", 128)
+	assert.Equal(t, 1, histogramValueSamples(t, s, "op.targets_count", map[string]string{}))
+}
+
+func TestRecordCountBucketsOverride(t *testing.T) {
+	custom := tally.MustMakeLinearValueBuckets(0, 1, 5)
+	s := tally.NewTestScope("", nil)
+	e := New(s)
+	e.RecordCount("op", "n", 2, WithValueBuckets(custom))
+	assert.Equal(t, 1, histogramValueSamples(t, s, "op.n", map[string]string{}))
+}