Allow configuring max concurrent ALTS handshakes based on an environment variable. (#32672)

The logic is straightforward: attempt to read the
`GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES` environment variable and, if it
set to an integer, instantiate the handshake queues based on this
integer.

Based on go/grpc-alts-concurrent-handshake-cap.

BUILD

@@ -3208,7 +3208,10 @@ grpc_cc_library(
         "//src/core:tsi/alts/handshaker/alts_tsi_handshaker_private.h",
         "//src/core:tsi/alts/handshaker/alts_tsi_utils.h",
     ],
-    external_deps = ["upb_lib"],
+    external_deps = [
+        "absl/strings",
+        "upb_lib",
+    ],
     language = "c++",
     visibility = ["@grpc:public"],
     deps = [
@@ -3221,6 +3224,7 @@ grpc_cc_library(
         "tsi_base",
         "//src/core:channel_args",
         "//src/core:closure",
+        "//src/core:env",
         "//src/core:pollset_set",
         "//src/core:slice",
     ],

src/core/tsi/alts/handshaker/alts_handshaker_client.cc

@@ -22,6 +22,7 @@
 
 #include <list>
 
+#include "absl/strings/numbers.h"
 #include "upb/upb.hpp"
 
 #include <grpc/byte_buffer.h>
@@ -29,6 +30,7 @@
 #include <grpc/support/log.h>
 
 #include "src/core/lib/gprpp/crash.h"
+#include "src/core/lib/gprpp/env.h"
 #include "src/core/lib/gprpp/sync.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/surface/call.h"
@@ -40,6 +42,8 @@
 #define TSI_ALTS_INITIAL_BUFFER_SIZE 256
 
 const int kHandshakerClientOpNum = 4;
+const char kMaxConcurrentStreamsEnvironmentVariable[] =
+    "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES";
 
 struct alts_handshaker_client {
   const alts_handshaker_client_vtable* vtable;
@@ -419,7 +423,8 @@ HandshakeQueue* g_client_handshake_queue;
 HandshakeQueue* g_server_handshake_queue;
 
 void DoHandshakeQueuesInit(void) {
-  const size_t per_queue_max_outstanding_handshakes = 40;
+  const size_t per_queue_max_outstanding_handshakes =
+      MaxNumberOfConcurrentHandshakes();
   g_client_handshake_queue =
       new HandshakeQueue(per_queue_max_outstanding_handshakes);
   g_server_handshake_queue =
@@ -924,3 +929,17 @@ void alts_handshaker_client_destroy(alts_handshaker_client* c) {
     alts_grpc_handshaker_client_unref(client);
   }
 }
+
+size_t MaxNumberOfConcurrentHandshakes() {
+  size_t max_concurrent_handshakes = 40;
+  absl::optional<std::string> env_var_max_concurrent_handshakes =
+      grpc_core::GetEnv(kMaxConcurrentStreamsEnvironmentVariable);
+  if (env_var_max_concurrent_handshakes.has_value()) {
+    size_t effective_max_concurrent_handshakes = 40;
+    if (absl::SimpleAtoi(*env_var_max_concurrent_handshakes,
+                         &effective_max_concurrent_handshakes)) {
+      max_concurrent_handshakes = effective_max_concurrent_handshakes;
+    }
+  }
+  return max_concurrent_handshakes;
+}

src/core/tsi/alts/handshaker/alts_handshaker_client.h

@@ -157,4 +157,9 @@ alts_handshaker_client* alts_grpc_handshaker_client_create(
 void alts_handshaker_client_handle_response(alts_handshaker_client* client,
                                             bool is_ok);
 
+// Returns the max number of concurrent handshakes that are permitted.
+//
+// Exposed for testing purposes only.
+size_t MaxNumberOfConcurrentHandshakes();
+
 #endif  // GRPC_SRC_CORE_TSI_ALTS_HANDSHAKER_ALTS_HANDSHAKER_CLIENT_H

test/core/tsi/alts/handshaker/BUILD

@@ -35,6 +35,7 @@ grpc_cc_test(
     deps = [
         ":alts_handshaker_service_api_test_lib",
         "//:grpc",
+        "//src/core:env",
         "//test/core/util:grpc_test_util",
     ],
 )

test/core/tsi/alts/handshaker/alts_handshaker_client_test.cc

@@ -25,6 +25,7 @@
 #include <grpc/grpc.h>
 #include <grpc/grpc_security.h>
 
+#include "src/core/lib/gprpp/env.h"
 #include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/tsi/alts/handshaker/alts_shared_resource.h"
 #include "src/core/tsi/alts/handshaker/alts_tsi_handshaker.h"
@@ -40,6 +41,8 @@
 #define ALTS_HANDSHAKER_CLIENT_TEST_TARGET_SERVICE_ACCOUNT2 "B@google.com"
 #define ALTS_HANDSHAKER_CLIENT_TEST_MAX_FRAME_SIZE (64 * 1024)
 
+const char kMaxConcurrentStreamsEnvironmentVariable[] =
+    "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES";
 const size_t kHandshakerClientOpNum = 4;
 const size_t kMaxRpcVersionMajor = 3;
 const size_t kMaxRpcVersionMinor = 2;
@@ -508,6 +511,26 @@ TEST(AltsHandshakerClientTest, ScheduleRequestGrpcCallFailureTest) {
   destroy_config(config);
 }
 
+TEST(MaxNumberOfConcurrentHandshakesTest, Default) {
+  grpc_core::UnsetEnv(kMaxConcurrentStreamsEnvironmentVariable);
+  EXPECT_EQ(MaxNumberOfConcurrentHandshakes(), 40);
+}
+
+TEST(MaxNumberOfConcurrentHandshakesTest, EnvVarNotInt) {
+  grpc_core::SetEnv(kMaxConcurrentStreamsEnvironmentVariable, "not-a-number");
+  EXPECT_EQ(MaxNumberOfConcurrentHandshakes(), 40);
+}
+
+TEST(MaxNumberOfConcurrentHandshakesTest, EnvVarNegative) {
+  grpc_core::SetEnv(kMaxConcurrentStreamsEnvironmentVariable, "-10");
+  EXPECT_EQ(MaxNumberOfConcurrentHandshakes(), 40);
+}
+
+TEST(MaxNumberOfConcurrentHandshakesTest, EnvVarSuccess) {
+  grpc_core::SetEnv(kMaxConcurrentStreamsEnvironmentVariable, "10");
+  EXPECT_EQ(MaxNumberOfConcurrentHandshakes(), 10);
+}
+
 int main(int argc, char** argv) {
   grpc::testing::TestEnvironment env(&argc, argv);
   ::testing::InitGoogleTest(&argc, argv);