From b22d56c8d71710aba3dadb00ad6245e53155e9f6 Mon Sep 17 00:00:00 2001 From: Dylan Tinianov Date: Thu, 15 Oct 2026 09:37:35 -0400 Subject: [PATCH 1/3] resourcemanager: drop resource_id from the utilization gauge for usage records Usage record resource ids are per-execution, so labeling the gauge with them creates an unbounded series set per node. On CRE staging zone-a this grew by thousands of series per hour until the OTLP export exceeded the collector's 16 MiB limit and every platform_engine_* metric was dropped. Snapshots keep the label: their resource id is the workflow id, which is bounded. Co-Authored-By: Claude Fable 5.1 --- pkg/resourcemanager/resourcemanager.go | 25 +++-- pkg/resourcemanager/utilization_gauge_test.go | 97 +++++++++++++++++++ 2 files changed, 109 insertions(+), 13 deletions(-) create mode 100644 pkg/resourcemanager/utilization_gauge_test.go diff --git a/pkg/resourcemanager/resourcemanager.go b/pkg/resourcemanager/resourcemanager.go index 752867ad0d..9c623d563e 100644 --- a/pkg/resourcemanager/resourcemanager.go +++ b/pkg/resourcemanager/resourcemanager.go @@ -95,11 +95,8 @@ const ( snapshotEmitFailureCounterName = "resource_manager_snapshot_emit_failure_total" ) -// utilizationGaugeName is the per-resource utilization gauge. Unlike the -// low-cardinality emit counters, it is labeled with EVERY ResourceIdentity -// dimension (including node_id and resource_id) so dashboards can break -// utilization down by any dimension. That is high cardinality by construction; -// it is the intended, requested behavior. +// utilizationGaugeName is labeled with every ResourceIdentity dimension; resource_id +// is only attached for snapshots, since usage record ids are per-execution. const utilizationGaugeName = "resource_manager_utilization" // DefaultSnapshotInterval is the recommended snapshot period. It is NOT @@ -377,7 +374,7 @@ func (rm *ResourceManager) emitSnapshot(ctx context.Context, m Meterable, tickTi bucketUnix, ) } - rm.recordUtilization(ctx, e.Identity, u) + rm.recordUtilization(ctx, e.Identity, u, true) } snapshot := &meteringpb.MeterSnapshot{ @@ -483,7 +480,7 @@ func (rm *ResourceManager) emitRecord(ctx context.Context, identity ResourceIden u.EventId = eventID } utilizations := []*meteringpb.Utilization{u} - rm.recordUtilization(ctx, identity, u) + rm.recordUtilization(ctx, identity, u, false) if rm.emitter == nil { return @@ -526,9 +523,8 @@ func (rm *ResourceManager) emitRecord(ctx context.Context, identity ResourceIden rm.countRecord(ctx, rm.emitSuccess, identity, action) } -// recordUtilization records value to the per-resource utilization gauge, -// labeled with every ResourceIdentity dimension plus utilization identity. -func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIdentity, u *meteringpb.Utilization) { +// recordUtilization records value to the utilization gauge; withResourceID adds the resource_id label. +func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIdentity, u *meteringpb.Utilization, withResourceID bool) { if rm.utilization == nil { return } @@ -547,7 +543,7 @@ func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIde ) return } - rm.utilization.Record(ctx, value, metric.WithAttributes( + attrs := []attribute.KeyValue{ attribute.String("product", id.Product), attribute.String("tenant", id.Tenant), attribute.String("numeric_tenant_id", id.NumericTenantID), @@ -559,9 +555,12 @@ func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIde attribute.String("resource_pool", id.ResourcePool), attribute.String("resource_pool_id", id.ResourcePoolID), attribute.String("resource_type", u.GetResourceType()), - attribute.String("resource_id", u.GetResourceId()), attribute.String("org_id", u.GetOrgId()), - )) + } + if withResourceID { + attrs = append(attrs, attribute.String("resource_id", u.GetResourceId())) + } + rm.utilization.Record(ctx, value, metric.WithAttributes(attrs...)) } // countRecord records a MeterRecord emit outcome. Labels are intentionally diff --git a/pkg/resourcemanager/utilization_gauge_test.go b/pkg/resourcemanager/utilization_gauge_test.go new file mode 100644 index 0000000000..86bbd72712 --- /dev/null +++ b/pkg/resourcemanager/utilization_gauge_test.go @@ -0,0 +1,97 @@ +package resourcemanager + +import ( + "testing" + "time" + + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" + meteringpb "github.com/smartcontractkit/chainlink-protos/metering/go" +) + +func collectUtilizationPoints(t *testing.T, reader *sdkmetric.ManualReader) []metricdata.DataPoint[int64] { + t.Helper() + var data metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &data)) + for _, sm := range data.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == utilizationGaugeName { + return m.Data.(metricdata.Gauge[int64]).DataPoints + } + } + } + return nil +} + +func TestUtilizationGauge_ResourceIDLabel(t *testing.T) { + reader := sdkmetric.NewManualReader() + gauge, err := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test").Int64Gauge(utilizationGaugeName) + require.NoError(t, err) + + t.Run("usage records omit resource_id", func(t *testing.T) { + rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{MeterRecordsEnabled: true, Emitter: &fakeEmitter{}}) + rm.utilization = gauge + + for _, exec := range []string{"exec-1", "exec-2"} { + rm.EmitUsage(t.Context(), testIdentity, exec, 7, UtilizationFields{ + ResourceType: "cre:workflow:compute", + ResourceID: "wf-1:" + exec, + OrgID: "org-1", + }) + } + + points := collectUtilizationPoints(t, reader) + require.Len(t, points, 1) + _, hasID := points[0].Attributes.Value("resource_id") + assert.False(t, hasID) + rt, _ := points[0].Attributes.Value("resource_type") + assert.Equal(t, "cre:workflow:compute", rt.AsString()) + }) + + t.Run("snapshots keep resource_id", func(t *testing.T) { + emitter := &fakeEmitter{} + clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)) + rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{ + MeterRecordsEnabled: true, + MeterSnapshotsEnabled: true, + Emitter: emitter, + SnapshotInterval: 30 * time.Second, + Clock: clock, + }) + rm.utilization = gauge + + m := &fakeMeterable{identity: testIdentity} + m.set([]SnapshotEntry{{ + Identity: testIdentity, + Utilizations: []*meteringpb.Utilization{ + NewUtilizationInt(3, UtilizationFields{ResourceType: "operations", ResourceID: "trigger-1", OrgID: "org-1"}), + }, + }}) + unregister := rm.Register(m) + defer unregister() + + servicetest.Run(t, rm) + require.NoError(t, clock.BlockUntilContext(t.Context(), 1)) + clock.Advance(30 * time.Second) + waitForSnapshotCount(t, emitter, 1) + + var found bool + for _, p := range collectUtilizationPoints(t, reader) { + rt, _ := p.Attributes.Value("resource_type") + if rt.AsString() != "operations" { + continue + } + found = true + id, hasID := p.Attributes.Value("resource_id") + assert.True(t, hasID) + assert.Equal(t, "trigger-1", id.AsString()) + } + assert.True(t, found) + }) +} From 463eb552470cdfef48651f9f3466abae6e1f3216 Mon Sep 17 00:00:00 2001 From: Dylan Tinianov Date: Thu, 15 Oct 2026 10:00:50 -0400 Subject: [PATCH 2/3] resourcemanager: keep resource_id on the gauge for delta records Only usage records have per-execution ids. Deltas address durable resources and must share a series with the matching snapshot. Co-Authored-By: Claude Fable 5.1 --- pkg/resourcemanager/resourcemanager.go | 6 +- pkg/resourcemanager/utilization_gauge_test.go | 64 +++++++++++++++---- 2 files changed, 54 insertions(+), 16 deletions(-) diff --git a/pkg/resourcemanager/resourcemanager.go b/pkg/resourcemanager/resourcemanager.go index 9c623d563e..0edf2f8a2c 100644 --- a/pkg/resourcemanager/resourcemanager.go +++ b/pkg/resourcemanager/resourcemanager.go @@ -95,8 +95,8 @@ const ( snapshotEmitFailureCounterName = "resource_manager_snapshot_emit_failure_total" ) -// utilizationGaugeName is labeled with every ResourceIdentity dimension; resource_id -// is only attached for snapshots, since usage record ids are per-execution. +// utilizationGaugeName is labeled with every ResourceIdentity dimension; usage +// records omit resource_id because their ids are per-execution. const utilizationGaugeName = "resource_manager_utilization" // DefaultSnapshotInterval is the recommended snapshot period. It is NOT @@ -480,7 +480,7 @@ func (rm *ResourceManager) emitRecord(ctx context.Context, identity ResourceIden u.EventId = eventID } utilizations := []*meteringpb.Utilization{u} - rm.recordUtilization(ctx, identity, u, false) + rm.recordUtilization(ctx, identity, u, action != meteringpb.MeterAction_METER_ACTION_USAGE) if rm.emitter == nil { return diff --git a/pkg/resourcemanager/utilization_gauge_test.go b/pkg/resourcemanager/utilization_gauge_test.go index 86bbd72712..4f385a95df 100644 --- a/pkg/resourcemanager/utilization_gauge_test.go +++ b/pkg/resourcemanager/utilization_gauge_test.go @@ -7,6 +7,7 @@ import ( "github.com/jonboulle/clockwork" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" @@ -54,7 +55,32 @@ func TestUtilizationGauge_ResourceIDLabel(t *testing.T) { assert.Equal(t, "cre:workflow:compute", rt.AsString()) }) - t.Run("snapshots keep resource_id", func(t *testing.T) { + t.Run("deltas keep resource_id", func(t *testing.T) { + rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{MeterRecordsEnabled: true, Emitter: &fakeEmitter{}}) + rm.utilization = gauge + + for _, id := range []string{"trigger-1", "trigger-2"} { + rm.EmitDelta(t.Context(), testIdentity, "register:wf-1:"+id, 1, UtilizationFields{ + ResourceType: "registrations", + ResourceID: id, + OrgID: "org-1", + }) + } + + var ids []string + for _, p := range collectUtilizationPoints(t, reader) { + rt, _ := p.Attributes.Value("resource_type") + if rt.AsString() != "registrations" { + continue + } + id, hasID := p.Attributes.Value("resource_id") + require.True(t, hasID) + ids = append(ids, id.AsString()) + } + assert.ElementsMatch(t, []string{"trigger-1", "trigger-2"}, ids) + }) + + t.Run("snapshots keep resource_id and share the delta series", func(t *testing.T) { emitter := &fakeEmitter{} clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)) rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{ @@ -66,6 +92,11 @@ func TestUtilizationGauge_ResourceIDLabel(t *testing.T) { }) rm.utilization = gauge + rm.EmitDelta(t.Context(), testIdentity, "register:wf-1:trigger-1", 1, UtilizationFields{ + ResourceType: "operations", ResourceID: "trigger-1", OrgID: "org-1", + }) + deltaSet := attributeSetFor(t, collectUtilizationPoints(t, reader), "operations") + m := &fakeMeterable{identity: testIdentity} m.set([]SnapshotEntry{{ Identity: testIdentity, @@ -81,17 +112,24 @@ func TestUtilizationGauge_ResourceIDLabel(t *testing.T) { clock.Advance(30 * time.Second) waitForSnapshotCount(t, emitter, 1) - var found bool - for _, p := range collectUtilizationPoints(t, reader) { - rt, _ := p.Attributes.Value("resource_type") - if rt.AsString() != "operations" { - continue - } - found = true - id, hasID := p.Attributes.Value("resource_id") - assert.True(t, hasID) - assert.Equal(t, "trigger-1", id.AsString()) - } - assert.True(t, found) + points := collectUtilizationPoints(t, reader) + snapSet := attributeSetFor(t, points, "operations") + id, hasID := snapSet.Value("resource_id") + require.True(t, hasID) + assert.Equal(t, "trigger-1", id.AsString()) + assert.True(t, deltaSet.Equals(&snapSet), "delta and snapshot must land on the same series") }) } + +// attributeSetFor returns the attribute set of the single point with the given resource_type. +func attributeSetFor(t *testing.T, points []metricdata.DataPoint[int64], resourceType string) attribute.Set { + t.Helper() + var sets []attribute.Set + for _, p := range points { + if rt, _ := p.Attributes.Value("resource_type"); rt.AsString() == resourceType { + sets = append(sets, p.Attributes) + } + } + require.Len(t, sets, 1) + return sets[0] +} From 19c347cb653ed16bf7b00205534a118f5818249f Mon Sep 17 00:00:00 2001 From: Dylan Tinianov Date: Thu, 15 Oct 2026 10:26:27 -0400 Subject: [PATCH 3/3] resourcemanager: never label the utilization gauge with resource_id The id can be per-execution, so it stays on the durable MeterRecord and MeterSnapshot events only. No caller option. Co-Authored-By: Claude Fable 5.1 --- pkg/resourcemanager/resourcemanager.go | 20 +-- pkg/resourcemanager/utilization_gauge_test.go | 145 +++++++----------- 2 files changed, 61 insertions(+), 104 deletions(-) diff --git a/pkg/resourcemanager/resourcemanager.go b/pkg/resourcemanager/resourcemanager.go index 0edf2f8a2c..264b9d16b1 100644 --- a/pkg/resourcemanager/resourcemanager.go +++ b/pkg/resourcemanager/resourcemanager.go @@ -95,8 +95,8 @@ const ( snapshotEmitFailureCounterName = "resource_manager_snapshot_emit_failure_total" ) -// utilizationGaugeName is labeled with every ResourceIdentity dimension; usage -// records omit resource_id because their ids are per-execution. +// utilizationGaugeName is labeled with every ResourceIdentity dimension but never +// resource_id, which can be per-execution. const utilizationGaugeName = "resource_manager_utilization" // DefaultSnapshotInterval is the recommended snapshot period. It is NOT @@ -374,7 +374,7 @@ func (rm *ResourceManager) emitSnapshot(ctx context.Context, m Meterable, tickTi bucketUnix, ) } - rm.recordUtilization(ctx, e.Identity, u, true) + rm.recordUtilization(ctx, e.Identity, u) } snapshot := &meteringpb.MeterSnapshot{ @@ -480,7 +480,7 @@ func (rm *ResourceManager) emitRecord(ctx context.Context, identity ResourceIden u.EventId = eventID } utilizations := []*meteringpb.Utilization{u} - rm.recordUtilization(ctx, identity, u, action != meteringpb.MeterAction_METER_ACTION_USAGE) + rm.recordUtilization(ctx, identity, u) if rm.emitter == nil { return @@ -523,8 +523,8 @@ func (rm *ResourceManager) emitRecord(ctx context.Context, identity ResourceIden rm.countRecord(ctx, rm.emitSuccess, identity, action) } -// recordUtilization records value to the utilization gauge; withResourceID adds the resource_id label. -func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIdentity, u *meteringpb.Utilization, withResourceID bool) { +// recordUtilization records value to the utilization gauge. +func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIdentity, u *meteringpb.Utilization) { if rm.utilization == nil { return } @@ -543,7 +543,7 @@ func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIde ) return } - attrs := []attribute.KeyValue{ + rm.utilization.Record(ctx, value, metric.WithAttributes( attribute.String("product", id.Product), attribute.String("tenant", id.Tenant), attribute.String("numeric_tenant_id", id.NumericTenantID), @@ -556,11 +556,7 @@ func (rm *ResourceManager) recordUtilization(ctx context.Context, id ResourceIde attribute.String("resource_pool_id", id.ResourcePoolID), attribute.String("resource_type", u.GetResourceType()), attribute.String("org_id", u.GetOrgId()), - } - if withResourceID { - attrs = append(attrs, attribute.String("resource_id", u.GetResourceId())) - } - rm.utilization.Record(ctx, value, metric.WithAttributes(attrs...)) + )) } // countRecord records a MeterRecord emit outcome. Labels are intentionally diff --git a/pkg/resourcemanager/utilization_gauge_test.go b/pkg/resourcemanager/utilization_gauge_test.go index 4f385a95df..caa0440d76 100644 --- a/pkg/resourcemanager/utilization_gauge_test.go +++ b/pkg/resourcemanager/utilization_gauge_test.go @@ -7,9 +7,9 @@ import ( "github.com/jonboulle/clockwork" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "go.opentelemetry.io/otel/attribute" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" + "google.golang.org/protobuf/proto" "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" @@ -30,106 +30,67 @@ func collectUtilizationPoints(t *testing.T, reader *sdkmetric.ManualReader) []me return nil } -func TestUtilizationGauge_ResourceIDLabel(t *testing.T) { +func TestUtilizationGauge_NoResourceIDLabel(t *testing.T) { reader := sdkmetric.NewManualReader() gauge, err := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test").Int64Gauge(utilizationGaugeName) require.NoError(t, err) - t.Run("usage records omit resource_id", func(t *testing.T) { - rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{MeterRecordsEnabled: true, Emitter: &fakeEmitter{}}) - rm.utilization = gauge - - for _, exec := range []string{"exec-1", "exec-2"} { - rm.EmitUsage(t.Context(), testIdentity, exec, 7, UtilizationFields{ - ResourceType: "cre:workflow:compute", - ResourceID: "wf-1:" + exec, - OrgID: "org-1", - }) - } - - points := collectUtilizationPoints(t, reader) - require.Len(t, points, 1) - _, hasID := points[0].Attributes.Value("resource_id") - assert.False(t, hasID) - rt, _ := points[0].Attributes.Value("resource_type") - assert.Equal(t, "cre:workflow:compute", rt.AsString()) + emitter := &fakeEmitter{} + clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)) + rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{ + MeterRecordsEnabled: true, + MeterSnapshotsEnabled: true, + Emitter: emitter, + SnapshotInterval: 30 * time.Second, + Clock: clock, }) + rm.utilization = gauge - t.Run("deltas keep resource_id", func(t *testing.T) { - rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{MeterRecordsEnabled: true, Emitter: &fakeEmitter{}}) - rm.utilization = gauge - - for _, id := range []string{"trigger-1", "trigger-2"} { - rm.EmitDelta(t.Context(), testIdentity, "register:wf-1:"+id, 1, UtilizationFields{ - ResourceType: "registrations", - ResourceID: id, - OrgID: "org-1", - }) - } - - var ids []string - for _, p := range collectUtilizationPoints(t, reader) { - rt, _ := p.Attributes.Value("resource_type") - if rt.AsString() != "registrations" { - continue - } - id, hasID := p.Attributes.Value("resource_id") - require.True(t, hasID) - ids = append(ids, id.AsString()) - } - assert.ElementsMatch(t, []string{"trigger-1", "trigger-2"}, ids) - }) - - t.Run("snapshots keep resource_id and share the delta series", func(t *testing.T) { - emitter := &fakeEmitter{} - clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)) - rm := NewResourceManager(logger.Test(t), ResourceManagerConfig{ - MeterRecordsEnabled: true, - MeterSnapshotsEnabled: true, - Emitter: emitter, - SnapshotInterval: 30 * time.Second, - Clock: clock, + for _, exec := range []string{"exec-1", "exec-2"} { + rm.EmitUsage(t.Context(), testIdentity, exec, 7, UtilizationFields{ + ResourceType: "cre:workflow:compute", ResourceID: "wf-1:" + exec, OrgID: "org-1", }) - rm.utilization = gauge - - rm.EmitDelta(t.Context(), testIdentity, "register:wf-1:trigger-1", 1, UtilizationFields{ - ResourceType: "operations", ResourceID: "trigger-1", OrgID: "org-1", + } + for _, id := range []string{"trigger-1", "trigger-2"} { + rm.EmitDelta(t.Context(), testIdentity, "register:wf-1:"+id, 1, UtilizationFields{ + ResourceType: "registrations", ResourceID: id, OrgID: "org-1", }) - deltaSet := attributeSetFor(t, collectUtilizationPoints(t, reader), "operations") - - m := &fakeMeterable{identity: testIdentity} - m.set([]SnapshotEntry{{ - Identity: testIdentity, - Utilizations: []*meteringpb.Utilization{ - NewUtilizationInt(3, UtilizationFields{ResourceType: "operations", ResourceID: "trigger-1", OrgID: "org-1"}), - }, - }}) - unregister := rm.Register(m) - defer unregister() - - servicetest.Run(t, rm) - require.NoError(t, clock.BlockUntilContext(t.Context(), 1)) - clock.Advance(30 * time.Second) - waitForSnapshotCount(t, emitter, 1) - - points := collectUtilizationPoints(t, reader) - snapSet := attributeSetFor(t, points, "operations") - id, hasID := snapSet.Value("resource_id") - require.True(t, hasID) - assert.Equal(t, "trigger-1", id.AsString()) - assert.True(t, deltaSet.Equals(&snapSet), "delta and snapshot must land on the same series") - }) -} - -// attributeSetFor returns the attribute set of the single point with the given resource_type. -func attributeSetFor(t *testing.T, points []metricdata.DataPoint[int64], resourceType string) attribute.Set { - t.Helper() - var sets []attribute.Set + } + m := &fakeMeterable{identity: testIdentity} + m.set([]SnapshotEntry{{ + Identity: testIdentity, + Utilizations: []*meteringpb.Utilization{ + NewUtilizationInt(3, UtilizationFields{ResourceType: "operations", ResourceID: "wf-1", OrgID: "org-1"}), + }, + }}) + unregister := rm.Register(m) + defer unregister() + servicetest.Run(t, rm) + require.NoError(t, clock.BlockUntilContext(t.Context(), 1)) + clock.Advance(30 * time.Second) + waitForSnapshotCount(t, emitter, 1) + + points := collectUtilizationPoints(t, reader) + byType := map[string]int{} for _, p := range points { - if rt, _ := p.Attributes.Value("resource_type"); rt.AsString() == resourceType { - sets = append(sets, p.Attributes) + _, hasID := p.Attributes.Value("resource_id") + assert.False(t, hasID, "resource_id must never be a metric label") + rt, _ := p.Attributes.Value("resource_type") + byType[rt.AsString()]++ + } + // Distinct resource ids collapse onto one series per resource_type. + assert.Equal(t, map[string]int{"cre:workflow:compute": 1, "registrations": 1, "operations": 1}, byType) + + // The durable events still carry the resource id. + ids := map[string]bool{} + for _, call := range emitter.snapshot() { + var record meteringpb.MeterRecord + if proto.Unmarshal(call.body, &record) != nil { + continue + } + for _, u := range record.GetUtilizations() { + ids[u.GetResourceId()] = true } } - require.Len(t, sets, 1) - return sets[0] + assert.True(t, ids["wf-1:exec-1"] && ids["wf-1:exec-2"] && ids["trigger-1"] && ids["trigger-2"]) }