Allow multiple async callbacks, allow callbacks to be removed

[jack-berg]

There has been discussion in the java SIG and at the spec level about allowing asynchronous instruments to have multiple callbacks. This PR demonstrates how that might function. One of the issues that comes up with supporting multiple callbacks is the need to remove or unregister them at a later time. This PR adds a remove() method to the Observable* instrument interfaces.

Related java issues: #4013, #3549

Related spec issues: #2249, #2232

[anuraaga]

It doesn't seem like removing a callback should cause total unregistration. From my reading of the code, it looks like this should but wouldn't work

addCallback
removeCallback
addCallback

Second addCallback would be unregistered

If the storage gets recreated at a higher level then maybe it's ok but I'd expect symmetry either at this level or a higher level, currently add and remove aren't symmetric and makes me wonder if there is possible thread safety issues at that higher level if this is true.

[anuraaga]

Also I wonder if unregistration causes the metric to stop being reported (not sure so just confirming) - if that is true, then we should avoid it for another reason, we'd want async and sync behavior to be consistent. There is currently no way for a sync metric to stop being reported, so we would want to solve both at the same time rather than just one.

[jack-berg]

If the storage gets recreated at a higher level then maybe it's ok

That's what happening - the second time you call addCallback a new AynchronousMetricStorage is registered with the MetricStorageRegistry because none currently exists at that point. Its likely an unnecessary optimization, since it only frees up resources of asynchronous metrics that have no more callbacks registered at all, which is probably unusual.

Also I wonder if unregistration causes the metric to stop being reported (not sure so just confirming)

Asynchronous metrics are different from synchronous metrics in that we delegate the management of the metrics stream to the callback. So today (i.e. without this code) if a callback stops reporting a metric stream, it will stop being exported to cumulative exporters. In contrast, synchronous instruments continue to report values to exporters even when there is no data reported to the metric stream during the collection interval. This PR doesn't address that behavior.

Here's a little demo of what I mean:

    InMemoryMetricReader cumulativeReader = InMemoryMetricReader.create();
    Meter meter = SdkMeterProvider.builder()
        .registerMetricReader(cumulativeReader)
        .setMinimumCollectionInterval(Duration.ofSeconds(0))
        .build()
        .get("test-meter");

    AtomicLong counter = new AtomicLong(2);
    meter.counterBuilder("foo").buildWithCallback(measurement -> {
      long count = counter.decrementAndGet();
      if (count >= 0) {
        measurement.record(10);
      }
    });

    for (int i = 0; i < 5; i++) {
      System.out.println(cumulativeReader.collectAllMetrics());
    }

That code yields:

[...data...]
[...data...]
[]
[]
[]

[codecov]

Codecov Report

Merging #4143 (7dd81c7) into main (c053393) will decrease coverage by 0.01%.
The diff coverage is 91.30%.

@@             Coverage Diff              @@
##               main    #4143      +/-   ##
============================================
- Coverage     90.24%   90.23%   -0.02%     
- Complexity     4744     4745       +1     
============================================
  Files           547      554       +7     
  Lines         14570    14591      +21     
  Branches       1400     1402       +2     
============================================
+ Hits          13149    13166      +17     
- Misses          958      966       +8     
+ Partials        463      459       -4     

Impacted Files	Coverage Δ
...telemetry/api/metrics/ObservableDoubleCounter.java	0.00% <0.00%> (ø)
...entelemetry/api/metrics/ObservableDoubleGauge.java	0.00% <0.00%> (ø)
...try/api/metrics/ObservableDoubleUpDownCounter.java	0.00% <0.00%> (ø)
...entelemetry/api/metrics/ObservableLongCounter.java	0.00% <0.00%> (ø)
...opentelemetry/api/metrics/ObservableLongGauge.java	0.00% <0.00%> (ø)
...metry/api/metrics/ObservableLongUpDownCounter.java	0.00% <0.00%> (ø)
.../metrics/internal/descriptor/MetricDescriptor.java	100.00% <ø> (+15.78%)	⬆️
...lemetry/sdk/metrics/internal/state/DebugUtils.java	100.00% <ø> (ø)
.../metrics/internal/state/MetricStorageRegistry.java	100.00% <ø> (ø)
...y/sdk/metrics/internal/state/MeterSharedState.java	95.38% <93.75%> (+1.15%)	⬆️
... and 13 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update c053393...7dd81c7. Read the comment docs.

[jack-berg]

Now that #2317 has been merged in the spec, we should be unblocked to merge this. Before we do, a question: Is it clear / intuitive that when you call Observable{Type}{Instrument}#close() that the callback associated with the #buildWithCallback() is the one removed? Put another way, is it obvious that #close() removes one callback and not all the callbacks associated with the instrument?

[jack-berg]

@jkwatson are you comfortable with merging this?

[anuraaga]

Nit, I think

if (!compareAndSet) {
 LOG
 return
}

is a bit of a better pattern so that logic can be increased at the unindented level, the false case is constant 2 LoC no matter what happens.