Merge branch 'main' into middleware-always-set-status-code-on-duratio…

…n-attrs

.pylintrc

@@ -45,6 +45,9 @@ suggestion-mode=yes
 # active Python interpreter and may run arbitrary code.
 unsafe-load-any-extension=no
 
+# Run python dependant checks considering the baseline version
+py-version=3.8
+
 
 [MESSAGES CONTROL]
 

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-instrumentation-django` Handle exceptions from request/response hooks
+  ([#2153](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2153))
+- `opentelemetry-instrumentation-asyncio` instrumented `asyncio.wait_for` properly raises `asyncio.TimeoutError` as expected
+  ([#2637](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2637))
 - `opentelemetry-instrumentation-aws-lambda` Bugfix: AWS Lambda event source key incorrect for SNS in instrumentation library. 
   ([#2612](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2612))
 - `opentelemetry-instrumentation-system-metrics` Permit to use psutil 6.0+.
@@ -23,6 +27,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2573](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2573))
 - `opentelemetry-instrumentation-confluent-kafka` Add support for version 2.4.0 of confluent_kafka
   ([#2616](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2616))
+- `opentelemetry-instrumentation-confluent-kafka` Add support for produce purge
+  ([#2638](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2638))
 
 ### Breaking changes
 
@@ -39,6 +45,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2590](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2590))
 - Reference symbols from generated semantic conventions
   ([#2611](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2611))
+- `opentelemetry-instrumentation-psycopg` Bugfix: Handle empty statement.
+  ([#2644](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2644))
+- `opentelemetry-instrumentation-confluent-kafka` Confluent Kafka: Ensure consume span is ended when consumer is closed
+  ([#2640](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2640))
 
 ## Version 1.25.0/0.46b0 (2024-05-31)
 
@@ -146,7 +156,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2136](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2136))
 - `opentelemetry-resource-detector-azure` Suppress instrumentation for `urllib` call
   ([#2178](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2178))
-- AwsLambdaInstrumentor handles and re-raises function exception ([#2245](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2245))
+- AwsLambdaInstrumentor handles and re-raises function exception
+  ([#2245](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2245))
 
 ### Added
 

CONTRIBUTING.md

@@ -60,13 +60,13 @@ $ pip install tox
 
 You can run `tox` with the following arguments:
 
-- `tox` to run all existing tox commands, including unit tests for all packages
+* `tox` to run all existing tox commands, including unit tests for all packages
   under multiple Python versions
-- `tox -e docs` to regenerate the API docs
-- `tox -e py312-test-instrumentation-aiopg` to e.g. run the aiopg instrumentation unit tests under a specific
+* `tox -e docs` to regenerate the API docs
+* `tox -e py312-test-instrumentation-aiopg` to e.g. run the aiopg instrumentation unit tests under a specific
   Python version
-- `tox -e spellcheck` to run a spellcheck on all the code
-- `tox -e lint-some-package` to run lint checks on `some-package`
+* `tox -e spellcheck` to run a spellcheck on all the code
+* `tox -e lint-some-package` to run lint checks on `some-package`
 
 `black` and `isort` are executed when `tox -e lint` is run. The reported errors can be tedious to fix manually.
 An easier way to do so is:
@@ -84,6 +84,7 @@ You can also configure it to run lint tools automatically before committing with
 
 ```console
 $ pre-commit install
+```
 
 See
 [`tox.ini`](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/tox.ini)
@@ -161,6 +162,7 @@ Open a pull request against the main `opentelemetry-python-contrib` repo.
 
 * If the PR is not ready for review, please put `[WIP]` in the title, tag it
   as `work-in-progress`, or mark it as [`draft`](https://github.blog/2019-02-14-introducing-draft-pull-requests/).
+* Make sure tests and lint are passing locally before requesting a review.
 * Make sure CLA is signed and CI is clear.
 
 ### How to Get PRs Reviewed
@@ -216,13 +218,26 @@ For a deeper discussion, see: https://github.com/open-telemetry/opentelemetry-sp
 2. Make sure you have `tox` installed. `pip install tox`.
 3. Run `tox` without any arguments to run tests for all the packages. Read more about [tox](https://tox.readthedocs.io/en/latest/).
 
+Some tests can be slow due to pre-steps that do dependencies installs. To help with that, you can run tox a first time, and after that run the tests using previous installed dependencies in toxdir as following:
+
+1. First time run (e.g., opentelemetry-instrumentation-aiopg)
+```console
+tox -e py312-test-instrumentation-aiopg
+```
+2. Run tests again without pre-steps:
+```console
+.tox/py312-test-instrumentation-aiopg/bin/pytest instrumentation/opentelemetry-instrumentation-aiopg
+```
+
 ### Testing against a different Core repo branch/commit
 
 Some of the tox targets install packages from the [OpenTelemetry Python Core Repository](https://github.com/open-telemetry/opentelemetry-python) via pip. The version of the packages installed defaults to the main branch in that repository when tox is run locally. It is possible to install packages tagged with a specific git commit hash by setting an environment variable before running tox as per the following example:
 
+```sh
 CORE_REPO_SHA=c49ad57bfe35cfc69bfa863d74058ca9bec55fc3 tox
+```
 
-The continuation integration overrides that environment variable with as per the configuration [here](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/2518a4ac07cb62ad6587dd8f6cbb5f8663a7e179/.github/workflows/test.yml#L9).
+The continuous integration overrides that environment variable with as per the configuration [here](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/.github/workflows/test.yml#L9).
 
 ## Style Guide
 
@@ -260,3 +275,24 @@ Below is a checklist of things to be mindful of when implementing a new instrume
 ## Expectations from contributors
 
 OpenTelemetry is an open source community, and as such, greatly encourages contributions from anyone interested in the project. With that being said, there is a certain level of expectation from contributors even after a pull request is merged, specifically pertaining to instrumentations. The OpenTelemetry Python community expects contributors to maintain a level of support and interest in the instrumentations they contribute. This is to ensure that the instrumentation does not become stale and still functions the way the original contributor intended. Some instrumentations also pertain to libraries that the current members of the community are not so familiar with, so it is necessary to rely on the expertise of the original contributing parties.
+
+## Updating supported Python versions
+
+### Bumping the Python baseline
+
+When updating the minimum supported Python version remember to:
+
+- Remove the version in `pyproject.toml` trove classifiers
+- Remove the version from `tox.ini`
+- Search for `sys.version_info` usage and remove code for unsupported versions
+- Bump `py-version` in `.pylintrc` for Python version dependent checks
+
+### Adding support for a new Python release
+
+When adding support for a new Python release remember to:
+
+- Add the version in `tox.ini`
+- Add the version in `pyproject.toml` trove classifiers
+- Update github workflows accordingly; lint and benchmarks use the latest supported version
+- Update `.pre-commit-config.yaml`
+- Update tox examples in the documentation

README.md

@@ -46,7 +46,6 @@ The Python auto-instrumentation libraries for [OpenTelemetry](https://openteleme
 * [Releasing](#releasing)
   * [Releasing a package as `1.0` stable](#releasing-a-package-as-10-stable)
 * [Contributing](#contributing)
-* [Running Tests Locally](#running-tests-locally)
 * [Thanks to all the people who already contributed](#thanks-to-all-the-people-who-already-contributed)
 
 ## Installation
@@ -143,14 +142,6 @@ Emeritus Maintainers:
 
 *Find more about the maintainer role in [community repository](https://github.com/open-telemetry/community/blob/main/community-membership.md#maintainer).*
 
-## Running Tests Locally
-
-1. Go to your Contrib repo directory. `cd ~/git/opentelemetry-python-contrib`.
-2. Create a virtual env in your Contrib repo directory. `python3 -m venv my_test_venv`.
-3. Activate your virtual env. `source my_test_venv/bin/activate`.
-4. Make sure you have `tox` installed. `pip install tox`.
-5. Run tests for a package. (e.g. `tox -e test-instrumentation-flask`.)
-
 ### Thanks to all the people who already contributed
 
 <a href="https://github.com/open-telemetry/opentelemetry-python-contrib/graphs/contributors">

instrumentation/opentelemetry-instrumentation-asyncio/src/opentelemetry/instrumentation/asyncio/__init__.py

@@ -280,8 +280,11 @@ async def trace_coroutine(self, coro):
         # CancelledError is raised when a coroutine is cancelled
         # before it has a chance to run. We don't want to record
         # this as an error.
+        # Still it needs to be raised in order for `asyncio.wait_for`
+        # to properly work with timeout and raise accordingly `asyncio.TimeoutError`
         except asyncio.CancelledError:
             attr["state"] = "cancelled"
+            raise
         except Exception as exc:
             exception = exc
             state = determine_state(exception)

instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_wait.py

@@ -68,6 +68,19 @@ async def main():
         spans = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans), 2)
 
+    def test_asyncio_wait_for_with_timeout(self):
+        expected_timeout_error = None
+
+        async def main():
+            nonlocal expected_timeout_error
+            try:
+                await asyncio.wait_for(async_func(), 0.01)
+            except asyncio.TimeoutError as timeout_error:
+                expected_timeout_error = timeout_error
+
+        asyncio.run(main())
+        self.assertNotEqual(expected_timeout_error, None)
+
     def test_asyncio_as_completed(self):
         async def main():
             if sys.version_info >= (3, 11):

instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py

@@ -144,6 +144,10 @@ def consume(
     ):  # pylint: disable=useless-super-delegation
         return super().consume(*args, **kwargs)
 
+    # This method is deliberately implemented in order to allow wrapt to wrap this function
+    def close(self):  # pylint: disable=useless-super-delegation
+        return super().close()
+
 
 class ProxiedProducer(Producer):
     def __init__(self, producer: Producer, tracer: Tracer):
@@ -156,6 +160,9 @@ def flush(self, timeout=-1):
     def poll(self, timeout=-1):
         return self._producer.poll(timeout)
 
+    def purge(self, in_queue=True, in_flight=True, blocking=True):
+        self._producer.purge(in_queue, in_flight, blocking)
+
     def produce(
         self, topic, value=None, *args, **kwargs
     ):  # pylint: disable=keyword-arg-before-vararg
@@ -178,6 +185,11 @@ def __init__(self, consumer: Consumer, tracer: Tracer):
         self._current_consume_span = None
         self._current_context_token = None
 
+    def close(self):
+        return ConfluentKafkaInstrumentor.wrap_close(
+            self._consumer.close, self
+        )
+
     def committed(self, partitions, timeout=-1):
         return self._consumer.committed(partitions, timeout)
 
@@ -300,6 +312,9 @@ def _inner_wrap_consume(func, instance, args, kwargs):
                 func, instance, self._tracer, args, kwargs
             )
 
+        def _inner_wrap_close(func, instance):
+            return ConfluentKafkaInstrumentor.wrap_close(func, instance)
+
         wrapt.wrap_function_wrapper(
             AutoInstrumentedProducer,
             "produce",
@@ -318,6 +333,12 @@ def _inner_wrap_consume(func, instance, args, kwargs):
             _inner_wrap_consume,
         )
 
+        wrapt.wrap_function_wrapper(
+            AutoInstrumentedConsumer,
+            "close",
+            _inner_wrap_close,
+        )
+
     def _uninstrument(self, **kwargs):
         confluent_kafka.Producer = self._original_kafka_producer
         confluent_kafka.Consumer = self._original_kafka_consumer
@@ -400,3 +421,9 @@ def wrap_consume(func, instance, tracer, args, kwargs):
         )
 
         return records
+
+    @staticmethod
+    def wrap_close(func, instance):
+        if instance._current_consume_span:
+            _end_current_consume_span(instance)
+        func()

instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py

@@ -237,7 +237,44 @@ def test_consume(self) -> None:
         span_list = self.memory_exporter.get_finished_spans()
         self._compare_spans(span_list, expected_spans)
 
+    def test_close(self) -> None:
+        instrumentation = ConfluentKafkaInstrumentor()
+        mocked_messages = [
+            MockedMessage("topic-a", 0, 0, []),
+        ]
+        expected_spans = [
+            {"name": "recv", "attributes": {}},
+            {
+                "name": "topic-a process",
+                "attributes": {
+                    SpanAttributes.MESSAGING_OPERATION: "process",
+                    SpanAttributes.MESSAGING_KAFKA_PARTITION: 0,
+                    SpanAttributes.MESSAGING_SYSTEM: "kafka",
+                    SpanAttributes.MESSAGING_DESTINATION: "topic-a",
+                    SpanAttributes.MESSAGING_DESTINATION_KIND: MessagingDestinationKindValues.QUEUE.value,
+                    SpanAttributes.MESSAGING_MESSAGE_ID: "topic-a.0.0",
+                },
+            },
+        ]
+
+        consumer = MockConsumer(
+            mocked_messages,
+            {
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            },
+        )
+        self.memory_exporter.clear()
+        consumer = instrumentation.instrument_consumer(consumer)
+        consumer.poll()
+        consumer.close()
+
+        span_list = self.memory_exporter.get_finished_spans()
+        self._compare_spans(span_list, expected_spans)
+
     def _compare_spans(self, spans, expected_spans):
+        self.assertEqual(len(spans), len(expected_spans))
         for span, expected_span in zip(spans, expected_spans):
             self.assertEqual(expected_span["name"], span.name)
             for attribute_key, expected_attribute_value in expected_span[

instrumentation/opentelemetry-instrumentation-django/src/opentelemetry/instrumentation/django/middleware/otel_middleware.py

@@ -187,6 +187,7 @@ def _get_span_name(request):
             return request.method
 
     # pylint: disable=too-many-locals
+    # pylint: disable=too-many-branches
     def process_request(self, request):
         # request.META is a dictionary containing all available HTTP headers
         # Read more about request.META here:
@@ -286,9 +287,14 @@ def process_request(self, request):
             request.META[self._environ_token] = token
 
         if _DjangoMiddleware._otel_request_hook:
-            _DjangoMiddleware._otel_request_hook(  # pylint: disable=not-callable
-                span, request
-            )
+            try:
+                _DjangoMiddleware._otel_request_hook(  # pylint: disable=not-callable
+                    span, request
+                )
+            except Exception:  # pylint: disable=broad-exception-caught
+                # Raising an exception here would leak the request span since process_response
+                # would not be called. Log the exception instead.
+                _logger.exception("Exception raised by request_hook")
 
     # pylint: disable=unused-argument
     def process_view(self, request, view_func, *args, **kwargs):
@@ -385,10 +391,14 @@ def process_response(self, request, response):
 
             # record any exceptions raised while processing the request
             exception = request.META.pop(self._environ_exception_key, None)
+
             if _DjangoMiddleware._otel_response_hook:
-                _DjangoMiddleware._otel_response_hook(  # pylint: disable=not-callable
-                    span, request, response
-                )
+                try:
+                    _DjangoMiddleware._otel_response_hook(  # pylint: disable=not-callable
+                        span, request, response
+                    )
+                except Exception:  # pylint: disable=broad-exception-caught
+                    _logger.exception("Exception raised by response_hook")
 
             if exception:
                 activation.__exit__(

instrumentation/opentelemetry-instrumentation-django/tests/test_middleware.py

@@ -392,6 +392,32 @@ def response_hook(span, request, response):
         self.assertIsInstance(response_hook_args[2], HttpResponse)
         self.assertEqual(response_hook_args[2], response)
 
+    def test_request_hook_exception(self):
+        def request_hook(span, request):
+            # pylint: disable=broad-exception-raised
+            raise Exception("request hook exception")
+
+        _DjangoMiddleware._otel_request_hook = request_hook
+        Client().get("/span_name/1234/")
+        _DjangoMiddleware._otel_request_hook = None
+
+        # ensure that span ended
+        finished_spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(finished_spans), 1)
+
+    def test_response_hook_exception(self):
+        def response_hook(span, request, response):
+            # pylint: disable=broad-exception-raised
+            raise Exception("response hook exception")
+
+        _DjangoMiddleware._otel_response_hook = response_hook
+        Client().get("/span_name/1234/")
+        _DjangoMiddleware._otel_response_hook = None
+
+        # ensure that span ended
+        finished_spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(finished_spans), 1)
+
     def test_trace_parent(self):
         id_generator = RandomIdGenerator()
         trace_id = format_trace_id(id_generator.generate_trace_id())

instrumentation/opentelemetry-instrumentation-flask/tests/test_programmatic.py

@@ -497,7 +497,7 @@ def test_flask_metrics_new_semconv(self):
         self.client.get("/hello/123")
         self.client.get("/hello/321")
         self.client.get("/hello/756")
-        duration = max(round((default_timer() - start) * 1000), 0)
+        duration_s = max(default_timer() - start, 0)
         metrics_list = self.memory_metrics_reader.get_metrics_data()
         number_data_point_seen = False
         histogram_data_point_seen = False
@@ -514,7 +514,7 @@ def test_flask_metrics_new_semconv(self):
                         if isinstance(point, HistogramDataPoint):
                             self.assertEqual(point.count, 3)
                             self.assertAlmostEqual(
-                                duration, point.sum, delta=10
+                                duration_s, point.sum, places=2
                             )
                             histogram_data_point_seen = True
                         if isinstance(point, NumberDataPoint):