bypass server selection; ensure connections are established in single-threaded

kevinAlbs · kevinAlbs · commit 472242b250ce · 2021-07-14T20:43:21.000-04:00
diff --git a/src/libmongoc/src/mongoc/mongoc-topology-background-monitoring.c b/src/libmongoc/src/mongoc/mongoc-topology-background-monitoring.c
@@ -148,6 +148,7 @@ _mongoc_topology_background_monitoring_start (mongoc_topology_t *topology)
    _mongoc_topology_description_monitor_opening (&topology->description);
    if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
       /* Do not proceed to start monitoring threads. */
+      TRACE ("%s", "disabling monitoring for load balanced topology");
       return;
    }
 
diff --git a/src/libmongoc/src/mongoc/mongoc-topology.c b/src/libmongoc/src/mongoc/mongoc-topology.c
@@ -131,6 +131,12 @@ _mongoc_topology_scanner_setup_err_cb (uint32_t id,
 
    topology = (mongoc_topology_t *) data;
 
+   if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
+      /* In load balanced mode, scanning is only for connection establishment. It must not modify the topology description. */
+      MONGOC_DEBUG ("Ignoring scanner error in load balanced mode");
+      return;
+   }
+
    mongoc_topology_description_handle_hello (&topology->description,
                                              id,
                                              NULL /* hello reply */,
@@ -167,7 +173,14 @@ _mongoc_topology_scanner_cb (uint32_t id,
 
    topology = (mongoc_topology_t *) data;
 
+   if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
+      /* In load balanced mode, scanning is only for connection establishment. It must not modify the topology description. */
+      MONGOC_DEBUG ("Ignoring hello response in callback in load balanced mode");
+      return;
+   }
+
    bson_mutex_lock (&topology->mutex);
+
    sd = mongoc_topology_description_server_by_id (
       &topology->description, id, NULL);
 
@@ -818,7 +831,9 @@ _mongoc_topology_do_blocking_scan (mongoc_topology_t *topology,
 {
    _mongoc_handshake_freeze ();
 
+   // LBTODO: don't even bother locking the mutex, note that this is for single-threaded clients only.
    bson_mutex_lock (&topology->mutex);
+   // LBTODO if this is a load balanced cluster, then do not obey cooldown.
    mongoc_topology_scan_once (topology, true /* obey cooldown */);
    bson_mutex_unlock (&topology->mutex);
    mongoc_topology_scanner_get_error (topology->scanner, error);
@@ -945,6 +960,64 @@ mongoc_topology_select (mongoc_topology_t *topology,
    }
 }
 
+/* Bypasses normal server selection behavior for a load balanced topology. Returns the id of the one load balancer server. Returns 0 on failure.
+ * Successful post-condition: On a single threaded client, a connection will have been established.
+*/
+static uint32_t
+_mongoc_topology_select_server_id_loadbalanced (mongoc_topology_t *topology,
+                                                bson_error_t *error)
+{
+   mongoc_server_description_t *selected_server;
+   int32_t selected_server_id;
+   mongoc_topology_scanner_node_t *node;
+   bson_error_t scanner_error = {0};
+
+   bson_mutex_lock (&topology->mutex);
+
+   BSON_ASSERT (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED);
+
+   /* Emit the opening SDAM events if they have not emitted already. */
+   _mongoc_topology_description_monitor_opening (&topology->description);
+   selected_server =
+      mongoc_topology_description_select (&topology->description,
+                                          MONGOC_SS_WRITE,
+                                          NULL /* read prefs */,
+                                          0 /* local threshold */);
+
+   if (!selected_server) {
+      _mongoc_server_selection_error (
+         "No suitable server found in load balanced deployment", NULL, error);
+      bson_mutex_unlock (&topology->mutex);
+      return 0;
+   }
+
+   selected_server_id = selected_server->id;
+   bson_mutex_unlock (&topology->mutex);
+
+   if (!topology->single_threaded) {
+      return selected_server_id;
+   }
+
+   /* If this is a single threaded topology, we must ensure that a connection is available to this server. Wrapping drivers make the assumption that successful server selection implies a connection is available. */
+   node = mongoc_topology_scanner_get_node (topology->scanner, selected_server_id);
+   if (!node) {
+      _mongoc_server_selection_error ("Topology scanner in invalid state; cannot find load balancer", NULL, error);
+      return 0;
+   }
+
+   if (!node->stream) {
+      MONGOC_DEBUG ("server selection performing scan since no connection has been established");
+      _mongoc_topology_do_blocking_scan (topology, &scanner_error);
+   }
+
+   if (!node->stream) {
+      _mongoc_server_selection_error ("Topology scanner in invalid state; cannot find load balancer", &scanner_error, error);
+      return 0;
+   }
+
+   return selected_server_id;
+}
+
 /*
  *-------------------------------------------------------------------------
  *
@@ -999,6 +1072,13 @@ mongoc_topology_select_server_id (mongoc_topology_t *topology,
       bson_mutex_unlock (&topology->mutex);
       return 0;
    }
+
+   if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
+      bson_mutex_unlock (&topology->mutex);
+      MONGOC_DEBUG ("bypassing server selection for load balanced topology");
+      return _mongoc_topology_select_server_id_loadbalanced (topology, error);   
+   }
+
    bson_mutex_unlock (&topology->mutex);
 
    heartbeat_msec = topology->description.heartbeat_msec;
@@ -1011,21 +1091,6 @@ mongoc_topology_select_server_id (mongoc_topology_t *topology,
    if (topology->single_threaded) {
       _mongoc_topology_description_monitor_opening (&topology->description);
 
-      if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
-         /* Bypass server selection loop. Always select the only server. */
-         // LBTODO: this will not work for PHP. Single threaded server selection must imply connection establishment.
-         selected_server = mongoc_topology_description_select (
-            &topology->description, optype, read_prefs, local_threshold_ms);
-
-         if (!selected_server) {
-            _mongoc_server_selection_error (
-               "No suitable server found in load balanced deployment",
-               NULL,
-               error);
-            return 0;
-         }
-      }
-
       tried_once = false;
       next_update = topology->last_scan + heartbeat_msec * 1000;
       if (next_update < loop_start) {
@@ -1123,18 +1188,6 @@ mongoc_topology_select_server_id (mongoc_topology_t *topology,
       selected_server = mongoc_topology_description_select (
          &topology->description, optype, read_prefs, local_threshold_ms);
 
-      if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
-         /* Bypass server selection loop. Always select the only server. */
-         if (!selected_server) {
-            bson_mutex_unlock (&topology->mutex);
-            _mongoc_server_selection_error (
-               "No suitable server found in load balanced deployment",
-               NULL,
-               error);
-            return 0;
-         }
-      }
-
       if (!selected_server) {
          TRACE (
             "server selection requesting an immediate scan, want %s",
@@ -1304,6 +1357,8 @@ mongoc_topology_invalidate_server (mongoc_topology_t *topology,
 {
    BSON_ASSERT (error);
 
+   // LBTODO: do not update in load balanced mode.
+
    bson_mutex_lock (&topology->mutex);
    mongoc_topology_description_invalidate_server (
       &topology->description, id, error);
@@ -1332,6 +1387,14 @@ _mongoc_topology_update_from_handshake (mongoc_topology_t *topology,
 
    bson_mutex_lock (&topology->mutex);
 
+   // LBTODO: do not update the topology if this is load balanced mode.
+   if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
+      /* In load balanced mode, scanning is only for connection establishment. It must not modify the topology description. */
+      MONGOC_DEBUG ("Ignoring handshake response in load balanced mode");
+      bson_mutex_unlock (&topology->mutex);
+      return true;
+   }
+
    /* return false if server was removed from topology */
    has_server = _mongoc_topology_update_no_lock (sd->id,
                                                  &sd->last_hello_response,
@@ -1743,6 +1806,7 @@ _mongoc_topology_handle_app_error (mongoc_topology_t *topology,
    }
 
    if (type == MONGOC_SDAM_APP_ERROR_NETWORK) {
+      // LBTODO: bypass this in load balanced mode.
       /* Mark server as unknown. */
       mongoc_topology_description_invalidate_server (
          &topology->description, server_id, why);
@@ -1758,6 +1822,7 @@ _mongoc_topology_handle_app_error (mongoc_topology_t *topology,
          return false;
       }
       /* Mark server as unknown. */
+      // LBTODO: bypass this in load balanced mode.
       mongoc_topology_description_invalidate_server (
          &topology->description, server_id, why);
       _mongoc_topology_clear_connection_pool (topology, server_id);
@@ -1812,6 +1877,7 @@ _mongoc_topology_handle_app_error (mongoc_topology_t *topology,
        * error and the error's topologyVersion is strictly greater than the
        * current ServerDescription's topologyVersion it MUST replace the
        * server's description with a ServerDescription of type Unknown. */
+      // LBTODO: bypass this in load balanced mode.
       mongoc_topology_description_invalidate_server (
          &topology->description, server_id, &cmd_error);
 
diff --git a/src/libmongoc/tests/test-mongoc-loadbalanced.c b/src/libmongoc/tests/test-mongoc-loadbalanced.c
@@ -133,6 +133,109 @@ test_loadbalanced_client_uri_validation (void *unused)
    mongoc_client_destroy (client);
 }
 
+/* Test basic connectivity to a load balanced cluster. */
+static void
+test_loadbalanced_connect_single (void *unused)
+{
+   mongoc_client_t *client;
+   char *uristr = loadbalanced_uri ();
+   bson_error_t error;
+   bool ok;
+   mongoc_server_description_t *monitor_sd;
+
+   client = mongoc_client_new (uristr);
+   ok = mongoc_client_command_simple (client,
+                                      "admin",
+                                      tmp_bson ("{'ping': 1}"),
+                                      NULL /* read prefs */,
+                                      NULL /* reply */,
+                                      &error);
+   ASSERT_OR_PRINT (ok, error);
+
+   /* Ensure the server description is unchanged and remains as type LoadBalancer. */
+   monitor_sd = mongoc_client_select_server (
+      client, true /* for writes */, NULL /* read prefs */, &error);
+   ASSERT_OR_PRINT (monitor_sd, error);
+   ASSERT_CMPSTR ("LoadBalancer", mongoc_server_description_type (monitor_sd));
+
+   mongoc_server_description_destroy (monitor_sd);
+   bson_free (uristr);
+   mongoc_client_destroy (client);
+}
+
+static void
+test_loadbalanced_connect_pooled (void *unused)
+{
+   mongoc_client_pool_t *pool;
+   mongoc_client_t *client;
+   char *uristr;
+   mongoc_uri_t *uri;
+   bson_error_t error;
+   bool ok;
+   mongoc_server_description_t *monitor_sd;
+
+   uristr = loadbalanced_uri ();
+   uri = mongoc_uri_new (uristr);
+   pool = mongoc_client_pool_new (uri);
+   client = mongoc_client_pool_pop (pool);
+
+   ok = mongoc_client_command_simple (client,
+                                    "admin",
+                                    tmp_bson ("{'ping': 1}"),
+                                    NULL /* read prefs */,
+                                    NULL /* reply */,
+                                    &error);
+   ASSERT_OR_PRINT (ok, error);
+
+   /* Ensure the server description is unchanged and remains as type LoadBalancer. */
+   monitor_sd = mongoc_client_select_server (
+      client, true /* for writes */, NULL /* read prefs */, &error);
+   ASSERT_OR_PRINT (monitor_sd, error);
+   ASSERT_CMPSTR ("LoadBalancer", mongoc_server_description_type (monitor_sd));
+
+   mongoc_server_description_destroy (monitor_sd);
+   bson_free (uristr);
+   mongoc_uri_destroy (uri);
+   mongoc_client_pool_push (pool, client);
+   mongoc_client_pool_destroy (pool);
+}
+
+/* Ensure that server selection on single threaded clients establishes a
+ * connection against load balanced clusters. */
+static void
+test_loadbalanced_server_selection_establishes_connection_single (void *unused)
+{
+   mongoc_client_t *client;
+   char *uristr = loadbalanced_uri ();
+   bson_error_t error;
+   mongoc_server_description_t *monitor_sd;
+   mongoc_server_description_t *handshake_sd;
+
+   client = mongoc_client_new (uristr);
+   monitor_sd = mongoc_client_select_server (
+      client, true /* for writes */, NULL /* read prefs */, &error);
+   ASSERT_OR_PRINT (monitor_sd, error);
+   ASSERT_CMPSTR ("LoadBalancer", mongoc_server_description_type (monitor_sd));
+
+   /* Ensure that a connection has been established by getting the handshake's
+    * server description. */
+   handshake_sd = mongoc_client_get_handshake_description (
+      client, monitor_sd->id, NULL /* opts */, &error);
+   ASSERT_OR_PRINT (handshake_sd, error);
+   ASSERT_CMPSTR ("Mongos", mongoc_server_description_type (handshake_sd));
+
+   bson_free (uristr);
+   mongoc_server_description_destroy (monitor_sd);
+   mongoc_server_description_destroy (handshake_sd);
+   mongoc_client_destroy (client);
+}
+
+/* Test that the 5 second cooldown does not apply when establishing a new connection to the load balancer after a network error. */
+static void
+test_loadbalanced_network_error_bypasses_cooldown_single (void *unused) {
+
+}
+
 static int
 skip_if_not_loadbalanced (void)
 {
@@ -165,4 +268,26 @@ test_loadbalanced_install (TestSuite *suite)
                       NULL /* ctx */,
                       NULL /* dtor */,
                       NULL);
+
+   TestSuite_AddFull (suite,
+                      "/loadbalanced/connect/single",
+                      test_loadbalanced_connect_single,
+                      NULL /* ctx */,
+                      NULL /* dtor */,
+                      skip_if_not_loadbalanced);
+
+   TestSuite_AddFull (suite,
+                      "/loadbalanced/connect/pooled",
+                      test_loadbalanced_connect_pooled,
+                      NULL /* ctx */,
+                      NULL /* dtor */,
+                      skip_if_not_loadbalanced);
+
+   TestSuite_AddFull (
+      suite,
+      "/loadbalanced/server_selection_establishes_connection/single",
+      test_loadbalanced_server_selection_establishes_connection_single,
+      NULL /* ctx */,
+      NULL /* dtor */,
+      skip_if_not_loadbalanced);
 }

Original file line number	Diff line number	Diff line change
`@@ -148,6 +148,7 @@ _mongoc_topology_background_monitoring_start (mongoc_topology_t *topology)`
`148`	`148`	`_mongoc_topology_description_monitor_opening (&topology->description);`
`149`	`149`	`if (topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED) {`
`150`	`150`	`/* Do not proceed to start monitoring threads. */`
	`151`	`+ TRACE ("%s", "disabling monitoring for load balanced topology");`
`151`	`152`	`return;`
`152`	`153`	`}`
`153`	`154`