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

CHANGELOG.md

@@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4397](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/4397))
 - Switch to SPDX license headers and add CI enforcement
   ([#4533](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4533))
+- `opentelemetry-instrumentation-asyncpg`: Add `exclude_queries` option to `AsyncPGInstrumentor`
+  to silence specific queries from being traced
+  ([#4373](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4373))
 - Bump `pylint` to `4.0.5`
   ([#4244](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4244))
 - `opentelemetry-instrumentation-sqlite3`: Add uninstrument, error status, suppress, and no-op tests

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

@@ -25,6 +25,9 @@
     # You can optionally pass a custom TracerProvider to AsyncPGInstrumentor.instrument()
     AsyncPGInstrumentor().instrument()
 
+    # OR opt into not tracing connection pool cleanup queries
+    AsyncPGInstrumentor(exclude_queries=AsyncPGInstrumentor._CLEANUP_QUERIES).instrument()
+
     async def main():
         conn = await asyncpg.connect(user='user', password='password')
 
@@ -103,11 +106,25 @@ 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_excluded_query(self, query: str) -> bool:
+        if not query:
+            return False
+        return any(q in query for q in self.exclude_queries)
+
+    def __init__(self, capture_parameters=False, exclude_queries=None):
         super().__init__()
         self.capture_parameters = capture_parameters
+        self.exclude_queries = exclude_queries or set()
 
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
@@ -173,6 +190,8 @@ async def _do_execute(self, func, instance, args, kwargs):
             args[0],
             args[1:] if self.capture_parameters else None,
         )
+        if self._is_excluded_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

@@ -147,3 +147,84 @@ 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 exclude_queries=AsyncPGInstrumentor._CLEANUP_QUERIES."""
+        AsyncPGInstrumentor().uninstrument()
+        apg = AsyncPGInstrumentor(
+            exclude_queries=AsyncPGInstrumentor._CLEANUP_QUERIES
+        )
+        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 exclude_queries=set()."""
+        AsyncPGInstrumentor().uninstrument()
+        apg = AsyncPGInstrumentor(exclude_queries=set())
+        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()