We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Expand sqlalchemy pool.name to follow the semantic conventions
Spec ref: https://github.com/open-telemetry/opentelemetry-specification/pull/3050

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.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 import os
15 import re
16 
17 from sqlalchemy.event import (  # pylint: disable=no-name-in-module
18     listen,
19     remove,
20 )
21 
22 from opentelemetry import trace
23 from opentelemetry.instrumentation.sqlalchemy.version import __version__
24 from opentelemetry.instrumentation.sqlcommenter_utils import _add_sql_comment
25 from opentelemetry.instrumentation.utils import _get_opentelemetry_values
26 from opentelemetry.semconv.trace import NetTransportValues, SpanAttributes
27 from opentelemetry.trace.status import Status, StatusCode
28 
29 
30 def _normalize_vendor(vendor):
31     """Return a canonical name for a type of database."""
32     if not vendor:
33         return "db"  # should this ever happen?
34 
35     if "sqlite" in vendor:
36         return "sqlite"
37 
38     if "postgres" in vendor or vendor == "psycopg2":
39         return "postgresql"
40 
41     return vendor
42 
43 
44 def _wrap_create_async_engine(
45     tracer, connections_usage, enable_commenter=False
46 ):
47     # pylint: disable=unused-argument
48     def _wrap_create_async_engine_internal(func, module, args, kwargs):
49         """Trace the SQLAlchemy engine, creating an `EngineTracer`
50         object that will listen to SQLAlchemy events.
51         """
52         engine = func(*args, **kwargs)
53         EngineTracer(
54             tracer, engine.sync_engine, connections_usage, enable_commenter
55         )
56         return engine
57 
58     return _wrap_create_async_engine_internal
59 
60 
61 def _wrap_create_engine(tracer, connections_usage, enable_commenter=False):
62     def _wrap_create_engine_internal(func, _module, args, kwargs):
63         """Trace the SQLAlchemy engine, creating an `EngineTracer`
64         object that will listen to SQLAlchemy events.
65         """
66         engine = func(*args, **kwargs)
67         EngineTracer(tracer, engine, connections_usage, enable_commenter)
68         return engine
69 
70     return _wrap_create_engine_internal
71 
72 
73 def _wrap_connect(tracer):
74     # pylint: disable=unused-argument
75     def _wrap_connect_internal(func, module, args, kwargs):
76         with tracer.start_as_current_span(
77             "connect", kind=trace.SpanKind.CLIENT
78         ) as span:
79             if span.is_recording():
80                 attrs, _ = _get_attributes_from_url(module.url)
81                 span.set_attributes(attrs)
82                 span.set_attribute(
83                     SpanAttributes.DB_SYSTEM, _normalize_vendor(module.name)
84                 )
85             return func(*args, **kwargs)
86 
87     return _wrap_connect_internal
88 
89 
90 class EngineTracer:
91     _remove_event_listener_params = []
92 
93     def __init__(
94         self,
95         tracer,
96         engine,
97         connections_usage,
98         enable_commenter=False,
99         commenter_options=None,
100     ):
101         self.tracer = tracer
102         self.engine = engine
103         self.connections_usage = connections_usage
104         self.vendor = _normalize_vendor(engine.name)
105         self.enable_commenter = enable_commenter
106         self.commenter_options = commenter_options if commenter_options else {}
107         self._leading_comment_remover = re.compile(r"^/\*.*?\*/")
108 
109         self._register_event_listener(
110             engine, "before_cursor_execute", self._before_cur_exec, retval=True
111         )
112         self._register_event_listener(
113             engine, "after_cursor_execute", _after_cur_exec
114         )
115         self._register_event_listener(engine, "handle_error", _handle_error)
116         self._register_event_listener(engine, "connect", self._pool_connect)
117         self._register_event_listener(engine, "close", self._pool_close)
118         self._register_event_listener(engine, "checkin", self._pool_checkin)
119         self._register_event_listener(engine, "checkout", self._pool_checkout)
120 
121     def _get_pool_name(self):
122         return self.engine.pool.logging_name or ""
123 
124     def _add_idle_to_connection_usage(self, value):
125         self.connections_usage.add(
126             value,
127             attributes={
128                 "pool.name": self._get_pool_name(),
129                 "state": "idle",
130             },
131         )
132 
133     def _add_used_to_connection_usage(self, value):
134         self.connections_usage.add(
135             value,
136             attributes={
137                 "pool.name": self._get_pool_name(),
138                 "state": "used",
139             },
140         )
141 
142     def _pool_connect(self, _dbapi_connection, _connection_record):
143         self._add_idle_to_connection_usage(1)
144 
145     def _pool_close(self, _dbapi_connection, _connection_record):
146         self._add_idle_to_connection_usage(-1)
147 
148     # Called when a connection returns to the pool.
149     def _pool_checkin(self, _dbapi_connection, _connection_record):
150         self._add_used_to_connection_usage(-1)
151         self._add_idle_to_connection_usage(1)
152 
153     # Called when a connection is retrieved from the Pool.
154     def _pool_checkout(
155         self, _dbapi_connection, _connection_record, _connection_proxy
156     ):
157         self._add_idle_to_connection_usage(-1)
158         self._add_used_to_connection_usage(1)
159 
160     @classmethod
161     def _register_event_listener(cls, target, identifier, func, *args, **kw):
162         listen(target, identifier, func, *args, **kw)
163         cls._remove_event_listener_params.append((target, identifier, func))
164 
165     @classmethod
166     def remove_all_event_listeners(cls):
167         for remove_params in cls._remove_event_listener_params:
168             remove(*remove_params)
169         cls._remove_event_listener_params.clear()
170 
171     def _operation_name(self, db_name, statement):
172         parts = []
173         if isinstance(statement, str):
174             # otel spec recommends against parsing SQL queries. We are not trying to parse SQL
175             # but simply truncating the statement to the first word. This covers probably >95%
176             # use cases and uses the SQL statement in span name correctly as per the spec.
177             # For some very special cases it might not record the correct statement if the SQL
178             # dialect is too weird but in any case it shouldn't break anything.
179             # Strip leading comments so we get the operation name.
180             parts.append(
181                 self._leading_comment_remover.sub("", statement).split()[0]
182             )
183         if db_name:
184             parts.append(db_name)
185         if not parts:
186             return self.vendor
187         return " ".join(parts)
188 
189     def _before_cur_exec(
190         self, conn, cursor, statement, params, context, _executemany
191     ):
192         attrs, found = _get_attributes_from_url(conn.engine.url)
193         if not found:
194             attrs = _get_attributes_from_cursor(self.vendor, cursor, attrs)
195 
196         db_name = attrs.get(SpanAttributes.DB_NAME, "")
197         span = self.tracer.start_span(
198             self._operation_name(db_name, statement),
199             kind=trace.SpanKind.CLIENT,
200         )
201         with trace.use_span(span, end_on_exit=False):
202             if span.is_recording():
203                 span.set_attribute(SpanAttributes.DB_STATEMENT, statement)
204                 span.set_attribute(SpanAttributes.DB_SYSTEM, self.vendor)
205                 for key, value in attrs.items():
206                     span.set_attribute(key, value)
207             if self.enable_commenter:
208                 commenter_data = dict(
209                     db_driver=conn.engine.driver,
210                     # Driver/framework centric information.
211                     db_framework=f"sqlalchemy:{__version__}",
212                 )
213 
214                 if self.commenter_options.get("opentelemetry_values", True):
215                     commenter_data.update(**_get_opentelemetry_values())
216 
217                 # Filter down to just the requested attributes.
218                 commenter_data = {
219                     k: v
220                     for k, v in commenter_data.items()
221                     if self.commenter_options.get(k, True)
222                 }
223 
224                 statement = _add_sql_comment(statement, **commenter_data)
225 
226         context._otel_span = span
227 
228         return statement, params
229 
230 
231 # pylint: disable=unused-argument
232 def _after_cur_exec(conn, cursor, statement, params, context, executemany):
233     span = getattr(context, "_otel_span", None)
234     if span is None:
235         return
236 
237     span.end()
238 
239 
240 def _handle_error(context):
241     span = getattr(context.execution_context, "_otel_span", None)
242     if span is None:
243         return
244 
245     if span.is_recording():
246         span.set_status(
247             Status(
248                 StatusCode.ERROR,
249                 str(context.original_exception),
250             )
251         )
252     span.end()
253 
254 
255 def _get_attributes_from_url(url):
256     """Set connection tags from the url. return true if successful."""
257     attrs = {}
258     if url.host:
259         attrs[SpanAttributes.NET_PEER_NAME] = url.host
260     if url.port:
261         attrs[SpanAttributes.NET_PEER_PORT] = url.port
262     if url.database:
263         attrs[SpanAttributes.DB_NAME] = url.database
264     if url.username:
265         attrs[SpanAttributes.DB_USER] = url.username
266     return attrs, bool(url.host)
267 
268 
269 def _get_attributes_from_cursor(vendor, cursor, attrs):
270     """Attempt to set db connection attributes by introspecting the cursor."""
271     if vendor == "postgresql":
272         info = getattr(getattr(cursor, "connection", None), "info", None)
273         if not info:
274             return attrs
275 
276         attrs[SpanAttributes.DB_NAME] = info.dbname
277         is_unix_socket = info.host and info.host.startswith("/")
278 
279         if is_unix_socket:
280             attrs[SpanAttributes.NET_TRANSPORT] = NetTransportValues.UNIX.value
281             if info.port:
282                 # postgresql enforces this pattern on all socket names
283                 attrs[SpanAttributes.NET_PEER_NAME] = os.path.join(
284                     info.host, f".s.PGSQL.{info.port}"
285                 )
286         else:
287             attrs[
288                 SpanAttributes.NET_TRANSPORT
289             ] = NetTransportValues.IP_TCP.value
290             attrs[SpanAttributes.NET_PEER_NAME] = info.host
291             if info.port:
292                 attrs[SpanAttributes.NET_PEER_PORT] = int(info.port)
293     return attrs
294
```
--- 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/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py
--- a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py
+++ b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py
@@ -118,8 +118,17 @@
         self._register_event_listener(engine, "checkin", self._pool_checkin)
         self._register_event_listener(engine, "checkout", self._pool_checkout)
 
+    def _get_connection_string(self):
+        drivername = self.engine.url.drivername or ""
+        host = self.engine.url.host or ""
+        port = self.engine.url.port or ""
+        database = self.engine.url.database or ""
+        return f"{drivername}://{host}:{port}/{database}"
+
     def _get_pool_name(self):
-        return self.engine.pool.logging_name or ""
+        if self.engine.pool.logging_name is not None:
+            return self.engine.pool.logging_name
+        return self._get_connection_string()
 
     def _add_idle_to_connection_usage(self, value):
         self.connections_usage.add(

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py\n--- a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py\n+++ b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py\n@@ -118,8 +118,17 @@\n         self._register_event_listener(engine, \"checkin\", self._pool_checkin)\n         self._register_event_listener(engine, \"checkout\", self._pool_checkout)\n \n+    def _get_connection_string(self):\n+        drivername = self.engine.url.drivername or \"\"\n+        host = self.engine.url.host or \"\"\n+        port = self.engine.url.port or \"\"\n+        database = self.engine.url.database or \"\"\n+        return f\"{drivername}://{host}:{port}/{database}\"\n+\n     def _get_pool_name(self):\n-        return self.engine.pool.logging_name or \"\"\n+        if self.engine.pool.logging_name is not None:\n+            return self.engine.pool.logging_name\n+        return self._get_connection_string()\n \n     def _add_idle_to_connection_usage(self, value):\n         self.connections_usage.add(\n", "issue": "Expand sqlalchemy pool.name to follow the semantic conventions\nSpec ref: https://github.com/open-telemetry/opentelemetry-specification/pull/3050\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.\nimport os\nimport re\n\nfrom sqlalchemy.event import (  # pylint: disable=no-name-in-module\n    listen,\n    remove,\n)\n\nfrom opentelemetry import trace\nfrom opentelemetry.instrumentation.sqlalchemy.version import __version__\nfrom opentelemetry.instrumentation.sqlcommenter_utils import _add_sql_comment\nfrom opentelemetry.instrumentation.utils import _get_opentelemetry_values\nfrom opentelemetry.semconv.trace import NetTransportValues, SpanAttributes\nfrom opentelemetry.trace.status import Status, StatusCode\n\n\ndef _normalize_vendor(vendor):\n    \"\"\"Return a canonical name for a type of database.\"\"\"\n    if not vendor:\n        return \"db\"  # should this ever happen?\n\n    if \"sqlite\" in vendor:\n        return \"sqlite\"\n\n    if \"postgres\" in vendor or vendor == \"psycopg2\":\n        return \"postgresql\"\n\n    return vendor\n\n\ndef _wrap_create_async_engine(\n    tracer, connections_usage, enable_commenter=False\n):\n    # pylint: disable=unused-argument\n    def _wrap_create_async_engine_internal(func, module, args, kwargs):\n        \"\"\"Trace the SQLAlchemy engine, creating an `EngineTracer`\n        object that will listen to SQLAlchemy events.\n        \"\"\"\n        engine = func(*args, **kwargs)\n        EngineTracer(\n            tracer, engine.sync_engine, connections_usage, enable_commenter\n        )\n        return engine\n\n    return _wrap_create_async_engine_internal\n\n\ndef _wrap_create_engine(tracer, connections_usage, enable_commenter=False):\n    def _wrap_create_engine_internal(func, _module, args, kwargs):\n        \"\"\"Trace the SQLAlchemy engine, creating an `EngineTracer`\n        object that will listen to SQLAlchemy events.\n        \"\"\"\n        engine = func(*args, **kwargs)\n        EngineTracer(tracer, engine, connections_usage, enable_commenter)\n        return engine\n\n    return _wrap_create_engine_internal\n\n\ndef _wrap_connect(tracer):\n    # pylint: disable=unused-argument\n    def _wrap_connect_internal(func, module, args, kwargs):\n        with tracer.start_as_current_span(\n            \"connect\", kind=trace.SpanKind.CLIENT\n        ) as span:\n            if span.is_recording():\n                attrs, _ = _get_attributes_from_url(module.url)\n                span.set_attributes(attrs)\n                span.set_attribute(\n                    SpanAttributes.DB_SYSTEM, _normalize_vendor(module.name)\n                )\n            return func(*args, **kwargs)\n\n    return _wrap_connect_internal\n\n\nclass EngineTracer:\n    _remove_event_listener_params = []\n\n    def __init__(\n        self,\n        tracer,\n        engine,\n        connections_usage,\n        enable_commenter=False,\n        commenter_options=None,\n    ):\n        self.tracer = tracer\n        self.engine = engine\n        self.connections_usage = connections_usage\n        self.vendor = _normalize_vendor(engine.name)\n        self.enable_commenter = enable_commenter\n        self.commenter_options = commenter_options if commenter_options else {}\n        self._leading_comment_remover = re.compile(r\"^/\\*.*?\\*/\")\n\n        self._register_event_listener(\n            engine, \"before_cursor_execute\", self._before_cur_exec, retval=True\n        )\n        self._register_event_listener(\n            engine, \"after_cursor_execute\", _after_cur_exec\n        )\n        self._register_event_listener(engine, \"handle_error\", _handle_error)\n        self._register_event_listener(engine, \"connect\", self._pool_connect)\n        self._register_event_listener(engine, \"close\", self._pool_close)\n        self._register_event_listener(engine, \"checkin\", self._pool_checkin)\n        self._register_event_listener(engine, \"checkout\", self._pool_checkout)\n\n    def _get_pool_name(self):\n        return self.engine.pool.logging_name or \"\"\n\n    def _add_idle_to_connection_usage(self, value):\n        self.connections_usage.add(\n            value,\n            attributes={\n                \"pool.name\": self._get_pool_name(),\n                \"state\": \"idle\",\n            },\n        )\n\n    def _add_used_to_connection_usage(self, value):\n        self.connections_usage.add(\n            value,\n            attributes={\n                \"pool.name\": self._get_pool_name(),\n                \"state\": \"used\",\n            },\n        )\n\n    def _pool_connect(self, _dbapi_connection, _connection_record):\n        self._add_idle_to_connection_usage(1)\n\n    def _pool_close(self, _dbapi_connection, _connection_record):\n        self._add_idle_to_connection_usage(-1)\n\n    # Called when a connection returns to the pool.\n    def _pool_checkin(self, _dbapi_connection, _connection_record):\n        self._add_used_to_connection_usage(-1)\n        self._add_idle_to_connection_usage(1)\n\n    # Called when a connection is retrieved from the Pool.\n    def _pool_checkout(\n        self, _dbapi_connection, _connection_record, _connection_proxy\n    ):\n        self._add_idle_to_connection_usage(-1)\n        self._add_used_to_connection_usage(1)\n\n    @classmethod\n    def _register_event_listener(cls, target, identifier, func, *args, **kw):\n        listen(target, identifier, func, *args, **kw)\n        cls._remove_event_listener_params.append((target, identifier, func))\n\n    @classmethod\n    def remove_all_event_listeners(cls):\n        for remove_params in cls._remove_event_listener_params:\n            remove(*remove_params)\n        cls._remove_event_listener_params.clear()\n\n    def _operation_name(self, db_name, statement):\n        parts = []\n        if isinstance(statement, str):\n            # otel spec recommends against parsing SQL queries. We are not trying to parse SQL\n            # but simply truncating the statement to the first word. This covers probably >95%\n            # use cases and uses the SQL statement in span name correctly as per the spec.\n            # For some very special cases it might not record the correct statement if the SQL\n            # dialect is too weird but in any case it shouldn't break anything.\n            # Strip leading comments so we get the operation name.\n            parts.append(\n                self._leading_comment_remover.sub(\"\", statement).split()[0]\n            )\n        if db_name:\n            parts.append(db_name)\n        if not parts:\n            return self.vendor\n        return \" \".join(parts)\n\n    def _before_cur_exec(\n        self, conn, cursor, statement, params, context, _executemany\n    ):\n        attrs, found = _get_attributes_from_url(conn.engine.url)\n        if not found:\n            attrs = _get_attributes_from_cursor(self.vendor, cursor, attrs)\n\n        db_name = attrs.get(SpanAttributes.DB_NAME, \"\")\n        span = self.tracer.start_span(\n            self._operation_name(db_name, statement),\n            kind=trace.SpanKind.CLIENT,\n        )\n        with trace.use_span(span, end_on_exit=False):\n            if span.is_recording():\n                span.set_attribute(SpanAttributes.DB_STATEMENT, statement)\n                span.set_attribute(SpanAttributes.DB_SYSTEM, self.vendor)\n                for key, value in attrs.items():\n                    span.set_attribute(key, value)\n            if self.enable_commenter:\n                commenter_data = dict(\n                    db_driver=conn.engine.driver,\n                    # Driver/framework centric information.\n                    db_framework=f\"sqlalchemy:{__version__}\",\n                )\n\n                if self.commenter_options.get(\"opentelemetry_values\", True):\n                    commenter_data.update(**_get_opentelemetry_values())\n\n                # Filter down to just the requested attributes.\n                commenter_data = {\n                    k: v\n                    for k, v in commenter_data.items()\n                    if self.commenter_options.get(k, True)\n                }\n\n                statement = _add_sql_comment(statement, **commenter_data)\n\n        context._otel_span = span\n\n        return statement, params\n\n\n# pylint: disable=unused-argument\ndef _after_cur_exec(conn, cursor, statement, params, context, executemany):\n    span = getattr(context, \"_otel_span\", None)\n    if span is None:\n        return\n\n    span.end()\n\n\ndef _handle_error(context):\n    span = getattr(context.execution_context, \"_otel_span\", None)\n    if span is None:\n        return\n\n    if span.is_recording():\n        span.set_status(\n            Status(\n                StatusCode.ERROR,\n                str(context.original_exception),\n            )\n        )\n    span.end()\n\n\ndef _get_attributes_from_url(url):\n    \"\"\"Set connection tags from the url. return true if successful.\"\"\"\n    attrs = {}\n    if url.host:\n        attrs[SpanAttributes.NET_PEER_NAME] = url.host\n    if url.port:\n        attrs[SpanAttributes.NET_PEER_PORT] = url.port\n    if url.database:\n        attrs[SpanAttributes.DB_NAME] = url.database\n    if url.username:\n        attrs[SpanAttributes.DB_USER] = url.username\n    return attrs, bool(url.host)\n\n\ndef _get_attributes_from_cursor(vendor, cursor, attrs):\n    \"\"\"Attempt to set db connection attributes by introspecting the cursor.\"\"\"\n    if vendor == \"postgresql\":\n        info = getattr(getattr(cursor, \"connection\", None), \"info\", None)\n        if not info:\n            return attrs\n\n        attrs[SpanAttributes.DB_NAME] = info.dbname\n        is_unix_socket = info.host and info.host.startswith(\"/\")\n\n        if is_unix_socket:\n            attrs[SpanAttributes.NET_TRANSPORT] = NetTransportValues.UNIX.value\n            if info.port:\n                # postgresql enforces this pattern on all socket names\n                attrs[SpanAttributes.NET_PEER_NAME] = os.path.join(\n                    info.host, f\".s.PGSQL.{info.port}\"\n                )\n        else:\n            attrs[\n                SpanAttributes.NET_TRANSPORT\n            ] = NetTransportValues.IP_TCP.value\n            attrs[SpanAttributes.NET_PEER_NAME] = info.host\n            if info.port:\n                attrs[SpanAttributes.NET_PEER_PORT] = int(info.port)\n    return attrs\n", "path": "instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.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.\nimport os\nimport re\n\nfrom sqlalchemy.event import (  # pylint: disable=no-name-in-module\n    listen,\n    remove,\n)\n\nfrom opentelemetry import trace\nfrom opentelemetry.instrumentation.sqlalchemy.version import __version__\nfrom opentelemetry.instrumentation.sqlcommenter_utils import _add_sql_comment\nfrom opentelemetry.instrumentation.utils import _get_opentelemetry_values\nfrom opentelemetry.semconv.trace import NetTransportValues, SpanAttributes\nfrom opentelemetry.trace.status import Status, StatusCode\n\n\ndef _normalize_vendor(vendor):\n    \"\"\"Return a canonical name for a type of database.\"\"\"\n    if not vendor:\n        return \"db\"  # should this ever happen?\n\n    if \"sqlite\" in vendor:\n        return \"sqlite\"\n\n    if \"postgres\" in vendor or vendor == \"psycopg2\":\n        return \"postgresql\"\n\n    return vendor\n\n\ndef _wrap_create_async_engine(\n    tracer, connections_usage, enable_commenter=False\n):\n    # pylint: disable=unused-argument\n    def _wrap_create_async_engine_internal(func, module, args, kwargs):\n        \"\"\"Trace the SQLAlchemy engine, creating an `EngineTracer`\n        object that will listen to SQLAlchemy events.\n        \"\"\"\n        engine = func(*args, **kwargs)\n        EngineTracer(\n            tracer, engine.sync_engine, connections_usage, enable_commenter\n        )\n        return engine\n\n    return _wrap_create_async_engine_internal\n\n\ndef _wrap_create_engine(tracer, connections_usage, enable_commenter=False):\n    def _wrap_create_engine_internal(func, _module, args, kwargs):\n        \"\"\"Trace the SQLAlchemy engine, creating an `EngineTracer`\n        object that will listen to SQLAlchemy events.\n        \"\"\"\n        engine = func(*args, **kwargs)\n        EngineTracer(tracer, engine, connections_usage, enable_commenter)\n        return engine\n\n    return _wrap_create_engine_internal\n\n\ndef _wrap_connect(tracer):\n    # pylint: disable=unused-argument\n    def _wrap_connect_internal(func, module, args, kwargs):\n        with tracer.start_as_current_span(\n            \"connect\", kind=trace.SpanKind.CLIENT\n        ) as span:\n            if span.is_recording():\n                attrs, _ = _get_attributes_from_url(module.url)\n                span.set_attributes(attrs)\n                span.set_attribute(\n                    SpanAttributes.DB_SYSTEM, _normalize_vendor(module.name)\n                )\n            return func(*args, **kwargs)\n\n    return _wrap_connect_internal\n\n\nclass EngineTracer:\n    _remove_event_listener_params = []\n\n    def __init__(\n        self,\n        tracer,\n        engine,\n        connections_usage,\n        enable_commenter=False,\n        commenter_options=None,\n    ):\n        self.tracer = tracer\n        self.engine = engine\n        self.connections_usage = connections_usage\n        self.vendor = _normalize_vendor(engine.name)\n        self.enable_commenter = enable_commenter\n        self.commenter_options = commenter_options if commenter_options else {}\n        self._leading_comment_remover = re.compile(r\"^/\\*.*?\\*/\")\n\n        self._register_event_listener(\n            engine, \"before_cursor_execute\", self._before_cur_exec, retval=True\n        )\n        self._register_event_listener(\n            engine, \"after_cursor_execute\", _after_cur_exec\n        )\n        self._register_event_listener(engine, \"handle_error\", _handle_error)\n        self._register_event_listener(engine, \"connect\", self._pool_connect)\n        self._register_event_listener(engine, \"close\", self._pool_close)\n        self._register_event_listener(engine, \"checkin\", self._pool_checkin)\n        self._register_event_listener(engine, \"checkout\", self._pool_checkout)\n\n    def _get_connection_string(self):\n        drivername = self.engine.url.drivername or \"\"\n        host = self.engine.url.host or \"\"\n        port = self.engine.url.port or \"\"\n        database = self.engine.url.database or \"\"\n        return f\"{drivername}://{host}:{port}/{database}\"\n\n    def _get_pool_name(self):\n        if self.engine.pool.logging_name is not None:\n            return self.engine.pool.logging_name\n        return self._get_connection_string()\n\n    def _add_idle_to_connection_usage(self, value):\n        self.connections_usage.add(\n            value,\n            attributes={\n                \"pool.name\": self._get_pool_name(),\n                \"state\": \"idle\",\n            },\n        )\n\n    def _add_used_to_connection_usage(self, value):\n        self.connections_usage.add(\n            value,\n            attributes={\n                \"pool.name\": self._get_pool_name(),\n                \"state\": \"used\",\n            },\n        )\n\n    def _pool_connect(self, _dbapi_connection, _connection_record):\n        self._add_idle_to_connection_usage(1)\n\n    def _pool_close(self, _dbapi_connection, _connection_record):\n        self._add_idle_to_connection_usage(-1)\n\n    # Called when a connection returns to the pool.\n    def _pool_checkin(self, _dbapi_connection, _connection_record):\n        self._add_used_to_connection_usage(-1)\n        self._add_idle_to_connection_usage(1)\n\n    # Called when a connection is retrieved from the Pool.\n    def _pool_checkout(\n        self, _dbapi_connection, _connection_record, _connection_proxy\n    ):\n        self._add_idle_to_connection_usage(-1)\n        self._add_used_to_connection_usage(1)\n\n    @classmethod\n    def _register_event_listener(cls, target, identifier, func, *args, **kw):\n        listen(target, identifier, func, *args, **kw)\n        cls._remove_event_listener_params.append((target, identifier, func))\n\n    @classmethod\n    def remove_all_event_listeners(cls):\n        for remove_params in cls._remove_event_listener_params:\n            remove(*remove_params)\n        cls._remove_event_listener_params.clear()\n\n    def _operation_name(self, db_name, statement):\n        parts = []\n        if isinstance(statement, str):\n            # otel spec recommends against parsing SQL queries. We are not trying to parse SQL\n            # but simply truncating the statement to the first word. This covers probably >95%\n            # use cases and uses the SQL statement in span name correctly as per the spec.\n            # For some very special cases it might not record the correct statement if the SQL\n            # dialect is too weird but in any case it shouldn't break anything.\n            # Strip leading comments so we get the operation name.\n            parts.append(\n                self._leading_comment_remover.sub(\"\", statement).split()[0]\n            )\n        if db_name:\n            parts.append(db_name)\n        if not parts:\n            return self.vendor\n        return \" \".join(parts)\n\n    def _before_cur_exec(\n        self, conn, cursor, statement, params, context, _executemany\n    ):\n        attrs, found = _get_attributes_from_url(conn.engine.url)\n        if not found:\n            attrs = _get_attributes_from_cursor(self.vendor, cursor, attrs)\n\n        db_name = attrs.get(SpanAttributes.DB_NAME, \"\")\n        span = self.tracer.start_span(\n            self._operation_name(db_name, statement),\n            kind=trace.SpanKind.CLIENT,\n        )\n        with trace.use_span(span, end_on_exit=False):\n            if span.is_recording():\n                span.set_attribute(SpanAttributes.DB_STATEMENT, statement)\n                span.set_attribute(SpanAttributes.DB_SYSTEM, self.vendor)\n                for key, value in attrs.items():\n                    span.set_attribute(key, value)\n            if self.enable_commenter:\n                commenter_data = dict(\n                    db_driver=conn.engine.driver,\n                    # Driver/framework centric information.\n                    db_framework=f\"sqlalchemy:{__version__}\",\n                )\n\n                if self.commenter_options.get(\"opentelemetry_values\", True):\n                    commenter_data.update(**_get_opentelemetry_values())\n\n                # Filter down to just the requested attributes.\n                commenter_data = {\n                    k: v\n                    for k, v in commenter_data.items()\n                    if self.commenter_options.get(k, True)\n                }\n\n                statement = _add_sql_comment(statement, **commenter_data)\n\n        context._otel_span = span\n\n        return statement, params\n\n\n# pylint: disable=unused-argument\ndef _after_cur_exec(conn, cursor, statement, params, context, executemany):\n    span = getattr(context, \"_otel_span\", None)\n    if span is None:\n        return\n\n    span.end()\n\n\ndef _handle_error(context):\n    span = getattr(context.execution_context, \"_otel_span\", None)\n    if span is None:\n        return\n\n    if span.is_recording():\n        span.set_status(\n            Status(\n                StatusCode.ERROR,\n                str(context.original_exception),\n            )\n        )\n    span.end()\n\n\ndef _get_attributes_from_url(url):\n    \"\"\"Set connection tags from the url. return true if successful.\"\"\"\n    attrs = {}\n    if url.host:\n        attrs[SpanAttributes.NET_PEER_NAME] = url.host\n    if url.port:\n        attrs[SpanAttributes.NET_PEER_PORT] = url.port\n    if url.database:\n        attrs[SpanAttributes.DB_NAME] = url.database\n    if url.username:\n        attrs[SpanAttributes.DB_USER] = url.username\n    return attrs, bool(url.host)\n\n\ndef _get_attributes_from_cursor(vendor, cursor, attrs):\n    \"\"\"Attempt to set db connection attributes by introspecting the cursor.\"\"\"\n    if vendor == \"postgresql\":\n        info = getattr(getattr(cursor, \"connection\", None), \"info\", None)\n        if not info:\n            return attrs\n\n        attrs[SpanAttributes.DB_NAME] = info.dbname\n        is_unix_socket = info.host and info.host.startswith(\"/\")\n\n        if is_unix_socket:\n            attrs[SpanAttributes.NET_TRANSPORT] = NetTransportValues.UNIX.value\n            if info.port:\n                # postgresql enforces this pattern on all socket names\n                attrs[SpanAttributes.NET_PEER_NAME] = os.path.join(\n                    info.host, f\".s.PGSQL.{info.port}\"\n                )\n        else:\n            attrs[\n                SpanAttributes.NET_TRANSPORT\n            ] = NetTransportValues.IP_TCP.value\n            attrs[SpanAttributes.NET_PEER_NAME] = info.host\n            if info.port:\n                attrs[SpanAttributes.NET_PEER_PORT] = int(info.port)\n    return attrs\n", "path": "instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py"}]}