From a3cc4f613d6d53278c2b5decf5d0987d0d3462cd Mon Sep 17 00:00:00 2001
From: Eric Gribkoff <ericgribkoff@google.com>
Date: Fri, 24 Apr 2020 12:11:27 -0700
Subject: [PATCH] interop: add --fail_on_failed_rpc xds client flag (#3567)

---
 interop/xds/client/client.go | 18 +++++++++++-------
 test/kokoro/xds.sh           |  3 ++-
 2 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/interop/xds/client/client.go b/interop/xds/client/client.go
index 0164cb4b..77b4563b 100644
--- a/interop/xds/client/client.go
+++ b/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)
 			}
diff --git a/test/kokoro/xds.sh b/test/kokoro/xds.sh
index 788c7c9f..06d82638 100755
--- a/test/kokoro/xds.sh
+++ b/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}"