diff --git a/src/Instrumentation/PDO/src/PDOInstrumentation.php b/src/Instrumentation/PDO/src/PDOInstrumentation.php
index ddf81c91..1a3aa3f1 100644
--- a/src/Instrumentation/PDO/src/PDOInstrumentation.php
+++ b/src/Instrumentation/PDO/src/PDOInstrumentation.php
@@ -12,9 +12,11 @@
 use OpenTelemetry\Context\Context;
 use function OpenTelemetry\Instrumentation\hook;
 use OpenTelemetry\SDK\Common\Configuration\Configuration;
+use OpenTelemetry\SDK\Metrics\Util\TimerTrackerByObject;
 use OpenTelemetry\SemConv\TraceAttributes;
 use OpenTelemetry\SemConv\Version;
 use PDO;
+use PDOException;
 use PDOStatement;
 use Throwable;
 
@@ -30,6 +32,7 @@ public static function register(): void
             Version::VERSION_1_32_0->url(),
         );
         $pdoTracker = new PDOTracker();
+        $timersTracker = new TimerTrackerByObject();
 
         // Hook for the new PDO::connect static method
         if (method_exists(PDO::class, 'connect')) {
@@ -57,7 +60,7 @@ public static function register(): void
                     array $params,
                     $result,
                     ?Throwable $exception,
-                ) use ($pdoTracker) {
+                ) use ($instrumentation, $pdoTracker) {
                     $scope = Context::storage()->scope();
                     if (!$scope) {
                         return;
@@ -73,6 +76,25 @@ public static function register(): void
                     }
 
                     self::end($exception);
+                    
+                    $attributes = $pdoTracker->get($result);
+                    $parent = Context::getCurrent();
+
+                    $instrumentation->meter()
+                        ->createUpDownCounter('db.client.connection.count', '1')
+                        ->add(1, $attributes, $parent);
+
+                    $pdoTracker->trackPdoInstancesDestruction(
+                        $object,
+                        function ($pdoInstance) use ($instrumentation, $pdoTracker) {
+                            $parent = Context::getCurrent();
+
+                            $attributes = $pdoTracker->get($pdoInstance);
+                            $instrumentation->meter()
+                                ->createUpDownCounter('db.client.connection.count', '1')
+                                ->add(-1, $attributes, $parent);
+                        }
+                    );
                 }
             );
         }
@@ -80,7 +102,7 @@ public static function register(): void
         hook(
             PDO::class,
             '__construct',
-            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation) {
+            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker) {
                 /** @psalm-suppress ArgumentTypeCoercion */
                 $builder = self::makeBuilder($instrumentation, 'PDO::__construct', $function, $class, $filename, $lineno)
                     ->setSpanKind(SpanKind::KIND_CLIENT);
@@ -88,7 +110,7 @@ public static function register(): void
                 $span = $builder->startSpan();
                 Context::storage()->attach($span->storeInContext($parent));
             },
-            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) use ($pdoTracker) {
+            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) use ($instrumentation, $pdoTracker) {
                 $scope = Context::storage()->scope();
                 if (!$scope) {
                     return;
@@ -101,146 +123,211 @@ public static function register(): void
                 $span->setAttributes($attributes);
 
                 self::end($exception);
+
+                $attributes = $pdoTracker->get($pdo);
+                $parent = Context::getCurrent();
+
+                $instrumentation->meter()
+                    ->createUpDownCounter('db.client.connection.count', '1')
+                    ->add(1, $attributes, $parent);
+
+                $pdoTracker->trackPdoInstancesDestruction(
+                    $pdo,
+                    function ($pdoInstance) use ($instrumentation, $pdoTracker) {
+                        $parent = Context::getCurrent();
+
+                        $attributes = $pdoTracker->get($pdoInstance);
+                        $instrumentation->meter()
+                            ->createUpDownCounter('db.client.connection.count', '1')
+                            ->add(-1, $attributes, $parent);
+                    }
+                );
             }
         );
 
         hook(
             PDO::class,
             'query',
-            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker, $timersTracker) {
                 /** @psalm-suppress ArgumentTypeCoercion */
                 $builder = self::makeBuilder($instrumentation, 'PDO::query', $function, $class, $filename, $lineno)
                     ->setSpanKind(SpanKind::KIND_CLIENT);
+                $encodedQuery = mb_convert_encoding($params[0] ?? 'undefined', 'UTF-8');
                 if ($class === PDO::class) {
-                    $builder->setAttribute(TraceAttributes::DB_QUERY_TEXT, mb_convert_encoding($params[0] ?? 'undefined', 'UTF-8'));
+                    $builder->setAttribute(TraceAttributes::DB_QUERY_TEXT, $encodedQuery);
                 }
                 $parent = Context::getCurrent();
                 $span = $builder->startSpan();
 
-                $attributes = $pdoTracker->trackedAttributesForPdo($pdo);
+                $attributes = $pdoTracker->append($pdo, TraceAttributes::DB_QUERY_TEXT, $encodedQuery);
                 $span->setAttributes($attributes);
 
                 Context::storage()->attach($span->storeInContext($parent));
+
+                self::createPendingOperationMetric($instrumentation, $attributes, 1);
+                $timersTracker->start($pdo);
             },
-            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) {
-                self::end($exception);
+            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) use ($instrumentation, $pdoTracker, $timersTracker) {
+                $duration = $timersTracker->durationMs($pdo);
+                // this happens ONLY when error mode is set to silent
+                // this is an alternative to changes in the ::end method
+                //                if ($statement === false && $exception === null) {
+                //                    $exception = new class($pdo->errorInfo()) extends \PDOException {
+                //                        // to workaround setting code that is not INT
+                //                        public function __construct(array $errorInfo) {
+                //                            $this->message = $errorInfo[2] ?? 'PDO error';
+                //                            $this->code = $errorInfo[0] ?? 0;
+                //                        }
+                //                    };
+                //                }
+
+                self::end($exception, $statement === false ? $pdo->errorInfo() : []);
+
+                $attributes = $pdoTracker->get($pdo);
+                self::createDurationMetric($instrumentation, $attributes, $duration);
+                self::createPendingOperationMetric($instrumentation, $attributes, -1);
+                if ($statement instanceof PDOStatement && $statement->rowCount()) {
+                    self::createReturnedRowsMetric($instrumentation, $attributes, $statement->rowCount());
+                }
             }
         );
 
         hook(
             PDO::class,
             'exec',
-            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker, $timersTracker) {
                 /** @psalm-suppress ArgumentTypeCoercion */
                 $builder = self::makeBuilder($instrumentation, 'PDO::exec', $function, $class, $filename, $lineno)
                     ->setSpanKind(SpanKind::KIND_CLIENT);
+                $encodedQuery = mb_convert_encoding($params[0] ?? 'undefined', 'UTF-8');
                 if ($class === PDO::class) {
-                    $builder->setAttribute(TraceAttributes::DB_QUERY_TEXT, mb_convert_encoding($params[0] ?? 'undefined', 'UTF-8'));
+                    $builder->setAttribute(TraceAttributes::DB_QUERY_TEXT, $encodedQuery);
                 }
                 $parent = Context::getCurrent();
                 $span = $builder->startSpan();
 
-                $attributes = $pdoTracker->trackedAttributesForPdo($pdo);
+                $attributes = $pdoTracker->append($pdo, TraceAttributes::DB_QUERY_TEXT, $encodedQuery);
                 $span->setAttributes($attributes);
 
                 Context::storage()->attach($span->storeInContext($parent));
+
+                self::createPendingOperationMetric($instrumentation, $attributes, 1);
+
+                $timersTracker->start($pdo);
             },
-            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) {
-                self::end($exception);
+            post: static function (PDO $pdo, array $params, mixed $affectedRows, ?Throwable $exception) use ($instrumentation, $pdoTracker, $timersTracker) {
+                $duration = $timersTracker->durationMs($pdo);
+                self::end($exception, $affectedRows === false ? $pdo->errorInfo() : []);
+
+                $attributes = $pdoTracker->get($pdo);
+                self::createDurationMetric($instrumentation, $attributes, $duration);
+                self::createPendingOperationMetric($instrumentation, $attributes, -1);
+                if (!empty($affectedRows)) {
+                    self::createReturnedRowsMetric($instrumentation, $attributes, $affectedRows);
+                }
             }
         );
 
         hook(
             PDO::class,
             'prepare',
-            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker, $timersTracker) {
                 /** @psalm-suppress ArgumentTypeCoercion */
                 $builder = self::makeBuilder($instrumentation, 'PDO::prepare', $function, $class, $filename, $lineno)
                     ->setSpanKind(SpanKind::KIND_CLIENT);
+                $encodedQuery = mb_convert_encoding($params[0] ?? 'undefined', 'UTF-8');
                 if ($class === PDO::class) {
-                    $builder->setAttribute(TraceAttributes::DB_QUERY_TEXT, mb_convert_encoding($params[0] ?? 'undefined', 'UTF-8'));
+                    $builder->setAttribute(TraceAttributes::DB_QUERY_TEXT, $encodedQuery);
                 }
                 $parent = Context::getCurrent();
                 $span = $builder->startSpan();
 
-                $attributes = $pdoTracker->trackedAttributesForPdo($pdo);
+                $attributes = $pdoTracker->append($pdo, TraceAttributes::DB_QUERY_TEXT, $encodedQuery);
                 $span->setAttributes($attributes);
 
                 Context::storage()->attach($span->storeInContext($parent));
+
+                self::createPendingOperationMetric($instrumentation, $attributes, 1);
+                
+                $timersTracker->start($pdo);
             },
-            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) use ($pdoTracker) {
+            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) use ($instrumentation, $pdoTracker, $timersTracker) {
+                $duration = $timersTracker->durationMs($pdo);
                 if ($statement instanceof PDOStatement) {
                     $pdoTracker->trackStatement($statement, $pdo, Span::getCurrent()->getContext());
                 }
 
-                self::end($exception);
+                self::end($exception, $statement === false ? $pdo->errorInfo() : []);
+                $attributes = $pdoTracker->get($pdo);
+                self::createDurationMetric($instrumentation, $attributes, $duration);
+                self::createPendingOperationMetric($instrumentation, $attributes, -1);
             }
         );
 
         hook(
             PDO::class,
             'beginTransaction',
-            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker) {
                 /** @psalm-suppress ArgumentTypeCoercion */
                 $builder = self::makeBuilder($instrumentation, 'PDO::beginTransaction', $function, $class, $filename, $lineno)
                     ->setSpanKind(SpanKind::KIND_CLIENT);
                 $parent = Context::getCurrent();
                 $span = $builder->startSpan();
 
-                $attributes = $pdoTracker->trackedAttributesForPdo($pdo);
+                $attributes = $pdoTracker->get($pdo);
                 $span->setAttributes($attributes);
 
                 Context::storage()->attach($span->storeInContext($parent));
             },
-            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) {
-                self::end($exception);
+            post: static function (PDO $pdo, array $params, mixed $retval, ?Throwable $exception) {
+                self::end($exception, $retval === false ? $pdo->errorInfo() : []);
             }
         );
 
         hook(
             PDO::class,
             'commit',
-            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker) {
                 /** @psalm-suppress ArgumentTypeCoercion */
                 $builder = self::makeBuilder($instrumentation, 'PDO::commit', $function, $class, $filename, $lineno)
                     ->setSpanKind(SpanKind::KIND_CLIENT);
                 $parent = Context::getCurrent();
                 $span = $builder->startSpan();
 
-                $attributes = $pdoTracker->trackedAttributesForPdo($pdo);
+                $attributes = $pdoTracker->get($pdo);
                 $span->setAttributes($attributes);
 
                 Context::storage()->attach($span->storeInContext($parent));
             },
-            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) {
-                self::end($exception);
+            post: static function (PDO $pdo, array $params, mixed $retval, ?Throwable $exception) {
+                self::end($exception, $retval === false ? $pdo->errorInfo() : []);
             }
         );
 
         hook(
             PDO::class,
             'rollBack',
-            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDO $pdo, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker) {
                 /** @psalm-suppress ArgumentTypeCoercion */
                 $builder = self::makeBuilder($instrumentation, 'PDO::rollBack', $function, $class, $filename, $lineno)
                     ->setSpanKind(SpanKind::KIND_CLIENT);
                 $parent = Context::getCurrent();
                 $span = $builder->startSpan();
 
-                $attributes = $pdoTracker->trackedAttributesForPdo($pdo);
+                $attributes = $pdoTracker->get($pdo);
                 $span->setAttributes($attributes);
 
                 Context::storage()->attach($span->storeInContext($parent));
             },
-            post: static function (PDO $pdo, array $params, mixed $statement, ?Throwable $exception) {
-                self::end($exception);
+            post: static function (PDO $pdo, array $params, mixed $retval, ?Throwable $exception) {
+                self::end($exception, $retval === false ? $pdo->errorInfo() : []);
             }
         );
 
         hook(
             PDOStatement::class,
             'fetchAll',
-            pre: static function (PDOStatement $statement, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDOStatement $statement, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker) {
                 $attributes = $pdoTracker->trackedAttributesForStatement($statement);
                 if (self::isDistributeStatementToLinkedSpansEnabled()) {
                     /** @psalm-suppress InvalidArrayAssignment */
@@ -266,7 +353,7 @@ public static function register(): void
         hook(
             PDOStatement::class,
             'execute',
-            pre: static function (PDOStatement $statement, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($pdoTracker, $instrumentation) {
+            pre: static function (PDOStatement $statement, array $params, string $class, string $function, ?string $filename, ?int $lineno) use ($instrumentation, $pdoTracker, $timersTracker) {
                 $attributes = $pdoTracker->trackedAttributesForStatement($statement);
 
                 if (self::isDistributeStatementToLinkedSpansEnabled()) {
@@ -284,12 +371,21 @@ public static function register(): void
                 $parent = Context::getCurrent();
                 $span = $builder->startSpan();
                 Context::storage()->attach($span->storeInContext($parent));
+
+                self::createPendingOperationMetric($instrumentation, $attributes, 1);
+                $timersTracker->start($statement);
             },
-            post: static function (PDOStatement $statement, array $params, mixed $retval, ?Throwable $exception) {
-                self::end($exception);
+            post: static function (PDOStatement $statement, array $params, mixed $retval, ?Throwable $exception) use ($instrumentation, $pdoTracker, $timersTracker) {
+                $duration = $timersTracker->durationMs($statement);
+                self::end($exception, $retval === false ? $statement->errorInfo() : []);
+
+                $attributes = $pdoTracker->trackedAttributesForStatement($statement);
+                self::createDurationMetric($instrumentation, $attributes, $duration);
+                self::createPendingOperationMetric($instrumentation, $attributes, -1);
             }
         );
     }
+
     private static function makeBuilder(
         CachedInstrumentation $instrumentation,
         string $name,
@@ -305,7 +401,8 @@ private static function makeBuilder(
                     ->setAttribute(TraceAttributes::CODE_FILE_PATH, $filename)
                     ->setAttribute(TraceAttributes::CODE_LINE_NUMBER, $lineno);
     }
-    private static function end(?Throwable $exception): void
+
+    private static function end(?Throwable $exception, array $errorInfo = []): void
     {
         $scope = Context::storage()->scope();
         if (!$scope) {
@@ -316,6 +413,15 @@ private static function end(?Throwable $exception): void
         if ($exception) {
             $span->recordException($exception);
             $span->setStatus(StatusCode::STATUS_ERROR, $exception->getMessage());
+
+        } elseif (!empty($errorInfo[2]) && $errorMessage = $errorInfo[2]) {
+            $span->addEvent('exception', [
+                'exception.type' => PDOException::class,
+                'exception.message' => $errorMessage,
+                // @todo try to add stacktrace?
+            ]);
+
+            $span->setStatus(StatusCode::STATUS_ERROR, $errorMessage);
         }
 
         $span->end();
@@ -327,6 +433,39 @@ private static function isDistributeStatementToLinkedSpansEnabled(): bool
             return Configuration::getBoolean('OTEL_PHP_INSTRUMENTATION_PDO_DISTRIBUTE_STATEMENT_TO_LINKED_SPANS', false);
         }
 
-        return filter_var(get_cfg_var('otel.instrumentation.pdo.distribute_statement_to_linked_spans'), FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE) ?? false;
+        return get_cfg_var('otel.instrumentation.pdo.distribute_statement_to_linked_spans');
+    }
+
+    protected static function createPendingOperationMetric(
+        CachedInstrumentation $instrumentation,
+        array $attributes,
+        int $value,
+    ): void {
+        $parent = Context::getCurrent();
+        $instrumentation->meter()
+            ->createUpDownCounter('db.client.connection.pending_requests', '1')
+            ->add($value, $attributes, $parent);
+    }
+
+    protected static function createReturnedRowsMetric(
+        CachedInstrumentation $instrumentation,
+        array $attributes,
+        int $value,
+    ): void {
+        $parent = Context::getCurrent();
+        $instrumentation->meter()
+            ->createHistogram('db.client.response.returned_rows', '1')
+            ->record($value, $attributes, $parent);
+    }
+
+    protected static function createDurationMetric(
+        CachedInstrumentation $instrumentation,
+        array $attributes,
+        float $value,
+    ): void {
+        $parent = Context::getCurrent();
+        $instrumentation->meter()
+            ->createHistogram('db.client.operation.duration', 'ms')
+            ->record($value, $attributes, $parent);
     }
 }
diff --git a/src/Instrumentation/PDO/src/PDOTracker.php b/src/Instrumentation/PDO/src/PDOTracker.php
index 4753a0ba..d9bec5a8 100644
--- a/src/Instrumentation/PDO/src/PDOTracker.php
+++ b/src/Instrumentation/PDO/src/PDOTracker.php
@@ -4,7 +4,9 @@
 
 namespace OpenTelemetry\Contrib\Instrumentation\PDO;
 
+use Error;
 use OpenTelemetry\API\Trace\SpanContextInterface;
+use OpenTelemetry\SDK\Util\AttributeTrackerByObject;
 use OpenTelemetry\SemConv\TraceAttributes;
 use PDO;
 use PDOStatement;
@@ -14,7 +16,7 @@
 /**
  * @phan-file-suppress PhanNonClassMethodCall,PhanTypeArraySuspicious
  */
-final class PDOTracker
+final class PDOTracker extends AttributeTrackerByObject
 {
     /**
      * @var WeakMap<PDO, array<non-empty-string, bool|int|float|string|array|null>>
@@ -26,13 +28,15 @@ final class PDOTracker
     private WeakMap $statementMapToPdoMap;
     private WeakMap $preparedStatementToSpanMap;
 
+    private WeakMap $pdoInstances;
+
     public function __construct()
     {
-        /** @psalm-suppress PropertyTypeCoercion */
-        $this->pdoToAttributesMap = new WeakMap();
         /** @psalm-suppress PropertyTypeCoercion */
         $this->statementMapToPdoMap = new WeakMap();
         $this->preparedStatementToSpanMap = new WeakMap();
+        $this->pdoInstances = new WeakMap();
+        parent::__construct();
     }
 
     /**
@@ -58,7 +62,7 @@ public function trackedAttributesForStatement(PDOStatement $statement): array
         }
 
         /** @psalm-var array<non-empty-string, bool|int|float|string|array|null> */
-        return $this->pdoToAttributesMap[$pdo] ?? [];
+        return $this->get($pdo);
     }
 
     /**
@@ -75,15 +79,14 @@ public function trackPdoAttributes(PDO $pdo, string $dsn): array
             $dbSystem = $pdo->getAttribute(PDO::ATTR_DRIVER_NAME);
             /** @psalm-suppress InvalidArrayAssignment */
             $attributes[TraceAttributes::DB_SYSTEM_NAME] = self::mapDriverNameToAttribute($dbSystem);
-        } catch (\Error) {
-            // if we caught an exception, the driver is likely not supporting the operation, default to "other"
+        } catch (Error) {
+            // if we caught an exception, and it is not set from extractAttributesFromDSN,
+            // the driver is likely not supporting the operation, default to "other"
             /** @psalm-suppress PossiblyInvalidArrayAssignment */
-            $attributes[TraceAttributes::DB_SYSTEM_NAME] = 'other_sql';
+            $attributes[TraceAttributes::DB_SYSTEM_NAME] ??= 'other_sql';
         }
 
-        $this->pdoToAttributesMap[$pdo] = $attributes;
-
-        return $attributes;
+        return $this->add($pdo, $attributes);
     }
 
     /**
@@ -93,7 +96,22 @@ public function trackPdoAttributes(PDO $pdo, string $dsn): array
     public function trackedAttributesForPdo(PDO $pdo): array
     {
         /** @psalm-var array<non-empty-string, bool|int|float|string|array|null> */
-        return $this->pdoToAttributesMap[$pdo] ?? [];
+        return $this->get($pdo);
+    }
+
+    public function trackPdoInstancesDestruction(PDO $pdo, callable $callbackFunction): void
+    {
+        $this->pdoInstances[$pdo] = new class($pdo, $callbackFunction) {
+            public function __construct(
+                private PDO $instance,
+                private $callback
+            ) {
+            }
+            public function __destruct()
+            {
+                ($this->callback)($this->instance);
+            }
+        };
     }
 
     public function getSpanForPreparedStatement(PDOStatement $statement): ?SpanContextInterface
diff --git a/src/Instrumentation/PDO/tests/Integration/PDOInstrumentationTest.php b/src/Instrumentation/PDO/tests/Integration/PDOInstrumentationTest.php
index 9398c870..66c9fa0a 100644
--- a/src/Instrumentation/PDO/tests/Integration/PDOInstrumentationTest.php
+++ b/src/Instrumentation/PDO/tests/Integration/PDOInstrumentationTest.php
@@ -5,11 +5,25 @@
 namespace OpenTelemetry\Tests\Instrumentation\PDO\Integration;
 
 use ArrayObject;
+use OpenTelemetry\API\Common\Time\TestClock;
+use OpenTelemetry\API\Globals;
 use OpenTelemetry\API\Instrumentation\Configurator;
 use OpenTelemetry\API\Trace\SpanInterface;
 use OpenTelemetry\API\Trace\SpanKind;
 use OpenTelemetry\Context\Context;
 use OpenTelemetry\Context\ScopeInterface;
+use OpenTelemetry\SDK\Common\Attribute\Attributes;
+use OpenTelemetry\SDK\Common\Export\InMemoryStorageManager;
+use OpenTelemetry\SDK\Common\Instrumentation\InstrumentationScopeFactory;
+use OpenTelemetry\SDK\Metrics\Data\Metric;
+use OpenTelemetry\SDK\Metrics\Data\Sum;
+use OpenTelemetry\SDK\Metrics\Data\Temporality;
+use OpenTelemetry\SDK\Metrics\Exemplar\ExemplarFilter\AllExemplarFilter;
+use OpenTelemetry\SDK\Metrics\MeterProvider;
+use OpenTelemetry\SDK\Metrics\MetricReader\ExportingReader;
+use OpenTelemetry\SDK\Metrics\StalenessHandler\DelayedStalenessHandlerFactory;
+use OpenTelemetry\SDK\Metrics\View\CriteriaViewRegistry;
+use OpenTelemetry\SDK\Resource\ResourceInfoFactory;
 use OpenTelemetry\SDK\Trace\ImmutableSpan;
 use OpenTelemetry\SDK\Trace\SpanExporter\InMemoryExporter;
 use OpenTelemetry\SDK\Trace\SpanProcessor\SimpleSpanProcessor;
@@ -17,6 +31,7 @@
 use OpenTelemetry\SemConv\TraceAttributes;
 use OpenTelemetry\TestUtils\TraceStructureAssertionTrait;
 use PDO;
+use PDOException;
 use PHPUnit\Framework\TestCase;
 
 class PDOInstrumentationTest extends TestCase
@@ -26,6 +41,9 @@ class PDOInstrumentationTest extends TestCase
     private ScopeInterface $scope;
     /** @var ArrayObject<array-key, mixed> */
     private ArrayObject $storage;
+    private $metricReader;
+    private $meterProvider;
+    private $metricExporter;
 
     private function createDB(): PDO
     {
@@ -67,9 +85,27 @@ public function setUp(): void
                 new InMemoryExporter($this->storage)
             )
         );
+        $clock = new TestClock();
+        $this->metricExporter = new \OpenTelemetry\SDK\Metrics\MetricExporter\InMemoryExporter(
+            InMemoryStorageManager::metrics(),
+            Temporality::CUMULATIVE
+        );
+        $this->metricReader = new ExportingReader($this->metricExporter);
+        $this->meterProvider = new MeterProvider(
+            null,
+            ResourceInfoFactory::emptyResource(),
+            $clock,
+            Attributes::factory(),
+            new InstrumentationScopeFactory(Attributes::factory()),
+            [$this->metricReader],
+            new CriteriaViewRegistry(),
+            new AllExemplarFilter(),
+            new DelayedStalenessHandlerFactory($clock, 20),
+        );
 
         $this->scope = Configurator::create()
             ->withTracerProvider($tracerProvider)
+            ->withMeterProvider($this->meterProvider)
             ->activate();
     }
 
@@ -135,7 +171,7 @@ public function test_pdo_sqlite_subclass(): void
 
     public function test_constructor_exception(): void
     {
-        $this->expectException(\PDOException::class);
+        $this->expectException(PDOException::class);
         $this->expectExceptionMessage('could not find driver');
         new PDO('unknown:foo');
     }
@@ -402,4 +438,84 @@ public function test_span_hierarchy_with_pdo_operations(): void
             ]
         );
     }
+
+    public function test_connection_metrics(): void
+    {
+        $db = self::createDB();
+        $db->exec($this->fillDB());
+
+        $non_utf8_id = mb_convert_encoding('rückwärts', 'ISO-8859-1', 'UTF-8');
+
+        $stmt = $db->prepare("SELECT id FROM technology WHERE id = '{$non_utf8_id}'");
+        $span_db_prepare = $this->storage->offsetGet(2);
+        $this->assertTrue(mb_check_encoding($span_db_prepare->getAttributes()->get(TraceAttributes::DB_QUERY_TEXT), 'UTF-8'));
+        $this->assertCount(3, $this->storage);
+
+        $stmt->execute();
+        $stmt->fetchAll();
+        unset($db);
+
+        $meterProvider = Globals::meterProvider();
+        if ($meterProvider instanceof MeterProvider) {
+            $meterProvider->forceFlush();
+            $meterProvider->shutdown();
+        }
+
+        $metricStats = [];
+        /** @var Metric $metric */
+        foreach (InMemoryStorageManager::metrics() as $metric) {
+            $metricStats[$metric->name] ??= [];
+            $metricStats[$metric->name][] = $metric->data;
+        }
+        self::assertCount(2, $metricStats['db.client.connection.count']);
+        self::assertCount(2, $metricStats['db.client.connection.pending_requests']);
+        self::assertCount(2, $metricStats['db.client.operation.duration']);
+        self::assertCount(2, $metricStats['db.client.response.returned_rows']);
+
+        // check count metrics:
+        $this->checkMetricSum($metricStats['db.client.connection.count'], 2, 2);
+        $this->checkMetricSum($metricStats['db.client.connection.pending_requests'], 0, 2);
+        $this->checkMetricDataPointsCount($metricStats['db.client.operation.duration'], 2);
+        $this->checkMetricDataPointsCount($metricStats['db.client.response.returned_rows'], 2);
+    }
+
+    public function test_error_info_in_silent_mode(): void
+    {
+        $db = self::createDB();
+        $db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_SILENT);
+        $db->exec($this->fillDB());
+
+        $db->query('SELECT id FROM non_existing_table');
+
+        $meterProvider = Globals::meterProvider();
+        if ($meterProvider instanceof MeterProvider) {
+            $meterProvider->forceFlush();
+            $meterProvider->shutdown();
+        }
+        $metrics = InMemoryStorageManager::metrics();
+        self::assertTrue(true);
+    }
+
+    public function checkMetricDataPointsCount($metrics, $expectedCount): void
+    {
+        $sum = 0;
+        /** @var Sum $point */
+        foreach ($metrics as $point) {
+            $sum += $point->dataPoints[0]->count;
+        }
+        self::assertEquals($expectedCount, $sum);
+    }
+
+    public function checkMetricSum($metrics, $expectedSum, ?int $count = null): void
+    {
+        $sum = 0;
+        /** @var Sum $point */
+        foreach ($metrics as $point) {
+            $sum += $point->dataPoints[0]->value;
+        }
+        self::assertEquals($expectedSum, $sum);
+        if ($count) {
+            self::assertCount($count, $metrics);
+        }
+    }
 }