WL#15629: Add OpenTelemetry tracing

With this worklog, OTel (OpenTelemetry) tracing support is added.
OTel already provides an `opentelemetry.MySQLInstrumentor`
supporting mysql-connector-python which is compatible with this
new feature, however, a new dedicated MySQL instrumentor is also
provided as part of this new feature.

This new instrumentor, which can be found in
`mysql.connector.opentelemetry`, exposes the same API and has
usage similar to `opentelemetry.MySQLInstrumentor`, nevertheless,
the trace morphology is different.

Additionally, a new connection property named
`otel_context_propagation` is added, and it can be used to control
the trace context propagation behavior. A value of `True` is used
as the default meaning that the trace context in progress (if any)
will be propagated to the server. If the user DOES NOT want the
propagation to happen no matter what, `otel_context_propagation`
should be set to `False`.

Change-Id: I0203924344ec6ad555fcf02531dd65b59c5a5193

Author: oscpache

.pre-commit-config.yaml

@@ -43,8 +43,12 @@ repos:
           django-stubs,
           gssapi==1.8.1,
           dnspython==2.1.0,
+          Deprecated>=1.2.6,
+          typing-extensions>=3.7.4,
+          zipp>=0.5,
         ]
         files: lib
+        exclude: lib/mysql/opentelemetry
         args: [
           --disallow-untyped-defs,
           --show-error-codes,

CHANGES.txt

@@ -14,6 +14,7 @@ v8.1.0
 - WL#15749: Remove DMG and MSI support
 - WL#15672: Upgrade Python Protobuf version to 4.21.12
 - WL#15630: Remove Python 3.7 support
+- WL#15629: Add OpenTelemetry tracing
 - WL#15591: Improve the network module
 - BUG#35425076: Fix deallocating None error
 - BUG#35349093: Compression doesn't work with C extension API

cpydist/__init__.py

@@ -119,6 +119,31 @@
 LOGGER.setLevel(logging.WARNING)
 
 
+def get_otel_src_package_data():
+    """Get a list including all py.typed and dist-info files corresponding
+    to opentelemetry-python (see [1]) located at
+    `mysql/opentelemetry`.
+
+    Returns:
+        package_data: List[str].
+
+    References:
+    [1]: https://github.com/open-telemetry/opentelemetry-python
+    """
+    path_otel = os.path.join("lib", "mysql", "opentelemetry")
+
+    package_data = []
+    for root, dirs, filenames in os.walk(os.path.join(os.getcwd(), path_otel, "")):
+        offset = root.replace(os.path.join(os.getcwd(), path_otel, ""), "")
+        for _dir in dirs:
+            if _dir.endswith(".dist-info"):
+                package_data.append(os.path.join(offset, _dir, "*"))
+        for filename in filenames:
+            if filename == "py.typed":
+                package_data.append(os.path.join(offset, filename))
+    return package_data
+
+
 class BaseCommand(Command):
     """Base command class for Connector/Python."""
 
@@ -228,6 +253,7 @@ def finalize_options(self):
         self.distribution.package_data = {
             "mysql.connector": ["py.typed"],
             "mysqlx": ["py.typed"],
+            "mysql.opentelemetry": get_otel_src_package_data(),
         }
         if not cmd_build_ext.skip_vendor:
             self._copy_vendor_libraries()
@@ -519,10 +545,9 @@ def _copy_vendor_libraries(self):
                 "vendor/private/*",
                 "vendor/private/sasl2/*",
             ],
-            "mysql.connector": [
-                "py.typed",
-            ],
+            "mysql.connector": ["py.typed"],
             "mysqlx": ["py.typed"],
+            "mysql.opentelemetry": get_otel_src_package_data(),
         }
 
 

lib/mysql/connector/abstracts.py

@@ -86,12 +86,26 @@
     OperationalError,
     ProgrammingError,
 )
+from .opentelemetry.constants import (
+    CONNECTION_SPAN_NAME,
+    OPTION_CNX_SPAN,
+    OPTION_CNX_TRACER,
+    OTEL_ENABLED,
+)
+
+if OTEL_ENABLED:
+    from .opentelemetry.instrumentation import (
+        end_span,
+        record_exception_event,
+        set_connection_span_attrs,
+        trace,
+    )
+
 from .optionfiles import read_option_files
 from .types import (
     ConnAttrsType,
     DescriptionType,
     HandShakeType,
-    QueryAttrType,
     StrOrBytes,
     SupportedMysqlBinaryProtocolTypes,
     WarningType,
@@ -142,6 +156,11 @@ class MySQLConnectionAbstract(ABC):
 
     def __init__(self) -> None:
         """Initialize"""
+        # opentelemetry related
+        self._tracer: Any = None
+        self._span: Any = None
+        self.otel_context_propagation: bool = True
+
         self._client_flags: int = ClientFlag.get_default()
         self._charset_id: int = 45
         self._sql_mode: Optional[str] = None
@@ -187,7 +206,7 @@ def __init__(self) -> None:
         ]
 
         self._prepared_statements: Any = None
-        self._query_attrs: QueryAttrType = []
+        self._query_attrs: Dict[str, Any] = {}
 
         self._ssl_active: bool = False
         self._auth_plugin: Optional[str] = None
@@ -233,19 +252,32 @@ def have_next_result(self) -> bool:
         return self._have_next_result
 
     @property
-    def query_attrs(self) -> QueryAttrType:
+    def query_attrs(self) -> List[Tuple[str, Any]]:
         """Return query attributes list."""
-        return self._query_attrs
+        return list(self._query_attrs.items())
 
     def query_attrs_append(
         self, value: Tuple[str, SupportedMysqlBinaryProtocolTypes]
     ) -> None:
-        """Add element to the query attributes list."""
-        self._query_attrs.append(value)
+        """Add element to the query attributes list.
+
+        If an element in the query attributes list already matches
+        the attribute name provided, the new element will NOT be added.
+        """
+        attr_name, attr_value = value
+        if attr_name not in self._query_attrs:
+            self._query_attrs[attr_name] = attr_value
+
+    def query_attrs_remove(self, name: str) -> Any:
+        """Remove element by name from the query attributes list.
+
+        If no match, `None` is returned; else the corresponding value is returned.
+        """
+        return self._query_attrs.pop(name, None)
 
     def query_attrs_clear(self) -> None:
         """Clear query attributes list."""
-        del self._query_attrs[:]
+        self._query_attrs = {}
 
     def _validate_tls_ciphersuites(self) -> None:
         """Validates the tls_ciphersuites option."""
@@ -470,6 +502,10 @@ def config(self, **kwargs: Any) -> None:
 
         Raises on errors.
         """
+        # opentelemetry related
+        self._span = kwargs.pop(OPTION_CNX_SPAN, None)
+        self._tracer = kwargs.pop(OPTION_CNX_TRACER, None)
+
         config = kwargs.copy()
         if "dsn" in config:
             raise NotSupportedError("Data source name is not supported")
@@ -1198,22 +1234,40 @@ def reconnect(self, attempts: int = 1, delay: int = 0) -> None:
         Raises InterfaceError on errors.
         """
         counter = 0
-        while counter != attempts:
-            counter = counter + 1
-            try:
-                self.disconnect()
-                self.connect()
-                if self.is_connected():
-                    break
-            except (Error, IOError) as err:
-                if counter == attempts:
-                    msg = (
-                        f"Can not reconnect to MySQL after {attempts} "
-                        f"attempt(s): {err}"
-                    )
-                    raise InterfaceError(msg) from err
-            if delay > 0:
-                sleep(delay)
+        span = None
+
+        if self._tracer:
+            span = self._tracer.start_span(
+                name=CONNECTION_SPAN_NAME, kind=trace.SpanKind.CLIENT
+            )
+
+        try:
+            while counter != attempts:
+                counter = counter + 1
+                try:
+                    self.disconnect()
+                    self.connect()
+                    if self.is_connected():
+                        break
+                except (Error, IOError) as err:
+                    if counter == attempts:
+                        msg = (
+                            f"Can not reconnect to MySQL after {attempts} "
+                            f"attempt(s): {err}"
+                        )
+                        raise InterfaceError(msg) from err
+                if delay > 0:
+                    sleep(delay)
+        except InterfaceError as interface_err:
+            if OTEL_ENABLED:
+                set_connection_span_attrs(self, span)
+                record_exception_event(span, interface_err)
+                end_span(span)
+            raise
+
+        self._span = span
+        if OTEL_ENABLED:
+            set_connection_span_attrs(self, self._span)
 
     @abstractmethod
     def is_connected(self) -> Any:
@@ -1553,7 +1607,7 @@ def close(self) -> Any:
     @abstractmethod
     def execute(
         self,
-        operation: Any,
+        operation: str,
         params: Union[Sequence[Any], Dict[str, Any]] = (),
         multi: bool = False,
     ) -> Any:
@@ -1577,7 +1631,7 @@ def execute(
 
     @abstractmethod
     def executemany(
-        self, operation: Any, seq_params: Sequence[Union[Sequence[Any], Dict[str, Any]]]
+        self, operation: str, seq_params: Sequence[Union[Sequence[Any], Dict[str, Any]]]
     ) -> Any:
         """Execute the given operation multiple times
 
@@ -1710,7 +1764,7 @@ def fetchwarnings(self) -> Optional[List[WarningType]]:
         """Returns Warnings."""
         return self._warnings
 
-    def get_attributes(self) -> Optional[List[Tuple[Any, Any]]]:
+    def get_attributes(self) -> Optional[List[Tuple[str, Any]]]:
         """Get the added query attributes so far."""
         if hasattr(self, "_cnx"):
             return self._cnx.query_attrs
@@ -1731,6 +1785,19 @@ def add_attribute(self, name: str, value: Any) -> None:
         elif hasattr(self, "_connection"):
             self._connection.query_attrs_append((name, value))
 
+    def remove_attribute(self, name: str) -> Any:
+        """Remove a query attribute by name.
+
+        If no match, `None` is returned; else the corresponding value is returned.
+        """
+        if not isinstance(name, str):
+            raise ProgrammingError("Parameter `name` must be a string type")
+        if hasattr(self, "_cnx"):
+            return self._cnx.query_attrs_remove(name)
+        if hasattr(self, "_connection"):
+            return self._connection.query_attrs_remove(name)
+        return None
+
     def clear_attributes(self) -> None:
         """Remove all the query attributes."""
         if hasattr(self, "_cnx"):