Support cumulative, delta, and pass-through exporters

[jmacd]

With these changes, Cumulative and Delta-oriented Exporters are able to request the correct kind of aggregation. This is the final step in a re-do of #799, an effort which included #808, #812, and #835.

Resolves open-telemetry/opentelemetry-proto#147.

[james-bebbington]

I made an attempt at reviewing this without me having a lot of context yet. Overall based on the concept / tests, this looks like it makes sense. I don't have enough knowledge of the rest of the SDK pipeline, however, and couldn't quite follow the logic in simple.go. I left a couple of comments / questions that may or may not make sense - feel free to correct me if what I said was totally wrong 😄

[MrAlias]

Overall, this looks good. The simple integrator processing pipeline can use some work to help explain the processing and possibly could use some refactoring like @james-bebbington pointed out.

[jmacd]

@james-bebbington and @MrAlias I think I've addressed your concerns, but you should be the judge. Thanks.

[MrAlias]

Looks good 👍

Thanks for the verbose comments in the simple integrator's Process, they are really helpful.

[james-bebbington]

Overall this PR is great btw. The addition of explicitly describing metrics as Sync/Async, Adding/Grouping, etc, adds some nice clarity & cleared up a few things for me.

I have a few more suggestions around how the clarity of the integrator code might be improved. Partially raising these just to ensure my own understanding is correct 😄

[jmacd]

Thanks @MrAlias and @james-bebbington!

[jmacd]

I failed to push my final commits in response to @james-bebbington. Sorry! I will roll them into a separate PR.

[jmacd]

(it was this chunk)

diff --git a/sdk/metric/integrator/simple/simple.go b/sdk/metric/integrator/simple/simple.go
index 40ccce33..def51169 100644
--- a/sdk/metric/integrator/simple/simple.go
+++ b/sdk/metric/integrator/simple/simple.go
@@ -151,9 +151,20 @@ func (b *Integrator) Process(accum export.Accumulation) error {
 	sameCollection := b.state.finishedCollection == value.updated
 	value.updated = b.state.finishedCollection
 
-	// An existing value will be found for some stateKey when:
-	// (a) stateful aggregation is being used
-	// (b) multiple accumulators are being used.
+	// At this point in the code, we have located an existing
+	// value for some stateKey.  This can be because:
+	//
+	// (a) stateful aggregation is being used, the entry was
+	// entered during a prior collection, and this is the first
+	// time processing an accumulation for this stateKey in the
+	// current collection.  Since this is the first time
+	// processing an accumulation for this stateKey during this
+	// collection, we don't know yet whether there are multiple
+	// accumulators at work.  If there are multiple accumulators,
+	// they'll hit case (b) the second time through.
+	//
+	// (b) multiple accumulators are being used, whether stateful
+	// or not.
 	//
 	// Case (a) occurs when the instrument and the exporter
 	// require memory to work correctly, either because the
@@ -273,9 +284,10 @@ func (b *Integrator) FinishCollection() error {
 		// delta or a cumulative aggregation.
 		var err error
 		if mkind.PrecomputedSum() {
-			// delta_value = current_cumulative_value - previous_cumulative_value
-			if subt, ok := value.current.(export.Subtractor); ok {
-				err = subt.Subtract(value.cumulative, value.delta, key.descriptor)
+			if currentSubtractor, ok := value.current.(export.Subtractor); ok {
+				// This line is equivalent to:
+				// value.delta = currentSubtractor - value.cumulative
+				err = currentSubtractor.Subtract(value.cumulative, value.delta, key.descriptor)
 
 				if err == nil {
 					err = value.current.SynchronizedCopy(value.cumulative, key.descriptor)
@@ -284,7 +296,8 @@ func (b *Integrator) FinishCollection() error {
 				err = aggregation.ErrNoSubtraction
 			}
 		} else {
-			// cumulative_value = previous_cumulative_value + current_delta_value
+			// This line is equivalent to:
+			// value.cumulative = value.cumulative + value.delta
 			err = value.cumulative.Merge(value.current, key.descriptor)
 		}
 		if err != nil {