From d63fdd9feb1b5ffd9d49fea4a7373b83a2a7d4db Mon Sep 17 00:00:00 2001 From: Xiaoyang Tan Date: Tue, 7 Jul 2026 16:45:38 -0700 Subject: [PATCH] feat(observability/metrics): add Emitter, options, buckets, and shared names Emitter is an interface; the default implementation wraps tally.Scope with op-as-subscope semantics, base tags via Tagged, default histogram buckets, and TrackInFlight. New(nil) is safe (NoopScope). --- observability/metrics/BUILD.bazel | 28 ++++ observability/metrics/buckets.go | 25 ++++ observability/metrics/emitter.go | 141 +++++++++++++++++++ observability/metrics/emitter_test.go | 182 +++++++++++++++++++++++++ observability/metrics/inflight_test.go | 97 +++++++++++++ observability/metrics/names.go | 71 ++++++++++ observability/metrics/options.go | 60 ++++++++ 7 files changed, 604 insertions(+) create mode 100644 observability/metrics/BUILD.bazel create mode 100644 observability/metrics/buckets.go create mode 100644 observability/metrics/emitter.go create mode 100644 observability/metrics/emitter_test.go create mode 100644 observability/metrics/inflight_test.go create mode 100644 observability/metrics/names.go create mode 100644 observability/metrics/options.go diff --git a/observability/metrics/BUILD.bazel b/observability/metrics/BUILD.bazel new file mode 100644 index 0000000..b4ec2ff --- /dev/null +++ b/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", + ], +) diff --git a/observability/metrics/buckets.go b/observability/metrics/buckets.go new file mode 100644 index 0000000..6924751 --- /dev/null +++ b/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) diff --git a/observability/metrics/emitter.go b/observability/metrics/emitter.go new file mode 100644 index 0000000..08c9e25 --- /dev/null +++ b/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)) +} diff --git a/observability/metrics/emitter_test.go b/observability/metrics/emitter_test.go new file mode 100644 index 0000000..1e7e38f --- /dev/null +++ b/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{})) +} diff --git a/observability/metrics/inflight_test.go b/observability/metrics/inflight_test.go new file mode 100644 index 0000000..5240016 --- /dev/null +++ b/observability/metrics/inflight_test.go @@ -0,0 +1,97 @@ +// 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 ( + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" +) + +func TestTrackInFlightBalances(t *testing.T) { + s := tally.NewTestScope("", nil) + e := New(s) + release1 := e.TrackInFlight(OpGetTargetGraph) + release2 := e.TrackInFlight(OpGetTargetGraph) + + v, ok := gaugeValue(t, s, InFlightRequests, map[string]string{TagOperation: OpGetTargetGraph}) + require.True(t, ok) + assert.Equal(t, float64(2), v) + + release1() + release2() + + v, _ = gaugeValue(t, s, InFlightRequests, map[string]string{TagOperation: OpGetTargetGraph}) + assert.Equal(t, float64(0), v) +} + +func TestTrackInFlightSeparateOps(t *testing.T) { + s := tally.NewTestScope("", nil) + e := New(s) + defer e.TrackInFlight(OpGetTargetGraph)() + defer e.TrackInFlight(OpGetChangedTargets)() + defer e.TrackInFlight(OpGetChangedTargets)() + + v1, _ := gaugeValue(t, s, InFlightRequests, map[string]string{TagOperation: OpGetTargetGraph}) + v2, _ := gaugeValue(t, s, InFlightRequests, map[string]string{TagOperation: OpGetChangedTargets}) + assert.Equal(t, float64(1), v1) + assert.Equal(t, float64(2), v2) +} + +// A Tagged child must share the parent's counter — otherwise gauges leak if a +// handler tags between the increment and the deferred release. +func TestTrackInFlightSharedAcrossTaggedChildren(t *testing.T) { + s := tally.NewTestScope("", nil) + parent := New(s) + child := parent.Tagged(map[string]string{TagRepo: "acme"}) + + release := parent.TrackInFlight(OpGetTargetGraph) + release2 := child.TrackInFlight(OpGetTargetGraph) + + v, _ := gaugeValue(t, s, InFlightRequests, map[string]string{TagOperation: OpGetTargetGraph}) + assert.Equal(t, float64(2), v) + + release() + release2() + + v, _ = gaugeValue(t, s, InFlightRequests, map[string]string{TagOperation: OpGetTargetGraph}) + assert.Equal(t, float64(0), v) +} + +func TestTrackInFlightConcurrent(t *testing.T) { + s := tally.NewTestScope("", nil) + e := New(s) + + const workers = 32 + const iterations = 500 + + var wg sync.WaitGroup + wg.Add(workers) + for i := 0; i < workers; i++ { + go func() { + defer wg.Done() + for j := 0; j < iterations; j++ { + e.TrackInFlight(OpGetTargetGraph)() + } + }() + } + wg.Wait() + + v, _ := gaugeValue(t, s, InFlightRequests, map[string]string{TagOperation: OpGetTargetGraph}) + assert.Equal(t, float64(0), v) +} diff --git a/observability/metrics/names.go b/observability/metrics/names.go new file mode 100644 index 0000000..463a723 --- /dev/null +++ b/observability/metrics/names.go @@ -0,0 +1,71 @@ +// 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 + +// Operation names. Internal-only ops should be declared next to their emit +// site rather than added here. +const ( + OpGetTargetGraph = "get_target_graph" + OpGetChangedTargets = "get_changed_targets" + OpGetChangedTargetGraph = "get_changed_target_graph" + OpGetGraph = "get_graph" + OpCompareTargetGraphs = "compare_target_graphs" + OpNativeOrchestrator = "native_orchestrator" + OpGraphRunner = "graph_runner" +) + +// Metric names. +const ( + Requests = "requests" + TreehashCacheLookup = "treehash_cache_lookup" + GraphCacheLookup = "graph_cache_lookup" + ChangedTargetsCacheLookup = "changed_targets_cache_lookup" + GraphCacheFetchDuration = "graph_cache_fetch_duration" + ChangedTargetsCacheFetchDuration = "changed_targets_cache_fetch_duration" + CompareDuration = "compare_duration" + ChangedTargetsCount = "changed_targets_count" + TotalDuration = "total_duration" + Patch = "patch" + PatchDuration = "patch_duration" + BazelQueryDuration = "bazel_query_duration" + GitFileHashesDuration = "git_file_hashes_duration" + TargetHashDuration = "target_hash_duration" + TargetsCount = "targets_count" + StorageUploadDuration = "storage_upload_duration" + InFlightRequests = "in_flight_requests" + WorkspacesInFlight = "in_flight_workspaces" +) + +// Tag keys. +const ( + TagRepo = "repo" + TagEmitter = "emitter" + TagResult = "result" + TagOperation = "operation" + TagFailureType = "failure_type" + TagFailureReason = "failure_reason" +) + +// Result is the value type for TagResult. +type Result string + +// Result values for TagResult. +const ( + ResultUnknown Result = "unknown" + ResultSuccess Result = "success" + ResultFail Result = "fail" + ResultHit Result = "hit" + ResultMiss Result = "miss" +) diff --git a/observability/metrics/options.go b/observability/metrics/options.go new file mode 100644 index 0000000..76e3000 --- /dev/null +++ b/observability/metrics/options.go @@ -0,0 +1,60 @@ +// 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 "github.com/uber-go/tally" + +// Option customizes a single emission. Options compose; later WithTags +// override earlier on key collision. +type Option interface { + apply(*emitOpts) +} + +type emitOpts struct { + tags map[string]string + durationBuckets tally.DurationBuckets + valueBuckets tally.ValueBuckets +} + +type optFunc func(*emitOpts) + +func (f optFunc) apply(o *emitOpts) { f(o) } + +// WithTags attaches key/value tags to a single emission. +func WithTags(tags map[string]string) Option { + return optFunc(func(o *emitOpts) { + if len(tags) == 0 { + return + } + if o.tags == nil { + o.tags = make(map[string]string, len(tags)) + } + for k, v := range tags { + o.tags[k] = v + } + }) +} + +// WithDurationBuckets overrides the emitter's default duration buckets for +// a single RecordDur call. +func WithDurationBuckets(b tally.DurationBuckets) Option { + return optFunc(func(o *emitOpts) { o.durationBuckets = b }) +} + +// WithValueBuckets overrides the emitter's default value buckets for a +// single RecordCount call. +func WithValueBuckets(b tally.ValueBuckets) Option { + return optFunc(func(o *emitOpts) { o.valueBuckets = b }) +}