CDRIVER-4054 support sessions in loadbalanced mode (#821)

Author: kevinAlbs

src/libmongoc/CMakeLists.txt

@@ -960,6 +960,7 @@ set (test-libmongoc-sources
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-interrupt.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-linux-distro-scanner.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-list.c
+   ${PROJECT_SOURCE_DIR}/tests/test-mongoc-loadbalanced.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-log.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-long-namespace.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-matcher.c

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

@@ -626,6 +626,7 @@ _mongoc_cmd_parts_assemble_mongod (mongoc_cmd_parts_t *parts,
          break;
       case MONGOC_TOPOLOGY_SHARDED:
       case MONGOC_TOPOLOGY_UNKNOWN:
+      case MONGOC_TOPOLOGY_LOAD_BALANCED:
       case MONGOC_TOPOLOGY_DESCRIPTION_TYPES:
       default:
          /* must not call this function w/ sharded or unknown topology type */

src/libmongoc/src/mongoc/mongoc-read-prefs.c

@@ -378,6 +378,7 @@ assemble_query (const mongoc_read_prefs_t *read_prefs,
       break;
 
    case MONGOC_TOPOLOGY_SHARDED:
+   case MONGOC_TOPOLOGY_LOAD_BALANCED:
       _apply_read_preferences_mongos (read_prefs, query_bson, result);
       break;
 

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_LOAD_BALANCER,
    MONGOC_SERVER_DESCRIPTION_TYPES,
 } mongoc_server_description_type_t;
 

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_LOAD_BALANCER:
+      return "LoadBalancer";
    case MONGOC_SERVER_DESCRIPTION_TYPES:
    default:
       MONGOC_ERROR ("Invalid mongoc_server_description_t type");

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_LOAD_BALANCED,
    MONGOC_TOPOLOGY_DESCRIPTION_TYPES
 } mongoc_topology_description_type_t;
 

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

@@ -36,6 +36,7 @@ _is_data_node (mongoc_server_description_t *sd)
    case MONGOC_SERVER_STANDALONE:
    case MONGOC_SERVER_RS_SECONDARY:
    case MONGOC_SERVER_RS_PRIMARY:
+   case MONGOC_SERVER_LOAD_BALANCER:
       return true;
    case MONGOC_SERVER_RS_OTHER:
    case MONGOC_SERVER_RS_ARBITER:
@@ -2139,6 +2140,8 @@ mongoc_topology_description_type (const mongoc_topology_description_t *td)
       return "ReplicaSetWithPrimary";
    case MONGOC_TOPOLOGY_SINGLE:
       return "Single";
+   case MONGOC_TOPOLOGY_LOAD_BALANCED:
+      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

@@ -375,11 +375,6 @@ mongoc_topology_new (const mongoc_uri_t *uri, bool single_threaded)
          GOTO (srv_fail);
       }
 
-      if (!mongoc_uri_finalize_loadbalanced (topology->uri,
-                                             &topology->scanner->error)) {
-         GOTO (srv_fail);
-      }
-
       topology->srv_polling_last_scan_ms = bson_get_monotonic_time () / 1000;
       /* TODO (CDRIVER-4047) use BSON_MIN */
       topology->srv_polling_rescan_interval_ms = BSON_MAX (
@@ -394,6 +389,11 @@ mongoc_topology_new (const mongoc_uri_t *uri, bool single_threaded)
       topology_valid = true;
    }
 
+   if (!mongoc_uri_finalize_loadbalanced (topology->uri,
+                                          &topology->scanner->error)) {
+      topology_valid = false;
+   }
+
    /*
     * Set topology type from URI:
     *   + if directConnection=true
@@ -413,7 +413,12 @@ 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 loadBalanced is enabled, directConnection is disabled. This was
+    * validated in mongoc_uri_finalize_loadbalanced. */
+   if (mongoc_uri_get_option_as_bool (
+          topology->uri, MONGOC_URI_LOADBALANCED, false)) {
+      init_type = MONGOC_TOPOLOGY_LOAD_BALANCED;
+   } else if (service && !has_directconnection) {
       init_type = MONGOC_TOPOLOGY_UNKNOWN;
    } else if (has_directconnection) {
       if (directconnection) {
@@ -653,11 +658,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)) {
@@ -1436,15 +1436,18 @@ _mongoc_topology_pop_server_session (mongoc_topology_t *topology,
    int64_t timeout;
    mongoc_server_session_t *ss = NULL;
    mongoc_topology_description_t *td;
+   bool loadbalanced;
 
    ENTRY;
 
    bson_mutex_lock (&topology->mutex);
 
    td = &topology->description;
    timeout = td->session_timeout_minutes;
+   loadbalanced = td->type == MONGOC_TOPOLOGY_LOAD_BALANCED;
 
-   if (timeout == MONGOC_NO_SESSIONS) {
+   /* When the topology type is LoadBalanced, sessions are always supported. */
+   if (!loadbalanced && timeout == MONGOC_NO_SESSIONS) {
       /* if needed, connect and check for session timeout again */
       if (!mongoc_topology_description_has_data_node (td)) {
          bson_mutex_unlock (&topology->mutex);
@@ -1470,6 +1473,11 @@ _mongoc_topology_pop_server_session (mongoc_topology_t *topology,
    while (topology->session_pool) {
       ss = topology->session_pool;
       CDL_DELETE (topology->session_pool, ss);
+      /* Sessions do not expire when the topology type is load balanced. */
+      if (loadbalanced) {
+         break;
+      }
+
       if (_mongoc_server_session_timed_out (ss, timeout)) {
          _mongoc_server_session_destroy (ss);
          ss = NULL;
@@ -1503,17 +1511,20 @@ _mongoc_topology_push_server_session (mongoc_topology_t *topology,
 {
    int64_t timeout;
    mongoc_server_session_t *ss;
+   bool loadbalanced;
 
    ENTRY;
 
    bson_mutex_lock (&topology->mutex);
 
    timeout = topology->description.session_timeout_minutes;
+   loadbalanced = topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED;
 
    /* start at back of queue and reap timed-out sessions */
    while (topology->session_pool && topology->session_pool->prev) {
       ss = topology->session_pool->prev;
-      if (_mongoc_server_session_timed_out (ss, timeout)) {
+      /* Sessions do not expire when the topology type is load balanced. */
+      if (!loadbalanced && _mongoc_server_session_timed_out (ss, timeout)) {
          BSON_ASSERT (ss->next); /* silences clang scan-build */
          CDL_DELETE (topology->session_pool, ss);
          _mongoc_server_session_destroy (ss);
@@ -1524,8 +1535,10 @@ _mongoc_topology_push_server_session (mongoc_topology_t *topology,
    }
 
    /* If session is expiring or "dirty" (a network error occurred on it), do not
-    * return it to the pool. */
-   if (_mongoc_server_session_timed_out (server_session, timeout) ||
+    * return it to the pool. Sessions do not expire when the topology type is
+    * load balanced. */
+   if ((!loadbalanced &&
+        _mongoc_server_session_timed_out (server_session, timeout)) ||
        server_session->dirty) {
       _mongoc_server_session_destroy (server_session);
    } else {

src/libmongoc/tests/test-libmongoc.c

@@ -272,6 +272,8 @@ extern void
 test_bson_util_install (TestSuite *suite);
 extern void
 test_result_install (TestSuite *suite);
+extern void
+test_loadbalanced_install (TestSuite *suite);
 
 typedef struct {
    mongoc_log_level_t level;
@@ -1851,7 +1853,8 @@ mongoc_client_pool_t *
 test_framework_new_default_client_pool ()
 {
    mongoc_uri_t *test_uri = test_framework_get_uri ();
-   mongoc_client_pool_t *pool = test_framework_client_pool_new_from_uri (test_uri, NULL);
+   mongoc_client_pool_t *pool =
+      test_framework_client_pool_new_from_uri (test_uri, NULL);
 
    BSON_ASSERT (pool);
    test_framework_set_pool_ssl_opts (pool);
@@ -2905,6 +2908,7 @@ main (int argc, char *argv[])
    test_bson_match_install (&suite);
    test_bson_util_install (&suite);
    test_result_install (&suite);
+   test_loadbalanced_install (&suite);
 
    ret = TestSuite_Run (&suite);
 

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

@@ -0,0 +1,168 @@
+/*
+ * Copyright 2021-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "mongoc/mongoc.h"
+#include "mongoc/mongoc-client-session-private.h"
+#include "test-conveniences.h"
+#include "test-libmongoc.h"
+#include "TestSuite.h"
+
+static char *
+loadbalanced_uri (void)
+{
+   /* TODO (CDRIVER-4062): This will need to add TLS and auth to the URI when
+    * run in evergreen. */
+   return test_framework_getenv ("SINGLE_MONGOS_LB_URI");
+}
+
+static void
+test_loadbalanced_sessions_supported (void *unused)
+{
+   mongoc_client_t *client;
+   mongoc_client_session_t *session;
+   char *uristr = loadbalanced_uri ();
+   bson_error_t error;
+
+   client = mongoc_client_new (uristr);
+   session = mongoc_client_start_session (client, NULL /* opts */, &error);
+   ASSERT_OR_PRINT (session, error);
+
+   mongoc_client_session_destroy (session);
+   bson_free (uristr);
+   mongoc_client_destroy (client);
+}
+
+static void
+test_loadbalanced_sessions_do_not_expire (void *unused)
+{
+   mongoc_client_t *client;
+   mongoc_client_session_t *session1;
+   mongoc_client_session_t *session2;
+   char *uristr = loadbalanced_uri ();
+   bson_error_t error;
+   bson_t *session1_lsid;
+   bson_t *session2_lsid;
+
+   client = mongoc_client_new (uristr);
+   /* Mock a timeout so session expiration applies. */
+   client->topology->description.session_timeout_minutes = 1;
+
+   /* Start two sessions, to ensure that pooled sessions remain in the pool when
+    * the pool is accessed. */
+   session1 = mongoc_client_start_session (client, NULL /* opts */, &error);
+   ASSERT_OR_PRINT (session1, error);
+   session2 = mongoc_client_start_session (client, NULL /* opts */, &error);
+   ASSERT_OR_PRINT (session2, error);
+
+   session1_lsid = bson_copy (mongoc_client_session_get_lsid (session1));
+   session2_lsid = bson_copy (mongoc_client_session_get_lsid (session2));
+
+   /* Expire both sessions. */
+   session1->server_session->last_used_usec = 1;
+   session2->server_session->last_used_usec = 1;
+   mongoc_client_session_destroy (session1);
+   mongoc_client_session_destroy (session2);
+
+   /* Get a new session, it should reuse the most recently pushed session2. */
+   session2 = mongoc_client_start_session (client, NULL /* opts */, &error);
+   ASSERT_OR_PRINT (session2, error);
+   if (!bson_equal (mongoc_client_session_get_lsid (session2), session2_lsid)) {
+      test_error ("Session not reused: %s != %s",
+                  tmp_json (mongoc_client_session_get_lsid (session2)),
+                  tmp_json (session2_lsid));
+   }
+
+   session1 = mongoc_client_start_session (client, NULL /* opts */, &error);
+   ASSERT_OR_PRINT (session1, error);
+   if (!bson_equal (mongoc_client_session_get_lsid (session1), session1_lsid)) {
+      test_error ("Session not reused: %s != %s",
+                  tmp_json (mongoc_client_session_get_lsid (session1)),
+                  tmp_json (session1_lsid));
+   }
+
+   bson_destroy (session1_lsid);
+   bson_destroy (session2_lsid);
+   bson_free (uristr);
+   mongoc_client_session_destroy (session1);
+   mongoc_client_session_destroy (session2);
+   mongoc_client_destroy (client);
+}
+
+/* Test that invalid loadBalanced URI configurations are validated during client
+ * construction. */
+static void
+test_loadbalanced_client_uri_validation (void *unused)
+{
+   mongoc_client_t *client;
+   mongoc_uri_t *uri;
+   bson_error_t error;
+   bool ret;
+
+   uri = mongoc_uri_new ("mongodb://localhost:27017");
+   mongoc_uri_set_option_as_bool (uri, MONGOC_URI_LOADBALANCED, true);
+   mongoc_uri_set_option_as_bool (uri, MONGOC_URI_DIRECTCONNECTION, true);
+   client = mongoc_client_new_from_uri (uri);
+
+   ret = mongoc_client_command_simple (client,
+                                       "admin",
+                                       tmp_bson ("{'ping': 1}"),
+                                       NULL /* read prefs */,
+                                       NULL /* reply */,
+                                       &error);
+   ASSERT_ERROR_CONTAINS (error,
+                          MONGOC_ERROR_SERVER_SELECTION,
+                          MONGOC_ERROR_SERVER_SELECTION_FAILURE,
+                          "URI with \"loadBalanced\" enabled must not contain "
+                          "option \"directConnection\" enabled");
+   BSON_ASSERT (!ret);
+
+   mongoc_uri_destroy (uri);
+   mongoc_client_destroy (client);
+}
+
+static int
+skip_if_not_loadbalanced (void)
+{
+   char *val = loadbalanced_uri ();
+   if (!val) {
+      return 0;
+   }
+   bson_free (val);
+   return 1;
+}
+
+void
+test_loadbalanced_install (TestSuite *suite)
+{
+   TestSuite_AddFull (suite,
+                      "/loadbalanced/sessions/supported",
+                      test_loadbalanced_sessions_supported,
+                      NULL /* ctx */,
+                      NULL /* dtor */,
+                      skip_if_not_loadbalanced);
+   TestSuite_AddFull (suite,
+                      "/loadbalanced/sessions/do_not_expire",
+                      test_loadbalanced_sessions_do_not_expire,
+                      NULL /* ctx */,
+                      NULL /* dtor */,
+                      skip_if_not_loadbalanced);
+   TestSuite_AddFull (suite,
+                      "/loadbalanced/client_uri_validation",
+                      test_loadbalanced_client_uri_validation,
+                      NULL /* ctx */,
+                      NULL /* dtor */,
+                      NULL);
+}