diff --git a/pom.xml b/pom.xml
index dc8eafd64..759ad46a9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -62,6 +62,8 @@
         <spring-context-version>5.3.2</spring-context-version>
         <spring-core-version>5.3.2</spring-core-version>
         <mule-spring-module-version>1.3.7</mule-spring-module-version>
+        <muleSdkCompatibilityApiVersion>1.0.0-SNAPSHOT</muleSdkCompatibilityApiVersion>
+        <muleSdkApiVersion>1.0.0-SNAPSHOT</muleSdkApiVersion>
     </properties>
 
     <dependencies>
@@ -174,6 +176,16 @@
             <version>${xmlunit.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mule.sdk</groupId>
+            <artifactId>mule-sdk-compatibility-api</artifactId>
+            <version>${muleSdkCompatibilityApiVersion}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mule.sdk</groupId>
+            <artifactId>mule-sdk-api</artifactId>
+            <version>${muleSdkApiVersion}</version>
+        </dependency>
 
         <!-- DB Drivers -->
         <dependency>
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/ConnectionBasedDbConnectionTracingMetadata.java b/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/ConnectionBasedDbConnectionTracingMetadata.java
new file mode 100644
index 000000000..6c12f482c
--- /dev/null
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/ConnectionBasedDbConnectionTracingMetadata.java
@@ -0,0 +1,109 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.domain.connection.datasource;
+
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getHostFrom;
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getProtocolFrom;
+import static org.mule.extension.db.internal.domain.connection.datasource.NoConnectionMetadata.getMetadata;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import static java.util.Optional.empty;
+import static java.util.Optional.ofNullable;
+
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.SQLException;
+import java.util.Optional;
+
+import org.slf4j.Logger;
+
+/**
+ * A {@link DbConnectionTracingMetadata} based on the connection metadata
+ */
+public class ConnectionBasedDbConnectionTracingMetadata implements DbConnectionTracingMetadata {
+
+  public static final String UNSET = "unset";
+  private static final Logger LOGGER = getLogger(ConnectionBasedDbConnectionTracingMetadata.class);
+  private final Connection connection;
+  private DatabaseMetaData metadata;
+
+  public ConnectionBasedDbConnectionTracingMetadata(Connection connection) {
+    this.connection = connection;
+  }
+
+  @Override
+  public String getDbSystem() {
+    try {
+      return getMetaData().getDatabaseProductName();
+    } catch (SQLException e) {
+      LOGGER.debug("Unable to retrieve connection metadata for tracing.", e);
+    }
+
+    return UNSET;
+  }
+
+  private DatabaseMetaData getMetaData() throws SQLException {
+    // In case the information cannot be retrieved it will not be present.
+    if (metadata == null) {
+      try {
+        metadata = connection.getMetaData();
+      } catch (Throwable e) {
+        LOGGER.debug("Unable to retrieve database metadata from connection", e);
+        metadata = getMetadata();
+      }
+
+    }
+
+    return metadata;
+  }
+
+  @Override
+  public String getConnectionString() {
+    try {
+      return getMetaData().getURL();
+    } catch (SQLException e) {
+      LOGGER.debug("Unable to retrieve connection metadata for tracing.", e);
+    }
+
+    return UNSET;
+  }
+
+  @Override
+  public String getUser() {
+    try {
+      return getMetaData().getUserName();
+    } catch (SQLException e) {
+      LOGGER.debug("Unable to retrieve connection metadata for tracing.", e);
+    }
+
+    return UNSET;
+  }
+
+  @Override
+  public Optional<String> getPeerName() {
+    try {
+      return ofNullable(getHostFrom(getMetaData().getURL()));
+    } catch (SQLException e) {
+      LOGGER.debug("Unable to retrieve connection metadata for tracing.", e);
+    }
+
+    return empty();
+  }
+
+  @Override
+  public Optional<String> getPeerTransport() {
+    try {
+      return ofNullable(getProtocolFrom(getMetaData().getURL()));
+    } catch (SQLException e) {
+      LOGGER.debug("Unable to retrieve connection metadata for tracing.", e);
+    }
+
+    return empty();
+  }
+}
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/DbDataSourceReferenceConnectionProvider.java b/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/DbDataSourceReferenceConnectionProvider.java
index e0c66e3f6..a256c41b3 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/DbDataSourceReferenceConnectionProvider.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/DbDataSourceReferenceConnectionProvider.java
@@ -12,6 +12,7 @@
 
 import javax.sql.DataSource;
 
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
 import org.mule.db.commons.internal.domain.connection.DbConnection;
 import org.mule.db.commons.internal.domain.connection.datasource.DataSourceReferenceConnectionProvider;
 import org.mule.db.commons.internal.domain.type.ResolvedDbType;
@@ -41,11 +42,12 @@ public class DbDataSourceReferenceConnectionProvider extends DataSourceReference
 
   @Override
   protected DbConnection createDbConnection(Connection connection) throws Exception {
+    DbConnectionTracingMetadata dbConnectionTracingMetadata = new ConnectionBasedDbConnectionTracingMetadata(connection);
     if (isOracle(connection)) {
       return new OracleDbConnection(connection, super.resolveCustomTypes(), resolvedDbTypesCache,
-                                    super.getCacheQueryTemplateSize());
+                                    super.getCacheQueryTemplateSize(), dbConnectionTracingMetadata);
     } else {
-      return super.createDbConnection(connection);
+      return super.createDbConnection(connection, dbConnectionTracingMetadata);
     }
   }
 
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/NoConnectionMetadata.java b/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/NoConnectionMetadata.java
new file mode 100644
index 000000000..9404979b5
--- /dev/null
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/datasource/NoConnectionMetadata.java
@@ -0,0 +1,920 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.domain.connection.datasource;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.RowIdLifetime;
+import java.sql.SQLException;
+
+/**
+ * A {@link DatabaseMetaData} to use when no metadata is available.
+ */
+public class NoConnectionMetadata implements DatabaseMetaData {
+
+  public static final String UNSET = "unset";
+
+  private static final NoConnectionMetadata INSTANCE = new NoConnectionMetadata();
+
+  private NoConnectionMetadata() {}
+
+  public static DatabaseMetaData getMetadata() {
+    return INSTANCE;
+  }
+
+  @Override
+  public boolean allProceduresAreCallable() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean allTablesAreSelectable() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public String getURL() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getUserName() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public boolean isReadOnly() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean nullsAreSortedHigh() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean nullsAreSortedLow() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean nullsAreSortedAtStart() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean nullsAreSortedAtEnd() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public String getDatabaseProductName() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getDatabaseProductVersion() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getDriverName() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getDriverVersion() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public int getDriverMajorVersion() {
+    return 0;
+  }
+
+  @Override
+  public int getDriverMinorVersion() {
+    return 0;
+  }
+
+  @Override
+  public boolean usesLocalFiles() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean usesLocalFilePerTable() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsMixedCaseIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean storesUpperCaseIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean storesLowerCaseIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean storesMixedCaseIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsMixedCaseQuotedIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean storesUpperCaseQuotedIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean storesLowerCaseQuotedIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean storesMixedCaseQuotedIdentifiers() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public String getIdentifierQuoteString() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getSQLKeywords() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getNumericFunctions() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getStringFunctions() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getSystemFunctions() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getTimeDateFunctions() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getSearchStringEscape() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getExtraNameCharacters() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public boolean supportsAlterTableWithAddColumn() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsAlterTableWithDropColumn() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsColumnAliasing() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean nullPlusNonNullIsNull() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsConvert() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsConvert(int fromType, int toType) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsTableCorrelationNames() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsDifferentTableCorrelationNames() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsExpressionsInOrderBy() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsOrderByUnrelated() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsGroupBy() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsGroupByUnrelated() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsGroupByBeyondSelect() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsLikeEscapeClause() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsMultipleResultSets() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsMultipleTransactions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsNonNullableColumns() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsMinimumSQLGrammar() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsCoreSQLGrammar() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsExtendedSQLGrammar() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsANSI92EntryLevelSQL() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsANSI92IntermediateSQL() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsANSI92FullSQL() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsIntegrityEnhancementFacility() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsOuterJoins() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsFullOuterJoins() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsLimitedOuterJoins() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public String getSchemaTerm() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getProcedureTerm() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public String getCatalogTerm() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public boolean isCatalogAtStart() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public String getCatalogSeparator() throws SQLException {
+    return UNSET;
+  }
+
+  @Override
+  public boolean supportsSchemasInDataManipulation() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSchemasInProcedureCalls() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSchemasInTableDefinitions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSchemasInIndexDefinitions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSchemasInPrivilegeDefinitions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsCatalogsInDataManipulation() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsCatalogsInProcedureCalls() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsCatalogsInTableDefinitions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsCatalogsInIndexDefinitions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsCatalogsInPrivilegeDefinitions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsPositionedDelete() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsPositionedUpdate() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSelectForUpdate() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsStoredProcedures() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSubqueriesInComparisons() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSubqueriesInExists() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSubqueriesInIns() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsSubqueriesInQuantifieds() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsCorrelatedSubqueries() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsUnion() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsUnionAll() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsOpenCursorsAcrossCommit() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsOpenCursorsAcrossRollback() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsOpenStatementsAcrossCommit() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsOpenStatementsAcrossRollback() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public int getMaxBinaryLiteralLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxCharLiteralLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxColumnNameLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxColumnsInGroupBy() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxColumnsInIndex() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxColumnsInOrderBy() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxColumnsInSelect() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxColumnsInTable() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxConnections() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxCursorNameLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxIndexLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxSchemaNameLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxProcedureNameLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxCatalogNameLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxRowSize() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public boolean doesMaxRowSizeIncludeBlobs() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public int getMaxStatementLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxStatements() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxTableNameLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxTablesInSelect() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getMaxUserNameLength() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getDefaultTransactionIsolation() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public boolean supportsTransactions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsTransactionIsolationLevel(int level) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsDataDefinitionAndDataManipulationTransactions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsDataManipulationTransactionsOnly() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean dataDefinitionCausesTransactionCommit() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean dataDefinitionIgnoredInTransactions() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public ResultSet getProcedures(String catalog, String schemaPattern, String procedureNamePattern) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getProcedureColumns(String catalog, String schemaPattern, String procedureNamePattern,
+                                       String columnNamePattern)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getTables(String catalog, String schemaPattern, String tableNamePattern, String[] types) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getSchemas() throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getCatalogs() throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getTableTypes() throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getColumnPrivileges(String catalog, String schema, String table, String columnNamePattern)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getTablePrivileges(String catalog, String schemaPattern, String tableNamePattern) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getBestRowIdentifier(String catalog, String schema, String table, int scope, boolean nullable)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getVersionColumns(String catalog, String schema, String table) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getPrimaryKeys(String catalog, String schema, String table) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getImportedKeys(String catalog, String schema, String table) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getExportedKeys(String catalog, String schema, String table) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getCrossReference(String parentCatalog, String parentSchema, String parentTable, String foreignCatalog,
+                                     String foreignSchema, String foreignTable)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getTypeInfo() throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getIndexInfo(String catalog, String schema, String table, boolean unique, boolean approximate)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public boolean supportsResultSetType(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsResultSetConcurrency(int type, int concurrency) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean ownUpdatesAreVisible(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean ownDeletesAreVisible(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean ownInsertsAreVisible(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean othersUpdatesAreVisible(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean othersDeletesAreVisible(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean othersInsertsAreVisible(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean updatesAreDetected(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean deletesAreDetected(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean insertsAreDetected(int type) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsBatchUpdates() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public ResultSet getUDTs(String catalog, String schemaPattern, String typeNamePattern, int[] types) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public Connection getConnection() throws SQLException {
+    return null;
+  }
+
+  @Override
+  public boolean supportsSavepoints() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsNamedParameters() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsMultipleOpenResults() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsGetGeneratedKeys() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public ResultSet getSuperTypes(String catalog, String schemaPattern, String typeNamePattern) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getSuperTables(String catalog, String schemaPattern, String tableNamePattern) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getAttributes(String catalog, String schemaPattern, String typeNamePattern, String attributeNamePattern)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public boolean supportsResultSetHoldability(int holdability) throws SQLException {
+    return false;
+  }
+
+  @Override
+  public int getResultSetHoldability() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getDatabaseMajorVersion() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getDatabaseMinorVersion() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getJDBCMajorVersion() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getJDBCMinorVersion() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public int getSQLStateType() throws SQLException {
+    return 0;
+  }
+
+  @Override
+  public boolean locatorsUpdateCopy() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean supportsStatementPooling() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public RowIdLifetime getRowIdLifetime() throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public boolean supportsStoredFunctionsUsingCallSyntax() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public boolean autoCommitFailureClosesAllResultSets() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public ResultSet getClientInfoProperties() throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getFunctions(String catalog, String schemaPattern, String functionNamePattern) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getFunctionColumns(String catalog, String schemaPattern, String functionNamePattern, String columnNamePattern)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public ResultSet getPseudoColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern)
+      throws SQLException {
+    return null;
+  }
+
+  @Override
+  public boolean generatedKeyAlwaysReturned() throws SQLException {
+    return false;
+  }
+
+  @Override
+  public <T> T unwrap(Class<T> iface) throws SQLException {
+    return null;
+  }
+
+  @Override
+  public boolean isWrapperFor(Class<?> iface) throws SQLException {
+    return false;
+  }
+}
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnection.java b/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnection.java
index 0ff903e62..bacff25e6 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnection.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnection.java
@@ -6,8 +6,11 @@
  */
 package org.mule.extension.db.internal.domain.connection.derby;
 
+import static org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata.getNotResolvedDbConnectionTracingMetadata;
+
 import org.mule.db.commons.internal.domain.connection.DefaultDbConnection;
 import org.mule.db.commons.internal.domain.type.DbType;
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
 
 import java.sql.Connection;
 import java.util.List;
@@ -20,7 +23,12 @@
 public class DerbyConnection extends DefaultDbConnection {
 
   DerbyConnection(Connection connection, List<DbType> dbTypes, long cacheQueryTemplateSize) {
-    super(connection, dbTypes, cacheQueryTemplateSize);
+    this(connection, dbTypes, cacheQueryTemplateSize, getNotResolvedDbConnectionTracingMetadata());
+  }
+
+  DerbyConnection(Connection connection, List<DbType> dbTypes, long cacheQueryTemplateSize,
+                  DbConnectionTracingMetadata dbConnectionTracingMetadata) {
+    super(connection, dbTypes, cacheQueryTemplateSize, dbConnectionTracingMetadata);
   }
 
   // We are disabling content streaming for Derby because of a incompatibility between the connector logic and the
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnectionProvider.java b/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnectionProvider.java
index 6b0373094..1a762f07e 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnectionProvider.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/derby/DerbyConnectionProvider.java
@@ -27,6 +27,7 @@
 import org.mule.db.commons.internal.domain.connection.DataSourceConfig;
 import org.mule.db.commons.internal.domain.connection.DbConnection;
 import org.mule.db.commons.internal.domain.connection.DbConnectionProvider;
+import org.mule.extension.db.internal.domain.connection.derby.tracing.DerbyDbConnectionTracingMetadata;
 import org.mule.runtime.extension.api.annotation.Alias;
 import org.mule.runtime.extension.api.annotation.ExternalLib;
 import org.mule.runtime.extension.api.annotation.param.ParameterGroup;
@@ -64,7 +65,8 @@ public java.util.Optional<DataSourceConfig> getDataSourceConfig() {
 
   @Override
   protected DbConnection createDbConnection(Connection connection) throws Exception {
-    return new DerbyConnection(connection, resolveCustomTypes(), super.getCacheQueryTemplateSize());
+    return new DerbyConnection(connection, resolveCustomTypes(), super.getCacheQueryTemplateSize(),
+                               new DerbyDbConnectionTracingMetadata(derbyParameters));
   }
 
   @Override
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/derby/tracing/DerbyDbConnectionTracingMetadata.java b/src/main/java/org/mule/extension/db/internal/domain/connection/derby/tracing/DerbyDbConnectionTracingMetadata.java
new file mode 100644
index 000000000..4785991c1
--- /dev/null
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/derby/tracing/DerbyDbConnectionTracingMetadata.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.domain.connection.derby.tracing;
+
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getHostFrom;
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getProtocolFrom;
+
+import static java.util.Optional.ofNullable;
+
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
+import org.mule.extension.db.internal.domain.connection.derby.DerbyConnectionParameters;
+
+import java.util.Optional;
+
+/**
+ * a {@link DbConnectionTracingMetadata} for derby.
+ */
+public class DerbyDbConnectionTracingMetadata implements DbConnectionTracingMetadata {
+
+  public static final String DERBY_DB_SYSTEM = "DERBY";
+  private final DerbyConnectionParameters derbyParameters;
+
+  public DerbyDbConnectionTracingMetadata(DerbyConnectionParameters derbyParameters) {
+    this.derbyParameters = derbyParameters;
+  }
+
+  @Override
+  public String getDbSystem() {
+    return DERBY_DB_SYSTEM;
+  }
+
+  @Override
+  public String getConnectionString() {
+    return derbyParameters.getUrl();
+  }
+
+  @Override
+  public String getUser() {
+    return derbyParameters.getUser();
+  }
+
+  @Override
+  public Optional<String> getPeerName() {
+    return ofNullable(getHostFrom(derbyParameters.getUrl()));
+  }
+
+  @Override
+  public Optional<String> getPeerTransport() {
+    return ofNullable(getProtocolFrom(derbyParameters.getUser()));
+  }
+}
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/generic/DbGenericConnectionProvider.java b/src/main/java/org/mule/extension/db/internal/domain/connection/generic/DbGenericConnectionProvider.java
index 306215754..ed4ca590f 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/generic/DbGenericConnectionProvider.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/generic/DbGenericConnectionProvider.java
@@ -13,6 +13,7 @@
 import org.mule.db.commons.internal.domain.connection.DbConnection;
 import org.mule.db.commons.internal.domain.connection.generic.GenericConnectionProvider;
 import org.mule.db.commons.internal.domain.type.ResolvedDbType;
+import org.mule.extension.db.internal.domain.connection.datasource.ConnectionBasedDbConnectionTracingMetadata;
 import org.mule.extension.db.internal.domain.connection.oracle.OracleDbConnection;
 import org.mule.runtime.extension.api.annotation.Alias;
 import org.mule.runtime.extension.api.annotation.ExternalLib;
@@ -38,10 +39,13 @@ public class DbGenericConnectionProvider extends GenericConnectionProvider {
 
   @Override
   protected DbConnection createDbConnection(Connection connection) throws Exception {
+    ConnectionBasedDbConnectionTracingMetadata connectionTracingMetadata =
+        new ConnectionBasedDbConnectionTracingMetadata(connection);
     if (isOracle(connection)) {
-      return new OracleDbConnection(connection, resolveCustomTypes(), resolvedDbTypesCache, super.getCacheQueryTemplateSize());
+      return new OracleDbConnection(connection, resolveCustomTypes(), resolvedDbTypesCache, super.getCacheQueryTemplateSize(),
+                                    connectionTracingMetadata);
     } else {
-      return super.createDbConnection(connection);
+      return super.createDbConnection(connection, connectionTracingMetadata);
     }
   }
 
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/mysql/MySqlConnectionProvider.java b/src/main/java/org/mule/extension/db/internal/domain/connection/mysql/MySqlConnectionProvider.java
index 319feaabc..b19b3c50f 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/mysql/MySqlConnectionProvider.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/mysql/MySqlConnectionProvider.java
@@ -23,6 +23,8 @@
 import org.mule.db.commons.api.exception.connection.DbError;
 import org.mule.db.commons.internal.domain.connection.DataSourceConfig;
 import org.mule.db.commons.internal.domain.connection.DbConnectionProvider;
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
+import org.mule.extension.db.internal.domain.connection.mysql.tracing.MySqlConnectionTracingMetadata;
 import org.mule.runtime.extension.api.annotation.Alias;
 import org.mule.runtime.extension.api.annotation.ExternalLib;
 import org.mule.runtime.extension.api.annotation.param.ParameterGroup;
@@ -72,4 +74,8 @@ public java.util.Optional<DbError> getDbVendorErrorType(SQLException e) {
     return empty();
   }
 
+  @Override
+  protected DbConnectionTracingMetadata getDbConnectionTracingMetadata() {
+    return new MySqlConnectionTracingMetadata(mySqlParameters);
+  }
 }
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/mysql/tracing/MySqlConnectionTracingMetadata.java b/src/main/java/org/mule/extension/db/internal/domain/connection/mysql/tracing/MySqlConnectionTracingMetadata.java
new file mode 100644
index 000000000..8b705bb2b
--- /dev/null
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/mysql/tracing/MySqlConnectionTracingMetadata.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.domain.connection.mysql.tracing;
+
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
+import org.mule.extension.db.internal.domain.connection.mysql.MySqlConnectionParameters;
+
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getHostFrom;
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getProtocolFrom;
+
+import static java.util.Optional.ofNullable;
+
+import java.util.Optional;
+
+/**
+ * a {@link DbConnectionTracingMetadata} for mysql.
+ */
+public class MySqlConnectionTracingMetadata implements DbConnectionTracingMetadata {
+
+  public static final String MYSQL = "mysql";
+  private final MySqlConnectionParameters mySqlParameter;
+
+  public MySqlConnectionTracingMetadata(MySqlConnectionParameters mySqlParameters) {
+    this.mySqlParameter = mySqlParameters;
+  }
+
+  @Override
+  public String getDbSystem() {
+    return MYSQL;
+  }
+
+  @Override
+  public String getConnectionString() {
+    return mySqlParameter.getUrl();
+  }
+
+  @Override
+  public String getUser() {
+    return mySqlParameter.getUser();
+  }
+
+  @Override
+  public Optional<String> getPeerName() {
+    return ofNullable(getHostFrom(mySqlParameter.getUrl()));
+  }
+
+  @Override
+  public Optional<String> getPeerTransport() {
+    return ofNullable(getProtocolFrom(mySqlParameter.getUrl()));
+  }
+}
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnection.java b/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnection.java
index a7ba6e61f..a8298bb16 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnection.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnection.java
@@ -10,6 +10,7 @@
 import static java.util.Optional.empty;
 import static java.util.Optional.ofNullable;
 import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata.getNotResolvedDbConnectionTracingMetadata;
 import static org.mule.extension.db.internal.domain.connection.oracle.OracleConnectionUtils.getOwnerFrom;
 import static org.mule.extension.db.internal.domain.connection.oracle.OracleConnectionUtils.getTypeSimpleName;
 
@@ -26,6 +27,7 @@
 import org.mule.extension.db.internal.domain.connection.oracle.types.OracleSQLXMLType;
 import org.mule.extension.db.internal.domain.connection.oracle.types.OracleSYSXMLType;
 import org.mule.extension.db.internal.domain.connection.oracle.types.OracleXMLType;
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
 
 import java.sql.Array;
 import java.sql.Connection;
@@ -77,10 +79,16 @@ public class OracleDbConnection extends DefaultDbConnection {
 
   private final Map<String, Map<Integer, ResolvedDbType>> resolvedDbTypesCache;
 
-
   public OracleDbConnection(Connection jdbcConnection, List<DbType> customDataTypes,
                             Map<String, Map<Integer, ResolvedDbType>> resolvedDbTypesCache, long cacheQueryTemplateSize) {
-    super(jdbcConnection, customDataTypes, cacheQueryTemplateSize);
+    this(jdbcConnection, customDataTypes, resolvedDbTypesCache, cacheQueryTemplateSize,
+         getNotResolvedDbConnectionTracingMetadata());
+  }
+
+  public OracleDbConnection(Connection jdbcConnection, List<DbType> customDataTypes,
+                            Map<String, Map<Integer, ResolvedDbType>> resolvedDbTypesCache, long cacheQueryTemplateSize,
+                            DbConnectionTracingMetadata dbConnectionTracingMetadata) {
+    super(jdbcConnection, customDataTypes, cacheQueryTemplateSize, dbConnectionTracingMetadata);
     this.resolvedDbTypesCache = resolvedDbTypesCache;
   }
 
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnectionProvider.java b/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnectionProvider.java
index 09ea28576..48dcff936 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnectionProvider.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/oracle/OracleDbConnectionProvider.java
@@ -7,10 +7,6 @@
 
 package org.mule.extension.db.internal.domain.connection.oracle;
 
-import static java.util.Optional.empty;
-import static java.util.Optional.of;
-import static java.util.Optional.ofNullable;
-
 import static org.mule.db.commons.api.exception.connection.DbError.CANNOT_REACH;
 import static org.mule.db.commons.api.exception.connection.DbError.INVALID_CREDENTIALS;
 import static org.mule.db.commons.api.exception.connection.DbError.INVALID_DATABASE;
@@ -21,6 +17,10 @@
 import static org.mule.runtime.extension.api.annotation.param.ParameterGroup.CONNECTION;
 import static org.mule.extension.db.internal.util.MigrationUtils.mapDataSourceConfig;
 
+import static java.util.Optional.empty;
+import static java.util.Optional.of;
+import static java.util.Optional.ofNullable;
+
 import javax.sql.DataSource;
 import java.sql.Connection;
 import java.sql.SQLException;
@@ -30,6 +30,7 @@
 
 import org.mule.db.commons.api.exception.connection.DbError;
 import org.mule.db.commons.internal.domain.connection.DataSourceConfig;
+import org.mule.db.commons.internal.domain.connection.GenericConfigDbConnectionTracingMetadata;
 import org.mule.db.commons.internal.domain.connection.DbConnection;
 import org.mule.db.commons.internal.domain.connection.DbConnectionProvider;
 import org.mule.db.commons.internal.domain.type.ResolvedDbType;
@@ -79,7 +80,8 @@ public java.util.Optional<DataSourceConfig> getDataSourceConfig() {
   @Override
   protected DbConnection createDbConnection(Connection connection) throws Exception {
     return new OracleDbConnection(connection, super.resolveCustomTypes(), resolvedDbTypesCache,
-                                  super.getCacheQueryTemplateSize());
+                                  super.getCacheQueryTemplateSize(),
+                                  new GenericConfigDbConnectionTracingMetadata(getDataSourceConfig().get()));
   }
 
   @Override
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnection.java b/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnection.java
index bb501c0d0..77928f9dd 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnection.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnection.java
@@ -6,11 +6,14 @@
  */
 package org.mule.extension.db.internal.domain.connection.sqlserver;
 
+import static org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata.getNotResolvedDbConnectionTracingMetadata;
+
 import org.mule.db.commons.internal.domain.connection.DbConnection;
 import org.mule.db.commons.internal.domain.connection.DefaultDbConnection;
 import org.mule.extension.db.internal.domain.connection.sqlserver.types.SqlServerBinaryDbType;
 import org.mule.extension.db.internal.domain.connection.sqlserver.types.SqlServerVarBinaryDbType;
 import org.mule.db.commons.internal.domain.type.DbType;
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
 
 import java.sql.Connection;
 import java.util.ArrayList;
@@ -24,7 +27,12 @@
 public class SqlServerConnection extends DefaultDbConnection {
 
   SqlServerConnection(Connection jdbcConnection, List<DbType> customDataTypes, long cacheQueryTemplateSize) {
-    super(jdbcConnection, customDataTypes, cacheQueryTemplateSize);
+    super(jdbcConnection, customDataTypes, cacheQueryTemplateSize, getNotResolvedDbConnectionTracingMetadata());
+  }
+
+  SqlServerConnection(Connection jdbcConnection, List<DbType> customDataTypes, long cacheQueryTemplateSize,
+                      DbConnectionTracingMetadata dbConnectionTracingMetadata) {
+    super(jdbcConnection, customDataTypes, cacheQueryTemplateSize, dbConnectionTracingMetadata);
   }
 
   /**
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnectionProvider.java b/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnectionProvider.java
index a3b384554..f89bc02df 100644
--- a/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnectionProvider.java
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/SqlServerConnectionProvider.java
@@ -23,10 +23,12 @@
 import javax.sql.DataSource;
 import java.sql.Connection;
 import java.sql.SQLException;
+
 import org.mule.db.commons.api.exception.connection.DbError;
 import org.mule.db.commons.internal.domain.connection.DataSourceConfig;
 import org.mule.db.commons.internal.domain.connection.DbConnection;
 import org.mule.db.commons.internal.domain.connection.DbConnectionProvider;
+import org.mule.extension.db.internal.domain.connection.sqlserver.tracing.SqlServerDbConnectionTracingMetadata;
 import org.mule.runtime.extension.api.annotation.Alias;
 import org.mule.runtime.extension.api.annotation.ExternalLib;
 import org.mule.runtime.extension.api.annotation.param.ParameterGroup;
@@ -56,7 +58,8 @@ public class SqlServerConnectionProvider extends DbConnectionProvider {
 
   @Override
   protected DbConnection createDbConnection(Connection connection) throws Exception {
-    return new SqlServerConnection(connection, super.resolveCustomTypes(), super.getCacheQueryTemplateSize());
+    return new SqlServerConnection(connection, super.resolveCustomTypes(), super.getCacheQueryTemplateSize(),
+                                   new SqlServerDbConnectionTracingMetadata(connectionParameters));
   }
 
   @Override
diff --git a/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/tracing/SqlServerDbConnectionTracingMetadata.java b/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/tracing/SqlServerDbConnectionTracingMetadata.java
new file mode 100644
index 000000000..357f6518a
--- /dev/null
+++ b/src/main/java/org/mule/extension/db/internal/domain/connection/sqlserver/tracing/SqlServerDbConnectionTracingMetadata.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.domain.connection.sqlserver.tracing;
+
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
+import org.mule.extension.db.internal.domain.connection.sqlserver.SqlServerConnectionParameters;
+
+import java.util.Optional;
+
+import static java.util.Optional.ofNullable;
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getHostFrom;
+import static org.mule.db.commons.internal.domain.connection.ConnectionTracingMetadataUtils.getProtocolFrom;
+
+/**
+ * a {@link DbConnectionTracingMetadata} for sql server.
+ */
+public class SqlServerDbConnectionTracingMetadata implements DbConnectionTracingMetadata {
+
+  public static final String SQL_SERVER_DB_SYSTEM = "SQL_SERVER";
+  private final SqlServerConnectionParameters connectionParameters;
+
+  public SqlServerDbConnectionTracingMetadata(SqlServerConnectionParameters connectionParameters) {
+    this.connectionParameters = connectionParameters;
+  }
+
+  @Override
+  public String getDbSystem() {
+    return SQL_SERVER_DB_SYSTEM;
+  }
+
+  @Override
+  public String getConnectionString() {
+    return connectionParameters.getUrl();
+  }
+
+  @Override
+  public String getUser() {
+    return connectionParameters.getUser();
+  }
+
+  @Override
+  public Optional<String> getPeerName() {
+    return ofNullable(getHostFrom(connectionParameters.getUrl()));
+  }
+
+  @Override
+  public Optional<String> getPeerTransport() {
+    return ofNullable(getProtocolFrom(connectionParameters.getUrl()));
+  }
+}
diff --git a/src/main/java/org/mule/extension/db/internal/operation/DbBulkOperations.java b/src/main/java/org/mule/extension/db/internal/operation/DbBulkOperations.java
index 4a0c36c0a..9416f6605 100644
--- a/src/main/java/org/mule/extension/db/internal/operation/DbBulkOperations.java
+++ b/src/main/java/org/mule/extension/db/internal/operation/DbBulkOperations.java
@@ -8,6 +8,7 @@
 
 import org.mule.db.commons.AbstractDbConnector;
 import org.mule.db.commons.internal.domain.connection.DbConnection;
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
 import org.mule.db.commons.internal.domain.metadata.DbInputMetadataResolver;
 import org.mule.db.commons.internal.operation.BulkOperations;
 import org.mule.db.commons.internal.operation.OperationErrorTypeProvider;
@@ -25,13 +26,19 @@
 import org.mule.extension.db.api.param.BulkQueryDefinition;
 import org.mule.extension.db.api.param.BulkScript;
 import org.mule.extension.db.api.param.QuerySettings;
+import org.mule.sdk.api.runtime.parameter.CorrelationInfo;
+import org.mule.sdk.compatibility.api.utils.ForwardCompatibilityHelper;
 
+import javax.inject.Inject;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Map;
 
+import static org.mule.db.commons.internal.domain.query.QueryType.DELETE;
+import static org.mule.db.commons.internal.domain.query.QueryType.UPDATE;
 import static org.mule.db.commons.internal.operation.BaseDbOperations.QUERY_GROUP;
 import static org.mule.db.commons.internal.operation.BaseDbOperations.QUERY_SETTINGS;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.setAttributesForDbClientOperation;
 import static org.mule.extension.db.internal.util.MigrationUtils.mapBulkQueryDefinition;
 import static org.mule.extension.db.internal.util.MigrationUtils.mapBulkScript;
 import static org.mule.extension.db.internal.util.MigrationUtils.mapQuerySettings;
@@ -47,6 +54,9 @@ public class DbBulkOperations implements Initialisable {
 
   private BulkOperations bulkOperations;
 
+  @Inject
+  private java.util.Optional<ForwardCompatibilityHelper> forwardCompatibilityHelper;
+
   @Override
   public void initialise() throws InitialisationException {
     this.bulkOperations = new BulkOperations.Builder().build();
@@ -70,8 +80,15 @@ public int[] bulkInsert(@DisplayName("Input Parameters") @Content @Placement(
                           @ParameterGroup(name = QUERY_GROUP) BulkQueryDefinition query,
                           @Config AbstractDbConnector connector,
                           @Connection DbConnection connection,
+                          CorrelationInfo correlationInfo,
                           StreamingHelper streamingHelper)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        query.getSql());
+    });
     return bulkOperations.bulkInsert(bulkInputParameters, mapBulkQueryDefinition(query), connector, connection, streamingHelper);
   }
 
@@ -94,8 +111,16 @@ public int[] bulkUpdate(@DisplayName("Input Parameters") @Content @Placement(
                           @ParameterGroup(name = QUERY_GROUP) BulkQueryDefinition query,
                           @Config AbstractDbConnector connector,
                           @Connection DbConnection connection,
+                          CorrelationInfo correlationInfo,
                           StreamingHelper streamingHelper)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        UPDATE.name() + " " + dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        query.getSql());
+    });
     return bulkOperations.bulkUpdate(bulkInputParameters, mapBulkQueryDefinition(query), connector, connection, streamingHelper);
   }
 
@@ -117,8 +142,16 @@ public int[] bulkDelete(@DisplayName("Input Parameters") @Content @Placement(
                           @ParameterGroup(name = QUERY_GROUP) BulkQueryDefinition query,
                           @Config AbstractDbConnector connector,
                           @Connection DbConnection connection,
+                          CorrelationInfo correlationInfo,
                           StreamingHelper streamingHelper)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        DELETE.name() + " " + dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        query.getSql());
+    });
     return bulkOperations.bulkDelete(bulkInputParameters, mapBulkQueryDefinition(query), connector, connection, streamingHelper);
   }
 
@@ -135,8 +168,16 @@ public int[] bulkDelete(@DisplayName("Input Parameters") @Content @Placement(
    */
   public int[] executeScript(@ParameterGroup(name = QUERY_GROUP) BulkScript script,
                              @ParameterGroup(name = QUERY_SETTINGS) QuerySettings settings,
-                             @Connection DbConnection connection)
+                             @Connection DbConnection connection,
+                             CorrelationInfo correlationInfo)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        "<script>");
+    });
     return bulkOperations.executeScript(mapBulkScript(script), mapQuerySettings(settings), connection);
   }
 
diff --git a/src/main/java/org/mule/extension/db/internal/operation/DbDmlOperations.java b/src/main/java/org/mule/extension/db/internal/operation/DbDmlOperations.java
index 33b75318c..384817931 100644
--- a/src/main/java/org/mule/extension/db/internal/operation/DbDmlOperations.java
+++ b/src/main/java/org/mule/extension/db/internal/operation/DbDmlOperations.java
@@ -9,6 +9,7 @@
 import org.mule.db.commons.AbstractDbConnector;
 
 import org.mule.db.commons.internal.domain.connection.DbConnection;
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
 import org.mule.db.commons.internal.domain.metadata.QuerySingleMetadataResolver;
 import org.mule.db.commons.internal.domain.metadata.SelectMetadataResolver;
 import org.mule.db.commons.internal.domain.metadata.StoredProcedureMetadataResolver;
@@ -30,17 +31,24 @@
 import org.mule.runtime.extension.api.runtime.operation.FlowListener;
 import org.mule.runtime.extension.api.runtime.streaming.PagingProvider;
 import org.mule.runtime.extension.api.runtime.streaming.StreamingHelper;
+import org.mule.sdk.api.runtime.parameter.CorrelationInfo;
+import org.mule.sdk.compatibility.api.utils.ForwardCompatibilityHelper;
 
+import javax.inject.Inject;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
-import static org.mule.extension.db.internal.util.MigrationUtils.mapQueryDefinition;
-import static org.mule.extension.db.internal.util.MigrationUtils.mapStoredProcedureCall;
 import static org.mule.db.commons.internal.domain.autogeneratedkey.AutoGenerateKeysAttributes.AUTO_GENERATE_KEYS;
+import static org.mule.db.commons.internal.domain.query.QueryType.DELETE;
+import static org.mule.db.commons.internal.domain.query.QueryType.INSERT;
+import static org.mule.db.commons.internal.domain.query.QueryType.UPDATE;
 import static org.mule.db.commons.internal.operation.BaseDbOperations.QUERY_GROUP;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.setAttributesForDbClientOperation;
+import static org.mule.extension.db.internal.util.MigrationUtils.mapQueryDefinition;
+import static org.mule.extension.db.internal.util.MigrationUtils.mapStoredProcedureCall;
 import static org.mule.runtime.extension.api.annotation.param.display.Placement.ADVANCED_TAB;
 
 /**
@@ -58,6 +66,9 @@ public void initialise() throws InitialisationException {
     dmlOperations = new DmlOperations.Builder().build();
   }
 
+  @Inject
+  private java.util.Optional<ForwardCompatibilityHelper> forwardCompatibilityHelper;
+
   /**
    * Selects data from a database.
    *
@@ -75,6 +86,7 @@ public PagingProvider<DbConnection, Map<String, Object>> select(
                                                                       tab = ADVANCED_TAB) QueryDefinition query,
                                                                   @Config AbstractDbConnector connector,
                                                                   StreamingHelper streamingHelper,
+                                                                  CorrelationInfo correlationInfo,
                                                                   FlowListener flowListener)
       throws SQLException {
     return dmlOperations.select(mapQueryDefinition(query), connector, streamingHelper, flowListener);
@@ -96,8 +108,22 @@ public PagingProvider<DbConnection, Map<String, Object>> select(
   @OutputResolver(output = QuerySingleMetadataResolver.class)
   public Map<String, Object> querySingle(@ParameterGroup(name = QUERY_GROUP) @Placement(
       tab = ADVANCED_TAB) QueryDefinition query, @Config AbstractDbConnector connector, @Connection DbConnection connection,
+                                         CorrelationInfo correlationInfo,
                                          StreamingHelper streamingHelper)
       throws SQLException {
+
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      // We set the name because it is the most generic operation.
+      // We follow the semantic conventions in this part:
+      // "It is not recommended to attempt any client-side parsing of db.statement just to get these properties,
+      // they should only be used if the library being instrumented already provides them. When it?s otherwise
+      // impossible to get any meaningful span name, db.name or the tech-specific database name MAY be used."
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        query.getSql());
+    });
     return dmlOperations.querySingle(mapQueryDefinition(query), connector, connection, streamingHelper);
   }
 
@@ -116,8 +142,16 @@ public StatementResult insert(@ParameterGroup(name = QUERY_GROUP) @Placement(tab
                                     name = AUTO_GENERATE_KEYS) DbAutoGenerateKeysAttributes autoGenerateKeysAttributes,
                                 @Config AbstractDbConnector connector,
                                 @Connection DbConnection connection,
+                                CorrelationInfo correlationInfo,
                                 StreamingHelper streamingHelper)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        INSERT.name() + " " + dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        query.getSql());
+    });
     return new StatementResult(dmlOperations.insert(mapQueryDefinition(query), autoGenerateKeysAttributes, connector, connection,
                                                     streamingHelper));
   }
@@ -137,8 +171,16 @@ public StatementResult update(@ParameterGroup(name = QUERY_GROUP) QueryDefinitio
                                     name = AUTO_GENERATE_KEYS) DbAutoGenerateKeysAttributes autoGenerateKeysAttributes,
                                 @Config AbstractDbConnector connector,
                                 @Connection DbConnection connection,
+                                CorrelationInfo correlationInfo,
                                 StreamingHelper streamingHelper)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        UPDATE.name() + " " + dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        query.getSql());
+    });
     return new StatementResult(dmlOperations.update(mapQueryDefinition(query), autoGenerateKeysAttributes, connector, connection,
                                                     streamingHelper));
 
@@ -156,8 +198,16 @@ public StatementResult update(@ParameterGroup(name = QUERY_GROUP) QueryDefinitio
   public int delete(@ParameterGroup(name = QUERY_GROUP) QueryDefinition query,
                     @Config AbstractDbConnector connector,
                     @Connection DbConnection connection,
+                    CorrelationInfo correlationInfo,
                     StreamingHelper streamingHelper)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        DELETE.name() + " " + dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        query.getSql());
+    });
     return dmlOperations.delete(mapQueryDefinition(query), connector, connection, streamingHelper);
   }
 
@@ -183,8 +233,16 @@ public Map<String, Object> storedProcedure(@ParameterGroup(name = QUERY_GROUP) S
                                              @Config AbstractDbConnector connector,
                                              @Connection DbConnection connection,
                                              StreamingHelper streamingHelper,
+                                             CorrelationInfo correlationInfo,
                                              FlowListener flowListener)
       throws SQLException {
+    forwardCompatibilityHelper.ifPresent(fwh -> {
+      DbConnectionTracingMetadata dbConnectionTracingMetadata = connection.getDbConnectionTracingMetadata();
+      setAttributesForDbClientOperation(dbConnectionTracingMetadata,
+                                        dbConnectionTracingMetadata.getDbSystem(),
+                                        fwh.getDistributedTraceContextManager(correlationInfo),
+                                        call.getSql());
+    });
     return dmlOperations.storedProcedure(mapStoredProcedureCall(call), autoGenerateKeysAttributes, connector, connection,
                                          streamingHelper, flowListener);
   }
diff --git a/src/main/java/org/mule/extension/db/internal/operation/tracing/TracingUtils.java b/src/main/java/org/mule/extension/db/internal/operation/tracing/TracingUtils.java
new file mode 100644
index 000000000..ea619dd4e
--- /dev/null
+++ b/src/main/java/org/mule/extension/db/internal/operation/tracing/TracingUtils.java
@@ -0,0 +1,215 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.operation.tracing;
+
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
+import org.mule.sdk.api.runtime.source.DistributedTraceContextManager;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Utils for runtime tracing.
+ */
+public class TracingUtils {
+
+  public static final String SPAN_KIND_CLIENT_NAME = "CLIENT";
+
+  private static final Logger LOGGER = getLogger(TracingUtils.class);
+  static final String SPAN_KIND = "span.kind.override";
+  static final String DB_CONNECTION_STRING_KEY = "db.connection_string";
+  static final String DB_SYSTEM_KEY = "db.system";
+  static final String DB_USER_KEY = "db.user";
+  static final String NET_PEER_NAME_KEY = "net.peer.name";
+  static final String DB_STATEMENT = "db.statement";
+
+  /**
+   * It sets the attributes for the db client operation corresponding
+   * to the semantic conventions for Optel.
+   *
+   * @link <a href="https://opentelemetry.io/docs/reference/specification/trace/semantic_conventions/database/">Semantic conventions for database client calls</a>
+   *
+   * @param dbConnectionTracingMetadata dbConnectionTracingMetadata the metadata from the tracing connection to get information for the span.
+   * @param spanName the updated name of the current span
+   * @param distributedTraceContextManager the {@link DistributedTraceContextManager} to communicate with the runtime.
+   * @param query the query string for the operation.
+   */
+  public static void setAttributesForDbClientOperation(DbConnectionTracingMetadata dbConnectionTracingMetadata,
+                                                       String spanName,
+                                                       DistributedTraceContextManager distributedTraceContextManager,
+                                                       String query) {
+    try {
+      // We set the name because it is the most generic operation.
+      // We follow the semantic conventions in this part:
+      // "It is not recommended to attempt any client-side parsing of db.statement just to get these properties,
+      // they should only be used if the library being instrumented already provides them. When it?s otherwise
+      // impossible to get any meaningful span name, db.name or the tech-specific database name MAY be used."
+      traceDbClientOperation(spanName,
+                             distributedTraceContextManager,
+                             dbConnectionTracingMetadata,
+                             query);
+    } catch (Throwable e) {
+      LOGGER.warn("Exception on adding span attributes", e);
+    }
+  }
+
+  /**
+   * It sets the attributes for the db client operation corresponding
+   * to the semantic conventions for Optel.
+   * As the span name the db system will be used.
+   *
+   * @link <a href="https://opentelemetry.io/docs/reference/specification/trace/semantic_conventions/database/">Semantic conventions for database client calls</a>
+   *
+   * @param dbConnectionTracingMetadata dbConnectionTracingMetadata the metadata from the tracing connection to get information for the span.
+   * @param distributedTraceContextManager the {@link DistributedTraceContextManager} to communicate with the runtime.
+   * @param query the query string for the operation.
+   */
+  public static void setAttributesForDbClientOperation(DbConnectionTracingMetadata dbConnectionTracingMetadata,
+                                                       DistributedTraceContextManager distributedTraceContextManager,
+                                                       String query) {
+
+    try {
+      // We set the name because it is the most generic operation.
+      // We follow the semantic conventions in this part:
+      // "It is not recommended to attempt any client-side parsing of db.statement just to get these properties,
+      // they should only be used if the library being instrumented already provides them. When it?s otherwise
+      // impossible to get any meaningful span name, db.name or the tech-specific database name MAY be used."
+      traceDbClientOperation(dbConnectionTracingMetadata.getDbSystem(),
+                             distributedTraceContextManager,
+                             dbConnectionTracingMetadata,
+                             query);
+    } catch (Throwable e) {
+      LOGGER.warn("Exception on adding span attributes", e);
+    }
+
+  }
+
+  private static void traceDbClientOperation(String spanName, DistributedTraceContextManager distributedTraceContextManager,
+                                             DbConnectionTracingMetadata dbConnectionTracingMetadata, String sqlQuery) {
+    distributedTraceContextManager.setCurrentSpanName(spanName);
+    distributedTraceContextManager.addCurrentSpanAttributes(new LazySpanAttributesMap(sqlQuery, dbConnectionTracingMetadata));
+  }
+
+  /**
+   * A lazy span attributes map to not affect the performance of the operations.
+   */
+
+  private final static class LazySpanAttributesMap implements Map<String, String> {
+
+    private final DbConnectionTracingMetadata dbConnectionTracingMetadata;
+
+    private final String sqlQuery;
+
+    private Map<String, String> delegate;
+
+    public LazySpanAttributesMap(String sqlQuery, DbConnectionTracingMetadata dbConnectionTracingMetadata) {
+      this.sqlQuery = sqlQuery;
+      this.dbConnectionTracingMetadata = dbConnectionTracingMetadata;
+    }
+
+    @Override
+    public int size() {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.size();
+    }
+
+    private Map<String, String> createDelegateMap() {
+      Map<String, String> delegate = new HashMap<>();
+      delegate.put(DB_STATEMENT, sqlQuery);
+      delegate.put(DB_CONNECTION_STRING_KEY, dbConnectionTracingMetadata.getConnectionString());
+      delegate.put(DB_SYSTEM_KEY, dbConnectionTracingMetadata.getDbSystem());
+      delegate.put(DB_USER_KEY, dbConnectionTracingMetadata.getUser());
+      dbConnectionTracingMetadata.getPeerName()
+          .ifPresent(peerName -> delegate.put(NET_PEER_NAME_KEY, peerName));
+      delegate.put(SPAN_KIND, SPAN_KIND_CLIENT_NAME);
+      return delegate;
+    }
+
+    @Override
+    public boolean isEmpty() {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.isEmpty();
+    }
+
+    @Override
+    public boolean containsKey(Object key) {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.containsKey(key);
+    }
+
+    @Override
+    public boolean containsValue(Object value) {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.containsValue(value);
+    }
+
+    @Override
+    public String get(Object key) {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.get(key);
+    }
+
+    @Override
+    public String put(String key, String value) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public String remove(Object key) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void putAll(Map<? extends String, ? extends String> m) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void clear() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Set<String> keySet() {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.keySet();
+    }
+
+    @Override
+    public Collection<String> values() {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.values();
+    }
+
+    @Override
+    public Set<Entry<String, String>> entrySet() {
+      if (delegate == null) {
+        delegate = createDelegateMap();
+      }
+      return delegate.entrySet();
+    }
+  }
+}
diff --git a/src/test/java/org/mule/extension/db/internal/domain/connection/datasource/ConnectionBasedDbConnectionTracingMetadataTest.java b/src/test/java/org/mule/extension/db/internal/domain/connection/datasource/ConnectionBasedDbConnectionTracingMetadataTest.java
new file mode 100644
index 000000000..b9728e6d4
--- /dev/null
+++ b/src/test/java/org/mule/extension/db/internal/domain/connection/datasource/ConnectionBasedDbConnectionTracingMetadataTest.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.domain.connection.datasource;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import static java.util.Optional.empty;
+import static java.util.Optional.of;
+
+import org.junit.Test;
+import scala.NotImplementedError;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+
+public class ConnectionBasedDbConnectionTracingMetadataTest {
+
+  public static final String EXPECTED_URL = "tcp://localhost:8081/db";
+  public static final String EXPECTED_DB_SYSTEM = "DB_SYSTEM";
+  public static final String EXPECTED_PEER_NAME = "localhost";
+  public static final String EXPECTED_PEER_PROTOCOL = "tcp";
+  public static final String EXPECTED_USER = "expectedUser";
+  public static final String UNSET = "unset";
+
+  @Test
+  public void testDbConnectionTracingMetadata() throws Exception {
+    Connection connection = mock(Connection.class);
+    DatabaseMetaData metadata = mock(DatabaseMetaData.class);
+    when(connection.getMetaData()).thenReturn(metadata);
+    when(metadata.getURL()).thenReturn(EXPECTED_URL);
+    when(metadata.getDatabaseProductName()).thenReturn(EXPECTED_DB_SYSTEM);
+    when(metadata.getUserName()).thenReturn(EXPECTED_USER);
+
+    ConnectionBasedDbConnectionTracingMetadata connectionBasedDbConnectionTracingMetadata =
+        new ConnectionBasedDbConnectionTracingMetadata(connection);
+
+    assertEquals(EXPECTED_URL, connectionBasedDbConnectionTracingMetadata.getConnectionString());
+    assertEquals(EXPECTED_DB_SYSTEM, connectionBasedDbConnectionTracingMetadata.getDbSystem());
+    assertEquals(of(EXPECTED_PEER_NAME), connectionBasedDbConnectionTracingMetadata.getPeerName());
+    assertEquals(of(EXPECTED_PEER_PROTOCOL), connectionBasedDbConnectionTracingMetadata.getPeerTransport());
+    assertEquals(EXPECTED_USER, connectionBasedDbConnectionTracingMetadata.getUser());
+  }
+
+  @Test
+  public void testDoesNotFailWhenTryingToRetrieveMetadata() throws Exception {
+    Connection connection = mock(Connection.class);
+    when(connection.getMetaData()).thenThrow(NotImplementedError.class);
+
+    ConnectionBasedDbConnectionTracingMetadata connectionBasedDbConnectionTracingMetadata =
+        new ConnectionBasedDbConnectionTracingMetadata(connection);
+    assertEquals(UNSET, connectionBasedDbConnectionTracingMetadata.getConnectionString());
+    assertEquals(UNSET, connectionBasedDbConnectionTracingMetadata.getDbSystem());
+    assertEquals(empty(), connectionBasedDbConnectionTracingMetadata.getPeerName());
+    assertEquals(empty(), connectionBasedDbConnectionTracingMetadata.getPeerTransport());
+    assertEquals(UNSET, connectionBasedDbConnectionTracingMetadata.getUser());
+  }
+}
diff --git a/src/test/java/org/mule/extension/db/internal/operation/tracing/TracingUtilsTestCase.java b/src/test/java/org/mule/extension/db/internal/operation/tracing/TracingUtilsTestCase.java
new file mode 100644
index 000000000..dac296540
--- /dev/null
+++ b/src/test/java/org/mule/extension/db/internal/operation/tracing/TracingUtilsTestCase.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.extension.db.internal.operation.tracing;
+
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.DB_CONNECTION_STRING_KEY;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.DB_STATEMENT;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.DB_SYSTEM_KEY;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.DB_USER_KEY;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.NET_PEER_NAME_KEY;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.SPAN_KIND;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.SPAN_KIND_CLIENT_NAME;
+import static org.mule.extension.db.internal.operation.tracing.TracingUtils.setAttributesForDbClientOperation;
+
+import static java.util.Optional.of;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.aMapWithSize;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
+
+import org.mule.db.commons.internal.domain.connection.DbConnectionTracingMetadata;
+import org.mule.sdk.api.runtime.source.DistributedTraceContextManager;
+
+import java.util.Map;
+
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+public class TracingUtilsTestCase {
+
+  public static final String TEST_SPAN_NAME = "test_span_name";
+  public static final String TEST_QUERY = "test_query";
+  public static final String TEST_CONNECTION_STRING = "connection_string";
+  public static final String TEST_DB_SYSTEM = "db_system";
+  public static final String TEST_PEER_NAME = "peer_name";
+  public static final String TEST_USER = "user";
+
+  @Test
+  public void tracingSemanticConventionAttributes() {
+    DbConnectionTracingMetadata dbConnectionTracingMetadata = mock(DbConnectionTracingMetadata.class);
+    DistributedTraceContextManager distributedTraceContext = mock(DistributedTraceContextManager.class);
+    when(dbConnectionTracingMetadata.getConnectionString()).thenReturn(TEST_CONNECTION_STRING);
+    when(dbConnectionTracingMetadata.getDbSystem()).thenReturn(TEST_DB_SYSTEM);
+    when(dbConnectionTracingMetadata.getPeerName()).thenReturn(of(TEST_PEER_NAME));
+    when(dbConnectionTracingMetadata.getUser()).thenReturn(TEST_USER);
+    setAttributesForDbClientOperation(dbConnectionTracingMetadata, TEST_SPAN_NAME, distributedTraceContext, TEST_QUERY);
+    verify(distributedTraceContext).setCurrentSpanName(any());
+    ArgumentCaptor<Map<String, String>> argumentCaptor = ArgumentCaptor.forClass(Map.class);
+    verify(distributedTraceContext).addCurrentSpanAttributes(argumentCaptor.capture());
+    Map<String, String> attributes = argumentCaptor.getValue();
+    assertThat(attributes, aMapWithSize(6));
+    assertThat(attributes, hasEntry(DB_CONNECTION_STRING_KEY, TEST_CONNECTION_STRING));
+    assertThat(attributes, hasEntry(SPAN_KIND, SPAN_KIND_CLIENT_NAME));
+    assertThat(attributes, hasEntry(DB_SYSTEM_KEY, TEST_DB_SYSTEM));
+    assertThat(attributes, hasEntry(DB_USER_KEY, TEST_USER));
+    assertThat(attributes, hasEntry(NET_PEER_NAME_KEY, TEST_PEER_NAME));
+    assertThat(attributes, hasEntry(DB_STATEMENT, TEST_QUERY));
+  }
+
+  @Test
+  public void failsafeAttributesPropagation() {
+    DbConnectionTracingMetadata dbConnectionTracingMetadata = mock(DbConnectionTracingMetadata.class);
+    DistributedTraceContextManager distributedTraceContext = mock(DistributedTraceContextManager.class);
+    setAttributesForDbClientOperation(dbConnectionTracingMetadata, TEST_SPAN_NAME, distributedTraceContext, TEST_QUERY);
+
+  }
+}