fix(sdk-metrics): merge uncollected delta accumulations

open-telemetry · Mar 10, 2023 · 20e8415 · 20e8415
1 parent ecb5ebe
commit 20e8415
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 0 deletions.
diff --git a/packages/sdk-metrics/src/state/DeltaMetricProcessor.ts b/packages/sdk-metrics/src/state/DeltaMetricProcessor.ts
@@ -57,6 +57,7 @@ export class DeltaMetricProcessor<T extends Maybe<Accumulation>> {
           this._aggregator.createAccumulation(collectionTime);
         accumulation?.record(value);
         let delta = accumulation;
+        // Diff with recorded cumulative memo.
         if (this._cumulativeMemoStorage.has(attributes, hashCode)) {
           // has() returned true, previous is present.
           // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
@@ -66,6 +67,13 @@ export class DeltaMetricProcessor<T extends Maybe<Accumulation>> {
           )!;
           delta = this._aggregator.diff(previous, accumulation);
         }
+        // Merge with uncollected active delta.
+        if (this._activeCollectionStorage.has(attributes, hashCode)) {
+          // has() returned true, previous is present.
+          // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+          const active = this._activeCollectionStorage.get(attributes, hashCode)!;
+          delta = this._aggregator.merge(active, delta);
+        }
 
         // Save the current record and the delta record.
         this._cumulativeMemoStorage.set(attributes, accumulation, hashCode);

diff --git a/packages/sdk-metrics/test/state/DeltaMetricProcessor.test.ts b/packages/sdk-metrics/test/state/DeltaMetricProcessor.test.ts
@@ -100,6 +100,26 @@ describe('DeltaMetricProcessor', () => {
         assert.strictEqual(accumulation?.toPointValue(), 11);
       }
     });
+
+    it('should merge with active delta of accumulations', () => {
+      const metricProcessor = new DeltaMetricProcessor(new SumAggregator(true));
+
+      {
+        const measurements = new AttributeHashMap<number>();
+        measurements.set({}, 10);
+        metricProcessor.batchCumulate(measurements, [0, 0]);
+      }
+
+      {
+        const measurements = new AttributeHashMap<number>();
+        measurements.set({}, 20);
+        metricProcessor.batchCumulate(measurements, [1, 1]);
+      }
+
+      const accumulations = metricProcessor.collect();
+      const accumulation = accumulations.get({});
+      assert.strictEqual(accumulation?.toPointValue(), 20);
+    });
   });
 
   describe('collect', () => {