ext-mongodb 2.0 support (#324)

Being able to retrieve host, port + info from a CommandStartedEvent is deprecated in 1.20.0 and will be removed in 2.0. we have been advised to use SDAM subscription instead.

- update tests to allow defining mongo server from env
- require ext-mongodb 1.13 (which provides SDAM subscriber for server info)
- use SDAM subscriber to get server info attributes. Subscribe to the serverChanged event and store the attributes in a class variable, organised by host/port.
- document mongo todos

composer.json

@@ -6,10 +6,9 @@
   "homepage": "https://opentelemetry.io/docs/php",
   "readme": "./README.md",
   "license": "Apache-2.0",
-  "minimum-stability": "dev",
   "require": {
     "php": ">=7.4",
-    "ext-mongodb": "*",
+    "ext-mongodb": "^1.13",
     "ext-json": "*",
     "mongodb/mongodb": "^1.15",
     "open-telemetry/api": "^1.0",
@@ -40,7 +39,8 @@
   },
   "config": {
     "allow-plugins": {
-      "php-http/discovery": false
+      "php-http/discovery": false,
+      "tbachert/spi": false
     }
   }
 }

phpstan.neon.dist

@@ -7,3 +7,8 @@ parameters:
     paths:
         - src
         - tests
+    ignoreErrors:
+        -
+            message: "#Call to an undefined method .*#"
+            paths:
+                - src/MongoDBInstrumentationSubscriber.php

src/MongoDBInstrumentationSubscriber.php

@@ -9,6 +9,16 @@
 use MongoDB\Driver\Monitoring\CommandStartedEvent;
 use MongoDB\Driver\Monitoring\CommandSubscriber;
 use MongoDB\Driver\Monitoring\CommandSucceededEvent;
+use MongoDB\Driver\Monitoring\SDAMSubscriber;
+use MongoDB\Driver\Monitoring\ServerChangedEvent;
+use MongoDB\Driver\Monitoring\ServerClosedEvent;
+use MongoDB\Driver\Monitoring\ServerHeartbeatFailedEvent;
+use MongoDB\Driver\Monitoring\ServerHeartbeatStartedEvent;
+use MongoDB\Driver\Monitoring\ServerHeartbeatSucceededEvent;
+use MongoDB\Driver\Monitoring\ServerOpeningEvent;
+use MongoDB\Driver\Monitoring\TopologyChangedEvent;
+use MongoDB\Driver\Monitoring\TopologyClosedEvent;
+use MongoDB\Driver\Monitoring\TopologyOpeningEvent;
 use OpenTelemetry\API\Instrumentation\CachedInstrumentation;
 use OpenTelemetry\API\Trace\Span;
 use OpenTelemetry\API\Trace\SpanBuilderInterface;
@@ -18,13 +28,17 @@
 use OpenTelemetry\SemConv\TraceAttributes;
 use Throwable;
 
-final class MongoDBInstrumentationSubscriber implements CommandSubscriber
+final class MongoDBInstrumentationSubscriber implements CommandSubscriber, SDAMSubscriber
 {
     private CachedInstrumentation $instrumentation;
     /**
      * @var Closure(object):?string
      */
     private Closure $commandSerializer;
+    /**
+     * @var array<string, array<int, array<string, mixed>>>
+     */
+    private array $serverAttributes = [];
 
     /**
      * @param (callable(object):?string) $commandSerializer
@@ -41,16 +55,26 @@ public function __construct(CachedInstrumentation $instrumentation, callable $co
         };
     }
 
+    /**
+     * @psalm-suppress MixedAssignment,MixedArrayTypeCoercion,MixedArrayOffset,MixedArgument
+     */
     public function commandStarted(CommandStartedEvent $event): void
     {
         $command = $event->getCommand();
         $collectionName = MongoDBCollectionExtractor::extract($command);
         $databaseName = $event->getDatabaseName();
         $commandName = $event->getCommandName();
-        $server = $event->getServer();
-        $info = $server->getInfo();
-        $port = $server->getPort();
-        $host = $server->getHost();
+        /** @phpstan-ignore-next-line */
+        if (version_compare(phpversion('mongodb'), '1.20.0', '>=')) {
+            $host = $event->getHost();
+            $port = $event->getPort();
+        } else {
+            $server = $event->getServer();
+            $host = $server->getHost();
+            $port = $server->getPort();
+        }
+        $attributes = $this->serverAttributes[$host][$port] ?? [];
+
         $isSocket = str_starts_with($host, '/');
         /** @psalm-suppress RiskyTruthyFalsyComparison **/
         $scopedCommand = ($collectionName ? $collectionName . '.' : '') . $commandName;
@@ -65,17 +89,10 @@ public function commandStarted(CommandStartedEvent $event): void
             ->setAttribute(TraceAttributes::NETWORK_TRANSPORT, $isSocket ? 'unix' : 'tcp')
             ->setAttribute(TraceAttributes::DB_STATEMENT, ($this->commandSerializer)($command))
             ->setAttribute(TraceAttributes::DB_MONGODB_COLLECTION, $collectionName)
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_MASTER, $info['ismaster'] ?? null)
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_READ_ONLY, $info['readOnly'] ?? null)
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_CONNECTION_ID, $info['connectionId'] ?? null)
             ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_REQUEST_ID, $event->getRequestId())
             ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_OPERATION_ID, $event->getOperationId())
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_MAX_WIRE_VERSION, $info['maxWireVersion'] ?? null)
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_MIN_WIRE_VERSION, $info['minWireVersion'] ?? null)
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_MAX_BSON_OBJECT_SIZE_BYTES, $info['maxBsonObjectSize'] ?? null)
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_MAX_MESSAGE_SIZE_BYTES, $info['maxMessageSizeBytes'] ?? null)
-            ->setAttribute(MongoDBTraceAttributes::DB_MONGODB_MAX_WRITE_BATCH_SIZE, $info['maxWriteBatchSize'] ?? null);
-
+            ->setAttributes($attributes)
+        ;
         $parent = Context::getCurrent();
         $span = $builder->startSpan();
         Context::storage()->attach($span->storeInContext($parent));
@@ -118,4 +135,61 @@ private static function endSpan(?Throwable $exception = null): void
 
         $span->end();
     }
+
+    /**
+     * @todo In a load-balanced scenario, the hello response may be empty.
+     */
+    public function serverChanged(ServerChangedEvent $event): void
+    {
+        $host = $event->getHost();
+        $port = $event->getPort();
+        $info = $event->getNewDescription()->getHelloResponse();
+        $attributes = [
+            MongoDBTraceAttributes::DB_MONGODB_MASTER => $info['ismaster'] ?? null,
+            MongoDBTraceAttributes::DB_MONGODB_READ_ONLY => $info['readOnly'] ?? null,
+            MongoDBTraceAttributes::DB_MONGODB_CONNECTION_ID => $info['connectionId'] ?? null,
+            MongoDBTraceAttributes::DB_MONGODB_MAX_WIRE_VERSION => $info['maxWireVersion'] ?? null,
+            MongoDBTraceAttributes::DB_MONGODB_MIN_WIRE_VERSION => $info['minWireVersion'] ?? null,
+            MongoDBTraceAttributes::DB_MONGODB_MAX_BSON_OBJECT_SIZE_BYTES => $info['maxBsonObjectSize'] ?? null,
+            MongoDBTraceAttributes::DB_MONGODB_MAX_MESSAGE_SIZE_BYTES => $info['maxMessageSizeBytes'] ?? null,
+            MongoDBTraceAttributes::DB_MONGODB_MAX_WRITE_BATCH_SIZE => $info['maxWriteBatchSize'] ?? null,
+        ];
+        $this->serverAttributes[$host][$port] = $attributes;
+    }
+
+    public function serverOpened(ServerOpeningEvent $event): void
+    {
+    }
+
+    public function serverClosed(ServerClosedEvent $event): void
+    {
+    }
+
+    public function serverOpening(ServerOpeningEvent $event): void
+    {
+    }
+
+    public function serverHeartbeatFailed(ServerHeartbeatFailedEvent $event): void
+    {
+    }
+
+    public function serverHeartbeatStarted(ServerHeartbeatStartedEvent $event): void
+    {
+    }
+
+    public function serverHeartbeatSucceeded(ServerHeartbeatSucceededEvent $event): void
+    {
+    }
+
+    public function topologyChanged(TopologyChangedEvent $event): void
+    {
+    }
+
+    public function topologyClosed(TopologyClosedEvent $event): void
+    {
+    }
+
+    public function topologyOpening(TopologyOpeningEvent $event): void
+    {
+    }
 }

src/MongoDBTraceAttributes.php

@@ -4,6 +4,10 @@
 
 namespace OpenTelemetry\Contrib\Instrumentation\MongoDB;
 
+/**
+ * @todo These attributes are not part of the specification and should be removed before this package goes stable,
+ * unless they are on the way to being added to the specification. See https://github.com/open-telemetry/opentelemetry-specification/blob/v1.40.0/specification/telemetry-stability.md
+ */
 interface MongoDBTraceAttributes
 {
     public const DB_MONGODB_MASTER = 'db.mongodb.master';

tests/Integration/MongoDBInstrumentationTest.php

@@ -22,13 +22,19 @@ class MongoDBInstrumentationTest extends TestCase
 {
     private const DATABASE_NAME = 'db';
     private const COLLECTION_NAME = 'coll';
+    private string $host;
+    private int $port;
+    private string $uri;
     private ScopeInterface $scope;
     /** @var ArrayObject<int,ImmutableSpan> */
     private ArrayObject $storage;
     private ?ImmutableSpan $span = null;
 
     public function setUp(): void
     {
+        $this->host = $_SERVER['MONGODB_HOST'] ?? '127.0.0.1';
+        $this->port = (int) ($_SERVER['MONGODB_PORT'] ?? 27017);
+        $this->uri = "mongodb://$this->host:$this->port";
         /** @psalm-suppress MixedPropertyTypeCoercion */
         $this->storage = new ArrayObject();
         $tracerProvider = new TracerProvider(
@@ -49,7 +55,7 @@ public function tearDown(): void
 
     public function test_mongodb_find_one(): void
     {
-        $manager = new Manager('mongodb://127.0.0.1:27017');
+        $manager = new Manager($this->uri);
 
         $find = new FindOne(self::DATABASE_NAME, self::COLLECTION_NAME, ['a' => 'b']);
 
@@ -67,8 +73,8 @@ public function test_mongodb_find_one(): void
         self::assertSame(self::DATABASE_NAME, $attributes->get(TraceAttributes::DB_NAME));
         self::assertSame('find', $attributes->get(TraceAttributes::DB_OPERATION));
         self::assertSame(self::COLLECTION_NAME, $attributes->get(TraceAttributes::DB_MONGODB_COLLECTION));
-        self::assertSame('127.0.0.1', $attributes->get(TraceAttributes::SERVER_ADDRESS));
-        self::assertSame(27017, $attributes->get(TraceAttributes::SERVER_PORT));
+        self::assertSame($this->host, $attributes->get(TraceAttributes::SERVER_ADDRESS));
+        self::assertSame($this->port, $attributes->get(TraceAttributes::SERVER_PORT));
         self::assertSame('tcp', $attributes->get(TraceAttributes::NETWORK_TRANSPORT));
         self::assertTrue($attributes->get(MongoDBTraceAttributes::DB_MONGODB_MASTER));
         self::assertFalse($attributes->get(MongoDBTraceAttributes::DB_MONGODB_READ_ONLY));