feat(asyncpg): add capture_connection_cleanup option to silence pool cleanup spans

CHANGELOG.md

@@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 > Use [this search for a list of all CHANGELOG.md files in this repo](https://github.com/search?q=repo%3Aopen-telemetry%2Fopentelemetry-python-contrib+path%3A**%2FCHANGELOG.md&type=code).
 
 ## Unreleased
-  
+
 ### Added
 
 - `opentelemetry-instrumentation-confluent-kafka`: Loosen confluent-kafka upper bound to <3.0.0
@@ -21,6 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4212](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4212))
 - `opentelemetry-instrumentation-botocore`: Add support for instrumenting `aiobotocore`
   ([#4049](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4049))
+  - `opentelemetry-instrumentation-asyncpg`: Add `capture_connection_cleanup` option to `AsyncPGInstrumentor` to silence connection pool cleanup spans
+  ([#4373](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4373))
 
 ### Fixed
 

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

@@ -35,6 +35,7 @@
 
     # You can optionally pass a custom TracerProvider to AsyncPGInstrumentor.instrument()
     AsyncPGInstrumentor().instrument()
+    AsyncPGInstrumentor(capture_connection_cleanup=False).instrument()
 
     async def main():
         conn = await asyncpg.connect(user='user', password='password')
@@ -114,11 +115,27 @@ def _hydrate_span_from_args(connection, query, parameters) -> dict:
 
 class AsyncPGInstrumentor(BaseInstrumentor):
     _leading_comment_remover = re.compile(r"^/\*.*?\*/")
+    _CLEANUP_QUERIES = frozenset(
+        [
+            "SELECT pg_advisory_unlock_all()",
+            "CLOSE ALL",
+            "UNLISTEN *",
+            "RESET ALL",
+        ]
+    )
     _tracer = None
 
-    def __init__(self, capture_parameters=False):
+    def _is_cleanup_query(self, query: str) -> bool:
+        if query is None:
+            return False
+        return any(q in query for q in self._CLEANUP_QUERIES)
+
+    def __init__(
+        self, capture_parameters=False, capture_connection_cleanup=True
+    ):
         super().__init__()
         self.capture_parameters = capture_parameters
+        self.capture_connection_cleanup = capture_connection_cleanup
 
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
@@ -184,6 +201,10 @@ async def _do_execute(self, func, instance, args, kwargs):
             args[0],
             args[1:] if self.capture_parameters else None,
         )
+        if not self.capture_connection_cleanup and self._is_cleanup_query(
+            args[0]
+        ):
+            return await func(*args, **kwargs)
 
         with self._tracer.start_as_current_span(
             name, kind=SpanKind.CLIENT, attributes=span_attributes

instrumentation/opentelemetry-instrumentation-asyncpg/tests/test_asyncpg_wrapper.py

@@ -144,3 +144,82 @@ async def exec_mock(*args, **kwargs):
 
         spans = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans), 0)
+
+    def test_capture_connection_cleanup_false(self):
+        """Test that cleanup queries are not traced when capture_connection_cleanup=False."""
+        AsyncPGInstrumentor().uninstrument()
+        apg = AsyncPGInstrumentor(capture_connection_cleanup=False)
+        apg.instrument(tracer_provider=self.tracer_provider)
+
+        async def mock_execute(*args, **kwargs):
+            return None
+
+        conn = mock.Mock()
+        conn._params = mock.Mock()
+        conn._params.database = "testdb"
+        conn._params.user = "testuser"
+        conn._addr = ("localhost", 5432)
+
+        for cleanup_query in [
+            "SELECT pg_advisory_unlock_all()",
+            "CLOSE ALL",
+            "UNLISTEN *",
+            "RESET ALL",
+        ]:
+            asyncio.run(
+                apg._do_execute(mock_execute, conn, (cleanup_query,), {})
+            )
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 0)
+        AsyncPGInstrumentor().uninstrument()
+
+    def test_capture_connection_cleanup_true_by_default(self):
+        """Test that cleanup queries are traced by default."""
+        AsyncPGInstrumentor().uninstrument()
+        apg = AsyncPGInstrumentor()
+        apg.instrument(tracer_provider=self.tracer_provider)
+
+        async def mock_execute(*args, **kwargs):
+            return None
+
+        conn = mock.Mock()
+        conn._params = mock.Mock()
+        conn._params.database = "testdb"
+        conn._params.user = "testuser"
+        conn._addr = ("localhost", 5432)
+
+        asyncio.run(
+            apg._do_execute(
+                mock_execute, conn, ("SELECT pg_advisory_unlock_all()",), {}
+            )
+        )
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        AsyncPGInstrumentor().uninstrument()
+
+    def test_capture_connection_cleanup_explicitly_true(self):
+        """Test that cleanup queries are traced when capture_connection_cleanup=True."""
+        AsyncPGInstrumentor().uninstrument()
+        apg = AsyncPGInstrumentor(capture_connection_cleanup=True)
+        apg.instrument(tracer_provider=self.tracer_provider)
+
+        async def mock_execute(*args, **kwargs):
+            return None
+
+        conn = mock.Mock()
+        conn._params = mock.Mock()
+        conn._params.database = "testdb"
+        conn._params.user = "testuser"
+        conn._addr = ("localhost", 5432)
+
+        asyncio.run(
+            apg._do_execute(
+                mock_execute, conn, ("SELECT pg_advisory_unlock_all()",), {}
+            )
+        )
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        AsyncPGInstrumentor().uninstrument()