mongodb 2.0 support

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.

Author: brettmc

docker-compose.yaml

@@ -14,6 +14,8 @@ services:
       PHP_IDE_CONFIG: ${PHP_IDE_CONFIG:-''}
       RABBIT_HOST: ${RABBIT_HOST:-rabbitmq}
       KAFKA_HOST: ${KAFKA_HOST:-kafka}
+      MONGODB_HOST: ${MONGODB_HOST:-mongodb}
+      MONGODB_PORT: ${MONGODB_PORT:-27017}
 
   zipkin:
     image: openzipkin/zipkin-slim
@@ -60,5 +62,9 @@ services:
     command: "bash -c '/tmp/update_run.sh && /etc/confluent/docker/run'"
     volumes:
       - ./docker/kafka/update_run.sh:/tmp/update_run.sh
-
+  mongodb:
+    image: mongo:4
+    hostname: mongodb
+    ports:
+      - "27017:27017/tcp"
 

src/Instrumentation/MongoDB/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
     }
   }
 }

src/Instrumentation/MongoDB/phpstan.neon.dist

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

src/Instrumentation/MongoDB/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,58 @@ private static function endSpan(?Throwable $exception = null): void
 
         $span->end();
     }
+
+    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/Instrumentation/MongoDB/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));