diff --git a/CHANGELOG.md b/CHANGELOG.md
index a243091b1d..1e09da0110 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -48,6 +48,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2942](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2942))
 - `opentelemetry-instrumentation-click`: new instrumentation to trace click commands
   ([#2994](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2994))
+- `opentelemetry-instrumentation-mysql` Add sqlcommenter support
+  ([#3023](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3023))
 
 ### Fixed
 
diff --git a/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py b/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py
index d8db967f47..c2b63e59fe 100644
--- a/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py
@@ -45,6 +45,10 @@
 import wrapt
 
 from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation.dbapi.proxy import (
+    BaseTracedConnectionProxy,
+    BaseTracedCursorProxy,
+)
 from opentelemetry.instrumentation.dbapi.version import __version__
 from opentelemetry.instrumentation.sqlcommenter_utils import _add_sql_comment
 from opentelemetry.instrumentation.utils import (
@@ -191,6 +195,8 @@ def instrument_connection(
     enable_commenter: bool = False,
     commenter_options: dict = None,
     connect_module: typing.Callable[..., typing.Any] = None,
+    db_api_integration_factory: typing.ClassVar = None,
+    get_cnx_proxy: typing.Callable[..., typing.Any] = None,
 ):
     """Enable instrumentation in a database connection.
 
@@ -206,6 +212,10 @@ def instrument_connection(
         enable_commenter: Flag to enable/disable sqlcommenter.
         commenter_options: Configurations for tags to be appended at the sql query.
         connect_module: Module name where connect method is available.
+        db_api_integration_factory: The `DatabaseApiIntegration` to use. If none is passed the
+            default one is used.
+        get_cnx_proxy: Method to get traced connextion proxy.  If none is passed the
+            default one is used.
 
     Returns:
         An instrumented connection.
@@ -214,7 +224,11 @@ def instrument_connection(
         _logger.warning("Connection already instrumented")
         return connection
 
-    db_integration = DatabaseApiIntegration(
+    db_api_integration_factory = (
+        db_api_integration_factory or DatabaseApiIntegration
+    )
+
+    db_integration = db_api_integration_factory(
         name,
         database_system,
         connection_attributes=connection_attributes,
@@ -226,7 +240,9 @@ def instrument_connection(
         connect_module=connect_module,
     )
     db_integration.get_connection_attributes(connection)
-    return get_traced_connection_proxy(connection, db_integration)
+
+    get_cnx_proxy = get_cnx_proxy or get_traced_connection_proxy
+    return get_cnx_proxy(connection, db_integration)
 
 
 def uninstrument_connection(connection):
@@ -397,31 +413,12 @@ def get_traced_connection_proxy(
     connection, db_api_integration, *args, **kwargs
 ):
     # pylint: disable=abstract-method
-    class TracedConnectionProxy(wrapt.ObjectProxy):
-        # pylint: disable=unused-argument
-        def __init__(self, connection, *args, **kwargs):
-            wrapt.ObjectProxy.__init__(self, connection)
-
-        def __getattribute__(self, name):
-            if object.__getattribute__(self, name):
-                return object.__getattribute__(self, name)
-
-            return object.__getattribute__(
-                object.__getattribute__(self, "_connection"), name
-            )
-
+    class TracedConnectionProxy(BaseTracedConnectionProxy):
         def cursor(self, *args, **kwargs):
             return get_traced_cursor_proxy(
                 self.__wrapped__.cursor(*args, **kwargs), db_api_integration
             )
 
-        def __enter__(self):
-            self.__wrapped__.__enter__()
-            return self
-
-        def __exit__(self, *args, **kwargs):
-            self.__wrapped__.__exit__(*args, **kwargs)
-
     return TracedConnectionProxy(connection, *args, **kwargs)
 
 
@@ -547,34 +544,11 @@ def traced_execution(
 
 
 def get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):
-    _cursor_tracer = CursorTracer(db_api_integration)
-
-    # pylint: disable=abstract-method
-    class TracedCursorProxy(wrapt.ObjectProxy):
-        # pylint: disable=unused-argument
-        def __init__(self, cursor, *args, **kwargs):
-            wrapt.ObjectProxy.__init__(self, cursor)
-
-        def execute(self, *args, **kwargs):
-            return _cursor_tracer.traced_execution(
-                self.__wrapped__, self.__wrapped__.execute, *args, **kwargs
+    class TracedCursorProxy(BaseTracedCursorProxy):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self._cursor_tracer = CursorTracer(
+                db_api_integration,
             )
 
-        def executemany(self, *args, **kwargs):
-            return _cursor_tracer.traced_execution(
-                self.__wrapped__, self.__wrapped__.executemany, *args, **kwargs
-            )
-
-        def callproc(self, *args, **kwargs):
-            return _cursor_tracer.traced_execution(
-                self.__wrapped__, self.__wrapped__.callproc, *args, **kwargs
-            )
-
-        def __enter__(self):
-            self.__wrapped__.__enter__()
-            return self
-
-        def __exit__(self, *args, **kwargs):
-            self.__wrapped__.__exit__(*args, **kwargs)
-
     return TracedCursorProxy(cursor, *args, **kwargs)
diff --git a/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/proxy.py b/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/proxy.py
new file mode 100644
index 0000000000..5b6a4872af
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/proxy.py
@@ -0,0 +1,100 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from abc import ABC, ABCMeta, abstractmethod
+
+import wrapt
+
+
+class BaseMeta(ABCMeta, type(wrapt.ObjectProxy)):
+    """Metaclass compatibility helper for PyPy for derived classes"""
+
+
+class BaseTracedConnectionProxy(ABC, wrapt.ObjectProxy, metaclass=BaseMeta):
+    """ABC for traced database client connection proxy.
+
+    Child classes are used to wrap Connection objects and
+    generate telemetry based on provided integration settings.
+
+    Child classes must implement cursor(), which should return
+    a traced database client cursor proxy depending on database
+    driver needs.
+    """
+
+    # pylint: disable=unused-argument
+    def __init__(self, connection, *args, **kwargs):
+        wrapt.ObjectProxy.__init__(self, connection)
+
+    def __getattribute__(self, name):
+        if object.__getattribute__(self, name):
+            return object.__getattribute__(self, name)
+
+        return object.__getattribute__(
+            object.__getattribute__(self, "_connection"), name
+        )
+
+    @abstractmethod
+    def cursor(self, *args, **kwargs):
+        """Returns instrumented database query cursor"""
+
+    def __enter__(self):
+        self.__wrapped__.__enter__()
+        return self
+
+    def __exit__(self, *args, **kwargs):
+        self.__wrapped__.__exit__(*args, **kwargs)
+
+
+# pylint: disable=abstract-method
+class BaseTracedCursorProxy(ABC, wrapt.ObjectProxy, metaclass=BaseMeta):
+    """ABC for traced database client cursor proxy.
+
+    Child classes are used to wrap Cursor objects and
+    generate telemetry based on provided integration settings.
+
+    Child classes must implement __init__(), which should set
+    class variable _cursor_tracer as a CursorTracer depending
+    on database driver needs.
+    """
+
+    _cursor_tracer = None
+
+    # pylint: disable=unused-argument
+    @abstractmethod
+    def __init__(self, cursor, *args, **kwargs):
+        """Wrap db client cursor for tracing"""
+        wrapt.ObjectProxy.__init__(self, cursor)
+
+    def callproc(self, *args, **kwargs):
+        return self._cursor_tracer.traced_execution(
+            self.__wrapped__, self.__wrapped__.callproc, *args, **kwargs
+        )
+
+    def execute(self, *args, **kwargs):
+        return self._cursor_tracer.traced_execution(
+            self.__wrapped__, self.__wrapped__.execute, *args, **kwargs
+        )
+
+    def executemany(self, *args, **kwargs):
+        return self._cursor_tracer.traced_execution(
+            self.__wrapped__, self.__wrapped__.executemany, *args, **kwargs
+        )
+
+    def __enter__(self):
+        self.__wrapped__.__enter__()
+        return self
+
+    def __exit__(self, *args, **kwargs):
+        self.__wrapped__.__exit__(*args, **kwargs)
diff --git a/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_baseclass.py b/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_baseclass.py
new file mode 100644
index 0000000000..90965edef5
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_baseclass.py
@@ -0,0 +1,144 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+
+from opentelemetry.instrumentation.dbapi import (
+    BaseTracedConnectionProxy,
+    BaseTracedCursorProxy,
+)
+from opentelemetry.test.test_base import TestBase
+
+
+class TestDBApiBaseClasses(TestBase):
+    class TestConnectionProxy(BaseTracedConnectionProxy):
+        def cursor(self, *args, **kwargs):
+            return "foo-cursor"
+
+    def test_base_traced_connection_proxy_init(self):
+        mock_connection = mock.Mock()
+        proxy = self.TestCursorProxy(mock_connection)
+        self.assertIs(proxy.__wrapped__, mock_connection)
+
+    def test_base_traced_connection_proxy_getattr_direct_attr(self):
+        mock_connection = mock.Mock()
+        proxy = self.TestConnectionProxy(mock_connection)
+        proxy.foo_attribute = "bar"  # pylint: disable=attribute-defined-outside-init
+        self.assertEqual(proxy.foo_attribute, "bar")
+
+    def test_base_traced_connection_proxy_getattr_cnx_attr(self):
+        mock_connection = mock.Mock()
+        mock_connection.foo_attribute = "bar"
+        proxy = self.TestConnectionProxy(mock_connection)
+        self.assertEqual(proxy.foo_attribute, "bar")
+
+    def test_base_traced_connection_proxy_cursor(self):
+        assert self.TestConnectionProxy("foo-cnx").cursor() == "foo-cursor"
+
+    def test_base_traced_connection_proxy_enter(self):
+        mock_connection = mock.Mock()
+        mock_connection.__enter__ = mock.Mock()
+        mock_connection.__exit__ = mock.Mock()
+        proxy = self.TestConnectionProxy(mock_connection)
+        with proxy:
+            mock_connection.__enter__.assert_called_once()
+
+    def test_base_traced_connection_proxy_exit(self):
+        mock_connection = mock.Mock()
+        mock_connection.__enter__ = mock.Mock()
+        mock_connection.__exit__ = mock.Mock()
+        proxy = self.TestConnectionProxy(mock_connection)
+        with proxy:
+            pass
+        mock_connection.__exit__.assert_called_once()
+
+    class TestCursorProxy(BaseTracedCursorProxy):
+        def __init__(self, cursor, *args, **kwargs):
+            super().__init__(cursor, *args, **kwargs)
+            self._cursor_tracer = kwargs.get("test_cursor_tracer", "foo")
+
+    def test_base_traced_cursor_proxy_init(self):
+        mock_cursor = mock.Mock()
+        mock_cursor_tracer = mock.Mock()
+        proxy = self.TestCursorProxy(
+            mock_cursor,
+            test_cursor_tracer=mock_cursor_tracer,
+        )
+        self.assertIs(proxy.__wrapped__, mock_cursor)
+        assert proxy._cursor_tracer == mock_cursor_tracer
+
+    def test_base_traced_cursor_proxy_callproc(self):
+        mock_cursor = mock.Mock()
+        mock_cursor.callproc = mock.Mock()
+        mock_cursor_tracer = mock.Mock()
+        mock_cursor_tracer.traced_execution = mock.Mock()
+        proxy = self.TestCursorProxy(
+            mock_cursor,
+            test_cursor_tracer=mock_cursor_tracer,
+        )
+        args = ("foo_name",)
+        kwargs = {"foo": "bar"}
+        proxy.callproc(*args, **kwargs)
+        mock_cursor_tracer.traced_execution.assert_called_once_with(
+            mock_cursor, mock_cursor.callproc, *args, **kwargs
+        )
+
+    def test_base_traced_cursor_proxy_execute(self):
+        mock_cursor = mock.Mock()
+        mock_cursor.execute = mock.Mock()
+        mock_cursor_tracer = mock.Mock()
+        mock_cursor_tracer.traced_execution = mock.Mock()
+        proxy = self.TestCursorProxy(
+            mock_cursor,
+            test_cursor_tracer=mock_cursor_tracer,
+        )
+        args = ("foo_name",)
+        kwargs = {"foo": "bar"}
+        proxy.execute(*args, **kwargs)
+        mock_cursor_tracer.traced_execution.assert_called_once_with(
+            mock_cursor, mock_cursor.execute, *args, **kwargs
+        )
+
+    def test_base_traced_cursor_proxy_executemany(self):
+        mock_cursor = mock.Mock()
+        mock_cursor.executemany = mock.Mock()
+        mock_cursor_tracer = mock.Mock()
+        mock_cursor_tracer.traced_execution = mock.Mock()
+        proxy = self.TestCursorProxy(
+            mock_cursor,
+            test_cursor_tracer=mock_cursor_tracer,
+        )
+        args = ("foo_name",)
+        kwargs = {"foo": "bar"}
+        proxy.executemany(*args, **kwargs)
+        mock_cursor_tracer.traced_execution.assert_called_once_with(
+            mock_cursor, mock_cursor.executemany, *args, **kwargs
+        )
+
+    def test_base_traced_cursor_proxy_enter(self):
+        mock_cursor = mock.Mock()
+        mock_cursor.__enter__ = mock.Mock()
+        mock_cursor.__exit__ = mock.Mock()
+        proxy = self.TestCursorProxy(mock_cursor)
+        with proxy:
+            mock_cursor.__enter__.assert_called_once()
+
+    def test_base_traced_cursor_proxy_exit(self):
+        mock_cursor = mock.Mock()
+        mock_cursor.__enter__ = mock.Mock()
+        mock_cursor.__exit__ = mock.Mock()
+        proxy = self.TestCursorProxy(mock_cursor)
+        with proxy:
+            pass
+        mock_cursor.__exit__.assert_called_once()
diff --git a/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py b/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py
index 2ffa2f3d5b..04312cc9cf 100644
--- a/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py
+++ b/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py
@@ -592,9 +592,15 @@ def test_instrument_connection(self):
         connection2 = dbapi.instrument_connection(self.tracer, connection, "-")
         self.assertIs(connection2.__wrapped__, connection)
 
+    @mock.patch(
+        "opentelemetry.instrumentation.dbapi.get_traced_connection_proxy"
+    )
     @mock.patch("opentelemetry.instrumentation.dbapi.DatabaseApiIntegration")
-    def test_instrument_connection_kwargs_defaults(self, mock_dbapiint):
-        dbapi.instrument_connection(self.tracer, mock.Mock(), "foo")
+    def test_instrument_connection_kwargs_defaults(
+        self, mock_dbapiint, mock_get_cnx_proxy
+    ):
+        mock_get_cnx_proxy.return_value = "foo_cnx"
+        cnx = dbapi.instrument_connection(self.tracer, mock.Mock(), "foo")
         kwargs = mock_dbapiint.call_args[1]
         self.assertEqual(kwargs["connection_attributes"], None)
         self.assertEqual(kwargs["version"], "")
@@ -603,11 +609,19 @@ def test_instrument_connection_kwargs_defaults(self, mock_dbapiint):
         self.assertEqual(kwargs["enable_commenter"], False)
         self.assertEqual(kwargs["commenter_options"], None)
         self.assertEqual(kwargs["connect_module"], None)
+        assert cnx == "foo_cnx"
 
+    @mock.patch(
+        "opentelemetry.instrumentation.dbapi.get_traced_connection_proxy"
+    )
     @mock.patch("opentelemetry.instrumentation.dbapi.DatabaseApiIntegration")
-    def test_instrument_connection_kwargs_provided(self, mock_dbapiint):
+    def test_instrument_connection_kwargs_provided(
+        self, mock_dbapiint, mock_get_cnx_proxy
+    ):
         mock_tracer_provider = mock.MagicMock()
         mock_connect_module = mock.MagicMock()
+        mock_custom_dbapiint = mock.MagicMock()
+        mock_custom_get_cnx_proxy = mock.MagicMock()
         dbapi.instrument_connection(
             self.tracer,
             mock.Mock(),
@@ -619,8 +633,11 @@ def test_instrument_connection_kwargs_provided(self, mock_dbapiint):
             enable_commenter=True,
             commenter_options={"foo": "bar"},
             connect_module=mock_connect_module,
+            db_api_integration_factory=mock_custom_dbapiint,
+            get_cnx_proxy=mock_custom_get_cnx_proxy,
         )
-        kwargs = mock_dbapiint.call_args[1]
+        mock_dbapiint.assert_not_called()
+        kwargs = mock_custom_dbapiint.call_args[1]
         self.assertEqual(kwargs["connection_attributes"], {"foo": "bar"})
         self.assertEqual(kwargs["version"], "test")
         self.assertIs(kwargs["tracer_provider"], mock_tracer_provider)
@@ -628,6 +645,8 @@ def test_instrument_connection_kwargs_provided(self, mock_dbapiint):
         self.assertEqual(kwargs["enable_commenter"], True)
         self.assertEqual(kwargs["commenter_options"], {"foo": "bar"})
         self.assertIs(kwargs["connect_module"], mock_connect_module)
+        mock_get_cnx_proxy.assert_not_called()
+        mock_custom_get_cnx_proxy.assert_called_once()
 
     def test_uninstrument_connection(self):
         connection = mock.Mock()
diff --git a/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py b/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py
index d88dffde2b..208a399cc9 100644
--- a/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py
@@ -34,19 +34,95 @@
     cursor.close()
     cnx.close()
 
+SQLCOMMENTER
+*****************************************
+You can optionally configure mysql-connector instrumentation to enable sqlcommenter which enriches
+the query with contextual information.
+
+Usage
+-----
+
+.. code:: python
+
+    import mysql.connector
+    from opentelemetry.instrumentation.mysql import MySQLInstrumentor
+
+    MySQLInstrumentor().instrument(enable_commenter=True, commenter_options={})
+
+    cnx = mysql.connector.connect(database="MySQL_Database")
+    cursor = cnx.cursor()
+    cursor.execute("INSERT INTO test (testField) VALUES (123)")
+    cursor.close()
+    cnx.close()
+
+
+For example,
+::
+
+   Invoking cursor.execute("INSERT INTO test (testField) VALUES (123)") will lead to sql query "INSERT INTO test (testField) VALUES (123)" but when SQLCommenter is enabled
+   the query will get appended with some configurable tags like "INSERT INTO test (testField) VALUES (123) /*tag=value*/;"
+
+
+Note that sqlcommenter for mysql-connector is not supported if cursor is initialized with `prepared=True`, which would natively prepare and execute MySQL statements. Adding sqlcommenting would introduce a severe performance penalty by repeating `Prepare` of statements by mysql-connector that are made unique by traceparent in sqlcomment. Therefore, sqlcommenter is not supported for this use case.
+
+
+SQLCommenter Configurations
+***************************
+We can configure the tags to be appended to the sqlquery log by adding configuration inside commenter_options(default:{}) keyword
+
+db_driver = True(Default) or False
+
+For example,
+::
+Enabling this flag will add mysql.connector and its version, e.g. /*mysql.connector%%3A1.2.3*/
+
+dbapi_threadsafety = True(Default) or False
+
+For example,
+::
+Enabling this flag will add threadsafety /*dbapi_threadsafety=2*/
+
+dbapi_level = True(Default) or False
+
+For example,
+::
+Enabling this flag will add dbapi_level /*dbapi_level='2.0'*/
+
+mysql_client_version = True(Default) or False
+
+For example,
+::
+Enabling this flag will add mysql_client_version /*mysql_client_version='123'*/
+
+driver_paramstyle = True(Default) or False
+
+For example,
+::
+Enabling this flag will add driver_paramstyle /*driver_paramstyle='pyformat'*/
+
+opentelemetry_values = True(Default) or False
+
+For example,
+::
+Enabling this flag will add traceparent values /*traceparent='00-03afa25236b8cd948fa853d67038ac79-405ff022e8247c46-01'*/
+
 API
 ---
 """
 
-from typing import Collection
+import logging
+import typing
 
 import mysql.connector
 
+from opentelemetry import trace as trace_api
 from opentelemetry.instrumentation import dbapi
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.mysql.package import _instruments
 from opentelemetry.instrumentation.mysql.version import __version__
 
+_logger = logging.getLogger(__name__)
+
 
 class MySQLInstrumentor(BaseInstrumentor):
     _CONNECTION_ATTRIBUTES = {
@@ -58,7 +134,7 @@ class MySQLInstrumentor(BaseInstrumentor):
 
     _DATABASE_SYSTEM = "mysql"
 
-    def instrumentation_dependencies(self) -> Collection[str]:
+    def instrumentation_dependencies(self) -> typing.Collection[str]:
         return _instruments
 
     def _instrument(self, **kwargs):
@@ -66,6 +142,8 @@ def _instrument(self, **kwargs):
         https://dev.mysql.com/doc/connector-python/en/
         """
         tracer_provider = kwargs.get("tracer_provider")
+        enable_sqlcommenter = kwargs.get("enable_commenter", False)
+        commenter_options = kwargs.get("commenter_options", {})
 
         dbapi.wrap_connect(
             __name__,
@@ -75,6 +153,9 @@ def _instrument(self, **kwargs):
             self._CONNECTION_ATTRIBUTES,
             version=__version__,
             tracer_provider=tracer_provider,
+            db_api_integration_factory=DatabaseApiIntegration,
+            enable_commenter=enable_sqlcommenter,
+            commenter_options=commenter_options,
         )
 
     def _uninstrument(self, **kwargs):
@@ -82,13 +163,21 @@ def _uninstrument(self, **kwargs):
         dbapi.unwrap_connect(mysql.connector, "connect")
 
     # pylint:disable=no-self-use
-    def instrument_connection(self, connection, tracer_provider=None):
+    def instrument_connection(
+        self,
+        connection,
+        tracer_provider: typing.Optional[trace_api.TracerProvider] = None,
+        enable_commenter: bool = False,
+        commenter_options: dict = None,
+    ):
         """Enable instrumentation in a MySQL connection.
 
         Args:
             connection: The connection to instrument.
             tracer_provider: The optional tracer provider to use. If omitted
                 the current globally configured one is used.
+            enable_commenter: Flag to enable/disable sqlcommenter.
+            commenter_options: Configurations for tags to be appended at the sql query.
 
         Returns:
             An instrumented connection.
@@ -100,9 +189,17 @@ def instrument_connection(self, connection, tracer_provider=None):
             self._CONNECTION_ATTRIBUTES,
             version=__version__,
             tracer_provider=tracer_provider,
+            enable_commenter=enable_commenter,
+            commenter_options=commenter_options,
+            connect_module=mysql.connector,
+            db_api_integration_factory=DatabaseApiIntegration,
+            get_cnx_proxy=get_traced_connection_proxy,
         )
 
-    def uninstrument_connection(self, connection):
+    def uninstrument_connection(
+        self,
+        connection,
+    ):
         """Disable instrumentation in a MySQL connection.
 
         Args:
@@ -112,3 +209,83 @@ def uninstrument_connection(self, connection):
             An uninstrumented connection.
         """
         return dbapi.uninstrument_connection(connection)
+
+
+class DatabaseApiIntegration(dbapi.DatabaseApiIntegration):
+    def wrapped_connection(
+        self,
+        connect_method: typing.Callable[..., typing.Any],
+        args: typing.Tuple[typing.Any, typing.Any],
+        kwargs: typing.Dict[typing.Any, typing.Any],
+    ):
+        """Add object proxy to connection object that checks cursor type."""
+        connection = connect_method(*args, **kwargs)
+        self.get_connection_attributes(connection)
+        return get_traced_connection_proxy(connection, self)
+
+
+def get_traced_connection_proxy(
+    connection, db_api_integration, *args, **kwargs
+):
+    # pylint: disable=abstract-method
+    class TracedConnectionProxy(dbapi.BaseTracedConnectionProxy):
+        def cursor(self, *args, **kwargs):
+            wrapped_cursor = self.__wrapped__.cursor(*args, **kwargs)
+
+            # It's common to have multiple db client cursors per app,
+            # so enable_commenter is calculated for the cursor level for
+            # traced query execution, based on original configuration.
+            enable_commenter_cursor = db_api_integration.enable_commenter
+
+            # If a mysql-connector cursor was created with prepared=True,
+            # then MySQL statements will be prepared and executed natively.
+            # Adding sqlcommenting would introduce a severe performance
+            # penalty by repeating `Prepare` of statements by mysql-connector
+            # that are made unique by traceparent in sqlcomment. Therefore,
+            # sqlcomment is not supported for this use case. It is not an
+            # issue if prepared=False.
+            # This is here because wrapped cursor is created when application
+            # side creates cursor. After that, the instrumentor knows what
+            # kind of cursor was initialized.
+            if enable_commenter_cursor:
+                is_prepared = kwargs.get("prepared", False)
+                if is_prepared:
+                    _logger.warning(
+                        "sqlcomment is not supported for query statements executed by cursors with native prepared statement support. Disabling sqlcommenting for instrumentation of %s.",
+                        db_api_integration.connect_module.__name__,
+                    )
+                    enable_commenter_cursor = False
+            return get_traced_cursor_proxy(
+                wrapped_cursor,
+                db_api_integration,
+                enable_commenter_cursor,
+            )
+
+    return TracedConnectionProxy(connection, *args, **kwargs)
+
+
+class CursorTracer(dbapi.CursorTracer):
+    def __init__(
+        self,
+        db_api_integration: DatabaseApiIntegration,
+        enable_commenter: bool = False,
+    ) -> None:
+        super().__init__(db_api_integration)
+        # It's common to have multiple db client cursors per app,
+        # so enable_commenter is set at the cursor level and used
+        # during CursorTracer.traced_execution for mysql-connector
+        self._commenter_enabled = enable_commenter
+
+
+def get_traced_cursor_proxy(
+    cursor, db_api_integration, enable_commenter_cursor, *args, **kwargs
+):
+    class TracedCursorProxy(dbapi.BaseTracedCursorProxy):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self._cursor_tracer = CursorTracer(
+                db_api_integration,
+                enable_commenter_cursor,
+            )
+
+    return TracedCursorProxy(cursor, *args, **kwargs)
diff --git a/instrumentation/opentelemetry-instrumentation-mysql/tests/test_mysql_integration.py b/instrumentation/opentelemetry-instrumentation-mysql/tests/test_mysql_integration.py
index 79399cce7f..158e188e9a 100644
--- a/instrumentation/opentelemetry-instrumentation-mysql/tests/test_mysql_integration.py
+++ b/instrumentation/opentelemetry-instrumentation-mysql/tests/test_mysql_integration.py
@@ -15,10 +15,15 @@
 from unittest import mock
 
 import mysql.connector
+import pytest
 
 import opentelemetry.instrumentation.mysql
 from opentelemetry import trace as trace_api
-from opentelemetry.instrumentation.mysql import MySQLInstrumentor
+from opentelemetry.instrumentation.mysql import (
+    MySQLInstrumentor,
+    get_traced_connection_proxy,
+    get_traced_cursor_proxy,
+)
 from opentelemetry.sdk import resources
 from opentelemetry.test.test_base import TestBase
 
@@ -33,6 +38,10 @@ def connect_and_execute_query():
 
 
 class TestMysqlIntegration(TestBase):
+    @pytest.fixture(autouse=True)
+    def inject_fixtures(self, caplog):
+        self.caplog = caplog  # pylint: disable=attribute-defined-outside-init
+
     def tearDown(self):
         super().tearDown()
         with self.disable_logging():
@@ -62,6 +71,133 @@ def test_instrumentor(self, mock_connect):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
 
+    @mock.patch("opentelemetry.instrumentation.dbapi.wrap_connect")
+    @mock.patch("mysql.connector.connect")
+    def test_instrumentor_sqlcomment_enabled_dbapi_kwargs(
+        self,
+        mock_connect,
+        mock_wrap_connect,
+    ):
+        MySQLInstrumentor().instrument(
+            enable_commenter=True,
+            commenter_options={"foo": True},
+        )
+        kwargs = mock_wrap_connect.call_args[1]
+        self.assertEqual(kwargs["enable_commenter"], True)
+        self.assertEqual(kwargs["commenter_options"], {"foo": True})
+
+    def test_instrument_with_dbapi_sqlcomment_enabled(
+        self,
+    ):
+        mock_connect_module = mock.MagicMock(
+            __name__="mysql.connector",
+            __version__="foobar",
+            threadsafety="123",
+            apilevel="123",
+            paramstyle="test",
+        )
+        mock_cursor = mock_connect_module.connect().cursor()
+        mock_cursor._cnx._cmysql.get_client_info.return_value = "foobaz"
+        mock_connection = mock.MagicMock()
+        mock_connection.cursor.return_value = mock_cursor
+
+        with mock.patch(
+            "opentelemetry.instrumentation.mysql.mysql.connector",
+            mock_connect_module,
+        ), mock.patch(
+            "opentelemetry.instrumentation.dbapi.util_version",
+            return_value="foobar",
+        ):
+            MySQLInstrumentor()._instrument(
+                enable_commenter=True,
+            )
+            cnx = mock_connect_module.connect(database="test")
+            cursor = cnx.cursor()
+            cursor.execute("Select 1;")
+
+            spans_list = self.memory_exporter.get_finished_spans()
+            span = spans_list[0]
+            span_id = format(span.get_span_context().span_id, "016x")
+            trace_id = format(span.get_span_context().trace_id, "032x")
+            self.assertEqual(
+                mock_cursor.execute.call_args[0][0],
+                f"Select 1 /*db_driver='mysql.connector%%3Afoobar',dbapi_level='123',dbapi_threadsafety='123',driver_paramstyle='test',mysql_client_version='foobaz',traceparent='00-{trace_id}-{span_id}-01'*/;",
+            )
+
+    def test_instrument_with_dbapi_sqlcomment_enabled_with_options(
+        self,
+    ):
+        mock_connect_module = mock.MagicMock(
+            __name__="mysql.connector",
+            __version__="foobar",
+            threadsafety="123",
+            apilevel="123",
+            paramstyle="test",
+        )
+        mock_cursor = mock_connect_module.connect().cursor()
+        mock_cursor._cnx._cmysql.get_client_info.return_value = "foobaz"
+        mock_connection = mock.MagicMock()
+        mock_connection.cursor.return_value = mock_cursor
+
+        with mock.patch(
+            "opentelemetry.instrumentation.mysql.mysql.connector",
+            mock_connect_module,
+        ), mock.patch(
+            "opentelemetry.instrumentation.dbapi.util_version",
+            return_value="foobar",
+        ):
+            MySQLInstrumentor()._instrument(
+                enable_commenter=True,
+                commenter_options={
+                    "dbapi_level": False,
+                    "dbapi_threadsafety": True,
+                    "driver_paramstyle": False,
+                },
+            )
+            cnx = mock_connect_module.connect(database="test")
+            cursor = cnx.cursor()
+            cursor.execute("Select 1;")
+
+            spans_list = self.memory_exporter.get_finished_spans()
+            span = spans_list[0]
+            span_id = format(span.get_span_context().span_id, "016x")
+            trace_id = format(span.get_span_context().trace_id, "032x")
+            self.assertEqual(
+                mock_cursor.execute.call_args[0][0],
+                f"Select 1 /*db_driver='mysql.connector%%3Afoobar',dbapi_threadsafety='123',mysql_client_version='foobaz',traceparent='00-{trace_id}-{span_id}-01'*/;",
+            )
+
+    def test_instrument_with_dbapi_sqlcomment_not_enabled_default(
+        self,
+    ):
+        mock_connect_module = mock.MagicMock(
+            __name__="mysql.connector",
+            __version__="foobar",
+            threadsafety="123",
+            apilevel="123",
+            paramstyle="test",
+        )
+        mock_cursor = mock_connect_module.connect().cursor()
+        mock_cursor._cnx._cmysql.get_client_info.return_value = "foobaz"
+        mock_connection = mock.MagicMock()
+        mock_connection.cursor.return_value = mock_cursor
+
+        with mock.patch(
+            "opentelemetry.instrumentation.mysql.mysql.connector",
+            mock_connect_module,
+        ), mock.patch(
+            "opentelemetry.instrumentation.dbapi.util_version",
+            return_value="foobar",
+        ):
+            MySQLInstrumentor()._instrument()
+            cnx = mock_connect_module.connect(database="test")
+            cursor = cnx.cursor()
+            cursor.execute("Select 1;")
+            self.assertEqual(
+                mock_cursor.execute.call_args[0][0],
+                "Select 1;",
+            )
+
     @mock.patch("mysql.connector.connect")
     # pylint: disable=unused-argument
     def test_custom_tracer_provider(self, mock_connect):
@@ -102,6 +238,134 @@ def test_instrument_connection_no_op_tracer_provider(self, mock_connect):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 0)
 
+    @mock.patch("opentelemetry.instrumentation.mysql.DatabaseApiIntegration")
+    @mock.patch("mysql.connector.connect")
+    # pylint: disable=unused-argument
+    def test_instrument_connection_enable_commenter_dbapi_kwargs(
+        self,
+        mock_connect,
+        mock_mysql_dbapi,
+    ):
+        cnx, query = connect_and_execute_query()
+        cnx = MySQLInstrumentor().instrument_connection(
+            cnx,
+            enable_commenter=True,
+            commenter_options={"foo": True},
+        )
+        cursor = cnx.cursor()
+        cursor.execute(query)
+        kwargs = mock_mysql_dbapi.call_args[1]
+        self.assertEqual(kwargs["enable_commenter"], True)
+        self.assertEqual(kwargs["commenter_options"], {"foo": True})
+
+    def test_instrument_connection_with_dbapi_sqlcomment_enabled(self):
+        mock_connect_module = mock.MagicMock(
+            __name__="mysql.connector",
+            __version__="foobar",
+            threadsafety="123",
+            apilevel="123",
+            paramstyle="test",
+        )
+        mock_cursor = mock_connect_module.connect().cursor()
+        mock_cursor._cnx._cmysql.get_client_info.return_value = "foobaz"
+        mock_connection = mock.MagicMock()
+        mock_connection.cursor.return_value = mock_cursor
+
+        with mock.patch(
+            "opentelemetry.instrumentation.mysql.mysql.connector",
+            mock_connect_module,
+        ), mock.patch(
+            "opentelemetry.instrumentation.dbapi.util_version",
+            return_value="foobar",
+        ):
+            cnx_proxy = MySQLInstrumentor().instrument_connection(
+                mock_connection,
+                enable_commenter=True,
+            )
+            cnx_proxy.cursor().execute("Select 1;")
+
+            spans_list = self.memory_exporter.get_finished_spans()
+            span = spans_list[0]
+            span_id = format(span.get_span_context().span_id, "016x")
+            trace_id = format(span.get_span_context().trace_id, "032x")
+            self.assertEqual(
+                mock_cursor.execute.call_args[0][0],
+                f"Select 1 /*db_driver='mysql.connector%%3Afoobar',dbapi_level='123',dbapi_threadsafety='123',driver_paramstyle='test',mysql_client_version='foobaz',traceparent='00-{trace_id}-{span_id}-01'*/;",
+            )
+
+    def test_instrument_connection_with_dbapi_sqlcomment_enabled_with_options(
+        self,
+    ):
+        mock_connect_module = mock.MagicMock(
+            __name__="mysql.connector",
+            __version__="foobar",
+            threadsafety="123",
+            apilevel="123",
+            paramstyle="test",
+        )
+        mock_cursor = mock_connect_module.connect().cursor()
+        mock_cursor._cnx._cmysql.get_client_info.return_value = "foobaz"
+        mock_connection = mock.MagicMock()
+        mock_connection.cursor.return_value = mock_cursor
+
+        with mock.patch(
+            "opentelemetry.instrumentation.mysql.mysql.connector",
+            mock_connect_module,
+        ), mock.patch(
+            "opentelemetry.instrumentation.dbapi.util_version",
+            return_value="foobar",
+        ):
+            cnx_proxy = MySQLInstrumentor().instrument_connection(
+                mock_connection,
+                enable_commenter=True,
+                commenter_options={
+                    "dbapi_level": False,
+                    "dbapi_threadsafety": True,
+                    "driver_paramstyle": False,
+                },
+            )
+            cnx_proxy.cursor().execute("Select 1;")
+
+            spans_list = self.memory_exporter.get_finished_spans()
+            span = spans_list[0]
+            span_id = format(span.get_span_context().span_id, "016x")
+            trace_id = format(span.get_span_context().trace_id, "032x")
+            self.assertEqual(
+                mock_cursor.execute.call_args[0][0],
+                f"Select 1 /*db_driver='mysql.connector%%3Afoobar',dbapi_threadsafety='123',mysql_client_version='foobaz',traceparent='00-{trace_id}-{span_id}-01'*/;",
+            )
+
+    def test_instrument_connection_with_dbapi_sqlcomment_not_enabled_default(
+        self,
+    ):
+        mock_connect_module = mock.MagicMock(
+            __name__="mysql.connector",
+            __version__="foobar",
+            threadsafety="123",
+            apilevel="123",
+            paramstyle="test",
+        )
+        mock_cursor = mock_connect_module.connect().cursor()
+        mock_cursor._cnx._cmysql.get_client_info.return_value = "foobaz"
+        mock_connection = mock.MagicMock()
+        mock_connection.cursor.return_value = mock_cursor
+
+        with mock.patch(
+            "opentelemetry.instrumentation.mysql.mysql.connector",
+            mock_connect_module,
+        ), mock.patch(
+            "opentelemetry.instrumentation.dbapi.util_version",
+            return_value="foobar",
+        ):
+            cnx_proxy = MySQLInstrumentor().instrument_connection(
+                mock_connection,
+            )
+            cnx_proxy.cursor().execute("Select 1;")
+            self.assertEqual(
+                mock_cursor.execute.call_args[0][0],
+                "Select 1;",
+            )
+
     @mock.patch("mysql.connector.connect")
     # pylint: disable=unused-argument
     def test_uninstrument_connection(self, mock_connect):
@@ -117,3 +381,87 @@ def test_uninstrument_connection(self, mock_connect):
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
+
+    def test_get_traced_cursor_proxy_with_enable_commenter_cursor(self):
+        mock_cursor = mock.Mock()
+        mock_dbapiint = mock.Mock()
+        mock_enable = mock.Mock()
+        proxy = get_traced_cursor_proxy(
+            mock_cursor,
+            mock_dbapiint,
+            mock_enable,
+        )
+        self.assertIs(proxy.__wrapped__, mock_cursor)
+        self.assertIs(proxy._cursor_tracer._db_api_integration, mock_dbapiint)
+        self.assertIs(proxy._cursor_tracer._commenter_enabled, mock_enable)
+
+    def test_get_traced_connection_proxy_not_dbapi_enable_commenter(self):
+        mock_connection = mock.Mock()
+        mock_dbapiint = mock.Mock()
+        mock_dbapiint.enable_commenter = False
+        cnx_proxy = get_traced_connection_proxy(
+            mock_connection,
+            mock_dbapiint,
+        )
+        self.assertIs(cnx_proxy.__wrapped__, mock_connection)
+        cur_proxy = cnx_proxy.cursor()
+        self.assertIs(
+            cur_proxy._cursor_tracer._db_api_integration, mock_dbapiint
+        )
+        self.assertFalse(cur_proxy._cursor_tracer._commenter_enabled)
+
+    def test_get_traced_connection_proxy_dbapi_enable_commenter_none_prepared_cursor(
+        self,
+    ):
+        mock_connection = mock.Mock()
+        mock_dbapiint = mock.Mock()
+        mock_dbapiint.enable_commenter = True
+        cnx_proxy = get_traced_connection_proxy(
+            mock_connection,
+            mock_dbapiint,
+        )
+        self.assertIs(cnx_proxy.__wrapped__, mock_connection)
+        cur_proxy = cnx_proxy.cursor()
+        self.assertIs(
+            cur_proxy._cursor_tracer._db_api_integration, mock_dbapiint
+        )
+        self.assertTrue(cur_proxy._cursor_tracer._commenter_enabled)
+
+    def test_get_traced_connection_proxy_dbapi_enable_commenter_not_prepared_cursor(
+        self,
+    ):
+        mock_connection = mock.Mock()
+        mock_dbapiint = mock.Mock()
+        mock_dbapiint.enable_commenter = True
+        cnx_proxy = get_traced_connection_proxy(
+            mock_connection,
+            mock_dbapiint,
+        )
+        self.assertIs(cnx_proxy.__wrapped__, mock_connection)
+        cur_proxy = cnx_proxy.cursor(prepared=False)
+        self.assertIs(
+            cur_proxy._cursor_tracer._db_api_integration, mock_dbapiint
+        )
+        self.assertTrue(cur_proxy._cursor_tracer._commenter_enabled)
+
+    def test_get_traced_connection_proxy_dbapi_enable_commenter_prepared_cursor(
+        self,
+    ):
+        mock_connection = mock.Mock()
+        mock_dbapiint = mock.Mock()
+        mock_dbapiint.enable_commenter = True
+        mock_dbapiint.connect_module.__name__ = "foo-module"
+        cnx_proxy = get_traced_connection_proxy(
+            mock_connection,
+            mock_dbapiint,
+        )
+        self.assertIs(cnx_proxy.__wrapped__, mock_connection)
+        cur_proxy = cnx_proxy.cursor(prepared=True)
+        self.assertIs(
+            cur_proxy._cursor_tracer._db_api_integration, mock_dbapiint
+        )
+        self.assertFalse(cur_proxy._cursor_tracer._commenter_enabled)
+        assert (
+            self.caplog.records[0].getMessage()
+            == "sqlcomment is not supported for query statements executed by cursors with native prepared statement support. Disabling sqlcommenting for instrumentation of foo-module."
+        )