CDRIVER-4500: SDAM structured log and unified test support (#1842)

* Resolves CDRIVER-4500 (SDAM structured logs + updated unified tests)
* Resolves CDRIVER-4758 (Prose test for heartbeat event order: /server_discovery_and_monitoring/prose/heartbeat)
* Resolves CDRIVER-4137 (Missing server close events, now tested by /server_discovery_and_monitoring/unified)
* Completes outstanding issues in the CDRIVER-3775 epic (libmongoc structured logging)

Details:

* Synchronized SDAM spec tests from the specifications repository, commit d795d493c41022cb8ed15006ae5ac5ad85936f40

* Added /server_discovery_and_monitoring/unified tests

* Refactoring: new log-and-monitor type for apm monitoring state plus structured log instance

* Implemented SDAM log messages for topology changes and monitoring lifecycle

* Fixes for lifecycle events generated for Structured Log and APM consumers:
  * ServerOpeningEvent is deferred based on topology 'opening' state, not based on when callbacks are installed.
  * Required ServerClosedEvent and topology Unknown state changes are emitted prior to TopologyClosedEvent.

* New Structured Log items:
  * oid() for a plain ObjectID without hex representation
  * topology_description_as_json() for plain topology descriptions not inside a topology

* Added test skips for SDAM unified tests that require pool support, driver-generated connection IDs, thread entities.
* Test skips for CDRIVER-5870 (Spec compliant response to authentication errors)
* Additional debug output from loadbalancer tests

* CDRIVER-4758: added /server_discovery_and_monitoring/prose/heartbeat

* Unified test runner changes
  * Error instead of segfault when checking events/logs on missing client entity
  * Support for "closing" some entity types without deleting (For example, to inspect events logged when closing a client)
  * Added debug logging about log suppression
  * Added per-entity structured log filter stack (entity_map_log_filter_push/pop) used internally to refine the apparent behavior of waitForEvent
  * Added SDAM APM events, and a suitable eventType filtering implementation.
  * New operations: assertEventCount, recordTopologyDescription, assertTopologyType, waitForPrimaryChange
  * entity-map: pass server monitoring mode URI option
  * runCommand should report reply as result for matching
  * waitForEvent: do an entire stream-selection, to advance blocking topology scans
  * Fix entity-map destructor for findcursor without cursor

Author: mdbmes

NEWS

@@ -1,6 +1,12 @@
 libmongoc 1.30.0 (Unreleased)
 =============================
 
+Fixes:
+
+  * Additional APM events required by the SDAM specification will now be delivered:
+    * Servers that have seen `server_opening` will now see a `server_closed` prior to `topology_closed`.
+    * Before `topology_closed`, a `topology_changed` event will transition to `Unknown` topology type.
+
 Deprecated:
 
   * Support for Debian 9 and Debian 10.

src/libmongoc/CMakeLists.txt

@@ -623,6 +623,7 @@ set (MONGOC_SOURCES
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-list.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-linux-distro-scanner.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-log.c
+   ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-log-and-monitor-private.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-matcher.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-matcher-op.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-memcmp.c

src/libmongoc/src/mongoc/mongoc-client-pool.c

@@ -23,6 +23,7 @@
 #include <mongoc/mongoc-client-pool.h>
 #include <mongoc/mongoc-client-private.h>
 #include <mongoc/mongoc-client-side-encryption-private.h>
+#include <mongoc/mongoc-log-and-monitor-private.h>
 #include <mongoc/mongoc-queue-private.h>
 #include <mongoc/mongoc-thread-private.h>
 #include <mongoc/mongoc-topology-private.h>
@@ -54,8 +55,6 @@ struct _mongoc_client_pool_t {
    bool error_api_set;
    bool structured_log_opts_set;
    bool client_initialized;
-   mongoc_apm_callbacks_t apm_callbacks;
-   void *apm_context;
    int32_t error_api_version;
    mongoc_server_api_t *api;
    // `last_known_serverids` is a sorted array of uint32_t.
@@ -283,7 +282,6 @@ _initialize_new_client (mongoc_client_pool_t *pool, mongoc_client_t *client)
 
    pool->client_initialized = true;
    client->error_api_version = pool->error_api_version;
-   _mongoc_client_set_apm_callbacks_private (client, &pool->apm_callbacks, pool->apm_context);
 
    client->api = mongoc_server_api_copy (pool->api);
 
@@ -577,35 +575,28 @@ bool
 mongoc_client_pool_set_apm_callbacks (mongoc_client_pool_t *pool, mongoc_apm_callbacks_t *callbacks, void *context)
 {
    BSON_ASSERT_PARAM (pool);
+   BSON_OPTIONAL_PARAM (callbacks);
+   BSON_OPTIONAL_PARAM (context);
 
-   mongoc_topology_t *const topology = BSON_ASSERT_PTR_INLINE (pool)->topology;
-   mc_tpld_modification tdmod = mc_tpld_modify_begin (topology);
-
-   // Prevent setting callbacks more than once
+   // Enforce documented thread-safety restrictions
    if (pool->apm_callbacks_set) {
-      mc_tpld_modify_drop (tdmod);
-      MONGOC_ERROR ("Can only set callbacks once");
+      MONGOC_ERROR ("mongoc_client_pool_set_apm_callbacks can only be called once per pool");
       return false;
-   }
+   } else if (pool->client_initialized) {
+      MONGOC_ERROR ("mongoc_client_pool_set_apm_callbacks can only be called before mongoc_client_pool_pop");
+      /* @todo Since 2017 this requirement has been documented but not actually enforced. For now we are leaving it
+       * unenforced, for backward compatibility. This usage remains unsafe and incorrect. When possible, this should be
+       * modified to return false without modifying the APM callbacks. */
+      mongoc_log_and_monitor_instance_set_apm_callbacks (&pool->topology->log_and_monitor, callbacks, context);
+      pool->apm_callbacks_set = true;
+      return true;
 
-   // Update callbacks on the pool
-   if (callbacks) {
-      pool->apm_callbacks = *callbacks;
    } else {
-      pool->apm_callbacks = (mongoc_apm_callbacks_t){0};
+      // Now we can be sure no other threads are relying on concurrent access to the instance yet.
+      mongoc_log_and_monitor_instance_set_apm_callbacks (&pool->topology->log_and_monitor, callbacks, context);
+      pool->apm_callbacks_set = true;
+      return true;
    }
-   pool->apm_context = context;
-
-   // Update callbacks on the topology
-   mongoc_topology_set_apm_callbacks (topology, tdmod.new_td, callbacks, context);
-
-   // Signal that we have already set the callbacks
-   pool->apm_callbacks_set = true;
-
-   // Save our updated topology
-   mc_tpld_modify_commit (tdmod);
-
-   return true;
 }
 
 bool
@@ -618,14 +609,14 @@ mongoc_client_pool_set_structured_log_opts (mongoc_client_pool_t *pool, const mo
     * and only before the first client is initialized. Structured logging is generally
     * expected to warn but not quit when encountering initialization errors. */
    if (pool->structured_log_opts_set) {
-      MONGOC_WARNING ("mongoc_client_pool_set_structured_log_opts can only be called once per pool");
+      MONGOC_ERROR ("mongoc_client_pool_set_structured_log_opts can only be called once per pool");
       return false;
    } else if (pool->client_initialized) {
-      MONGOC_WARNING ("mongoc_client_pool_set_structured_log_opts can only be called before mongoc_client_pool_pop");
+      MONGOC_ERROR ("mongoc_client_pool_set_structured_log_opts can only be called before mongoc_client_pool_pop");
       return false;
    } else {
       // Now we can be sure no other threads are relying on concurrent access to the instance yet.
-      mongoc_topology_set_structured_log_opts (pool->topology, opts);
+      mongoc_log_and_monitor_instance_set_structured_log_opts (&pool->topology->log_and_monitor, opts);
       pool->structured_log_opts_set = true;
       return true;
    }

src/libmongoc/src/mongoc/mongoc-client-private.h

@@ -107,9 +107,6 @@ struct _mongoc_client_t {
    mongoc_read_concern_t *read_concern;
    mongoc_write_concern_t *write_concern;
 
-   mongoc_apm_callbacks_t apm_callbacks;
-   void *apm_context;
-
    int32_t error_api_version;
    bool error_api_set;
 
@@ -151,9 +148,6 @@ _mongoc_client_get_rr (const char *hostname,
 mongoc_client_t *
 _mongoc_client_new_from_topology (mongoc_topology_t *topology);
 
-bool
-_mongoc_client_set_apm_callbacks_private (mongoc_client_t *client, mongoc_apm_callbacks_t *callbacks, void *context);
-
 mongoc_stream_t *
 mongoc_client_default_stream_initiator (const mongoc_uri_t *uri,
                                         const mongoc_host_list_t *host,

src/libmongoc/src/mongoc/mongoc-client.c

@@ -1122,7 +1122,7 @@ _mongoc_client_new_from_topology (mongoc_topology_t *topology)
    }
 #endif
 
-   mongoc_structured_log (topology->structured_log,
+   mongoc_structured_log (topology->log_and_monitor.structured_log,
                           MONGOC_STRUCTURED_LOG_LEVEL_DEBUG,
                           MONGOC_STRUCTURED_LOG_COMPONENT_CONNECTION,
                           "Client created");
@@ -1646,8 +1646,6 @@ _mongoc_client_retryable_read_command_with_stream (mongoc_client_t *client,
     * server does not support retryable reads, fall through and allow the
     * original error to be reported. */
    if (is_retryable && _mongoc_read_error_get_type (ret, error, reply) == MONGOC_READ_ERR_RETRY) {
-      bson_error_t ignored_error;
-
       /* each read command may be retried at most once */
       is_retryable = false;
 
@@ -1667,8 +1665,13 @@ _mongoc_client_retryable_read_command_with_stream (mongoc_client_t *client,
             .has_operation_id = true,
             .operation_id = parts->assembled.operation_id,
          };
-         retry_server_stream = mongoc_cluster_stream_for_reads (
-            &client->cluster, &ss_log_context, parts->read_prefs, parts->assembled.session, ds, NULL, &ignored_error);
+         retry_server_stream = mongoc_cluster_stream_for_reads (&client->cluster,
+                                                                &ss_log_context,
+                                                                parts->read_prefs,
+                                                                parts->assembled.session,
+                                                                ds,
+                                                                NULL /* reply */,
+                                                                NULL /* error */);
 
          mongoc_deprioritized_servers_destroy (ds);
       }
@@ -2165,14 +2168,14 @@ _mongoc_client_monitor_op_killcursors (mongoc_cluster_t *cluster,
                                        const char *collection)
 {
    bson_t doc;
-   mongoc_client_t *client;
    mongoc_apm_command_started_t event;
 
    ENTRY;
 
-   client = cluster->client;
+   mongoc_client_t *client = cluster->client;
+   const mongoc_log_and_monitor_instance_t *log_and_monitor = &client->topology->log_and_monitor;
 
-   if (!client->apm_callbacks.started) {
+   if (!log_and_monitor->apm_callbacks.started) {
       return;
    }
 
@@ -2189,9 +2192,9 @@ _mongoc_client_monitor_op_killcursors (mongoc_cluster_t *cluster,
                                     &server_stream->sd->service_id,
                                     server_stream->sd->server_connection_id,
                                     NULL,
-                                    client->apm_context);
+                                    log_and_monitor->apm_context);
 
-   client->apm_callbacks.started (&event);
+   log_and_monitor->apm_callbacks.started (&event);
    mongoc_apm_command_started_cleanup (&event);
    bson_destroy (&doc);
 
@@ -2207,16 +2210,16 @@ _mongoc_client_monitor_op_killcursors_succeeded (mongoc_cluster_t *cluster,
                                                  int64_t operation_id,
                                                  const char *db)
 {
-   mongoc_client_t *client;
    bson_t doc;
    bson_array_builder_t *cursors_unknown;
    mongoc_apm_command_succeeded_t event;
 
    ENTRY;
 
-   client = cluster->client;
+   mongoc_client_t *client = cluster->client;
+   const mongoc_log_and_monitor_instance_t *log_and_monitor = &client->topology->log_and_monitor;
 
-   if (!client->apm_callbacks.succeeded) {
+   if (!log_and_monitor->apm_callbacks.succeeded) {
       EXIT;
    }
 
@@ -2239,9 +2242,9 @@ _mongoc_client_monitor_op_killcursors_succeeded (mongoc_cluster_t *cluster,
                                       &server_stream->sd->service_id,
                                       server_stream->sd->server_connection_id,
                                       false,
-                                      client->apm_context);
+                                      log_and_monitor->apm_context);
 
-   client->apm_callbacks.succeeded (&event);
+   log_and_monitor->apm_callbacks.succeeded (&event);
 
    mongoc_apm_command_succeeded_cleanup (&event);
    bson_destroy (&doc);
@@ -2256,15 +2259,15 @@ _mongoc_client_monitor_op_killcursors_failed (mongoc_cluster_t *cluster,
                                               int64_t operation_id,
                                               const char *db)
 {
-   mongoc_client_t *client;
    bson_t doc;
    mongoc_apm_command_failed_t event;
 
    ENTRY;
 
-   client = cluster->client;
+   mongoc_client_t *client = cluster->client;
+   const mongoc_log_and_monitor_instance_t *log_and_monitor = &client->topology->log_and_monitor;
 
-   if (!client->apm_callbacks.failed) {
+   if (!log_and_monitor->apm_callbacks.failed) {
       EXIT;
    }
 
@@ -2285,9 +2288,9 @@ _mongoc_client_monitor_op_killcursors_failed (mongoc_cluster_t *cluster,
                                    &server_stream->sd->service_id,
                                    server_stream->sd->server_connection_id,
                                    false,
-                                   client->apm_context);
+                                   log_and_monitor->apm_context);
 
-   client->apm_callbacks.failed (&event);
+   log_and_monitor->apm_callbacks.failed (&event);
 
    mongoc_apm_command_failed_cleanup (&event);
    bson_destroy (&doc);
@@ -2576,33 +2579,6 @@ mongoc_client_set_stream_initiator (mongoc_client_t *client, mongoc_stream_initi
 }
 
 
-bool
-_mongoc_client_set_apm_callbacks_private (mongoc_client_t *client, mongoc_apm_callbacks_t *callbacks, void *context)
-{
-   BSON_ASSERT_PARAM (client);
-
-   if (callbacks) {
-      memcpy (&client->apm_callbacks, callbacks, sizeof (mongoc_apm_callbacks_t));
-   } else {
-      memset (&client->apm_callbacks, 0, sizeof (mongoc_apm_callbacks_t));
-   }
-
-   client->apm_context = context;
-
-   /* A client pool sets APM callbacks for the entire pool. */
-   if (client->topology->single_threaded) {
-      mongoc_topology_set_apm_callbacks (client->topology,
-                                         /* We are safe to modify the shared_descr directly, since we are
-                                          * single-threaded */
-                                         mc_tpld_unsafe_get_mutable (client->topology),
-                                         callbacks,
-                                         context);
-   }
-
-   return true;
-}
-
-
 bool
 mongoc_client_set_apm_callbacks (mongoc_client_t *client, mongoc_apm_callbacks_t *callbacks, void *context)
 {
@@ -2614,7 +2590,9 @@ mongoc_client_set_apm_callbacks (mongoc_client_t *client, mongoc_apm_callbacks_t
       return false;
    }
 
-   return _mongoc_client_set_apm_callbacks_private (client, callbacks, context);
+   mongoc_log_and_monitor_instance_set_apm_callbacks (&client->topology->log_and_monitor, callbacks, context);
+
+   return true;
 }
 
 
@@ -2625,7 +2603,7 @@ mongoc_client_set_structured_log_opts (mongoc_client_t *client, const mongoc_str
    BSON_OPTIONAL_PARAM (opts);
 
    if (client->topology->single_threaded) {
-      mongoc_topology_set_structured_log_opts (client->topology, opts);
+      mongoc_log_and_monitor_instance_set_structured_log_opts (&client->topology->log_and_monitor, opts);
       return true;
    } else {
       MONGOC_WARNING ("Cannot set structured log options on a pooled client, use "