Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Adds a cache to the metric insrument pipeline. #3181

Closed
wants to merge 7 commits into from
Closed

Adds a cache to the metric insrument pipeline. #3181

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
2a575d7
add instrument caching
MadVikingGod Aug 30, 2022
a295f2b
Fixes for sync instruments
MadVikingGod Sep 6, 2022
4b159b3
fix lint
MadVikingGod Sep 8, 2022
2baf437
Add changelog entry
MadVikingGod Sep 16, 2022
3134e76
made cache miss a const
MadVikingGod Sep 19, 2022
dd156c2
Merge remote-tracking branch 'upstream/main' into mvg/instrument_cache
MadVikingGod Sep 19, 2022
a316885
Merge remote-tracking branch 'upstream/main' into mvg/instrument_cache
MadVikingGod Sep 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
24 changes: 12 additions & 12 deletions sdk/metric/instrument_provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ var _ asyncint64.InstrumentProvider = asyncInt64Provider{}
func (p asyncInt64Provider) Counter(name string, opts ...instrument.Option) (asyncint64.Counter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[int64](p.registry, view.Instrument{
aggs, err := p.registry.createInt64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -59,7 +59,7 @@ func (p asyncInt64Provider) Counter(name string, opts ...instrument.Option) (asy
func (p asyncInt64Provider) UpDownCounter(name string, opts ...instrument.Option) (asyncint64.UpDownCounter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[int64](p.registry, view.Instrument{
aggs, err := p.registry.createInt64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -77,7 +77,7 @@ func (p asyncInt64Provider) UpDownCounter(name string, opts ...instrument.Option
func (p asyncInt64Provider) Gauge(name string, opts ...instrument.Option) (asyncint64.Gauge, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[int64](p.registry, view.Instrument{
aggs, err := p.registry.createInt64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -102,7 +102,7 @@ var _ asyncfloat64.InstrumentProvider = asyncFloat64Provider{}
func (p asyncFloat64Provider) Counter(name string, opts ...instrument.Option) (asyncfloat64.Counter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[float64](p.registry, view.Instrument{
aggs, err := p.registry.createFloat64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -120,7 +120,7 @@ func (p asyncFloat64Provider) Counter(name string, opts ...instrument.Option) (a
func (p asyncFloat64Provider) UpDownCounter(name string, opts ...instrument.Option) (asyncfloat64.UpDownCounter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[float64](p.registry, view.Instrument{
aggs, err := p.registry.createFloat64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -138,7 +138,7 @@ func (p asyncFloat64Provider) UpDownCounter(name string, opts ...instrument.Opti
func (p asyncFloat64Provider) Gauge(name string, opts ...instrument.Option) (asyncfloat64.Gauge, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[float64](p.registry, view.Instrument{
aggs, err := p.registry.createFloat64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -163,7 +163,7 @@ var _ syncint64.InstrumentProvider = syncInt64Provider{}
func (p syncInt64Provider) Counter(name string, opts ...instrument.Option) (syncint64.Counter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[int64](p.registry, view.Instrument{
aggs, err := p.registry.createInt64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -181,7 +181,7 @@ func (p syncInt64Provider) Counter(name string, opts ...instrument.Option) (sync
func (p syncInt64Provider) UpDownCounter(name string, opts ...instrument.Option) (syncint64.UpDownCounter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[int64](p.registry, view.Instrument{
aggs, err := p.registry.createInt64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -199,7 +199,7 @@ func (p syncInt64Provider) UpDownCounter(name string, opts ...instrument.Option)
func (p syncInt64Provider) Histogram(name string, opts ...instrument.Option) (syncint64.Histogram, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[int64](p.registry, view.Instrument{
aggs, err := p.registry.createInt64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -224,7 +224,7 @@ var _ syncfloat64.InstrumentProvider = syncFloat64Provider{}
func (p syncFloat64Provider) Counter(name string, opts ...instrument.Option) (syncfloat64.Counter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[float64](p.registry, view.Instrument{
aggs, err := p.registry.createFloat64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -242,7 +242,7 @@ func (p syncFloat64Provider) Counter(name string, opts ...instrument.Option) (sy
func (p syncFloat64Provider) UpDownCounter(name string, opts ...instrument.Option) (syncfloat64.UpDownCounter, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[float64](p.registry, view.Instrument{
aggs, err := p.registry.createFloat64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -260,7 +260,7 @@ func (p syncFloat64Provider) UpDownCounter(name string, opts ...instrument.Optio
func (p syncFloat64Provider) Histogram(name string, opts ...instrument.Option) (syncfloat64.Histogram, error) {
cfg := instrument.NewConfig(opts...)

aggs, err := createAggregators[float64](p.registry, view.Instrument{
aggs, err := p.registry.createFloat64Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand Down
112 changes: 112 additions & 0 deletions sdk/metric/meter_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -515,3 +515,115 @@ func TestMetersProvideScope(t *testing.T) {
assert.NoError(t, err)
metricdatatest.AssertEqual(t, want, got, metricdatatest.IgnoreTimestamp())
}

func TestDuplicateInstruments(t *testing.T) {
testCases := []struct {
name string
generateInstrument func(*testing.T, metric.Meter) any
}{
{
name: "AsyncInt64Counter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.AsyncInt64().Counter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "AsyncInt64UpDownCounter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.AsyncInt64().UpDownCounter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "AsyncInt64Gauge",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.AsyncInt64().Gauge("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "AsyncFloat64Counter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.AsyncFloat64().Counter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "AsyncFloat64UpDownCounter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.AsyncFloat64().UpDownCounter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "AsyncFloat64Gauge",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.AsyncFloat64().Gauge("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "SyncInt64Counter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.SyncInt64().Counter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "SyncInt64UpDownCounter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.SyncInt64().UpDownCounter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "SyncInt64Histogram",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.SyncInt64().Histogram("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "SyncFloat64Counter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.SyncFloat64().Counter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "SyncFloat64UpDownCounter",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.SyncFloat64().UpDownCounter("duplicate")
assert.NoError(t, err)
return inst
},
},
{
name: "SyncFloat64Histogram",
generateInstrument: func(t *testing.T, m metric.Meter) any {
inst, err := m.SyncFloat64().Histogram("duplicate")
assert.NoError(t, err)
return inst
},
},
}
for _, tt := range testCases {
t.Run(tt.name, func(t *testing.T) {
m := NewMeterProvider(WithReader(NewManualReader())).Meter("DuplicateInstruments")
inst := tt.generateInstrument(t, m)

assert.Equal(t, inst, tt.generateInstrument(t, m))
})
}
}
75 changes: 75 additions & 0 deletions sdk/metric/pipeline.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,16 @@ func (p *pipeline) produce(ctx context.Context) (metricdata.ResourceMetrics, err
type pipelineRegistry struct {
views map[Reader][]view.View
pipelines map[Reader]*pipeline

cacheLock sync.Mutex
cacheInt map[instrumentID]pipelineCacheResult[int64]
cacheFloat map[instrumentID]pipelineCacheResult[float64]
}

type pipelineCacheResult[N int64 | float64] struct {
aggregators []internal.Aggregator[N]
err error
instrumentKind view.InstrumentKind
}

func newPipelineRegistries(views map[Reader][]view.View) *pipelineRegistry {
Expand All @@ -175,7 +185,72 @@ func newPipelineRegistries(views map[Reader][]view.View) *pipelineRegistry {
return &pipelineRegistry{
views: views,
pipelines: pipelines,

cacheInt: map[instrumentID]pipelineCacheResult[int64]{},
cacheFloat: map[instrumentID]pipelineCacheResult[float64]{},
}
}

func (reg *pipelineRegistry) createInt64Aggregators(inst view.Instrument, instUnit unit.Unit) ([]internal.Aggregator[int64], error) {
reg.cacheLock.Lock()
defer reg.cacheLock.Unlock()
key := instrumentID{
scope: inst.Scope,
name: inst.Name,
description: inst.Description,
}

if result, ok := reg.cacheInt[key]; ok {
if inst.Kind != result.instrumentKind {
return nil, fmt.Errorf("instruments collide")
}
return result.aggregators, result.err
}

aggs, err := createAggregators[int64](reg, inst, instUnit)
reg.cacheInt[key] = pipelineCacheResult[int64]{
aggregators: aggs,
err: err,
instrumentKind: inst.Kind,
}
missedCacheInstrumentKind := view.InstrumentKind(254)
MadVikingGod marked this conversation as resolved.
Show resolved Hide resolved
reg.cacheFloat[key] = pipelineCacheResult[float64]{
aggregators: nil,
err: nil,
instrumentKind: missedCacheInstrumentKind,
}
return aggs, err
}

func (reg *pipelineRegistry) createFloat64Aggregators(inst view.Instrument, instUnit unit.Unit) ([]internal.Aggregator[float64], error) {
reg.cacheLock.Lock()
defer reg.cacheLock.Unlock()
key := instrumentID{
scope: inst.Scope,
name: inst.Name,
description: inst.Description,
}

if result, ok := reg.cacheFloat[key]; ok {
if inst.Kind != result.instrumentKind {
return nil, fmt.Errorf("instruments collide")
}
return result.aggregators, result.err
}

aggs, err := createAggregators[float64](reg, inst, instUnit)
reg.cacheFloat[key] = pipelineCacheResult[float64]{
aggregators: aggs,
err: err,
instrumentKind: inst.Kind,
}
missedCacheInstrumentKind := view.InstrumentKind(254)
reg.cacheInt[key] = pipelineCacheResult[int64]{
aggregators: nil,
err: nil,
instrumentKind: missedCacheInstrumentKind,
}
return aggs, err
}

// TODO (#3053) Only register callbacks if any instrument matches in a view.
Expand Down