VOL-2138 Use v2 import paths for voltha-lib-go;
migrate from voltha-go to voltha-lib-go
Change-Id: I3db6759f3c0cea3c2164889b3d36eae708b19bde
diff --git a/vendor/google.golang.org/grpc/.travis.yml b/vendor/google.golang.org/grpc/.travis.yml
index 8f36b81..f0f723f 100644
--- a/vendor/google.golang.org/grpc/.travis.yml
+++ b/vendor/google.golang.org/grpc/.travis.yml
@@ -2,16 +2,16 @@
matrix:
include:
- - go: 1.12beta2
- env: GO111MODULE=on
- - go: 1.11.x
+ - go: 1.12.x
env: VET=1 GO111MODULE=on
- - go: 1.11.x
+ - go: 1.12.x
env: RACE=1 GO111MODULE=on
- - go: 1.11.x
+ - go: 1.12.x
env: RUN386=1
- - go: 1.11.x
+ - go: 1.12.x
env: GRPC_GO_RETRY=on
+ - go: 1.11.x
+ env: GO111MODULE=on
- go: 1.10.x
- go: 1.9.x
- go: 1.9.x
@@ -23,7 +23,7 @@
- if [[ "${GO111MODULE}" = "on" ]]; then mkdir "${HOME}/go"; export GOPATH="${HOME}/go"; fi
- if [[ -n "${RUN386}" ]]; then export GOARCH=386; fi
- if [[ "${TRAVIS_EVENT_TYPE}" = "cron" && -z "${RUN386}" ]]; then RACE=1; fi
- - if [[ "${TRAVIS_EVENT_TYPE}" != "cron" ]]; then VET_SKIP_PROTO=1; fi
+ - if [[ "${TRAVIS_EVENT_TYPE}" != "cron" ]]; then export VET_SKIP_PROTO=1; fi
install:
- try3() { eval "$*" || eval "$*" || eval "$*"; }
diff --git a/vendor/google.golang.org/grpc/CODE-OF-CONDUCT.md b/vendor/google.golang.org/grpc/CODE-OF-CONDUCT.md
new file mode 100644
index 0000000..9d4213e
--- /dev/null
+++ b/vendor/google.golang.org/grpc/CODE-OF-CONDUCT.md
@@ -0,0 +1,3 @@
+## Community Code of Conduct
+
+gRPC follows the [CNCF Code of Conduct](https://github.com/cncf/foundation/blob/master/code-of-conduct.md).
diff --git a/vendor/google.golang.org/grpc/CONTRIBUTING.md b/vendor/google.golang.org/grpc/CONTRIBUTING.md
index ca34e8a..4f1567e 100644
--- a/vendor/google.golang.org/grpc/CONTRIBUTING.md
+++ b/vendor/google.golang.org/grpc/CONTRIBUTING.md
@@ -1,6 +1,8 @@
# How to contribute
-We definitely welcome your patches and contributions to gRPC!
+We definitely welcome your patches and contributions to gRPC! Please read the gRPC
+organization's [governance rules](https://github.com/grpc/grpc-community/blob/master/governance.md)
+and [contribution guidelines](https://github.com/grpc/grpc-community/blob/master/CONTRIBUTING.md) before proceeding.
If you are new to github, please start by reading [Pull Request howto](https://help.github.com/articles/about-pull-requests/)
@@ -11,22 +13,46 @@
## Guidelines for Pull Requests
How to get your contributions merged smoothly and quickly.
-
-- Create **small PRs** that are narrowly focused on **addressing a single concern**. We often times receive PRs that are trying to fix several things at a time, but only one fix is considered acceptable, nothing gets merged and both author's & review's time is wasted. Create more PRs to address different concerns and everyone will be happy.
-
-- For speculative changes, consider opening an issue and discussing it first. If you are suggesting a behavioral or API change, consider starting with a [gRFC proposal](https://github.com/grpc/proposal).
-
-- Provide a good **PR description** as a record of **what** change is being made and **why** it was made. Link to a github issue if it exists.
-
-- Don't fix code style and formatting unless you are already changing that line to address an issue. PRs with irrelevant changes won't be merged. If you do want to fix formatting or style, do that in a separate PR.
-
-- Unless your PR is trivial, you should expect there will be reviewer comments that you'll need to address before merging. We expect you to be reasonably responsive to those comments, otherwise the PR will be closed after 2-3 weeks of inactivity.
-
-- Maintain **clean commit history** and use **meaningful commit messages**. PRs with messy commit history are difficult to review and won't be merged. Use `rebase -i upstream/master` to curate your commit history and/or to bring in latest changes from master (but avoid rebasing in the middle of a code review).
-
-- Keep your PR up to date with upstream/master (if there are merge conflicts, we can't really merge your change).
-
-- **All tests need to be passing** before your change can be merged. We recommend you **run tests locally** before creating your PR to catch breakages early on.
+
+- Create **small PRs** that are narrowly focused on **addressing a single
+ concern**. We often times receive PRs that are trying to fix several things at
+ a time, but only one fix is considered acceptable, nothing gets merged and
+ both author's & review's time is wasted. Create more PRs to address different
+ concerns and everyone will be happy.
+
+- The grpc package should only depend on standard Go packages and a small number
+ of exceptions. If your contribution introduces new dependencies which are NOT
+ in the [list](https://godoc.org/google.golang.org/grpc?imports), you need a
+ discussion with gRPC-Go authors and consultants.
+
+- For speculative changes, consider opening an issue and discussing it first. If
+ you are suggesting a behavioral or API change, consider starting with a [gRFC
+ proposal](https://github.com/grpc/proposal).
+
+- Provide a good **PR description** as a record of **what** change is being made
+ and **why** it was made. Link to a github issue if it exists.
+
+- Don't fix code style and formatting unless you are already changing that line
+ to address an issue. PRs with irrelevant changes won't be merged. If you do
+ want to fix formatting or style, do that in a separate PR.
+
+- Unless your PR is trivial, you should expect there will be reviewer comments
+ that you'll need to address before merging. We expect you to be reasonably
+ responsive to those comments, otherwise the PR will be closed after 2-3 weeks
+ of inactivity.
+
+- Maintain **clean commit history** and use **meaningful commit messages**. PRs
+ with messy commit history are difficult to review and won't be merged. Use
+ `rebase -i upstream/master` to curate your commit history and/or to bring in
+ latest changes from master (but avoid rebasing in the middle of a code
+ review).
+
+- Keep your PR up to date with upstream/master (if there are merge conflicts, we
+ can't really merge your change).
+
+- **All tests need to be passing** before your change can be merged. We
+ recommend you **run tests locally** before creating your PR to catch breakages
+ early on.
- `make all` to test everything, OR
- `make vet` to catch vet errors
- `make test` to run the tests
@@ -34,4 +60,3 @@
- optional `make testappengine` to run tests with appengine
- Exceptions to the rules can be made if there's a compelling reason for doing so.
-
diff --git a/vendor/google.golang.org/grpc/GOVERNANCE.md b/vendor/google.golang.org/grpc/GOVERNANCE.md
new file mode 100644
index 0000000..d6ff267
--- /dev/null
+++ b/vendor/google.golang.org/grpc/GOVERNANCE.md
@@ -0,0 +1 @@
+This repository is governed by the gRPC organization's [governance rules](https://github.com/grpc/grpc-community/blob/master/governance.md).
diff --git a/vendor/google.golang.org/grpc/MAINTAINERS.md b/vendor/google.golang.org/grpc/MAINTAINERS.md
new file mode 100644
index 0000000..093c82b
--- /dev/null
+++ b/vendor/google.golang.org/grpc/MAINTAINERS.md
@@ -0,0 +1,27 @@
+This page lists all active maintainers of this repository. If you were a
+maintainer and would like to add your name to the Emeritus list, please send us a
+PR.
+
+See [GOVERNANCE.md](https://github.com/grpc/grpc-community/blob/master/governance.md)
+for governance guidelines and how to become a maintainer.
+See [CONTRIBUTING.md](https://github.com/grpc/grpc-community/blob/master/CONTRIBUTING.md)
+for general contribution guidelines.
+
+## Maintainers (in alphabetical order)
+- [canguler](https://github.com/canguler), Google LLC
+- [cesarghali](https://github.com/cesarghali), Google LLC
+- [dfawley](https://github.com/dfawley), Google LLC
+- [easwars](https://github.com/easwars), Google LLC
+- [jadekler](https://github.com/jadekler), Google LLC
+- [menghanl](https://github.com/menghanl), Google LLC
+- [srini100](https://github.com/srini100), Google LLC
+
+## Emeritus Maintainers (in alphabetical order)
+- [adelez](https://github.com/adelez), Google LLC
+- [iamqizhao](https://github.com/iamqizhao), Google LLC
+- [jtattermusch](https://github.com/jtattermusch), Google LLC
+- [lyuxuan](https://github.com/lyuxuan), Google LLC
+- [makmukhi](https://github.com/makmukhi), Google LLC
+- [matt-kwong](https://github.com/matt-kwong), Google LLC
+- [nicolasnoble](https://github.com/nicolasnoble), Google LLC
+- [yongni](https://github.com/yongni), Google LLC
diff --git a/vendor/google.golang.org/grpc/README.md b/vendor/google.golang.org/grpc/README.md
index f5eec67..afbc43d 100644
--- a/vendor/google.golang.org/grpc/README.md
+++ b/vendor/google.golang.org/grpc/README.md
@@ -1,42 +1,96 @@
# gRPC-Go
-[](https://travis-ci.org/grpc/grpc-go) [](https://godoc.org/google.golang.org/grpc) [](https://goreportcard.com/report/github.com/grpc/grpc-go)
+[](https://travis-ci.org/grpc/grpc-go)
+[](https://godoc.org/google.golang.org/grpc)
+[](https://goreportcard.com/report/github.com/grpc/grpc-go)
-The Go implementation of [gRPC](https://grpc.io/): A high performance, open source, general RPC framework that puts mobile and HTTP/2 first. For more information see the [gRPC Quick Start: Go](https://grpc.io/docs/quickstart/go.html) guide.
+The Go implementation of [gRPC](https://grpc.io/): A high performance, open
+source, general RPC framework that puts mobile and HTTP/2 first. For more
+information see the [gRPC Quick Start:
+Go](https://grpc.io/docs/quickstart/go.html) guide.
Installation
------------
-To install this package, you need to install Go and setup your Go workspace on your computer. The simplest way to install the library is to run:
+To install this package, you need to install Go and setup your Go workspace on
+your computer. The simplest way to install the library is to run:
```
$ go get -u google.golang.org/grpc
```
+With Go module support (Go 1.11+), simply `import "google.golang.org/grpc"` in
+your source code and `go [build|run|test]` will automatically download the
+necessary dependencies ([Go modules
+ref](https://github.com/golang/go/wiki/Modules)).
+
+If you are trying to access grpc-go from within China, please see the
+[FAQ](#FAQ) below.
+
Prerequisites
-------------
-
gRPC-Go requires Go 1.9 or later.
-Constraints
------------
-The grpc package should only depend on standard Go packages and a small number of exceptions. If your contribution introduces new dependencies which are NOT in the [list](https://godoc.org/google.golang.org/grpc?imports), you need a discussion with gRPC-Go authors and consultants.
-
Documentation
-------------
-See [API documentation](https://godoc.org/google.golang.org/grpc) for package and API descriptions and find examples in the [examples directory](examples/).
+- See [godoc](https://godoc.org/google.golang.org/grpc) for package and API
+ descriptions.
+- Documentation on specific topics can be found in the [Documentation
+ directory](Documentation/).
+- Examples can be found in the [examples directory](examples/).
Performance
-----------
-See the current benchmarks for some of the languages supported in [this dashboard](https://performance-dot-grpc-testing.appspot.com/explore?dashboard=5652536396611584&widget=490377658&container=1286539696).
+Performance benchmark data for grpc-go and other languages is maintained in
+[this
+dashboard](https://performance-dot-grpc-testing.appspot.com/explore?dashboard=5652536396611584&widget=490377658&container=1286539696).
Status
------
-General Availability [Google Cloud Platform Launch Stages](https://cloud.google.com/terms/launch-stages).
+General Availability [Google Cloud Platform Launch
+Stages](https://cloud.google.com/terms/launch-stages).
FAQ
---
+#### I/O Timeout Errors
+
+The `golang.org` domain may be blocked from some countries. `go get` usually
+produces an error like the following when this happens:
+
+```
+$ go get -u google.golang.org/grpc
+package google.golang.org/grpc: unrecognized import path "google.golang.org/grpc" (https fetch: Get https://google.golang.org/grpc?go-get=1: dial tcp 216.239.37.1:443: i/o timeout)
+```
+
+To build Go code, there are several options:
+
+- Set up a VPN and access google.golang.org through that.
+
+- Without Go module support: `git clone` the repo manually:
+
+ ```
+ git clone https://github.com/grpc/grpc-go.git $GOPATH/src/google.golang.org/grpc
+ ```
+
+ You will need to do the same for all of grpc's dependencies in `golang.org`,
+ e.g. `golang.org/x/net`.
+
+- With Go module support: it is possible to use the `replace` feature of `go
+ mod` to create aliases for golang.org packages. In your project's directory:
+
+ ```
+ go mod edit -replace=google.golang.org/grpc=github.com/grpc/grpc-go@latest
+ go mod tidy
+ go mod vendor
+ go build -mod=vendor
+ ```
+
+ Again, this will need to be done for all transitive dependencies hosted on
+ golang.org as well. Please refer to [this
+ issue](https://github.com/golang/go/issues/28652) in the golang repo regarding
+ this concern.
+
#### Compiling error, undefined: grpc.SupportPackageIsVersion
Please update proto package, gRPC package and rebuild the proto files:
diff --git a/vendor/google.golang.org/grpc/balancer.go b/vendor/google.golang.org/grpc/balancer.go
index a78e702..a8eb0f4 100644
--- a/vendor/google.golang.org/grpc/balancer.go
+++ b/vendor/google.golang.org/grpc/balancer.go
@@ -43,7 +43,7 @@
// BalancerConfig specifies the configurations for Balancer.
//
-// Deprecated: please use package balancer.
+// Deprecated: please use package balancer. May be removed in a future 1.x release.
type BalancerConfig struct {
// DialCreds is the transport credential the Balancer implementation can
// use to dial to a remote load balancer server. The Balancer implementations
@@ -57,7 +57,7 @@
// BalancerGetOptions configures a Get call.
//
-// Deprecated: please use package balancer.
+// Deprecated: please use package balancer. May be removed in a future 1.x release.
type BalancerGetOptions struct {
// BlockingWait specifies whether Get should block when there is no
// connected address.
@@ -66,7 +66,7 @@
// Balancer chooses network addresses for RPCs.
//
-// Deprecated: please use package balancer.
+// Deprecated: please use package balancer. May be removed in a future 1.x release.
type Balancer interface {
// Start does the initialization work to bootstrap a Balancer. For example,
// this function may start the name resolution and watch the updates. It will
@@ -120,7 +120,7 @@
// RoundRobin returns a Balancer that selects addresses round-robin. It uses r to watch
// the name resolution updates and updates the addresses available correspondingly.
//
-// Deprecated: please use package balancer/roundrobin.
+// Deprecated: please use package balancer/roundrobin. May be removed in a future 1.x release.
func RoundRobin(r naming.Resolver) Balancer {
return &roundRobin{r: r}
}
diff --git a/vendor/google.golang.org/grpc/balancer/balancer.go b/vendor/google.golang.org/grpc/balancer/balancer.go
index 67518de..c266f4e 100644
--- a/vendor/google.golang.org/grpc/balancer/balancer.go
+++ b/vendor/google.golang.org/grpc/balancer/balancer.go
@@ -22,6 +22,7 @@
import (
"context"
+ "encoding/json"
"errors"
"net"
"strings"
@@ -31,6 +32,7 @@
"google.golang.org/grpc/internal"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/resolver"
+ "google.golang.org/grpc/serviceconfig"
)
var (
@@ -39,7 +41,10 @@
)
// Register registers the balancer builder to the balancer map. b.Name
-// (lowercased) will be used as the name registered with this builder.
+// (lowercased) will be used as the name registered with this builder. If the
+// Builder implements ConfigParser, ParseConfig will be called when new service
+// configs are received by the resolver, and the result will be provided to the
+// Balancer in UpdateClientConnState.
//
// NOTE: this function must only be called during initialization time (i.e. in
// an init() function), and is not thread-safe. If multiple Balancers are
@@ -138,6 +143,8 @@
ResolveNow(resolver.ResolveNowOption)
// Target returns the dial target for this ClientConn.
+ //
+ // Deprecated: Use the Target field in the BuildOptions instead.
Target() string
}
@@ -155,6 +162,10 @@
Dialer func(context.Context, string) (net.Conn, error)
// ChannelzParentID is the entity parent's channelz unique identification number.
ChannelzParentID int64
+ // Target contains the parsed address info of the dial target. It is the same resolver.Target as
+ // passed to the resolver.
+ // See the documentation for the resolver.Target type for details about what it contains.
+ Target resolver.Target
}
// Builder creates a balancer.
@@ -166,14 +177,19 @@
Name() string
}
+// ConfigParser parses load balancer configs.
+type ConfigParser interface {
+ // ParseConfig parses the JSON load balancer config provided into an
+ // internal form or returns an error if the config is invalid. For future
+ // compatibility reasons, unknown fields in the config should be ignored.
+ ParseConfig(LoadBalancingConfigJSON json.RawMessage) (serviceconfig.LoadBalancingConfig, error)
+}
+
// PickOptions contains addition information for the Pick operation.
type PickOptions struct {
// FullMethodName is the method name that NewClientStream() is called
// with. The canonical format is /service/Method.
FullMethodName string
- // Header contains the metadata from the RPC's client header. The metadata
- // should not be modified; make a copy first if needed.
- Header metadata.MD
}
// DoneInfo contains additional information for done.
@@ -186,6 +202,11 @@
BytesSent bool
// BytesReceived indicates if any byte has been received from the server.
BytesReceived bool
+ // ServerLoad is the load received from server. It's usually sent as part of
+ // trailing metadata.
+ //
+ // The only supported type now is *orca_v1.LoadReport.
+ ServerLoad interface{}
}
var (
@@ -215,8 +236,10 @@
//
// If a SubConn is returned:
// - If it is READY, gRPC will send the RPC on it;
- // - If it is not ready, or becomes not ready after it's returned, gRPC will block
- // until UpdateBalancerState() is called and will call pick on the new picker.
+ // - If it is not ready, or becomes not ready after it's returned, gRPC will
+ // block until UpdateBalancerState() is called and will call pick on the
+ // new picker. The done function returned from Pick(), if not nil, will be
+ // called with nil error, no bytes sent and no bytes received.
//
// If the returned error is not nil:
// - If the error is ErrNoSubConnAvailable, gRPC will block until UpdateBalancerState()
@@ -249,18 +272,55 @@
// that back to gRPC.
// Balancer should also generate and update Pickers when its internal state has
// been changed by the new state.
+ //
+ // Deprecated: if V2Balancer is implemented by the Balancer,
+ // UpdateSubConnState will be called instead.
HandleSubConnStateChange(sc SubConn, state connectivity.State)
// HandleResolvedAddrs is called by gRPC to send updated resolved addresses to
// balancers.
// Balancer can create new SubConn or remove SubConn with the addresses.
// An empty address slice and a non-nil error will be passed if the resolver returns
// non-nil error to gRPC.
+ //
+ // Deprecated: if V2Balancer is implemented by the Balancer,
+ // UpdateClientConnState will be called instead.
HandleResolvedAddrs([]resolver.Address, error)
// Close closes the balancer. The balancer is not required to call
// ClientConn.RemoveSubConn for its existing SubConns.
Close()
}
+// SubConnState describes the state of a SubConn.
+type SubConnState struct {
+ ConnectivityState connectivity.State
+ // TODO: add last connection error
+}
+
+// ClientConnState describes the state of a ClientConn relevant to the
+// balancer.
+type ClientConnState struct {
+ ResolverState resolver.State
+ // The parsed load balancing configuration returned by the builder's
+ // ParseConfig method, if implemented.
+ BalancerConfig serviceconfig.LoadBalancingConfig
+}
+
+// V2Balancer is defined for documentation purposes. If a Balancer also
+// implements V2Balancer, its UpdateClientConnState method will be called
+// instead of HandleResolvedAddrs and its UpdateSubConnState will be called
+// instead of HandleSubConnStateChange.
+type V2Balancer interface {
+ // UpdateClientConnState is called by gRPC when the state of the ClientConn
+ // changes.
+ UpdateClientConnState(ClientConnState)
+ // UpdateSubConnState is called by gRPC when the state of a SubConn
+ // changes.
+ UpdateSubConnState(SubConn, SubConnState)
+ // Close closes the balancer. The balancer is not required to call
+ // ClientConn.RemoveSubConn for its existing SubConns.
+ Close()
+}
+
// ConnectivityStateEvaluator takes the connectivity states of multiple SubConns
// and returns one aggregated connectivity state.
//
diff --git a/vendor/google.golang.org/grpc/balancer/base/balancer.go b/vendor/google.golang.org/grpc/balancer/base/balancer.go
index 245785e..1af88f0 100644
--- a/vendor/google.golang.org/grpc/balancer/base/balancer.go
+++ b/vendor/google.golang.org/grpc/balancer/base/balancer.go
@@ -67,14 +67,18 @@
}
func (b *baseBalancer) HandleResolvedAddrs(addrs []resolver.Address, err error) {
- if err != nil {
- grpclog.Infof("base.baseBalancer: HandleResolvedAddrs called with error %v", err)
- return
+ panic("not implemented")
+}
+
+func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) {
+ // TODO: handle s.ResolverState.Err (log if not nil) once implemented.
+ // TODO: handle s.ResolverState.ServiceConfig?
+ if grpclog.V(2) {
+ grpclog.Infoln("base.baseBalancer: got new ClientConn state: ", s)
}
- grpclog.Infoln("base.baseBalancer: got new resolved addresses: ", addrs)
// addrsSet is the set converted from addrs, it's used for quick lookup of an address.
addrsSet := make(map[resolver.Address]struct{})
- for _, a := range addrs {
+ for _, a := range s.ResolverState.Addresses {
addrsSet[a] = struct{}{}
if _, ok := b.subConns[a]; !ok {
// a is a new address (not existing in b.subConns).
@@ -120,10 +124,19 @@
}
func (b *baseBalancer) HandleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
- grpclog.Infof("base.baseBalancer: handle SubConn state change: %p, %v", sc, s)
+ panic("not implemented")
+}
+
+func (b *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
+ s := state.ConnectivityState
+ if grpclog.V(2) {
+ grpclog.Infof("base.baseBalancer: handle SubConn state change: %p, %v", sc, s)
+ }
oldS, ok := b.scStates[sc]
if !ok {
- grpclog.Infof("base.baseBalancer: got state changes for an unknown SubConn: %p, %v", sc, s)
+ if grpclog.V(2) {
+ grpclog.Infof("base.baseBalancer: got state changes for an unknown SubConn: %p, %v", sc, s)
+ }
return
}
b.scStates[sc] = s
diff --git a/vendor/google.golang.org/grpc/balancer_conn_wrappers.go b/vendor/google.golang.org/grpc/balancer_conn_wrappers.go
index 7233ade..8df4095 100644
--- a/vendor/google.golang.org/grpc/balancer_conn_wrappers.go
+++ b/vendor/google.golang.org/grpc/balancer_conn_wrappers.go
@@ -82,20 +82,13 @@
return b.c
}
-// resolverUpdate contains the new resolved addresses or error if there's
-// any.
-type resolverUpdate struct {
- addrs []resolver.Address
- err error
-}
-
// ccBalancerWrapper is a wrapper on top of cc for balancers.
// It implements balancer.ClientConn interface.
type ccBalancerWrapper struct {
cc *ClientConn
balancer balancer.Balancer
stateChangeQueue *scStateUpdateBuffer
- resolverUpdateCh chan *resolverUpdate
+ ccUpdateCh chan *balancer.ClientConnState
done chan struct{}
mu sync.Mutex
@@ -106,7 +99,7 @@
ccb := &ccBalancerWrapper{
cc: cc,
stateChangeQueue: newSCStateUpdateBuffer(),
- resolverUpdateCh: make(chan *resolverUpdate, 1),
+ ccUpdateCh: make(chan *balancer.ClientConnState, 1),
done: make(chan struct{}),
subConns: make(map[*acBalancerWrapper]struct{}),
}
@@ -128,15 +121,23 @@
return
default:
}
- ccb.balancer.HandleSubConnStateChange(t.sc, t.state)
- case t := <-ccb.resolverUpdateCh:
+ if ub, ok := ccb.balancer.(balancer.V2Balancer); ok {
+ ub.UpdateSubConnState(t.sc, balancer.SubConnState{ConnectivityState: t.state})
+ } else {
+ ccb.balancer.HandleSubConnStateChange(t.sc, t.state)
+ }
+ case s := <-ccb.ccUpdateCh:
select {
case <-ccb.done:
ccb.balancer.Close()
return
default:
}
- ccb.balancer.HandleResolvedAddrs(t.addrs, t.err)
+ if ub, ok := ccb.balancer.(balancer.V2Balancer); ok {
+ ub.UpdateClientConnState(*s)
+ } else {
+ ccb.balancer.HandleResolvedAddrs(s.ResolverState.Addresses, nil)
+ }
case <-ccb.done:
}
@@ -150,9 +151,11 @@
for acbw := range scs {
ccb.cc.removeAddrConn(acbw.getAddrConn(), errConnDrain)
}
+ ccb.UpdateBalancerState(connectivity.Connecting, nil)
return
default:
}
+ ccb.cc.firstResolveEvent.Fire()
}
}
@@ -177,37 +180,24 @@
})
}
-func (ccb *ccBalancerWrapper) handleResolvedAddrs(addrs []resolver.Address, err error) {
+func (ccb *ccBalancerWrapper) updateClientConnState(ccs *balancer.ClientConnState) {
if ccb.cc.curBalancerName != grpclbName {
- var containsGRPCLB bool
- for _, a := range addrs {
- if a.Type == resolver.GRPCLB {
- containsGRPCLB = true
- break
+ // Filter any grpclb addresses since we don't have the grpclb balancer.
+ s := &ccs.ResolverState
+ for i := 0; i < len(s.Addresses); {
+ if s.Addresses[i].Type == resolver.GRPCLB {
+ copy(s.Addresses[i:], s.Addresses[i+1:])
+ s.Addresses = s.Addresses[:len(s.Addresses)-1]
+ continue
}
- }
- if containsGRPCLB {
- // The current balancer is not grpclb, but addresses contain grpclb
- // address. This means we failed to switch to grpclb, most likely
- // because grpclb is not registered. Filter out all grpclb addresses
- // from addrs before sending to balancer.
- tempAddrs := make([]resolver.Address, 0, len(addrs))
- for _, a := range addrs {
- if a.Type != resolver.GRPCLB {
- tempAddrs = append(tempAddrs, a)
- }
- }
- addrs = tempAddrs
+ i++
}
}
select {
- case <-ccb.resolverUpdateCh:
+ case <-ccb.ccUpdateCh:
default:
}
- ccb.resolverUpdateCh <- &resolverUpdate{
- addrs: addrs,
- err: err,
- }
+ ccb.ccUpdateCh <- ccs
}
func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
diff --git a/vendor/google.golang.org/grpc/balancer_v1_wrapper.go b/vendor/google.golang.org/grpc/balancer_v1_wrapper.go
index 29bda63..66e9a44 100644
--- a/vendor/google.golang.org/grpc/balancer_v1_wrapper.go
+++ b/vendor/google.golang.org/grpc/balancer_v1_wrapper.go
@@ -20,7 +20,6 @@
import (
"context"
- "strings"
"sync"
"google.golang.org/grpc/balancer"
@@ -34,13 +33,7 @@
}
func (bwb *balancerWrapperBuilder) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer {
- targetAddr := cc.Target()
- targetSplitted := strings.Split(targetAddr, ":///")
- if len(targetSplitted) >= 2 {
- targetAddr = targetSplitted[1]
- }
-
- bwb.b.Start(targetAddr, BalancerConfig{
+ bwb.b.Start(opts.Target.Endpoint, BalancerConfig{
DialCreds: opts.DialCreds,
Dialer: opts.Dialer,
})
@@ -49,7 +42,7 @@
balancer: bwb.b,
pickfirst: pickfirst,
cc: cc,
- targetAddr: targetAddr,
+ targetAddr: opts.Target.Endpoint,
startCh: make(chan struct{}),
conns: make(map[resolver.Address]balancer.SubConn),
connSt: make(map[balancer.SubConn]*scState),
@@ -120,7 +113,7 @@
}
for addrs := range notifyCh {
- grpclog.Infof("balancerWrapper: got update addr from Notify: %v\n", addrs)
+ grpclog.Infof("balancerWrapper: got update addr from Notify: %v", addrs)
if bw.pickfirst {
var (
oldA resolver.Address
diff --git a/vendor/google.golang.org/grpc/clientconn.go b/vendor/google.golang.org/grpc/clientconn.go
index df1bb94..a7643df 100644
--- a/vendor/google.golang.org/grpc/clientconn.go
+++ b/vendor/google.golang.org/grpc/clientconn.go
@@ -38,14 +38,13 @@
"google.golang.org/grpc/grpclog"
"google.golang.org/grpc/internal/backoff"
"google.golang.org/grpc/internal/channelz"
- "google.golang.org/grpc/internal/envconfig"
"google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/internal/transport"
"google.golang.org/grpc/keepalive"
- "google.golang.org/grpc/metadata"
"google.golang.org/grpc/resolver"
_ "google.golang.org/grpc/resolver/dns" // To register dns resolver.
_ "google.golang.org/grpc/resolver/passthrough" // To register passthrough resolver.
+ "google.golang.org/grpc/serviceconfig"
"google.golang.org/grpc/status"
)
@@ -69,11 +68,9 @@
errConnClosing = errors.New("grpc: the connection is closing")
// errBalancerClosed indicates that the balancer is closed.
errBalancerClosed = errors.New("grpc: balancer is closed")
- // We use an accessor so that minConnectTimeout can be
- // atomically read and updated while testing.
- getMinConnectTimeout = func() time.Duration {
- return minConnectTimeout
- }
+ // invalidDefaultServiceConfigErrPrefix is used to prefix the json parsing error for the default
+ // service config.
+ invalidDefaultServiceConfigErrPrefix = "grpc: the provided default service config is invalid"
)
// The following errors are returned from Dial and DialContext
@@ -140,6 +137,15 @@
opt.apply(&cc.dopts)
}
+ chainUnaryClientInterceptors(cc)
+ chainStreamClientInterceptors(cc)
+
+ defer func() {
+ if err != nil {
+ cc.Close()
+ }
+ }()
+
if channelz.IsOn() {
if cc.dopts.channelzParentID != 0 {
cc.channelzID = channelz.RegisterChannel(&channelzChannel{cc}, cc.dopts.channelzParentID, target)
@@ -179,6 +185,13 @@
}
}
+ if cc.dopts.defaultServiceConfigRawJSON != nil {
+ sc, err := parseServiceConfig(*cc.dopts.defaultServiceConfigRawJSON)
+ if err != nil {
+ return nil, fmt.Errorf("%s: %v", invalidDefaultServiceConfigErrPrefix, err)
+ }
+ cc.dopts.defaultServiceConfig = sc
+ }
cc.mkp = cc.dopts.copts.KeepaliveParams
if cc.dopts.copts.Dialer == nil {
@@ -201,17 +214,12 @@
ctx, cancel = context.WithTimeout(ctx, cc.dopts.timeout)
defer cancel()
}
-
defer func() {
select {
case <-ctx.Done():
conn, err = nil, ctx.Err()
default:
}
-
- if err != nil {
- cc.Close()
- }
}()
scSet := false
@@ -220,7 +228,7 @@
select {
case sc, ok := <-cc.dopts.scChan:
if ok {
- cc.sc = sc
+ cc.sc = &sc
scSet = true
}
default:
@@ -266,7 +274,7 @@
select {
case sc, ok := <-cc.dopts.scChan:
if ok {
- cc.sc = sc
+ cc.sc = &sc
}
case <-ctx.Done():
return nil, ctx.Err()
@@ -285,6 +293,7 @@
CredsBundle: cc.dopts.copts.CredsBundle,
Dialer: cc.dopts.copts.Dialer,
ChannelzParentID: cc.channelzID,
+ Target: cc.parsedTarget,
}
// Build the resolver.
@@ -322,6 +331,68 @@
return cc, nil
}
+// chainUnaryClientInterceptors chains all unary client interceptors into one.
+func chainUnaryClientInterceptors(cc *ClientConn) {
+ interceptors := cc.dopts.chainUnaryInts
+ // Prepend dopts.unaryInt to the chaining interceptors if it exists, since unaryInt will
+ // be executed before any other chained interceptors.
+ if cc.dopts.unaryInt != nil {
+ interceptors = append([]UnaryClientInterceptor{cc.dopts.unaryInt}, interceptors...)
+ }
+ var chainedInt UnaryClientInterceptor
+ if len(interceptors) == 0 {
+ chainedInt = nil
+ } else if len(interceptors) == 1 {
+ chainedInt = interceptors[0]
+ } else {
+ chainedInt = func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error {
+ return interceptors[0](ctx, method, req, reply, cc, getChainUnaryInvoker(interceptors, 0, invoker), opts...)
+ }
+ }
+ cc.dopts.unaryInt = chainedInt
+}
+
+// getChainUnaryInvoker recursively generate the chained unary invoker.
+func getChainUnaryInvoker(interceptors []UnaryClientInterceptor, curr int, finalInvoker UnaryInvoker) UnaryInvoker {
+ if curr == len(interceptors)-1 {
+ return finalInvoker
+ }
+ return func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error {
+ return interceptors[curr+1](ctx, method, req, reply, cc, getChainUnaryInvoker(interceptors, curr+1, finalInvoker), opts...)
+ }
+}
+
+// chainStreamClientInterceptors chains all stream client interceptors into one.
+func chainStreamClientInterceptors(cc *ClientConn) {
+ interceptors := cc.dopts.chainStreamInts
+ // Prepend dopts.streamInt to the chaining interceptors if it exists, since streamInt will
+ // be executed before any other chained interceptors.
+ if cc.dopts.streamInt != nil {
+ interceptors = append([]StreamClientInterceptor{cc.dopts.streamInt}, interceptors...)
+ }
+ var chainedInt StreamClientInterceptor
+ if len(interceptors) == 0 {
+ chainedInt = nil
+ } else if len(interceptors) == 1 {
+ chainedInt = interceptors[0]
+ } else {
+ chainedInt = func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error) {
+ return interceptors[0](ctx, desc, cc, method, getChainStreamer(interceptors, 0, streamer), opts...)
+ }
+ }
+ cc.dopts.streamInt = chainedInt
+}
+
+// getChainStreamer recursively generate the chained client stream constructor.
+func getChainStreamer(interceptors []StreamClientInterceptor, curr int, finalStreamer Streamer) Streamer {
+ if curr == len(interceptors)-1 {
+ return finalStreamer
+ }
+ return func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
+ return interceptors[curr+1](ctx, desc, cc, method, getChainStreamer(interceptors, curr+1, finalStreamer), opts...)
+ }
+}
+
// connectivityStateManager keeps the connectivity.State of ClientConn.
// This struct will eventually be exported so the balancers can access it.
type connectivityStateManager struct {
@@ -388,14 +459,11 @@
mu sync.RWMutex
resolverWrapper *ccResolverWrapper
- sc ServiceConfig
- scRaw string
+ sc *ServiceConfig
conns map[*addrConn]struct{}
// Keepalive parameter can be updated if a GoAway is received.
mkp keepalive.ClientParameters
curBalancerName string
- preBalancerName string // previous balancer name.
- curAddresses []resolver.Address
balancerWrapper *ccBalancerWrapper
retryThrottler atomic.Value
@@ -437,8 +505,7 @@
cc.mu.Lock()
// TODO: load balance policy runtime change is ignored.
// We may revisit this decision in the future.
- cc.sc = sc
- cc.scRaw = ""
+ cc.sc = &sc
cc.mu.Unlock()
case <-cc.ctx.Done():
return
@@ -465,48 +532,45 @@
}
}
-func (cc *ClientConn) handleResolvedAddrs(addrs []resolver.Address, err error) {
+func (cc *ClientConn) updateResolverState(s resolver.State) error {
cc.mu.Lock()
defer cc.mu.Unlock()
+ // Check if the ClientConn is already closed. Some fields (e.g.
+ // balancerWrapper) are set to nil when closing the ClientConn, and could
+ // cause nil pointer panic if we don't have this check.
if cc.conns == nil {
- // cc was closed.
- return
+ return nil
}
- if reflect.DeepEqual(cc.curAddresses, addrs) {
- return
+ if cc.dopts.disableServiceConfig || s.ServiceConfig == nil {
+ if cc.dopts.defaultServiceConfig != nil && cc.sc == nil {
+ cc.applyServiceConfig(cc.dopts.defaultServiceConfig)
+ }
+ } else if sc, ok := s.ServiceConfig.(*ServiceConfig); ok {
+ cc.applyServiceConfig(sc)
}
- cc.curAddresses = addrs
- cc.firstResolveEvent.Fire()
-
+ var balCfg serviceconfig.LoadBalancingConfig
if cc.dopts.balancerBuilder == nil {
// Only look at balancer types and switch balancer if balancer dial
// option is not set.
- var isGRPCLB bool
- for _, a := range addrs {
- if a.Type == resolver.GRPCLB {
- isGRPCLB = true
- break
- }
- }
var newBalancerName string
- if isGRPCLB {
- newBalancerName = grpclbName
+ if cc.sc != nil && cc.sc.lbConfig != nil {
+ newBalancerName = cc.sc.lbConfig.name
+ balCfg = cc.sc.lbConfig.cfg
} else {
- // Address list doesn't contain grpclb address. Try to pick a
- // non-grpclb balancer.
- newBalancerName = cc.curBalancerName
- // If current balancer is grpclb, switch to the previous one.
- if newBalancerName == grpclbName {
- newBalancerName = cc.preBalancerName
+ var isGRPCLB bool
+ for _, a := range s.Addresses {
+ if a.Type == resolver.GRPCLB {
+ isGRPCLB = true
+ break
+ }
}
- // The following could be true in two cases:
- // - the first time handling resolved addresses
- // (curBalancerName="")
- // - the first time handling non-grpclb addresses
- // (curBalancerName="grpclb", preBalancerName="")
- if newBalancerName == "" {
+ if isGRPCLB {
+ newBalancerName = grpclbName
+ } else if cc.sc != nil && cc.sc.LB != nil {
+ newBalancerName = *cc.sc.LB
+ } else {
newBalancerName = PickFirstBalancerName
}
}
@@ -514,10 +578,12 @@
} else if cc.balancerWrapper == nil {
// Balancer dial option was set, and this is the first time handling
// resolved addresses. Build a balancer with dopts.balancerBuilder.
+ cc.curBalancerName = cc.dopts.balancerBuilder.Name()
cc.balancerWrapper = newCCBalancerWrapper(cc, cc.dopts.balancerBuilder, cc.balancerBuildOpts)
}
- cc.balancerWrapper.handleResolvedAddrs(addrs, nil)
+ cc.balancerWrapper.updateClientConnState(&balancer.ClientConnState{ResolverState: s, BalancerConfig: balCfg})
+ return nil
}
// switchBalancer starts the switching from current balancer to the balancer
@@ -529,11 +595,7 @@
//
// Caller must hold cc.mu.
func (cc *ClientConn) switchBalancer(name string) {
- if cc.conns == nil {
- return
- }
-
- if strings.ToLower(cc.curBalancerName) == strings.ToLower(name) {
+ if strings.EqualFold(cc.curBalancerName, name) {
return
}
@@ -542,15 +604,11 @@
grpclog.Infoln("ignoring balancer switching: Balancer DialOption used instead")
return
}
- // TODO(bar switching) change this to two steps: drain and close.
- // Keep track of sc in wrapper.
if cc.balancerWrapper != nil {
cc.balancerWrapper.close()
}
builder := balancer.Get(name)
- // TODO(yuxuanli): If user send a service config that does not contain a valid balancer name, should
- // we reuse previous one?
if channelz.IsOn() {
if builder == nil {
channelz.AddTraceEvent(cc.channelzID, &channelz.TraceEventDesc{
@@ -569,7 +627,6 @@
builder = newPickfirstBuilder()
}
- cc.preBalancerName = cc.curBalancerName
cc.curBalancerName = builder.Name()
cc.balancerWrapper = newCCBalancerWrapper(cc, builder, cc.balancerBuildOpts)
}
@@ -677,6 +734,8 @@
ac.mu.Unlock()
return nil
}
+ // Update connectivity state within the lock to prevent subsequent or
+ // concurrent calls from resetting the transport more than once.
ac.updateConnectivityState(connectivity.Connecting)
ac.mu.Unlock()
@@ -687,7 +746,16 @@
// tryUpdateAddrs tries to update ac.addrs with the new addresses list.
//
-// It checks whether current connected address of ac is in the new addrs list.
+// If ac is Connecting, it returns false. The caller should tear down the ac and
+// create a new one. Note that the backoff will be reset when this happens.
+//
+// If ac is TransientFailure, it updates ac.addrs and returns true. The updated
+// addresses will be picked up by retry in the next iteration after backoff.
+//
+// If ac is Shutdown or Idle, it updates ac.addrs and returns true.
+//
+// If ac is Ready, it checks whether current connected address of ac is in the
+// new addrs list.
// - If true, it updates ac.addrs and returns true. The ac will keep using
// the existing connection.
// - If false, it does nothing and returns false.
@@ -695,17 +763,18 @@
ac.mu.Lock()
defer ac.mu.Unlock()
grpclog.Infof("addrConn: tryUpdateAddrs curAddr: %v, addrs: %v", ac.curAddr, addrs)
- if ac.state == connectivity.Shutdown {
+ if ac.state == connectivity.Shutdown ||
+ ac.state == connectivity.TransientFailure ||
+ ac.state == connectivity.Idle {
ac.addrs = addrs
return true
}
- // Unless we're busy reconnecting already, let's reconnect from the top of
- // the list.
- if ac.state != connectivity.Ready {
+ if ac.state == connectivity.Connecting {
return false
}
+ // ac.state is Ready, try to find the connected address.
var curAddrFound bool
for _, a := range addrs {
if reflect.DeepEqual(ac.curAddr, a) {
@@ -732,6 +801,9 @@
// TODO: Avoid the locking here.
cc.mu.RLock()
defer cc.mu.RUnlock()
+ if cc.sc == nil {
+ return MethodConfig{}
+ }
m, ok := cc.sc.Methods[method]
if !ok {
i := strings.LastIndex(method, "/")
@@ -743,14 +815,15 @@
func (cc *ClientConn) healthCheckConfig() *healthCheckConfig {
cc.mu.RLock()
defer cc.mu.RUnlock()
+ if cc.sc == nil {
+ return nil
+ }
return cc.sc.healthCheckConfig
}
func (cc *ClientConn) getTransport(ctx context.Context, failfast bool, method string) (transport.ClientTransport, func(balancer.DoneInfo), error) {
- hdr, _ := metadata.FromOutgoingContext(ctx)
t, done, err := cc.blockingpicker.pick(ctx, failfast, balancer.PickOptions{
FullMethodName: method,
- Header: hdr,
})
if err != nil {
return nil, nil, toRPCErr(err)
@@ -758,65 +831,25 @@
return t, done, nil
}
-// handleServiceConfig parses the service config string in JSON format to Go native
-// struct ServiceConfig, and store both the struct and the JSON string in ClientConn.
-func (cc *ClientConn) handleServiceConfig(js string) error {
- if cc.dopts.disableServiceConfig {
- return nil
+func (cc *ClientConn) applyServiceConfig(sc *ServiceConfig) error {
+ if sc == nil {
+ // should never reach here.
+ return fmt.Errorf("got nil pointer for service config")
}
- if cc.scRaw == js {
- return nil
- }
- if channelz.IsOn() {
- channelz.AddTraceEvent(cc.channelzID, &channelz.TraceEventDesc{
- // The special formatting of \"%s\" instead of %q is to provide nice printing of service config
- // for human consumption.
- Desc: fmt.Sprintf("Channel has a new service config \"%s\"", js),
- Severity: channelz.CtINFO,
- })
- }
- sc, err := parseServiceConfig(js)
- if err != nil {
- return err
- }
- cc.mu.Lock()
- // Check if the ClientConn is already closed. Some fields (e.g.
- // balancerWrapper) are set to nil when closing the ClientConn, and could
- // cause nil pointer panic if we don't have this check.
- if cc.conns == nil {
- cc.mu.Unlock()
- return nil
- }
- cc.scRaw = js
cc.sc = sc
- if sc.retryThrottling != nil {
+ if cc.sc.retryThrottling != nil {
newThrottler := &retryThrottler{
- tokens: sc.retryThrottling.MaxTokens,
- max: sc.retryThrottling.MaxTokens,
- thresh: sc.retryThrottling.MaxTokens / 2,
- ratio: sc.retryThrottling.TokenRatio,
+ tokens: cc.sc.retryThrottling.MaxTokens,
+ max: cc.sc.retryThrottling.MaxTokens,
+ thresh: cc.sc.retryThrottling.MaxTokens / 2,
+ ratio: cc.sc.retryThrottling.TokenRatio,
}
cc.retryThrottler.Store(newThrottler)
} else {
cc.retryThrottler.Store((*retryThrottler)(nil))
}
- if sc.LB != nil && *sc.LB != grpclbName { // "grpclb" is not a valid balancer option in service config.
- if cc.curBalancerName == grpclbName {
- // If current balancer is grpclb, there's at least one grpclb
- // balancer address in the resolved list. Don't switch the balancer,
- // but change the previous balancer name, so if a new resolved
- // address list doesn't contain grpclb address, balancer will be
- // switched to *sc.LB.
- cc.preBalancerName = *sc.LB
- } else {
- cc.switchBalancer(*sc.LB)
- cc.balancerWrapper.handleResolvedAddrs(cc.curAddresses, nil)
- }
- }
-
- cc.mu.Unlock()
return nil
}
@@ -892,7 +925,7 @@
}
channelz.AddTraceEvent(cc.channelzID, ted)
// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to
- // the entity beng deleted, and thus prevent it from being deleted right away.
+ // the entity being deleted, and thus prevent it from being deleted right away.
channelz.RemoveEntry(cc.channelzID)
}
return nil
@@ -921,8 +954,6 @@
// Use updateConnectivityState for updating addrConn's connectivity state.
state connectivity.State
- tearDownErr error // The reason this addrConn is torn down.
-
backoffIdx int // Needs to be stateful for resetConnectBackoff.
resetBackoff chan struct{}
@@ -963,191 +994,147 @@
func (ac *addrConn) resetTransport() {
for i := 0; ; i++ {
- tryNextAddrFromStart := grpcsync.NewEvent()
-
- ac.mu.Lock()
if i > 0 {
ac.cc.resolveNow(resolver.ResolveNowOption{})
}
- addrs := ac.addrs
- backoffFor := ac.dopts.bs.Backoff(ac.backoffIdx)
- // This will be the duration that dial gets to finish.
- dialDuration := getMinConnectTimeout()
- if dialDuration < backoffFor {
- // Give dial more time as we keep failing to connect.
- dialDuration = backoffFor
- }
- connectDeadline := time.Now().Add(dialDuration)
- ac.mu.Unlock()
-
- addrLoop:
- for _, addr := range addrs {
- ac.mu.Lock()
-
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return
- }
- ac.updateConnectivityState(connectivity.Connecting)
- ac.transport = nil
-
- ac.cc.mu.RLock()
- ac.dopts.copts.KeepaliveParams = ac.cc.mkp
- ac.cc.mu.RUnlock()
-
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return
- }
-
- copts := ac.dopts.copts
- if ac.scopts.CredsBundle != nil {
- copts.CredsBundle = ac.scopts.CredsBundle
- }
- hctx, hcancel := context.WithCancel(ac.ctx)
- defer hcancel()
- ac.mu.Unlock()
-
- if channelz.IsOn() {
- channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Subchannel picks a new address %q to connect", addr.Addr),
- Severity: channelz.CtINFO,
- })
- }
-
- reconnect := grpcsync.NewEvent()
- prefaceReceived := make(chan struct{})
- newTr, err := ac.createTransport(addr, copts, connectDeadline, reconnect, prefaceReceived)
- if err == nil {
- ac.mu.Lock()
- ac.curAddr = addr
- ac.transport = newTr
- ac.mu.Unlock()
-
- healthCheckConfig := ac.cc.healthCheckConfig()
- // LB channel health checking is only enabled when all the four requirements below are met:
- // 1. it is not disabled by the user with the WithDisableHealthCheck DialOption,
- // 2. the internal.HealthCheckFunc is set by importing the grpc/healthcheck package,
- // 3. a service config with non-empty healthCheckConfig field is provided,
- // 4. the current load balancer allows it.
- healthcheckManagingState := false
- if !ac.cc.dopts.disableHealthCheck && healthCheckConfig != nil && ac.scopts.HealthCheckEnabled {
- if ac.cc.dopts.healthCheckFunc == nil {
- // TODO: add a link to the health check doc in the error message.
- grpclog.Error("the client side LB channel health check function has not been set.")
- } else {
- // TODO(deklerk) refactor to just return transport
- go ac.startHealthCheck(hctx, newTr, addr, healthCheckConfig.ServiceName)
- healthcheckManagingState = true
- }
- }
- if !healthcheckManagingState {
- ac.mu.Lock()
- ac.updateConnectivityState(connectivity.Ready)
- ac.mu.Unlock()
- }
- } else {
- hcancel()
- if err == errConnClosing {
- return
- }
-
- if tryNextAddrFromStart.HasFired() {
- break addrLoop
- }
- continue
- }
-
- backoffFor = 0
- ac.mu.Lock()
- reqHandshake := ac.dopts.reqHandshake
- ac.mu.Unlock()
-
- <-reconnect.Done()
- hcancel()
-
- if reqHandshake == envconfig.RequireHandshakeHybrid {
- // In RequireHandshakeHybrid mode, we must check to see whether
- // server preface has arrived yet to decide whether to start
- // reconnecting at the top of the list (server preface received)
- // or continue with the next addr in the list as if the
- // connection were not successful (server preface not received).
- select {
- case <-prefaceReceived:
- // We received a server preface - huzzah! We consider this
- // a success and restart from the top of the addr list.
- ac.mu.Lock()
- ac.backoffIdx = 0
- ac.mu.Unlock()
- break addrLoop
- default:
- // Despite having set state to READY, in hybrid mode we
- // consider this a failure and continue connecting at the
- // next addr in the list.
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return
- }
-
- ac.updateConnectivityState(connectivity.TransientFailure)
- ac.mu.Unlock()
-
- if tryNextAddrFromStart.HasFired() {
- break addrLoop
- }
- }
- } else {
- // In RequireHandshakeOn mode, we would have already waited for
- // the server preface, so we consider this a success and restart
- // from the top of the addr list. In RequireHandshakeOff mode,
- // we don't care to wait for the server preface before
- // considering this a success, so we also restart from the top
- // of the addr list.
- ac.mu.Lock()
- ac.backoffIdx = 0
- ac.mu.Unlock()
- break addrLoop
- }
- }
-
- // After exhausting all addresses, or after need to reconnect after a
- // READY, the addrConn enters TRANSIENT_FAILURE.
ac.mu.Lock()
if ac.state == connectivity.Shutdown {
ac.mu.Unlock()
return
}
- ac.updateConnectivityState(connectivity.TransientFailure)
- // Backoff.
- b := ac.resetBackoff
- timer := time.NewTimer(backoffFor)
- acctx := ac.ctx
+ addrs := ac.addrs
+ backoffFor := ac.dopts.bs.Backoff(ac.backoffIdx)
+ // This will be the duration that dial gets to finish.
+ dialDuration := minConnectTimeout
+ if ac.dopts.minConnectTimeout != nil {
+ dialDuration = ac.dopts.minConnectTimeout()
+ }
+
+ if dialDuration < backoffFor {
+ // Give dial more time as we keep failing to connect.
+ dialDuration = backoffFor
+ }
+ // We can potentially spend all the time trying the first address, and
+ // if the server accepts the connection and then hangs, the following
+ // addresses will never be tried.
+ //
+ // The spec doesn't mention what should be done for multiple addresses.
+ // https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md#proposed-backoff-algorithm
+ connectDeadline := time.Now().Add(dialDuration)
+
+ ac.updateConnectivityState(connectivity.Connecting)
+ ac.transport = nil
ac.mu.Unlock()
- select {
- case <-timer.C:
+ newTr, addr, reconnect, err := ac.tryAllAddrs(addrs, connectDeadline)
+ if err != nil {
+ // After exhausting all addresses, the addrConn enters
+ // TRANSIENT_FAILURE.
ac.mu.Lock()
- ac.backoffIdx++
+ if ac.state == connectivity.Shutdown {
+ ac.mu.Unlock()
+ return
+ }
+ ac.updateConnectivityState(connectivity.TransientFailure)
+
+ // Backoff.
+ b := ac.resetBackoff
ac.mu.Unlock()
- case <-b:
- timer.Stop()
- case <-acctx.Done():
- timer.Stop()
+
+ timer := time.NewTimer(backoffFor)
+ select {
+ case <-timer.C:
+ ac.mu.Lock()
+ ac.backoffIdx++
+ ac.mu.Unlock()
+ case <-b:
+ timer.Stop()
+ case <-ac.ctx.Done():
+ timer.Stop()
+ return
+ }
+ continue
+ }
+
+ ac.mu.Lock()
+ if ac.state == connectivity.Shutdown {
+ ac.mu.Unlock()
+ newTr.Close()
return
}
+ ac.curAddr = addr
+ ac.transport = newTr
+ ac.backoffIdx = 0
+
+ hctx, hcancel := context.WithCancel(ac.ctx)
+ ac.startHealthCheck(hctx)
+ ac.mu.Unlock()
+
+ // Block until the created transport is down. And when this happens,
+ // we restart from the top of the addr list.
+ <-reconnect.Done()
+ hcancel()
+ // restart connecting - the top of the loop will set state to
+ // CONNECTING. This is against the current connectivity semantics doc,
+ // however it allows for graceful behavior for RPCs not yet dispatched
+ // - unfortunate timing would otherwise lead to the RPC failing even
+ // though the TRANSIENT_FAILURE state (called for by the doc) would be
+ // instantaneous.
+ //
+ // Ideally we should transition to Idle here and block until there is
+ // RPC activity that leads to the balancer requesting a reconnect of
+ // the associated SubConn.
}
}
-// createTransport creates a connection to one of the backends in addrs. It
-// sets ac.transport in the success case, or it returns an error if it was
-// unable to successfully create a transport.
-//
-// If waitForHandshake is enabled, it blocks until server preface arrives.
-func (ac *addrConn) createTransport(addr resolver.Address, copts transport.ConnectOptions, connectDeadline time.Time, reconnect *grpcsync.Event, prefaceReceived chan struct{}) (transport.ClientTransport, error) {
+// tryAllAddrs tries to creates a connection to the addresses, and stop when at the
+// first successful one. It returns the transport, the address and a Event in
+// the successful case. The Event fires when the returned transport disconnects.
+func (ac *addrConn) tryAllAddrs(addrs []resolver.Address, connectDeadline time.Time) (transport.ClientTransport, resolver.Address, *grpcsync.Event, error) {
+ for _, addr := range addrs {
+ ac.mu.Lock()
+ if ac.state == connectivity.Shutdown {
+ ac.mu.Unlock()
+ return nil, resolver.Address{}, nil, errConnClosing
+ }
+
+ ac.cc.mu.RLock()
+ ac.dopts.copts.KeepaliveParams = ac.cc.mkp
+ ac.cc.mu.RUnlock()
+
+ copts := ac.dopts.copts
+ if ac.scopts.CredsBundle != nil {
+ copts.CredsBundle = ac.scopts.CredsBundle
+ }
+ ac.mu.Unlock()
+
+ if channelz.IsOn() {
+ channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
+ Desc: fmt.Sprintf("Subchannel picks a new address %q to connect", addr.Addr),
+ Severity: channelz.CtINFO,
+ })
+ }
+
+ newTr, reconnect, err := ac.createTransport(addr, copts, connectDeadline)
+ if err == nil {
+ return newTr, addr, reconnect, nil
+ }
+ ac.cc.blockingpicker.updateConnectionError(err)
+ }
+
+ // Couldn't connect to any address.
+ return nil, resolver.Address{}, nil, fmt.Errorf("couldn't connect to any address")
+}
+
+// createTransport creates a connection to addr. It returns the transport and a
+// Event in the successful case. The Event fires when the returned transport
+// disconnects.
+func (ac *addrConn) createTransport(addr resolver.Address, copts transport.ConnectOptions, connectDeadline time.Time) (transport.ClientTransport, *grpcsync.Event, error) {
+ prefaceReceived := make(chan struct{})
onCloseCalled := make(chan struct{})
+ reconnect := grpcsync.NewEvent()
target := transport.TargetInfo{
Addr: addr.Addr,
@@ -1155,24 +1142,41 @@
Authority: ac.cc.authority,
}
- prefaceTimer := time.NewTimer(time.Until(connectDeadline))
-
+ once := sync.Once{}
onGoAway := func(r transport.GoAwayReason) {
ac.mu.Lock()
ac.adjustParams(r)
+ once.Do(func() {
+ if ac.state == connectivity.Ready {
+ // Prevent this SubConn from being used for new RPCs by setting its
+ // state to Connecting.
+ //
+ // TODO: this should be Idle when grpc-go properly supports it.
+ ac.updateConnectivityState(connectivity.Connecting)
+ }
+ })
ac.mu.Unlock()
reconnect.Fire()
}
onClose := func() {
+ ac.mu.Lock()
+ once.Do(func() {
+ if ac.state == connectivity.Ready {
+ // Prevent this SubConn from being used for new RPCs by setting its
+ // state to Connecting.
+ //
+ // TODO: this should be Idle when grpc-go properly supports it.
+ ac.updateConnectivityState(connectivity.Connecting)
+ }
+ })
+ ac.mu.Unlock()
close(onCloseCalled)
- prefaceTimer.Stop()
reconnect.Fire()
}
onPrefaceReceipt := func() {
close(prefaceReceived)
- prefaceTimer.Stop()
}
connectCtx, cancel := context.WithDeadline(ac.ctx, connectDeadline)
@@ -1182,107 +1186,105 @@
}
newTr, err := transport.NewClientTransport(connectCtx, ac.cc.ctx, target, copts, onPrefaceReceipt, onGoAway, onClose)
-
- if err == nil {
- if ac.dopts.reqHandshake == envconfig.RequireHandshakeOn {
- select {
- case <-prefaceTimer.C:
- // We didn't get the preface in time.
- newTr.Close()
- err = errors.New("timed out waiting for server handshake")
- case <-prefaceReceived:
- // We got the preface - huzzah! things are good.
- case <-onCloseCalled:
- // The transport has already closed - noop.
- return nil, errors.New("connection closed")
- }
- } else if ac.dopts.reqHandshake == envconfig.RequireHandshakeHybrid {
- go func() {
- select {
- case <-prefaceTimer.C:
- // We didn't get the preface in time.
- newTr.Close()
- case <-prefaceReceived:
- // We got the preface just in the nick of time - huzzah!
- case <-onCloseCalled:
- // The transport has already closed - noop.
- }
- }()
- }
- }
-
if err != nil {
// newTr is either nil, or closed.
- ac.cc.blockingpicker.updateConnectionError(err)
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- // ac.tearDown(...) has been invoked.
- ac.mu.Unlock()
-
- return nil, errConnClosing
- }
- ac.mu.Unlock()
grpclog.Warningf("grpc: addrConn.createTransport failed to connect to %v. Err :%v. Reconnecting...", addr, err)
- return nil, err
+ return nil, nil, err
}
- // Now there is a viable transport to be use, so set ac.transport to reflect the new viable transport.
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
+ select {
+ case <-time.After(connectDeadline.Sub(time.Now())):
+ // We didn't get the preface in time.
newTr.Close()
- return nil, errConnClosing
+ grpclog.Warningf("grpc: addrConn.createTransport failed to connect to %v: didn't receive server preface in time. Reconnecting...", addr)
+ return nil, nil, errors.New("timed out waiting for server handshake")
+ case <-prefaceReceived:
+ // We got the preface - huzzah! things are good.
+ case <-onCloseCalled:
+ // The transport has already closed - noop.
+ return nil, nil, errors.New("connection closed")
+ // TODO(deklerk) this should bail on ac.ctx.Done(). Add a test and fix.
}
- ac.mu.Unlock()
-
- // Now there is a viable transport to be use, so set ac.transport to reflect the new viable transport.
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- newTr.Close()
- return nil, errConnClosing
- }
- ac.mu.Unlock()
-
- return newTr, nil
+ return newTr, reconnect, nil
}
-func (ac *addrConn) startHealthCheck(ctx context.Context, newTr transport.ClientTransport, addr resolver.Address, serviceName string) {
- // Set up the health check helper functions
- newStream := func() (interface{}, error) {
- return ac.newClientStream(ctx, &StreamDesc{ServerStreams: true}, "/grpc.health.v1.Health/Watch", newTr)
+// startHealthCheck starts the health checking stream (RPC) to watch the health
+// stats of this connection if health checking is requested and configured.
+//
+// LB channel health checking is enabled when all requirements below are met:
+// 1. it is not disabled by the user with the WithDisableHealthCheck DialOption
+// 2. internal.HealthCheckFunc is set by importing the grpc/healthcheck package
+// 3. a service config with non-empty healthCheckConfig field is provided
+// 4. the load balancer requests it
+//
+// It sets addrConn to READY if the health checking stream is not started.
+//
+// Caller must hold ac.mu.
+func (ac *addrConn) startHealthCheck(ctx context.Context) {
+ var healthcheckManagingState bool
+ defer func() {
+ if !healthcheckManagingState {
+ ac.updateConnectivityState(connectivity.Ready)
+ }
+ }()
+
+ if ac.cc.dopts.disableHealthCheck {
+ return
}
- firstReady := true
- reportHealth := func(ok bool) {
+ healthCheckConfig := ac.cc.healthCheckConfig()
+ if healthCheckConfig == nil {
+ return
+ }
+ if !ac.scopts.HealthCheckEnabled {
+ return
+ }
+ healthCheckFunc := ac.cc.dopts.healthCheckFunc
+ if healthCheckFunc == nil {
+ // The health package is not imported to set health check function.
+ //
+ // TODO: add a link to the health check doc in the error message.
+ grpclog.Error("Health check is requested but health check function is not set.")
+ return
+ }
+
+ healthcheckManagingState = true
+
+ // Set up the health check helper functions.
+ currentTr := ac.transport
+ newStream := func(method string) (interface{}, error) {
+ ac.mu.Lock()
+ if ac.transport != currentTr {
+ ac.mu.Unlock()
+ return nil, status.Error(codes.Canceled, "the provided transport is no longer valid to use")
+ }
+ ac.mu.Unlock()
+ return newNonRetryClientStream(ctx, &StreamDesc{ServerStreams: true}, method, currentTr, ac)
+ }
+ setConnectivityState := func(s connectivity.State) {
ac.mu.Lock()
defer ac.mu.Unlock()
- if ac.transport != newTr {
+ if ac.transport != currentTr {
return
}
- if ok {
- if firstReady {
- firstReady = false
- ac.curAddr = addr
- }
- ac.updateConnectivityState(connectivity.Ready)
- } else {
- ac.updateConnectivityState(connectivity.TransientFailure)
- }
+ ac.updateConnectivityState(s)
}
- err := ac.cc.dopts.healthCheckFunc(ctx, newStream, reportHealth, serviceName)
- if err != nil {
- if status.Code(err) == codes.Unimplemented {
- if channelz.IsOn() {
- channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
- Desc: "Subchannel health check is unimplemented at server side, thus health check is disabled",
- Severity: channelz.CtError,
- })
+ // Start the health checking stream.
+ go func() {
+ err := ac.cc.dopts.healthCheckFunc(ctx, newStream, setConnectivityState, healthCheckConfig.ServiceName)
+ if err != nil {
+ if status.Code(err) == codes.Unimplemented {
+ if channelz.IsOn() {
+ channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
+ Desc: "Subchannel health check is unimplemented at server side, thus health check is disabled",
+ Severity: channelz.CtError,
+ })
+ }
+ grpclog.Error("Subchannel health check is unimplemented at server side, thus health check is disabled")
+ } else {
+ grpclog.Errorf("HealthCheckFunc exits with unexpected error %v", err)
}
- grpclog.Error("Subchannel health check is unimplemented at server side, thus health check is disabled")
- } else {
- grpclog.Errorf("HealthCheckFunc exits with unexpected error %v", err)
}
- }
+ }()
}
func (ac *addrConn) resetConnectBackoff() {
@@ -1332,7 +1334,6 @@
// between setting the state and logic that waits on context cancelation / etc.
ac.updateConnectivityState(connectivity.Shutdown)
ac.cancel()
- ac.tearDownErr = err
ac.curAddr = resolver.Address{}
if err == errConnDrain && curTr != nil {
// GracefulClose(...) may be executed multiple times when
diff --git a/vendor/google.golang.org/grpc/codes/codes.go b/vendor/google.golang.org/grpc/codes/codes.go
index d9b9d57..0273883 100644
--- a/vendor/google.golang.org/grpc/codes/codes.go
+++ b/vendor/google.golang.org/grpc/codes/codes.go
@@ -132,7 +132,8 @@
// Unavailable indicates the service is currently unavailable.
// This is a most likely a transient condition and may be corrected
- // by retrying with a backoff.
+ // by retrying with a backoff. Note that it is not always safe to retry
+ // non-idempotent operations.
//
// See litmus test above for deciding between FailedPrecondition,
// Aborted, and Unavailable.
diff --git a/vendor/google.golang.org/grpc/credentials/credentials.go b/vendor/google.golang.org/grpc/credentials/credentials.go
index a851560..8ea3d4a 100644
--- a/vendor/google.golang.org/grpc/credentials/credentials.go
+++ b/vendor/google.golang.org/grpc/credentials/credentials.go
@@ -36,9 +36,6 @@
"google.golang.org/grpc/credentials/internal"
)
-// alpnProtoStr are the specified application level protocols for gRPC.
-var alpnProtoStr = []string{"h2"}
-
// PerRPCCredentials defines the common interface for the credentials which need to
// attach security information to every RPC (e.g., oauth2).
type PerRPCCredentials interface {
@@ -208,10 +205,23 @@
return nil
}
+const alpnProtoStrH2 = "h2"
+
+func appendH2ToNextProtos(ps []string) []string {
+ for _, p := range ps {
+ if p == alpnProtoStrH2 {
+ return ps
+ }
+ }
+ ret := make([]string, 0, len(ps)+1)
+ ret = append(ret, ps...)
+ return append(ret, alpnProtoStrH2)
+}
+
// NewTLS uses c to construct a TransportCredentials based on TLS.
func NewTLS(c *tls.Config) TransportCredentials {
tc := &tlsCreds{cloneTLSConfig(c)}
- tc.config.NextProtos = alpnProtoStr
+ tc.config.NextProtos = appendH2ToNextProtos(tc.config.NextProtos)
return tc
}
@@ -268,24 +278,22 @@
// TLSChannelzSecurityValue defines the struct that TLS protocol should return
// from GetSecurityValue(), containing security info like cipher and certificate used.
type TLSChannelzSecurityValue struct {
+ ChannelzSecurityValue
StandardName string
LocalCertificate []byte
RemoteCertificate []byte
}
-func (*TLSChannelzSecurityValue) isChannelzSecurityValue() {}
-
// OtherChannelzSecurityValue defines the struct that non-TLS protocol should return
// from GetSecurityValue(), which contains protocol specific security info. Note
// the Value field will be sent to users of channelz requesting channel info, and
// thus sensitive info should better be avoided.
type OtherChannelzSecurityValue struct {
+ ChannelzSecurityValue
Name string
Value proto.Message
}
-func (*OtherChannelzSecurityValue) isChannelzSecurityValue() {}
-
var cipherSuiteLookup = map[uint16]string{
tls.TLS_RSA_WITH_RC4_128_SHA: "TLS_RSA_WITH_RC4_128_SHA",
tls.TLS_RSA_WITH_3DES_EDE_CBC_SHA: "TLS_RSA_WITH_3DES_EDE_CBC_SHA",
diff --git a/vendor/google.golang.org/grpc/dialoptions.go b/vendor/google.golang.org/grpc/dialoptions.go
index 537b258..e8f34d0 100644
--- a/vendor/google.golang.org/grpc/dialoptions.go
+++ b/vendor/google.golang.org/grpc/dialoptions.go
@@ -39,8 +39,12 @@
// dialOptions configure a Dial call. dialOptions are set by the DialOption
// values passed to Dial.
type dialOptions struct {
- unaryInt UnaryClientInterceptor
- streamInt StreamClientInterceptor
+ unaryInt UnaryClientInterceptor
+ streamInt StreamClientInterceptor
+
+ chainUnaryInts []UnaryClientInterceptor
+ chainStreamInts []StreamClientInterceptor
+
cp Compressor
dc Decompressor
bs backoff.Strategy
@@ -55,13 +59,15 @@
// balancer, and also by WithBalancerName dial option.
balancerBuilder balancer.Builder
// This is to support grpclb.
- resolverBuilder resolver.Builder
- reqHandshake envconfig.RequireHandshakeSetting
- channelzParentID int64
- disableServiceConfig bool
- disableRetry bool
- disableHealthCheck bool
- healthCheckFunc internal.HealthChecker
+ resolverBuilder resolver.Builder
+ channelzParentID int64
+ disableServiceConfig bool
+ disableRetry bool
+ disableHealthCheck bool
+ healthCheckFunc internal.HealthChecker
+ minConnectTimeout func() time.Duration
+ defaultServiceConfig *ServiceConfig // defaultServiceConfig is parsed from defaultServiceConfigRawJSON.
+ defaultServiceConfigRawJSON *string
}
// DialOption configures how we set up the connection.
@@ -93,17 +99,6 @@
}
}
-// WithWaitForHandshake blocks until the initial settings frame is received from
-// the server before assigning RPCs to the connection.
-//
-// Deprecated: this is the default behavior, and this option will be removed
-// after the 1.18 release.
-func WithWaitForHandshake() DialOption {
- return newFuncDialOption(func(o *dialOptions) {
- o.reqHandshake = envconfig.RequireHandshakeOn
- })
-}
-
// WithWriteBufferSize determines how much data can be batched before doing a
// write on the wire. The corresponding memory allocation for this buffer will
// be twice the size to keep syscalls low. The default value for this buffer is
@@ -149,7 +144,8 @@
// WithMaxMsgSize returns a DialOption which sets the maximum message size the
// client can receive.
//
-// Deprecated: use WithDefaultCallOptions(MaxCallRecvMsgSize(s)) instead.
+// Deprecated: use WithDefaultCallOptions(MaxCallRecvMsgSize(s)) instead. Will
+// be supported throughout 1.x.
func WithMaxMsgSize(s int) DialOption {
return WithDefaultCallOptions(MaxCallRecvMsgSize(s))
}
@@ -165,7 +161,8 @@
// WithCodec returns a DialOption which sets a codec for message marshaling and
// unmarshaling.
//
-// Deprecated: use WithDefaultCallOptions(ForceCodec(_)) instead.
+// Deprecated: use WithDefaultCallOptions(ForceCodec(_)) instead. Will be
+// supported throughout 1.x.
func WithCodec(c Codec) DialOption {
return WithDefaultCallOptions(CallCustomCodec(c))
}
@@ -174,7 +171,7 @@
// message compression. It has lower priority than the compressor set by the
// UseCompressor CallOption.
//
-// Deprecated: use UseCompressor instead.
+// Deprecated: use UseCompressor instead. Will be supported throughout 1.x.
func WithCompressor(cp Compressor) DialOption {
return newFuncDialOption(func(o *dialOptions) {
o.cp = cp
@@ -189,7 +186,8 @@
// message. If no compressor is registered for the encoding, an Unimplemented
// status error will be returned.
//
-// Deprecated: use encoding.RegisterCompressor instead.
+// Deprecated: use encoding.RegisterCompressor instead. Will be supported
+// throughout 1.x.
func WithDecompressor(dc Decompressor) DialOption {
return newFuncDialOption(func(o *dialOptions) {
o.dc = dc
@@ -200,7 +198,7 @@
// Name resolver will be ignored if this DialOption is specified.
//
// Deprecated: use the new balancer APIs in balancer package and
-// WithBalancerName.
+// WithBalancerName. Will be removed in a future 1.x release.
func WithBalancer(b Balancer) DialOption {
return newFuncDialOption(func(o *dialOptions) {
o.balancerBuilder = &balancerWrapperBuilder{
@@ -216,7 +214,8 @@
// The balancer cannot be overridden by balancer option specified by service
// config.
//
-// This is an EXPERIMENTAL API.
+// Deprecated: use WithDefaultServiceConfig and WithDisableServiceConfig
+// instead. Will be removed in a future 1.x release.
func WithBalancerName(balancerName string) DialOption {
builder := balancer.Get(balancerName)
if builder == nil {
@@ -237,9 +236,10 @@
// WithServiceConfig returns a DialOption which has a channel to read the
// service configuration.
//
-// Deprecated: service config should be received through name resolver, as
-// specified here.
-// https://github.com/grpc/grpc/blob/master/doc/service_config.md
+// Deprecated: service config should be received through name resolver or via
+// WithDefaultServiceConfig, as specified at
+// https://github.com/grpc/grpc/blob/master/doc/service_config.md. Will be
+// removed in a future 1.x release.
func WithServiceConfig(c <-chan ServiceConfig) DialOption {
return newFuncDialOption(func(o *dialOptions) {
o.scChan = c
@@ -322,7 +322,8 @@
// WithTimeout returns a DialOption that configures a timeout for dialing a
// ClientConn initially. This is valid if and only if WithBlock() is present.
//
-// Deprecated: use DialContext and context.WithTimeout instead.
+// Deprecated: use DialContext and context.WithTimeout instead. Will be
+// supported throughout 1.x.
func WithTimeout(d time.Duration) DialOption {
return newFuncDialOption(func(o *dialOptions) {
o.timeout = d
@@ -349,7 +350,8 @@
// is returned by f, gRPC checks the error's Temporary() method to decide if it
// should try to reconnect to the network address.
//
-// Deprecated: use WithContextDialer instead
+// Deprecated: use WithContextDialer instead. Will be supported throughout
+// 1.x.
func WithDialer(f func(string, time.Duration) (net.Conn, error)) DialOption {
return WithContextDialer(
func(ctx context.Context, addr string) (net.Conn, error) {
@@ -411,6 +413,17 @@
})
}
+// WithChainUnaryInterceptor returns a DialOption that specifies the chained
+// interceptor for unary RPCs. The first interceptor will be the outer most,
+// while the last interceptor will be the inner most wrapper around the real call.
+// All interceptors added by this method will be chained, and the interceptor
+// defined by WithUnaryInterceptor will always be prepended to the chain.
+func WithChainUnaryInterceptor(interceptors ...UnaryClientInterceptor) DialOption {
+ return newFuncDialOption(func(o *dialOptions) {
+ o.chainUnaryInts = append(o.chainUnaryInts, interceptors...)
+ })
+}
+
// WithStreamInterceptor returns a DialOption that specifies the interceptor for
// streaming RPCs.
func WithStreamInterceptor(f StreamClientInterceptor) DialOption {
@@ -419,6 +432,17 @@
})
}
+// WithChainStreamInterceptor returns a DialOption that specifies the chained
+// interceptor for unary RPCs. The first interceptor will be the outer most,
+// while the last interceptor will be the inner most wrapper around the real call.
+// All interceptors added by this method will be chained, and the interceptor
+// defined by WithStreamInterceptor will always be prepended to the chain.
+func WithChainStreamInterceptor(interceptors ...StreamClientInterceptor) DialOption {
+ return newFuncDialOption(func(o *dialOptions) {
+ o.chainStreamInts = append(o.chainStreamInts, interceptors...)
+ })
+}
+
// WithAuthority returns a DialOption that specifies the value to be used as the
// :authority pseudo-header. This value only works with WithInsecure and has no
// effect if TransportCredentials are present.
@@ -437,15 +461,32 @@
})
}
-// WithDisableServiceConfig returns a DialOption that causes grpc to ignore any
+// WithDisableServiceConfig returns a DialOption that causes gRPC to ignore any
// service config provided by the resolver and provides a hint to the resolver
// to not fetch service configs.
+//
+// Note that this dial option only disables service config from resolver. If
+// default service config is provided, gRPC will use the default service config.
func WithDisableServiceConfig() DialOption {
return newFuncDialOption(func(o *dialOptions) {
o.disableServiceConfig = true
})
}
+// WithDefaultServiceConfig returns a DialOption that configures the default
+// service config, which will be used in cases where:
+//
+// 1. WithDisableServiceConfig is also used.
+// 2. Resolver does not return a service config or if the resolver returns an
+// invalid service config.
+//
+// This API is EXPERIMENTAL.
+func WithDefaultServiceConfig(s string) DialOption {
+ return newFuncDialOption(func(o *dialOptions) {
+ o.defaultServiceConfigRawJSON = &s
+ })
+}
+
// WithDisableRetry returns a DialOption that disables retries, even if the
// service config enables them. This does not impact transparent retries, which
// will happen automatically if no data is written to the wire or if the RPC is
@@ -470,7 +511,8 @@
})
}
-// WithDisableHealthCheck disables the LB channel health checking for all SubConns of this ClientConn.
+// WithDisableHealthCheck disables the LB channel health checking for all
+// SubConns of this ClientConn.
//
// This API is EXPERIMENTAL.
func WithDisableHealthCheck() DialOption {
@@ -479,8 +521,8 @@
})
}
-// withHealthCheckFunc replaces the default health check function with the provided one. It makes
-// tests easier to change the health check function.
+// withHealthCheckFunc replaces the default health check function with the
+// provided one. It makes tests easier to change the health check function.
//
// For testing purpose only.
func withHealthCheckFunc(f internal.HealthChecker) DialOption {
@@ -492,7 +534,6 @@
func defaultDialOptions() dialOptions {
return dialOptions{
disableRetry: !envconfig.Retry,
- reqHandshake: envconfig.RequireHandshake,
healthCheckFunc: internal.HealthCheckFunc,
copts: transport.ConnectOptions{
WriteBufferSize: defaultWriteBufSize,
@@ -500,3 +541,14 @@
},
}
}
+
+// withGetMinConnectDeadline specifies the function that clientconn uses to
+// get minConnectDeadline. This can be used to make connection attempts happen
+// faster/slower.
+//
+// For testing purpose only.
+func withMinConnectDeadline(f func() time.Duration) DialOption {
+ return newFuncDialOption(func(o *dialOptions) {
+ o.minConnectTimeout = f
+ })
+}
diff --git a/vendor/google.golang.org/grpc/encoding/encoding.go b/vendor/google.golang.org/grpc/encoding/encoding.go
index ade8b7c..30a75da 100644
--- a/vendor/google.golang.org/grpc/encoding/encoding.go
+++ b/vendor/google.golang.org/grpc/encoding/encoding.go
@@ -102,10 +102,10 @@
if codec == nil {
panic("cannot register a nil Codec")
}
- contentSubtype := strings.ToLower(codec.Name())
- if contentSubtype == "" {
- panic("cannot register Codec with empty string result for String()")
+ if codec.Name() == "" {
+ panic("cannot register Codec with empty string result for Name()")
}
+ contentSubtype := strings.ToLower(codec.Name())
registeredCodecs[contentSubtype] = codec
}
diff --git a/vendor/google.golang.org/grpc/go.mod b/vendor/google.golang.org/grpc/go.mod
index 0418807..c7f3fa3 100644
--- a/vendor/google.golang.org/grpc/go.mod
+++ b/vendor/google.golang.org/grpc/go.mod
@@ -1,20 +1,21 @@
module google.golang.org/grpc
+go 1.11
+
require (
cloud.google.com/go v0.26.0 // indirect
github.com/BurntSushi/toml v0.3.1 // indirect
github.com/client9/misspell v0.3.4
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
github.com/golang/mock v1.1.1
- github.com/golang/protobuf v1.2.0
- golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3
- golang.org/x/net v0.0.0-20180826012351-8a410e7b638d
+ github.com/golang/protobuf v1.3.2
+ github.com/google/go-cmp v0.2.0
+ golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3
+ golang.org/x/net v0.0.0-20190311183353-d8887717615a
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be
- golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f // indirect
- golang.org/x/sys v0.0.0-20180830151530-49385e6e1522
- golang.org/x/text v0.3.0 // indirect
- golang.org/x/tools v0.0.0-20190114222345-bf090417da8b
+ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a
+ golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135
google.golang.org/appengine v1.1.0 // indirect
google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8
- honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099
+ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc
)
diff --git a/vendor/google.golang.org/grpc/go.sum b/vendor/google.golang.org/grpc/go.sum
index a79939d..7faff49 100644
--- a/vendor/google.golang.org/grpc/go.sum
+++ b/vendor/google.golang.org/grpc/go.sum
@@ -8,25 +8,30 @@
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/mock v1.1.1 h1:G5FRp8JnTd7RQH5kemVNlMeyXQAztQ3mOWV95KxsXH8=
github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
-github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
-github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3 h1:x/bBzNauLQAlE3fLku/xy92Y8QwKX5HZymrMz2IiKFc=
-golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
-golang.org/x/net v0.0.0-20180826012351-8a410e7b638d h1:g9qWBGx4puODJTMVyoPrpoxPFgVGd+z1DZwjfRu4d0I=
-golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/google/go-cmp v0.2.0 h1:+dTQ8DZQJz0Mb/HjFlkptS1FeQ4cWSnN941F8aEG4SQ=
+github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3 h1:XQyxROzUlZH+WIQwySDgnISgOivlhjIEwaQaJEJrrN0=
+golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a h1:oWX7TPOiFAMXLq8o0ikBYfCJVlRHBcsciT5bXOrH628=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be h1:vEDujvNQGv4jgYKudGeI/+DAX4Jffq6hpD55MmoEvKs=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
-golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA=
-golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sys v0.0.0-20180830151530-49385e6e1522 h1:Ve1ORMCxvRmSXBwJK+t3Oy+V2vRW2OetUQBq4rJIkZE=
-golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv3hCI0z56oJR5vAMgBU=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/tools v0.0.0-20190114222345-bf090417da8b h1:qMK98NmNCRVDIYFycQ5yVRkvgDUFfdP8Ip4KqmDEB7g=
-golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190311212946-11955173bddd h1:/e+gpKk9r3dJobndpTytxS2gOy6m5uvpg+ISQoEcusQ=
+golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135 h1:5Beo0mZN8dRzgrMMkDp0jc8YXQKx9DiJ2k1dkvGsn5A=
+golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
google.golang.org/appengine v1.1.0 h1:igQkv0AAhEIvTEpD5LIpAfav2eeVO9HBTjvKHVJPRSs=
google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8 h1:Nw54tB0rB7hY/N0NQvRW8DG4Yk3Q6T9cu9RcFQDu1tc=
google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
-honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099 h1:XJP7lxbSxWLOMNdBE4B/STaqVy6L73o0knwj2vIlxnw=
-honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc h1:/hemPrYIhOhy8zYrNj+069zDB68us2sMGsfkFJO0iZs=
+honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
diff --git a/vendor/google.golang.org/grpc/grpclog/grpclog.go b/vendor/google.golang.org/grpc/grpclog/grpclog.go
index 1fabb11..51bb945 100644
--- a/vendor/google.golang.org/grpc/grpclog/grpclog.go
+++ b/vendor/google.golang.org/grpc/grpclog/grpclog.go
@@ -18,7 +18,7 @@
// Package grpclog defines logging for grpc.
//
-// All logs in transport package only go to verbose level 2.
+// All logs in transport and grpclb packages only go to verbose level 2.
// All logs in other packages in grpc are logged in spite of the verbosity level.
//
// In the default logger,
diff --git a/vendor/google.golang.org/grpc/internal/balancerload/load.go b/vendor/google.golang.org/grpc/internal/balancerload/load.go
new file mode 100644
index 0000000..3a905d9
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/balancerload/load.go
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019 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.
+ */
+
+// Package balancerload defines APIs to parse server loads in trailers. The
+// parsed loads are sent to balancers in DoneInfo.
+package balancerload
+
+import (
+ "google.golang.org/grpc/metadata"
+)
+
+// Parser converts loads from metadata into a concrete type.
+type Parser interface {
+ // Parse parses loads from metadata.
+ Parse(md metadata.MD) interface{}
+}
+
+var parser Parser
+
+// SetParser sets the load parser.
+//
+// Not mutex-protected, should be called before any gRPC functions.
+func SetParser(lr Parser) {
+ parser = lr
+}
+
+// Parse calls parser.Read().
+func Parse(md metadata.MD) interface{} {
+ if parser == nil {
+ return nil
+ }
+ return parser.Parse(md)
+}
diff --git a/vendor/google.golang.org/grpc/internal/channelz/funcs.go b/vendor/google.golang.org/grpc/internal/channelz/funcs.go
index 041520d..f0744f9 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/funcs.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/funcs.go
@@ -24,6 +24,7 @@
package channelz
import (
+ "fmt"
"sort"
"sync"
"sync/atomic"
@@ -95,9 +96,14 @@
// NewChannelzStorage initializes channelz data storage and id generator.
//
+// This function returns a cleanup function to wait for all channelz state to be reset by the
+// grpc goroutines when those entities get closed. By using this cleanup function, we make sure tests
+// don't mess up each other, i.e. lingering goroutine from previous test doing entity removal happen
+// to remove some entity just register by the new test, since the id space is the same.
+//
// Note: This function is exported for testing purpose only. User should not call
// it in most cases.
-func NewChannelzStorage() {
+func NewChannelzStorage() (cleanup func() error) {
db.set(&channelMap{
topLevelChannels: make(map[int64]struct{}),
channels: make(map[int64]*channel),
@@ -107,6 +113,28 @@
subChannels: make(map[int64]*subChannel),
})
idGen.reset()
+ return func() error {
+ var err error
+ cm := db.get()
+ if cm == nil {
+ return nil
+ }
+ for i := 0; i < 1000; i++ {
+ cm.mu.Lock()
+ if len(cm.topLevelChannels) == 0 && len(cm.servers) == 0 && len(cm.channels) == 0 && len(cm.subChannels) == 0 && len(cm.listenSockets) == 0 && len(cm.normalSockets) == 0 {
+ cm.mu.Unlock()
+ // all things stored in the channelz map have been cleared.
+ return nil
+ }
+ cm.mu.Unlock()
+ time.Sleep(10 * time.Millisecond)
+ }
+
+ cm.mu.Lock()
+ err = fmt.Errorf("after 10s the channelz map has not been cleaned up yet, topchannels: %d, servers: %d, channels: %d, subchannels: %d, listen sockets: %d, normal sockets: %d", len(cm.topLevelChannels), len(cm.servers), len(cm.channels), len(cm.subChannels), len(cm.listenSockets), len(cm.normalSockets))
+ cm.mu.Unlock()
+ return err
+ }
}
// GetTopChannels returns a slice of top channel's ChannelMetric, along with a
diff --git a/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go b/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
index 62ed0f2..3ee8740 100644
--- a/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
+++ b/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
@@ -25,47 +25,11 @@
)
const (
- prefix = "GRPC_GO_"
- retryStr = prefix + "RETRY"
- requireHandshakeStr = prefix + "REQUIRE_HANDSHAKE"
-)
-
-// RequireHandshakeSetting describes the settings for handshaking.
-type RequireHandshakeSetting int
-
-const (
- // RequireHandshakeHybrid (default, deprecated) indicates to not wait for
- // handshake before considering a connection ready, but wait before
- // considering successful.
- RequireHandshakeHybrid RequireHandshakeSetting = iota
- // RequireHandshakeOn (default after the 1.17 release) indicates to wait
- // for handshake before considering a connection ready/successful.
- RequireHandshakeOn
- // RequireHandshakeOff indicates to not wait for handshake before
- // considering a connection ready/successful.
- RequireHandshakeOff
+ prefix = "GRPC_GO_"
+ retryStr = prefix + "RETRY"
)
var (
// Retry is set if retry is explicitly enabled via "GRPC_GO_RETRY=on".
Retry = strings.EqualFold(os.Getenv(retryStr), "on")
- // RequireHandshake is set based upon the GRPC_GO_REQUIRE_HANDSHAKE
- // environment variable.
- //
- // Will be removed after the 1.18 release.
- RequireHandshake RequireHandshakeSetting
)
-
-func init() {
- switch strings.ToLower(os.Getenv(requireHandshakeStr)) {
- case "on":
- fallthrough
- default:
- RequireHandshake = RequireHandshakeOn
- case "off":
- RequireHandshake = RequireHandshakeOff
- case "hybrid":
- // Will be removed after the 1.17 release.
- RequireHandshake = RequireHandshakeHybrid
- }
-}
diff --git a/vendor/google.golang.org/grpc/internal/internal.go b/vendor/google.golang.org/grpc/internal/internal.go
index c1d2c69..bc1f99a 100644
--- a/vendor/google.golang.org/grpc/internal/internal.go
+++ b/vendor/google.golang.org/grpc/internal/internal.go
@@ -23,6 +23,8 @@
import (
"context"
"time"
+
+ "google.golang.org/grpc/connectivity"
)
var (
@@ -37,10 +39,25 @@
// KeepaliveMinPingTime is the minimum ping interval. This must be 10s by
// default, but tests may wish to set it lower for convenience.
KeepaliveMinPingTime = 10 * time.Second
+ // ParseServiceConfig is a function to parse JSON service configs into
+ // opaque data structures.
+ ParseServiceConfig func(sc string) (interface{}, error)
+ // StatusRawProto is exported by status/status.go. This func returns a
+ // pointer to the wrapped Status proto for a given status.Status without a
+ // call to proto.Clone(). The returned Status proto should not be mutated by
+ // the caller.
+ StatusRawProto interface{} // func (*status.Status) *spb.Status
)
// HealthChecker defines the signature of the client-side LB channel health checking function.
-type HealthChecker func(ctx context.Context, newStream func() (interface{}, error), reportHealth func(bool), serviceName string) error
+//
+// The implementation is expected to create a health checking RPC stream by
+// calling newStream(), watch for the health status of serviceName, and report
+// it's health back by calling setConnectivityState().
+//
+// The health checking protocol is defined at:
+// https://github.com/grpc/grpc/blob/master/doc/health-checking.md
+type HealthChecker func(ctx context.Context, newStream func(string) (interface{}, error), setConnectivityState func(connectivity.State), serviceName string) error
const (
// CredsBundleModeFallback switches GoogleDefaultCreds to fallback mode.
diff --git a/vendor/google.golang.org/grpc/internal/syscall/syscall_nonlinux.go b/vendor/google.golang.org/grpc/internal/syscall/syscall_nonlinux.go
index 61678fe..d3fd9da 100644
--- a/vendor/google.golang.org/grpc/internal/syscall/syscall_nonlinux.go
+++ b/vendor/google.golang.org/grpc/internal/syscall/syscall_nonlinux.go
@@ -22,18 +22,24 @@
import (
"net"
+ "sync"
"time"
"google.golang.org/grpc/grpclog"
)
-func init() {
- grpclog.Info("CPU time info is unavailable on non-linux or appengine environment.")
+var once sync.Once
+
+func log() {
+ once.Do(func() {
+ grpclog.Info("CPU time info is unavailable on non-linux or appengine environment.")
+ })
}
// GetCPUTime returns the how much CPU time has passed since the start of this process.
// It always returns 0 under non-linux or appengine environment.
func GetCPUTime() int64 {
+ log()
return 0
}
@@ -42,22 +48,26 @@
// GetRusage is a no-op function under non-linux or appengine environment.
func GetRusage() (rusage *Rusage) {
+ log()
return nil
}
// CPUTimeDiff returns the differences of user CPU time and system CPU time used
// between two Rusage structs. It a no-op function for non-linux or appengine environment.
func CPUTimeDiff(first *Rusage, latest *Rusage) (float64, float64) {
+ log()
return 0, 0
}
// SetTCPUserTimeout is a no-op function under non-linux or appengine environments
func SetTCPUserTimeout(conn net.Conn, timeout time.Duration) error {
+ log()
return nil
}
// GetTCPUserTimeout is a no-op function under non-linux or appengine environments
// a negative return value indicates the operation is not supported
func GetTCPUserTimeout(conn net.Conn) (int, error) {
+ log()
return -1, nil
}
diff --git a/vendor/google.golang.org/grpc/internal/transport/controlbuf.go b/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
index 204ba15..ddee20b 100644
--- a/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
+++ b/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
@@ -23,6 +23,7 @@
"fmt"
"runtime"
"sync"
+ "sync/atomic"
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
@@ -84,24 +85,40 @@
// the control buffer of transport. They represent different aspects of
// control tasks, e.g., flow control, settings, streaming resetting, etc.
+// maxQueuedTransportResponseFrames is the most queued "transport response"
+// frames we will buffer before preventing new reads from occurring on the
+// transport. These are control frames sent in response to client requests,
+// such as RST_STREAM due to bad headers or settings acks.
+const maxQueuedTransportResponseFrames = 50
+
+type cbItem interface {
+ isTransportResponseFrame() bool
+}
+
// registerStream is used to register an incoming stream with loopy writer.
type registerStream struct {
streamID uint32
wq *writeQuota
}
+func (*registerStream) isTransportResponseFrame() bool { return false }
+
// headerFrame is also used to register stream on the client-side.
type headerFrame struct {
streamID uint32
hf []hpack.HeaderField
- endStream bool // Valid on server side.
- initStream func(uint32) (bool, error) // Used only on the client side.
+ endStream bool // Valid on server side.
+ initStream func(uint32) error // Used only on the client side.
onWrite func()
wq *writeQuota // write quota for the stream created.
cleanup *cleanupStream // Valid on the server side.
onOrphaned func(error) // Valid on client-side
}
+func (h *headerFrame) isTransportResponseFrame() bool {
+ return h.cleanup != nil && h.cleanup.rst // Results in a RST_STREAM
+}
+
type cleanupStream struct {
streamID uint32
rst bool
@@ -109,6 +126,8 @@
onWrite func()
}
+func (c *cleanupStream) isTransportResponseFrame() bool { return c.rst } // Results in a RST_STREAM
+
type dataFrame struct {
streamID uint32
endStream bool
@@ -119,27 +138,41 @@
onEachWrite func()
}
+func (*dataFrame) isTransportResponseFrame() bool { return false }
+
type incomingWindowUpdate struct {
streamID uint32
increment uint32
}
+func (*incomingWindowUpdate) isTransportResponseFrame() bool { return false }
+
type outgoingWindowUpdate struct {
streamID uint32
increment uint32
}
+func (*outgoingWindowUpdate) isTransportResponseFrame() bool {
+ return false // window updates are throttled by thresholds
+}
+
type incomingSettings struct {
ss []http2.Setting
}
+func (*incomingSettings) isTransportResponseFrame() bool { return true } // Results in a settings ACK
+
type outgoingSettings struct {
ss []http2.Setting
}
+func (*outgoingSettings) isTransportResponseFrame() bool { return false }
+
type incomingGoAway struct {
}
+func (*incomingGoAway) isTransportResponseFrame() bool { return false }
+
type goAway struct {
code http2.ErrCode
debugData []byte
@@ -147,15 +180,21 @@
closeConn bool
}
+func (*goAway) isTransportResponseFrame() bool { return false }
+
type ping struct {
ack bool
data [8]byte
}
+func (*ping) isTransportResponseFrame() bool { return true }
+
type outFlowControlSizeRequest struct {
resp chan uint32
}
+func (*outFlowControlSizeRequest) isTransportResponseFrame() bool { return false }
+
type outStreamState int
const (
@@ -238,6 +277,14 @@
consumerWaiting bool
list *itemList
err error
+
+ // transportResponseFrames counts the number of queued items that represent
+ // the response of an action initiated by the peer. trfChan is created
+ // when transportResponseFrames >= maxQueuedTransportResponseFrames and is
+ // closed and nilled when transportResponseFrames drops below the
+ // threshold. Both fields are protected by mu.
+ transportResponseFrames int
+ trfChan atomic.Value // *chan struct{}
}
func newControlBuffer(done <-chan struct{}) *controlBuffer {
@@ -248,12 +295,24 @@
}
}
-func (c *controlBuffer) put(it interface{}) error {
+// throttle blocks if there are too many incomingSettings/cleanupStreams in the
+// controlbuf.
+func (c *controlBuffer) throttle() {
+ ch, _ := c.trfChan.Load().(*chan struct{})
+ if ch != nil {
+ select {
+ case <-*ch:
+ case <-c.done:
+ }
+ }
+}
+
+func (c *controlBuffer) put(it cbItem) error {
_, err := c.executeAndPut(nil, it)
return err
}
-func (c *controlBuffer) executeAndPut(f func(it interface{}) bool, it interface{}) (bool, error) {
+func (c *controlBuffer) executeAndPut(f func(it interface{}) bool, it cbItem) (bool, error) {
var wakeUp bool
c.mu.Lock()
if c.err != nil {
@@ -271,6 +330,15 @@
c.consumerWaiting = false
}
c.list.enqueue(it)
+ if it.isTransportResponseFrame() {
+ c.transportResponseFrames++
+ if c.transportResponseFrames == maxQueuedTransportResponseFrames {
+ // We are adding the frame that puts us over the threshold; create
+ // a throttling channel.
+ ch := make(chan struct{})
+ c.trfChan.Store(&ch)
+ }
+ }
c.mu.Unlock()
if wakeUp {
select {
@@ -304,7 +372,17 @@
return nil, c.err
}
if !c.list.isEmpty() {
- h := c.list.dequeue()
+ h := c.list.dequeue().(cbItem)
+ if h.isTransportResponseFrame() {
+ if c.transportResponseFrames == maxQueuedTransportResponseFrames {
+ // We are removing the frame that put us over the
+ // threshold; close and clear the throttling channel.
+ ch := c.trfChan.Load().(*chan struct{})
+ close(*ch)
+ c.trfChan.Store((*chan struct{})(nil))
+ }
+ c.transportResponseFrames--
+ }
c.mu.Unlock()
return h, nil
}
@@ -559,21 +637,17 @@
func (l *loopyWriter) originateStream(str *outStream) error {
hdr := str.itl.dequeue().(*headerFrame)
- sendPing, err := hdr.initStream(str.id)
- if err != nil {
+ if err := hdr.initStream(str.id); err != nil {
if err == ErrConnClosing {
return err
}
// Other errors(errStreamDrain) need not close transport.
return nil
}
- if err = l.writeHeader(str.id, hdr.endStream, hdr.hf, hdr.onWrite); err != nil {
+ if err := l.writeHeader(str.id, hdr.endStream, hdr.hf, hdr.onWrite); err != nil {
return err
}
l.estdStreams[str.id] = str
- if sendPing {
- return l.pingHandler(&ping{data: [8]byte{}})
- }
return nil
}
diff --git a/vendor/google.golang.org/grpc/internal/transport/flowcontrol.go b/vendor/google.golang.org/grpc/internal/transport/flowcontrol.go
index 5ea997a..f262edd 100644
--- a/vendor/google.golang.org/grpc/internal/transport/flowcontrol.go
+++ b/vendor/google.golang.org/grpc/internal/transport/flowcontrol.go
@@ -149,6 +149,7 @@
n = uint32(math.MaxInt32)
}
f.mu.Lock()
+ defer f.mu.Unlock()
// estSenderQuota is the receiver's view of the maximum number of bytes the sender
// can send without a window update.
estSenderQuota := int32(f.limit - (f.pendingData + f.pendingUpdate))
@@ -169,10 +170,8 @@
// is padded; We will fallback on the current available window(at least a 1/4th of the limit).
f.delta = n
}
- f.mu.Unlock()
return f.delta
}
- f.mu.Unlock()
return 0
}
diff --git a/vendor/google.golang.org/grpc/internal/transport/handler_server.go b/vendor/google.golang.org/grpc/internal/transport/handler_server.go
index 73b41ea..78f9ddc 100644
--- a/vendor/google.golang.org/grpc/internal/transport/handler_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/handler_server.go
@@ -24,6 +24,7 @@
package transport
import (
+ "bytes"
"context"
"errors"
"fmt"
@@ -63,9 +64,6 @@
if _, ok := w.(http.Flusher); !ok {
return nil, errors.New("gRPC requires a ResponseWriter supporting http.Flusher")
}
- if _, ok := w.(http.CloseNotifier); !ok {
- return nil, errors.New("gRPC requires a ResponseWriter supporting http.CloseNotifier")
- }
st := &serverHandlerTransport{
rw: w,
@@ -176,17 +174,11 @@
// do runs fn in the ServeHTTP goroutine.
func (ht *serverHandlerTransport) do(fn func()) error {
- // Avoid a panic writing to closed channel. Imperfect but maybe good enough.
select {
case <-ht.closedCh:
return ErrConnClosing
- default:
- select {
- case ht.writes <- fn:
- return nil
- case <-ht.closedCh:
- return ErrConnClosing
- }
+ case ht.writes <- fn:
+ return nil
}
}
@@ -237,7 +229,6 @@
if ht.stats != nil {
ht.stats.HandleRPC(s.Context(), &stats.OutTrailer{})
}
- close(ht.writes)
}
ht.Close()
return err
@@ -315,19 +306,13 @@
ctx, cancel = context.WithCancel(ctx)
}
- // requestOver is closed when either the request's context is done
- // or the status has been written via WriteStatus.
+ // requestOver is closed when the status has been written via WriteStatus.
requestOver := make(chan struct{})
-
- // clientGone receives a single value if peer is gone, either
- // because the underlying connection is dead or because the
- // peer sends an http2 RST_STREAM.
- clientGone := ht.rw.(http.CloseNotifier).CloseNotify()
go func() {
select {
case <-requestOver:
case <-ht.closedCh:
- case <-clientGone:
+ case <-ht.req.Context().Done():
}
cancel()
ht.Close()
@@ -363,7 +348,7 @@
ht.stats.HandleRPC(s.ctx, inHeader)
}
s.trReader = &transportReader{
- reader: &recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: s.buf},
+ reader: &recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: s.buf, freeBuffer: func(*bytes.Buffer) {}},
windowHandler: func(int) {},
}
@@ -377,7 +362,7 @@
for buf := make([]byte, readSize); ; {
n, err := req.Body.Read(buf)
if n > 0 {
- s.buf.put(recvMsg{data: buf[:n:n]})
+ s.buf.put(recvMsg{buffer: bytes.NewBuffer(buf[:n:n])})
buf = buf[n:]
}
if err != nil {
@@ -407,10 +392,7 @@
func (ht *serverHandlerTransport) runStream() {
for {
select {
- case fn, ok := <-ht.writes:
- if !ok {
- return
- }
+ case fn := <-ht.writes:
fn()
case <-ht.closedCh:
return
diff --git a/vendor/google.golang.org/grpc/internal/transport/http2_client.go b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
index ff8f4db..9bd8c27 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_client.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
@@ -62,8 +62,6 @@
// goAway is closed to notify the upper layer (i.e., addrConn.transportMonitor)
// that the server sent GoAway on this transport.
goAway chan struct{}
- // awakenKeepalive is used to wake up keepalive when after it has gone dormant.
- awakenKeepalive chan struct{}
framer *framer
// controlBuf delivers all the control related tasks (e.g., window
@@ -110,6 +108,16 @@
// goAwayReason records the http2.ErrCode and debug data received with the
// GoAway frame.
goAwayReason GoAwayReason
+ // A condition variable used to signal when the keepalive goroutine should
+ // go dormant. The condition for dormancy is based on the number of active
+ // streams and the `PermitWithoutStream` keepalive client parameter. And
+ // since the number of active streams is guarded by the above mutex, we use
+ // the same for this condition variable as well.
+ kpDormancyCond *sync.Cond
+ // A boolean to track whether the keepalive goroutine is dormant or not.
+ // This is checked before attempting to signal the above condition
+ // variable.
+ kpDormant bool
// Fields below are for channelz metric collection.
channelzID int64 // channelz unique identification number
@@ -117,6 +125,8 @@
onGoAway func(GoAwayReason)
onClose func()
+
+ bufferPool *bufferPool
}
func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error), addr string) (net.Conn, error) {
@@ -230,7 +240,6 @@
readerDone: make(chan struct{}),
writerDone: make(chan struct{}),
goAway: make(chan struct{}),
- awakenKeepalive: make(chan struct{}, 1),
framer: newFramer(conn, writeBufSize, readBufSize, maxHeaderListSize),
fc: &trInFlow{limit: uint32(icwz)},
scheme: scheme,
@@ -249,6 +258,7 @@
onGoAway: onGoAway,
onClose: onClose,
keepaliveEnabled: keepaliveEnabled,
+ bufferPool: newBufferPool(),
}
t.controlBuf = newControlBuffer(t.ctxDone)
if opts.InitialWindowSize >= defaultWindowSize {
@@ -261,9 +271,6 @@
updateFlowControl: t.updateFlowControl,
}
}
- // Make sure awakenKeepalive can't be written upon.
- // keepalive routine will make it writable, if need be.
- t.awakenKeepalive <- struct{}{}
if t.statsHandler != nil {
t.ctx = t.statsHandler.TagConn(t.ctx, &stats.ConnTagInfo{
RemoteAddr: t.remoteAddr,
@@ -278,6 +285,7 @@
t.channelzID = channelz.RegisterNormalSocket(t, opts.ChannelzParentID, fmt.Sprintf("%s -> %s", t.localAddr, t.remoteAddr))
}
if t.keepaliveEnabled {
+ t.kpDormancyCond = sync.NewCond(&t.mu)
go t.keepalive()
}
// Start the reader goroutine for incoming message. Each transport has
@@ -367,6 +375,7 @@
closeStream: func(err error) {
t.CloseStream(s, err)
},
+ freeBuffer: t.bufferPool.put,
},
windowHandler: func(n int) {
t.updateWindow(s, uint32(n))
@@ -437,6 +446,15 @@
if md, added, ok := metadata.FromOutgoingContextRaw(ctx); ok {
var k string
+ for k, vv := range md {
+ // HTTP doesn't allow you to set pseudoheaders after non pseudoheaders were set.
+ if isReservedHeader(k) {
+ continue
+ }
+ for _, v := range vv {
+ headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
+ }
+ }
for _, vv := range added {
for i, v := range vv {
if i%2 == 0 {
@@ -450,15 +468,6 @@
headerFields = append(headerFields, hpack.HeaderField{Name: strings.ToLower(k), Value: encodeMetadataHeader(k, v)})
}
}
- for k, vv := range md {
- // HTTP doesn't allow you to set pseudoheaders after non pseudoheaders were set.
- if isReservedHeader(k) {
- continue
- }
- for _, v := range vv {
- headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
- }
- }
}
if md, ok := t.md.(*metadata.MD); ok {
for k, vv := range *md {
@@ -489,6 +498,9 @@
}
func (t *http2Client) getTrAuthData(ctx context.Context, audience string) (map[string]string, error) {
+ if len(t.perRPCCreds) == 0 {
+ return nil, nil
+ }
authData := map[string]string{}
for _, c := range t.perRPCCreds {
data, err := c.GetRequestMetadata(ctx, audience)
@@ -509,7 +521,7 @@
}
func (t *http2Client) getCallAuthData(ctx context.Context, audience string, callHdr *CallHdr) (map[string]string, error) {
- callAuthData := map[string]string{}
+ var callAuthData map[string]string
// Check if credentials.PerRPCCredentials were provided via call options.
// Note: if these credentials are provided both via dial options and call
// options, then both sets of credentials will be applied.
@@ -521,6 +533,7 @@
if err != nil {
return nil, status.Errorf(codes.Internal, "transport: %v", err)
}
+ callAuthData = make(map[string]string, len(data))
for k, v := range data {
// Capital header names are illegal in HTTP/2
k = strings.ToLower(k)
@@ -549,15 +562,14 @@
s.write(recvMsg{err: err})
close(s.done)
// If headerChan isn't closed, then close it.
- if atomic.SwapUint32(&s.headerDone, 1) == 0 {
+ if atomic.CompareAndSwapUint32(&s.headerChanClosed, 0, 1) {
close(s.headerChan)
}
-
}
hdr := &headerFrame{
hf: headerFields,
endStream: false,
- initStream: func(id uint32) (bool, error) {
+ initStream: func(id uint32) error {
t.mu.Lock()
if state := t.state; state != reachable {
t.mu.Unlock()
@@ -567,29 +579,19 @@
err = ErrConnClosing
}
cleanup(err)
- return false, err
+ return err
}
t.activeStreams[id] = s
if channelz.IsOn() {
atomic.AddInt64(&t.czData.streamsStarted, 1)
atomic.StoreInt64(&t.czData.lastStreamCreatedTime, time.Now().UnixNano())
}
- var sendPing bool
- // If the number of active streams change from 0 to 1, then check if keepalive
- // has gone dormant. If so, wake it up.
- if len(t.activeStreams) == 1 && t.keepaliveEnabled {
- select {
- case t.awakenKeepalive <- struct{}{}:
- sendPing = true
- // Fill the awakenKeepalive channel again as this channel must be
- // kept non-writable except at the point that the keepalive()
- // goroutine is waiting either to be awaken or shutdown.
- t.awakenKeepalive <- struct{}{}
- default:
- }
+ // If the keepalive goroutine has gone dormant, wake it up.
+ if t.kpDormant {
+ t.kpDormancyCond.Signal()
}
t.mu.Unlock()
- return sendPing, nil
+ return nil
},
onOrphaned: cleanup,
wq: s.wq,
@@ -713,7 +715,7 @@
s.write(recvMsg{err: err})
}
// If headerChan isn't closed, then close it.
- if atomic.SwapUint32(&s.headerDone, 1) == 0 {
+ if atomic.CompareAndSwapUint32(&s.headerChanClosed, 0, 1) {
s.noHeaders = true
close(s.headerChan)
}
@@ -765,9 +767,17 @@
t.mu.Unlock()
return nil
}
+ // Call t.onClose before setting the state to closing to prevent the client
+ // from attempting to create new streams ASAP.
+ t.onClose()
t.state = closing
streams := t.activeStreams
t.activeStreams = nil
+ if t.kpDormant {
+ // If the keepalive goroutine is blocked on this condition variable, we
+ // should unblock it so that the goroutine eventually exits.
+ t.kpDormancyCond.Signal()
+ }
t.mu.Unlock()
t.controlBuf.finish()
t.cancel()
@@ -785,7 +795,6 @@
}
t.statsHandler.HandleConn(t.ctx, connEnd)
}
- t.onClose()
return err
}
@@ -794,21 +803,21 @@
// stream is closed. If there are no active streams, the transport is closed
// immediately. This does nothing if the transport is already draining or
// closing.
-func (t *http2Client) GracefulClose() error {
+func (t *http2Client) GracefulClose() {
t.mu.Lock()
// Make sure we move to draining only from active.
if t.state == draining || t.state == closing {
t.mu.Unlock()
- return nil
+ return
}
t.state = draining
active := len(t.activeStreams)
t.mu.Unlock()
if active == 0 {
- return t.Close()
+ t.Close()
+ return
}
t.controlBuf.put(&incomingGoAway{})
- return nil
}
// Write formats the data into HTTP2 data frame(s) and sends it out. The caller
@@ -844,11 +853,11 @@
return t.controlBuf.put(df)
}
-func (t *http2Client) getStream(f http2.Frame) (*Stream, bool) {
+func (t *http2Client) getStream(f http2.Frame) *Stream {
t.mu.Lock()
- defer t.mu.Unlock()
- s, ok := t.activeStreams[f.Header().StreamID]
- return s, ok
+ s := t.activeStreams[f.Header().StreamID]
+ t.mu.Unlock()
+ return s
}
// adjustWindow sends out extra window update over the initial window size
@@ -928,8 +937,8 @@
t.controlBuf.put(bdpPing)
}
// Select the right stream to dispatch.
- s, ok := t.getStream(f)
- if !ok {
+ s := t.getStream(f)
+ if s == nil {
return
}
if size > 0 {
@@ -946,9 +955,10 @@
// guarantee f.Data() is consumed before the arrival of next frame.
// Can this copy be eliminated?
if len(f.Data()) > 0 {
- data := make([]byte, len(f.Data()))
- copy(data, f.Data())
- s.write(recvMsg{data: data})
+ buffer := t.bufferPool.get()
+ buffer.Reset()
+ buffer.Write(f.Data())
+ s.write(recvMsg{buffer: buffer})
}
}
// The server has closed the stream without sending trailers. Record that
@@ -959,8 +969,8 @@
}
func (t *http2Client) handleRSTStream(f *http2.RSTStreamFrame) {
- s, ok := t.getStream(f)
- if !ok {
+ s := t.getStream(f)
+ if s == nil {
return
}
if f.ErrCode == http2.ErrCodeRefusedStream {
@@ -973,9 +983,9 @@
statusCode = codes.Unknown
}
if statusCode == codes.Canceled {
- // Our deadline was already exceeded, and that was likely the cause of
- // this cancelation. Alter the status code accordingly.
- if d, ok := s.ctx.Deadline(); ok && d.After(time.Now()) {
+ if d, ok := s.ctx.Deadline(); ok && !d.After(time.Now()) {
+ // Our deadline was already exceeded, and that was likely the cause
+ // of this cancelation. Alter the status code accordingly.
statusCode = codes.DeadlineExceeded
}
}
@@ -1080,11 +1090,12 @@
default:
t.setGoAwayReason(f)
close(t.goAway)
- t.state = draining
t.controlBuf.put(&incomingGoAway{})
-
- // This has to be a new goroutine because we're still using the current goroutine to read in the transport.
+ // Notify the clientconn about the GOAWAY before we set the state to
+ // draining, to allow the client to stop attempting to create streams
+ // before disallowing new streams on this connection.
t.onGoAway(t.goAwayReason)
+ t.state = draining
}
// All streams with IDs greater than the GoAwayId
// and smaller than the previous GoAway ID should be killed.
@@ -1136,20 +1147,30 @@
// operateHeaders takes action on the decoded headers.
func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
- s, ok := t.getStream(frame)
- if !ok {
+ s := t.getStream(frame)
+ if s == nil {
return
}
+ endStream := frame.StreamEnded()
atomic.StoreUint32(&s.bytesReceived, 1)
- var state decodeState
- if err := state.decodeHeader(frame); err != nil {
- t.closeStream(s, err, true, http2.ErrCodeProtocol, status.New(codes.Internal, err.Error()), nil, false)
- // Something wrong. Stops reading even when there is remaining.
+ initialHeader := atomic.LoadUint32(&s.headerChanClosed) == 0
+
+ if !initialHeader && !endStream {
+ // As specified by gRPC over HTTP2, a HEADERS frame (and associated CONTINUATION frames) can only appear at the start or end of a stream. Therefore, second HEADERS frame must have EOS bit set.
+ st := status.New(codes.Internal, "a HEADERS frame cannot appear in the middle of a stream")
+ t.closeStream(s, st.Err(), true, http2.ErrCodeProtocol, st, nil, false)
return
}
- endStream := frame.StreamEnded()
- var isHeader bool
+ state := &decodeState{}
+ // Initialize isGRPC value to be !initialHeader, since if a gRPC Response-Headers has already been received, then it means that the peer is speaking gRPC and we are in gRPC mode.
+ state.data.isGRPC = !initialHeader
+ if err := state.decodeHeader(frame); err != nil {
+ t.closeStream(s, err, true, http2.ErrCodeProtocol, status.Convert(err), nil, endStream)
+ return
+ }
+
+ isHeader := false
defer func() {
if t.statsHandler != nil {
if isHeader {
@@ -1167,29 +1188,33 @@
}
}
}()
- // If headers haven't been received yet.
- if atomic.SwapUint32(&s.headerDone, 1) == 0 {
+
+ // If headerChan hasn't been closed yet
+ if atomic.CompareAndSwapUint32(&s.headerChanClosed, 0, 1) {
if !endStream {
- // Headers frame is not actually a trailers-only frame.
+ // HEADERS frame block carries a Response-Headers.
isHeader = true
// These values can be set without any synchronization because
// stream goroutine will read it only after seeing a closed
// headerChan which we'll close after setting this.
- s.recvCompress = state.encoding
- if len(state.mdata) > 0 {
- s.header = state.mdata
+ s.recvCompress = state.data.encoding
+ if len(state.data.mdata) > 0 {
+ s.header = state.data.mdata
}
} else {
+ // HEADERS frame block carries a Trailers-Only.
s.noHeaders = true
}
close(s.headerChan)
}
+
if !endStream {
return
}
+
// if client received END_STREAM from server while stream was still active, send RST_STREAM
rst := s.getState() == streamActive
- t.closeStream(s, io.EOF, rst, http2.ErrCodeNo, state.status(), state.mdata, true)
+ t.closeStream(s, io.EOF, rst, http2.ErrCodeNo, state.status(), state.data.mdata, true)
}
// reader runs as a separate goroutine in charge of reading data from network
@@ -1220,6 +1245,7 @@
// loop to keep reading incoming messages on this transport.
for {
+ t.controlBuf.throttle()
frame, err := t.framer.fr.ReadFrame()
if t.keepaliveEnabled {
atomic.CompareAndSwapUint32(&t.activity, 0, 1)
@@ -1277,29 +1303,32 @@
timer.Reset(t.kp.Time)
continue
}
- // Check if keepalive should go dormant.
t.mu.Lock()
- if len(t.activeStreams) < 1 && !t.kp.PermitWithoutStream {
- // Make awakenKeepalive writable.
- <-t.awakenKeepalive
+ if t.state == closing {
+ // If the transport is closing, we should exit from the
+ // keepalive goroutine here. If not, we could have a race
+ // between the call to Signal() from Close() and the call to
+ // Wait() here, whereby the keepalive goroutine ends up
+ // blocking on the condition variable which will never be
+ // signalled again.
t.mu.Unlock()
- select {
- case <-t.awakenKeepalive:
- // If the control gets here a ping has been sent
- // need to reset the timer with keepalive.Timeout.
- case <-t.ctx.Done():
- return
- }
- } else {
- t.mu.Unlock()
- if channelz.IsOn() {
- atomic.AddInt64(&t.czData.kpCount, 1)
- }
- // Send ping.
- t.controlBuf.put(p)
+ return
}
+ if len(t.activeStreams) < 1 && !t.kp.PermitWithoutStream {
+ t.kpDormant = true
+ t.kpDormancyCond.Wait()
+ }
+ t.kpDormant = false
+ t.mu.Unlock()
- // By the time control gets here a ping has been sent one way or the other.
+ if channelz.IsOn() {
+ atomic.AddInt64(&t.czData.kpCount, 1)
+ }
+ // We get here either because we were dormant and a new stream was
+ // created which unblocked the Wait() call, or because the
+ // keepalive timer expired. In both cases, we need to send a ping.
+ t.controlBuf.put(p)
+
timer.Reset(t.kp.Timeout)
select {
case <-timer.C:
@@ -1307,6 +1336,7 @@
timer.Reset(t.kp.Time)
continue
}
+ infof("transport: closing client transport due to idleness.")
t.Close()
return
case <-t.ctx.Done():
@@ -1356,6 +1386,8 @@
return &s
}
+func (t *http2Client) RemoteAddr() net.Addr { return t.remoteAddr }
+
func (t *http2Client) IncrMsgSent() {
atomic.AddInt64(&t.czData.msgSent, 1)
atomic.StoreInt64(&t.czData.lastMsgSentTime, time.Now().UnixNano())
diff --git a/vendor/google.golang.org/grpc/internal/transport/http2_server.go b/vendor/google.golang.org/grpc/internal/transport/http2_server.go
index d038b2d..33686a1 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_server.go
@@ -35,9 +35,11 @@
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
+ spb "google.golang.org/genproto/googleapis/rpc/status"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/grpclog"
+ "google.golang.org/grpc/internal"
"google.golang.org/grpc/internal/channelz"
"google.golang.org/grpc/internal/grpcrand"
"google.golang.org/grpc/keepalive"
@@ -55,13 +57,15 @@
// ErrHeaderListSizeLimitViolation indicates that the header list size is larger
// than the limit set by peer.
ErrHeaderListSizeLimitViolation = errors.New("transport: trying to send header list size larger than the limit set by peer")
+ // statusRawProto is a function to get to the raw status proto wrapped in a
+ // status.Status without a proto.Clone().
+ statusRawProto = internal.StatusRawProto.(func(*status.Status) *spb.Status)
)
// http2Server implements the ServerTransport interface with HTTP2.
type http2Server struct {
ctx context.Context
- ctxDone <-chan struct{} // Cache the context.Done() chan
- cancel context.CancelFunc
+ done chan struct{}
conn net.Conn
loopy *loopyWriter
readerDone chan struct{} // sync point to enable testing.
@@ -119,6 +123,7 @@
// Fields below are for channelz metric collection.
channelzID int64 // channelz unique identification number
czData *channelzData
+ bufferPool *bufferPool
}
// newHTTP2Server constructs a ServerTransport based on HTTP2. ConnectionError is
@@ -132,7 +137,10 @@
}
framer := newFramer(conn, writeBufSize, readBufSize, maxHeaderListSize)
// Send initial settings as connection preface to client.
- var isettings []http2.Setting
+ isettings := []http2.Setting{{
+ ID: http2.SettingMaxFrameSize,
+ Val: http2MaxFrameLen,
+ }}
// TODO(zhaoq): Have a better way to signal "no limit" because 0 is
// permitted in the HTTP2 spec.
maxStreams := config.MaxStreams
@@ -197,11 +205,10 @@
if kep.MinTime == 0 {
kep.MinTime = defaultKeepalivePolicyMinTime
}
- ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan struct{})
t := &http2Server{
- ctx: ctx,
- cancel: cancel,
- ctxDone: ctx.Done(),
+ ctx: context.Background(),
+ done: done,
conn: conn,
remoteAddr: conn.RemoteAddr(),
localAddr: conn.LocalAddr(),
@@ -220,8 +227,9 @@
kep: kep,
initialWindowSize: iwz,
czData: new(channelzData),
+ bufferPool: newBufferPool(),
}
- t.controlBuf = newControlBuffer(t.ctxDone)
+ t.controlBuf = newControlBuffer(t.done)
if dynamicWindow {
t.bdpEst = &bdpEstimator{
bdp: initialWindowSize,
@@ -286,7 +294,9 @@
// operateHeader takes action on the decoded headers.
func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (fatal bool) {
streamID := frame.Header().StreamID
- state := decodeState{serverSide: true}
+ state := &decodeState{
+ serverSide: true,
+ }
if err := state.decodeHeader(frame); err != nil {
if se, ok := status.FromError(err); ok {
t.controlBuf.put(&cleanupStream{
@@ -305,16 +315,16 @@
st: t,
buf: buf,
fc: &inFlow{limit: uint32(t.initialWindowSize)},
- recvCompress: state.encoding,
- method: state.method,
- contentSubtype: state.contentSubtype,
+ recvCompress: state.data.encoding,
+ method: state.data.method,
+ contentSubtype: state.data.contentSubtype,
}
if frame.StreamEnded() {
// s is just created by the caller. No lock needed.
s.state = streamReadDone
}
- if state.timeoutSet {
- s.ctx, s.cancel = context.WithTimeout(t.ctx, state.timeout)
+ if state.data.timeoutSet {
+ s.ctx, s.cancel = context.WithTimeout(t.ctx, state.data.timeout)
} else {
s.ctx, s.cancel = context.WithCancel(t.ctx)
}
@@ -327,19 +337,19 @@
}
s.ctx = peer.NewContext(s.ctx, pr)
// Attach the received metadata to the context.
- if len(state.mdata) > 0 {
- s.ctx = metadata.NewIncomingContext(s.ctx, state.mdata)
+ if len(state.data.mdata) > 0 {
+ s.ctx = metadata.NewIncomingContext(s.ctx, state.data.mdata)
}
- if state.statsTags != nil {
- s.ctx = stats.SetIncomingTags(s.ctx, state.statsTags)
+ if state.data.statsTags != nil {
+ s.ctx = stats.SetIncomingTags(s.ctx, state.data.statsTags)
}
- if state.statsTrace != nil {
- s.ctx = stats.SetIncomingTrace(s.ctx, state.statsTrace)
+ if state.data.statsTrace != nil {
+ s.ctx = stats.SetIncomingTrace(s.ctx, state.data.statsTrace)
}
if t.inTapHandle != nil {
var err error
info := &tap.Info{
- FullMethodName: state.method,
+ FullMethodName: state.data.method,
}
s.ctx, err = t.inTapHandle(s.ctx, info)
if err != nil {
@@ -350,12 +360,14 @@
rstCode: http2.ErrCodeRefusedStream,
onWrite: func() {},
})
+ s.cancel()
return false
}
}
t.mu.Lock()
if t.state != reachable {
t.mu.Unlock()
+ s.cancel()
return false
}
if uint32(len(t.activeStreams)) >= t.maxStreams {
@@ -366,12 +378,14 @@
rstCode: http2.ErrCodeRefusedStream,
onWrite: func() {},
})
+ s.cancel()
return false
}
if streamID%2 != 1 || streamID <= t.maxStreamID {
t.mu.Unlock()
// illegal gRPC stream id.
errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", streamID)
+ s.cancel()
return true
}
t.maxStreamID = streamID
@@ -403,9 +417,10 @@
s.wq = newWriteQuota(defaultWriteQuota, s.ctxDone)
s.trReader = &transportReader{
reader: &recvBufferReader{
- ctx: s.ctx,
- ctxDone: s.ctxDone,
- recv: s.buf,
+ ctx: s.ctx,
+ ctxDone: s.ctxDone,
+ recv: s.buf,
+ freeBuffer: t.bufferPool.put,
},
windowHandler: func(n int) {
t.updateWindow(s, uint32(n))
@@ -426,6 +441,7 @@
func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.Context, string) context.Context) {
defer close(t.readerDone)
for {
+ t.controlBuf.throttle()
frame, err := t.framer.fr.ReadFrame()
atomic.StoreUint32(&t.activity, 1)
if err != nil {
@@ -435,7 +451,7 @@
s := t.activeStreams[se.StreamID]
t.mu.Unlock()
if s != nil {
- t.closeStream(s, true, se.Code, nil, false)
+ t.closeStream(s, true, se.Code, false)
} else {
t.controlBuf.put(&cleanupStream{
streamID: se.StreamID,
@@ -577,7 +593,7 @@
}
if size > 0 {
if err := s.fc.onData(size); err != nil {
- t.closeStream(s, true, http2.ErrCodeFlowControl, nil, false)
+ t.closeStream(s, true, http2.ErrCodeFlowControl, false)
return
}
if f.Header().Flags.Has(http2.FlagDataPadded) {
@@ -589,9 +605,10 @@
// guarantee f.Data() is consumed before the arrival of next frame.
// Can this copy be eliminated?
if len(f.Data()) > 0 {
- data := make([]byte, len(f.Data()))
- copy(data, f.Data())
- s.write(recvMsg{data: data})
+ buffer := t.bufferPool.get()
+ buffer.Reset()
+ buffer.Write(f.Data())
+ s.write(recvMsg{buffer: buffer})
}
}
if f.Header().Flags.Has(http2.FlagDataEndStream) {
@@ -602,11 +619,18 @@
}
func (t *http2Server) handleRSTStream(f *http2.RSTStreamFrame) {
- s, ok := t.getStream(f)
- if !ok {
+ // If the stream is not deleted from the transport's active streams map, then do a regular close stream.
+ if s, ok := t.getStream(f); ok {
+ t.closeStream(s, false, 0, false)
return
}
- t.closeStream(s, false, 0, nil, false)
+ // If the stream is already deleted from the active streams map, then put a cleanupStream item into controlbuf to delete the stream from loopy writer's established streams map.
+ t.controlBuf.put(&cleanupStream{
+ streamID: f.Header().StreamID,
+ rst: false,
+ rstCode: 0,
+ onWrite: func() {},
+ })
}
func (t *http2Server) handleSettings(f *http2.SettingsFrame) {
@@ -748,6 +772,10 @@
return nil
}
+func (t *http2Server) setResetPingStrikes() {
+ atomic.StoreUint32(&t.resetPingStrikes, 1)
+}
+
func (t *http2Server) writeHeaderLocked(s *Stream) error {
// TODO(mmukhi): Benchmark if the performance gets better if count the metadata and other header fields
// first and create a slice of that exact size.
@@ -762,15 +790,13 @@
streamID: s.id,
hf: headerFields,
endStream: false,
- onWrite: func() {
- atomic.StoreUint32(&t.resetPingStrikes, 1)
- },
+ onWrite: t.setResetPingStrikes,
})
if !success {
if err != nil {
return err
}
- t.closeStream(s, true, http2.ErrCodeInternal, nil, false)
+ t.closeStream(s, true, http2.ErrCodeInternal, false)
return ErrHeaderListSizeLimitViolation
}
if t.stats != nil {
@@ -808,7 +834,7 @@
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-status", Value: strconv.Itoa(int(st.Code()))})
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-message", Value: encodeGrpcMessage(st.Message())})
- if p := st.Proto(); p != nil && len(p.Details) > 0 {
+ if p := statusRawProto(st); p != nil && len(p.Details) > 0 {
stBytes, err := proto.Marshal(p)
if err != nil {
// TODO: return error instead, when callers are able to handle it.
@@ -824,9 +850,7 @@
streamID: s.id,
hf: headerFields,
endStream: true,
- onWrite: func() {
- atomic.StoreUint32(&t.resetPingStrikes, 1)
- },
+ onWrite: t.setResetPingStrikes,
}
s.hdrMu.Unlock()
success, err := t.controlBuf.execute(t.checkForHeaderListSize, trailingHeader)
@@ -834,10 +858,12 @@
if err != nil {
return err
}
- t.closeStream(s, true, http2.ErrCodeInternal, nil, false)
+ t.closeStream(s, true, http2.ErrCodeInternal, false)
return ErrHeaderListSizeLimitViolation
}
- t.closeStream(s, false, 0, trailingHeader, true)
+ // Send a RST_STREAM after the trailers if the client has not already half-closed.
+ rst := s.getState() == streamActive
+ t.finishStream(s, rst, http2.ErrCodeNo, trailingHeader, true)
if t.stats != nil {
t.stats.HandleRPC(s.Context(), &stats.OutTrailer{})
}
@@ -849,6 +875,9 @@
func (t *http2Server) Write(s *Stream, hdr []byte, data []byte, opts *Options) error {
if !s.isHeaderSent() { // Headers haven't been written yet.
if err := t.WriteHeader(s, nil); err != nil {
+ if _, ok := err.(ConnectionError); ok {
+ return err
+ }
// TODO(mmukhi, dfawley): Make sure this is the right code to return.
return status.Errorf(codes.Internal, "transport: %v", err)
}
@@ -858,7 +887,7 @@
// TODO(mmukhi, dfawley): Should the server write also return io.EOF?
s.cancel()
select {
- case <-t.ctx.Done():
+ case <-t.done:
return ErrConnClosing
default:
}
@@ -873,16 +902,14 @@
hdr = append(hdr, data[:emptyLen]...)
data = data[emptyLen:]
df := &dataFrame{
- streamID: s.id,
- h: hdr,
- d: data,
- onEachWrite: func() {
- atomic.StoreUint32(&t.resetPingStrikes, 1)
- },
+ streamID: s.id,
+ h: hdr,
+ d: data,
+ onEachWrite: t.setResetPingStrikes,
}
if err := s.wq.get(int32(len(hdr) + len(data))); err != nil {
select {
- case <-t.ctx.Done():
+ case <-t.done:
return ErrConnClosing
default:
}
@@ -944,10 +971,11 @@
select {
case <-maxAge.C:
// Close the connection after grace period.
+ infof("transport: closing server transport due to maximum connection age.")
t.Close()
// Resetting the timer so that the clean-up doesn't deadlock.
maxAge.Reset(infinity)
- case <-t.ctx.Done():
+ case <-t.done:
}
return
case <-keepalive.C:
@@ -957,6 +985,7 @@
continue
}
if pingSent {
+ infof("transport: closing server transport due to idleness.")
t.Close()
// Resetting the timer so that the clean-up doesn't deadlock.
keepalive.Reset(infinity)
@@ -968,7 +997,7 @@
}
t.controlBuf.put(p)
keepalive.Reset(t.kp.Timeout)
- case <-t.ctx.Done():
+ case <-t.done:
return
}
}
@@ -988,7 +1017,7 @@
t.activeStreams = nil
t.mu.Unlock()
t.controlBuf.finish()
- t.cancel()
+ close(t.done)
err := t.conn.Close()
if channelz.IsOn() {
channelz.RemoveEntry(t.channelzID)
@@ -1006,15 +1035,17 @@
// deleteStream deletes the stream s from transport's active streams.
func (t *http2Server) deleteStream(s *Stream, eosReceived bool) {
- t.mu.Lock()
- if _, ok := t.activeStreams[s.id]; !ok {
- t.mu.Unlock()
- return
- }
+ // In case stream sending and receiving are invoked in separate
+ // goroutines (e.g., bi-directional streaming), cancel needs to be
+ // called to interrupt the potential blocking on other goroutines.
+ s.cancel()
- delete(t.activeStreams, s.id)
- if len(t.activeStreams) == 0 {
- t.idle = time.Now()
+ t.mu.Lock()
+ if _, ok := t.activeStreams[s.id]; ok {
+ delete(t.activeStreams, s.id)
+ if len(t.activeStreams) == 0 {
+ t.idle = time.Now()
+ }
}
t.mu.Unlock()
@@ -1027,51 +1058,36 @@
}
}
-// closeStream clears the footprint of a stream when the stream is not needed
-// any more.
-func (t *http2Server) closeStream(s *Stream, rst bool, rstCode http2.ErrCode, hdr *headerFrame, eosReceived bool) {
- // Mark the stream as done
+// finishStream closes the stream and puts the trailing headerFrame into controlbuf.
+func (t *http2Server) finishStream(s *Stream, rst bool, rstCode http2.ErrCode, hdr *headerFrame, eosReceived bool) {
oldState := s.swapState(streamDone)
+ if oldState == streamDone {
+ // If the stream was already done, return.
+ return
+ }
- // In case stream sending and receiving are invoked in separate
- // goroutines (e.g., bi-directional streaming), cancel needs to be
- // called to interrupt the potential blocking on other goroutines.
- s.cancel()
+ hdr.cleanup = &cleanupStream{
+ streamID: s.id,
+ rst: rst,
+ rstCode: rstCode,
+ onWrite: func() {
+ t.deleteStream(s, eosReceived)
+ },
+ }
+ t.controlBuf.put(hdr)
+}
- // Deletes the stream from active streams
+// closeStream clears the footprint of a stream when the stream is not needed any more.
+func (t *http2Server) closeStream(s *Stream, rst bool, rstCode http2.ErrCode, eosReceived bool) {
+ s.swapState(streamDone)
t.deleteStream(s, eosReceived)
- cleanup := &cleanupStream{
+ t.controlBuf.put(&cleanupStream{
streamID: s.id,
rst: rst,
rstCode: rstCode,
onWrite: func() {},
- }
-
- // No trailer. Puts cleanupFrame into transport's control buffer.
- if hdr == nil {
- t.controlBuf.put(cleanup)
- return
- }
-
- // We do the check here, because of the following scenario:
- // 1. closeStream is called first with a trailer. A trailer item with a piggybacked cleanup item
- // is put to control buffer.
- // 2. Loopy writer is waiting on a stream quota. It will never get it because client errored at
- // some point. So loopy can't act on trailer
- // 3. Client sends a RST_STREAM due to the error. Then closeStream is called without a trailer as
- // the result of the received RST_STREAM.
- // If we do this check at the beginning of the closeStream, then we won't put a cleanup item in
- // response to received RST_STREAM into the control buffer and outStream in loopy writer will
- // never get cleaned up.
-
- // If the stream is already done, don't send the trailer.
- if oldState == streamDone {
- return
- }
-
- hdr.cleanup = cleanup
- t.controlBuf.put(hdr)
+ })
}
func (t *http2Server) RemoteAddr() net.Addr {
@@ -1141,7 +1157,7 @@
select {
case <-t.drainChan:
case <-timer.C:
- case <-t.ctx.Done():
+ case <-t.done:
return
}
t.controlBuf.put(&goAway{code: g.code, debugData: g.debugData})
@@ -1191,7 +1207,7 @@
select {
case sz := <-resp:
return int64(sz)
- case <-t.ctxDone:
+ case <-t.done:
return -1
case <-timer.C:
return -2
diff --git a/vendor/google.golang.org/grpc/internal/transport/http_util.go b/vendor/google.golang.org/grpc/internal/transport/http_util.go
index 77a2cfa..8f5f334 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http_util.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http_util.go
@@ -78,7 +78,8 @@
codes.ResourceExhausted: http2.ErrCodeEnhanceYourCalm,
codes.PermissionDenied: http2.ErrCodeInadequateSecurity,
}
- httpStatusConvTab = map[int]codes.Code{
+ // HTTPStatusConvTab is the HTTP status code to gRPC error code conversion table.
+ HTTPStatusConvTab = map[int]codes.Code{
// 400 Bad Request - INTERNAL.
http.StatusBadRequest: codes.Internal,
// 401 Unauthorized - UNAUTHENTICATED.
@@ -98,9 +99,7 @@
}
)
-// Records the states during HPACK decoding. Must be reset once the
-// decoding of the entire headers are finished.
-type decodeState struct {
+type parsedHeaderData struct {
encoding string
// statusGen caches the stream status received from the trailer the server
// sent. Client side only. Do not access directly. After all trailers are
@@ -120,8 +119,30 @@
statsTags []byte
statsTrace []byte
contentSubtype string
+
+ // isGRPC field indicates whether the peer is speaking gRPC (otherwise HTTP).
+ //
+ // We are in gRPC mode (peer speaking gRPC) if:
+ // * We are client side and have already received a HEADER frame that indicates gRPC peer.
+ // * The header contains valid a content-type, i.e. a string starts with "application/grpc"
+ // And we should handle error specific to gRPC.
+ //
+ // Otherwise (i.e. a content-type string starts without "application/grpc", or does not exist), we
+ // are in HTTP fallback mode, and should handle error specific to HTTP.
+ isGRPC bool
+ grpcErr error
+ httpErr error
+ contentTypeErr string
+}
+
+// decodeState configures decoding criteria and records the decoded data.
+type decodeState struct {
// whether decoding on server side or not
serverSide bool
+
+ // Records the states during HPACK decoding. It will be filled with info parsed from HTTP HEADERS
+ // frame once decodeHeader function has been invoked and returned.
+ data parsedHeaderData
}
// isReservedHeader checks whether hdr belongs to HTTP2 headers
@@ -202,11 +223,11 @@
}
func (d *decodeState) status() *status.Status {
- if d.statusGen == nil {
+ if d.data.statusGen == nil {
// No status-details were provided; generate status using code/msg.
- d.statusGen = status.New(codes.Code(int32(*(d.rawStatusCode))), d.rawStatusMsg)
+ d.data.statusGen = status.New(codes.Code(int32(*(d.data.rawStatusCode))), d.data.rawStatusMsg)
}
- return d.statusGen
+ return d.data.statusGen
}
const binHdrSuffix = "-bin"
@@ -244,113 +265,146 @@
if frame.Truncated {
return status.Error(codes.Internal, "peer header list size exceeded limit")
}
+
for _, hf := range frame.Fields {
- if err := d.processHeaderField(hf); err != nil {
- return err
+ d.processHeaderField(hf)
+ }
+
+ if d.data.isGRPC {
+ if d.data.grpcErr != nil {
+ return d.data.grpcErr
}
- }
-
- if d.serverSide {
+ if d.serverSide {
+ return nil
+ }
+ if d.data.rawStatusCode == nil && d.data.statusGen == nil {
+ // gRPC status doesn't exist.
+ // Set rawStatusCode to be unknown and return nil error.
+ // So that, if the stream has ended this Unknown status
+ // will be propagated to the user.
+ // Otherwise, it will be ignored. In which case, status from
+ // a later trailer, that has StreamEnded flag set, is propagated.
+ code := int(codes.Unknown)
+ d.data.rawStatusCode = &code
+ }
return nil
}
- // If grpc status exists, no need to check further.
- if d.rawStatusCode != nil || d.statusGen != nil {
- return nil
+ // HTTP fallback mode
+ if d.data.httpErr != nil {
+ return d.data.httpErr
}
- // If grpc status doesn't exist and http status doesn't exist,
- // then it's a malformed header.
- if d.httpStatus == nil {
- return status.Error(codes.Internal, "malformed header: doesn't contain status(gRPC or HTTP)")
- }
+ var (
+ code = codes.Internal // when header does not include HTTP status, return INTERNAL
+ ok bool
+ )
- if *(d.httpStatus) != http.StatusOK {
- code, ok := httpStatusConvTab[*(d.httpStatus)]
+ if d.data.httpStatus != nil {
+ code, ok = HTTPStatusConvTab[*(d.data.httpStatus)]
if !ok {
code = codes.Unknown
}
- return status.Error(code, http.StatusText(*(d.httpStatus)))
}
- // gRPC status doesn't exist and http status is OK.
- // Set rawStatusCode to be unknown and return nil error.
- // So that, if the stream has ended this Unknown status
- // will be propagated to the user.
- // Otherwise, it will be ignored. In which case, status from
- // a later trailer, that has StreamEnded flag set, is propagated.
- code := int(codes.Unknown)
- d.rawStatusCode = &code
- return nil
+ return status.Error(code, d.constructHTTPErrMsg())
+}
+
+// constructErrMsg constructs error message to be returned in HTTP fallback mode.
+// Format: HTTP status code and its corresponding message + content-type error message.
+func (d *decodeState) constructHTTPErrMsg() string {
+ var errMsgs []string
+
+ if d.data.httpStatus == nil {
+ errMsgs = append(errMsgs, "malformed header: missing HTTP status")
+ } else {
+ errMsgs = append(errMsgs, fmt.Sprintf("%s: HTTP status code %d", http.StatusText(*(d.data.httpStatus)), *d.data.httpStatus))
+ }
+
+ if d.data.contentTypeErr == "" {
+ errMsgs = append(errMsgs, "transport: missing content-type field")
+ } else {
+ errMsgs = append(errMsgs, d.data.contentTypeErr)
+ }
+
+ return strings.Join(errMsgs, "; ")
}
func (d *decodeState) addMetadata(k, v string) {
- if d.mdata == nil {
- d.mdata = make(map[string][]string)
+ if d.data.mdata == nil {
+ d.data.mdata = make(map[string][]string)
}
- d.mdata[k] = append(d.mdata[k], v)
+ d.data.mdata[k] = append(d.data.mdata[k], v)
}
-func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
+func (d *decodeState) processHeaderField(f hpack.HeaderField) {
switch f.Name {
case "content-type":
contentSubtype, validContentType := contentSubtype(f.Value)
if !validContentType {
- return status.Errorf(codes.Internal, "transport: received the unexpected content-type %q", f.Value)
+ d.data.contentTypeErr = fmt.Sprintf("transport: received the unexpected content-type %q", f.Value)
+ return
}
- d.contentSubtype = contentSubtype
+ d.data.contentSubtype = contentSubtype
// TODO: do we want to propagate the whole content-type in the metadata,
// or come up with a way to just propagate the content-subtype if it was set?
// ie {"content-type": "application/grpc+proto"} or {"content-subtype": "proto"}
// in the metadata?
d.addMetadata(f.Name, f.Value)
+ d.data.isGRPC = true
case "grpc-encoding":
- d.encoding = f.Value
+ d.data.encoding = f.Value
case "grpc-status":
code, err := strconv.Atoi(f.Value)
if err != nil {
- return status.Errorf(codes.Internal, "transport: malformed grpc-status: %v", err)
+ d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status: %v", err)
+ return
}
- d.rawStatusCode = &code
+ d.data.rawStatusCode = &code
case "grpc-message":
- d.rawStatusMsg = decodeGrpcMessage(f.Value)
+ d.data.rawStatusMsg = decodeGrpcMessage(f.Value)
case "grpc-status-details-bin":
v, err := decodeBinHeader(f.Value)
if err != nil {
- return status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
+ d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
+ return
}
s := &spb.Status{}
if err := proto.Unmarshal(v, s); err != nil {
- return status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
+ d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
+ return
}
- d.statusGen = status.FromProto(s)
+ d.data.statusGen = status.FromProto(s)
case "grpc-timeout":
- d.timeoutSet = true
+ d.data.timeoutSet = true
var err error
- if d.timeout, err = decodeTimeout(f.Value); err != nil {
- return status.Errorf(codes.Internal, "transport: malformed time-out: %v", err)
+ if d.data.timeout, err = decodeTimeout(f.Value); err != nil {
+ d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed time-out: %v", err)
}
case ":path":
- d.method = f.Value
+ d.data.method = f.Value
case ":status":
code, err := strconv.Atoi(f.Value)
if err != nil {
- return status.Errorf(codes.Internal, "transport: malformed http-status: %v", err)
+ d.data.httpErr = status.Errorf(codes.Internal, "transport: malformed http-status: %v", err)
+ return
}
- d.httpStatus = &code
+ d.data.httpStatus = &code
case "grpc-tags-bin":
v, err := decodeBinHeader(f.Value)
if err != nil {
- return status.Errorf(codes.Internal, "transport: malformed grpc-tags-bin: %v", err)
+ d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-tags-bin: %v", err)
+ return
}
- d.statsTags = v
+ d.data.statsTags = v
d.addMetadata(f.Name, string(v))
case "grpc-trace-bin":
v, err := decodeBinHeader(f.Value)
if err != nil {
- return status.Errorf(codes.Internal, "transport: malformed grpc-trace-bin: %v", err)
+ d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-trace-bin: %v", err)
+ return
}
- d.statsTrace = v
+ d.data.statsTrace = v
d.addMetadata(f.Name, string(v))
default:
if isReservedHeader(f.Name) && !isWhitelistedHeader(f.Name) {
@@ -359,11 +413,10 @@
v, err := decodeMetadataHeader(f.Name, f.Value)
if err != nil {
errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
- return nil
+ return
}
d.addMetadata(f.Name, v)
}
- return nil
}
type timeoutUnit uint8
@@ -614,6 +667,7 @@
writer: w,
fr: http2.NewFramer(w, r),
}
+ f.fr.SetMaxReadFrameSize(http2MaxFrameLen)
// Opt-in to Frame reuse API on framer to reduce garbage.
// Frames aren't safe to read from after a subsequent call to ReadFrame.
f.fr.SetReuseFrames()
diff --git a/vendor/google.golang.org/grpc/internal/transport/transport.go b/vendor/google.golang.org/grpc/internal/transport/transport.go
index 2580aa7..1c1d106 100644
--- a/vendor/google.golang.org/grpc/internal/transport/transport.go
+++ b/vendor/google.golang.org/grpc/internal/transport/transport.go
@@ -22,6 +22,7 @@
package transport
import (
+ "bytes"
"context"
"errors"
"fmt"
@@ -39,10 +40,32 @@
"google.golang.org/grpc/tap"
)
+type bufferPool struct {
+ pool sync.Pool
+}
+
+func newBufferPool() *bufferPool {
+ return &bufferPool{
+ pool: sync.Pool{
+ New: func() interface{} {
+ return new(bytes.Buffer)
+ },
+ },
+ }
+}
+
+func (p *bufferPool) get() *bytes.Buffer {
+ return p.pool.Get().(*bytes.Buffer)
+}
+
+func (p *bufferPool) put(b *bytes.Buffer) {
+ p.pool.Put(b)
+}
+
// recvMsg represents the received msg from the transport. All transport
// protocol specific info has been removed.
type recvMsg struct {
- data []byte
+ buffer *bytes.Buffer
// nil: received some data
// io.EOF: stream is completed. data is nil.
// other non-nil error: transport failure. data is nil.
@@ -117,8 +140,9 @@
ctx context.Context
ctxDone <-chan struct{} // cache of ctx.Done() (for performance).
recv *recvBuffer
- last []byte // Stores the remaining data in the previous calls.
+ last *bytes.Buffer // Stores the remaining data in the previous calls.
err error
+ freeBuffer func(*bytes.Buffer)
}
// Read reads the next len(p) bytes from last. If last is drained, it tries to
@@ -128,10 +152,13 @@
if r.err != nil {
return 0, r.err
}
- if r.last != nil && len(r.last) > 0 {
+ if r.last != nil {
// Read remaining data left in last call.
- copied := copy(p, r.last)
- r.last = r.last[copied:]
+ copied, _ := r.last.Read(p)
+ if r.last.Len() == 0 {
+ r.freeBuffer(r.last)
+ r.last = nil
+ }
return copied, nil
}
if r.closeStream != nil {
@@ -157,6 +184,19 @@
// r.readAdditional acts on that message and returns the necessary error.
select {
case <-r.ctxDone:
+ // Note that this adds the ctx error to the end of recv buffer, and
+ // reads from the head. This will delay the error until recv buffer is
+ // empty, thus will delay ctx cancellation in Recv().
+ //
+ // It's done this way to fix a race between ctx cancel and trailer. The
+ // race was, stream.Recv() may return ctx error if ctxDone wins the
+ // race, but stream.Trailer() may return a non-nil md because the stream
+ // was not marked as done when trailer is received. This closeStream
+ // call will mark stream as done, thus fix the race.
+ //
+ // TODO: delaying ctx error seems like a unnecessary side effect. What
+ // we really want is to mark the stream as done, and return ctx error
+ // faster.
r.closeStream(ContextErr(r.ctx.Err()))
m := <-r.recv.get()
return r.readAdditional(m, p)
@@ -170,8 +210,13 @@
if m.err != nil {
return 0, m.err
}
- copied := copy(p, m.data)
- r.last = m.data[copied:]
+ copied, _ := m.buffer.Read(p)
+ if m.buffer.Len() == 0 {
+ r.freeBuffer(m.buffer)
+ r.last = nil
+ } else {
+ r.last = m.buffer
+ }
return copied, nil
}
@@ -204,8 +249,8 @@
// is used to adjust flow control, if needed.
requestRead func(int)
- headerChan chan struct{} // closed to indicate the end of header metadata.
- headerDone uint32 // set when headerChan is closed. Used to avoid closing headerChan multiple times.
+ headerChan chan struct{} // closed to indicate the end of header metadata.
+ headerChanClosed uint32 // set when headerChan is closed. Used to avoid closing headerChan multiple times.
// hdrMu protects header and trailer metadata on the server-side.
hdrMu sync.Mutex
@@ -266,6 +311,14 @@
}
select {
case <-s.ctx.Done():
+ // We prefer success over failure when reading messages because we delay
+ // context error in stream.Read(). To keep behavior consistent, we also
+ // prefer success here.
+ select {
+ case <-s.headerChan:
+ return nil
+ default:
+ }
return ContextErr(s.ctx.Err())
case <-s.headerChan:
return nil
@@ -327,8 +380,7 @@
if err != nil {
return false, err
}
- // if !headerDone, some other connection error occurred.
- return s.noHeaders && atomic.LoadUint32(&s.headerDone) == 1, nil
+ return s.noHeaders, nil
}
// Trailer returns the cached trailer metedata. Note that if it is not called
@@ -579,9 +631,12 @@
// is called only once.
Close() error
- // GracefulClose starts to tear down the transport. It stops accepting
- // new RPCs and wait the completion of the pending RPCs.
- GracefulClose() error
+ // GracefulClose starts to tear down the transport: the transport will stop
+ // accepting new RPCs and NewStream will return error. Once all streams are
+ // finished, the transport will close.
+ //
+ // It does not block.
+ GracefulClose()
// Write sends the data for the given stream. A nil stream indicates
// the write is to be performed on the transport as a whole.
@@ -611,6 +666,9 @@
// GetGoAwayReason returns the reason why GoAway frame was received.
GetGoAwayReason() GoAwayReason
+ // RemoteAddr returns the remote network address.
+ RemoteAddr() net.Addr
+
// IncrMsgSent increments the number of message sent through this transport.
IncrMsgSent()
diff --git a/vendor/google.golang.org/grpc/naming/naming.go b/vendor/google.golang.org/grpc/naming/naming.go
index c99fdbe..f4c1c8b 100644
--- a/vendor/google.golang.org/grpc/naming/naming.go
+++ b/vendor/google.golang.org/grpc/naming/naming.go
@@ -17,9 +17,8 @@
*/
// Package naming defines the naming API and related data structures for gRPC.
-// The interface is EXPERIMENTAL and may be subject to change.
//
-// Deprecated: please use package resolver.
+// This package is deprecated: please use package resolver instead.
package naming
// Operation defines the corresponding operations for a name resolution change.
diff --git a/vendor/google.golang.org/grpc/picker_wrapper.go b/vendor/google.golang.org/grpc/picker_wrapper.go
index a2575c9..45baa2a 100644
--- a/vendor/google.golang.org/grpc/picker_wrapper.go
+++ b/vendor/google.golang.org/grpc/picker_wrapper.go
@@ -120,6 +120,14 @@
bp.mu.Unlock()
select {
case <-ctx.Done():
+ if connectionErr := bp.connectionError(); connectionErr != nil {
+ switch ctx.Err() {
+ case context.DeadlineExceeded:
+ return nil, nil, status.Errorf(codes.DeadlineExceeded, "latest connection error: %v", connectionErr)
+ case context.Canceled:
+ return nil, nil, status.Errorf(codes.Canceled, "latest connection error: %v", connectionErr)
+ }
+ }
return nil, nil, ctx.Err()
case <-ch:
}
@@ -165,6 +173,11 @@
}
return t, done, nil
}
+ if done != nil {
+ // Calling done with nil error, no bytes sent and no bytes received.
+ // DoneInfo with default value works.
+ done(balancer.DoneInfo{})
+ }
grpclog.Infof("blockingPicker: the picked transport is not ready, loop back to repick")
// If ok == false, ac.state is not READY.
// A valid picker always returns READY subConn. This means the state of ac
diff --git a/vendor/google.golang.org/grpc/pickfirst.go b/vendor/google.golang.org/grpc/pickfirst.go
index d1e38aa..ed05b02 100644
--- a/vendor/google.golang.org/grpc/pickfirst.go
+++ b/vendor/google.golang.org/grpc/pickfirst.go
@@ -51,14 +51,18 @@
func (b *pickfirstBalancer) HandleResolvedAddrs(addrs []resolver.Address, err error) {
if err != nil {
- grpclog.Infof("pickfirstBalancer: HandleResolvedAddrs called with error %v", err)
+ if grpclog.V(2) {
+ grpclog.Infof("pickfirstBalancer: HandleResolvedAddrs called with error %v", err)
+ }
return
}
if b.sc == nil {
b.sc, err = b.cc.NewSubConn(addrs, balancer.NewSubConnOptions{})
if err != nil {
//TODO(yuxuanli): why not change the cc state to Idle?
- grpclog.Errorf("pickfirstBalancer: failed to NewSubConn: %v", err)
+ if grpclog.V(2) {
+ grpclog.Errorf("pickfirstBalancer: failed to NewSubConn: %v", err)
+ }
return
}
b.cc.UpdateBalancerState(connectivity.Idle, &picker{sc: b.sc})
@@ -70,9 +74,13 @@
}
func (b *pickfirstBalancer) HandleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
- grpclog.Infof("pickfirstBalancer: HandleSubConnStateChange: %p, %v", sc, s)
+ if grpclog.V(2) {
+ grpclog.Infof("pickfirstBalancer: HandleSubConnStateChange: %p, %v", sc, s)
+ }
if b.sc != sc {
- grpclog.Infof("pickfirstBalancer: ignored state change because sc is not recognized")
+ if grpclog.V(2) {
+ grpclog.Infof("pickfirstBalancer: ignored state change because sc is not recognized")
+ }
return
}
if s == connectivity.Shutdown {
diff --git a/vendor/google.golang.org/grpc/preloader.go b/vendor/google.golang.org/grpc/preloader.go
new file mode 100644
index 0000000..76acbbc
--- /dev/null
+++ b/vendor/google.golang.org/grpc/preloader.go
@@ -0,0 +1,64 @@
+/*
+ *
+ * Copyright 2019 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.
+ *
+ */
+
+package grpc
+
+import (
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+// PreparedMsg is responsible for creating a Marshalled and Compressed object.
+//
+// This API is EXPERIMENTAL.
+type PreparedMsg struct {
+ // Struct for preparing msg before sending them
+ encodedData []byte
+ hdr []byte
+ payload []byte
+}
+
+// Encode marshalls and compresses the message using the codec and compressor for the stream.
+func (p *PreparedMsg) Encode(s Stream, msg interface{}) error {
+ ctx := s.Context()
+ rpcInfo, ok := rpcInfoFromContext(ctx)
+ if !ok {
+ return status.Errorf(codes.Internal, "grpc: unable to get rpcInfo")
+ }
+
+ // check if the context has the relevant information to prepareMsg
+ if rpcInfo.preloaderInfo == nil {
+ return status.Errorf(codes.Internal, "grpc: rpcInfo.preloaderInfo is nil")
+ }
+ if rpcInfo.preloaderInfo.codec == nil {
+ return status.Errorf(codes.Internal, "grpc: rpcInfo.preloaderInfo.codec is nil")
+ }
+
+ // prepare the msg
+ data, err := encode(rpcInfo.preloaderInfo.codec, msg)
+ if err != nil {
+ return err
+ }
+ p.encodedData = data
+ compData, err := compress(data, rpcInfo.preloaderInfo.cp, rpcInfo.preloaderInfo.comp)
+ if err != nil {
+ return err
+ }
+ p.hdr, p.payload = msgHeader(data, compData)
+ return nil
+}
diff --git a/vendor/google.golang.org/grpc/resolver/dns/dns_resolver.go b/vendor/google.golang.org/grpc/resolver/dns/dns_resolver.go
index 2d8da33..297492e 100644
--- a/vendor/google.golang.org/grpc/resolver/dns/dns_resolver.go
+++ b/vendor/google.golang.org/grpc/resolver/dns/dns_resolver.go
@@ -47,6 +47,8 @@
defaultFreq = time.Minute * 30
defaultDNSSvrPort = "53"
golang = "GO"
+ // txtPrefix is the prefix string to be prepended to the host name for txt record lookup.
+ txtPrefix = "_grpc_config."
// In DNS, service config is encoded in a TXT record via the mechanism
// described in RFC-1464 using the attribute name grpc_config.
txtAttribute = "grpc_config="
@@ -64,6 +66,9 @@
var (
defaultResolver netResolver = net.DefaultResolver
+ // To prevent excessive re-resolution, we enforce a rate limit on DNS
+ // resolution requests.
+ minDNSResRate = 30 * time.Second
)
var customAuthorityDialler = func(authority string) func(ctx context.Context, network, address string) (net.Conn, error) {
@@ -239,7 +244,13 @@
return
case <-d.t.C:
case <-d.rn:
+ if !d.t.Stop() {
+ // Before resetting a timer, it should be stopped to prevent racing with
+ // reads on it's channel.
+ <-d.t.C
+ }
}
+
result, sc := d.lookup()
// Next lookup should happen within an interval defined by d.freq. It may be
// more often due to exponential retry on empty address list.
@@ -252,6 +263,16 @@
}
d.cc.NewServiceConfig(sc)
d.cc.NewAddress(result)
+
+ // Sleep to prevent excessive re-resolutions. Incoming resolution requests
+ // will be queued in d.rn.
+ t := time.NewTimer(minDNSResRate)
+ select {
+ case <-t.C:
+ case <-d.ctx.Done():
+ t.Stop()
+ return
+ }
}
}
@@ -282,7 +303,7 @@
}
func (d *dnsResolver) lookupTXT() string {
- ss, err := d.resolver.LookupTXT(d.ctx, d.host)
+ ss, err := d.resolver.LookupTXT(d.ctx, txtPrefix+d.host)
if err != nil {
grpclog.Infof("grpc: failed dns TXT record lookup due to %v.\n", err)
return ""
diff --git a/vendor/google.golang.org/grpc/resolver/passthrough/passthrough.go b/vendor/google.golang.org/grpc/resolver/passthrough/passthrough.go
index b76010d..893d5d1 100644
--- a/vendor/google.golang.org/grpc/resolver/passthrough/passthrough.go
+++ b/vendor/google.golang.org/grpc/resolver/passthrough/passthrough.go
@@ -45,7 +45,7 @@
}
func (r *passthroughResolver) start() {
- r.cc.NewAddress([]resolver.Address{{Addr: r.target.Endpoint}})
+ r.cc.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: r.target.Endpoint}}})
}
func (*passthroughResolver) ResolveNow(o resolver.ResolveNowOption) {}
diff --git a/vendor/google.golang.org/grpc/resolver/resolver.go b/vendor/google.golang.org/grpc/resolver/resolver.go
index 145cf47..e83da34 100644
--- a/vendor/google.golang.org/grpc/resolver/resolver.go
+++ b/vendor/google.golang.org/grpc/resolver/resolver.go
@@ -20,6 +20,10 @@
// All APIs in this package are experimental.
package resolver
+import (
+ "google.golang.org/grpc/serviceconfig"
+)
+
var (
// m is a map from scheme to resolver builder.
m = make(map[string]Builder)
@@ -98,6 +102,16 @@
DisableServiceConfig bool
}
+// State contains the current Resolver state relevant to the ClientConn.
+type State struct {
+ Addresses []Address // Resolved addresses for the target
+ // ServiceConfig is the parsed service config; obtained from
+ // serviceconfig.Parse.
+ ServiceConfig serviceconfig.Config
+
+ // TODO: add Err error
+}
+
// ClientConn contains the callbacks for resolver to notify any updates
// to the gRPC ClientConn.
//
@@ -106,17 +120,38 @@
// testing, the new implementation should embed this interface. This allows
// gRPC to add new methods to this interface.
type ClientConn interface {
+ // UpdateState updates the state of the ClientConn appropriately.
+ UpdateState(State)
// NewAddress is called by resolver to notify ClientConn a new list
// of resolved addresses.
// The address list should be the complete list of resolved addresses.
+ //
+ // Deprecated: Use UpdateState instead.
NewAddress(addresses []Address)
// NewServiceConfig is called by resolver to notify ClientConn a new
// service config. The service config should be provided as a json string.
+ //
+ // Deprecated: Use UpdateState instead.
NewServiceConfig(serviceConfig string)
}
// Target represents a target for gRPC, as specified in:
// https://github.com/grpc/grpc/blob/master/doc/naming.md.
+// It is parsed from the target string that gets passed into Dial or DialContext by the user. And
+// grpc passes it to the resolver and the balancer.
+//
+// If the target follows the naming spec, and the parsed scheme is registered with grpc, we will
+// parse the target string according to the spec. e.g. "dns://some_authority/foo.bar" will be parsed
+// into &Target{Scheme: "dns", Authority: "some_authority", Endpoint: "foo.bar"}
+//
+// If the target does not contain a scheme, we will apply the default scheme, and set the Target to
+// be the full target string. e.g. "foo.bar" will be parsed into
+// &Target{Scheme: resolver.GetDefaultScheme(), Endpoint: "foo.bar"}.
+//
+// If the parsed scheme is not registered (i.e. no corresponding resolver available to resolve the
+// endpoint), we set the Scheme to be the default scheme, and set the Endpoint to be the full target
+// string. e.g. target string "unknown_scheme://authority/endpoint" will be parsed into
+// &Target{Scheme: resolver.GetDefaultScheme(), Endpoint: "unknown_scheme://authority/endpoint"}.
type Target struct {
Scheme string
Authority string
diff --git a/vendor/google.golang.org/grpc/resolver_conn_wrapper.go b/vendor/google.golang.org/grpc/resolver_conn_wrapper.go
index 50991ea..6934905 100644
--- a/vendor/google.golang.org/grpc/resolver_conn_wrapper.go
+++ b/vendor/google.golang.org/grpc/resolver_conn_wrapper.go
@@ -21,6 +21,7 @@
import (
"fmt"
"strings"
+ "sync/atomic"
"google.golang.org/grpc/grpclog"
"google.golang.org/grpc/internal/channelz"
@@ -30,12 +31,12 @@
// ccResolverWrapper is a wrapper on top of cc for resolvers.
// It implements resolver.ClientConnection interface.
type ccResolverWrapper struct {
- cc *ClientConn
- resolver resolver.Resolver
- addrCh chan []resolver.Address
- scCh chan string
- done chan struct{}
- lastAddressesCount int
+ cc *ClientConn
+ resolver resolver.Resolver
+ addrCh chan []resolver.Address
+ scCh chan string
+ done uint32 // accessed atomically; set to 1 when closed.
+ curState resolver.State
}
// split2 returns the values from strings.SplitN(s, sep, 2).
@@ -82,7 +83,6 @@
cc: cc,
addrCh: make(chan []resolver.Address, 1),
scCh: make(chan string, 1),
- done: make(chan struct{}),
}
var err error
@@ -99,57 +99,70 @@
func (ccr *ccResolverWrapper) close() {
ccr.resolver.Close()
- close(ccr.done)
+ atomic.StoreUint32(&ccr.done, 1)
}
-// NewAddress is called by the resolver implemenetion to send addresses to gRPC.
-func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
- select {
- case <-ccr.done:
+func (ccr *ccResolverWrapper) isDone() bool {
+ return atomic.LoadUint32(&ccr.done) == 1
+}
+
+func (ccr *ccResolverWrapper) UpdateState(s resolver.State) {
+ if ccr.isDone() {
return
- default:
+ }
+ grpclog.Infof("ccResolverWrapper: sending update to cc: %v", s)
+ if channelz.IsOn() {
+ ccr.addChannelzTraceEvent(s)
+ }
+ ccr.cc.updateResolverState(s)
+ ccr.curState = s
+}
+
+// NewAddress is called by the resolver implementation to send addresses to gRPC.
+func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
+ if ccr.isDone() {
+ return
}
grpclog.Infof("ccResolverWrapper: sending new addresses to cc: %v", addrs)
if channelz.IsOn() {
- ccr.addChannelzTraceEvent(addrs)
+ ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig})
}
- ccr.cc.handleResolvedAddrs(addrs, nil)
+ ccr.curState.Addresses = addrs
+ ccr.cc.updateResolverState(ccr.curState)
}
-// NewServiceConfig is called by the resolver implemenetion to send service
+// NewServiceConfig is called by the resolver implementation to send service
// configs to gRPC.
func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
- select {
- case <-ccr.done:
+ if ccr.isDone() {
return
- default:
}
grpclog.Infof("ccResolverWrapper: got new service config: %v", sc)
- ccr.cc.handleServiceConfig(sc)
+ c, err := parseServiceConfig(sc)
+ if err != nil {
+ return
+ }
+ if channelz.IsOn() {
+ ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: c})
+ }
+ ccr.curState.ServiceConfig = c
+ ccr.cc.updateResolverState(ccr.curState)
}
-func (ccr *ccResolverWrapper) addChannelzTraceEvent(addrs []resolver.Address) {
- if len(addrs) == 0 && ccr.lastAddressesCount != 0 {
- channelz.AddTraceEvent(ccr.cc.channelzID, &channelz.TraceEventDesc{
- Desc: "Resolver returns an empty address list",
- Severity: channelz.CtWarning,
- })
- } else if len(addrs) != 0 && ccr.lastAddressesCount == 0 {
- var s string
- for i, a := range addrs {
- if a.ServerName != "" {
- s += a.Addr + "(" + a.ServerName + ")"
- } else {
- s += a.Addr
- }
- if i != len(addrs)-1 {
- s += " "
- }
- }
- channelz.AddTraceEvent(ccr.cc.channelzID, &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Resolver returns a non-empty address list (previous one was empty) %q", s),
- Severity: channelz.CtINFO,
- })
+func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) {
+ var updates []string
+ oldSC, oldOK := ccr.curState.ServiceConfig.(*ServiceConfig)
+ newSC, newOK := s.ServiceConfig.(*ServiceConfig)
+ if oldOK != newOK || (oldOK && newOK && oldSC.rawJSONString != newSC.rawJSONString) {
+ updates = append(updates, "service config updated")
}
- ccr.lastAddressesCount = len(addrs)
+ if len(ccr.curState.Addresses) > 0 && len(s.Addresses) == 0 {
+ updates = append(updates, "resolver returned an empty address list")
+ } else if len(ccr.curState.Addresses) == 0 && len(s.Addresses) > 0 {
+ updates = append(updates, "resolver returned new addresses")
+ }
+ channelz.AddTraceEvent(ccr.cc.channelzID, &channelz.TraceEventDesc{
+ Desc: fmt.Sprintf("Resolver state updated: %+v (%v)", s, strings.Join(updates, "; ")),
+ Severity: channelz.CtINFO,
+ })
}
diff --git a/vendor/google.golang.org/grpc/rpc_util.go b/vendor/google.golang.org/grpc/rpc_util.go
index 2a59562..088c3f1 100644
--- a/vendor/google.golang.org/grpc/rpc_util.go
+++ b/vendor/google.golang.org/grpc/rpc_util.go
@@ -694,14 +694,34 @@
return nil
}
+// Information about RPC
type rpcInfo struct {
- failfast bool
+ failfast bool
+ preloaderInfo *compressorInfo
+}
+
+// Information about Preloader
+// Responsible for storing codec, and compressors
+// If stream (s) has context s.Context which stores rpcInfo that has non nil
+// pointers to codec, and compressors, then we can use preparedMsg for Async message prep
+// and reuse marshalled bytes
+type compressorInfo struct {
+ codec baseCodec
+ cp Compressor
+ comp encoding.Compressor
}
type rpcInfoContextKey struct{}
-func newContextWithRPCInfo(ctx context.Context, failfast bool) context.Context {
- return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{failfast: failfast})
+func newContextWithRPCInfo(ctx context.Context, failfast bool, codec baseCodec, cp Compressor, comp encoding.Compressor) context.Context {
+ return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{
+ failfast: failfast,
+ preloaderInfo: &compressorInfo{
+ codec: codec,
+ cp: cp,
+ comp: comp,
+ },
+ })
}
func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) {
diff --git a/vendor/google.golang.org/grpc/server.go b/vendor/google.golang.org/grpc/server.go
index 33272a4..f064b73 100644
--- a/vendor/google.golang.org/grpc/server.go
+++ b/vendor/google.golang.org/grpc/server.go
@@ -42,6 +42,7 @@
"google.golang.org/grpc/grpclog"
"google.golang.org/grpc/internal/binarylog"
"google.golang.org/grpc/internal/channelz"
+ "google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/internal/transport"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/metadata"
@@ -56,6 +57,8 @@
defaultServerMaxSendMessageSize = math.MaxInt32
)
+var statusOK = status.New(codes.OK, "")
+
type methodHandler func(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor UnaryServerInterceptor) (interface{}, error)
// MethodDesc represents an RPC service's method specification.
@@ -86,21 +89,19 @@
// Server is a gRPC server to serve RPC requests.
type Server struct {
- opts options
+ opts serverOptions
mu sync.Mutex // guards following
lis map[net.Listener]bool
- conns map[io.Closer]bool
+ conns map[transport.ServerTransport]bool
serve bool
drain bool
cv *sync.Cond // signaled when connections close for GracefulStop
m map[string]*service // service name -> service info
events trace.EventLog
- quit chan struct{}
- done chan struct{}
- quitOnce sync.Once
- doneOnce sync.Once
+ quit *grpcsync.Event
+ done *grpcsync.Event
channelzRemoveOnce sync.Once
serveWG sync.WaitGroup // counts active Serve goroutines for GracefulStop
@@ -108,7 +109,7 @@
czData *channelzData
}
-type options struct {
+type serverOptions struct {
creds credentials.TransportCredentials
codec baseCodec
cp Compressor
@@ -131,7 +132,7 @@
maxHeaderListSize *uint32
}
-var defaultServerOptions = options{
+var defaultServerOptions = serverOptions{
maxReceiveMessageSize: defaultServerMaxReceiveMessageSize,
maxSendMessageSize: defaultServerMaxSendMessageSize,
connectionTimeout: 120 * time.Second,
@@ -140,7 +141,33 @@
}
// A ServerOption sets options such as credentials, codec and keepalive parameters, etc.
-type ServerOption func(*options)
+type ServerOption interface {
+ apply(*serverOptions)
+}
+
+// EmptyServerOption does not alter the server configuration. It can be embedded
+// in another structure to build custom server options.
+//
+// This API is EXPERIMENTAL.
+type EmptyServerOption struct{}
+
+func (EmptyServerOption) apply(*serverOptions) {}
+
+// funcServerOption wraps a function that modifies serverOptions into an
+// implementation of the ServerOption interface.
+type funcServerOption struct {
+ f func(*serverOptions)
+}
+
+func (fdo *funcServerOption) apply(do *serverOptions) {
+ fdo.f(do)
+}
+
+func newFuncServerOption(f func(*serverOptions)) *funcServerOption {
+ return &funcServerOption{
+ f: f,
+ }
+}
// WriteBufferSize determines how much data can be batched before doing a write on the wire.
// The corresponding memory allocation for this buffer will be twice the size to keep syscalls low.
@@ -148,9 +175,9 @@
// Zero will disable the write buffer such that each write will be on underlying connection.
// Note: A Send call may not directly translate to a write.
func WriteBufferSize(s int) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.writeBufferSize = s
- }
+ })
}
// ReadBufferSize lets you set the size of read buffer, this determines how much data can be read at most
@@ -159,25 +186,25 @@
// Zero will disable read buffer for a connection so data framer can access the underlying
// conn directly.
func ReadBufferSize(s int) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.readBufferSize = s
- }
+ })
}
// InitialWindowSize returns a ServerOption that sets window size for stream.
// The lower bound for window size is 64K and any value smaller than that will be ignored.
func InitialWindowSize(s int32) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.initialWindowSize = s
- }
+ })
}
// InitialConnWindowSize returns a ServerOption that sets window size for a connection.
// The lower bound for window size is 64K and any value smaller than that will be ignored.
func InitialConnWindowSize(s int32) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.initialConnWindowSize = s
- }
+ })
}
// KeepaliveParams returns a ServerOption that sets keepalive and max-age parameters for the server.
@@ -187,25 +214,25 @@
kp.Time = time.Second
}
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.keepaliveParams = kp
- }
+ })
}
// KeepaliveEnforcementPolicy returns a ServerOption that sets keepalive enforcement policy for the server.
func KeepaliveEnforcementPolicy(kep keepalive.EnforcementPolicy) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.keepalivePolicy = kep
- }
+ })
}
// CustomCodec returns a ServerOption that sets a codec for message marshaling and unmarshaling.
//
// This will override any lookups by content-subtype for Codecs registered with RegisterCodec.
func CustomCodec(codec Codec) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.codec = codec
- }
+ })
}
// RPCCompressor returns a ServerOption that sets a compressor for outbound
@@ -216,9 +243,9 @@
//
// Deprecated: use encoding.RegisterCompressor instead.
func RPCCompressor(cp Compressor) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.cp = cp
- }
+ })
}
// RPCDecompressor returns a ServerOption that sets a decompressor for inbound
@@ -227,9 +254,9 @@
//
// Deprecated: use encoding.RegisterCompressor instead.
func RPCDecompressor(dc Decompressor) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.dc = dc
- }
+ })
}
// MaxMsgSize returns a ServerOption to set the max message size in bytes the server can receive.
@@ -243,73 +270,73 @@
// MaxRecvMsgSize returns a ServerOption to set the max message size in bytes the server can receive.
// If this is not set, gRPC uses the default 4MB.
func MaxRecvMsgSize(m int) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.maxReceiveMessageSize = m
- }
+ })
}
// MaxSendMsgSize returns a ServerOption to set the max message size in bytes the server can send.
// If this is not set, gRPC uses the default `math.MaxInt32`.
func MaxSendMsgSize(m int) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.maxSendMessageSize = m
- }
+ })
}
// MaxConcurrentStreams returns a ServerOption that will apply a limit on the number
// of concurrent streams to each ServerTransport.
func MaxConcurrentStreams(n uint32) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.maxConcurrentStreams = n
- }
+ })
}
// Creds returns a ServerOption that sets credentials for server connections.
func Creds(c credentials.TransportCredentials) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.creds = c
- }
+ })
}
// UnaryInterceptor returns a ServerOption that sets the UnaryServerInterceptor for the
// server. Only one unary interceptor can be installed. The construction of multiple
// interceptors (e.g., chaining) can be implemented at the caller.
func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
if o.unaryInt != nil {
panic("The unary server interceptor was already set and may not be reset.")
}
o.unaryInt = i
- }
+ })
}
// StreamInterceptor returns a ServerOption that sets the StreamServerInterceptor for the
// server. Only one stream interceptor can be installed.
func StreamInterceptor(i StreamServerInterceptor) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
if o.streamInt != nil {
panic("The stream server interceptor was already set and may not be reset.")
}
o.streamInt = i
- }
+ })
}
// InTapHandle returns a ServerOption that sets the tap handle for all the server
// transport to be created. Only one can be installed.
func InTapHandle(h tap.ServerInHandle) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
if o.inTapHandle != nil {
panic("The tap handle was already set and may not be reset.")
}
o.inTapHandle = h
- }
+ })
}
// StatsHandler returns a ServerOption that sets the stats handler for the server.
func StatsHandler(h stats.Handler) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.statsHandler = h
- }
+ })
}
// UnknownServiceHandler returns a ServerOption that allows for adding a custom
@@ -319,7 +346,7 @@
// The handling function has full access to the Context of the request and the
// stream, and the invocation bypasses interceptors.
func UnknownServiceHandler(streamHandler StreamHandler) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.unknownStreamDesc = &StreamDesc{
StreamName: "unknown_service_handler",
Handler: streamHandler,
@@ -327,7 +354,7 @@
ClientStreams: true,
ServerStreams: true,
}
- }
+ })
}
// ConnectionTimeout returns a ServerOption that sets the timeout for
@@ -337,17 +364,17 @@
//
// This API is EXPERIMENTAL.
func ConnectionTimeout(d time.Duration) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.connectionTimeout = d
- }
+ })
}
// MaxHeaderListSize returns a ServerOption that sets the max (uncompressed) size
// of header list that the server is prepared to accept.
func MaxHeaderListSize(s uint32) ServerOption {
- return func(o *options) {
+ return newFuncServerOption(func(o *serverOptions) {
o.maxHeaderListSize = &s
- }
+ })
}
// NewServer creates a gRPC server which has no service registered and has not
@@ -355,15 +382,15 @@
func NewServer(opt ...ServerOption) *Server {
opts := defaultServerOptions
for _, o := range opt {
- o(&opts)
+ o.apply(&opts)
}
s := &Server{
lis: make(map[net.Listener]bool),
opts: opts,
- conns: make(map[io.Closer]bool),
+ conns: make(map[transport.ServerTransport]bool),
m: make(map[string]*service),
- quit: make(chan struct{}),
- done: make(chan struct{}),
+ quit: grpcsync.NewEvent(),
+ done: grpcsync.NewEvent(),
czData: new(channelzData),
}
s.cv = sync.NewCond(&s.mu)
@@ -530,11 +557,9 @@
s.serveWG.Add(1)
defer func() {
s.serveWG.Done()
- select {
- // Stop or GracefulStop called; block until done and return nil.
- case <-s.quit:
- <-s.done
- default:
+ if s.quit.HasFired() {
+ // Stop or GracefulStop called; block until done and return nil.
+ <-s.done.Done()
}
}()
@@ -577,7 +602,7 @@
timer := time.NewTimer(tempDelay)
select {
case <-timer.C:
- case <-s.quit:
+ case <-s.quit.Done():
timer.Stop()
return nil
}
@@ -587,10 +612,8 @@
s.printf("done serving; Accept = %v", err)
s.mu.Unlock()
- select {
- case <-s.quit:
+ if s.quit.HasFired() {
return nil
- default:
}
return err
}
@@ -611,29 +634,26 @@
// handleRawConn forks a goroutine to handle a just-accepted connection that
// has not had any I/O performed on it yet.
func (s *Server) handleRawConn(rawConn net.Conn) {
+ if s.quit.HasFired() {
+ rawConn.Close()
+ return
+ }
rawConn.SetDeadline(time.Now().Add(s.opts.connectionTimeout))
conn, authInfo, err := s.useTransportAuthenticator(rawConn)
if err != nil {
- s.mu.Lock()
- s.errorf("ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err)
- s.mu.Unlock()
- grpclog.Warningf("grpc: Server.Serve failed to complete security handshake from %q: %v", rawConn.RemoteAddr(), err)
- // If serverHandshake returns ErrConnDispatched, keep rawConn open.
+ // ErrConnDispatched means that the connection was dispatched away from
+ // gRPC; those connections should be left open.
if err != credentials.ErrConnDispatched {
+ s.mu.Lock()
+ s.errorf("ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err)
+ s.mu.Unlock()
+ grpclog.Warningf("grpc: Server.Serve failed to complete security handshake from %q: %v", rawConn.RemoteAddr(), err)
rawConn.Close()
}
rawConn.SetDeadline(time.Time{})
return
}
- s.mu.Lock()
- if s.conns == nil {
- s.mu.Unlock()
- conn.Close()
- return
- }
- s.mu.Unlock()
-
// Finish handshaking (HTTP2)
st := s.newHTTP2Transport(conn, authInfo)
if st == nil {
@@ -741,6 +761,9 @@
// traceInfo returns a traceInfo and associates it with stream, if tracing is enabled.
// If tracing is not enabled, it returns nil.
func (s *Server) traceInfo(st transport.ServerTransport, stream *transport.Stream) (trInfo *traceInfo) {
+ if !EnableTracing {
+ return nil
+ }
tr, ok := trace.FromContext(stream.Context())
if !ok {
return nil
@@ -748,37 +771,38 @@
trInfo = &traceInfo{
tr: tr,
+ firstLine: firstLine{
+ client: false,
+ remoteAddr: st.RemoteAddr(),
+ },
}
- trInfo.firstLine.client = false
- trInfo.firstLine.remoteAddr = st.RemoteAddr()
-
if dl, ok := stream.Context().Deadline(); ok {
trInfo.firstLine.deadline = time.Until(dl)
}
return trInfo
}
-func (s *Server) addConn(c io.Closer) bool {
+func (s *Server) addConn(st transport.ServerTransport) bool {
s.mu.Lock()
defer s.mu.Unlock()
if s.conns == nil {
- c.Close()
+ st.Close()
return false
}
if s.drain {
// Transport added after we drained our existing conns: drain it
// immediately.
- c.(transport.ServerTransport).Drain()
+ st.Drain()
}
- s.conns[c] = true
+ s.conns[st] = true
return true
}
-func (s *Server) removeConn(c io.Closer) {
+func (s *Server) removeConn(st transport.ServerTransport) {
s.mu.Lock()
defer s.mu.Unlock()
if s.conns != nil {
- delete(s.conns, c)
+ delete(s.conns, st)
s.cv.Broadcast()
}
}
@@ -859,7 +883,6 @@
}
if trInfo != nil {
defer trInfo.tr.Finish()
- trInfo.firstLine.client = false
trInfo.tr.LazyLog(&trInfo.firstLine, false)
defer func() {
if err != nil && err != io.EOF {
@@ -951,10 +974,11 @@
}
if sh != nil {
sh.HandleRPC(stream.Context(), &stats.InPayload{
- RecvTime: time.Now(),
- Payload: v,
- Data: d,
- Length: len(d),
+ RecvTime: time.Now(),
+ Payload: v,
+ WireLength: payInfo.wireLength,
+ Data: d,
+ Length: len(d),
})
}
if binlog != nil {
@@ -1050,7 +1074,7 @@
// TODO: Should we be logging if writing status failed here, like above?
// Should the logging be in WriteStatus? Should we ignore the WriteStatus
// error or allow the stats handler to see it?
- err = t.WriteStatus(stream, status.New(codes.OK, ""))
+ err = t.WriteStatus(stream, statusOK)
if binlog != nil {
binlog.Log(&binarylog.ServerTrailer{
Trailer: stream.Trailer(),
@@ -1208,7 +1232,7 @@
ss.trInfo.tr.LazyLog(stringer("OK"), false)
ss.mu.Unlock()
}
- err = t.WriteStatus(ss.s, status.New(codes.OK, ""))
+ err = t.WriteStatus(ss.s, statusOK)
if ss.binlog != nil {
ss.binlog.Log(&binarylog.ServerTrailer{
Trailer: ss.s.Trailer(),
@@ -1245,7 +1269,8 @@
service := sm[:pos]
method := sm[pos+1:]
- if srv, ok := s.m[service]; ok {
+ srv, knownService := s.m[service]
+ if knownService {
if md, ok := srv.md[method]; ok {
s.processUnaryRPC(t, stream, srv, md, trInfo)
return
@@ -1260,11 +1285,16 @@
s.processStreamingRPC(t, stream, nil, unknownDesc, trInfo)
return
}
+ var errDesc string
+ if !knownService {
+ errDesc = fmt.Sprintf("unknown service %v", service)
+ } else {
+ errDesc = fmt.Sprintf("unknown method %v for service %v", method, service)
+ }
if trInfo != nil {
- trInfo.tr.LazyLog(&fmtStringer{"Unknown service %v", []interface{}{service}}, true)
+ trInfo.tr.LazyPrintf("%s", errDesc)
trInfo.tr.SetError()
}
- errDesc := fmt.Sprintf("unknown service %v", service)
if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
if trInfo != nil {
trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
@@ -1319,15 +1349,11 @@
// pending RPCs on the client side will get notified by connection
// errors.
func (s *Server) Stop() {
- s.quitOnce.Do(func() {
- close(s.quit)
- })
+ s.quit.Fire()
defer func() {
s.serveWG.Wait()
- s.doneOnce.Do(func() {
- close(s.done)
- })
+ s.done.Fire()
}()
s.channelzRemoveOnce.Do(func() {
@@ -1364,15 +1390,8 @@
// accepting new connections and RPCs and blocks until all the pending RPCs are
// finished.
func (s *Server) GracefulStop() {
- s.quitOnce.Do(func() {
- close(s.quit)
- })
-
- defer func() {
- s.doneOnce.Do(func() {
- close(s.done)
- })
- }()
+ s.quit.Fire()
+ defer s.done.Fire()
s.channelzRemoveOnce.Do(func() {
if channelz.IsOn() {
@@ -1390,8 +1409,8 @@
}
s.lis = nil
if !s.drain {
- for c := range s.conns {
- c.(transport.ServerTransport).Drain()
+ for st := range s.conns {
+ st.Drain()
}
s.drain = true
}
diff --git a/vendor/google.golang.org/grpc/service_config.go b/vendor/google.golang.org/grpc/service_config.go
index 162857e..686ad7b 100644
--- a/vendor/google.golang.org/grpc/service_config.go
+++ b/vendor/google.golang.org/grpc/service_config.go
@@ -25,8 +25,11 @@
"strings"
"time"
+ "google.golang.org/grpc/balancer"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/grpclog"
+ "google.golang.org/grpc/internal"
+ "google.golang.org/grpc/serviceconfig"
)
const maxInt = int(^uint(0) >> 1)
@@ -61,6 +64,11 @@
retryPolicy *retryPolicy
}
+type lbConfig struct {
+ name string
+ cfg serviceconfig.LoadBalancingConfig
+}
+
// ServiceConfig is provided by the service provider and contains parameters for how
// clients that connect to the service should behave.
//
@@ -68,10 +76,18 @@
// through name resolver, as specified here
// https://github.com/grpc/grpc/blob/master/doc/service_config.md
type ServiceConfig struct {
- // LB is the load balancer the service providers recommends. The balancer specified
- // via grpc.WithBalancer will override this.
+ serviceconfig.Config
+
+ // LB is the load balancer the service providers recommends. The balancer
+ // specified via grpc.WithBalancer will override this. This is deprecated;
+ // lbConfigs is preferred. If lbConfig and LB are both present, lbConfig
+ // will be used.
LB *string
+ // lbConfig is the service config's load balancing configuration. If
+ // lbConfig and LB are both present, lbConfig will be used.
+ lbConfig *lbConfig
+
// Methods contains a map for the methods in this service. If there is an
// exact match for a method (i.e. /service/method) in the map, use the
// corresponding MethodConfig. If there's no exact match, look for the
@@ -99,6 +115,9 @@
// healthCheckConfig must be set as one of the requirement to enable LB channel
// health check.
healthCheckConfig *healthCheckConfig
+ // rawJSONString stores service config json string that get parsed into
+ // this service config struct.
+ rawJSONString string
}
// healthCheckConfig defines the go-native version of the LB channel health check config.
@@ -230,34 +249,80 @@
RetryPolicy *jsonRetryPolicy
}
+type loadBalancingConfig map[string]json.RawMessage
+
// TODO(lyuxuan): delete this struct after cleaning up old service config implementation.
type jsonSC struct {
LoadBalancingPolicy *string
+ LoadBalancingConfig *[]loadBalancingConfig
MethodConfig *[]jsonMC
RetryThrottling *retryThrottlingPolicy
HealthCheckConfig *healthCheckConfig
}
-func parseServiceConfig(js string) (ServiceConfig, error) {
+func init() {
+ internal.ParseServiceConfig = func(sc string) (interface{}, error) {
+ return parseServiceConfig(sc)
+ }
+}
+
+func parseServiceConfig(js string) (*ServiceConfig, error) {
if len(js) == 0 {
- return ServiceConfig{}, fmt.Errorf("no JSON service config provided")
+ return nil, fmt.Errorf("no JSON service config provided")
}
var rsc jsonSC
err := json.Unmarshal([]byte(js), &rsc)
if err != nil {
grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
- return ServiceConfig{}, err
+ return nil, err
}
sc := ServiceConfig{
LB: rsc.LoadBalancingPolicy,
Methods: make(map[string]MethodConfig),
retryThrottling: rsc.RetryThrottling,
healthCheckConfig: rsc.HealthCheckConfig,
+ rawJSONString: js,
}
- if rsc.MethodConfig == nil {
- return sc, nil
+ if rsc.LoadBalancingConfig != nil {
+ for i, lbcfg := range *rsc.LoadBalancingConfig {
+ if len(lbcfg) != 1 {
+ err := fmt.Errorf("invalid loadBalancingConfig: entry %v does not contain exactly 1 policy/config pair: %q", i, lbcfg)
+ grpclog.Warningf(err.Error())
+ return nil, err
+ }
+ var name string
+ var jsonCfg json.RawMessage
+ for name, jsonCfg = range lbcfg {
+ }
+ builder := balancer.Get(name)
+ if builder == nil {
+ continue
+ }
+ sc.lbConfig = &lbConfig{name: name}
+ if parser, ok := builder.(balancer.ConfigParser); ok {
+ var err error
+ sc.lbConfig.cfg, err = parser.ParseConfig(jsonCfg)
+ if err != nil {
+ return nil, fmt.Errorf("error parsing loadBalancingConfig for policy %q: %v", name, err)
+ }
+ } else if string(jsonCfg) != "{}" {
+ grpclog.Warningf("non-empty balancer configuration %q, but balancer does not implement ParseConfig", string(jsonCfg))
+ }
+ break
+ }
+ if sc.lbConfig == nil {
+ // We had a loadBalancingConfig field but did not encounter a
+ // supported policy. The config is considered invalid in this
+ // case.
+ err := fmt.Errorf("invalid loadBalancingConfig: no supported policies found")
+ grpclog.Warningf(err.Error())
+ return nil, err
+ }
}
+ if rsc.MethodConfig == nil {
+ return &sc, nil
+ }
for _, m := range *rsc.MethodConfig {
if m.Name == nil {
continue
@@ -265,7 +330,7 @@
d, err := parseDuration(m.Timeout)
if err != nil {
grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
- return ServiceConfig{}, err
+ return nil, err
}
mc := MethodConfig{
@@ -274,7 +339,7 @@
}
if mc.retryPolicy, err = convertRetryPolicy(m.RetryPolicy); err != nil {
grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
- return ServiceConfig{}, err
+ return nil, err
}
if m.MaxRequestMessageBytes != nil {
if *m.MaxRequestMessageBytes > int64(maxInt) {
@@ -298,14 +363,14 @@
}
if sc.retryThrottling != nil {
- if sc.retryThrottling.MaxTokens <= 0 ||
- sc.retryThrottling.MaxTokens >= 1000 ||
- sc.retryThrottling.TokenRatio <= 0 {
- // Illegal throttling config; disable throttling.
- sc.retryThrottling = nil
+ if mt := sc.retryThrottling.MaxTokens; mt <= 0 || mt > 1000 {
+ return nil, fmt.Errorf("invalid retry throttling config: maxTokens (%v) out of range (0, 1000]", mt)
+ }
+ if tr := sc.retryThrottling.TokenRatio; tr <= 0 {
+ return nil, fmt.Errorf("invalid retry throttling config: tokenRatio (%v) may not be negative", tr)
}
}
- return sc, nil
+ return &sc, nil
}
func convertRetryPolicy(jrp *jsonRetryPolicy) (p *retryPolicy, err error) {
diff --git a/vendor/google.golang.org/grpc/serviceconfig/serviceconfig.go b/vendor/google.golang.org/grpc/serviceconfig/serviceconfig.go
new file mode 100644
index 0000000..53b2787
--- /dev/null
+++ b/vendor/google.golang.org/grpc/serviceconfig/serviceconfig.go
@@ -0,0 +1,48 @@
+/*
+ *
+ * Copyright 2019 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.
+ *
+ */
+
+// Package serviceconfig defines types and methods for operating on gRPC
+// service configs.
+//
+// This package is EXPERIMENTAL.
+package serviceconfig
+
+import (
+ "google.golang.org/grpc/internal"
+)
+
+// Config represents an opaque data structure holding a service config.
+type Config interface {
+ isConfig()
+}
+
+// LoadBalancingConfig represents an opaque data structure holding a load
+// balancer config.
+type LoadBalancingConfig interface {
+ isLoadBalancingConfig()
+}
+
+// Parse parses the JSON service config provided into an internal form or
+// returns an error if the config is invalid.
+func Parse(ServiceConfigJSON string) (Config, error) {
+ c, err := internal.ParseServiceConfig(ServiceConfigJSON)
+ if err != nil {
+ return nil, err
+ }
+ return c.(Config), err
+}
diff --git a/vendor/google.golang.org/grpc/stats/stats.go b/vendor/google.golang.org/grpc/stats/stats.go
index 84f77da..f3f593c 100644
--- a/vendor/google.golang.org/grpc/stats/stats.go
+++ b/vendor/google.golang.org/grpc/stats/stats.go
@@ -27,6 +27,8 @@
"context"
"net"
"time"
+
+ "google.golang.org/grpc/metadata"
)
// RPCStats contains stats information about RPCs.
@@ -172,6 +174,9 @@
BeginTime time.Time
// EndTime is the time when the RPC ends.
EndTime time.Time
+ // Trailer contains the trailer metadata received from the server. This
+ // field is only valid if this End is from the client side.
+ Trailer metadata.MD
// Error is the error the RPC ended with. It is an error generated from
// status.Status and can be converted back to status.Status using
// status.FromError if non-nil.
diff --git a/vendor/google.golang.org/grpc/status/status.go b/vendor/google.golang.org/grpc/status/status.go
index ed36681..a1348e9 100644
--- a/vendor/google.golang.org/grpc/status/status.go
+++ b/vendor/google.golang.org/grpc/status/status.go
@@ -36,8 +36,15 @@
"github.com/golang/protobuf/ptypes"
spb "google.golang.org/genproto/googleapis/rpc/status"
"google.golang.org/grpc/codes"
+ "google.golang.org/grpc/internal"
)
+func init() {
+ internal.StatusRawProto = statusRawProto
+}
+
+func statusRawProto(s *Status) *spb.Status { return s.s }
+
// statusError is an alias of a status proto. It implements error and Status,
// and a nil statusError should never be returned by this package.
type statusError spb.Status
@@ -51,6 +58,17 @@
return &Status{s: (*spb.Status)(se)}
}
+// Is implements future error.Is functionality.
+// A statusError is equivalent if the code and message are identical.
+func (se *statusError) Is(target error) bool {
+ tse, ok := target.(*statusError)
+ if !ok {
+ return false
+ }
+
+ return proto.Equal((*spb.Status)(se), (*spb.Status)(tse))
+}
+
// Status represents an RPC status code, message, and details. It is immutable
// and should be created with New, Newf, or FromProto.
type Status struct {
@@ -125,7 +143,7 @@
// Status is returned with codes.Unknown and the original error message.
func FromError(err error) (s *Status, ok bool) {
if err == nil {
- return &Status{s: &spb.Status{Code: int32(codes.OK)}}, true
+ return nil, true
}
if se, ok := err.(interface {
GRPCStatus() *Status
@@ -199,7 +217,7 @@
func FromContextError(err error) *Status {
switch err {
case nil:
- return New(codes.OK, "")
+ return nil
case context.DeadlineExceeded:
return New(codes.DeadlineExceeded, err.Error())
case context.Canceled:
diff --git a/vendor/google.golang.org/grpc/stream.go b/vendor/google.golang.org/grpc/stream.go
index ccf996b..134a624 100644
--- a/vendor/google.golang.org/grpc/stream.go
+++ b/vendor/google.golang.org/grpc/stream.go
@@ -30,9 +30,9 @@
"golang.org/x/net/trace"
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/codes"
- "google.golang.org/grpc/connectivity"
"google.golang.org/grpc/encoding"
"google.golang.org/grpc/grpclog"
+ "google.golang.org/grpc/internal/balancerload"
"google.golang.org/grpc/internal/binarylog"
"google.golang.org/grpc/internal/channelz"
"google.golang.org/grpc/internal/grpcrand"
@@ -230,17 +230,21 @@
if c.creds != nil {
callHdr.Creds = c.creds
}
- var trInfo traceInfo
+ var trInfo *traceInfo
if EnableTracing {
- trInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method)
- trInfo.firstLine.client = true
+ trInfo = &traceInfo{
+ tr: trace.New("grpc.Sent."+methodFamily(method), method),
+ firstLine: firstLine{
+ client: true,
+ },
+ }
if deadline, ok := ctx.Deadline(); ok {
trInfo.firstLine.deadline = time.Until(deadline)
}
trInfo.tr.LazyLog(&trInfo.firstLine, false)
ctx = trace.NewContext(ctx, trInfo.tr)
}
- ctx = newContextWithRPCInfo(ctx, c.failFast)
+ ctx = newContextWithRPCInfo(ctx, c.failFast, c.codec, cp, comp)
sh := cc.dopts.copts.StatsHandler
var beginTime time.Time
if sh != nil {
@@ -323,13 +327,23 @@
return cs, nil
}
-func (cs *clientStream) newAttemptLocked(sh stats.Handler, trInfo traceInfo) error {
- cs.attempt = &csAttempt{
+// newAttemptLocked creates a new attempt with a transport.
+// If it succeeds, then it replaces clientStream's attempt with this new attempt.
+func (cs *clientStream) newAttemptLocked(sh stats.Handler, trInfo *traceInfo) (retErr error) {
+ newAttempt := &csAttempt{
cs: cs,
dc: cs.cc.dopts.dc,
statsHandler: sh,
trInfo: trInfo,
}
+ defer func() {
+ if retErr != nil {
+ // This attempt is not set in the clientStream, so it's finish won't
+ // be called. Call it here for stats and trace in case they are not
+ // nil.
+ newAttempt.finish(retErr)
+ }
+ }()
if err := cs.ctx.Err(); err != nil {
return toRPCErr(err)
@@ -338,8 +352,12 @@
if err != nil {
return err
}
- cs.attempt.t = t
- cs.attempt.done = done
+ if trInfo != nil {
+ trInfo.firstLine.SetRemoteAddr(t.RemoteAddr())
+ }
+ newAttempt.t = t
+ newAttempt.done = done
+ cs.attempt = newAttempt
return nil
}
@@ -388,11 +406,18 @@
serverHeaderBinlogged bool
mu sync.Mutex
- firstAttempt bool // if true, transparent retry is valid
- numRetries int // exclusive of transparent retry attempt(s)
- numRetriesSincePushback int // retries since pushback; to reset backoff
- finished bool // TODO: replace with atomic cmpxchg or sync.Once?
- attempt *csAttempt // the active client stream attempt
+ firstAttempt bool // if true, transparent retry is valid
+ numRetries int // exclusive of transparent retry attempt(s)
+ numRetriesSincePushback int // retries since pushback; to reset backoff
+ finished bool // TODO: replace with atomic cmpxchg or sync.Once?
+ // attempt is the active client stream attempt.
+ // The only place where it is written is the newAttemptLocked method and this method never writes nil.
+ // So, attempt can be nil only inside newClientStream function when clientStream is first created.
+ // One of the first things done after clientStream's creation, is to call newAttemptLocked which either
+ // assigns a non nil value to the attempt or returns an error. If an error is returned from newAttemptLocked,
+ // then newClientStream calls finish on the clientStream and returns. So, finish method is the only
+ // place where we need to check if the attempt is nil.
+ attempt *csAttempt
// TODO(hedging): hedging will have multiple attempts simultaneously.
committed bool // active attempt committed for retry?
buffer []func(a *csAttempt) error // operations to replay on retry
@@ -414,9 +439,10 @@
decompSet bool
mu sync.Mutex // guards trInfo.tr
+ // trInfo may be nil (if EnableTracing is false).
// trInfo.tr is set when created (if EnableTracing is true),
// and cleared when the finish method is called.
- trInfo traceInfo
+ trInfo *traceInfo
statsHandler stats.Handler
}
@@ -449,8 +475,8 @@
if cs.attempt.s != nil {
<-cs.attempt.s.Done()
}
- if cs.firstAttempt && !cs.callInfo.failFast && (cs.attempt.s == nil || cs.attempt.s.Unprocessed()) {
- // First attempt, wait-for-ready, stream unprocessed: transparently retry.
+ if cs.firstAttempt && (cs.attempt.s == nil || cs.attempt.s.Unprocessed()) {
+ // First attempt, stream unprocessed: transparently retry.
cs.firstAttempt = false
return nil
}
@@ -540,7 +566,7 @@
cs.commitAttemptLocked()
return err
}
- if err := cs.newAttemptLocked(nil, traceInfo{}); err != nil {
+ if err := cs.newAttemptLocked(nil, nil); err != nil {
return err
}
if lastErr = cs.replayBufferLocked(); lastErr == nil {
@@ -668,15 +694,13 @@
if !cs.desc.ClientStreams {
cs.sentLast = true
}
- data, err := encode(cs.codec, m)
+
+ // load hdr, payload, data
+ hdr, payload, data, err := prepareMsg(m, cs.codec, cs.cp, cs.comp)
if err != nil {
return err
}
- compData, err := compress(data, cs.cp, cs.comp)
- if err != nil {
- return err
- }
- hdr, payload := msgHeader(data, compData)
+
// TODO(dfawley): should we be checking len(data) instead?
if len(payload) > *cs.callInfo.maxSendMessageSize {
return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.callInfo.maxSendMessageSize)
@@ -799,11 +823,11 @@
}
if cs.attempt != nil {
cs.attempt.finish(err)
- }
- // after functions all rely upon having a stream.
- if cs.attempt.s != nil {
- for _, o := range cs.opts {
- o.after(cs.callInfo)
+ // after functions all rely upon having a stream.
+ if cs.attempt.s != nil {
+ for _, o := range cs.opts {
+ o.after(cs.callInfo)
+ }
}
}
cs.cancel()
@@ -811,7 +835,7 @@
func (a *csAttempt) sendMsg(m interface{}, hdr, payld, data []byte) error {
cs := a.cs
- if EnableTracing {
+ if a.trInfo != nil {
a.mu.Lock()
if a.trInfo.tr != nil {
a.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
@@ -868,7 +892,7 @@
}
return toRPCErr(err)
}
- if EnableTracing {
+ if a.trInfo != nil {
a.mu.Lock()
if a.trInfo.tr != nil {
a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
@@ -881,8 +905,9 @@
RecvTime: time.Now(),
Payload: m,
// TODO truncate large payload.
- Data: payInfo.uncompressedBytes,
- Length: len(payInfo.uncompressedBytes),
+ Data: payInfo.uncompressedBytes,
+ WireLength: payInfo.wireLength,
+ Length: len(payInfo.uncompressedBytes),
})
}
if channelz.IsOn() {
@@ -915,22 +940,23 @@
// Ending a stream with EOF indicates a success.
err = nil
}
+ var tr metadata.MD
if a.s != nil {
a.t.CloseStream(a.s, err)
+ tr = a.s.Trailer()
}
if a.done != nil {
br := false
- var tr metadata.MD
if a.s != nil {
br = a.s.BytesReceived()
- tr = a.s.Trailer()
}
a.done(balancer.DoneInfo{
Err: err,
Trailer: tr,
BytesSent: a.s != nil,
BytesReceived: br,
+ ServerLoad: balancerload.Parse(tr),
})
}
if a.statsHandler != nil {
@@ -938,11 +964,12 @@
Client: true,
BeginTime: a.cs.beginTime,
EndTime: time.Now(),
+ Trailer: tr,
Error: err,
}
a.statsHandler.HandleRPC(a.cs.ctx, end)
}
- if a.trInfo.tr != nil {
+ if a.trInfo != nil && a.trInfo.tr != nil {
if err == nil {
a.trInfo.tr.LazyPrintf("RPC: [OK]")
} else {
@@ -955,19 +982,18 @@
a.mu.Unlock()
}
-func (ac *addrConn) newClientStream(ctx context.Context, desc *StreamDesc, method string, t transport.ClientTransport, opts ...CallOption) (_ ClientStream, err error) {
- ac.mu.Lock()
- if ac.transport != t {
- ac.mu.Unlock()
- return nil, status.Error(codes.Canceled, "the provided transport is no longer valid to use")
- }
- // transition to CONNECTING state when an attempt starts
- if ac.state != connectivity.Connecting {
- ac.updateConnectivityState(connectivity.Connecting)
- ac.cc.handleSubConnStateChange(ac.acbw, ac.state)
- }
- ac.mu.Unlock()
-
+// newClientStream creates a ClientStream with the specified transport, on the
+// given addrConn.
+//
+// It's expected that the given transport is either the same one in addrConn, or
+// is already closed. To avoid race, transport is specified separately, instead
+// of using ac.transpot.
+//
+// Main difference between this and ClientConn.NewStream:
+// - no retry
+// - no service config (or wait for service config)
+// - no tracing or stats
+func newNonRetryClientStream(ctx context.Context, desc *StreamDesc, method string, t transport.ClientTransport, ac *addrConn, opts ...CallOption) (_ ClientStream, err error) {
if t == nil {
// TODO: return RPC error here?
return nil, errors.New("transport provided is nil")
@@ -975,14 +1001,6 @@
// defaultCallInfo contains unnecessary info(i.e. failfast, maxRetryRPCBufferSize), so we just initialize an empty struct.
c := &callInfo{}
- for _, o := range opts {
- if err := o.before(c); err != nil {
- return nil, toRPCErr(err)
- }
- }
- c.maxReceiveMessageSize = getMaxSize(nil, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize)
- c.maxSendMessageSize = getMaxSize(nil, c.maxSendMessageSize, defaultServerMaxSendMessageSize)
-
// Possible context leak:
// The cancel function for the child context we create will only be called
// when RecvMsg returns a non-nil error, if the ClientConn is closed, or if
@@ -995,6 +1013,13 @@
}
}()
+ for _, o := range opts {
+ if err := o.before(c); err != nil {
+ return nil, toRPCErr(err)
+ }
+ }
+ c.maxReceiveMessageSize = getMaxSize(nil, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize)
+ c.maxSendMessageSize = getMaxSize(nil, c.maxSendMessageSize, defaultServerMaxSendMessageSize)
if err := setCallInfoCodec(c); err != nil {
return nil, err
}
@@ -1027,6 +1052,7 @@
callHdr.Creds = c.creds
}
+ // Use a special addrConnStream to avoid retry.
as := &addrConnStream{
callHdr: callHdr,
ac: ac,
@@ -1138,15 +1164,13 @@
if !as.desc.ClientStreams {
as.sentLast = true
}
- data, err := encode(as.codec, m)
+
+ // load hdr, payload, data
+ hdr, payld, _, err := prepareMsg(m, as.codec, as.cp, as.comp)
if err != nil {
return err
}
- compData, err := compress(data, as.cp, as.comp)
- if err != nil {
- return err
- }
- hdr, payld := msgHeader(data, compData)
+
// TODO(dfawley): should we be checking len(data) instead?
if len(payld) > *as.callInfo.maxSendMessageSize {
return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payld), *as.callInfo.maxSendMessageSize)
@@ -1383,15 +1407,13 @@
ss.t.IncrMsgSent()
}
}()
- data, err := encode(ss.codec, m)
+
+ // load hdr, payload, data
+ hdr, payload, data, err := prepareMsg(m, ss.codec, ss.cp, ss.comp)
if err != nil {
return err
}
- compData, err := compress(data, ss.cp, ss.comp)
- if err != nil {
- return err
- }
- hdr, payload := msgHeader(data, compData)
+
// TODO(dfawley): should we be checking len(data) instead?
if len(payload) > ss.maxSendMessageSize {
return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), ss.maxSendMessageSize)
@@ -1466,8 +1488,9 @@
RecvTime: time.Now(),
Payload: m,
// TODO truncate large payload.
- Data: payInfo.uncompressedBytes,
- Length: len(payInfo.uncompressedBytes),
+ Data: payInfo.uncompressedBytes,
+ WireLength: payInfo.wireLength,
+ Length: len(payInfo.uncompressedBytes),
})
}
if ss.binlog != nil {
@@ -1483,3 +1506,24 @@
func MethodFromServerStream(stream ServerStream) (string, bool) {
return Method(stream.Context())
}
+
+// prepareMsg returns the hdr, payload and data
+// using the compressors passed or using the
+// passed preparedmsg
+func prepareMsg(m interface{}, codec baseCodec, cp Compressor, comp encoding.Compressor) (hdr, payload, data []byte, err error) {
+ if preparedMsg, ok := m.(*PreparedMsg); ok {
+ return preparedMsg.hdr, preparedMsg.payload, preparedMsg.encodedData, nil
+ }
+ // The input interface is not a prepared msg.
+ // Marshal and Compress the data at this point
+ data, err = encode(codec, m)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ compData, err := compress(data, cp, comp)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ hdr, payload = msgHeader(data, compData)
+ return hdr, payload, data, nil
+}
diff --git a/vendor/google.golang.org/grpc/trace.go b/vendor/google.golang.org/grpc/trace.go
index c1c96de..0a57b99 100644
--- a/vendor/google.golang.org/grpc/trace.go
+++ b/vendor/google.golang.org/grpc/trace.go
@@ -24,6 +24,7 @@
"io"
"net"
"strings"
+ "sync"
"time"
"golang.org/x/net/trace"
@@ -53,13 +54,25 @@
}
// firstLine is the first line of an RPC trace.
+// It may be mutated after construction; remoteAddr specifically may change
+// during client-side use.
type firstLine struct {
+ mu sync.Mutex
client bool // whether this is a client (outgoing) RPC
remoteAddr net.Addr
deadline time.Duration // may be zero
}
+func (f *firstLine) SetRemoteAddr(addr net.Addr) {
+ f.mu.Lock()
+ f.remoteAddr = addr
+ f.mu.Unlock()
+}
+
func (f *firstLine) String() string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
var line bytes.Buffer
io.WriteString(&line, "RPC: ")
if f.client {
diff --git a/vendor/google.golang.org/grpc/version.go b/vendor/google.golang.org/grpc/version.go
index 035b939..483ef89 100644
--- a/vendor/google.golang.org/grpc/version.go
+++ b/vendor/google.golang.org/grpc/version.go
@@ -19,4 +19,4 @@
package grpc
// Version is the current grpc version.
-const Version = "1.19.1"
+const Version = "1.24.0"
diff --git a/vendor/google.golang.org/grpc/vet.sh b/vendor/google.golang.org/grpc/vet.sh
index 7209aa5..2d79b1c 100644
--- a/vendor/google.golang.org/grpc/vet.sh
+++ b/vendor/google.golang.org/grpc/vet.sh
@@ -67,7 +67,7 @@
fi
# - Ensure all source files contain a copyright message.
-git ls-files "*.go" | xargs grep -L "\(Copyright [0-9]\{4,\} gRPC authors\)\|DO NOT EDIT" 2>&1 | fail_on_output
+(! git grep -L "\(Copyright [0-9]\{4,\} gRPC authors\)\|DO NOT EDIT" -- '*.go')
# - Make sure all tests in grpc and grpc/test use leakcheck via Teardown.
(! grep 'func Test[^(]' *_test.go)
@@ -75,10 +75,10 @@
# - Do not import math/rand for real library code. Use internal/grpcrand for
# thread safety.
-git ls-files "*.go" | xargs grep -l '"math/rand"' 2>&1 | (! grep -v '^examples\|^stress\|grpcrand')
+git grep -l '"math/rand"' -- "*.go" 2>&1 | (! grep -v '^examples\|^stress\|grpcrand\|wrr_test')
# - Ensure all ptypes proto packages are renamed when importing.
-git ls-files "*.go" | (! xargs grep "\(import \|^\s*\)\"github.com/golang/protobuf/ptypes/")
+(! git grep "\(import \|^\s*\)\"github.com/golang/protobuf/ptypes/" -- "*.go")
# - Check imports that are illegal in appengine (until Go 1.11).
# TODO: Remove when we drop Go 1.10 support
@@ -86,9 +86,9 @@
# - gofmt, goimports, golint (with exceptions for generated code), go vet.
gofmt -s -d -l . 2>&1 | fail_on_output
-goimports -l . 2>&1 | fail_on_output
+goimports -l . 2>&1 | (! grep -vE "(_mock|\.pb)\.go") | fail_on_output
golint ./... 2>&1 | (! grep -vE "(_mock|\.pb)\.go:")
-go tool vet -all .
+go vet -all .
# - Check that generated proto files are up to date.
if [[ -z "${VET_SKIP_PROTO}" ]]; then
@@ -105,17 +105,29 @@
fi
# - Collection of static analysis checks
-# TODO(menghanl): fix errors in transport_test.
+# TODO(dfawley): don't use deprecated functions in examples.
staticcheck -go 1.9 -checks 'inherit,-ST1015' -ignore '
google.golang.org/grpc/balancer.go:SA1019
-google.golang.org/grpc/balancer_test.go:SA1019
-google.golang.org/grpc/clientconn_test.go:SA1019
+google.golang.org/grpc/balancer/grpclb/grpclb_remote_balancer.go:SA1019
google.golang.org/grpc/balancer/roundrobin/roundrobin_test.go:SA1019
+google.golang.org/grpc/xds/internal/balancer/edsbalancer/balancergroup.go:SA1019
+google.golang.org/grpc/xds/internal/resolver/xds_resolver.go:SA1019
+google.golang.org/grpc/xds/internal/balancer/xds.go:SA1019
+google.golang.org/grpc/xds/internal/balancer/xds_client.go:SA1019
+google.golang.org/grpc/balancer_conn_wrappers.go:SA1019
+google.golang.org/grpc/balancer_test.go:SA1019
google.golang.org/grpc/benchmark/benchmain/main.go:SA1019
google.golang.org/grpc/benchmark/worker/benchmark_client.go:SA1019
+google.golang.org/grpc/clientconn.go:S1024
+google.golang.org/grpc/clientconn_state_transition_test.go:SA1019
+google.golang.org/grpc/clientconn_test.go:SA1019
+google.golang.org/grpc/examples/features/debugging/client/main.go:SA1019
+google.golang.org/grpc/examples/features/load_balancing/client/main.go:SA1019
google.golang.org/grpc/internal/transport/handler_server.go:SA1019
google.golang.org/grpc/internal/transport/handler_server_test.go:SA1019
+google.golang.org/grpc/resolver/dns/dns_resolver.go:SA1019
google.golang.org/grpc/stats/stats_test.go:SA1019
+google.golang.org/grpc/test/balancer_test.go:SA1019
google.golang.org/grpc/test/channelz_test.go:SA1019
google.golang.org/grpc/test/end2end_test.go:SA1019
google.golang.org/grpc/test/healthcheck_test.go:SA1019