Disable SDAM on LoadBalanced, add events

Author: kevinAlbs

src/libmongoc/src/mongoc/mongoc-server-description-private.h

@@ -54,6 +54,7 @@ typedef enum {
    MONGOC_SERVER_RS_ARBITER,
    MONGOC_SERVER_RS_OTHER,
    MONGOC_SERVER_RS_GHOST,
+   MONGOC_SERVER_LOADBALANCER,
    MONGOC_SERVER_DESCRIPTION_TYPES,
 } mongoc_server_description_type_t;
 
@@ -107,7 +108,8 @@ struct _mongoc_server_description_t {
    1. a monitor receives a network error
    2. an app thread receives any network error before completing a handshake
    3. an app thread receives a non-timeout network error after the handshake
-   4. an app thread receives a "not primary" or "node is recovering" error from a
+   4. an app thread receives a "not primary" or "node is recovering" error from
+   a
    pre-4.2 server.
    */
    uint32_t generation;

src/libmongoc/src/mongoc/mongoc-server-description.c

@@ -361,6 +361,8 @@ mongoc_server_description_type (const mongoc_server_description_t *description)
       return "RSOther";
    case MONGOC_SERVER_RS_GHOST:
       return "RSGhost";
+   case MONGOC_SERVER_LOADBALANCER:
+      return "LoadBalancer";
    case MONGOC_SERVER_DESCRIPTION_TYPES:
    default:
       MONGOC_ERROR ("Invalid mongoc_server_description_t type");
@@ -803,6 +805,9 @@ mongoc_server_description_new_copy (
                                               &description->error);
    } else {
       mongoc_server_description_reset (copy);
+      /* preserve the original server description type, which is manually set
+       * for a LoadBalancer server */
+      copy->type = description->type;
    }
 
    /* Preserve the error */

src/libmongoc/src/mongoc/mongoc-topology-background-monitoring.c

@@ -146,6 +146,10 @@ _mongoc_topology_background_monitoring_start (mongoc_topology_t *topology)
 
    _mongoc_handshake_freeze ();
    _mongoc_topology_description_monitor_opening (&topology->description);
+   if (topology->description.type == MONGOC_TOPOLOGY_LOADBALANCED) {
+      /* Do not proceed to start monitoring threads. */
+      return;
+   }
 
    /* Reconcile to create the first server monitors. */
    _mongoc_topology_background_monitoring_reconcile (topology);

src/libmongoc/src/mongoc/mongoc-topology-description-apm.c

@@ -27,8 +27,7 @@
 /* ServerOpeningEvent */
 void
 _mongoc_topology_description_monitor_server_opening (
-   const mongoc_topology_description_t *td,
-   mongoc_server_description_t *sd)
+   const mongoc_topology_description_t *td, mongoc_server_description_t *sd)
 {
    if (td->apm_callbacks.server_opening && !sd->opened) {
       mongoc_apm_server_opening_t event;
@@ -115,10 +114,33 @@ _mongoc_topology_description_monitor_opening (mongoc_topology_description_t *td)
    for (i = 0; i < td->servers->items_len; i++) {
       sd = (mongoc_server_description_t *) mongoc_set_get_item (td->servers,
                                                                 (int) i);
-
       _mongoc_topology_description_monitor_server_opening (td, sd);
    }
 
+   /* If this is a load balanced topology:
+    * - update the one server description to be LoadBalancer
+    * - emit a server changed event Unknown => LoadBalancer
+    * - emit a topology changed event
+    */
+   if (td->type == MONGOC_TOPOLOGY_LOADBALANCED) {
+      mongoc_server_description_t *prev_sd;
+
+      /* LoadBalanced deployments must have exactly one host listed. */
+      BSON_ASSERT (td->servers->items_len == 1);
+      sd = (mongoc_server_description_t *) mongoc_set_get_item (td->servers,
+                                                                (int) 0);
+      prev_sd = mongoc_server_description_new_copy (sd);
+      if (td->apm_callbacks.topology_changed) {
+         _mongoc_topology_description_copy_to (td, prev_td);
+      }
+      sd->type = MONGOC_SERVER_LOADBALANCER;
+      _mongoc_topology_description_monitor_server_changed (td, prev_sd, sd);
+      mongoc_server_description_destroy (prev_sd);
+      if (td->apm_callbacks.topology_changed) {
+         _mongoc_topology_description_monitor_changed (prev_td, td);
+      }
+   }
+
    if (prev_td) {
       mongoc_topology_description_destroy (prev_td);
       bson_free (prev_td);
@@ -152,6 +174,15 @@ _mongoc_topology_description_monitor_closed (
    if (td->apm_callbacks.topology_closed) {
       mongoc_apm_topology_closed_t event;
 
+      if (td->type == MONGOC_TOPOLOGY_LOADBALANCED) {
+         mongoc_server_description_t *sd;
+
+         /* LoadBalanced deployments must have exactly one host listed. */
+         BSON_ASSERT (td->servers->items_len == 1);
+         sd = (mongoc_server_description_t *) mongoc_set_get_item (td->servers,
+                                                                   (int) 0);
+         _mongoc_topology_description_monitor_server_closed (td, sd);
+      }
       bson_oid_copy (&td->topology_id, &event.topology_id);
       event.context = td->apm_context;
       td->apm_callbacks.topology_closed (&event);

src/libmongoc/src/mongoc/mongoc-topology-description-private.h

@@ -32,6 +32,7 @@ typedef enum {
    MONGOC_TOPOLOGY_RS_NO_PRIMARY,
    MONGOC_TOPOLOGY_RS_WITH_PRIMARY,
    MONGOC_TOPOLOGY_SINGLE,
+   MONGOC_TOPOLOGY_LOADBALANCED,
    MONGOC_TOPOLOGY_DESCRIPTION_TYPES
 } mongoc_topology_description_type_t;
 

src/libmongoc/src/mongoc/mongoc-topology-description.c

@@ -2139,6 +2139,8 @@ mongoc_topology_description_type (const mongoc_topology_description_t *td)
       return "ReplicaSetWithPrimary";
    case MONGOC_TOPOLOGY_SINGLE:
       return "Single";
+   case MONGOC_TOPOLOGY_LOADBALANCED:
+      return "LoadBalanced";
    case MONGOC_TOPOLOGY_DESCRIPTION_TYPES:
    default:
       fprintf (stderr, "ERROR: Unknown topology type %d\n", td->type);

src/libmongoc/src/mongoc/mongoc-topology.c

@@ -413,7 +413,10 @@ mongoc_topology_new (const mongoc_uri_t *uri, bool single_threaded)
       has_directconnection &&
       mongoc_uri_get_option_as_bool (uri, MONGOC_URI_DIRECTCONNECTION, false);
    hl = mongoc_uri_get_hosts (topology->uri);
-   if (service && !has_directconnection) {
+   if (mongoc_uri_get_option_as_bool (
+          topology->uri, MONGOC_URI_LOADBALANCED, false)) {
+      init_type = MONGOC_TOPOLOGY_LOADBALANCED;
+   } else if (service && !has_directconnection) {
       init_type = MONGOC_TOPOLOGY_UNKNOWN;
    } else if (has_directconnection) {
       if (directconnection) {
@@ -653,11 +656,6 @@ mongoc_topology_should_rescan_srv (mongoc_topology_t *topology)
       return false;
    }
 
-   /* TODO: rely on topology->description.type instead of URI option. */
-   if (mongoc_uri_get_option_as_bool (
-          topology->uri, MONGOC_URI_LOADBALANCED, false)) {
-      return false;
-   }
 
    if ((topology->description.type != MONGOC_TOPOLOGY_SHARDED) &&
        (topology->description.type != MONGOC_TOPOLOGY_UNKNOWN)) {
@@ -818,6 +816,10 @@ _mongoc_topology_do_blocking_scan (mongoc_topology_t *topology,
 {
    _mongoc_handshake_freeze ();
 
+   if (topology->description.type == MONGOC_TOPOLOGY_LOADBALANCED) {
+      MONGOC_DEBUG ("bypassing server selection logic for single threaded");
+      return;
+   }
    bson_mutex_lock (&topology->mutex);
    mongoc_topology_scan_once (topology, true /* obey cooldown */);
    bson_mutex_unlock (&topology->mutex);

src/libmongoc/tests/json-test.c

@@ -52,6 +52,8 @@ topology_type_from_test (const char *type)
       return MONGOC_TOPOLOGY_SINGLE;
    } else if (strcmp (type, "Sharded") == 0) {
       return MONGOC_TOPOLOGY_SHARDED;
+   } else if (strcmp (type, "LoadBalanced") == 0) {
+      return MONGOC_TOPOLOGY_LOADBALANCED;
    }
 
    fprintf (stderr, "can't parse this: %s", type);
@@ -80,6 +82,8 @@ server_type_from_test (const char *type)
       return MONGOC_SERVER_RS_GHOST;
    } else if (strcmp (type, "Unknown") == 0) {
       return MONGOC_SERVER_UNKNOWN;
+   } else if (strcmp (type, "LoadBalancer") == 0) {
+      return MONGOC_SERVER_LOADBALANCER;
    }
    fprintf (stderr, "ERROR: Unknown server type %s\n", type);
    BSON_ASSERT (0);
@@ -175,8 +179,7 @@ server_description_by_hostname (mongoc_topology_description_t *topology,
  *-----------------------------------------------------------------------
  */
 void
-process_sdam_test_hello_responses (bson_t *phase,
-                                   mongoc_topology_t *topology)
+process_sdam_test_hello_responses (bson_t *phase, mongoc_topology_t *topology)
 {
    mongoc_topology_description_t *td;
    mongoc_server_description_t *sd;
@@ -190,7 +193,7 @@ process_sdam_test_hello_responses (bson_t *phase,
       description = bson_iter_utf8 (&phase_field_iter, NULL);
       MONGOC_DEBUG ("phase: %s", description);
    }
-   /* grab hello responses out and feed them to topology */
+   /* grab hello responses (if present) and feed them to topology */
    if (bson_iter_init_find (&phase_field_iter, phase, "responses")) {
       bson_t hellos;
       bson_t hello;
@@ -317,9 +320,6 @@ process_sdam_test_hello_responses (bson_t *phase,
                                             generation);
          bson_mutex_unlock (&topology->mutex);
       }
-   } else {
-      test_error (
-         "test does not have 'responses' or 'applicationErrors' array");
    }
 }
 
@@ -1739,9 +1739,8 @@ run_json_general_test (const json_test_config_t *config)
 
       /* expect "operation was interrupted", ignore "command not found" or "is
        * not supported" */
-      if (!r &&
-          (error.domain != MONGOC_ERROR_SERVER ||
-           (error.code != 11601 && error.code != 59)) &&
+      if (!r && (error.domain != MONGOC_ERROR_SERVER ||
+                 (error.code != 11601 && error.code != 59)) &&
           (strstr (error.message, "is unsupported") == NULL)) {
          MONGOC_WARNING ("Error in killAllSessions: %s", error.message);
       }

src/libmongoc/tests/test-mongoc-sdam.c

@@ -6,6 +6,7 @@
 
 #include "mongoc/mongoc-client-private.h"
 #include "mongoc/mongoc-topology-private.h"
+#include "mongoc/mongoc-topology-description-apm-private.h"
 #include "test-libmongoc.h"
 
 #ifdef BSON_HAVE_STRINGS_H
@@ -33,6 +34,8 @@ _topology_has_description (mongoc_topology_description_t *topology,
          if (set_name) {
             BSON_ASSERT (sd->set_name);
             ASSERT_CMPSTR (sd->set_name, set_name);
+         } else if (sd->set_name) {
+            test_error ("expected NULL setName, got: %s", sd->set_name);
          }
       } else if (strcmp ("type", bson_iter_key (&server_iter)) == 0) {
          server_type = bson_iter_utf8 (&server_iter, NULL);
@@ -86,6 +89,41 @@ _topology_has_description (mongoc_topology_description_t *topology,
                       BSON_ITER_HOLDS_INT32 (&iter));
          expected_generation = bson_iter_int32 (&iter);
          ASSERT_CMPINT32 (expected_generation, ==, sd->generation);
+      } else if (strcmp ("logicalSessionTimeoutMinutes",
+                         bson_iter_key (&server_iter)) == 0) {
+         if (BSON_ITER_HOLDS_NULL (&server_iter)) {
+            if (sd->session_timeout_minutes != MONGOC_NO_SESSIONS) {
+               test_error ("ERROR: expected unset value for "
+                           "logicalSessionTimeoutMinutes but got: %" PRId64,
+                           sd->session_timeout_minutes);
+            }
+         } else {
+            ASSERT_CMPINT64 (bson_iter_as_int64 (&server_iter),
+                             ==,
+                             sd->session_timeout_minutes);
+         }
+      } else if (strcmp ("minWireVersion", bson_iter_key (&server_iter)) == 0) {
+         if (BSON_ITER_HOLDS_NULL (&server_iter)) {
+            if (sd->min_wire_version != 0) {
+               test_error ("ERROR: expected unset value for minWireVersion but "
+                           "got: %" PRId32,
+                           sd->min_wire_version);
+            }
+         } else {
+            ASSERT_CMPINT32 (
+               bson_iter_int32 (&server_iter), ==, sd->min_wire_version);
+         }
+      } else if (strcmp ("maxWireVersion", bson_iter_key (&server_iter)) == 0) {
+         if (BSON_ITER_HOLDS_NULL (&server_iter)) {
+            if (sd->max_wire_version != 0) {
+               test_error ("ERROR: expected unset value for maxWireVersion but "
+                           "got: %" PRId32,
+                           sd->max_wire_version);
+            }
+         } else {
+            ASSERT_CMPINT32 (
+               bson_iter_int32 (&server_iter), ==, sd->max_wire_version);
+         }
       } else {
          fprintf (
             stderr, "ERROR: unparsed field %s\n", bson_iter_key (&server_iter));
@@ -129,6 +167,11 @@ test_sdam_cb (bson_t *test)
    bson_iter_bson (&iter, &phases);
    bson_iter_init (&phase_iter, &phases);
 
+   /* LoadBalanced topologies change the server from Unknown to LoadBalancer
+    * when SDAM monitoring begins. Force an opening, which would occur on the
+    * first operation on the client. */
+   _mongoc_topology_description_monitor_opening (td);
+
    while (bson_iter_next (&phase_iter)) {
       bson_iter_bson (&phase_iter, &phase);
 
@@ -160,6 +203,10 @@ test_sdam_cb (bson_t *test)
             if (set_name) {
                BSON_ASSERT (td->set_name);
                ASSERT_CMPSTR (td->set_name, set_name);
+            } else {
+               if (td->set_name) {
+                  test_error ("expected NULL setName, got: %s", td->set_name);
+               }
             }
          } else if (strcmp ("topologyType", bson_iter_key (&outcome_iter)) ==
                     0) {
@@ -187,13 +234,27 @@ test_sdam_cb (bson_t *test)
             }
          } else if (strcmp ("maxSetVersion", bson_iter_key (&outcome_iter)) ==
                     0) {
-            ASSERT_CMPINT64 (
-               bson_iter_as_int64 (&outcome_iter), ==, td->max_set_version);
+            if (BSON_ITER_HOLDS_NULL (&outcome_iter)) {
+               if (td->max_set_version != MONGOC_NO_SET_VERSION) {
+                  test_error ("ERROR: expected unset value for maxSetVersion "
+                              "but got: %" PRId64,
+                              td->max_set_version);
+               }
+            } else {
+               ASSERT_CMPINT64 (
+                  bson_iter_as_int64 (&outcome_iter), ==, td->max_set_version);
+            }
          } else if (strcmp ("maxElectionId", bson_iter_key (&outcome_iter)) ==
                     0) {
             const bson_oid_t *expected_oid;
+            bson_oid_t zeroed = {0};
+
             expected_oid = bson_iter_oid (&outcome_iter);
 
+            if (expected_oid == NULL) {
+               expected_oid = &zeroed;
+            }
+
             if (!bson_oid_equal (expected_oid, &td->max_election_id)) {
                char expected_oid_str[25];
                char actual_oid_str[25];
@@ -205,6 +266,19 @@ test_sdam_cb (bson_t *test)
                            expected_oid_str,
                            actual_oid_str);
             }
+         } else if (strcmp ("logicalSessionTimeoutMinutes",
+                            bson_iter_key (&outcome_iter)) == 0) {
+            if (BSON_ITER_HOLDS_NULL (&outcome_iter)) {
+               if (td->session_timeout_minutes != MONGOC_NO_SESSIONS) {
+                  test_error ("ERROR: expected unset value for "
+                              "logicalSessionTimeoutMinutes but got: %" PRId64,
+                              td->session_timeout_minutes);
+               }
+            } else {
+               ASSERT_CMPINT64 (bson_iter_as_int64 (&outcome_iter),
+                                ==,
+                                td->session_timeout_minutes);
+            }
          } else {
             fprintf (stderr,
                      "ERROR: unparsed test field %s\n",
@@ -481,6 +555,10 @@ test_all_spec_tests (TestSuite *suite)
                                        TestSuite_CheckLive,
                                        test_framework_skip_if_no_crypto,
                                        test_framework_skip_if_slow);
+
+   ASSERT (realpath (JSON_DIR "/server_discovery_and_monitoring/load-balanced",
+                     resolved));
+   install_json_test_suite (suite, resolved, &test_sdam_cb);
 }
 
 static void
@@ -773,8 +851,9 @@ test_prose_rtt (void *unused)
     * RTT_TEST_TIMEOUT_SEC seconds, consider it a failure. */
    satisfied = false;
    start_us = bson_get_monotonic_time ();
-   while (!satisfied && bson_get_monotonic_time () <
-                           start_us + RTT_TEST_TIMEOUT_SEC * 1000 * 1000) {
+   while (!satisfied &&
+          bson_get_monotonic_time () <
+             start_us + RTT_TEST_TIMEOUT_SEC * 1000 * 1000) {
       mongoc_server_description_t *sd;
 
       sd = mongoc_client_select_server (