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

Refactor Pipeline #3233

Merged
merged 19 commits into from Sep 28, 2022
Merged

Refactor Pipeline #3233

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
eb4369a
Add views field to pipeline
MrAlias Sep 23, 2022
48905fc
Rename newPipelineRegistries->newPipelineRegistry
MrAlias Sep 23, 2022
46a4ef7
Add Reader as field to pipeline
MrAlias Sep 23, 2022
9c08765
Replace createAggregators with resolver facilitator
MrAlias Sep 23, 2022
ba1433d
Replace create agg funcs with inserter facilitator
MrAlias Sep 23, 2022
638cea8
Correct documentation
MrAlias Sep 23, 2022
bcc091d
Replace pipelineRegistry with []pipeline type
MrAlias Sep 23, 2022
65dd4fe
Rename newPipelineRegistry->newPipelines
MrAlias Sep 23, 2022
95b570f
Fix pipeline_registry_test
MrAlias Sep 23, 2022
ce13072
Flatten isMonotonic into only use
MrAlias Sep 23, 2022
0b24e9d
Update FIXME into TODO
MrAlias Sep 23, 2022
6429b15
Rename instrument provider resolver field to resolve
MrAlias Sep 23, 2022
78cd494
Fix comment English
MrAlias Sep 23, 2022
2dd217c
Merge branch 'main' into pipeline
MrAlias Sep 23, 2022
baae592
Merge branch 'main' into pipeline
MrAlias Sep 26, 2022
64ac589
Merge branch 'main' into pipeline
MrAlias Sep 26, 2022
218d5e2
Merge branch 'main' into pipeline
MrAlias Sep 27, 2022
26b7a33
Fix drop aggregator detection
MrAlias Sep 27, 2022
d45bee9
Merge branch 'pipeline' of github.com:MrAlias/opentelemetry-go into p…
MrAlias Sep 27, 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
40 changes: 20 additions & 20 deletions sdk/metric/instrument_provider.go
View file
Open in desktop
Expand Up @@ -27,8 +27,8 @@ import (
)

type asyncInt64Provider struct {
scope instrumentation.Scope
registry *pipelineRegistry
scope instrumentation.Scope
resolve *resolver[int64]
}

var _ asyncint64.InstrumentProvider = asyncInt64Provider{}
Expand All @@ -37,7 +37,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -56,7 +56,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -74,7 +74,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -89,8 +89,8 @@ func (p asyncInt64Provider) Gauge(name string, opts ...instrument.Option) (async
}

type asyncFloat64Provider struct {
scope instrumentation.Scope
registry *pipelineRegistry
scope instrumentation.Scope
resolve *resolver[float64]
}

var _ asyncfloat64.InstrumentProvider = asyncFloat64Provider{}
Expand All @@ -99,7 +99,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -117,7 +117,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -135,7 +135,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -150,8 +150,8 @@ func (p asyncFloat64Provider) Gauge(name string, opts ...instrument.Option) (asy
}

type syncInt64Provider struct {
scope instrumentation.Scope
registry *pipelineRegistry
scope instrumentation.Scope
resolve *resolver[int64]
}

var _ syncint64.InstrumentProvider = syncInt64Provider{}
Expand All @@ -160,7 +160,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -178,7 +178,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -196,7 +196,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -211,8 +211,8 @@ func (p syncInt64Provider) Histogram(name string, opts ...instrument.Option) (sy
}

type syncFloat64Provider struct {
scope instrumentation.Scope
registry *pipelineRegistry
scope instrumentation.Scope
resolve *resolver[float64]
}

var _ syncfloat64.InstrumentProvider = syncFloat64Provider{}
Expand All @@ -221,7 +221,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -239,7 +239,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand All @@ -257,7 +257,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.resolve.Aggregators(view.Instrument{
Scope: p.scope,
Name: name,
Description: cfg.Description(),
Expand Down
22 changes: 11 additions & 11 deletions sdk/metric/meter.go
View file
Open in desktop
Expand Up @@ -42,7 +42,7 @@ type meterRegistry struct {

meters map[instrumentation.Scope]*meter

registry *pipelineRegistry
pipes pipelines
}

// Get returns a registered meter matching the instrumentation scope if it
Expand All @@ -56,8 +56,8 @@ func (r *meterRegistry) Get(s instrumentation.Scope) *meter {

if r.meters == nil {
m := &meter{
Scope: s,
registry: r.registry,
Scope: s,
pipes: r.pipes,
}
r.meters = map[instrumentation.Scope]*meter{s: m}
return m
Expand All @@ -69,8 +69,8 @@ func (r *meterRegistry) Get(s instrumentation.Scope) *meter {
}

m = &meter{
Scope: s,
registry: r.registry,
Scope: s,
pipes: r.pipes,
}
r.meters[s] = m
return m
Expand All @@ -83,35 +83,35 @@ func (r *meterRegistry) Get(s instrumentation.Scope) *meter {
type meter struct {
instrumentation.Scope

registry *pipelineRegistry
pipes pipelines
}

// Compile-time check meter implements metric.Meter.
var _ metric.Meter = (*meter)(nil)

// AsyncInt64 returns the asynchronous integer instrument provider.
func (m *meter) AsyncInt64() asyncint64.InstrumentProvider {
return asyncInt64Provider{scope: m.Scope, registry: m.registry}
return asyncInt64Provider{scope: m.Scope, resolve: newResolver[int64](m.pipes)}
}

// AsyncFloat64 returns the asynchronous floating-point instrument provider.
func (m *meter) AsyncFloat64() asyncfloat64.InstrumentProvider {
return asyncFloat64Provider{scope: m.Scope, registry: m.registry}
return asyncFloat64Provider{scope: m.Scope, resolve: newResolver[float64](m.pipes)}
}

// RegisterCallback registers the function f to be called when any of the
// insts Collect method is called.
func (m *meter) RegisterCallback(insts []instrument.Asynchronous, f func(context.Context)) error {
m.registry.registerCallback(f)
m.pipes.registerCallback(f)
return nil
}

// SyncInt64 returns the synchronous integer instrument provider.
func (m *meter) SyncInt64() syncint64.InstrumentProvider {
return syncInt64Provider{scope: m.Scope, registry: m.registry}
return syncInt64Provider{scope: m.Scope, resolve: newResolver[int64](m.pipes)}
}

// SyncFloat64 returns the synchronous floating-point instrument provider.
func (m *meter) SyncFloat64() syncfloat64.InstrumentProvider {
return syncFloat64Provider{scope: m.Scope, registry: m.registry}
return syncFloat64Provider{scope: m.Scope, resolve: newResolver[float64](m.pipes)}
}