Merge branch 'master' into sreek-ep

.github/lock.yml

@@ -0,0 +1,2 @@
+daysUntilLock: 90
+lockComment: false

doc/core/grpc-client-server-polling-engine-usage.md

@@ -0,0 +1,32 @@
+# Polling Engine Usage on gRPC client and Server
+
+_Author: Sree Kuchibhotla (@sreecha) - Sep 2018_
+
+
+This document talks about how polling engine is used in gRPC core (both on client and server code paths).
+
+## gRPC client
+
+### Relation between Call, Channel (sub-channels), Completion queue, `grpc_pollset` 
+- A gRPC Call is tied to a channel (more specifically a sub-channel) and a completion queue for the lifetime of the call.
+- Once a _sub-channel_ is picked for the call, the file-descriptor (socket fd in case of TCP channels) is added to the pollset corresponding to call's completion queue. (Recall that as per [grpc-cq](grpc-cq.md), a completion queue has a pollset by default)
+
+![image](../images/grpc-call-channel-cq.png)
+
+
+### Making progress on Async `connect()` on sub-channels  (`grpc_pollset_set` usecase)
+- A gRPC channel is created between a client and a 'target'. The 'target' may resolve in to one or more backend servers.
+- A sub-channel is the 'connection' from a client to the backend server
+- While establishing sub-cannels (i.e connections) to the backends, gRPC issues async [`connect()`](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/iomgr/tcp_client_posix.cc#L296) calls which may not complete right away.  When the `connect()` eventually succeeds, the socket fd is make 'writable'
+  - This means that the polling engine must be monitoring all these sub-channel `fd`s for writable events and we need to make sure there is a polling thread that monitors all these fds
+  - To accomplish this, the `grpc_pollset_set` is used the following way (see picture below)
+
+![image](../images/grpc-client-lb-pss.png)
+
+## gRPC server
+
+- The listening fd (i.e., the socket fd corresponding to the server listening port) is added to each of the server completion queues. Note that in gRPC we use SO_REUSEPORT option and create multiple listening fds but all of them map to the same listening port
+- A new incoming channel is assigned to some server completion queue picked randomly (note that we currently [round-robin](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/iomgr/tcp_server_posix.cc#L231) over the server completion queues)
+
+![image](../images/grpc-server-cq-fds.png)
+

src/core/lib/iomgr/ev_epoll1_linux.cc

@@ -273,6 +273,10 @@ static gpr_mu fork_fd_list_mu;
 static void fd_global_init(void) { gpr_mu_init(&fd_freelist_mu); }
 
 static void fd_global_shutdown(void) {
+  // TODO(guantaol): We don't have a reasonable explanation about this
+  // lock()/unlock() pattern. It can be a valid barrier if there is at most one
+  // pending lock() at this point. Otherwise, there is still a possibility of
+  // use-after-free race. Need to reason about the code and/or clean it up.
   gpr_mu_lock(&fd_freelist_mu);
   gpr_mu_unlock(&fd_freelist_mu);
   while (fd_freelist != nullptr) {

src/core/lib/iomgr/ev_epollex_linux.cc

@@ -403,6 +403,10 @@ static void unref_by(grpc_fd* fd, int n) {
 static void fd_global_init(void) { gpr_mu_init(&fd_freelist_mu); }
 
 static void fd_global_shutdown(void) {
+  // TODO(guantaol): We don't have a reasonable explanation about this
+  // lock()/unlock() pattern. It can be a valid barrier if there is at most one
+  // pending lock() at this point. Otherwise, there is still a possibility of
+  // use-after-free race. Need to reason about the code and/or clean it up.
   gpr_mu_lock(&fd_freelist_mu);
   gpr_mu_unlock(&fd_freelist_mu);
   while (fd_freelist != nullptr) {

src/cpp/server/channelz/channelz_service.cc

@@ -92,4 +92,21 @@ Status ChannelzService::GetSubchannel(
   return Status::OK;
 }
 
+Status ChannelzService::GetSocket(ServerContext* unused,
+                                  const channelz::v1::GetSocketRequest* request,
+                                  channelz::v1::GetSocketResponse* response) {
+  char* json_str = grpc_channelz_get_socket(request->socket_id());
+  gpr_log(GPR_ERROR, "%s", json_str);
+  if (json_str == nullptr) {
+    return Status(NOT_FOUND, "No object found for that SocketId");
+  }
+  google::protobuf::util::Status s =
+      google::protobuf::util::JsonStringToMessage(json_str, response);
+  gpr_free(json_str);
+  if (s != google::protobuf::util::Status::OK) {
+    return Status(INTERNAL, s.ToString());
+  }
+  return Status::OK;
+}
+
 }  // namespace grpc

src/cpp/server/channelz/channelz_service.h

@@ -44,6 +44,10 @@ class ChannelzService final : public channelz::v1::Channelz::Service {
   Status GetSubchannel(ServerContext* unused,
                        const channelz::v1::GetSubchannelRequest* request,
                        channelz::v1::GetSubchannelResponse* response) override;
+  // implementation of GetSocket rpc
+  Status GetSocket(ServerContext* unused,
+                   const channelz::v1::GetSocketRequest* request,
+                   channelz::v1::GetSocketResponse* response) override;
 };
 
 }  // namespace grpc

src/csharp/Grpc.Core.Tests/MetadataTest.cs

@@ -72,6 +72,23 @@ namespace Grpc.Core.Tests
             Assert.Throws(typeof(ArgumentException), () => new Metadata.Entry("abc/", "xyz"));
         }
 
+        [Test]
+        public void KeysAreNormalized_UppercaseKey()
+        {
+            var uppercaseKey = "ABC";
+            var entry = new Metadata.Entry(uppercaseKey, "XYZ");
+            Assert.AreEqual("abc", entry.Key);
+        }
+
+        [Test]
+        public void KeysAreNormalized_LowercaseKey()
+        {
+            var lowercaseKey = "abc";
+            var entry = new Metadata.Entry(lowercaseKey, "XYZ");
+            // no allocation if key already lowercase
+            Assert.AreSame(lowercaseKey, entry.Key);
+        }
+
         [Test]
         public void Entry_ConstructionPreconditions()
         {

src/csharp/Grpc.Core/ClientBase.cs

@@ -151,12 +151,12 @@ namespace Grpc.Core
         {
             private class ClientBaseConfigurationInterceptor : Interceptor
             {
-                readonly Func<IMethod, string, CallOptions, Tuple<string, CallOptions>> interceptor;
+                readonly Func<IMethod, string, CallOptions, ClientBaseConfigurationInfo> interceptor;
 
                 /// <summary>
                 /// Creates a new instance of ClientBaseConfigurationInterceptor given the specified header and host interceptor function.
                 /// </summary>
-                public ClientBaseConfigurationInterceptor(Func<IMethod, string, CallOptions, Tuple<string, CallOptions>> interceptor)
+                public ClientBaseConfigurationInterceptor(Func<IMethod, string, CallOptions, ClientBaseConfigurationInfo> interceptor)
                 {
                     this.interceptor = GrpcPreconditions.CheckNotNull(interceptor, nameof(interceptor));
                 }
@@ -166,7 +166,7 @@ namespace Grpc.Core
                     where TResponse : class
                 {
                     var newHostAndCallOptions = interceptor(context.Method, context.Host, context.Options);
-                    return new ClientInterceptorContext<TRequest, TResponse>(context.Method, newHostAndCallOptions.Item1, newHostAndCallOptions.Item2);
+                    return new ClientInterceptorContext<TRequest, TResponse>(context.Method, newHostAndCallOptions.Host, newHostAndCallOptions.CallOptions);
                 }
 
                 public override TResponse BlockingUnaryCall<TRequest, TResponse>(TRequest request, ClientInterceptorContext<TRequest, TResponse> context, BlockingUnaryCallContinuation<TRequest, TResponse> continuation)
@@ -195,6 +195,18 @@ namespace Grpc.Core
                 }
             }
 
+            internal struct ClientBaseConfigurationInfo
+            {
+                internal readonly string Host;
+                internal readonly CallOptions CallOptions;
+
+                internal ClientBaseConfigurationInfo(string host, CallOptions callOptions)
+                {
+                    Host = host;
+                    CallOptions = callOptions;
+                }
+            }
+
             readonly CallInvoker undecoratedCallInvoker;
             readonly string host;
 
@@ -206,7 +218,7 @@ namespace Grpc.Core
 
             internal CallInvoker CreateDecoratedCallInvoker()
             {
-                return undecoratedCallInvoker.Intercept(new ClientBaseConfigurationInterceptor((method, host, options) => Tuple.Create(this.host, options)));
+                return undecoratedCallInvoker.Intercept(new ClientBaseConfigurationInterceptor((method, host, options) => new ClientBaseConfigurationInfo(this.host, options)));
             }
 
             internal ClientBaseConfiguration WithHost(string host)

src/csharp/Grpc.Core/Internal/MarshalUtils.cs

@@ -35,6 +35,13 @@ namespace Grpc.Core.Internal
         /// </summary>
         public static string PtrToStringUTF8(IntPtr ptr, int len)
         {
+            if (len == 0)
+            {
+                return "";
+            }
+
+            // TODO(jtattermusch): once Span dependency is added,
+            // use Span-based API to decode the string without copying the buffer.
             var bytes = new byte[len];
             Marshal.Copy(ptr, bytes, 0, len);
             return EncodingUTF8.GetString(bytes);

src/csharp/Grpc.Core/Metadata.cs

@@ -135,7 +135,7 @@ namespace Grpc.Core
         }
 
         /// <summary>
-        /// <see cref="T:IList`1"/>
+        /// Adds a new ASCII-valued metadata entry. See <c>Metadata.Entry</c> constructor for params.
         /// </summary>
         public void Add(string key, string value)
         {
@@ -143,7 +143,7 @@ namespace Grpc.Core
         }
 
         /// <summary>
-        /// <see cref="T:IList`1"/>
+        /// Adds a new binary-valued metadata entry. See <c>Metadata.Entry</c> constructor for params.
         /// </summary>
         public void Add(string key, byte[] valueBytes)
         {
@@ -225,8 +225,6 @@ namespace Grpc.Core
         /// </summary>
         public class Entry
         {
-            private static readonly Regex ValidKeyRegex = new Regex("^[.a-z0-9_-]+$");
-
             readonly string key;
             readonly string value;
             readonly byte[] valueBytes;
@@ -241,7 +239,7 @@ namespace Grpc.Core
             /// <summary>
             /// Initializes a new instance of the <see cref="Grpc.Core.Metadata.Entry"/> struct with a binary value.
             /// </summary>
-            /// <param name="key">Metadata key, needs to have suffix indicating a binary valued metadata entry.</param>
+            /// <param name="key">Metadata key. Gets converted to lowercase. Needs to have suffix indicating a binary valued metadata entry. Can only contain lowercase alphanumeric characters, underscores, hyphens and dots.</param>
             /// <param name="valueBytes">Value bytes.</param>
             public Entry(string key, byte[] valueBytes)
             {
@@ -255,9 +253,9 @@ namespace Grpc.Core
             }
 
             /// <summary>
-            /// Initializes a new instance of the <see cref="Grpc.Core.Metadata.Entry"/> struct holding an ASCII value.
+            /// Initializes a new instance of the <see cref="Grpc.Core.Metadata.Entry"/> struct with an ASCII value.
             /// </summary>
-            /// <param name="key">Metadata key, must not use suffix indicating a binary valued metadata entry.</param>
+            /// <param name="key">Metadata key. Gets converted to lowercase. Must not use suffix indicating a binary valued metadata entry. Can only contain lowercase alphanumeric characters, underscores, hyphens and dots.</param>
             /// <param name="value">Value string. Only ASCII characters are allowed.</param>
             public Entry(string key, string value)
             {
@@ -358,10 +356,42 @@ namespace Grpc.Core
 
             private static string NormalizeKey(string key)
             {
-                var normalized = GrpcPreconditions.CheckNotNull(key, "key").ToLowerInvariant();
-                GrpcPreconditions.CheckArgument(ValidKeyRegex.IsMatch(normalized), 
+                GrpcPreconditions.CheckNotNull(key, "key");
+
+                GrpcPreconditions.CheckArgument(IsValidKey(key, out bool isLowercase), 
                     "Metadata entry key not valid. Keys can only contain lowercase alphanumeric characters, underscores, hyphens and dots.");
-                return normalized;
+                if (isLowercase)
+                {
+                    // save allocation of a new string if already lowercase
+                    return key;
+                }
+                
+                return key.ToLowerInvariant();
+            }
+
+            private static bool IsValidKey(string input, out bool isLowercase)
+            {
+                isLowercase = true;
+                for (int i = 0; i < input.Length; i++)
+                {
+                    char c = input[i];
+                    if ('a' <= c && c <= 'z' ||
+                        '0' <= c && c <= '9' ||
+                        c == '.' ||
+                        c == '_' || 
+                        c == '-' )
+                        continue;
+
+                    if ('A' <= c && c <= 'Z')
+                    {
+                        isLowercase = false;
+                        continue;
+                    }
+
+                    return false;
+                }
+
+                return true;
             }
 
             /// <summary>

test/cpp/end2end/channelz_service_test.cc

@@ -43,6 +43,8 @@ using grpc::channelz::v1::GetChannelRequest;
 using grpc::channelz::v1::GetChannelResponse;
 using grpc::channelz::v1::GetServersRequest;
 using grpc::channelz::v1::GetServersResponse;
+using grpc::channelz::v1::GetSocketRequest;
+using grpc::channelz::v1::GetSocketResponse;
 using grpc::channelz::v1::GetSubchannelRequest;
 using grpc::channelz::v1::GetSubchannelResponse;
 using grpc::channelz::v1::GetTopChannelsRequest;
@@ -71,6 +73,26 @@ class Proxy : public ::grpc::testing::EchoTestService::Service {
     return stubs_[idx]->Echo(client_context.get(), *request, response);
   }
 
+  Status BidiStream(ServerContext* server_context,
+                    ServerReaderWriter<EchoResponse, EchoRequest>*
+                        stream_from_client) override {
+    EchoRequest request;
+    EchoResponse response;
+    std::unique_ptr<ClientContext> client_context =
+        ClientContext::FromServerContext(*server_context);
+
+    // always use the first proxy for streaming
+    auto stream_to_backend = stubs_[0]->BidiStream(client_context.get());
+    while (stream_from_client->Read(&request)) {
+      stream_to_backend->Write(request);
+      stream_to_backend->Read(&response);
+      stream_from_client->Write(response);
+    }
+
+    stream_to_backend->WritesDone();
+    return stream_to_backend->Finish();
+  }
+
  private:
   std::vector<std::unique_ptr<::grpc::testing::EchoTestService::Stub>> stubs_;
 };
@@ -149,6 +171,21 @@ class ChannelzServerTest : public ::testing::Test {
     EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message();
   }
 
+  void SendSuccessfulStream(int num_messages) {
+    EchoRequest request;
+    EchoResponse response;
+    request.set_message("Hello channelz");
+    ClientContext context;
+    auto stream_to_proxy = echo_stub_->BidiStream(&context);
+    for (int i = 0; i < num_messages; ++i) {
+      EXPECT_TRUE(stream_to_proxy->Write(request));
+      EXPECT_TRUE(stream_to_proxy->Read(&response));
+    }
+    stream_to_proxy->WritesDone();
+    Status s = stream_to_proxy->Finish();
+    EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message();
+  }
+
   void SendFailedEcho(int channel_idx) {
     EchoRequest request;
     EchoResponse response;
@@ -448,6 +485,121 @@ TEST_F(ChannelzServerTest, ServerCallTest) {
             kNumSuccess + kNumFailed + 1);
 }
 
+TEST_F(ChannelzServerTest, ManySubchannelsAndSockets) {
+  ResetStubs();
+  const int kNumChannels = 4;
+  ConfigureProxy(kNumChannels);
+  const int kNumSuccess = 10;
+  const int kNumFailed = 11;
+  for (int i = 0; i < kNumSuccess; ++i) {
+    SendSuccessfulEcho(0);
+    SendSuccessfulEcho(2);
+  }
+  for (int i = 0; i < kNumFailed; ++i) {
+    SendFailedEcho(1);
+    SendFailedEcho(2);
+  }
+  GetTopChannelsRequest gtc_request;
+  GetTopChannelsResponse gtc_response;
+  gtc_request.set_start_channel_id(0);
+  ClientContext context;
+  Status s =
+      channelz_stub_->GetTopChannels(&context, gtc_request, &gtc_response);
+  EXPECT_TRUE(s.ok()) << s.error_message();
+  EXPECT_EQ(gtc_response.channel_size(), kNumChannels);
+  for (int i = 0; i < gtc_response.channel_size(); ++i) {
+    // if the channel sent no RPCs, then expect no subchannels to have been
+    // created.
+    if (gtc_response.channel(i).data().calls_started() == 0) {
+      EXPECT_EQ(gtc_response.channel(i).subchannel_ref_size(), 0);
+      continue;
+    }
+    // The resolver must return at least one address.
+    ASSERT_GT(gtc_response.channel(i).subchannel_ref_size(), 0);
+    // First grab the subchannel
+    GetSubchannelRequest get_subchannel_req;
+    GetSubchannelResponse get_subchannel_resp;
+    get_subchannel_req.set_subchannel_id(
+        gtc_response.channel(i).subchannel_ref(0).subchannel_id());
+    ClientContext get_subchannel_ctx;
+    Status s = channelz_stub_->GetSubchannel(
+        &get_subchannel_ctx, get_subchannel_req, &get_subchannel_resp);
+    EXPECT_TRUE(s.ok()) << s.error_message();
+    EXPECT_EQ(get_subchannel_resp.subchannel().socket_ref_size(), 1);
+    // Now grab the socket.
+    GetSocketRequest get_socket_req;
+    GetSocketResponse get_socket_resp;
+    ClientContext get_socket_ctx;
+    get_socket_req.set_socket_id(
+        get_subchannel_resp.subchannel().socket_ref(0).socket_id());
+    s = channelz_stub_->GetSocket(&get_socket_ctx, get_socket_req,
+                                  &get_socket_resp);
+    EXPECT_TRUE(s.ok()) << s.error_message();
+    // calls started == streams started AND stream succeeded. Since none of
+    // these RPCs were canceled, all of the streams will succeeded even though
+    // the RPCs they represent might have failed.
+    EXPECT_EQ(get_subchannel_resp.subchannel().data().calls_started(),
+              get_socket_resp.socket().data().streams_started());
+    EXPECT_EQ(get_subchannel_resp.subchannel().data().calls_started(),
+              get_socket_resp.socket().data().streams_succeeded());
+    // All of the calls were unary, so calls started == messages sent.
+    EXPECT_EQ(get_subchannel_resp.subchannel().data().calls_started(),
+              get_socket_resp.socket().data().messages_sent());
+    // We only get responses when the RPC was successful, so
+    // calls succeeded == messages received.
+    EXPECT_EQ(get_subchannel_resp.subchannel().data().calls_succeeded(),
+              get_socket_resp.socket().data().messages_received());
+  }
+}
+
+TEST_F(ChannelzServerTest, StreamingRPC) {
+  ResetStubs();
+  ConfigureProxy(1);
+  const int kNumMessages = 5;
+  SendSuccessfulStream(kNumMessages);
+  // Get the channel
+  GetChannelRequest get_channel_request;
+  GetChannelResponse get_channel_response;
+  get_channel_request.set_channel_id(GetChannelId(0));
+  ClientContext get_channel_context;
+  Status s = channelz_stub_->GetChannel(
+      &get_channel_context, get_channel_request, &get_channel_response);
+  EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message();
+  EXPECT_EQ(get_channel_response.channel().data().calls_started(), 1);
+  EXPECT_EQ(get_channel_response.channel().data().calls_succeeded(), 1);
+  EXPECT_EQ(get_channel_response.channel().data().calls_failed(), 0);
+  // Get the subchannel
+  ASSERT_GT(get_channel_response.channel().subchannel_ref_size(), 0);
+  GetSubchannelRequest get_subchannel_request;
+  GetSubchannelResponse get_subchannel_response;
+  ClientContext get_subchannel_context;
+  get_subchannel_request.set_subchannel_id(
+      get_channel_response.channel().subchannel_ref(0).subchannel_id());
+  s = channelz_stub_->GetSubchannel(&get_subchannel_context,
+                                    get_subchannel_request,
+                                    &get_subchannel_response);
+  EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message();
+  EXPECT_EQ(get_subchannel_response.subchannel().data().calls_started(), 1);
+  EXPECT_EQ(get_subchannel_response.subchannel().data().calls_succeeded(), 1);
+  EXPECT_EQ(get_subchannel_response.subchannel().data().calls_failed(), 0);
+  // Get the socket
+  ASSERT_GT(get_subchannel_response.subchannel().socket_ref_size(), 0);
+  GetSocketRequest get_socket_request;
+  GetSocketResponse get_socket_response;
+  ClientContext get_socket_context;
+  get_socket_request.set_socket_id(
+      get_subchannel_response.subchannel().socket_ref(0).socket_id());
+  s = channelz_stub_->GetSocket(&get_socket_context, get_socket_request,
+                                &get_socket_response);
+  EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message();
+  EXPECT_EQ(get_socket_response.socket().data().streams_started(), 1);
+  EXPECT_EQ(get_socket_response.socket().data().streams_succeeded(), 1);
+  EXPECT_EQ(get_socket_response.socket().data().streams_failed(), 0);
+  EXPECT_EQ(get_socket_response.socket().data().messages_sent(), kNumMessages);
+  EXPECT_EQ(get_socket_response.socket().data().messages_received(),
+            kNumMessages);
+}
+
 }  // namespace testing
 }  // namespace grpc
 

tools/doxygen/Doxyfile.core

@@ -771,6 +771,7 @@ doc/compression_cookbook.md \
 doc/connection-backoff-interop-test-description.md \
 doc/connection-backoff.md \
 doc/connectivity-semantics-and-api.md \
+doc/core/grpc-client-server-polling-engine-usage.md \
 doc/core/grpc-cq.md \
 doc/core/grpc-error.md \
 doc/core/moving-to-c++.md \

tools/doxygen/Doxyfile.core.internal

@@ -771,6 +771,7 @@ doc/compression_cookbook.md \
 doc/connection-backoff-interop-test-description.md \
 doc/connection-backoff.md \
 doc/connectivity-semantics-and-api.md \
+doc/core/grpc-client-server-polling-engine-usage.md \
 doc/core/grpc-cq.md \
 doc/core/grpc-error.md \
 doc/core/moving-to-c++.md \

tools/run_tests/run_interop_tests.py

@@ -777,12 +777,14 @@ def cloud_to_prod_jobspec(language,
     ]
     if transport_security == 'tls':
         transport_security_options = ['--use_tls=true']
-    elif transport_security == 'google_default_credentials' and language == 'c++':
+    elif transport_security == 'google_default_credentials' and str(
+            language) in ['c++', 'go']:
         transport_security_options = [
             '--custom_credentials_type=google_default_credentials'
         ]
     else:
-        print('Invalid transport security option.')
+        print('Invalid transport security option %s in cloud_to_prod_jobspec.' %
+              transport_security)
         sys.exit(1)
     cmdargs = cmdargs + transport_security_options
     environ = dict(language.cloud_to_prod_env(), **language.global_env())
@@ -817,8 +819,9 @@ def cloud_to_prod_jobspec(language,
         cmdline=cmdline,
         cwd=cwd,
         environ=environ,
-        shortname='%s:%s:%s:%s' % (suite_name, language, server_host_nickname,
-                                   test_case),
+        shortname='%s:%s:%s:%s:%s' %
+        (suite_name, language, server_host_nickname, test_case,
+         transport_security),
         timeout_seconds=_TEST_TIMEOUT,
         flake_retries=4 if args.allow_flakes else 0,
         timeout_retries=2 if args.allow_flakes else 0,
@@ -848,7 +851,8 @@ def cloud_to_cloud_jobspec(language,
     elif transport_security == 'insecure':
         interop_only_options += ['--use_tls=false']
     else:
-        print('Invalid transport security option.')
+        print('Invalid transport security option %s in cloud_to_cloud_jobspec.'
+              % transport_security)
         sys.exit(1)
 
     client_test_case = test_case
@@ -903,8 +907,8 @@ def cloud_to_cloud_jobspec(language,
         cmdline=cmdline,
         cwd=cwd,
         environ=environ,
-        shortname='cloud_to_cloud:%s:%s_server:%s' % (language, server_name,
-                                                      test_case),
+        shortname='cloud_to_cloud:%s:%s_server:%s:%s' %
+        (language, server_name, test_case, transport_security),
         timeout_seconds=_TEST_TIMEOUT,
         flake_retries=4 if args.allow_flakes else 0,
         timeout_retries=2 if args.allow_flakes else 0,
@@ -929,7 +933,8 @@ def server_jobspec(language,
     elif transport_security == 'insecure':
         server_cmd += ['--use_tls=false']
     else:
-        print('Invalid transport security option.')
+        print('Invalid transport security option %s in server_jobspec.' %
+              transport_security)
         sys.exit(1)
     cmdline = bash_cmdline(language.server_cmd(server_cmd))
     environ = language.global_env()
@@ -1318,7 +1323,7 @@ try:
                                 service_account_key_file,
                                 transport_security='tls')
                             jobs.append(tls_test_job)
-                            if language == 'c++':
+                            if str(language) in ['c++', 'go']:
                                 google_default_creds_test_job = cloud_to_prod_jobspec(
                                     language,
                                     test_case,
@@ -1370,7 +1375,9 @@ try:
                                 service_account_key_file,
                                 transport_security='tls')
                             jobs.append(tls_test_job)
-                            if language == 'c++':
+                            if str(language) in [
+                                    'go'
+                            ]:  # Add more languages to the list to turn on tests.
                                 google_default_creds_test_job = cloud_to_prod_jobspec(
                                     language,
                                     test_case,
@@ -1378,6 +1385,7 @@ try:
                                     prod_servers[server_host_nickname],
                                     docker_image=docker_images.get(
                                         str(language)),
+                                    auth=True,
                                     manual_cmd_log=client_manual_cmd_log,
                                     service_account_key_file=args.
                                     service_account_key_file,