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 all 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
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Expand Up @@ -8,6 +8,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm

## [Unreleased]

### Changed

- Add caching to instrument creation (#3181)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
- Add caching to instrument creation (#3181)
- Add caching to instrument creation.
Repeated requests for the same instrument will return the same instance. (#3181)


## [0.32.0] Revised Metric SDK (Alpha) - 2022-09-18

### Changed
Expand Down
24 changes: 12 additions & 12 deletions sdk/metric/instrument_provider.go
View file
Open in desktop
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
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
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]{},
}
}

const missedCacheInstrumentKind = view.InstrumentKind(254)

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,
}
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,
}
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