CDRIVER-4193 Ensure OP_MSG for handshakes and fix RPC op_egress counters

src/libmongoc/doc/basic-troubleshooting.rst

@@ -17,16 +17,16 @@ The following is a short list of things to check when you have a problem.
 Performance Counters
 --------------------
 
-The MongoDB C driver comes with an optional unique feature to help developers and sysadmins troubleshoot problems in production.
-Performance counters are available for each process using the driver.
+The MongoDB C Driver comes with an optional and unique feature to help developers and sysadmins troubleshoot problems in production.
+Performance counters are available for each process using the C Driver.
 If available, the counters can be accessed outside of the application process via a shared memory segment.
-This means that you can graph statistics about your application process easily from tools like Munin or Nagios.
-Your author often uses ``watch --interval=0.5 -d mongoc-stat $PID`` to monitor an application.
+The counters may be used graph statistics about your application process easily from tools like Munin or Nagios.
+For example, the command ``watch --interval=0.5 -d mongoc-stat $PID`` may be used to monitor an application.
 
-Performance counters are only available on Linux platforms and macOS arm64 platforms supporting shared memory segments.
-On supported platforms they are enabled by default.
-Applications can be built without the counters by specifying the cmake option ``-DENABLE_SHM_COUNTERS=OFF``. Additionally, if
-performance counters are already compiled, they can be disabled at runtime by specifying the environment variable ``MONGOC_DISABLE_SHM``.
+Performance counters are only available on Linux platforms and macOS arm64 platforms that support shared memory segments.
+On supported platforms, they are enabled by default.
+Applications can be built without the counters by specifying the cmake option ``-DENABLE_SHM_COUNTERS=OFF``.
+Additionally, if performance counters are already compiled, they can be disabled at runtime by specifying the environment variable ``MONGOC_DISABLE_SHM``.
 
 Performance counters keep track of the following:
 
@@ -37,6 +37,10 @@ Performance counters keep track of the following:
 * Authentication successes and failures.
 * Number of wire protocol errors.
 
+.. note::
+  An operation is considered "sent" when one or more bytes of the corresponding message is written to the stream, regardless of whether the entire message is successfully written or if the operation ultimately succeeds or fails.
+  This does not include bytes that may be written during the stream connection process, such as TLS handshake messages.
+
 To access counters for a given process, simply provide the process id to the ``mongoc-stat`` program installed with the MongoDB C Driver.
 
 .. code-block:: none

src/libmongoc/src/mongoc/mongoc-async-cmd.c

@@ -347,6 +347,7 @@ _mongoc_async_cmd_phase_send (mongoc_async_cmd_t *acmd)
       used_temp_iovec = true;
    }
 
+   _mongoc_rpc_op_egress_inc (&acmd->rpc);
    bytes = mongoc_stream_writev (acmd->stream, iovec, niovec, 0);
 
    if (used_temp_iovec) {

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

@@ -226,11 +226,17 @@ mongoc_client_connect (bool buffered,
                        bson_error_t *error);
 
 
-/* Returns true if a versioned server API has been selected,
- * otherwise returns false. */
+/* Returns true if a versioned server API has been selected, otherwise returns
+ * false. */
 bool
 mongoc_client_uses_server_api (const mongoc_client_t *client);
 
+
+/* Returns true if load balancing mode has been selected, otherwise returns
+ * false. */
+bool
+mongoc_client_uses_loadbalanced (const mongoc_client_t *client);
+
 BSON_END_DECLS
 
 #endif /* MONGOC_CLIENT_PRIVATE_H */

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

@@ -3184,3 +3184,9 @@ mongoc_client_uses_server_api (const mongoc_client_t *client)
 {
    return mongoc_topology_uses_server_api (client->topology);
 }
+
+bool
+mongoc_client_uses_loadbalanced (const mongoc_client_t *client)
+{
+   return mongoc_topology_uses_loadbalanced (client->topology);
+}

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

@@ -234,11 +234,16 @@ _mongoc_cluster_get_auth_cmd_x509 (const mongoc_uri_t *uri,
                                    bson_t *cmd /* OUT */,
                                    bson_error_t *error /* OUT */);
 
-/* Returns true if a versioned server API has been selected,
- * otherwise returns false. */
+/* Returns true if a versioned server API has been selected, otherwise returns
+ * false. */
 bool
 mongoc_cluster_uses_server_api (const mongoc_cluster_t *cluster);
 
+/* Returns true if load balancing mode has been selected, otherwise returns
+ * false. */
+bool
+mongoc_cluster_uses_loadbalanced (const mongoc_cluster_t *cluster);
+
 #ifdef MONGOC_ENABLE_CRYPTO
 void
 _mongoc_cluster_init_scram (const mongoc_cluster_t *cluster,

src/libmongoc/src/mongoc/mongoc-cluster.c

@@ -312,6 +312,7 @@ mongoc_cluster_run_command_opquery (mongoc_cluster_t *cluster,
    /*
     * send and receive
     */
+   _mongoc_rpc_op_egress_inc (&rpc);
    if (!_mongoc_stream_writev_full (stream,
                                     cluster->iov.data,
                                     cluster->iov.len,
@@ -635,6 +636,14 @@ mongoc_cluster_run_command_monitored (mongoc_cluster_t *cluster,
 }
 
 
+static bool
+_should_use_op_msg (const mongoc_cluster_t *cluster)
+{
+   return mongoc_cluster_uses_server_api (cluster) ||
+          mongoc_cluster_uses_loadbalanced (cluster);
+}
+
+
 /*
  *--------------------------------------------------------------------------
  *
@@ -676,7 +685,7 @@ mongoc_cluster_run_command_private (mongoc_cluster_t *cluster,
 
    server_stream = cmd->server_stream;
 
-   if (mongoc_cluster_uses_server_api (cluster) ||
+   if (_should_use_op_msg (cluster) ||
        server_stream->sd->max_wire_version >= WIRE_VERSION_MIN) {
       retval = mongoc_cluster_run_opmsg (cluster, cmd, reply, error);
    } else {
@@ -818,24 +827,22 @@ _stream_run_hello (mongoc_cluster_t *cluster,
    to either an op_msg or op_query: */
    memset (&hello_cmd, 0, sizeof (hello_cmd));
 
-
-   hello_cmd.db_name = "admin";
-   hello_cmd.command = &handshake_command;
-   hello_cmd.command_name = _mongoc_get_command_name (&handshake_command);
-   hello_cmd.server_stream = server_stream;
-
-   hello_cmd.is_acknowledged = true;
-
    /* Use OP_QUERY for the handshake, unless the user has specified an
     * API version; the correct hello_cmd has already been selected: */
-   if (!mongoc_cluster_uses_server_api (cluster)) {
+   if (!_should_use_op_msg (cluster)) {
       /* Complete OPCODE_QUERY setup: */
       hello_cmd.query_flags = MONGOC_QUERY_SECONDARY_OK;
    } else {
       /* We're using OP_MSG, and require some additional doctoring: */
       bson_append_utf8 (&handshake_command, "$db", 3, "admin", 5);
    }
 
+   hello_cmd.db_name = "admin";
+   hello_cmd.command = &handshake_command;
+   hello_cmd.command_name = _mongoc_get_command_name (&handshake_command);
+   hello_cmd.server_stream = server_stream;
+   hello_cmd.is_acknowledged = true;
+
    if (!mongoc_cluster_run_command_private (
           cluster, &hello_cmd, &reply, error)) {
       if (negotiate_sasl_supported_mechs) {
@@ -1345,9 +1352,17 @@ _mongoc_cluster_auth_node_x509 (mongoc_cluster_t *cluster,
 bool
 mongoc_cluster_uses_server_api (const mongoc_cluster_t *cluster)
 {
+   BSON_ASSERT_PARAM (cluster);
    return mongoc_client_uses_server_api (cluster->client);
 }
 
+bool
+mongoc_cluster_uses_loadbalanced (const mongoc_cluster_t *cluster)
+{
+   BSON_ASSERT_PARAM (cluster);
+   return mongoc_client_uses_loadbalanced (cluster->client);
+}
+
 #ifdef MONGOC_ENABLE_CRYPTO
 void
 _mongoc_cluster_init_scram (const mongoc_cluster_t *cluster,
@@ -3287,6 +3302,7 @@ mongoc_cluster_legacy_rpc_sendv_to_server (
       GOTO (done);
    }
 
+   _mongoc_rpc_op_egress_inc (rpc);
    if (!_mongoc_stream_writev_full (server_stream->stream,
                                     cluster->iov.data,
                                     cluster->iov.len,
@@ -3561,11 +3577,14 @@ mongoc_cluster_run_opmsg (mongoc_cluster_t *cluster,
          }
       }
    }
+
+   _mongoc_rpc_op_egress_inc (&rpc);
    ok = _mongoc_stream_writev_full (server_stream->stream,
                                     (mongoc_iovec_t *) cluster->iov.data,
                                     cluster->iov.len,
                                     cluster->sockettimeoutms,
                                     error);
+
    if (!ok) {
       /* add info about the command to writev_full's error message */
       RUN_CMD_ERR_DECORATE;

src/libmongoc/src/mongoc/mongoc-cmd.c

@@ -837,6 +837,7 @@ mongoc_cmd_parts_assemble (mongoc_cmd_parts_t *parts,
    }
 
    if (mongoc_client_uses_server_api (parts->client) ||
+       mongoc_client_uses_loadbalanced (parts->client) ||
        server_stream->sd->max_wire_version >= WIRE_VERSION_MIN) {
       if (!bson_has_field (parts->body, "$db")) {
          BSON_APPEND_UTF8 (&parts->extra, "$db", parts->assembled.db_name);

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

@@ -179,9 +179,23 @@ enum {
          bson_atomic_int64_exchange (counter, 0, bson_memory_order_seq_cst);   \
       }                                                                        \
       bson_atomic_thread_fence ();                                             \
+   }                                                                           \
+   static BSON_INLINE int32_t mongoc_counter_##ident##_count (void)            \
+   {                                                                           \
+      int32_t _sum = 0;                                                        \
+      uint32_t _i;                                                             \
+      for (_i = 0; _i < _mongoc_get_cpu_count (); _i++) {                      \
+         const int64_t *counter =                                              \
+            &BSON_CONCAT (__mongoc_counter_, ident)                            \
+                .cpus[_i]                                                      \
+                .slots[BSON_CONCAT (COUNTER_, ident) % SLOTS_PER_CACHELINE];   \
+         _sum += bson_atomic_int64_fetch (counter, bson_memory_order_seq_cst); \
+      }                                                                        \
+      return _sum;                                                             \
    }
 #include "mongoc-counters.defs"
 #undef COUNTER
+
 #else
 /* when counters are disabled, these functions are no-ops */
 #define COUNTER(ident, Category, Name, Description)                   \
@@ -196,6 +210,9 @@ enum {
    }                                                                  \
    static BSON_INLINE void mongoc_counter_##ident##_reset (void)      \
    {                                                                  \
+   }                                                                  \
+   static BSON_INLINE void mongoc_counter_##ident##_count (void)      \
+   {                                                                  \
    }
 #include "mongoc-counters.defs"
 #undef COUNTER

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

@@ -136,6 +136,8 @@ BSON_STATIC_ASSERT2 (sizeof_reply_header,
 void
 _mongoc_rpc_gather (mongoc_rpc_t *rpc, mongoc_array_t *array);
 void
+_mongoc_rpc_op_egress_inc (const mongoc_rpc_t *rpc);
+void
 _mongoc_rpc_swab_to_le (mongoc_rpc_t *rpc);
 void
 _mongoc_rpc_swab_from_le (mongoc_rpc_t *rpc);

src/libmongoc/src/mongoc/mongoc-rpc.c

@@ -577,67 +577,44 @@
 #undef CHECKSUM_FIELD
 
 
-/*
- *--------------------------------------------------------------------------
- *
- * _mongoc_rpc_gather --
- *
- *       Takes a (native endian) rpc struct and gathers the buffer.
- *       Caller should swab to little endian after calling gather.
- *
- *       Gather, swab, compress write.
- *       Read, scatter, uncompress, swab
- *
- *--------------------------------------------------------------------------
- */
-
 void
 _mongoc_rpc_gather (mongoc_rpc_t *rpc, mongoc_array_t *array)
 {
-   mongoc_counter_op_egress_total_inc ();
    switch ((mongoc_opcode_t) rpc->header.opcode) {
    case MONGOC_OPCODE_REPLY:
       _mongoc_rpc_gather_reply (&rpc->reply, &rpc->header, array);
       return;
 
    case MONGOC_OPCODE_MSG:
       _mongoc_rpc_gather_msg (&rpc->msg, &rpc->header, array);
-      mongoc_counter_op_egress_msg_inc ();
       return;
 
    case MONGOC_OPCODE_UPDATE:
       _mongoc_rpc_gather_update (&rpc->update, &rpc->header, array);
-      mongoc_counter_op_egress_update_inc ();
       return;
 
    case MONGOC_OPCODE_INSERT:
       _mongoc_rpc_gather_insert (&rpc->insert, &rpc->header, array);
-      mongoc_counter_op_egress_insert_inc ();
       return;
 
    case MONGOC_OPCODE_QUERY:
       _mongoc_rpc_gather_query (&rpc->query, &rpc->header, array);
-      mongoc_counter_op_egress_query_inc ();
       return;
 
    case MONGOC_OPCODE_GET_MORE:
       _mongoc_rpc_gather_get_more (&rpc->get_more, &rpc->header, array);
-      mongoc_counter_op_egress_getmore_inc ();
       return;
 
    case MONGOC_OPCODE_DELETE:
       _mongoc_rpc_gather_delete (&rpc->delete_, &rpc->header, array);
-      mongoc_counter_op_egress_delete_inc ();
       return;
 
    case MONGOC_OPCODE_KILL_CURSORS:
       _mongoc_rpc_gather_kill_cursors (&rpc->kill_cursors, &rpc->header, array);
-      mongoc_counter_op_egress_killcursors_inc ();
       return;
 
    case MONGOC_OPCODE_COMPRESSED:
       _mongoc_rpc_gather_compressed (&rpc->compressed, &rpc->header, array);
-      mongoc_counter_op_egress_compressed_inc ();
       return;
 
    default:
@@ -648,6 +625,67 @@ _mongoc_rpc_gather (mongoc_rpc_t *rpc, mongoc_array_t *array)
 }
 
 
+void
+_mongoc_rpc_op_egress_inc (const mongoc_rpc_t *rpc)
+{
+   mongoc_opcode_t opcode =
+      (mongoc_opcode_t) BSON_UINT32_FROM_LE (rpc->header.opcode);
+
+   if (opcode == MONGOC_OPCODE_COMPRESSED) {
+      mongoc_counter_op_egress_compressed_inc ();
+      mongoc_counter_op_egress_total_inc ();
+
+      opcode = (mongoc_opcode_t) BSON_UINT32_FROM_LE (
+         rpc->compressed.original_opcode);
+   }
+
+   mongoc_counter_op_egress_total_inc ();
+
+   switch (opcode) {
+   case MONGOC_OPCODE_REPLY:
+      return;
+
+   case MONGOC_OPCODE_MSG:
+      mongoc_counter_op_egress_msg_inc ();
+      return;
+
+   case MONGOC_OPCODE_UPDATE:
+      mongoc_counter_op_egress_update_inc ();
+      return;
+
+   case MONGOC_OPCODE_INSERT:
+      mongoc_counter_op_egress_insert_inc ();
+      return;
+
+   case MONGOC_OPCODE_QUERY:
+      mongoc_counter_op_egress_query_inc ();
+      return;
+
+   case MONGOC_OPCODE_GET_MORE:
+      mongoc_counter_op_egress_getmore_inc ();
+      return;
+
+   case MONGOC_OPCODE_DELETE:
+      mongoc_counter_op_egress_delete_inc ();
+      return;
+
+   case MONGOC_OPCODE_KILL_CURSORS:
+      mongoc_counter_op_egress_killcursors_inc ();
+      return;
+
+   case MONGOC_OPCODE_COMPRESSED:
+      MONGOC_WARNING ("Compressed an OP_COMPRESSED message!?");
+      BSON_ASSERT (false);
+      return;
+
+   default:
+      MONGOC_WARNING ("Unknown rpc type: 0x%08x", opcode);
+      BSON_ASSERT (false);
+      break;
+   }
+}
+
+
 void
 _mongoc_rpc_swab_to_le (mongoc_rpc_t *rpc)
 {