Add sdk/metric/reader package structure to new_sdk/main

[MrAlias]

Blocked by #2799

Add in the needed interfaces and high-level types for the new SDK design from new_sdk/example. This is not expected to be a fully working implementation.

sdk/metric/reader

• type Reader interface
  • type Registeree interface (might be a better idea to rename this idiomatic to Go Registerer)
  • type Producer interface
  • type Metrics struct
  • type Scope struct
  • type Instrument struct
  • type Point struct
• type Exporter interface
• type Option interface to be passed to the new Provider method option
  • type config struct stub

[MrAlias]

The Registeree interface is only embedded in the Reader interface. I do not see a reason this should not just become:

type Reader interface {
	Register(Producer)
	Flush(context.Context) error
	Shutdown(context.Context) error
}

[MrAlias]

The current package structure seems like an attempt to consolidate docs of the sdk/metric package instead of an intentional grouping of related concepts. There are interface exported here that are not used in the package, it doesn't match the trace span processor and exporter layout, and it seems to overlap with the sdkinstrument package's purpose.

I think this package structure could use a revised design attempt here before we progress with a PR.

[MrAlias]

After reading the metric SDK specification and looking at other SIG implementations I wonder if we can simplify the readers and make them more concrete in the implementation. Something more along these lines:

type Exporter interface {
	Export(context.Context, Metrics) error
	Flush(context.Context) error
	Shutdown(context.Context) error
}

type Reader struct{}

func NewReader(mp *MeterProvider, exp Exporter, options ...Option) *Reader

func (*Reader) Collect(context.Context) error

func (*Reader) Flush(context.Context) error

func (*Reader) Shutdown(context.Context) error

type PeriodicReader struct {
	Reader
}

func NewPeriodicReader(mp *MeterProvider, exp Exporter, options ...PeriodicOption) *PeriodicReader

This addresses a few issues:

• There is a uniqueness between Reader to MeterProvider that needs to be enforced. Constructing Readers this way ensures that only one MeterProvider is passed to a Reader. There is no extra guards need to handle a registration method.
• Simplifies implementation. There is no need for the additional Registeree or Producer interfaces.
• Each Reader can manage their own pool of Metrics and this optimization of passing an already allocated object to Collect is not exposed to the public API.

Issues with this approach:

• It is not user extensible; users cannot provide their own implementation of the Reader.
  • The Java SIG currently does not allow non-OTel Readers, so there is already precedence here.
  • The specifications language does not indicate this should be user defined.
  • If we want to change this in the future, the MeterProvider can have a Producer method added to it that user Readers can wrap with their Collect operations.
• Initialization of the metric pipeline switches to Reader(MeterProvider, Exporter) or even Exporter(Reader(MeterProvider)) instead of MeterProvider(Reader(Exporter)).
  • The periodic reader case could still be abstracted into a MeterProvicer option like func WithPeriodicReader(Exporter, ...PeriodicOption). This would be similar to the tracing WithBatcher option and, outside of the Prometheus exporter, would be the common use case.
• The prometheus exporter can no longer be an implementation of the Reader.
  • The Prometheus exporter can instead wrap a Reader that it calls (i.e. func New(*MeterProvider, ...Option) *Exporter can create and wrap a Reader)

@jmacd thoughts?

[MrAlias]

Abbreviated feedback from @MadVikingGod in SIG meeting on the above design is the Prometheus exporter needs to call Collect and in the same function scope receive the collected metrics. The new_sdk/example approach does this by implementing a reader and calling Produce.

[MrAlias]

Abbreviated feedback from @MadVikingGod in SIG meeting on the above design is the Prometheus exporter needs to call Collect and in the same function scope receive the collected metrics. The new_sdk/example approach does this by implementing a reader and calling Produce.

I think it is possible to handle this with the above design, but the complexity of passing responses from the out-of-band call to Export by the Reader seems like the opposite of what the design is intended to accomplish.

Why not just have a reader pass the Metrics as a return value and only have the PeriodicReader accept an Exporter?

E.g.

type Exporter interface {
	Export(context.Context, Metrics) error
	Flush(context.Context) error
	Shutdown(context.Context) error
}

type Reader struct{}

func NewReader(mp *MeterProvider, options ...Option) *Reader

func (*Reader) Collect(context.Context) (Metrics, error)

func (*Reader) Flush(context.Context) error

func (*Reader) Shutdown(context.Context) error

type PeriodicReader struct {
	Reader
}

func (*PeriodicReader) Collect(context.Context) error

func NewPeriodicReader(mp *MeterProvider, exp Exporter, options ...PeriodicOption) *PeriodicReader

[MadVikingGod]

How does this interact with the metric.WithReader() API? If we establish multiple kinds of readers, even if they imbed the concrete reader, then how do we add the other Readers to our sdk?

[MrAlias]

How does this interact with the metric.WithReader() API? If we establish multiple kinds of readers, even if they imbed the concrete reader, then how do we add the other Readers to our sdk?

They wouldn't interact as that would be the wrong approach with the alternate design. The Reader would accept the MeterProvider to it's creation function, there by ensuring only one MeterProvider is associated with it. The WithReader option would not exist.

If we wanted to add another Reader to the package we would add another concrete type. And it would follow with its own creation function that associated it with a MeterProvider.

[MadVikingGod]

What I don't think I understand how this will work is how prometheus can connect a specific call of its Collect(chan<-) with the export() that is a result of the Reader's Collect().

In prometheus the HTTP function calls Collect(chan<-); this would cause our prometheus export to call Collect() on the reader, which will then call Export(metrics), but it won't have any connection with the channel that was passed in the first call. How do these get associated? If prometheus calls Collect twice, how do we make sure the in-flight Export goes to the first channel while the second one is waiting (currently, there is a lock so only one collect in flight at a time).

A similar but different issue is how does the Reader collect the information? The MeterProvider doesn't have an exposed API for the reader of any kind to get this information. I think we may want to put the Collect(context.Context) (Metrics, error) on the MeterProvider, and have the Reader take an exporter as a construction argument. This would let prometheus bypass the reader, essentially play the role of the reader, and still allow for things like a Periodic Reader.

Edit: the reason the current API doesn't have a MeterProvider.Produce() (read Collect()) is that the Producer that is registered is not the MeterProvider but the ViewCompiler. This allows us to compile a reduced set of async instruments that must be processed when Produce() is eventually called.

[MrAlias]

What I don't think I understand how this will work is how prometheus can connect a specific call of its Collect(chan<-) with the export() that is a result of the Reader's Collect().

In prometheus the HTTP function calls Collect(chan<-); this would cause our prometheus export to call Collect() on the reader, which will then call Export(metrics), but it won't have any connection with the channel that was passed in the first call. How do these get associated? If prometheus calls Collect twice, how do we make sure the in-flight Export goes to the first channel while the second one is waiting (currently, there is a lock so only one collect in flight at a time).

I don't follow. Why can't the Prometheus exporter replace this line:

opentelemetry-go/exporters/prometheus/prometheus.go

Line 166 in 8b6f0c6

data := c.exp.producer.Produce(context.Background(), nil)

With

data, err := c.exp.reader.Collect(context.Background())

A similar but different issue is how does the Reader collect the information? The MeterProvider doesn't have an exposed API for the reader of any kind to get this information. I think we may want to put the Collect(context.Context) (Metrics, error) on the MeterProvider, and have the Reader take an exporter as a construction argument. This would let prometheus bypass the reader, essentially play the role of the reader, and still allow for things like a Periodic Reader.

Edit: the reason the current API doesn't have a MeterProvider.Produce() (read Collect()) is that the Producer that is registered is not the MeterProvider but the ViewCompiler. This allows us to compile a reduced set of async instruments that must be processed when Produce() is eventually called.

I think we have similar implementation ideas here. I see the MeterProvider having a method like this:

func (*MeterProvider) collectFunc(ConfigForAViewCompiler) func(contect.Context) (Metric, error)

The returned function would be a closure of this essentially.

When a reader is created, the MeterProvider it is passed would have that method called.

[MrAlias]

A similar but different issue is how does the Reader collect the information? The MeterProvider doesn't have an exposed API for the reader of any kind to get this information. I think we may want to put the Collect(context.Context) (Metrics, error) on the MeterProvider, and have the Reader take an exporter as a construction argument. This would let prometheus bypass the reader, essentially play the role of the reader, and still allow for things like a Periodic Reader.
Edit: the reason the current API doesn't have a MeterProvider.Produce() (read Collect()) is that the Producer that is registered is not the MeterProvider but the ViewCompiler. This allows us to compile a reduced set of async instruments that must be processed when Produce() is eventually called.

I think we have similar implementation ideas here. I see the MeterProvider having a method like this:

func (*MeterProvider) collectFunc(ConfigForAViewCompiler) func(contect.Context) (Metric, error)

The returned function would be a closure of this essentially.

When a reader is created, the MeterProvider it is passed would have that method called.

Given this is unexported, we would also be able to play with memory allocation optimizations. We could support this as well

func (*MeterProvider) collectOverwriteFunc(ConfigForAViewCompiler) func(contect.Context, *Metric) error

That way the function does not need overhead of checking how it should return data.

Having these functions on the MeterProvider is also the precursor to allowing third party readers if we ever needed them. We can test out these methods internally and if there is ever a need for external readers we could start exporting the method.

[MadVikingGod]

Maybe we don't want to create it with a MeterProvider, but with a ViewCompiler. The MeterProvider could instead of having a WithView option we could have a method that takes a View (config) and returns a ViewCompiler that the Reader could use.

So it would looks like:

mp := sdkmetric.NewMeterProvider(...)
view := mp.WithView(...)

rdr := reader.New(view...)

[MrAlias]

Maybe we don't want to create it with a MeterProvider, but with a ViewCompiler. The MeterProvider could instead of having a WithView option we could have a method that takes a View (config) and returns a ViewCompiler that the Reader could use.

So it would looks like:

mp := sdkmetric.NewMeterProvider(...)
view := mp.WithView(...)

rdr := reader.New(view...)

Ah, yeah! That looks like a promising approach. It would definitely simplify the viewstate logic; it would no longer need to keep its own mapping of reader to collectors at that point.

[MadVikingGod]

In the API i last proposed the views will need a more robust API. They couldn't be just a configuration block, because when you add a new instrument the sdk will have to find all views that it belongs in (and set up what transforms might need to happen) and an API for the reader to get data from the aggregators associated with a view.

So we it might looks something like:

// In a instrument creation
for _, view := range mp.views {
    if v := view.RegisterInstrument(name, description, kind, labels); v != nil {
        i.views = append(i.views, v)
    }
}

And we will have to be able to update instruments as views are added

[MrAlias]

Overall question that are being raised from design sync:

1. (spec level question) Are readers physical or logical?
2. (spec level question) Are exporters physical or logical?
3. Is this registration complexity justified?
   1. The information flow is from instrument -> meter -> MeterProvider -> view -> reader
      1. Is there bidirctionaly? Is there a cycle?
      2. Can our design be simplified by matching closer to this information flow?
4. Optimization for synchronous instrument (buffer per reader output), is it needed? Is there justification here?

cc @MadVikingGod @jmacd @Aneurysm9

[MadVikingGod]

To answer the first two questions we can look at other sigs. In particular I surveyed Java, python, and C#.
For readers it's both (kind of).
Java has an interface and you use those in the construction of the SDK MeterProvider. You can also
Python has an abstract base class, and you provide them when creating the MeterProvider.
C# uses a partial class, which I'm out of my league on, but it looks to be more concrete then python or java.

[jmacd]

I want to acknowledge @MrAlias and work to provide better answers than I did in today's meeting.
Here are some partial answers and discussion points.

For the question about bi-directionality, I believe the direction is always the same but we have different points where a trigger originates, so different paths: (1) the user responsible for registering the SDK can Flush and Shutdown (which impacts all readers); (2) the user who instruments code can Add and Record synchronous instruments (which impact all readers); (3) the user who instruments code can Observe async instruments (which impact one reader); (4) the reader can Collect observations (impacts one reader). So there are per-reader and all-reader flows, that's part of the confusion.

@MrAlias I think your question 3b deserves more study. I don't know that "simplified" is possible, but an arrangement of data types that is closer to your intuition is probably better, so possibly yes.

On question 4. I muddled through something in our meeting today. I have removed any optimizations on this topic in my branch already, meaning that each configured View behavior maps to exactly one Aggregator and if there are more than one reader or behavior then a generic multi-reader/multi-instrument type is used (which is aware of per-reader and all-reader flows).

I will follow on with more on this topic, investigating part 3b.

[MrAlias]

Following up to this thread. I still think there are merits registering a MeterProvider with a Reader on creation. And I think it would still conform to the specification.

However, my main reservation at this point is that it would be too dissimilar to other implementations and the tracing signal API.

With that in mind, I would like to focus on the registration API a bit.

1. Can we reduce the exported types and methods of the registration implementation?
   • Leaking implementation details with the Producer type adds bloat to the API. This bloat is not expected to be used by the SDK user, but only by developers extending the SDK. Can we still achieve this extensibility without the bloat?
2. Can we restrict the underlying reader registration to only be implemented by the OTel SDK?
   • Allowing user implementation of the Reader type to the MetricProvider will mean the Reader interface need to be stabilized and cannot change with a 1.0 release.
   • Allowing users to implement their own mechanism for registration will lock us into a registration/producer API, something we may want to optimize in the future.

I believe the answer to these questions to be yes.

The Reader interface registration method can be unexported:

// produceFunc is a closure to produce metrics for a specific reader.
type produceFunc func(context.Context) (Metrics, error)

type Reader interface {
	register(produceFunc)
	Flush(context.Context) error
	Shutdown(context.Context) error
}

Similarly, the MeterProvider registration mechanism can be unexported.

type MeterProvider struct{/*...*/}

// ...

func (*MeterProvider) produceFor(readerConfig) produceFunc

Which would result in a similarly redacted ManualReader.

type ManualReader struct {
	produce produceFunc
}

func (m *ManualReader) Collect(ctx context.Context) (Metrics, error)  { /*...*/ return m.produce(ctx) }
func (m *ManualReader) Flush(context.Context) error { /*...*/ }
func (m *ManualReader) Shutdown(context.Context) error  { /*...*/ }
func (m *ManualReader) register(f produceFunc) { m.produce = f }

Both the periodic reader and the Prometheus exporter can then wrap this with their own export logic. For example, the Prometheus exporter becomes:

type PrometheusExporter struct {
	// Embedding a ManualReader reader means the PrometheusExporter can be
	// registered as a Reader with a MeterProvider.
	ManualReader
	/*...*/
}

/* ... */

func (exp *PrometheusExporter) Collect(ch chan<- prometheus.Metric) {
	data, err := exp.ManualReader.Collect(context.Background())
	// Handle err and export data ...
}

[MrAlias]

Something I failed to mention above is that the Reader interface, and the ManualReader and PeriodicReader types would be moved into the sdk/metric package with the MeterProvider. This would be appropriate given we are trying to match the trace signal SDK and the types are related.

[jmacd]

Following the discussion in today's SIG meeting, here is the reader.Reader interface that abstracts all kinds of exporters including push-based and pull-based:

	// Reader is the interface used between the SDK and an
	// exporter.  Control flow is bi-directional through the
	// Reader, since the SDK initiates ForceFlush and Shutdown
	// while the initiates collection.  The Register() method here
	// informs the Reader that it can begin reading, signaling the
	// start of bi-directional control flow.
	//
	// Typically, push-based exporters that are periodic will
	// implement PeroidicExporter themselves and construct a
	// PeriodicReader to satisfy this interface.
	//
	// Pull-based exporters will typically implement Register
	// themselves, since they read on demand.
	Reader interface {
		// String describes this reader.
		String() string

		// Register is called when the SDK is fully
		// configured.  The Producer passed allows the
		// Reader to begin collecting metrics using its
		// Produce() method.
		Register(Producer)

		// ForceFlush is called when MeterProvider.ForceFlush() is called.
		ForceFlush(context.Context) error

		// Shutdown is called when MeterProvider.Shutdown() is called.
		Shutdown(context.Context) error
	}

with the same Producer API as before, having a single-method Produce() to perform collection on demand. You would pass reader.Reader objects along with ...reader.Option via metric.WithReader() when constructing a meter provider.

Under this scenario, the Prometheus "exporter" is a type that implements reader.Reader itself, mainly because there's little advantage to factoring out the "manual reader" interface for Prometheus to use. It's approximately 3 lines of code to implement Register() for any manual reader, since all you need to do is save the Producer.

Just to make that clear, here's the ManualReader implementation that I see as being useful for testing:

// ManualReader is a a simple Reader that allows an application to
// read metrics on demand.  It simply stores the Producer interface
// provided through registration.  Flush and Shutdown are no-ops.
type ManualReader struct {
	Name string
	Producer
}

var _ Reader = &ManualReader{}

// NewManualReader returns an Reader that stores the Producer for
// manual use and returns a configurable `name` as its String(),
func NewManualReader(name string) *ManualReader {
	return &ManualReader{
		Name: name,
	}
}

// String returns the name of this ManualReader.
func (mr *ManualReader) String() string {
	return mr.Name
}

// Register stores the Producer which enables the caller to read
// metrics on demand.
func (mr *ManualReader) Register(p Producer) {
	mr.Producer = p
}

// ForceFlush is a no-op, always returns nil.
func (mr *ManualReader) ForceFlush(context.Context) error {
	return nil
}

// Shutdown is a no-op, always returns nil.
func (mr *ManualReader) Shutdown(context.Context) error {
	return nil
}

The question we left the SIG meeting with was about periodic readers. Here's what I would implement, roughly speaking:

// PushExporter is an interface for push-based exporters.
type PushExporter interface {
	String() string
	ExportMetrics(context.Context, Metrics) error
	ShutdownMetrics(context.Context, Metrics) error
	ForceFlushMetrics(context.Context, Metrics) error
}

// PeriodicReader is an implementation of Reader that manages periodic
// exporter, flush, and shutdown.  This implementation re-uses data
// from one collection to the next, to lower memory costs.
type PeriodicReader struct {
	lock     sync.Mutex
	data     Metrics
	interval time.Duration
	exporter PushExporter
	producer Producer
	stop     context.CancelFunc
	wait     sync.WaitGroup
}

// NewPeriodicReader constructs a PeriodicReader from a push-based
// exporter given an interval (TODO: and options).
func NewPeriodicReader(exporter PushExporter, interval time.Duration /* opts ...Option*/) Reader {
	return &PeriodicReader{
		interval: interval,
		exporter: exporter,
	}
}

// String returns the exporter name and the configured interval.
func (pr *PeriodicReader) String() string {
	return fmt.Sprintf("%v interval %v", pr.exporter.String(), pr.interval)
}

// Register starts the periodic export loop.
func (pr *PeriodicReader) Register(producer Producer) {
	ctx, cancel := context.WithCancel(context.Background())

	pr.producer = producer
	pr.stop = cancel
	pr.wait.Add(1)

	go pr.start(ctx)
}

// start runs the export loop.
func (pr *PeriodicReader) start(ctx context.Context) {
	defer pr.wait.Done()
	ticker := time.NewTicker(pr.interval)
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			pr.collect(ctx, pr.exporter.ExportMetrics)
		}
	}
}

// Shutdown stops the stops the export loop, canceling its Context,
// and waits for it to return.  Then it issues a ShutdownMetrics with
// final data.
func (pr *PeriodicReader) Shutdown(ctx context.Context) error {
	pr.stop()
	pr.wait.Wait()

	return pr.collect(ctx, pr.exporter.ShutdownMetrics)
}

// ForceFlush immediately waits for an existing collection, otherwise
// immediately begins collection without regards to timing and calls
// ForceFlush with current data.
func (pr *PeriodicReader) ForceFlush(ctx context.Context) error {
	return pr.collect(ctx, pr.exporter.ForceFlushMetrics)
}

// collect serializes access to re-usable metrics data, in each case
// calling through to an underlying PushExporter method with current
// data.
func (pr *PeriodicReader) collect(ctx context.Context, method func(context.Context, Metrics) error) error {
	pr.lock.Lock()
	defer pr.lock.Unlock()

	// The lock ensures that re-use of `pr.data` is successful, it
	// means that shutdown, flush, and ordinary collection are
	// exclusive.  Note that shutdown will cancel an concurrent
	// (ordinary) export, while flush will wait on for a
	// concurrent export.
	pr.data = pr.producer.Produce(&pr.data)

	return method(ctx, pr.data)
}

[jmacd]

By the way, this is totally untested! Suspect incorrect use of time.Ticker 😁 .

Not shown above, but from the same branch in the sdk/metric package, note that the method used to configure the meter provider is:

func WithReader(r reader.Reader, opts ...reader.Option) Option { ... }

The SDK specification talks about SDK auto-configured exporters. For an auto-configured exporter pattern, I would like to see each potentially auto-configurable exporter package support returning the appropriate sdk/metric.Option (potentially based on the environment, e.g., temporality preference for OTLP exporters).

For a Prometheus exporter, this means returning one of the structs that implements Reader itself, e.g.,

return sdkmetric.WithReader(
        &PrometheusExporter{ /* defaults from environment */ },
        /* no reader defaults needed for Prometheus */,
)

For any push-based exporter, this will mean returning a PeriodicReader option, e.g.,

return sdkmetric.WithReader(
        reader.NewPeriodicReader(
                &MyPusher{ /* defaults from environment, e.g., formatting, etc. */ },
                /* periodic defaults from environment, e.g., interval */),
        /* reader defaults from environment, e.g., temporality */,
)

[MadVikingGod]

I've spent this morning tiring to implement the Manual Reader, and I think I've found an interesting problem. Conceptually should a reader always get the data streams from ALL instruments or just the ones from views associated with it?

If it's the former, then the reader will need some facility to filter/transform/reaggergate data streams it receives from produce(). I think we can add this facility into our concepts of views so every reader doesn't need to reimplement this.
If it's the latter, then when you register the reader you will need to present the configuration of the view to the MeterProvider (or it's delegate like in the example).

If we don't associate readers with views in some way I don't see how we could have multiple readers measuring different things, as implied by (3) in the data model examples https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/datamodel.md#example-use-cases

[jmacd]

@MadVikingGod FWIW the current SDK Specification says that Views are specified for the whole provider, not for the individual reader. I filed open-telemetry/opentelemetry-specification#2288 about this. The branch I'm working on in #2865 for the record has taken this route, since our 4/25/2022 "Metrics SDK design" special meeting, where it seemed everyone agreed to this position. That means, the WithViews() option became a per-reader option; the implementation was already per-reader-per-view anyway, so the change to per-reader-view was minor.

Prior to last week, I had been following the specification on this topic, only allowing one set of Views to be configured at the MeterProvider level. I'm not sure if this helps us answer the question(s) in this issue, though, about Readers.

[jmacd]

For the record, #2885 proposes a very similar structure to what's in my latest examples above. That PR is close enough, IMO, that we should continue discussion there.

[MrAlias]

Closed by #2885