Merge branch 'grpc:master' into master

Documentation/server-reflection-tutorial.md

@@ -2,8 +2,9 @@
 
 gRPC Server Reflection provides information about publicly-accessible gRPC
 services on a server, and assists clients at runtime to construct RPC requests
-and responses without precompiled service information. It is used by gRPC CLI,
-which can be used to introspect server protos and send/receive test RPCs.
+and responses without precompiled service information. It is used by
+[gRPCurl](https://github.com/fullstorydev/grpcurl), which can be used to
+introspect server protos and send/receive test RPCs.
 
 ## Enable Server Reflection
 
@@ -39,36 +40,41 @@ make the following changes:
 An example server with reflection registered can be found at
 `examples/features/reflection/server`.
 
-## gRPC CLI
+## gRPCurl
 
-After enabling Server Reflection in a server application, you can use gRPC CLI
-to check its services. gRPC CLI is only available in c++. Instructions on how to
-build and use gRPC CLI can be found at
-[command_line_tool.md](https://github.com/grpc/grpc/blob/master/doc/command_line_tool.md).
+After enabling Server Reflection in a server application, you can use gRPCurl
+to check its services. gRPCurl is built with Go and has packages available.
+Instructions on how to install and use gRPCurl can be found at
+[gRPCurl Installation](https://github.com/fullstorydev/grpcurl#installation).
 
-## Use gRPC CLI to check services
+## Use gRPCurl to check services
 
 First, start the helloworld server in grpc-go directory:
 
 ```sh
-$ cd <grpc-go-directory>
-$ go run examples/features/reflection/server/main.go
+$ cd <grpc-go-directory>/examples
+$ go run features/reflection/server/main.go
 ```
 
-Open a new terminal and make sure you are in the directory where grpc_cli lives:
-
+output:
 ```sh
-$ cd <grpc-cpp-directory>/bins/opt
+server listening at [::]:50051
 ```
 
-### List services
+After installing gRPCurl, open a new terminal and run the commands from the new
+terminal.
+
+**NOTE:** gRPCurl expects a TLS-encrypted connection by default. For all of
+the commands below, use the `-plaintext` flag to use an unencrypted connection.
 
-`grpc_cli ls` command lists services and methods exposed at a given port:
+### List services and methods
+
+The `list` command lists services exposed at a given port:
 
 - List all the services exposed at a given port
 
   ```sh
-  $ ./grpc_cli ls localhost:50051
+  $ grpcurl -plaintext localhost:50051 list
   ```
 
   output:
@@ -78,72 +84,88 @@ $ cd <grpc-cpp-directory>/bins/opt
   helloworld.Greeter
   ```
 
-- List one service with details
+- List all the methods of a service
 
-  `grpc_cli ls` command inspects a service given its full name (in the format of
-  \<package\>.\<service\>). It can print information with a long listing format
-  when `-l` flag is set. This flag can be used to get more details about a
-  service.
+  The `list` command lists methods given the full service name (in the format of
+  \<package\>.\<service\>).
 
   ```sh
-  $ ./grpc_cli ls localhost:50051 helloworld.Greeter -l
+  $ grpcurl -plaintext localhost:50051 list helloworld.Greeter
   ```
 
   output:
   ```sh
-  filename: helloworld.proto
-  package: helloworld;
-  service Greeter {
-    rpc SayHello(helloworld.HelloRequest) returns (helloworld.HelloReply) {}
-  }
+  helloworld.Greeter.SayHello
+  ```
+
+### Describe services and methods
+
+- Describe all services
 
+  The `describe` command inspects a service given its full name (in the format
+  of \<package\>.\<service\>). 
+
+  ```sh
+  $ grpcurl -plaintext localhost:50051 describe helloworld.Greeter
   ```
 
-### List methods
+  output:
+  ```sh
+  helloworld.Greeter is a service:
+  service Greeter {
+    rpc SayHello ( .helloworld.HelloRequest ) returns ( .helloworld.HelloReply );
+  }
+  ```
 
-- List one method with details
+- Describe all methods of a service
 
-  `grpc_cli ls` command also inspects a method given its full name (in the
-  format of \<package\>.\<service\>.\<method\>).
+  The `describe` command inspects a method given its full name (in the format of
+  \<package\>.\<service\>.\<method\>).
 
   ```sh
-  $ ./grpc_cli ls localhost:50051 helloworld.Greeter.SayHello -l
+  $ grpcurl -plaintext localhost:50051 describe helloworld.Greeter.SayHello
   ```
 
   output:
   ```sh
-    rpc SayHello(helloworld.HelloRequest) returns (helloworld.HelloReply) {}
+  helloworld.Greeter.SayHello is a method:
+  rpc SayHello ( .helloworld.HelloRequest ) returns ( .helloworld.HelloReply );
   ```
 
 ### Inspect message types
 
-We can use`grpc_cli type` command to inspect request/response types given the
+We can use the `describe` command to inspect request/response types given the
 full name of the type (in the format of \<package\>.\<type\>).
 
 - Get information about the request type
 
   ```sh
-  $ ./grpc_cli type localhost:50051 helloworld.HelloRequest
+  $ grpcurl -plaintext localhost:50051 describe helloworld.HelloRequest
   ```
 
   output:
   ```sh
+  helloworld.HelloRequest is a message:
   message HelloRequest {
-    optional string name = 1[json_name = "name"];
+    string name = 1;
   }
   ```
 
 ### Call a remote method
 
-We can send RPCs to a server and get responses using `grpc_cli call` command.
+We can send RPCs to a server and get responses using the full method name (in
+the format of \<package\>.\<service\>.\<method\>). The `-d <string>` flag
+represents the request data and the `-format text` flag indicates that the
+request data is in text format.
 
 - Call a unary method
 
   ```sh
-  $ ./grpc_cli call localhost:50051 SayHello "name: 'gRPC CLI'"
+  $ grpcurl -plaintext -format text -d 'name: "gRPCurl"' \
+    localhost:50051 helloworld.Greeter.SayHello
   ```
 
   output:
   ```sh
-  message: "Hello gRPC CLI"
+  message: "Hello gRPCurl"
   ```

balancer/balancer.go

@@ -279,6 +279,14 @@ type PickResult struct {
 	// type, Done may not be called.  May be nil if the balancer does not wish
 	// to be notified when the RPC completes.
 	Done func(DoneInfo)
+
+	// Metadata provides a way for LB policies to inject arbitrary per-call
+	// metadata. Any metadata returned here will be merged with existing
+	// metadata added by the client application.
+	//
+	// LB policies with child policies are responsible for propagating metadata
+	// injected by their children to the ClientConn, as part of Pick().
+	Metatada metadata.MD
 }
 
 // TransientFailureError returns e.  It exists for backward compatibility and

balancer/rls/balancer.go

@@ -91,14 +91,16 @@ func (rlsBB) Name() string {
 
 func (rlsBB) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer {
 	lb := &rlsBalancer{
-		done:          grpcsync.NewEvent(),
-		cc:            cc,
-		bopts:         opts,
-		purgeTicker:   dataCachePurgeTicker(),
-		lbCfg:         &lbConfig{},
-		pendingMap:    make(map[cacheKey]*backoffState),
-		childPolicies: make(map[string]*childPolicyWrapper),
-		updateCh:      buffer.NewUnbounded(),
+		closed:             grpcsync.NewEvent(),
+		done:               grpcsync.NewEvent(),
+		cc:                 cc,
+		bopts:              opts,
+		purgeTicker:        dataCachePurgeTicker(),
+		dataCachePurgeHook: dataCachePurgeHook,
+		lbCfg:              &lbConfig{},
+		pendingMap:         make(map[cacheKey]*backoffState),
+		childPolicies:      make(map[string]*childPolicyWrapper),
+		updateCh:           buffer.NewUnbounded(),
 	}
 	lb.logger = internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf("[rls-experimental-lb %p] ", lb))
 	lb.dataCache = newDataCache(maxCacheSize, lb.logger)
@@ -110,11 +112,13 @@ func (rlsBB) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.
 
 // rlsBalancer implements the RLS LB policy.
 type rlsBalancer struct {
-	done        *grpcsync.Event
-	cc          balancer.ClientConn
-	bopts       balancer.BuildOptions
-	purgeTicker *time.Ticker
-	logger      *internalgrpclog.PrefixLogger
+	closed             *grpcsync.Event // Fires when Close() is invoked. Guarded by stateMu.
+	done               *grpcsync.Event // Fires when Close() is done.
+	cc                 balancer.ClientConn
+	bopts              balancer.BuildOptions
+	purgeTicker        *time.Ticker
+	dataCachePurgeHook func()
+	logger             *internalgrpclog.PrefixLogger
 
 	// If both cacheMu and stateMu need to be acquired, the former must be
 	// acquired first to prevent a deadlock. This order restriction is due to the
@@ -167,7 +171,18 @@ type controlChannelReady struct{}
 // on to a channel that this goroutine will select on, thereby the handling of
 // the update will happen asynchronously.
 func (b *rlsBalancer) run() {
-	go b.purgeDataCache()
+	// We exit out of the for loop below only after `Close()` has been invoked.
+	// Firing the done event here will ensure that Close() returns only after
+	// all goroutines are done.
+	defer func() { b.done.Fire() }()
+
+	// Wait for purgeDataCache() goroutine to exit before returning from here.
+	doneCh := make(chan struct{})
+	defer func() {
+		<-doneCh
+	}()
+	go b.purgeDataCache(doneCh)
+
 	for {
 		select {
 		case u := <-b.updateCh.Get():
@@ -194,7 +209,7 @@ func (b *rlsBalancer) run() {
 			default:
 				b.logger.Errorf("Unsupported update type %T", update)
 			}
-		case <-b.done.Done():
+		case <-b.closed.Done():
 			return
 		}
 	}
@@ -203,10 +218,12 @@ func (b *rlsBalancer) run() {
 // purgeDataCache is a long-running goroutine which periodically deletes expired
 // entries. An expired entry is one for which both the expiryTime and
 // backoffExpiryTime are in the past.
-func (b *rlsBalancer) purgeDataCache() {
+func (b *rlsBalancer) purgeDataCache(doneCh chan struct{}) {
+	defer close(doneCh)
+
 	for {
 		select {
-		case <-b.done.Done():
+		case <-b.closed.Done():
 			return
 		case <-b.purgeTicker.C:
 			b.cacheMu.Lock()
@@ -215,19 +232,21 @@ func (b *rlsBalancer) purgeDataCache() {
 			if updatePicker {
 				b.sendNewPicker()
 			}
-			dataCachePurgeHook()
+			b.dataCachePurgeHook()
 		}
 	}
 }
 
 func (b *rlsBalancer) UpdateClientConnState(ccs balancer.ClientConnState) error {
 	defer clientConnUpdateHook()
-	if b.done.HasFired() {
+
+	b.stateMu.Lock()
+	if b.closed.HasFired() {
+		b.stateMu.Unlock()
 		b.logger.Warningf("Received service config after balancer close: %s", pretty.ToJSON(ccs.BalancerConfig))
 		return errBalancerClosed
 	}
 
-	b.stateMu.Lock()
 	newCfg := ccs.BalancerConfig.(*lbConfig)
 	if b.lbCfg.Equal(newCfg) {
 		b.stateMu.Unlock()
@@ -405,10 +424,9 @@ func (b *rlsBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.Sub
 }
 
 func (b *rlsBalancer) Close() {
-	b.done.Fire()
-
-	b.purgeTicker.Stop()
 	b.stateMu.Lock()
+	b.closed.Fire()
+	b.purgeTicker.Stop()
 	if b.ctrlCh != nil {
 		b.ctrlCh.close()
 	}
@@ -418,6 +436,8 @@ func (b *rlsBalancer) Close() {
 	b.cacheMu.Lock()
 	b.dataCache.stop()
 	b.cacheMu.Unlock()
+
+	<-b.done.Done()
 }
 
 func (b *rlsBalancer) ExitIdle() {
@@ -479,8 +499,11 @@ func (b *rlsBalancer) sendNewPickerLocked() {
 
 func (b *rlsBalancer) sendNewPicker() {
 	b.stateMu.Lock()
+	defer b.stateMu.Unlock()
+	if b.closed.HasFired() {
+		return
+	}
 	b.sendNewPickerLocked()
-	b.stateMu.Unlock()
 }
 
 // The aggregated connectivity state reported is determined as follows:

balancer/rls/balancer_test.go

@@ -1048,6 +1048,43 @@ func (s) TestUpdateStatePauses(t *testing.T) {
 	// Make sure an RLS request is sent out.
 	verifyRLSRequest(t, rlsReqCh, true)
 
+	// Wait for the control channel to become READY, before reading the states
+	// out of the wrapping top-level balancer.
+	//
+	// makeTestRPCAndExpectItToReachBackend repeatedly sends RPCs with short
+	// deadlines until one succeeds. See its docstring for details.
+	//
+	// The following sequence of events is possible:
+	// 1. When the first RPC is attempted above, a pending cache entry is
+	//    created, an RLS request is sent out, and the pick is queued. The
+	//    channel is in CONNECTING state.
+	// 2. When the RLS response arrives, the pending cache entry is moved to the
+	//    data cache, a child policy is created for the target specified in the
+	//    response and a new picker is returned. The channel is still in
+	//    CONNECTING, and retried pick is again queued.
+	// 3. The child policy moves through the standard set of states, IDLE -->
+	//    CONNECTING --> READY. And for each of these state changes, a new
+	//    picker is sent on the channel. But the overall connectivity state of
+	//    the channel is still CONNECTING.
+	// 4. Right around the time when the child policy becomes READY, the
+	//    deadline associated with the first RPC made by
+	//    makeTestRPCAndExpectItToReachBackend() could expire, and it could send
+	//    a new one. And because the internal state of the LB policy now
+	//    contains a child policy which is READY, this RPC will succeed. But the
+	//    RLS LB policy has yet to push a new picker on the channel.
+	// 5. If we read the states seen by the top-level wrapping LB policy without
+	//    waiting for the channel to become READY, there is a possibility that we
+	//    might not see the READY state in there. And if that happens, we will
+	//    see two extra states in the last check made in the test, and thereby
+	//    the test would fail. Waiting for the channel to become READY here
+	//    ensures that the test does not flake because of this rare sequence of
+	//    events.
+	for s := cc.GetState(); s != connectivity.Ready; s = cc.GetState() {
+		if !cc.WaitForStateChange(ctx, s) {
+			t.Fatal("Timeout when waiting for connectivity state to reach READY")
+		}
+	}
+
 	// Cache the state changes seen up to this point.
 	states0 := wb.getStates()