interop: add --fail_on_failed_rpc xds client flag (#3567)

interop/xds/client/client.go

@@ -47,12 +47,13 @@ type statsWatcher struct {
 }
 
 var (
-	numChannels   = flag.Int("num_channels", 1, "Num of channels")
-	printResponse = flag.Bool("print_response", false, "Write RPC response to stdout")
-	qps           = flag.Int("qps", 1, "QPS per channel")
-	rpcTimeout    = flag.Duration("rpc_timeout", 10*time.Second, "Per RPC timeout")
-	server        = flag.String("server", "localhost:8080", "Address of server to connect to")
-	statsPort     = flag.Int("stats_port", 8081, "Port to expose peer distribution stats service")
+	failOnFailedRPC = flag.Bool("fail_on_failed_rpc", false, "Fail client if any RPCs fail")
+	numChannels     = flag.Int("num_channels", 1, "Num of channels")
+	printResponse   = flag.Bool("print_response", false, "Write RPC response to stdout")
+	qps             = flag.Int("qps", 1, "QPS per channel")
+	rpcTimeout      = flag.Duration("rpc_timeout", 10*time.Second, "Per RPC timeout")
+	server          = flag.String("server", "localhost:8080", "Address of server to connect to")
+	statsPort       = flag.Int("stats_port", 8081, "Port to expose peer distribution stats service")
 
 	mu               sync.Mutex
 	currentRequestID int32
@@ -123,7 +124,7 @@ func main() {
 
 	clients := make([]testpb.TestServiceClient, *numChannels)
 	for i := 0; i < *numChannels; i++ {
-		conn, err := grpc.DialContext(context.Background(), *server, grpc.WithInsecure(), grpc.WithBlock())
+		conn, err := grpc.DialContext(context.Background(), *server, grpc.WithInsecure())
 		if err != nil {
 			grpclog.Fatalf("Fail to dial: %v", err)
 		}
@@ -161,6 +162,9 @@ func sendRPCs(clients []testpb.TestServiceClient, ticker *time.Ticker) {
 				watcher.c <- r
 			}
 
+			if err != nil && *failOnFailedRPC {
+				grpclog.Fatalf("RPC failed: %v", err)
+			}
 			if success && *printResponse {
 				fmt.Printf("Greeting: Hello world, this is %s, from %v\n", r.GetHostname(), p.Addr)
 			}

test/kokoro/xds.sh

@@ -30,4 +30,5 @@ GRPC_GO_LOG_VERBOSITY_LEVEL=99 GRPC_GO_LOG_SEVERITY_LEVEL=info \
     --client_cmd="grpc-go/interop/xds/client/client \
       --server=xds-experimental:///{server_uri} \
       --stats_port={stats_port} \
-      --qps={qps}"
+      --qps={qps} \
+      {fail_on_failed_rpc}"