Skip to content

Commit 91211b3

Browse files
authored
Align collection start times (#2679)
* Align collection start times Fixes #2677 * Undo unnecessary change * Use fixed values for more deterministic tests * Align start times * Rename variable * Add time align test case * Remove unnecessary views * Add more test cases
1 parent e8fbb08 commit 91211b3

File tree

7 files changed

+545
-162
lines changed

7 files changed

+545
-162
lines changed

opentelemetry-sdk/src/opentelemetry/sdk/_metrics/_internal/_view_instrument_match.py

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
from opentelemetry.sdk._metrics._internal.measurement import Measurement
2929
from opentelemetry.sdk._metrics._internal.point import DataPointT
3030
from opentelemetry.sdk._metrics._internal.view import View
31+
from opentelemetry.util._time import _time_ns
3132

3233
_logger = getLogger(__name__)
3334

@@ -39,6 +40,7 @@ def __init__(
3940
instrument: Instrument,
4041
instrument_class_aggregation: Dict[type, Aggregation],
4142
):
43+
self._start_time_unix_nano = _time_ns()
4244
self._view = view
4345
self._instrument = instrument
4446
self._attributes_aggregation: Dict[frozenset, _Aggregation] = {}
@@ -50,12 +52,12 @@ def __init__(
5052
)
5153
if not isinstance(self._view._aggregation, DefaultAggregation):
5254
self._aggregation = self._view._aggregation._create_aggregation(
53-
self._instrument, None
55+
self._instrument, None, 0
5456
)
5557
else:
5658
self._aggregation = self._instrument_class_aggregation[
5759
self._instrument.__class__
58-
]._create_aggregation(self._instrument, None)
60+
]._create_aggregation(self._instrument, None, 0)
5961

6062
def conflicts(self, other: "_ViewInstrumentMatch") -> bool:
6163
# pylint: disable=protected-access
@@ -103,21 +105,31 @@ def consume_measurement(self, measurement: Measurement) -> None:
103105
):
104106
aggregation = (
105107
self._view._aggregation._create_aggregation(
106-
self._instrument, attributes
108+
self._instrument,
109+
attributes,
110+
self._start_time_unix_nano,
107111
)
108112
)
109113
else:
110114
aggregation = self._instrument_class_aggregation[
111115
self._instrument.__class__
112-
]._create_aggregation(self._instrument, attributes)
116+
]._create_aggregation(
117+
self._instrument,
118+
attributes,
119+
self._start_time_unix_nano,
120+
)
113121
self._attributes_aggregation[attributes] = aggregation
114122

115123
self._attributes_aggregation[attributes].aggregate(measurement)
116124

117125
def collect(
118-
self, aggregation_temporality: AggregationTemporality
126+
self,
127+
aggregation_temporality: AggregationTemporality,
128+
collection_start_nanos: int,
119129
) -> Iterable[DataPointT]:
120130

121131
with self._lock:
122132
for aggregation in self._attributes_aggregation.values():
123-
yield aggregation.collect(aggregation_temporality)
133+
yield aggregation.collect(
134+
aggregation_temporality, collection_start_nanos
135+
)

0 commit comments

Comments
 (0)