CDRIVER-2370 monitor OP_MSG document sequences

Update Command Monitoring (also known as APM) to expose the document
sequence sent with an OP_MSG insert, update, or delete. Simulate a
document sequence if the wire protocol uses OP_QUERY commands instead
of OP_MSG.

In MongoDB 3.8 there will be document sequences in the OP_MSG reply to
"find", "aggregate", etc. Anticipate that wire protocol by simulating
a document sequence with command replies.

Author: ajdavis

doc/application-performance-monitoring.rst

@@ -22,33 +22,37 @@ This example program prints:
 .. code-block:: none
 
   Command drop started on 127.0.0.1:
-  { "drop" : "test" }
+  { "drop" : "test", "$db" : "test" }
 
-  Command drop failed:
-  "ns not found"
+  Command drop succeeded:
+  { "ns" : "test.test", "nIndexesWas" : 1, "ok" : 1.0 }
 
   Command insert started on 127.0.0.1:
-  { "insert" : "test", "documents" : [ { "_id" : 1 } ] }
+  { "insert" : "test", "ordered" : true, "$db" : "test" }
+      document 0 { "_id" : 0 }
+      document 1 { "_id" : 1 }
 
   Command insert succeeded:
-  { "ok" : 1, "n" : 1 }
+  { "n" : 1, "ok" : 1.0 }
 
   Command insert started on 127.0.0.1:
-  { "insert" : "test", "documents" : [ { "_id" : 1 } ] }
+  { "insert" : "test", "ordered" : true, "$db" : "test" }
+      document 0: { "_id" : 1 }
 
   Command insert succeeded:
-  { "ok" : 1,
+  {
     "n" : 0,
-    "writeErrors" : [ {
-       "index" : 0, "code" : 11000,
-       "errmsg" : "E11000 duplicate key error"
-  } ] }
+    "writeErrors" : [
+      { "index" : 0, "code" : 11000, "errmsg" : "duplicate key" }
+    ],
+    "ok" : 1.0
+  }
 
-  started: 3
-  succeeded: 2
-  failed: 1
+  started: 4
+  succeeded: 4
+  failed: 0
 
-In older versions of the MongoDB wire protocol, queries and write operations are sent to the server with special `opcodes <https://docs.mongodb.org/manual/reference/mongodb-wire-protocol/#request-opcodes>`_, not as commands. To provide consistent event notifications with any MongoDB version, these legacy opcodes are reported as if they had used modern commands.
+(Depending on your server configuration, messages may include metadata like logical session ids or cluster times that are not shown here.)
 
 The final "insert" command is considered successful, despite the writeError, because the server replied to the overall command with ``"ok": 1``.
 
@@ -126,6 +130,7 @@ The driver discovers the third member, "localhost:27019", and adds it to the top
     :maxdepth: 1
 
     mongoc_apm_callbacks_t
+    mongoc_apm_document_sequence_t
     mongoc_apm_command_failed_t
     mongoc_apm_command_started_t
     mongoc_apm_command_succeeded_t

doc/mongoc_apm_command_started_get_document_sequences.rst

@@ -0,0 +1,33 @@
+:man_page: mongoc_apm_command_started_get_document_sequences
+
+mongoc_apm_command_started_get_document_sequences()
+===================================================
+
+Synopsis
+--------
+
+.. code-block:: c
+
+  void
+  mongoc_apm_command_started_get_document_sequences (
+     const mongoc_apm_command_started_t *event,
+     const mongoc_apm_document_sequence_t **sequences,
+     size_t *n_sequences);
+
+Returns the document sequences sent to the server with this command. The "insert", "update", and "delete" commands have one document sequence; for other commands sequences is NULL and the number of sequences is 0.
+
+In MongoDB 3.4 and older, the "insert", "update", and "delete" commands include one document sequence as a BSON array within the command body. Starting in MongoDB 3.6, these commands each send one document sequence as a separate section of the OP_MSG, and the document sequence is not available in the command body. This function retrieves the document sequence sent to any version of MongoDB.
+
+The data is only valid in the scope of the callback that receives this event; copy it if it will be accessed after the callback returns.
+
+Parameters
+----------
+
+* ``event``: A :symbol:`mongoc_apm_command_started_t`.
+* ``sequences``: An array of pointers to a :symbol:`mongoc_apm_document_sequence_t`.
+* ``n_sequences``: The number of document sequences sent with the command.
+
+See Also
+--------
+
+:doc:`Introduction to Application Performance Monitoring <application-performance-monitoring>`

doc/mongoc_apm_command_started_t.rst

@@ -28,6 +28,7 @@ See Also
     mongoc_apm_command_started_get_command_name
     mongoc_apm_command_started_get_context
     mongoc_apm_command_started_get_database_name
+    mongoc_apm_command_started_get_document_sequences
     mongoc_apm_command_started_get_host
     mongoc_apm_command_started_get_operation_id
     mongoc_apm_command_started_get_request_id

doc/mongoc_apm_command_succeeded_get_document_sequences.rst

@@ -0,0 +1,33 @@
+:man_page: mongoc_apm_command_succeeded_get_document_sequences
+
+mongoc_apm_command_succeeded_get_document_sequences()
+=====================================================
+
+Synopsis
+--------
+
+.. code-block:: c
+
+  void
+  mongoc_apm_command_succeeded_get_document_sequences (
+     const mongoc_apm_command_succeeded_t *event,
+     const mongoc_apm_document_sequence_t **sequences,
+     size_t *n_sequences);
+
+Returns the document sequences sent by the server. Query-like commands such as "find" and "aggregate" return one document sequence, for other commands the document sequence is NULL and the number of documents is 0.
+
+In MongoDB 3.6 and older, all replies include their document sequences as BSON arrays within the reply body. Starting in MongoDB 3.8, document sequences are sent as separate sections of the OP_MSG, and are not available in the reply body. This function retrieves the document sequences sent by any version of MongoDB server.
+
+The data is only valid in the scope of the callback that receives this event; copy it if it will be accessed after the callback returns.
+
+Parameters
+----------
+
+* ``event``: A :symbol:`mongoc_apm_command_succeeded_t`.
+* ``sequences``: An array of pointers to a :symbol:`mongoc_apm_document_sequence_t`.
+* ``n_sequences``: The number of document sequences sent with the command.
+
+See Also
+--------
+
+:doc:`Introduction to Application Performance Monitoring <application-performance-monitoring>`

doc/mongoc_apm_command_succeeded_t.rst

@@ -26,6 +26,7 @@ See Also
 
     mongoc_apm_command_succeeded_get_command_name
     mongoc_apm_command_succeeded_get_context
+    mongoc_apm_command_succeeded_get_document_sequences
     mongoc_apm_command_succeeded_get_duration
     mongoc_apm_command_succeeded_get_host
     mongoc_apm_command_succeeded_get_operation_id

doc/mongoc_apm_document_sequence_t.rst

@@ -0,0 +1,34 @@
+:man_page: mongoc_apm_document_sequence_t
+
+mongoc_apm_document_sequence_t
+==============================
+
+Synopsis
+--------
+
+.. code-block:: c
+
+  typedef struct {
+     int32_t size;
+     const char *identifier;
+     bson_t **documents;
+  } mongoc_apm_document_sequence_t;
+
+Used in Application Performance Monitoring to represent the sequence of BSON documents sent to MongoDB with an "insert", "update", or "delete" command, or to represent the sequence of documents returned by a "find" or "aggregate" command.
+
+See Also
+--------
+
+:doc:`Introduction to Application Performance Monitoring <application-performance-monitoring>`
+
+.. only:: html
+
+  Functions
+  ---------
+
+  .. toctree::
+    :titlesonly:
+    :maxdepth: 1
+
+    mongoc_apm_command_started_get_document_sequences
+    mongoc_apm_command_succeeded_get_document_sequences

examples/example-command-monitoring.c

@@ -18,34 +18,65 @@ void
 command_started (const mongoc_apm_command_started_t *event)
 {
    char *s;
+   const mongoc_apm_document_sequence_t *sequences;
+   size_t n_sequences, i, j;
 
-   s = bson_as_canonical_extended_json (
+   s = bson_as_relaxed_extended_json (
       mongoc_apm_command_started_get_command (event), NULL);
-   printf ("Command %s started on %s:\n%s\n\n",
+
+   printf ("Command %s started on %s:\n%s\n",
            mongoc_apm_command_started_get_command_name (event),
            mongoc_apm_command_started_get_host (event)->host,
            s);
 
-   ((stats_t *) mongoc_apm_command_started_get_context (event))->started++;
-
    bson_free (s);
+
+   /* insert, update, and delete commands include a document sequence */
+   mongoc_apm_command_started_get_document_sequences (
+      event, &sequences, &n_sequences);
+
+   for (i = 0; i < n_sequences; i++) {
+      for (j = 0; j < sequences[i].size; j++) {
+         s = bson_as_relaxed_extended_json (sequences[i].documents[j], NULL);
+         printf ("    document %zi %s\n", j, s);
+         bson_free (s);
+      }
+   }
+
+   ((stats_t *) mongoc_apm_command_started_get_context (event))->started++;
 }
 
 
 void
 command_succeeded (const mongoc_apm_command_succeeded_t *event)
 {
    char *s;
+   const mongoc_apm_document_sequence_t *sequences;
+   size_t n_sequences, i, j;
 
-   s = bson_as_canonical_extended_json (
+   s = bson_as_relaxed_extended_json (
       mongoc_apm_command_succeeded_get_reply (event), NULL);
-   printf ("Command %s succeeded:\n%s\n\n",
+
+   printf ("Command %s succeeded on %s:\n%s\n",
            mongoc_apm_command_succeeded_get_command_name (event),
+           mongoc_apm_command_succeeded_get_host (event)->host,
            s);
 
-   ((stats_t *) mongoc_apm_command_succeeded_get_context (event))->succeeded++;
-
    bson_free (s);
+
+   /* replies to "find" or "aggregate" include a document sequence */
+   mongoc_apm_command_succeeded_get_document_sequences (
+      event, &sequences, &n_sequences);
+
+   for (i = 0; i < n_sequences; i++) {
+      for (j = 0; j < sequences[i].size; j++) {
+         s = bson_as_relaxed_extended_json (sequences[i].documents[j], NULL);
+         printf ("    document %zi %s\n", j, s);
+         bson_free (s);
+      }
+   }
+
+   ((stats_t *) mongoc_apm_command_succeeded_get_context (event))->succeeded++;
 }
 
 
@@ -72,7 +103,7 @@ main (int argc, char *argv[])
    mongoc_collection_t *collection;
    const char *uristr = "mongodb://127.0.0.1/?appname=cmd-monitoring-example";
    const char *collection_name = "test";
-   bson_t doc;
+   bson_t *docs[2];
 
    mongoc_init ();
 
@@ -95,16 +126,17 @@ main (int argc, char *argv[])
    mongoc_client_set_apm_callbacks (
       client, callbacks, (void *) &stats /* context pointer */);
 
-   bson_init (&doc);
-   BSON_APPEND_INT32 (&doc, "_id", 1);
-
    collection = mongoc_client_get_collection (client, "test", collection_name);
    mongoc_collection_drop (collection, NULL);
-   mongoc_collection_insert_one (collection, &doc, NULL, NULL, NULL);
+
+   docs[0] = BCON_NEW ("_id", BCON_INT32 (0));
+   docs[1] = BCON_NEW ("_id", BCON_INT32 (1));
+   mongoc_collection_insert_many (
+      collection, (const bson_t **) docs, 2, NULL, NULL, NULL);
+
    /* duplicate key error on the second insert */
-   mongoc_collection_insert_one (collection, &doc, NULL, NULL, NULL);
+   mongoc_collection_insert_one (collection, docs[0], NULL, NULL, NULL);
 
-   bson_destroy (&doc);
    mongoc_collection_destroy (collection);
    mongoc_apm_callbacks_destroy (callbacks);
    mongoc_client_destroy (client);
@@ -114,6 +146,9 @@ main (int argc, char *argv[])
            stats.succeeded,
            stats.failed);
 
+   bson_destroy (docs[0]);
+   bson_destroy (docs[1]);
+
    mongoc_cleanup ();
 
    return EXIT_SUCCESS;

src/mongoc/mongoc-apm-private.h

@@ -23,9 +23,13 @@
 
 #include <bson.h>
 #include "mongoc-apm.h"
+#include "mongoc-array-private.h"
 
 BSON_BEGIN_DECLS
 
+/* forward decl */
+typedef struct _mongoc_cmd_t mongoc_cmd_t;
+
 struct _mongoc_apm_callbacks_t {
    mongoc_apm_command_started_cb_t started;
    mongoc_apm_command_succeeded_cb_t succeeded;
@@ -54,6 +58,8 @@ struct _mongoc_apm_command_started_t {
    int64_t operation_id;
    const mongoc_host_list_t *host;
    uint32_t server_id;
+   /* for now, an OP_MSG has at most one document sequence */
+   mongoc_apm_document_sequence_t *sequence;
    void *context;
 };
 
@@ -65,6 +71,7 @@ struct _mongoc_apm_command_succeeded_t {
    int64_t operation_id;
    const mongoc_host_list_t *host;
    uint32_t server_id;
+   mongoc_apm_document_sequence_t *sequence;
    void *context;
 };
 
@@ -150,6 +157,12 @@ mongoc_apm_command_started_init (mongoc_apm_command_started_t *event,
                                  uint32_t server_id,
                                  void *context);
 
+void
+mongoc_apm_command_started_init_with_cmd (mongoc_apm_command_started_t *event,
+                                          mongoc_cmd_t *cmd,
+                                          int64_t request_id,
+                                          void *context);
+
 void
 mongoc_apm_command_started_cleanup (mongoc_apm_command_started_t *event);