Merge branch 'master' of https://github.com/grpc/grpc-java into trace-new-line

Author: YifeiZhuang

alts/src/main/java/io/grpc/alts/internal/AltsProtocolNegotiator.java

@@ -54,10 +54,12 @@
 // TODO(carl-mastrangelo): rename this AltsProtocolNegotiators.
 public final class AltsProtocolNegotiator {
   private static final Logger logger = Logger.getLogger(AltsProtocolNegotiator.class.getName());
-  // Avoid performing too many handshakes in parallel, as it may cause queuing in the handshake
-  // server and cause unbounded blocking on the event loop (b/168808426). This is a workaround until
-  // there is an async TSI handshaking API to avoid the blocking.
-  private static final AsyncSemaphore handshakeSemaphore = new AsyncSemaphore(32);
+
+  static final String ALTS_MAX_CONCURRENT_HANDSHAKES_ENV_VARIABLE =
+      "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES";
+  @VisibleForTesting static final int DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES = 32;
+  private static final AsyncSemaphore handshakeSemaphore =
+      new AsyncSemaphore(getAltsMaxConcurrentHandshakes());
 
   @Grpc.TransportAttr
   public static final Attributes.Key<TsiPeer> TSI_PEER_KEY =
@@ -424,5 +426,30 @@ public SecurityDetails validatePeerObject(Object peerObject) throws GeneralSecur
     }
   }
 
+  @VisibleForTesting
+  static int getAltsMaxConcurrentHandshakes(String altsMaxConcurrentHandshakes) {
+    if (altsMaxConcurrentHandshakes == null) {
+      return DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES;
+    }
+    try {
+      int effectiveMaxConcurrentHandshakes = Integer.parseInt(altsMaxConcurrentHandshakes);
+      if (effectiveMaxConcurrentHandshakes < 0) {
+        logger.warning(
+            "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES environment variable set to invalid value.");
+        return DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES;
+      }
+      return effectiveMaxConcurrentHandshakes;
+    } catch (NumberFormatException e) {
+      logger.warning(
+          "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES environment variable set to invalid value.");
+      return DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES;
+    }
+  }
+
+  private static int getAltsMaxConcurrentHandshakes() {
+    return getAltsMaxConcurrentHandshakes(
+        System.getenv(ALTS_MAX_CONCURRENT_HANDSHAKES_ENV_VARIABLE));
+  }
+
   private AltsProtocolNegotiator() {}
 }

alts/src/test/java/io/grpc/alts/internal/AltsProtocolNegotiatorTest.java

@@ -354,6 +354,29 @@ public void peerPropagated() throws Exception {
         .isEqualTo(SecurityLevel.PRIVACY_AND_INTEGRITY);
   }
 
+  @Test
+  public void getAltsMaxConcurrentHandshakes_success() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes("10")).isEqualTo(10);
+  }
+
+  @Test
+  public void getAltsMaxConcurrentHandshakes_envVariableNotSet() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes(null))
+        .isEqualTo(AltsProtocolNegotiator.DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES);
+  }
+
+  @Test
+  public void getAltsMaxConcurrentHandshakes_envVariableNotANumber() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes("not-a-number"))
+        .isEqualTo(AltsProtocolNegotiator.DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES);
+  }
+
+  @Test
+  public void getAltsMaxConcurrentHandshakes_envVariableNegative() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes("-10"))
+        .isEqualTo(AltsProtocolNegotiator.DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES);
+  }
+
   private void doHandshake() throws Exception {
     // Capture the client frame and add to the server.
     assertEquals(1, channel.outboundMessages().size());

gcp-observability/src/main/java/io/grpc/gcp/observability/GcpObservability.java

@@ -21,7 +21,6 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableSet;
 import io.grpc.ClientInterceptor;
-import io.grpc.ExperimentalApi;
 import io.grpc.InternalGlobalInterceptors;
 import io.grpc.ManagedChannelProvider.ProviderNotFoundException;
 import io.grpc.ServerInterceptor;
@@ -64,7 +63,6 @@
 import java.util.stream.Collectors;
 
 /** The main class for gRPC Google Cloud Platform Observability features. */
-@ExperimentalApi("https://github.com/grpc/grpc-java/issues/8869")
 public final class GcpObservability implements AutoCloseable {
 
   private static final Logger logger = Logger.getLogger(GcpObservability.class.getName());
@@ -90,12 +88,10 @@ public final class GcpObservability implements AutoCloseable {
    */
   public static synchronized GcpObservability grpcInit() throws IOException {
     if (instance == null) {
-      GlobalLocationTags globalLocationTags = new GlobalLocationTags();
       ObservabilityConfigImpl observabilityConfig = ObservabilityConfigImpl.getInstance();
       TraceLoggingHelper traceLoggingHelper = new TraceLoggingHelper(
           observabilityConfig.getProjectId());
-      Sink sink = new GcpLogSink(observabilityConfig.getProjectId(),
-          globalLocationTags.getLocationTags(), observabilityConfig,
+      Sink sink = new GcpLogSink(observabilityConfig.getProjectId(), observabilityConfig,
           SERVICES_TO_EXCLUDE, traceLoggingHelper);
       LogHelper helper = new LogHelper(sink);
       ConfigFilterHelper configFilterHelper = ConfigFilterHelper.getInstance(observabilityConfig);