Add telemetry pipeline structure to new_sdk/main

MrAlias commented 2 years ago

~Blocked by #2799~

MadVikingGod commented 2 years ago

Here are my ideas when conceiving of the internal view. Views seem to have two responsibilities with datastreams in the SDK:

Filter and transform data points from instruments and direct the points into the correct aggregation/storage.
On-demand produce all datastreams from all current aggregations/storages.

An easy optimization that can be made is that when you create an instrument the SDK should be able to query all views if the instrument matches that view or not. This will allow us to not check if the instrument matches when updating datapoints. To be able to match the instrument descriptor would need at least Name, Description, Unit, InsturmentAttributes, and any other match criteria.

When an instrument is used, eg. Add() or Observe(), it should Update() ALL Views that it matches. This method should block the caller for as little as possible. The responsibilities of Update() is to create the aggregation that results from applying the view if needed, and to update the aggregation with the provided value. The datastream Descriptor should describe the meter and the instrument that the stream belongs to, and the resolved attributes of that stream (both the instrument attributes and the attributes provided by the Observe()).

A View should be able to read from all of it's aggregations and produce the composite datastreams. This should have a similar structure to the Produce method that readers get registered with. Because this should minimize blocking of updates a snapshot should be taken of the current state of aggregations and the data read from that.

package view // import "go.opentelemetry.io/otel/sdk/metric/internal/view"

type View interface {

    // Match checks if the instrument matches this View's filter. Only
    // instruments that match should Update the view.
    Match(instrument.Descriptor) bool

    // Update is used by instruments to record a new datapoint.
    Update(datastream.Descriptor, number) 

    // Produce snapshots the current aggregations and serialize the data out
    Produce(context.Context?) (export.Metrics, error?)
    // The Context and error are only needed if this method will respect timeouts.  If this is deemed simple enough then both can be removed.

}

type SingleView struct {
    // config to do matching
    ...
    // config for transforms
    ...
    // All aggregations for datastreams
    data map[meter]map[instrument]Aggregation
}
var _ View = &SingleView{}

func NewView(cfg view.Config) *SingleView

To be able to support multiple views we should have an object that can both distribute and aggregate data for different SingleViews

type MultiView struct {
    views []SingleView
}

func NewMultiView(configs ...view.Config) *MultiView

MadVikingGod commented 2 years ago

From the SIG discussion modify the interface as such

type View interface {

    // Match checkes if the instrument matches this View's filter. Only
    // instruments that match should Update the view.
    Match(instrument.Descriptor) Updater

   // Produce snapshots the current aggregations and serialize the data out
    Produce(context.Context?) (export.Metrics, error?)
    // The Context and error are only needed if this method will respect timeouts.  If this is deemed simple enough then both can be removed.
}

type Updater interface {
    // Update is used by instruments to record a new datapoint.
    Update([]attribute.KeyValue, number) // could be attribute.Set
}