interop/client/client.go

/*
 *
 * Copyright 2014 gRPC authors.
 *
 * 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.
 *
 */

// Binary client is an interop client.
//
// See interop test case descriptions [here].
//
// [here]: https://github.com/grpc/grpc/blob/master/doc/interop-test-descriptions.md
package main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"flag"
	"log"
	"net"
	"os"
	"strconv"
	"strings"
	"time"

	"golang.org/x/oauth2"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/credentials/alts"
	"google.golang.org/grpc/credentials/google"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/credentials/oauth"
	"google.golang.org/grpc/grpclog"
	"google.golang.org/grpc/interop"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/testdata"

	_ "google.golang.org/grpc/balancer/grpclb"    // Register the grpclb load balancing policy.
	_ "google.golang.org/grpc/balancer/rls"       // Register the RLS load balancing policy.
	"google.golang.org/grpc/xds/googledirectpath" // Register xDS resolver required for c2p directpath.

	testgrpc "google.golang.org/grpc/interop/grpc_testing"
)

const (
	googleDefaultCredsName = "google_default_credentials"
	computeEngineCredsName = "compute_engine_channel_creds"
)

var (
	caFile                                 = flag.String("ca_file", "", "The file containing the CA root cert file")
	useTLS                                 = flag.Bool("use_tls", false, "Connection uses TLS if true")
	useALTS                                = flag.Bool("use_alts", false, "Connection uses ALTS if true (this option can only be used on GCP)")
	customCredentialsType                  = flag.String("custom_credentials_type", "", "Custom creds to use, excluding TLS or ALTS")
	altsHSAddr                             = flag.String("alts_handshaker_service_address", "", "ALTS handshaker gRPC service address")
	testCA                                 = flag.Bool("use_test_ca", false, "Whether to replace platform root CAs with test CA as the CA root")
	serviceAccountKeyFile                  = flag.String("service_account_key_file", "", "Path to service account json key file")
	oauthScope                             = flag.String("oauth_scope", "", "The scope for OAuth2 tokens")
	defaultServiceAccount                  = flag.String("default_service_account", "", "Email of GCE default service account")
	googleC2PUniverseDomain                = flag.String("google_c2p_universe_domain", "", "Universe domain for google-c2p resolve")
	serverHost                             = flag.String("server_host", "localhost", "The server host name")
	serverPort                             = flag.Int("server_port", 10000, "The server port number")
	serviceConfigJSON                      = flag.String("service_config_json", "", "Disables service config lookups and sets the provided string as the default service config.")
	soakIterations                         = flag.Int("soak_iterations", 10, "The number of iterations to use for the two soak tests: rpc_soak and channel_soak")
	soakMaxFailures                        = flag.Int("soak_max_failures", 0, "The number of iterations in soak tests that are allowed to fail (either due to non-OK status code or exceeding the per-iteration max acceptable latency).")
	soakPerIterationMaxAcceptableLatencyMs = flag.Int("soak_per_iteration_max_acceptable_latency_ms", 1000, "The number of milliseconds a single iteration in the two soak tests (rpc_soak and channel_soak) should take.")
	soakOverallTimeoutSeconds              = flag.Int("soak_overall_timeout_seconds", 10, "The overall number of seconds after which a soak test should stop and fail, if the desired number of iterations have not yet completed.")
	soakMinTimeMsBetweenRPCs               = flag.Int("soak_min_time_ms_between_rpcs", 0, "The minimum time in milliseconds between consecutive RPCs in a soak test (rpc_soak or channel_soak), useful for limiting QPS")
	soakRequestSize                        = flag.Int("soak_request_size", 271828, "The request size in a soak RPC. The default value is set based on the interop large unary test case.")
	soakResponseSize                       = flag.Int("soak_response_size", 314159, "The response size in a soak RPC. The default value is set based on the interop large unary test case.")
	soakNumThreads                         = flag.Int("soak_num_threads", 1, "The number of threads for concurrent execution of the soak tests (rpc_soak or channel_soak). The default value is set based on the interop large unary test case.")
	tlsServerName                          = flag.String("server_host_override", "", "The server name used to verify the hostname returned by TLS handshake if it is not empty. Otherwise, --server_host is used.")
	additionalMetadata                     = flag.String("additional_metadata", "", "Additional metadata to send in each request, as a semicolon-separated list of key:value pairs.")
	testCase                               = flag.String("test_case", "large_unary",
		`Configure different test cases. Valid options are:
        empty_unary : empty (zero bytes) request and response;
        large_unary : single request and (large) response;
        client_streaming : request streaming with single response;
        server_streaming : single request with response streaming;
        ping_pong : full-duplex streaming;
        empty_stream : full-duplex streaming with zero message;
        timeout_on_sleeping_server: fullduplex streaming on a sleeping server;
        compute_engine_creds: large_unary with compute engine auth;
        service_account_creds: large_unary with service account auth;
        jwt_token_creds: large_unary with jwt token auth;
        per_rpc_creds: large_unary with per rpc token;
        oauth2_auth_token: large_unary with oauth2 token auth;
        google_default_credentials: large_unary with google default credentials
        compute_engine_channel_credentials: large_unary with compute engine creds
        cancel_after_begin: cancellation after metadata has been sent but before payloads are sent;
        cancel_after_first_response: cancellation after receiving 1st message from the server;
        status_code_and_message: status code propagated back to client;
        special_status_message: Unicode and whitespace is correctly processed in status message;
        custom_metadata: server will echo custom metadata;
        unimplemented_method: client attempts to call unimplemented method;
        unimplemented_service: client attempts to call unimplemented service;
        pick_first_unary: all requests are sent to one server despite multiple servers are resolved;
        orca_per_rpc: the client verifies ORCA per-RPC metrics are provided;
        orca_oob: the client verifies ORCA out-of-band metrics are provided.`)

	logger = grpclog.Component("interop")
)

type credsMode uint8

const (
	credsNone credsMode = iota
	credsTLS
	credsALTS
	credsGoogleDefaultCreds
	credsComputeEngineCreds
)

// tlsBundle is a credentials.Bundle that combines TLS transport credentials
// with per-RPC credentials. It always uses the same TLS transport credentials,
// regardless of mode.
type tlsBundle struct {
	transportCreds credentials.TransportCredentials
	perRPCCreds    credentials.PerRPCCredentials
}

func (b *tlsBundle) TransportCredentials() credentials.TransportCredentials {
	return b.transportCreds
}

func (b *tlsBundle) PerRPCCredentials() credentials.PerRPCCredentials {
	return b.perRPCCreds
}

func (b *tlsBundle) NewWithMode(string) (credentials.Bundle, error) {
	return &tlsBundle{transportCreds: b.transportCreds.Clone(), perRPCCreds: b.perRPCCreds}, nil
}

// Parses the --additional_metadata flag and returns metadata to send on each RPC,
// formatted as per https://pkg.go.dev/google.golang.org/grpc/metadata#Pairs.
// Allow any character but semicolons in values. If the flag is empty, return a nil map.
func parseAdditionalMetadataFlag() []string {
	if len(*additionalMetadata) == 0 {
		return nil
	}
	r := *additionalMetadata
	addMd := make([]string, 0)
	for len(r) > 0 {
		i := strings.Index(r, ":")
		if i < 0 {
			logger.Fatalf("Error parsing --additional_metadata flag: missing colon separator")
		}
		addMd = append(addMd, r[:i]) // append key
		r = r[i+1:]
		i = strings.Index(r, ";")
		// append value
		if i < 0 {
			addMd = append(addMd, r)
			break
		}
		addMd = append(addMd, r[:i])
		r = r[i+1:]
	}
	return addMd
}

// createSoakTestConfig creates a shared configuration structure for soak tests.
func createBaseSoakConfig(serverAddr string) interop.SoakTestConfig {
	return interop.SoakTestConfig{
		RequestSize:                      *soakRequestSize,
		ResponseSize:                     *soakResponseSize,
		PerIterationMaxAcceptableLatency: time.Duration(*soakPerIterationMaxAcceptableLatencyMs) * time.Millisecond,
		MinTimeBetweenRPCs:               time.Duration(*soakMinTimeMsBetweenRPCs) * time.Millisecond,
		OverallTimeout:                   time.Duration(*soakOverallTimeoutSeconds) * time.Second,
		ServerAddr:                       serverAddr,
		NumWorkers:                       *soakNumThreads,
		Iterations:                       *soakIterations,
		MaxFailures:                      *soakMaxFailures,
	}
}

func main() {
	flag.Parse()
	logger.Infof("Client running with test case %q", *testCase)
	var useGDC bool // use google default creds
	var useCEC bool // use compute engine creds
	if *customCredentialsType != "" {
		switch *customCredentialsType {
		case googleDefaultCredsName:
			useGDC = true
		case computeEngineCredsName:
			useCEC = true
		default:
			logger.Fatalf("If set, custom_credentials_type can only be set to one of %v or %v",
				googleDefaultCredsName, computeEngineCredsName)
		}
	}
	if (*useTLS && *useALTS) || (*useTLS && useGDC) || (*useALTS && useGDC) || (*useTLS && useCEC) || (*useALTS && useCEC) {
		logger.Fatalf("only one of TLS, ALTS, google default creds, or compute engine creds can be used")
	}

	ctx := context.Background()

	var credsChosen credsMode
	switch {
	case *useTLS:
		credsChosen = credsTLS
	case *useALTS:
		credsChosen = credsALTS
	case useGDC:
		credsChosen = credsGoogleDefaultCreds
	case useCEC:
		credsChosen = credsComputeEngineCreds
	}

	resolver.SetDefaultScheme("dns")
	if len(*googleC2PUniverseDomain) > 0 {
		if err := googledirectpath.SetUniverseDomain(*googleC2PUniverseDomain); err != nil {
			log.Fatalf("googlec2p.SetUniverseDomain(%s) failed: %v", *googleC2PUniverseDomain, err)
		}
	}
	serverAddr := *serverHost
	if *serverPort != 0 {
		serverAddr = net.JoinHostPort(*serverHost, strconv.Itoa(*serverPort))
	}
	var opts []grpc.DialOption
	// tlsCreds is set when TLS is chosen. The dial option for it is added
	// below, once the per-RPC credentials for the test case are known.
	var tlsCreds credentials.TransportCredentials
	switch credsChosen {
	case credsTLS:
		var roots *x509.CertPool
		if *testCA {
			if *caFile == "" {
				*caFile = testdata.Path("ca.pem")
			}
			b, err := os.ReadFile(*caFile)
			if err != nil {
				logger.Fatalf("Failed to read root certificate file %q: %v", *caFile, err)
			}
			roots = x509.NewCertPool()
			if !roots.AppendCertsFromPEM(b) {
				logger.Fatalf("Failed to append certificates: %s", string(b))
			}
		}
		if *tlsServerName != "" {
			tlsCreds = credentials.NewClientTLSFromCert(roots, *tlsServerName)
		} else {
			tlsCreds = credentials.NewTLS(&tls.Config{RootCAs: roots})
		}
	case credsALTS:
		altsOpts := alts.DefaultClientOptions()
		if *altsHSAddr != "" {
			altsOpts.HandshakerServiceAddress = *altsHSAddr
		}
		altsTC := alts.NewClientCreds(altsOpts)
		opts = append(opts, grpc.WithTransportCredentials(altsTC))
	case credsGoogleDefaultCreds:
		opts = append(opts, grpc.WithCredentialsBundle(google.NewDefaultCredentials()))
	case credsComputeEngineCreds:
		opts = append(opts, grpc.WithCredentialsBundle(google.NewComputeEngineCredentials()))
	case credsNone:
		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
	default:
		logger.Fatal("Invalid creds")
	}
	if credsChosen == credsTLS {
		var perRPCCreds credentials.PerRPCCredentials
		if *testCase == "compute_engine_creds" {
			perRPCCreds = oauth.NewComputeEngine()
		} else if *testCase == "service_account_creds" {
			jwtCreds, err := oauth.NewServiceAccountFromFile(*serviceAccountKeyFile, *oauthScope)
			if err != nil {
				logger.Fatalf("Failed to create JWT credentials: %v", err)
			}
			perRPCCreds = jwtCreds
		} else if *testCase == "jwt_token_creds" {
			jwtCreds, err := oauth.NewJWTAccessFromFile(*serviceAccountKeyFile)
			if err != nil {
				logger.Fatalf("Failed to create JWT credentials: %v", err)
			}
			perRPCCreds = jwtCreds
		} else if *testCase == "oauth2_auth_token" {
			perRPCCreds = oauth.TokenSource{TokenSource: oauth2.StaticTokenSource(interop.GetToken(ctx, *serviceAccountKeyFile, *oauthScope))}
		}
		if perRPCCreds != nil {
			// Attach the per-RPC credentials to the channel credentials rather
			// than using grpc.WithPerRPCCredentials, so that channels created
			// on behalf of this channel (such as the RLS control channel)
			// inherit them as well.
			opts = append(opts, grpc.WithCredentialsBundle(&tlsBundle{transportCreds: tlsCreds, perRPCCreds: perRPCCreds}))
		} else {
			opts = append(opts, grpc.WithTransportCredentials(tlsCreds))
		}
	}
	if len(*serviceConfigJSON) > 0 {
		opts = append(opts, grpc.WithDisableServiceConfig(), grpc.WithDefaultServiceConfig(*serviceConfigJSON))
	}
	if addMd := parseAdditionalMetadataFlag(); addMd != nil {
		unaryAddMd := func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
			ctx = metadata.AppendToOutgoingContext(ctx, addMd...)
			return invoker(ctx, method, req, reply, cc, opts...)
		}
		streamingAddMd := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
			ctx = metadata.AppendToOutgoingContext(ctx, addMd...)
			return streamer(ctx, desc, cc, method, opts...)
		}
		opts = append(opts, grpc.WithUnaryInterceptor(unaryAddMd), grpc.WithStreamInterceptor(streamingAddMd))
	}
	conn, err := grpc.NewClient(serverAddr, opts...)
	if err != nil {
		logger.Fatalf("grpc.NewClient(%q) = %v", serverAddr, err)
	}
	defer conn.Close()
	tc := testgrpc.NewTestServiceClient(conn)
	ctxWithDeadline, cancel := context.WithTimeout(ctx, time.Duration(*soakOverallTimeoutSeconds)*time.Second)
	defer cancel()
	switch *testCase {
	case "empty_unary":
		interop.DoEmptyUnaryCall(ctx, tc)
		logger.Infoln("EmptyUnaryCall done")
	case "large_unary":
		interop.DoLargeUnaryCall(ctx, tc)
		logger.Infoln("LargeUnaryCall done")
	case "client_streaming":
		interop.DoClientStreaming(ctx, tc)
		logger.Infoln("ClientStreaming done")
	case "server_streaming":
		interop.DoServerStreaming(ctx, tc)
		logger.Infoln("ServerStreaming done")
	case "ping_pong":
		interop.DoPingPong(ctx, tc)
		logger.Infoln("Pingpong done")
	case "empty_stream":
		interop.DoEmptyStream(ctx, tc)
		logger.Infoln("Emptystream done")
	case "timeout_on_sleeping_server":
		interop.DoTimeoutOnSleepingServer(ctx, tc)
		logger.Infoln("TimeoutOnSleepingServer done")
	case "compute_engine_creds":
		if credsChosen != credsTLS {
			logger.Fatalf("TLS credentials need to be set for compute_engine_creds test case.")
		}
		interop.DoComputeEngineCreds(ctx, tc, *defaultServiceAccount, *oauthScope)
		logger.Infoln("ComputeEngineCreds done")
	case "service_account_creds":
		if credsChosen != credsTLS {
			logger.Fatalf("TLS credentials need to be set for service_account_creds test case.")
		}
		interop.DoServiceAccountCreds(ctx, tc, *serviceAccountKeyFile, *oauthScope)
		logger.Infoln("ServiceAccountCreds done")
	case "jwt_token_creds":
		if credsChosen != credsTLS {
			logger.Fatalf("TLS credentials need to be set for jwt_token_creds test case.")
		}
		interop.DoJWTTokenCreds(ctx, tc, *serviceAccountKeyFile)
		logger.Infoln("JWTtokenCreds done")
	case "per_rpc_creds":
		if credsChosen != credsTLS {
			logger.Fatalf("TLS credentials need to be set for per_rpc_creds test case.")
		}
		interop.DoPerRPCCreds(ctx, tc, *serviceAccountKeyFile, *oauthScope)
		logger.Infoln("PerRPCCreds done")
	case "oauth2_auth_token":
		if credsChosen != credsTLS {
			logger.Fatalf("TLS credentials need to be set for oauth2_auth_token test case.")
		}
		interop.DoOauth2TokenCreds(ctx, tc, *serviceAccountKeyFile, *oauthScope)
		logger.Infoln("Oauth2TokenCreds done")
	case "google_default_credentials":
		if credsChosen != credsGoogleDefaultCreds {
			logger.Fatalf("GoogleDefaultCredentials need to be set for google_default_credentials test case.")
		}
		interop.DoGoogleDefaultCredentials(ctx, tc, *defaultServiceAccount)
		logger.Infoln("GoogleDefaultCredentials done")
	case "compute_engine_channel_credentials":
		if credsChosen != credsComputeEngineCreds {
			logger.Fatalf("ComputeEngineCreds need to be set for compute_engine_channel_credentials test case.")
		}
		interop.DoComputeEngineChannelCredentials(ctx, tc, *defaultServiceAccount)
		logger.Infoln("ComputeEngineChannelCredentials done")
	case "cancel_after_begin":
		interop.DoCancelAfterBegin(ctx, tc)
		logger.Infoln("CancelAfterBegin done")
	case "cancel_after_first_response":
		interop.DoCancelAfterFirstResponse(ctx, tc)
		logger.Infoln("CancelAfterFirstResponse done")
	case "status_code_and_message":
		interop.DoStatusCodeAndMessage(ctx, tc)
		logger.Infoln("StatusCodeAndMessage done")
	case "special_status_message":
		interop.DoSpecialStatusMessage(ctx, tc)
		logger.Infoln("SpecialStatusMessage done")
	case "custom_metadata":
		interop.DoCustomMetadata(ctx, tc)
		logger.Infoln("CustomMetadata done")
	case "unimplemented_method":
		interop.DoUnimplementedMethod(ctx, conn)
		logger.Infoln("UnimplementedMethod done")
	case "unimplemented_service":
		interop.DoUnimplementedService(ctx, testgrpc.NewUnimplementedServiceClient(conn))
		logger.Infoln("UnimplementedService done")
	case "pick_first_unary":
		interop.DoPickFirstUnary(ctx, tc)
		logger.Infoln("PickFirstUnary done")
	case "rpc_soak":
		rpcSoakConfig := createBaseSoakConfig(serverAddr)
		rpcSoakConfig.ChannelForTest = func() (*grpc.ClientConn, func()) { return conn, func() {} }
		interop.DoSoakTest(ctxWithDeadline, rpcSoakConfig)
		logger.Infoln("RpcSoak done")
	case "channel_soak":
		channelSoakConfig := createBaseSoakConfig(serverAddr)
		channelSoakConfig.ChannelForTest = func() (*grpc.ClientConn, func()) {
			cc, err := grpc.NewClient(serverAddr, opts...)
			if err != nil {
				log.Fatalf("Failed to create shared channel: %v", err)
			}
			return cc, func() { cc.Close() }
		}
		interop.DoSoakTest(ctxWithDeadline, channelSoakConfig)
		logger.Infoln("ChannelSoak done")
	case "orca_per_rpc":
		interop.DoORCAPerRPCTest(ctx, tc)
		logger.Infoln("ORCAPerRPC done")
	case "orca_oob":
		interop.DoORCAOOBTest(ctx, tc)
		logger.Infoln("ORCAOOB done")
	default:
		logger.Fatal("Unsupported test case: ", *testCase)
	}
}