We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Implement Drop aggregation
See [Drop aggregation](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#drop-aggregation)

> The Drop Aggregation informs the SDK to ignore/drop all Instrument Measurements for this Aggregation.

This is a bit of a special case. The implementation should probably avoid creating a `ViewInstrumentMatch` altogether when we encounter this aggregation to save memory.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py`
Content:
```
1 # Copyright The OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from abc import ABC, abstractmethod
16 from bisect import bisect_left
17 from dataclasses import replace
18 from logging import getLogger
19 from math import inf
20 from threading import Lock
21 from typing import Generic, List, Optional, Sequence, TypeVar
22 
23 from opentelemetry._metrics.instrument import (
24     Asynchronous,
25     Instrument,
26     Synchronous,
27     _Monotonic,
28 )
29 from opentelemetry.sdk._metrics.measurement import Measurement
30 from opentelemetry.sdk._metrics.point import (
31     AggregationTemporality,
32     Gauge,
33     Histogram,
34     PointT,
35     Sum,
36 )
37 from opentelemetry.util._time import _time_ns
38 
39 _PointVarT = TypeVar("_PointVarT", bound=PointT)
40 
41 _logger = getLogger(__name__)
42 
43 
44 class _Aggregation(ABC, Generic[_PointVarT]):
45     def __init__(self):
46         self._lock = Lock()
47 
48     @abstractmethod
49     def aggregate(self, measurement: Measurement) -> None:
50         pass
51 
52     @abstractmethod
53     def collect(self) -> Optional[_PointVarT]:
54         pass
55 
56 
57 class _SumAggregation(_Aggregation[Sum]):
58     def __init__(
59         self,
60         instrument_is_monotonic: bool,
61         instrument_temporality: AggregationTemporality,
62     ):
63         super().__init__()
64 
65         self._start_time_unix_nano = _time_ns()
66         self._instrument_temporality = instrument_temporality
67         self._instrument_is_monotonic = instrument_is_monotonic
68 
69         if self._instrument_temporality is AggregationTemporality.DELTA:
70             self._value = 0
71         else:
72             self._value = None
73 
74     def aggregate(self, measurement: Measurement) -> None:
75         with self._lock:
76             if self._value is None:
77                 self._value = 0
78             self._value = self._value + measurement.value
79 
80     def collect(self) -> Optional[Sum]:
81         """
82         Atomically return a point for the current value of the metric and
83         reset the aggregation value.
84         """
85         now = _time_ns()
86 
87         if self._instrument_temporality is AggregationTemporality.DELTA:
88 
89             with self._lock:
90                 value = self._value
91                 start_time_unix_nano = self._start_time_unix_nano
92 
93                 self._value = 0
94                 self._start_time_unix_nano = now + 1
95 
96             return Sum(
97                 aggregation_temporality=AggregationTemporality.DELTA,
98                 is_monotonic=self._instrument_is_monotonic,
99                 start_time_unix_nano=start_time_unix_nano,
100                 time_unix_nano=now,
101                 value=value,
102             )
103 
104         with self._lock:
105             if self._value is None:
106                 return None
107             value = self._value
108             self._value = None
109 
110         return Sum(
111             aggregation_temporality=AggregationTemporality.CUMULATIVE,
112             is_monotonic=self._instrument_is_monotonic,
113             start_time_unix_nano=self._start_time_unix_nano,
114             time_unix_nano=now,
115             value=value,
116         )
117 
118 
119 class _LastValueAggregation(_Aggregation[Gauge]):
120     def __init__(self):
121         super().__init__()
122         self._value = None
123 
124     def aggregate(self, measurement: Measurement):
125         with self._lock:
126             self._value = measurement.value
127 
128     def collect(self) -> Optional[Gauge]:
129         """
130         Atomically return a point for the current value of the metric.
131         """
132         with self._lock:
133             if self._value is None:
134                 return None
135             value = self._value
136             self._value = None
137 
138         return Gauge(
139             time_unix_nano=_time_ns(),
140             value=value,
141         )
142 
143 
144 class _ExplicitBucketHistogramAggregation(_Aggregation[Histogram]):
145     def __init__(
146         self,
147         boundaries: Sequence[float] = (
148             0.0,
149             5.0,
150             10.0,
151             25.0,
152             50.0,
153             75.0,
154             100.0,
155             250.0,
156             500.0,
157             1000.0,
158         ),
159         record_min_max: bool = True,
160     ):
161         super().__init__()
162         self._boundaries = tuple(boundaries)
163         self._bucket_counts = self._get_empty_bucket_counts()
164         self._min = inf
165         self._max = -inf
166         self._sum = 0
167         self._record_min_max = record_min_max
168         self._start_time_unix_nano = _time_ns()
169 
170     def _get_empty_bucket_counts(self) -> List[int]:
171         return [0] * (len(self._boundaries) + 1)
172 
173     def aggregate(self, measurement: Measurement) -> None:
174 
175         value = measurement.value
176 
177         if self._record_min_max:
178             self._min = min(self._min, value)
179             self._max = max(self._max, value)
180 
181         self._sum += value
182 
183         self._bucket_counts[bisect_left(self._boundaries, value)] += 1
184 
185     def collect(self) -> Histogram:
186         """
187         Atomically return a point for the current value of the metric.
188         """
189         now = _time_ns()
190 
191         with self._lock:
192             value = self._bucket_counts
193             start_time_unix_nano = self._start_time_unix_nano
194             histogram_sum = self._sum
195 
196             self._bucket_counts = self._get_empty_bucket_counts()
197             self._start_time_unix_nano = now + 1
198             self._sum = 0
199 
200         return Histogram(
201             start_time_unix_nano=start_time_unix_nano,
202             time_unix_nano=now,
203             bucket_counts=tuple(value),
204             explicit_bounds=self._boundaries,
205             aggregation_temporality=AggregationTemporality.DELTA,
206             sum=histogram_sum,
207         )
208 
209 
210 # pylint: disable=too-many-return-statements,too-many-branches
211 def _convert_aggregation_temporality(
212     previous_point: Optional[_PointVarT],
213     current_point: _PointVarT,
214     aggregation_temporality: AggregationTemporality,
215 ) -> _PointVarT:
216     """Converts `current_point` to the requested `aggregation_temporality`
217     given the `previous_point`.
218 
219     `previous_point` must have `CUMULATIVE` temporality. `current_point` may
220     have `DELTA` or `CUMULATIVE` temporality.
221 
222     The output point will have temporality `aggregation_temporality`. Since
223     `GAUGE` points have no temporality, they are returned unchanged.
224     """
225 
226     current_point_type = type(current_point)
227 
228     if current_point_type is Gauge:
229         return current_point
230 
231     if previous_point is not None and type(previous_point) is not type(
232         current_point
233     ):
234         _logger.warning(
235             "convert_aggregation_temporality called with mismatched "
236             "point types: %s and %s",
237             type(previous_point),
238             current_point_type,
239         )
240 
241         return current_point
242 
243     if current_point_type is Sum:
244         if previous_point is None:
245             # Output CUMULATIVE for a synchronous instrument
246             # There is no previous value, return the delta point as a
247             # cumulative
248             return replace(
249                 current_point, aggregation_temporality=aggregation_temporality
250             )
251         if previous_point.aggregation_temporality is not (
252             AggregationTemporality.CUMULATIVE
253         ):
254             raise Exception(
255                 "previous_point aggregation temporality must be CUMULATIVE"
256             )
257 
258         if current_point.aggregation_temporality is aggregation_temporality:
259             # Output DELTA for a synchronous instrument
260             # Output CUMULATIVE for an asynchronous instrument
261             return current_point
262 
263         if aggregation_temporality is AggregationTemporality.DELTA:
264             # Output temporality DELTA for an asynchronous instrument
265             value = current_point.value - previous_point.value
266             output_start_time_unix_nano = previous_point.time_unix_nano
267 
268         else:
269             # Output CUMULATIVE for a synchronous instrument
270             value = current_point.value + previous_point.value
271             output_start_time_unix_nano = previous_point.start_time_unix_nano
272 
273         is_monotonic = (
274             previous_point.is_monotonic and current_point.is_monotonic
275         )
276 
277         return Sum(
278             start_time_unix_nano=output_start_time_unix_nano,
279             time_unix_nano=current_point.time_unix_nano,
280             value=value,
281             aggregation_temporality=aggregation_temporality,
282             is_monotonic=is_monotonic,
283         )
284 
285     if current_point_type is Histogram:
286         if previous_point is None:
287             return replace(
288                 current_point, aggregation_temporality=aggregation_temporality
289             )
290         if previous_point.aggregation_temporality is not (
291             AggregationTemporality.CUMULATIVE
292         ):
293             raise Exception(
294                 "previous_point aggregation temporality must be CUMULATIVE"
295             )
296 
297         if current_point.aggregation_temporality is aggregation_temporality:
298             return current_point
299 
300         if aggregation_temporality is AggregationTemporality.CUMULATIVE:
301             start_time_unix_nano = previous_point.start_time_unix_nano
302             sum_ = current_point.sum + previous_point.sum
303             bucket_counts = [
304                 curr_count + prev_count
305                 for curr_count, prev_count in zip(
306                     current_point.bucket_counts, previous_point.bucket_counts
307                 )
308             ]
309         else:
310             start_time_unix_nano = previous_point.time_unix_nano
311             sum_ = current_point.sum - previous_point.sum
312             bucket_counts = [
313                 curr_count - prev_count
314                 for curr_count, prev_count in zip(
315                     current_point.bucket_counts, previous_point.bucket_counts
316                 )
317             ]
318 
319         return Histogram(
320             start_time_unix_nano=start_time_unix_nano,
321             time_unix_nano=current_point.time_unix_nano,
322             bucket_counts=bucket_counts,
323             explicit_bounds=current_point.explicit_bounds,
324             sum=sum_,
325             aggregation_temporality=aggregation_temporality,
326         )
327     return None
328 
329 
330 class _AggregationFactory(ABC):
331     @abstractmethod
332     def _create_aggregation(self, instrument: Instrument) -> _Aggregation:
333         """Creates an aggregation"""
334 
335 
336 class ExplicitBucketHistogramAggregation(_AggregationFactory):
337     def __init__(
338         self,
339         boundaries: Sequence[float] = (
340             0.0,
341             5.0,
342             10.0,
343             25.0,
344             50.0,
345             75.0,
346             100.0,
347             250.0,
348             500.0,
349             1000.0,
350         ),
351         record_min_max: bool = True,
352     ) -> None:
353         self._boundaries = boundaries
354         self._record_min_max = record_min_max
355 
356     def _create_aggregation(self, instrument: Instrument) -> _Aggregation:
357         return _ExplicitBucketHistogramAggregation(
358             boundaries=self._boundaries,
359             record_min_max=self._record_min_max,
360         )
361 
362 
363 class SumAggregation(_AggregationFactory):
364     def _create_aggregation(self, instrument: Instrument) -> _Aggregation:
365 
366         temporality = AggregationTemporality.UNSPECIFIED
367         if isinstance(instrument, Synchronous):
368             temporality = AggregationTemporality.DELTA
369         elif isinstance(instrument, Asynchronous):
370             temporality = AggregationTemporality.CUMULATIVE
371 
372         return _SumAggregation(
373             isinstance(instrument, _Monotonic),
374             temporality,
375         )
376 
377 
378 class LastValueAggregation(_AggregationFactory):
379     def _create_aggregation(self, instrument: Instrument) -> _Aggregation:
380         return _LastValueAggregation()
381
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py
--- a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py
@@ -54,6 +54,14 @@
         pass
 
 
+class _DropAggregation(_Aggregation):
+    def aggregate(self, measurement: Measurement) -> None:
+        pass
+
+    def collect(self) -> Optional[_PointVarT]:
+        pass
+
+
 class _SumAggregation(_Aggregation[Sum]):
     def __init__(
         self,
@@ -378,3 +386,10 @@
 class LastValueAggregation(_AggregationFactory):
     def _create_aggregation(self, instrument: Instrument) -> _Aggregation:
         return _LastValueAggregation()
+
+
+class DropAggregation(_AggregationFactory):
+    """Using this aggregation will make all measurements be ignored."""
+
+    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:
+        return _DropAggregation()

{"golden_diff": "diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py\n--- a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py\n+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py\n@@ -54,6 +54,14 @@\n         pass\n \n \n+class _DropAggregation(_Aggregation):\n+    def aggregate(self, measurement: Measurement) -> None:\n+        pass\n+\n+    def collect(self) -> Optional[_PointVarT]:\n+        pass\n+\n+\n class _SumAggregation(_Aggregation[Sum]):\n     def __init__(\n         self,\n@@ -378,3 +386,10 @@\n class LastValueAggregation(_AggregationFactory):\n     def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n         return _LastValueAggregation()\n+\n+\n+class DropAggregation(_AggregationFactory):\n+    \"\"\"Using this aggregation will make all measurements be ignored.\"\"\"\n+\n+    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n+        return _DropAggregation()\n", "issue": "Implement Drop aggregation\nSee [Drop aggregation](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#drop-aggregation)\r\n\r\n> The Drop Aggregation informs the SDK to ignore/drop all Instrument Measurements for this Aggregation.\r\n\r\nThis is a bit of a special case. The implementation should probably avoid creating a `ViewInstrumentMatch` altogether when we encounter this aggregation to save memory.\n", "before_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom abc import ABC, abstractmethod\nfrom bisect import bisect_left\nfrom dataclasses import replace\nfrom logging import getLogger\nfrom math import inf\nfrom threading import Lock\nfrom typing import Generic, List, Optional, Sequence, TypeVar\n\nfrom opentelemetry._metrics.instrument import (\n    Asynchronous,\n    Instrument,\n    Synchronous,\n    _Monotonic,\n)\nfrom opentelemetry.sdk._metrics.measurement import Measurement\nfrom opentelemetry.sdk._metrics.point import (\n    AggregationTemporality,\n    Gauge,\n    Histogram,\n    PointT,\n    Sum,\n)\nfrom opentelemetry.util._time import _time_ns\n\n_PointVarT = TypeVar(\"_PointVarT\", bound=PointT)\n\n_logger = getLogger(__name__)\n\n\nclass _Aggregation(ABC, Generic[_PointVarT]):\n    def __init__(self):\n        self._lock = Lock()\n\n    @abstractmethod\n    def aggregate(self, measurement: Measurement) -> None:\n        pass\n\n    @abstractmethod\n    def collect(self) -> Optional[_PointVarT]:\n        pass\n\n\nclass _SumAggregation(_Aggregation[Sum]):\n    def __init__(\n        self,\n        instrument_is_monotonic: bool,\n        instrument_temporality: AggregationTemporality,\n    ):\n        super().__init__()\n\n        self._start_time_unix_nano = _time_ns()\n        self._instrument_temporality = instrument_temporality\n        self._instrument_is_monotonic = instrument_is_monotonic\n\n        if self._instrument_temporality is AggregationTemporality.DELTA:\n            self._value = 0\n        else:\n            self._value = None\n\n    def aggregate(self, measurement: Measurement) -> None:\n        with self._lock:\n            if self._value is None:\n                self._value = 0\n            self._value = self._value + measurement.value\n\n    def collect(self) -> Optional[Sum]:\n        \"\"\"\n        Atomically return a point for the current value of the metric and\n        reset the aggregation value.\n        \"\"\"\n        now = _time_ns()\n\n        if self._instrument_temporality is AggregationTemporality.DELTA:\n\n            with self._lock:\n                value = self._value\n                start_time_unix_nano = self._start_time_unix_nano\n\n                self._value = 0\n                self._start_time_unix_nano = now + 1\n\n            return Sum(\n                aggregation_temporality=AggregationTemporality.DELTA,\n                is_monotonic=self._instrument_is_monotonic,\n                start_time_unix_nano=start_time_unix_nano,\n                time_unix_nano=now,\n                value=value,\n            )\n\n        with self._lock:\n            if self._value is None:\n                return None\n            value = self._value\n            self._value = None\n\n        return Sum(\n            aggregation_temporality=AggregationTemporality.CUMULATIVE,\n            is_monotonic=self._instrument_is_monotonic,\n            start_time_unix_nano=self._start_time_unix_nano,\n            time_unix_nano=now,\n            value=value,\n        )\n\n\nclass _LastValueAggregation(_Aggregation[Gauge]):\n    def __init__(self):\n        super().__init__()\n        self._value = None\n\n    def aggregate(self, measurement: Measurement):\n        with self._lock:\n            self._value = measurement.value\n\n    def collect(self) -> Optional[Gauge]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        with self._lock:\n            if self._value is None:\n                return None\n            value = self._value\n            self._value = None\n\n        return Gauge(\n            time_unix_nano=_time_ns(),\n            value=value,\n        )\n\n\nclass _ExplicitBucketHistogramAggregation(_Aggregation[Histogram]):\n    def __init__(\n        self,\n        boundaries: Sequence[float] = (\n            0.0,\n            5.0,\n            10.0,\n            25.0,\n            50.0,\n            75.0,\n            100.0,\n            250.0,\n            500.0,\n            1000.0,\n        ),\n        record_min_max: bool = True,\n    ):\n        super().__init__()\n        self._boundaries = tuple(boundaries)\n        self._bucket_counts = self._get_empty_bucket_counts()\n        self._min = inf\n        self._max = -inf\n        self._sum = 0\n        self._record_min_max = record_min_max\n        self._start_time_unix_nano = _time_ns()\n\n    def _get_empty_bucket_counts(self) -> List[int]:\n        return [0] * (len(self._boundaries) + 1)\n\n    def aggregate(self, measurement: Measurement) -> None:\n\n        value = measurement.value\n\n        if self._record_min_max:\n            self._min = min(self._min, value)\n            self._max = max(self._max, value)\n\n        self._sum += value\n\n        self._bucket_counts[bisect_left(self._boundaries, value)] += 1\n\n    def collect(self) -> Histogram:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        now = _time_ns()\n\n        with self._lock:\n            value = self._bucket_counts\n            start_time_unix_nano = self._start_time_unix_nano\n            histogram_sum = self._sum\n\n            self._bucket_counts = self._get_empty_bucket_counts()\n            self._start_time_unix_nano = now + 1\n            self._sum = 0\n\n        return Histogram(\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=now,\n            bucket_counts=tuple(value),\n            explicit_bounds=self._boundaries,\n            aggregation_temporality=AggregationTemporality.DELTA,\n            sum=histogram_sum,\n        )\n\n\n# pylint: disable=too-many-return-statements,too-many-branches\ndef _convert_aggregation_temporality(\n    previous_point: Optional[_PointVarT],\n    current_point: _PointVarT,\n    aggregation_temporality: AggregationTemporality,\n) -> _PointVarT:\n    \"\"\"Converts `current_point` to the requested `aggregation_temporality`\n    given the `previous_point`.\n\n    `previous_point` must have `CUMULATIVE` temporality. `current_point` may\n    have `DELTA` or `CUMULATIVE` temporality.\n\n    The output point will have temporality `aggregation_temporality`. Since\n    `GAUGE` points have no temporality, they are returned unchanged.\n    \"\"\"\n\n    current_point_type = type(current_point)\n\n    if current_point_type is Gauge:\n        return current_point\n\n    if previous_point is not None and type(previous_point) is not type(\n        current_point\n    ):\n        _logger.warning(\n            \"convert_aggregation_temporality called with mismatched \"\n            \"point types: %s and %s\",\n            type(previous_point),\n            current_point_type,\n        )\n\n        return current_point\n\n    if current_point_type is Sum:\n        if previous_point is None:\n            # Output CUMULATIVE for a synchronous instrument\n            # There is no previous value, return the delta point as a\n            # cumulative\n            return replace(\n                current_point, aggregation_temporality=aggregation_temporality\n            )\n        if previous_point.aggregation_temporality is not (\n            AggregationTemporality.CUMULATIVE\n        ):\n            raise Exception(\n                \"previous_point aggregation temporality must be CUMULATIVE\"\n            )\n\n        if current_point.aggregation_temporality is aggregation_temporality:\n            # Output DELTA for a synchronous instrument\n            # Output CUMULATIVE for an asynchronous instrument\n            return current_point\n\n        if aggregation_temporality is AggregationTemporality.DELTA:\n            # Output temporality DELTA for an asynchronous instrument\n            value = current_point.value - previous_point.value\n            output_start_time_unix_nano = previous_point.time_unix_nano\n\n        else:\n            # Output CUMULATIVE for a synchronous instrument\n            value = current_point.value + previous_point.value\n            output_start_time_unix_nano = previous_point.start_time_unix_nano\n\n        is_monotonic = (\n            previous_point.is_monotonic and current_point.is_monotonic\n        )\n\n        return Sum(\n            start_time_unix_nano=output_start_time_unix_nano,\n            time_unix_nano=current_point.time_unix_nano,\n            value=value,\n            aggregation_temporality=aggregation_temporality,\n            is_monotonic=is_monotonic,\n        )\n\n    if current_point_type is Histogram:\n        if previous_point is None:\n            return replace(\n                current_point, aggregation_temporality=aggregation_temporality\n            )\n        if previous_point.aggregation_temporality is not (\n            AggregationTemporality.CUMULATIVE\n        ):\n            raise Exception(\n                \"previous_point aggregation temporality must be CUMULATIVE\"\n            )\n\n        if current_point.aggregation_temporality is aggregation_temporality:\n            return current_point\n\n        if aggregation_temporality is AggregationTemporality.CUMULATIVE:\n            start_time_unix_nano = previous_point.start_time_unix_nano\n            sum_ = current_point.sum + previous_point.sum\n            bucket_counts = [\n                curr_count + prev_count\n                for curr_count, prev_count in zip(\n                    current_point.bucket_counts, previous_point.bucket_counts\n                )\n            ]\n        else:\n            start_time_unix_nano = previous_point.time_unix_nano\n            sum_ = current_point.sum - previous_point.sum\n            bucket_counts = [\n                curr_count - prev_count\n                for curr_count, prev_count in zip(\n                    current_point.bucket_counts, previous_point.bucket_counts\n                )\n            ]\n\n        return Histogram(\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=current_point.time_unix_nano,\n            bucket_counts=bucket_counts,\n            explicit_bounds=current_point.explicit_bounds,\n            sum=sum_,\n            aggregation_temporality=aggregation_temporality,\n        )\n    return None\n\n\nclass _AggregationFactory(ABC):\n    @abstractmethod\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n        \"\"\"Creates an aggregation\"\"\"\n\n\nclass ExplicitBucketHistogramAggregation(_AggregationFactory):\n    def __init__(\n        self,\n        boundaries: Sequence[float] = (\n            0.0,\n            5.0,\n            10.0,\n            25.0,\n            50.0,\n            75.0,\n            100.0,\n            250.0,\n            500.0,\n            1000.0,\n        ),\n        record_min_max: bool = True,\n    ) -> None:\n        self._boundaries = boundaries\n        self._record_min_max = record_min_max\n\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n        return _ExplicitBucketHistogramAggregation(\n            boundaries=self._boundaries,\n            record_min_max=self._record_min_max,\n        )\n\n\nclass SumAggregation(_AggregationFactory):\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n\n        temporality = AggregationTemporality.UNSPECIFIED\n        if isinstance(instrument, Synchronous):\n            temporality = AggregationTemporality.DELTA\n        elif isinstance(instrument, Asynchronous):\n            temporality = AggregationTemporality.CUMULATIVE\n\n        return _SumAggregation(\n            isinstance(instrument, _Monotonic),\n            temporality,\n        )\n\n\nclass LastValueAggregation(_AggregationFactory):\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n        return _LastValueAggregation()\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py"}], "after_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom abc import ABC, abstractmethod\nfrom bisect import bisect_left\nfrom dataclasses import replace\nfrom logging import getLogger\nfrom math import inf\nfrom threading import Lock\nfrom typing import Generic, List, Optional, Sequence, TypeVar\n\nfrom opentelemetry._metrics.instrument import (\n    Asynchronous,\n    Instrument,\n    Synchronous,\n    _Monotonic,\n)\nfrom opentelemetry.sdk._metrics.measurement import Measurement\nfrom opentelemetry.sdk._metrics.point import (\n    AggregationTemporality,\n    Gauge,\n    Histogram,\n    PointT,\n    Sum,\n)\nfrom opentelemetry.util._time import _time_ns\n\n_PointVarT = TypeVar(\"_PointVarT\", bound=PointT)\n\n_logger = getLogger(__name__)\n\n\nclass _Aggregation(ABC, Generic[_PointVarT]):\n    def __init__(self):\n        self._lock = Lock()\n\n    @abstractmethod\n    def aggregate(self, measurement: Measurement) -> None:\n        pass\n\n    @abstractmethod\n    def collect(self) -> Optional[_PointVarT]:\n        pass\n\n\nclass _DropAggregation(_Aggregation):\n    def aggregate(self, measurement: Measurement) -> None:\n        pass\n\n    def collect(self) -> Optional[_PointVarT]:\n        pass\n\n\nclass _SumAggregation(_Aggregation[Sum]):\n    def __init__(\n        self,\n        instrument_is_monotonic: bool,\n        instrument_temporality: AggregationTemporality,\n    ):\n        super().__init__()\n\n        self._start_time_unix_nano = _time_ns()\n        self._instrument_temporality = instrument_temporality\n        self._instrument_is_monotonic = instrument_is_monotonic\n\n        if self._instrument_temporality is AggregationTemporality.DELTA:\n            self._value = 0\n        else:\n            self._value = None\n\n    def aggregate(self, measurement: Measurement) -> None:\n        with self._lock:\n            if self._value is None:\n                self._value = 0\n            self._value = self._value + measurement.value\n\n    def collect(self) -> Optional[Sum]:\n        \"\"\"\n        Atomically return a point for the current value of the metric and\n        reset the aggregation value.\n        \"\"\"\n        now = _time_ns()\n\n        if self._instrument_temporality is AggregationTemporality.DELTA:\n\n            with self._lock:\n                value = self._value\n                start_time_unix_nano = self._start_time_unix_nano\n\n                self._value = 0\n                self._start_time_unix_nano = now + 1\n\n            return Sum(\n                aggregation_temporality=AggregationTemporality.DELTA,\n                is_monotonic=self._instrument_is_monotonic,\n                start_time_unix_nano=start_time_unix_nano,\n                time_unix_nano=now,\n                value=value,\n            )\n\n        with self._lock:\n            if self._value is None:\n                return None\n            value = self._value\n            self._value = None\n\n        return Sum(\n            aggregation_temporality=AggregationTemporality.CUMULATIVE,\n            is_monotonic=self._instrument_is_monotonic,\n            start_time_unix_nano=self._start_time_unix_nano,\n            time_unix_nano=now,\n            value=value,\n        )\n\n\nclass _LastValueAggregation(_Aggregation[Gauge]):\n    def __init__(self):\n        super().__init__()\n        self._value = None\n\n    def aggregate(self, measurement: Measurement):\n        with self._lock:\n            self._value = measurement.value\n\n    def collect(self) -> Optional[Gauge]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        with self._lock:\n            if self._value is None:\n                return None\n            value = self._value\n            self._value = None\n\n        return Gauge(\n            time_unix_nano=_time_ns(),\n            value=value,\n        )\n\n\nclass _ExplicitBucketHistogramAggregation(_Aggregation[Histogram]):\n    def __init__(\n        self,\n        boundaries: Sequence[float] = (\n            0.0,\n            5.0,\n            10.0,\n            25.0,\n            50.0,\n            75.0,\n            100.0,\n            250.0,\n            500.0,\n            1000.0,\n        ),\n        record_min_max: bool = True,\n    ):\n        super().__init__()\n        self._boundaries = tuple(boundaries)\n        self._bucket_counts = self._get_empty_bucket_counts()\n        self._min = inf\n        self._max = -inf\n        self._sum = 0\n        self._record_min_max = record_min_max\n        self._start_time_unix_nano = _time_ns()\n\n    def _get_empty_bucket_counts(self) -> List[int]:\n        return [0] * (len(self._boundaries) + 1)\n\n    def aggregate(self, measurement: Measurement) -> None:\n\n        value = measurement.value\n\n        if self._record_min_max:\n            self._min = min(self._min, value)\n            self._max = max(self._max, value)\n\n        self._sum += value\n\n        self._bucket_counts[bisect_left(self._boundaries, value)] += 1\n\n    def collect(self) -> Histogram:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        now = _time_ns()\n\n        with self._lock:\n            value = self._bucket_counts\n            start_time_unix_nano = self._start_time_unix_nano\n            histogram_sum = self._sum\n\n            self._bucket_counts = self._get_empty_bucket_counts()\n            self._start_time_unix_nano = now + 1\n            self._sum = 0\n\n        return Histogram(\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=now,\n            bucket_counts=tuple(value),\n            explicit_bounds=self._boundaries,\n            aggregation_temporality=AggregationTemporality.DELTA,\n            sum=histogram_sum,\n        )\n\n\n# pylint: disable=too-many-return-statements,too-many-branches\ndef _convert_aggregation_temporality(\n    previous_point: Optional[_PointVarT],\n    current_point: _PointVarT,\n    aggregation_temporality: AggregationTemporality,\n) -> _PointVarT:\n    \"\"\"Converts `current_point` to the requested `aggregation_temporality`\n    given the `previous_point`.\n\n    `previous_point` must have `CUMULATIVE` temporality. `current_point` may\n    have `DELTA` or `CUMULATIVE` temporality.\n\n    The output point will have temporality `aggregation_temporality`. Since\n    `GAUGE` points have no temporality, they are returned unchanged.\n    \"\"\"\n\n    current_point_type = type(current_point)\n\n    if current_point_type is Gauge:\n        return current_point\n\n    if previous_point is not None and type(previous_point) is not type(\n        current_point\n    ):\n        _logger.warning(\n            \"convert_aggregation_temporality called with mismatched \"\n            \"point types: %s and %s\",\n            type(previous_point),\n            current_point_type,\n        )\n\n        return current_point\n\n    if current_point_type is Sum:\n        if previous_point is None:\n            # Output CUMULATIVE for a synchronous instrument\n            # There is no previous value, return the delta point as a\n            # cumulative\n            return replace(\n                current_point, aggregation_temporality=aggregation_temporality\n            )\n        if previous_point.aggregation_temporality is not (\n            AggregationTemporality.CUMULATIVE\n        ):\n            raise Exception(\n                \"previous_point aggregation temporality must be CUMULATIVE\"\n            )\n\n        if current_point.aggregation_temporality is aggregation_temporality:\n            # Output DELTA for a synchronous instrument\n            # Output CUMULATIVE for an asynchronous instrument\n            return current_point\n\n        if aggregation_temporality is AggregationTemporality.DELTA:\n            # Output temporality DELTA for an asynchronous instrument\n            value = current_point.value - previous_point.value\n            output_start_time_unix_nano = previous_point.time_unix_nano\n\n        else:\n            # Output CUMULATIVE for a synchronous instrument\n            value = current_point.value + previous_point.value\n            output_start_time_unix_nano = previous_point.start_time_unix_nano\n\n        is_monotonic = (\n            previous_point.is_monotonic and current_point.is_monotonic\n        )\n\n        return Sum(\n            start_time_unix_nano=output_start_time_unix_nano,\n            time_unix_nano=current_point.time_unix_nano,\n            value=value,\n            aggregation_temporality=aggregation_temporality,\n            is_monotonic=is_monotonic,\n        )\n\n    if current_point_type is Histogram:\n        if previous_point is None:\n            return replace(\n                current_point, aggregation_temporality=aggregation_temporality\n            )\n        if previous_point.aggregation_temporality is not (\n            AggregationTemporality.CUMULATIVE\n        ):\n            raise Exception(\n                \"previous_point aggregation temporality must be CUMULATIVE\"\n            )\n\n        if current_point.aggregation_temporality is aggregation_temporality:\n            return current_point\n\n        if aggregation_temporality is AggregationTemporality.CUMULATIVE:\n            start_time_unix_nano = previous_point.start_time_unix_nano\n            sum_ = current_point.sum + previous_point.sum\n            bucket_counts = [\n                curr_count + prev_count\n                for curr_count, prev_count in zip(\n                    current_point.bucket_counts, previous_point.bucket_counts\n                )\n            ]\n        else:\n            start_time_unix_nano = previous_point.time_unix_nano\n            sum_ = current_point.sum - previous_point.sum\n            bucket_counts = [\n                curr_count - prev_count\n                for curr_count, prev_count in zip(\n                    current_point.bucket_counts, previous_point.bucket_counts\n                )\n            ]\n\n        return Histogram(\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=current_point.time_unix_nano,\n            bucket_counts=bucket_counts,\n            explicit_bounds=current_point.explicit_bounds,\n            sum=sum_,\n            aggregation_temporality=aggregation_temporality,\n        )\n    return None\n\n\nclass _AggregationFactory(ABC):\n    @abstractmethod\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n        \"\"\"Creates an aggregation\"\"\"\n\n\nclass ExplicitBucketHistogramAggregation(_AggregationFactory):\n    def __init__(\n        self,\n        boundaries: Sequence[float] = (\n            0.0,\n            5.0,\n            10.0,\n            25.0,\n            50.0,\n            75.0,\n            100.0,\n            250.0,\n            500.0,\n            1000.0,\n        ),\n        record_min_max: bool = True,\n    ) -> None:\n        self._boundaries = boundaries\n        self._record_min_max = record_min_max\n\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n        return _ExplicitBucketHistogramAggregation(\n            boundaries=self._boundaries,\n            record_min_max=self._record_min_max,\n        )\n\n\nclass SumAggregation(_AggregationFactory):\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n\n        temporality = AggregationTemporality.UNSPECIFIED\n        if isinstance(instrument, Synchronous):\n            temporality = AggregationTemporality.DELTA\n        elif isinstance(instrument, Asynchronous):\n            temporality = AggregationTemporality.CUMULATIVE\n\n        return _SumAggregation(\n            isinstance(instrument, _Monotonic),\n            temporality,\n        )\n\n\nclass LastValueAggregation(_AggregationFactory):\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n        return _LastValueAggregation()\n\n\nclass DropAggregation(_AggregationFactory):\n    \"\"\"Using this aggregation will make all measurements be ignored.\"\"\"\n\n    def _create_aggregation(self, instrument: Instrument) -> _Aggregation:\n        return _DropAggregation()\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py"}]}