Load balancer: SDAM and Server Selection (#816)

CDRIVER-4053 Load balancer: SDAM changes
- Add a LoadBalanced topology type and LoadBalancer server type.
- Manually set the topology description to contain one server of type LoadBalancer when opening.
- Prevent updates to the topology description after opening.
- Disable SDAM monitoring for a LoadBalanced topology.
- Emit correct SDAM monitoring events for a LoadBalanced topology.
- Bypass the cooldown period for the single-threaded topology scanner.

CDRIVER-4055 Load balancer: Server Selection
- Always return the one load balancer in server selection.

Author: kevinAlbs

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

@@ -629,7 +629,7 @@ _mongoc_cmd_parts_assemble_mongod (mongoc_cmd_parts_t *parts,
       case MONGOC_TOPOLOGY_LOAD_BALANCED:
       case MONGOC_TOPOLOGY_DESCRIPTION_TYPES:
       default:
-         /* must not call this function w/ sharded or unknown topology type */
+         /* must not call this function w/ sharded, load balanced, or unknown topology type */
          BSON_ASSERT (false);
       }
    } /* if (!parts->is_write_command) */

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

@@ -807,6 +807,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,11 @@ _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_LOAD_BALANCED) {
+      /* Do not proceed to start monitoring threads. */
+      TRACE ("%s", "disabling monitoring for load balanced topology");
+      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,35 @@ _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_LOAD_BALANCED) {
+      mongoc_server_description_t *prev_sd;
+
+      /* LoadBalanced deployments must have exactly one host listed. Otherwise,
+       * an error would have occurred when constructing the topology. */
+      BSON_ASSERT (td->servers->items_len == 1);
+      sd = (mongoc_server_description_t *) mongoc_set_get_item (td->servers, 0);
+      prev_sd = mongoc_server_description_new_copy (sd);
+      BSON_ASSERT (prev_sd);
+      if (td->apm_callbacks.topology_changed) {
+         mongoc_topology_description_destroy (prev_td);
+         _mongoc_topology_description_copy_to (td, prev_td);
+      }
+      sd->type = MONGOC_SERVER_LOAD_BALANCER;
+      _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 +176,14 @@ _mongoc_topology_description_monitor_closed (
    if (td->apm_callbacks.topology_closed) {
       mongoc_apm_topology_closed_t event;
 
+      if (td->type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
+         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, 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.c

@@ -352,6 +352,14 @@ _mongoc_topology_description_server_is_candidate (
       default:
          return false;
       }
+
+   /* Note, there is no call path that leads to the
+    * MONGOC_TOPOLOGY_LOAD_BALANCED case. Server selection for load balanced
+    * topologies bypasses this logic. This silences compiler warnings on
+    * unhandled enum values. */
+   case MONGOC_TOPOLOGY_LOAD_BALANCED:
+      return desc_type == MONGOC_SERVER_LOAD_BALANCER;
+
    default:
       return false;
    }
@@ -712,6 +720,16 @@ mongoc_topology_description_suitable_servers (
          topology->servers, _mongoc_find_suitable_mongos_cb, &data);
    }
 
+   /* Load balanced clusters --
+    * Always select the only server. */
+   if (topology->type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
+      BSON_ASSERT (topology->servers->items_len == 1);
+      server = (mongoc_server_description_t *) mongoc_set_get_item (
+         topology->servers, 0);
+      _mongoc_array_append_val (set, server);
+      goto DONE;
+   }
+
    /* Ways to get here:
     *   - secondary read
     *   - secondary preferred read
@@ -1132,6 +1150,11 @@ mongoc_topology_description_invalidate_server (
 {
    BSON_ASSERT (error);
 
+   if (topology->type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
+      /* Load balancers must never be marked unknown. */
+      return;
+   }
+
    /* send NULL hello reply */
    mongoc_topology_description_handle_hello (
       topology, id, NULL, MONGOC_RTT_UNSET, error);
@@ -1793,6 +1816,8 @@ _mongoc_topology_description_type (mongoc_topology_description_t *topology)
       return "RSWithPrimary";
    case MONGOC_TOPOLOGY_SINGLE:
       return "Single";
+   case MONGOC_TOPOLOGY_LOAD_BALANCED:
+      return "LoadBalanced";
    case MONGOC_TOPOLOGY_DESCRIPTION_TYPES:
    default:
       MONGOC_ERROR ("Invalid mongoc_topology_description_type_t type");

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. */
+      return;
+   }
+
    mongoc_topology_description_handle_hello (&topology->description,
                                              id,
                                              NULL /* hello reply */,
@@ -167,6 +173,12 @@ _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. */
+      return;
+   }
+
    bson_mutex_lock (&topology->mutex);
    sd = mongoc_topology_description_server_by_id (
       &topology->description, id, NULL);
@@ -418,6 +430,15 @@ mongoc_topology_new (const mongoc_uri_t *uri, bool single_threaded)
    if (mongoc_uri_get_option_as_bool (
           topology->uri, MONGOC_URI_LOADBALANCED, false)) {
       init_type = MONGOC_TOPOLOGY_LOAD_BALANCED;
+      if (topology->single_threaded) {
+         /* Cooldown only applies to server monitoring for single-threaded
+          * clients. In load balanced mode, the topology scanner is used to
+          * create connections. The cooldown period does not apply. A network
+          * error to a load balanced connection does not imply subsequent
+          * connection attempts will be to the same server and that a delay
+          * should occur. */
+         _mongoc_topology_bypass_cooldown (topology);
+      }
    } else if (service && !has_directconnection) {
       init_type = MONGOC_TOPOLOGY_UNKNOWN;
    } else if (has_directconnection) {
@@ -945,6 +966,88 @@ 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) {
+      TRACE ("%s",
+             "Server selection performing scan since no connection has "
+             "been established");
+      _mongoc_topology_do_blocking_scan (topology, &scanner_error);
+   }
+
+   if (!node->stream) {
+      /* Use the same error domain / code that is returned in mongoc-cluster.c
+       * when fetching a stream fails. */
+      if (scanner_error.code) {
+         bson_set_error (error,
+                         MONGOC_ERROR_STREAM,
+                         MONGOC_ERROR_STREAM_NOT_ESTABLISHED,
+                         "Could not establish stream for node %s: %s",
+                         node->host.host_and_port,
+                         scanner_error.message);
+      } else {
+         bson_set_error (error,
+                         MONGOC_ERROR_STREAM,
+                         MONGOC_ERROR_STREAM_NOT_ESTABLISHED,
+                         "Could not establish stream for node %s",
+                         node->host.host_and_port);
+      }
+      return 0;
+   }
+
+   return selected_server_id;
+}
+
 /*
  *-------------------------------------------------------------------------
  *
@@ -999,6 +1102,12 @@ 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);
+      return _mongoc_topology_select_server_id_loadbalanced (topology, error);   
+   }
+
    bson_mutex_unlock (&topology->mutex);
 
    heartbeat_msec = topology->description.heartbeat_msec;
@@ -1305,6 +1414,13 @@ _mongoc_topology_update_from_handshake (mongoc_topology_t *topology,
 
    bson_mutex_lock (&topology->mutex);
 
+   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. */
+      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,

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_LOAD_BALANCED;
    }
 
    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_LOAD_BALANCER;
    }
    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");
    }
 }
 

src/libmongoc/tests/json/server_discovery_and_monitoring/load-balanced/discover_load_balancer.json

@@ -0,0 +1,28 @@
+{
+  "description": "Load balancer can be discovered and only has the address property set",
+  "uri": "mongodb://a/?loadBalanced=true",
+  "phases": [
+    {
+      "outcome": {
+        "servers": {
+          "a:27017": {
+            "type": "LoadBalancer",
+            "setName": null,
+            "setVersion": null,
+            "electionId": null,
+            "logicalSessionTimeoutMinutes": null,
+            "minWireVersion": null,
+            "maxWireVersion": null,
+            "topologyVersion": null
+          }
+        },
+        "topologyType": "LoadBalanced",
+        "setName": null,
+        "logicalSessionTimeoutMinutes": null,
+        "maxSetVersion": null,
+        "maxElectionId": null,
+        "compatible": true
+      }
+    }
+  ]
+}