Support multi-instrument observable callbacks and ability to unregister callbacks

See spec issues https://github.com/open-telemetry/opentelemetry-specification/issues/2280, https://github.com/open-telemetry/opentelemetry-specification/issues/2232 and https://github.com/open-telemetry/opentelemetry-specification/pull/2317 PR which adds this to the spec.

Our API for creating observable instruments currently looks like this:

def cpu_time_callback() -> Iterable[Measurement]:
    with open("/proc/stat") as procstat:
        procstat.readline()  # skip the first line
        for line in procstat:
            if not line.startswith("cpu"): break
            cpu, *states = line.split()
            yield Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"})
            yield Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"})

meter.create_observable_counter("system.cpu.time", cpu_time_callback)

To support multi instrument callbacks we have a few possibilities:

Approach 1

https://github.com/open-telemetry/opentelemetry-specification/pull/2317#discussion_r801936529 suggested this as a possible way to implement multi instrument callbacks given our current API:

cpuTime = meter.create_observable_counter("system.cpu.time")
procsRunning = meter.create_observable_updowncounter("system.procs_running")
procsBlocked = meter.create_observable_updowncounter("system.procs_blocked")

def proc_stat_observer() -> Iterable[Measurement]:
    with open("/proc/stat") as procstat:
        procstat.readline()  # skip the first line
        for line in procstat:
            if line.startswith("cpu"):
                cpu, *states = line.split()
                yield cpuTime.observe(int(states[0]) // 100, {"cpu": cpu, "state": "user"})
                yield cpuTime.observe(int(states[1]) // 100, {"cpu": cpu, "state": "nice"})
            if line.startswith("procs_running"):
                var, value = line.split()
                yield procsRunning.observe(value)
            if line.startswith("procs_blocked"):
                var, value = line.split()
                yield procsBlocked.observe(value)

callback = meter.register_callback(proc_stat_observer, [cpuTime, procsRunning, procsBlocked])

def stop():
    callback.unregister()

The actual API changes to implement this are

• remove callback as a parameter during observable instrument creation
• add Meter.register_callback(callback: Callable[[], Iterable[Measurement]]) -> CallbackHandle
• add something like a CallbackHandle which just has an unregister() method.
• add Asynchronous.observe(value: Union[int, float], attributes: Attributes = None) -> Measurement to the Asynchronous instrument base class, which emits a Measurement associated with that instrument.

Pros:

• Somewhat similar to what we already have
• Single vs multi instrument callbacks are handled the same way

Cons:

• yield instrument.observe(value, attributes) seems a little weird to me
• Makes the single instrument callback case more difficult

Approach 2

We could restructure the observable callback to accept an parameter to use for making observations on:

cpuTime = meter.create_observable_counter("system.cpu.time")
procsRunning = meter.create_observable_updowncounter("system.procs_running")
procsBlocked = meter.create_observable_updowncounter("system.procs_blocked")

def proc_stat_observer(observe: Callable[[Asynchronous, Measurement], None]) -> None:
    with open("/proc/stat") as procstat:
        procstat.readline()  # skip the first line
        for line in procstat:
            if line.startswith("cpu"):
                cpu, *states = line.split()
                observe(cpuTime, Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
                observe(cpuTime, Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
            if line.startswith("procs_running"):
                var, value = line.split()
                observe(procsRunning, Measurement(value))
            if line.startswith("procs_blocked"):
                var, value = line.split()
                observe(procsBlocked, Measurement(value))

callback = meter.register_callback(proc_stat_observer, [cpuTime, procsRunning, procsBlocked])

def stop():
    callback.unregister()

Alternatively, the observe function could be an object. This is somewhat similar to this example in the spec

The actual API changes to implement this are

• remove callback as a parameter during observable instrument creation
• add Meter.register_callback(callback: Callable[[], Iterable[Measurement]]) -> CallbackHandle
• add something like a CallbackHandle which just has an unregister() method.

Pros:

• Somewhat similar to what we already have
• Single vs multi instrument callbacks are handled the same way
• Avoid adding anything to the Asynchronous instruments

Cons:

• a little awkward IMO
• Makes the single instrument callback case more difficult

Approach 3

We could restructure the observable callback to simply call observe() on async instruments, similar to what is suggested in https://github.com/open-telemetry/opentelemetry-specification/issues/2280

asyncCounter := meter.NewAsyncCounter(...)
asyncGauge := meter.NewAsyncGauge(...)
cbfunc := func(ctx context.Context) {
    expensiveResult := expensiveCall()
    asyncCounter.Observe(ctx, expensiveResult.Count)
    asyncGauge.Observe(ctx, expensiveResult.Gauge)
}
meter.NewCallback(cbfunc, asyncCounter, asyncGauge)

Pros:

• Simpler API

Cons:

• In the Observe() function, you have to validate that it is being invoked from a callback. The code above I believe is trying to do this by using the ctx param to the callback as a token to pass to Observe(). This is a little weird in Python. We could probably hack around this with OTel implicit context or a custom contextvars.
• Makes the single instrument callback case more difficult

---

Any other ideas for handling this?

Start with the observable instrument base class in opentelemetry-api/src/opentelemetry/_metrics/instrument.py and the current Meter instrument-creation API. Read specification issues 2280 and 2232 plus pull request 2317 before choosing among the proposed callback designs. Done means the selected API supports callbacks covering multiple instruments and callback unregistration, with corresponding implementation and tests.