From def1a3b77fcf52e6ae2015865bcbbb2fe5033048 Mon Sep 17 00:00:00 2001 From: Gyu-Ho Lee Date: Mon, 10 Oct 2016 11:01:14 -0700 Subject: [PATCH 1/5] script/updatedep: update glide, glide-vc version --- scripts/updatedep.sh | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/scripts/updatedep.sh b/scripts/updatedep.sh index 66c74d24bda..3cb382f08a5 100755 --- a/scripts/updatedep.sh +++ b/scripts/updatedep.sh @@ -26,22 +26,28 @@ mv cmd/vendor vendor echo "manually deleting etcd-repo symlink in vendor" rm -f vendor/github.com/coreos/etcd -GLIDE_ROOT=$GOPATH/src/github.com/Masterminds/glide -go get -v -u github.com/Masterminds/glide -go get -v -u github.com/sgotti/glide-vc -GLIDE_ROOT=$GOPATH/src/github.com/Masterminds/glide -GLIDE_SHA=3e49dce57f4a3a1e9bc55475065235766000d2f0 +GLIDE_ROOT="$GOPATH/src/github.com/Masterminds/glide" +GLIDE_SHA=cfde1caa6b394a320fc65c5abc77646d18febff9 +go get -d -u github.com/Masterminds/glide pushd "${GLIDE_ROOT}" git reset --hard ${GLIDE_SHA} go install popd +GLIDE_VC_ROOT="$GOPATH/src/github.com/sgotti/glide-vc" +GLIDE_VC_SHA=d96375d23c85287e80296cdf48f9d21c227fa40a +go get -d -u github.com/sgotti/glide-vc +pushd "${GLIDE_VC_ROOT}" + git reset --hard ${GLIDE_VC_SHA} + go install +popd + if [ -n "$1" ]; then echo "glide get on $(echo $1)" - glide --verbose get --strip-vendor --strip-vcs --update-vendored --skip-test $1 + glide get --strip-vendor --skip-test $1 else echo "glide update on *" - glide --verbose update --delete --strip-vendor --strip-vcs --update-vendored --skip-test + glide update --strip-vendor --skip-test fi; echo "removing test files" From b9f3ef09e1e516e4f70421ab12e36633da960558 Mon Sep 17 00:00:00 2001 From: Gyu-Ho Lee Date: Mon, 10 Oct 2016 11:02:13 -0700 Subject: [PATCH 2/5] vendor: clean up dependencies (remove unused ones) --- .../github.com/cpuguy83/go-md2man/md2man.go | 44 -- .../github.com/golang/groupcache/byteview.go | 160 ------ .../golang/groupcache/groupcache.go | 489 ------------------ .../github.com/golang/groupcache/http.go | 227 -------- .../github.com/golang/groupcache/peers.go | 71 --- .../github.com/golang/groupcache/sinks.go | 322 ------------ glide.lock | 15 +- glide.yaml | 11 +- 8 files changed, 15 insertions(+), 1324 deletions(-) delete mode 100644 cmd/vendor/github.com/cpuguy83/go-md2man/md2man.go delete mode 100644 cmd/vendor/github.com/golang/groupcache/byteview.go delete mode 100644 cmd/vendor/github.com/golang/groupcache/groupcache.go delete mode 100644 cmd/vendor/github.com/golang/groupcache/http.go delete mode 100644 cmd/vendor/github.com/golang/groupcache/peers.go delete mode 100644 cmd/vendor/github.com/golang/groupcache/sinks.go diff --git a/cmd/vendor/github.com/cpuguy83/go-md2man/md2man.go b/cmd/vendor/github.com/cpuguy83/go-md2man/md2man.go deleted file mode 100644 index 1dc70f47a72..00000000000 --- a/cmd/vendor/github.com/cpuguy83/go-md2man/md2man.go +++ /dev/null @@ -1,44 +0,0 @@ -package main - -import ( - "flag" - "fmt" - "io/ioutil" - "os" - - "github.com/cpuguy83/go-md2man/md2man" -) - -var inFilePath = flag.String("in", "", "Path to file to be processed") -var outFilePath = flag.String("out", "", "Path to output processed file") - -func main() { - flag.Parse() - - inFile, err := os.Open(*inFilePath) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - defer inFile.Close() - - doc, err := ioutil.ReadAll(inFile) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - - out := md2man.Render(doc) - - outFile, err := os.Create(*outFilePath) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - defer outFile.Close() - _, err = outFile.Write(out) - if err != nil { - fmt.Println(err) - os.Exit(1) - } -} diff --git a/cmd/vendor/github.com/golang/groupcache/byteview.go b/cmd/vendor/github.com/golang/groupcache/byteview.go deleted file mode 100644 index 035a9ee4410..00000000000 --- a/cmd/vendor/github.com/golang/groupcache/byteview.go +++ /dev/null @@ -1,160 +0,0 @@ -/* -Copyright 2012 Google Inc. - -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 groupcache - -import ( - "bytes" - "errors" - "io" - "strings" -) - -// A ByteView holds an immutable view of bytes. -// Internally it wraps either a []byte or a string, -// but that detail is invisible to callers. -// -// A ByteView is meant to be used as a value type, not -// a pointer (like a time.Time). -type ByteView struct { - // If b is non-nil, b is used, else s is used. - b []byte - s string -} - -// Len returns the view's length. -func (v ByteView) Len() int { - if v.b != nil { - return len(v.b) - } - return len(v.s) -} - -// ByteSlice returns a copy of the data as a byte slice. -func (v ByteView) ByteSlice() []byte { - if v.b != nil { - return cloneBytes(v.b) - } - return []byte(v.s) -} - -// String returns the data as a string, making a copy if necessary. -func (v ByteView) String() string { - if v.b != nil { - return string(v.b) - } - return v.s -} - -// At returns the byte at index i. -func (v ByteView) At(i int) byte { - if v.b != nil { - return v.b[i] - } - return v.s[i] -} - -// Slice slices the view between the provided from and to indices. -func (v ByteView) Slice(from, to int) ByteView { - if v.b != nil { - return ByteView{b: v.b[from:to]} - } - return ByteView{s: v.s[from:to]} -} - -// SliceFrom slices the view from the provided index until the end. -func (v ByteView) SliceFrom(from int) ByteView { - if v.b != nil { - return ByteView{b: v.b[from:]} - } - return ByteView{s: v.s[from:]} -} - -// Copy copies b into dest and returns the number of bytes copied. -func (v ByteView) Copy(dest []byte) int { - if v.b != nil { - return copy(dest, v.b) - } - return copy(dest, v.s) -} - -// Equal returns whether the bytes in b are the same as the bytes in -// b2. -func (v ByteView) Equal(b2 ByteView) bool { - if b2.b == nil { - return v.EqualString(b2.s) - } - return v.EqualBytes(b2.b) -} - -// EqualString returns whether the bytes in b are the same as the bytes -// in s. -func (v ByteView) EqualString(s string) bool { - if v.b == nil { - return v.s == s - } - l := v.Len() - if len(s) != l { - return false - } - for i, bi := range v.b { - if bi != s[i] { - return false - } - } - return true -} - -// EqualBytes returns whether the bytes in b are the same as the bytes -// in b2. -func (v ByteView) EqualBytes(b2 []byte) bool { - if v.b != nil { - return bytes.Equal(v.b, b2) - } - l := v.Len() - if len(b2) != l { - return false - } - for i, bi := range b2 { - if bi != v.s[i] { - return false - } - } - return true -} - -// Reader returns an io.ReadSeeker for the bytes in v. -func (v ByteView) Reader() io.ReadSeeker { - if v.b != nil { - return bytes.NewReader(v.b) - } - return strings.NewReader(v.s) -} - -// ReadAt implements io.ReaderAt on the bytes in v. -func (v ByteView) ReadAt(p []byte, off int64) (n int, err error) { - if off < 0 { - return 0, errors.New("view: invalid offset") - } - if off >= int64(v.Len()) { - return 0, io.EOF - } - n = v.SliceFrom(int(off)).Copy(p) - if n < len(p) { - err = io.EOF - } - return -} diff --git a/cmd/vendor/github.com/golang/groupcache/groupcache.go b/cmd/vendor/github.com/golang/groupcache/groupcache.go deleted file mode 100644 index 9499dbb6a7d..00000000000 --- a/cmd/vendor/github.com/golang/groupcache/groupcache.go +++ /dev/null @@ -1,489 +0,0 @@ -/* -Copyright 2012 Google Inc. - -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 groupcache provides a data loading mechanism with caching -// and de-duplication that works across a set of peer processes. -// -// Each data Get first consults its local cache, otherwise delegates -// to the requested key's canonical owner, which then checks its cache -// or finally gets the data. In the common case, many concurrent -// cache misses across a set of peers for the same key result in just -// one cache fill. -package groupcache - -import ( - "errors" - "math/rand" - "strconv" - "sync" - "sync/atomic" - - pb "github.com/golang/groupcache/groupcachepb" - "github.com/golang/groupcache/lru" - "github.com/golang/groupcache/singleflight" -) - -// A Getter loads data for a key. -type Getter interface { - // Get returns the value identified by key, populating dest. - // - // The returned data must be unversioned. That is, key must - // uniquely describe the loaded data, without an implicit - // current time, and without relying on cache expiration - // mechanisms. - Get(ctx Context, key string, dest Sink) error -} - -// A GetterFunc implements Getter with a function. -type GetterFunc func(ctx Context, key string, dest Sink) error - -func (f GetterFunc) Get(ctx Context, key string, dest Sink) error { - return f(ctx, key, dest) -} - -var ( - mu sync.RWMutex - groups = make(map[string]*Group) - - initPeerServerOnce sync.Once - initPeerServer func() -) - -// GetGroup returns the named group previously created with NewGroup, or -// nil if there's no such group. -func GetGroup(name string) *Group { - mu.RLock() - g := groups[name] - mu.RUnlock() - return g -} - -// NewGroup creates a coordinated group-aware Getter from a Getter. -// -// The returned Getter tries (but does not guarantee) to run only one -// Get call at once for a given key across an entire set of peer -// processes. Concurrent callers both in the local process and in -// other processes receive copies of the answer once the original Get -// completes. -// -// The group name must be unique for each getter. -func NewGroup(name string, cacheBytes int64, getter Getter) *Group { - return newGroup(name, cacheBytes, getter, nil) -} - -// If peers is nil, the peerPicker is called via a sync.Once to initialize it. -func newGroup(name string, cacheBytes int64, getter Getter, peers PeerPicker) *Group { - if getter == nil { - panic("nil Getter") - } - mu.Lock() - defer mu.Unlock() - initPeerServerOnce.Do(callInitPeerServer) - if _, dup := groups[name]; dup { - panic("duplicate registration of group " + name) - } - g := &Group{ - name: name, - getter: getter, - peers: peers, - cacheBytes: cacheBytes, - loadGroup: &singleflight.Group{}, - } - if fn := newGroupHook; fn != nil { - fn(g) - } - groups[name] = g - return g -} - -// newGroupHook, if non-nil, is called right after a new group is created. -var newGroupHook func(*Group) - -// RegisterNewGroupHook registers a hook that is run each time -// a group is created. -func RegisterNewGroupHook(fn func(*Group)) { - if newGroupHook != nil { - panic("RegisterNewGroupHook called more than once") - } - newGroupHook = fn -} - -// RegisterServerStart registers a hook that is run when the first -// group is created. -func RegisterServerStart(fn func()) { - if initPeerServer != nil { - panic("RegisterServerStart called more than once") - } - initPeerServer = fn -} - -func callInitPeerServer() { - if initPeerServer != nil { - initPeerServer() - } -} - -// A Group is a cache namespace and associated data loaded spread over -// a group of 1 or more machines. -type Group struct { - name string - getter Getter - peersOnce sync.Once - peers PeerPicker - cacheBytes int64 // limit for sum of mainCache and hotCache size - - // mainCache is a cache of the keys for which this process - // (amongst its peers) is authoritative. That is, this cache - // contains keys which consistent hash on to this process's - // peer number. - mainCache cache - - // hotCache contains keys/values for which this peer is not - // authoritative (otherwise they would be in mainCache), but - // are popular enough to warrant mirroring in this process to - // avoid going over the network to fetch from a peer. Having - // a hotCache avoids network hotspotting, where a peer's - // network card could become the bottleneck on a popular key. - // This cache is used sparingly to maximize the total number - // of key/value pairs that can be stored globally. - hotCache cache - - // loadGroup ensures that each key is only fetched once - // (either locally or remotely), regardless of the number of - // concurrent callers. - loadGroup flightGroup - - // Stats are statistics on the group. - Stats Stats -} - -// flightGroup is defined as an interface which flightgroup.Group -// satisfies. We define this so that we may test with an alternate -// implementation. -type flightGroup interface { - // Done is called when Do is done. - Do(key string, fn func() (interface{}, error)) (interface{}, error) -} - -// Stats are per-group statistics. -type Stats struct { - Gets AtomicInt // any Get request, including from peers - CacheHits AtomicInt // either cache was good - PeerLoads AtomicInt // either remote load or remote cache hit (not an error) - PeerErrors AtomicInt - Loads AtomicInt // (gets - cacheHits) - LoadsDeduped AtomicInt // after singleflight - LocalLoads AtomicInt // total good local loads - LocalLoadErrs AtomicInt // total bad local loads - ServerRequests AtomicInt // gets that came over the network from peers -} - -// Name returns the name of the group. -func (g *Group) Name() string { - return g.name -} - -func (g *Group) initPeers() { - if g.peers == nil { - g.peers = getPeers() - } -} - -func (g *Group) Get(ctx Context, key string, dest Sink) error { - g.peersOnce.Do(g.initPeers) - g.Stats.Gets.Add(1) - if dest == nil { - return errors.New("groupcache: nil dest Sink") - } - value, cacheHit := g.lookupCache(key) - - if cacheHit { - g.Stats.CacheHits.Add(1) - return setSinkView(dest, value) - } - - // Optimization to avoid double unmarshalling or copying: keep - // track of whether the dest was already populated. One caller - // (if local) will set this; the losers will not. The common - // case will likely be one caller. - destPopulated := false - value, destPopulated, err := g.load(ctx, key, dest) - if err != nil { - return err - } - if destPopulated { - return nil - } - return setSinkView(dest, value) -} - -// load loads key either by invoking the getter locally or by sending it to another machine. -func (g *Group) load(ctx Context, key string, dest Sink) (value ByteView, destPopulated bool, err error) { - g.Stats.Loads.Add(1) - viewi, err := g.loadGroup.Do(key, func() (interface{}, error) { - // Check the cache again because singleflight can only dedup calls - // that overlap concurrently. It's possible for 2 concurrent - // requests to miss the cache, resulting in 2 load() calls. An - // unfortunate goroutine scheduling would result in this callback - // being run twice, serially. If we don't check the cache again, - // cache.nbytes would be incremented below even though there will - // be only one entry for this key. - // - // Consider the following serialized event ordering for two - // goroutines in which this callback gets called twice for hte - // same key: - // 1: Get("key") - // 2: Get("key") - // 1: lookupCache("key") - // 2: lookupCache("key") - // 1: load("key") - // 2: load("key") - // 1: loadGroup.Do("key", fn) - // 1: fn() - // 2: loadGroup.Do("key", fn) - // 2: fn() - if value, cacheHit := g.lookupCache(key); cacheHit { - g.Stats.CacheHits.Add(1) - return value, nil - } - g.Stats.LoadsDeduped.Add(1) - var value ByteView - var err error - if peer, ok := g.peers.PickPeer(key); ok { - value, err = g.getFromPeer(ctx, peer, key) - if err == nil { - g.Stats.PeerLoads.Add(1) - return value, nil - } - g.Stats.PeerErrors.Add(1) - // TODO(bradfitz): log the peer's error? keep - // log of the past few for /groupcachez? It's - // probably boring (normal task movement), so not - // worth logging I imagine. - } - value, err = g.getLocally(ctx, key, dest) - if err != nil { - g.Stats.LocalLoadErrs.Add(1) - return nil, err - } - g.Stats.LocalLoads.Add(1) - destPopulated = true // only one caller of load gets this return value - g.populateCache(key, value, &g.mainCache) - return value, nil - }) - if err == nil { - value = viewi.(ByteView) - } - return -} - -func (g *Group) getLocally(ctx Context, key string, dest Sink) (ByteView, error) { - err := g.getter.Get(ctx, key, dest) - if err != nil { - return ByteView{}, err - } - return dest.view() -} - -func (g *Group) getFromPeer(ctx Context, peer ProtoGetter, key string) (ByteView, error) { - req := &pb.GetRequest{ - Group: &g.name, - Key: &key, - } - res := &pb.GetResponse{} - err := peer.Get(ctx, req, res) - if err != nil { - return ByteView{}, err - } - value := ByteView{b: res.Value} - // TODO(bradfitz): use res.MinuteQps or something smart to - // conditionally populate hotCache. For now just do it some - // percentage of the time. - if rand.Intn(10) == 0 { - g.populateCache(key, value, &g.hotCache) - } - return value, nil -} - -func (g *Group) lookupCache(key string) (value ByteView, ok bool) { - if g.cacheBytes <= 0 { - return - } - value, ok = g.mainCache.get(key) - if ok { - return - } - value, ok = g.hotCache.get(key) - return -} - -func (g *Group) populateCache(key string, value ByteView, cache *cache) { - if g.cacheBytes <= 0 { - return - } - cache.add(key, value) - - // Evict items from cache(s) if necessary. - for { - mainBytes := g.mainCache.bytes() - hotBytes := g.hotCache.bytes() - if mainBytes+hotBytes <= g.cacheBytes { - return - } - - // TODO(bradfitz): this is good-enough-for-now logic. - // It should be something based on measurements and/or - // respecting the costs of different resources. - victim := &g.mainCache - if hotBytes > mainBytes/8 { - victim = &g.hotCache - } - victim.removeOldest() - } -} - -// CacheType represents a type of cache. -type CacheType int - -const ( - // The MainCache is the cache for items that this peer is the - // owner for. - MainCache CacheType = iota + 1 - - // The HotCache is the cache for items that seem popular - // enough to replicate to this node, even though it's not the - // owner. - HotCache -) - -// CacheStats returns stats about the provided cache within the group. -func (g *Group) CacheStats(which CacheType) CacheStats { - switch which { - case MainCache: - return g.mainCache.stats() - case HotCache: - return g.hotCache.stats() - default: - return CacheStats{} - } -} - -// cache is a wrapper around an *lru.Cache that adds synchronization, -// makes values always be ByteView, and counts the size of all keys and -// values. -type cache struct { - mu sync.RWMutex - nbytes int64 // of all keys and values - lru *lru.Cache - nhit, nget int64 - nevict int64 // number of evictions -} - -func (c *cache) stats() CacheStats { - c.mu.RLock() - defer c.mu.RUnlock() - return CacheStats{ - Bytes: c.nbytes, - Items: c.itemsLocked(), - Gets: c.nget, - Hits: c.nhit, - Evictions: c.nevict, - } -} - -func (c *cache) add(key string, value ByteView) { - c.mu.Lock() - defer c.mu.Unlock() - if c.lru == nil { - c.lru = &lru.Cache{ - OnEvicted: func(key lru.Key, value interface{}) { - val := value.(ByteView) - c.nbytes -= int64(len(key.(string))) + int64(val.Len()) - c.nevict++ - }, - } - } - c.lru.Add(key, value) - c.nbytes += int64(len(key)) + int64(value.Len()) -} - -func (c *cache) get(key string) (value ByteView, ok bool) { - c.mu.Lock() - defer c.mu.Unlock() - c.nget++ - if c.lru == nil { - return - } - vi, ok := c.lru.Get(key) - if !ok { - return - } - c.nhit++ - return vi.(ByteView), true -} - -func (c *cache) removeOldest() { - c.mu.Lock() - defer c.mu.Unlock() - if c.lru != nil { - c.lru.RemoveOldest() - } -} - -func (c *cache) bytes() int64 { - c.mu.RLock() - defer c.mu.RUnlock() - return c.nbytes -} - -func (c *cache) items() int64 { - c.mu.RLock() - defer c.mu.RUnlock() - return c.itemsLocked() -} - -func (c *cache) itemsLocked() int64 { - if c.lru == nil { - return 0 - } - return int64(c.lru.Len()) -} - -// An AtomicInt is an int64 to be accessed atomically. -type AtomicInt int64 - -// Add atomically adds n to i. -func (i *AtomicInt) Add(n int64) { - atomic.AddInt64((*int64)(i), n) -} - -// Get atomically gets the value of i. -func (i *AtomicInt) Get() int64 { - return atomic.LoadInt64((*int64)(i)) -} - -func (i *AtomicInt) String() string { - return strconv.FormatInt(i.Get(), 10) -} - -// CacheStats are returned by stats accessors on Group. -type CacheStats struct { - Bytes int64 - Items int64 - Gets int64 - Hits int64 - Evictions int64 -} diff --git a/cmd/vendor/github.com/golang/groupcache/http.go b/cmd/vendor/github.com/golang/groupcache/http.go deleted file mode 100644 index 14eb345a8fa..00000000000 --- a/cmd/vendor/github.com/golang/groupcache/http.go +++ /dev/null @@ -1,227 +0,0 @@ -/* -Copyright 2013 Google Inc. - -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 groupcache - -import ( - "bytes" - "fmt" - "io" - "net/http" - "net/url" - "strings" - "sync" - - "github.com/golang/groupcache/consistenthash" - pb "github.com/golang/groupcache/groupcachepb" - "github.com/golang/protobuf/proto" -) - -const defaultBasePath = "/_groupcache/" - -const defaultReplicas = 50 - -// HTTPPool implements PeerPicker for a pool of HTTP peers. -type HTTPPool struct { - // Context optionally specifies a context for the server to use when it - // receives a request. - // If nil, the server uses a nil Context. - Context func(*http.Request) Context - - // Transport optionally specifies an http.RoundTripper for the client - // to use when it makes a request. - // If nil, the client uses http.DefaultTransport. - Transport func(Context) http.RoundTripper - - // this peer's base URL, e.g. "https://example.net:8000" - self string - - // opts specifies the options. - opts HTTPPoolOptions - - mu sync.Mutex // guards peers and httpGetters - peers *consistenthash.Map - httpGetters map[string]*httpGetter // keyed by e.g. "http://10.0.0.2:8008" -} - -// HTTPPoolOptions are the configurations of a HTTPPool. -type HTTPPoolOptions struct { - // BasePath specifies the HTTP path that will serve groupcache requests. - // If blank, it defaults to "/_groupcache/". - BasePath string - - // Replicas specifies the number of key replicas on the consistent hash. - // If blank, it defaults to 50. - Replicas int - - // HashFn specifies the hash function of the consistent hash. - // If blank, it defaults to crc32.ChecksumIEEE. - HashFn consistenthash.Hash -} - -// NewHTTPPool initializes an HTTP pool of peers, and registers itself as a PeerPicker. -// For convenience, it also registers itself as an http.Handler with http.DefaultServeMux. -// The self argument be a valid base URL that points to the current server, -// for example "http://example.net:8000". -func NewHTTPPool(self string) *HTTPPool { - p := NewHTTPPoolOpts(self, nil) - http.Handle(p.opts.BasePath, p) - return p -} - -var httpPoolMade bool - -// NewHTTPPoolOpts initializes an HTTP pool of peers with the given options. -// Unlike NewHTTPPool, this function does not register the created pool as an HTTP handler. -// The returned *HTTPPool implements http.Handler and must be registered using http.Handle. -func NewHTTPPoolOpts(self string, o *HTTPPoolOptions) *HTTPPool { - if httpPoolMade { - panic("groupcache: NewHTTPPool must be called only once") - } - httpPoolMade = true - - p := &HTTPPool{ - self: self, - httpGetters: make(map[string]*httpGetter), - } - if o != nil { - p.opts = *o - } - if p.opts.BasePath == "" { - p.opts.BasePath = defaultBasePath - } - if p.opts.Replicas == 0 { - p.opts.Replicas = defaultReplicas - } - p.peers = consistenthash.New(p.opts.Replicas, p.opts.HashFn) - - RegisterPeerPicker(func() PeerPicker { return p }) - return p -} - -// Set updates the pool's list of peers. -// Each peer value should be a valid base URL, -// for example "http://example.net:8000". -func (p *HTTPPool) Set(peers ...string) { - p.mu.Lock() - defer p.mu.Unlock() - p.peers = consistenthash.New(p.opts.Replicas, p.opts.HashFn) - p.peers.Add(peers...) - p.httpGetters = make(map[string]*httpGetter, len(peers)) - for _, peer := range peers { - p.httpGetters[peer] = &httpGetter{transport: p.Transport, baseURL: peer + p.opts.BasePath} - } -} - -func (p *HTTPPool) PickPeer(key string) (ProtoGetter, bool) { - p.mu.Lock() - defer p.mu.Unlock() - if p.peers.IsEmpty() { - return nil, false - } - if peer := p.peers.Get(key); peer != p.self { - return p.httpGetters[peer], true - } - return nil, false -} - -func (p *HTTPPool) ServeHTTP(w http.ResponseWriter, r *http.Request) { - // Parse request. - if !strings.HasPrefix(r.URL.Path, p.opts.BasePath) { - panic("HTTPPool serving unexpected path: " + r.URL.Path) - } - parts := strings.SplitN(r.URL.Path[len(p.opts.BasePath):], "/", 2) - if len(parts) != 2 { - http.Error(w, "bad request", http.StatusBadRequest) - return - } - groupName := parts[0] - key := parts[1] - - // Fetch the value for this group/key. - group := GetGroup(groupName) - if group == nil { - http.Error(w, "no such group: "+groupName, http.StatusNotFound) - return - } - var ctx Context - if p.Context != nil { - ctx = p.Context(r) - } - - group.Stats.ServerRequests.Add(1) - var value []byte - err := group.Get(ctx, key, AllocatingByteSliceSink(&value)) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - // Write the value to the response body as a proto message. - body, err := proto.Marshal(&pb.GetResponse{Value: value}) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - w.Header().Set("Content-Type", "application/x-protobuf") - w.Write(body) -} - -type httpGetter struct { - transport func(Context) http.RoundTripper - baseURL string -} - -var bufferPool = sync.Pool{ - New: func() interface{} { return new(bytes.Buffer) }, -} - -func (h *httpGetter) Get(context Context, in *pb.GetRequest, out *pb.GetResponse) error { - u := fmt.Sprintf( - "%v%v/%v", - h.baseURL, - url.QueryEscape(in.GetGroup()), - url.QueryEscape(in.GetKey()), - ) - req, err := http.NewRequest("GET", u, nil) - if err != nil { - return err - } - tr := http.DefaultTransport - if h.transport != nil { - tr = h.transport(context) - } - res, err := tr.RoundTrip(req) - if err != nil { - return err - } - defer res.Body.Close() - if res.StatusCode != http.StatusOK { - return fmt.Errorf("server returned: %v", res.Status) - } - b := bufferPool.Get().(*bytes.Buffer) - b.Reset() - defer bufferPool.Put(b) - _, err = io.Copy(b, res.Body) - if err != nil { - return fmt.Errorf("reading response body: %v", err) - } - err = proto.Unmarshal(b.Bytes(), out) - if err != nil { - return fmt.Errorf("decoding response body: %v", err) - } - return nil -} diff --git a/cmd/vendor/github.com/golang/groupcache/peers.go b/cmd/vendor/github.com/golang/groupcache/peers.go deleted file mode 100644 index a74a79b8f47..00000000000 --- a/cmd/vendor/github.com/golang/groupcache/peers.go +++ /dev/null @@ -1,71 +0,0 @@ -/* -Copyright 2012 Google Inc. - -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. -*/ - -// peers.go defines how processes find and communicate with their peers. - -package groupcache - -import ( - pb "github.com/golang/groupcache/groupcachepb" -) - -// Context is an opaque value passed through calls to the -// ProtoGetter. It may be nil if your ProtoGetter implementation does -// not require a context. -type Context interface{} - -// ProtoGetter is the interface that must be implemented by a peer. -type ProtoGetter interface { - Get(context Context, in *pb.GetRequest, out *pb.GetResponse) error -} - -// PeerPicker is the interface that must be implemented to locate -// the peer that owns a specific key. -type PeerPicker interface { - // PickPeer returns the peer that owns the specific key - // and true to indicate that a remote peer was nominated. - // It returns nil, false if the key owner is the current peer. - PickPeer(key string) (peer ProtoGetter, ok bool) -} - -// NoPeers is an implementation of PeerPicker that never finds a peer. -type NoPeers struct{} - -func (NoPeers) PickPeer(key string) (peer ProtoGetter, ok bool) { return } - -var ( - portPicker func() PeerPicker -) - -// RegisterPeerPicker registers the peer initialization function. -// It is called once, when the first group is created. -func RegisterPeerPicker(fn func() PeerPicker) { - if portPicker != nil { - panic("RegisterPeerPicker called more than once") - } - portPicker = fn -} - -func getPeers() PeerPicker { - if portPicker == nil { - return NoPeers{} - } - pk := portPicker() - if pk == nil { - pk = NoPeers{} - } - return pk -} diff --git a/cmd/vendor/github.com/golang/groupcache/sinks.go b/cmd/vendor/github.com/golang/groupcache/sinks.go deleted file mode 100644 index cb42b41b4d7..00000000000 --- a/cmd/vendor/github.com/golang/groupcache/sinks.go +++ /dev/null @@ -1,322 +0,0 @@ -/* -Copyright 2012 Google Inc. - -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 groupcache - -import ( - "errors" - - "github.com/golang/protobuf/proto" -) - -// A Sink receives data from a Get call. -// -// Implementation of Getter must call exactly one of the Set methods -// on success. -type Sink interface { - // SetString sets the value to s. - SetString(s string) error - - // SetBytes sets the value to the contents of v. - // The caller retains ownership of v. - SetBytes(v []byte) error - - // SetProto sets the value to the encoded version of m. - // The caller retains ownership of m. - SetProto(m proto.Message) error - - // view returns a frozen view of the bytes for caching. - view() (ByteView, error) -} - -func cloneBytes(b []byte) []byte { - c := make([]byte, len(b)) - copy(c, b) - return c -} - -func setSinkView(s Sink, v ByteView) error { - // A viewSetter is a Sink that can also receive its value from - // a ByteView. This is a fast path to minimize copies when the - // item was already cached locally in memory (where it's - // cached as a ByteView) - type viewSetter interface { - setView(v ByteView) error - } - if vs, ok := s.(viewSetter); ok { - return vs.setView(v) - } - if v.b != nil { - return s.SetBytes(v.b) - } - return s.SetString(v.s) -} - -// StringSink returns a Sink that populates the provided string pointer. -func StringSink(sp *string) Sink { - return &stringSink{sp: sp} -} - -type stringSink struct { - sp *string - v ByteView - // TODO(bradfitz): track whether any Sets were called. -} - -func (s *stringSink) view() (ByteView, error) { - // TODO(bradfitz): return an error if no Set was called - return s.v, nil -} - -func (s *stringSink) SetString(v string) error { - s.v.b = nil - s.v.s = v - *s.sp = v - return nil -} - -func (s *stringSink) SetBytes(v []byte) error { - return s.SetString(string(v)) -} - -func (s *stringSink) SetProto(m proto.Message) error { - b, err := proto.Marshal(m) - if err != nil { - return err - } - s.v.b = b - *s.sp = string(b) - return nil -} - -// ByteViewSink returns a Sink that populates a ByteView. -func ByteViewSink(dst *ByteView) Sink { - if dst == nil { - panic("nil dst") - } - return &byteViewSink{dst: dst} -} - -type byteViewSink struct { - dst *ByteView - - // if this code ever ends up tracking that at least one set* - // method was called, don't make it an error to call set - // methods multiple times. Lorry's payload.go does that, and - // it makes sense. The comment at the top of this file about - // "exactly one of the Set methods" is overly strict. We - // really care about at least once (in a handler), but if - // multiple handlers fail (or multiple functions in a program - // using a Sink), it's okay to re-use the same one. -} - -func (s *byteViewSink) setView(v ByteView) error { - *s.dst = v - return nil -} - -func (s *byteViewSink) view() (ByteView, error) { - return *s.dst, nil -} - -func (s *byteViewSink) SetProto(m proto.Message) error { - b, err := proto.Marshal(m) - if err != nil { - return err - } - *s.dst = ByteView{b: b} - return nil -} - -func (s *byteViewSink) SetBytes(b []byte) error { - *s.dst = ByteView{b: cloneBytes(b)} - return nil -} - -func (s *byteViewSink) SetString(v string) error { - *s.dst = ByteView{s: v} - return nil -} - -// ProtoSink returns a sink that unmarshals binary proto values into m. -func ProtoSink(m proto.Message) Sink { - return &protoSink{ - dst: m, - } -} - -type protoSink struct { - dst proto.Message // authorative value - typ string - - v ByteView // encoded -} - -func (s *protoSink) view() (ByteView, error) { - return s.v, nil -} - -func (s *protoSink) SetBytes(b []byte) error { - err := proto.Unmarshal(b, s.dst) - if err != nil { - return err - } - s.v.b = cloneBytes(b) - s.v.s = "" - return nil -} - -func (s *protoSink) SetString(v string) error { - b := []byte(v) - err := proto.Unmarshal(b, s.dst) - if err != nil { - return err - } - s.v.b = b - s.v.s = "" - return nil -} - -func (s *protoSink) SetProto(m proto.Message) error { - b, err := proto.Marshal(m) - if err != nil { - return err - } - // TODO(bradfitz): optimize for same-task case more and write - // right through? would need to document ownership rules at - // the same time. but then we could just assign *dst = *m - // here. This works for now: - err = proto.Unmarshal(b, s.dst) - if err != nil { - return err - } - s.v.b = b - s.v.s = "" - return nil -} - -// AllocatingByteSliceSink returns a Sink that allocates -// a byte slice to hold the received value and assigns -// it to *dst. The memory is not retained by groupcache. -func AllocatingByteSliceSink(dst *[]byte) Sink { - return &allocBytesSink{dst: dst} -} - -type allocBytesSink struct { - dst *[]byte - v ByteView -} - -func (s *allocBytesSink) view() (ByteView, error) { - return s.v, nil -} - -func (s *allocBytesSink) setView(v ByteView) error { - if v.b != nil { - *s.dst = cloneBytes(v.b) - } else { - *s.dst = []byte(v.s) - } - s.v = v - return nil -} - -func (s *allocBytesSink) SetProto(m proto.Message) error { - b, err := proto.Marshal(m) - if err != nil { - return err - } - return s.setBytesOwned(b) -} - -func (s *allocBytesSink) SetBytes(b []byte) error { - return s.setBytesOwned(cloneBytes(b)) -} - -func (s *allocBytesSink) setBytesOwned(b []byte) error { - if s.dst == nil { - return errors.New("nil AllocatingByteSliceSink *[]byte dst") - } - *s.dst = cloneBytes(b) // another copy, protecting the read-only s.v.b view - s.v.b = b - s.v.s = "" - return nil -} - -func (s *allocBytesSink) SetString(v string) error { - if s.dst == nil { - return errors.New("nil AllocatingByteSliceSink *[]byte dst") - } - *s.dst = []byte(v) - s.v.b = nil - s.v.s = v - return nil -} - -// TruncatingByteSliceSink returns a Sink that writes up to len(*dst) -// bytes to *dst. If more bytes are available, they're silently -// truncated. If fewer bytes are available than len(*dst), *dst -// is shrunk to fit the number of bytes available. -func TruncatingByteSliceSink(dst *[]byte) Sink { - return &truncBytesSink{dst: dst} -} - -type truncBytesSink struct { - dst *[]byte - v ByteView -} - -func (s *truncBytesSink) view() (ByteView, error) { - return s.v, nil -} - -func (s *truncBytesSink) SetProto(m proto.Message) error { - b, err := proto.Marshal(m) - if err != nil { - return err - } - return s.setBytesOwned(b) -} - -func (s *truncBytesSink) SetBytes(b []byte) error { - return s.setBytesOwned(cloneBytes(b)) -} - -func (s *truncBytesSink) setBytesOwned(b []byte) error { - if s.dst == nil { - return errors.New("nil TruncatingByteSliceSink *[]byte dst") - } - n := copy(*s.dst, b) - if n < len(*s.dst) { - *s.dst = (*s.dst)[:n] - } - s.v.b = b - s.v.s = "" - return nil -} - -func (s *truncBytesSink) SetString(v string) error { - if s.dst == nil { - return errors.New("nil TruncatingByteSliceSink *[]byte dst") - } - n := copy(*s.dst, v) - if n < len(*s.dst) { - *s.dst = (*s.dst)[:n] - } - s.v.b = nil - s.v.s = v - return nil -} diff --git a/glide.lock b/glide.lock index a3b41fe3ed0..279d69470f9 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 0e7d50c7716873a35f2d14542747525a35af1e5bb734b3696bc1ff78d30107b4 -updated: 2016-09-23T12:52:52.300796603-07:00 +hash: 8363b260167c451ce8e3863fe01e234808b4598d1025ceefb8bdb11fb0ce83ca +updated: 2016-10-10T10:59:55.934552225-07:00 imports: - name: bitbucket.org/ww/goautoneg version: 75cd24fc2f2c2a2088577d12123ddee5f54e0675 @@ -100,10 +100,6 @@ imports: version: 1c44ec8d3f1552cac48999f9306da23c4d8a288b - name: github.com/spf13/pflag version: 08b1a584251b5b62f458943640fc8ebd4d50aaa5 -- name: github.com/stretchr/testify - version: 9cc77fa25329013ce07362c7742952ff887361f2 - subpackages: - - assert - name: github.com/ugorji/go version: f1f1a805ed361a0e078bb537e4ea78cd37dcf065 subpackages: @@ -148,4 +144,9 @@ imports: version: 29ad9b62f9e0274422d738242b94a5b89440bfa6 - name: gopkg.in/yaml.v2 version: 53feefa2559fb8dfa8d81baad31be332c97d6c77 -testImports: [] +testImports: +- name: github.com/stretchr/testify + version: "" + repo: https://github.com/stretchr/testify.git + subpackages: + - assert diff --git a/glide.yaml b/glide.yaml index 03eba779617..895e966e456 100644 --- a/glide.yaml +++ b/glide.yaml @@ -98,10 +98,6 @@ import: version: 1c44ec8d3f1552cac48999f9306da23c4d8a288b - package: github.com/spf13/pflag version: 08b1a584251b5b62f458943640fc8ebd4d50aaa5 -- package: github.com/stretchr/testify - version: 9cc77fa25329013ce07362c7742952ff887361f2 - subpackages: - - assert - package: github.com/ugorji/go version: f1f1a805ed361a0e078bb537e4ea78cd37dcf065 subpackages: @@ -146,3 +142,10 @@ import: version: 29ad9b62f9e0274422d738242b94a5b89440bfa6 - package: gopkg.in/yaml.v2 version: 53feefa2559fb8dfa8d81baad31be332c97d6c77 +testImport: +- package: github.com/stretchr/testify + repo: https://github.com/stretchr/testify.git + version: 976c720a22c8eb4eb6a0b4348ad85ad12491a506 + subpackages: + - assert + From 69ea359e627c399607c30effd7e9f976bf1a9a60 Mon Sep 17 00:00:00 2001 From: Gyu-Ho Lee Date: Mon, 10 Oct 2016 11:04:34 -0700 Subject: [PATCH 3/5] vendor: update glide.yaml with grpc-go v1.0.2 tag --- glide.lock | 6 +++--- glide.yaml | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/glide.lock b/glide.lock index 279d69470f9..59ab29bd7cf 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 8363b260167c451ce8e3863fe01e234808b4598d1025ceefb8bdb11fb0ce83ca -updated: 2016-10-10T10:59:55.934552225-07:00 +hash: f1f10632a41d55b08f5b1cc3fff1e3522fcc98663bb1ea13cb9aa9827c1c7a5f +updated: 2016-10-10T11:03:32.434091858-07:00 imports: - name: bitbucket.org/ww/goautoneg version: 75cd24fc2f2c2a2088577d12123ddee5f54e0675 @@ -130,7 +130,7 @@ imports: subpackages: - rate - name: google.golang.org/grpc - version: 231b4cfea0e79843053a33f5fe90bd4d84b23cd3 + version: b1a2821ca5a4fd6b6e48ddfbb7d6d7584d839d21 subpackages: - codes - credentials diff --git a/glide.yaml b/glide.yaml index 895e966e456..ed489c3687f 100644 --- a/glide.yaml +++ b/glide.yaml @@ -128,7 +128,7 @@ import: subpackages: - rate - package: google.golang.org/grpc - version: 231b4cfea0e79843053a33f5fe90bd4d84b23cd3 + version: v1.0.2 subpackages: - codes - credentials From e3558a64cf49e3a8b08233438353851f5badc317 Mon Sep 17 00:00:00 2001 From: Gyu-Ho Lee Date: Mon, 10 Oct 2016 11:05:14 -0700 Subject: [PATCH 4/5] vendor: update grpc-go v1.0.2 tag Fix https://github.com/coreos/etcd/issues/6529. --- cmd/vendor/google.golang.org/grpc/balancer.go | 20 ++- cmd/vendor/google.golang.org/grpc/call.go | 11 +- .../google.golang.org/grpc/clientconn.go | 125 ++++++++++++------ .../grpc/credentials/credentials.go | 49 ++++--- .../google.golang.org/grpc/interceptor.go | 16 +++ .../grpc/metadata/metadata.go | 13 +- cmd/vendor/google.golang.org/grpc/rpc_util.go | 4 +- cmd/vendor/google.golang.org/grpc/server.go | 26 ++-- cmd/vendor/google.golang.org/grpc/stream.go | 33 ++++- .../grpc/transport/handler_server.go | 4 +- .../grpc/transport/http2_client.go | 100 +++++++++----- .../grpc/transport/http2_server.go | 12 +- .../grpc/transport/http_util.go | 13 +- .../grpc/transport/transport.go | 31 ++--- 14 files changed, 305 insertions(+), 152 deletions(-) diff --git a/cmd/vendor/google.golang.org/grpc/balancer.go b/cmd/vendor/google.golang.org/grpc/balancer.go index 419e2146113..e217a2077c3 100644 --- a/cmd/vendor/google.golang.org/grpc/balancer.go +++ b/cmd/vendor/google.golang.org/grpc/balancer.go @@ -38,6 +38,7 @@ import ( "sync" "golang.org/x/net/context" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/grpclog" "google.golang.org/grpc/naming" ) @@ -52,6 +53,14 @@ type Address struct { Metadata interface{} } +// BalancerConfig specifies the configurations for Balancer. +type BalancerConfig struct { + // DialCreds is the transport credential the Balancer implementation can + // use to dial to a remote load balancer server. The Balancer implementations + // can ignore this if it does not need to talk to another party securely. + DialCreds credentials.TransportCredentials +} + // BalancerGetOptions configures a Get call. // This is the EXPERIMENTAL API and may be changed or extended in the future. type BalancerGetOptions struct { @@ -66,11 +75,11 @@ 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 // be called when dialing. - Start(target string) error + Start(target string, config BalancerConfig) error // Up informs the Balancer that gRPC has a connection to the server at // addr. It returns down which is called once the connection to addr gets // lost or closed. - // TODO: It is not clear how to construct and take advantage the meaningful error + // TODO: It is not clear how to construct and take advantage of the meaningful error // parameter for down. Need realistic demands to guide. Up(addr Address) (down func(error)) // Get gets the address of a server for the RPC corresponding to ctx. @@ -205,7 +214,12 @@ func (rr *roundRobin) watchAddrUpdates() error { return nil } -func (rr *roundRobin) Start(target string) error { +func (rr *roundRobin) Start(target string, config BalancerConfig) error { + rr.mu.Lock() + defer rr.mu.Unlock() + if rr.done { + return ErrClientConnClosing + } if rr.r == nil { // If there is no name resolver installed, it is not needed to // do name resolution. In this case, target is added into rr.addrs diff --git a/cmd/vendor/google.golang.org/grpc/call.go b/cmd/vendor/google.golang.org/grpc/call.go index fea07998d7b..788b3d92811 100644 --- a/cmd/vendor/google.golang.org/grpc/call.go +++ b/cmd/vendor/google.golang.org/grpc/call.go @@ -96,7 +96,7 @@ func sendRequest(ctx context.Context, codec Codec, compressor Compressor, callHd } outBuf, err := encode(codec, args, compressor, cbuf) if err != nil { - return nil, transport.StreamErrorf(codes.Internal, "grpc: %v", err) + return nil, Errorf(codes.Internal, "grpc: %v", err) } err = t.Write(stream, outBuf, opts) // t.NewStream(...) could lead to an early rejection of the RPC (e.g., the service/method @@ -112,7 +112,14 @@ func sendRequest(ctx context.Context, codec Codec, compressor Compressor, callHd // Invoke sends the RPC request on the wire and returns after response is received. // Invoke is called by generated code. Also users can call Invoke directly when it // is really needed in their use cases. -func Invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) (err error) { +func Invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) error { + if cc.dopts.unaryInt != nil { + return cc.dopts.unaryInt(ctx, method, args, reply, cc, invoke, opts...) + } + return invoke(ctx, method, args, reply, cc, opts...) +} + +func invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) (err error) { c := defaultCallInfo for _, o := range opts { if err := o.before(&c); err != nil { diff --git a/cmd/vendor/google.golang.org/grpc/clientconn.go b/cmd/vendor/google.golang.org/grpc/clientconn.go index 27e74e6f22b..11dce44fd3d 100644 --- a/cmd/vendor/google.golang.org/grpc/clientconn.go +++ b/cmd/vendor/google.golang.org/grpc/clientconn.go @@ -83,15 +83,17 @@ var ( // dialOptions configure a Dial call. dialOptions are set by the DialOption // values passed to Dial. type dialOptions struct { - codec Codec - cp Compressor - dc Decompressor - bs backoffStrategy - balancer Balancer - block bool - insecure bool - timeout time.Duration - copts transport.ConnectOptions + unaryInt UnaryClientInterceptor + streamInt StreamClientInterceptor + codec Codec + cp Compressor + dc Decompressor + bs backoffStrategy + balancer Balancer + block bool + insecure bool + timeout time.Duration + copts transport.ConnectOptions } // DialOption configures how we set up the connection. @@ -215,19 +217,48 @@ func WithUserAgent(s string) DialOption { } } +// WithUnaryInterceptor returns a DialOption that specifies the interceptor for unary RPCs. +func WithUnaryInterceptor(f UnaryClientInterceptor) DialOption { + return func(o *dialOptions) { + o.unaryInt = f + } +} + +// WithStreamInterceptor returns a DialOption that specifies the interceptor for streaming RPCs. +func WithStreamInterceptor(f StreamClientInterceptor) DialOption { + return func(o *dialOptions) { + o.streamInt = f + } +} + // Dial creates a client connection to the given target. func Dial(target string, opts ...DialOption) (*ClientConn, error) { return DialContext(context.Background(), target, opts...) } -// DialContext creates a client connection to the given target -// using the supplied context. -func DialContext(ctx context.Context, target string, opts ...DialOption) (*ClientConn, error) { +// DialContext creates a client connection to the given target. ctx can be used to +// cancel or expire the pending connecting. Once this function returns, the +// cancellation and expiration of ctx will be noop. Users should call ClientConn.Close +// to terminate all the pending operations after this function returns. +// This is the EXPERIMENTAL API. +func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) { cc := &ClientConn{ target: target, conns: make(map[Address]*addrConn), } - cc.ctx, cc.cancel = context.WithCancel(ctx) + cc.ctx, cc.cancel = context.WithCancel(context.Background()) + defer func() { + select { + case <-ctx.Done(): + conn, err = nil, ctx.Err() + default: + } + + if err != nil { + cc.Close() + } + }() + for _, opt := range opts { opt(&cc.dopts) } @@ -239,31 +270,47 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (*Clien if cc.dopts.bs == nil { cc.dopts.bs = DefaultBackoffConfig } - - var ( - ok bool - addrs []Address - ) - if cc.dopts.balancer == nil { - // Connect to target directly if balancer is nil. - addrs = append(addrs, Address{Addr: target}) + creds := cc.dopts.copts.TransportCredentials + if creds != nil && creds.Info().ServerName != "" { + cc.authority = creds.Info().ServerName } else { - if err := cc.dopts.balancer.Start(target); err != nil { - return nil, err + colonPos := strings.LastIndex(target, ":") + if colonPos == -1 { + colonPos = len(target) } - ch := cc.dopts.balancer.Notify() - if ch == nil { - // There is no name resolver installed. + cc.authority = target[:colonPos] + } + var ok bool + waitC := make(chan error, 1) + go func() { + var addrs []Address + if cc.dopts.balancer == nil { + // Connect to target directly if balancer is nil. addrs = append(addrs, Address{Addr: target}) } else { - addrs, ok = <-ch - if !ok || len(addrs) == 0 { - return nil, errNoAddr + var credsClone credentials.TransportCredentials + if creds != nil { + credsClone = creds.Clone() + } + config := BalancerConfig{ + DialCreds: credsClone, + } + if err := cc.dopts.balancer.Start(target, config); err != nil { + waitC <- err + return + } + ch := cc.dopts.balancer.Notify() + if ch == nil { + // There is no name resolver installed. + addrs = append(addrs, Address{Addr: target}) + } else { + addrs, ok = <-ch + if !ok || len(addrs) == 0 { + waitC <- errNoAddr + return + } } } - } - waitC := make(chan error, 1) - go func() { for _, a := range addrs { if err := cc.resetAddrConn(a, false, nil); err != nil { waitC <- err @@ -277,16 +324,13 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (*Clien timeoutCh = time.After(cc.dopts.timeout) } select { + case <-ctx.Done(): + return nil, ctx.Err() case err := <-waitC: if err != nil { - cc.Close() return nil, err } - case <-cc.ctx.Done(): - cc.Close() - return nil, cc.ctx.Err() case <-timeoutCh: - cc.Close() return nil, ErrClientConnTimeout } // If balancer is nil or balancer.Notify() is nil, ok will be false here. @@ -294,11 +338,6 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (*Clien if ok { go cc.lbWatcher() } - colonPos := strings.LastIndex(target, ":") - if colonPos == -1 { - colonPos = len(target) - } - cc.authority = target[:colonPos] return cc, nil } @@ -652,7 +691,7 @@ func (ac *addrConn) resetTransport(closeTransport bool) error { if e, ok := err.(transport.ConnectionError); ok && !e.Temporary() { return err } - grpclog.Printf("grpc: addrConn.resetTransport failed to create client transport: %v; Reconnecting to %q", err, ac.addr) + grpclog.Printf("grpc: addrConn.resetTransport failed to create client transport: %v; Reconnecting to %v", err, ac.addr) ac.mu.Lock() if ac.state == Shutdown { // ac.tearDown(...) has been invoked. diff --git a/cmd/vendor/google.golang.org/grpc/credentials/credentials.go b/cmd/vendor/google.golang.org/grpc/credentials/credentials.go index 3f17b70628e..5555ef024f6 100644 --- a/cmd/vendor/google.golang.org/grpc/credentials/credentials.go +++ b/cmd/vendor/google.golang.org/grpc/credentials/credentials.go @@ -40,6 +40,7 @@ package credentials // import "google.golang.org/grpc/credentials" import ( "crypto/tls" "crypto/x509" + "errors" "fmt" "io/ioutil" "net" @@ -71,7 +72,7 @@ type PerRPCCredentials interface { } // ProtocolInfo provides information regarding the gRPC wire protocol version, -// security protocol, security protocol version in use, etc. +// security protocol, security protocol version in use, server name, etc. type ProtocolInfo struct { // ProtocolVersion is the gRPC wire protocol version. ProtocolVersion string @@ -79,6 +80,8 @@ type ProtocolInfo struct { SecurityProtocol string // SecurityVersion is the security protocol version. SecurityVersion string + // ServerName is the user-configured server name. + ServerName string } // AuthInfo defines the common interface for the auth information the users are interested in. @@ -86,6 +89,12 @@ type AuthInfo interface { AuthType() string } +var ( + // ErrConnDispatched indicates that rawConn has been dispatched out of gRPC + // and the caller should not close rawConn. + ErrConnDispatched = errors.New("credentials: rawConn is dispatched out of gRPC") +) + // TransportCredentials defines the common interface for all the live gRPC wire // protocols and supported transport security protocols (e.g., TLS, SSL). type TransportCredentials interface { @@ -100,6 +109,12 @@ type TransportCredentials interface { ServerHandshake(net.Conn) (net.Conn, AuthInfo, error) // Info provides the ProtocolInfo of this TransportCredentials. Info() ProtocolInfo + // Clone makes a copy of this TransportCredentials. + Clone() TransportCredentials + // OverrideServerName overrides the server name used to verify the hostname on the returned certificates from the server. + // gRPC internals also use it to override the virtual hosting name if it is set. + // It must be called before dialing. Currently, this is only used by grpclb. + OverrideServerName(string) error } // TLSInfo contains the auth information for a TLS authenticated connection. @@ -123,19 +138,10 @@ func (c tlsCreds) Info() ProtocolInfo { return ProtocolInfo{ SecurityProtocol: "tls", SecurityVersion: "1.2", + ServerName: c.config.ServerName, } } -// GetRequestMetadata returns nil, nil since TLS credentials does not have -// metadata. -func (c *tlsCreds) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) { - return nil, nil -} - -func (c *tlsCreds) RequireTransportSecurity() bool { - return true -} - func (c *tlsCreds) ClientHandshake(ctx context.Context, addr string, rawConn net.Conn) (_ net.Conn, _ AuthInfo, err error) { // use local cfg to avoid clobbering ServerName if using multiple endpoints cfg := cloneTLSConfig(c.config) @@ -172,6 +178,15 @@ func (c *tlsCreds) ServerHandshake(rawConn net.Conn) (net.Conn, AuthInfo, error) return conn, TLSInfo{conn.ConnectionState()}, nil } +func (c *tlsCreds) Clone() TransportCredentials { + return NewTLS(c.config) +} + +func (c *tlsCreds) OverrideServerName(serverNameOverride string) error { + c.config.ServerName = serverNameOverride + return nil +} + // NewTLS uses c to construct a TransportCredentials based on TLS. func NewTLS(c *tls.Config) TransportCredentials { tc := &tlsCreds{cloneTLSConfig(c)} @@ -180,12 +195,16 @@ func NewTLS(c *tls.Config) TransportCredentials { } // NewClientTLSFromCert constructs a TLS from the input certificate for client. -func NewClientTLSFromCert(cp *x509.CertPool, serverName string) TransportCredentials { - return NewTLS(&tls.Config{ServerName: serverName, RootCAs: cp}) +// serverNameOverride is for testing only. If set to a non empty string, +// it will override the virtual host name of authority (e.g. :authority header field) in requests. +func NewClientTLSFromCert(cp *x509.CertPool, serverNameOverride string) TransportCredentials { + return NewTLS(&tls.Config{ServerName: serverNameOverride, RootCAs: cp}) } // NewClientTLSFromFile constructs a TLS from the input certificate file for client. -func NewClientTLSFromFile(certFile, serverName string) (TransportCredentials, error) { +// serverNameOverride is for testing only. If set to a non empty string, +// it will override the virtual host name of authority (e.g. :authority header field) in requests. +func NewClientTLSFromFile(certFile, serverNameOverride string) (TransportCredentials, error) { b, err := ioutil.ReadFile(certFile) if err != nil { return nil, err @@ -194,7 +213,7 @@ func NewClientTLSFromFile(certFile, serverName string) (TransportCredentials, er if !cp.AppendCertsFromPEM(b) { return nil, fmt.Errorf("credentials: failed to append certificates") } - return NewTLS(&tls.Config{ServerName: serverName, RootCAs: cp}), nil + return NewTLS(&tls.Config{ServerName: serverNameOverride, RootCAs: cp}), nil } // NewServerTLSFromCert constructs a TLS from the input certificate for server. diff --git a/cmd/vendor/google.golang.org/grpc/interceptor.go b/cmd/vendor/google.golang.org/grpc/interceptor.go index 588f59e5abf..8d932efed7e 100644 --- a/cmd/vendor/google.golang.org/grpc/interceptor.go +++ b/cmd/vendor/google.golang.org/grpc/interceptor.go @@ -37,6 +37,22 @@ import ( "golang.org/x/net/context" ) +// UnaryInvoker is called by UnaryClientInterceptor to complete RPCs. +type UnaryInvoker func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error + +// UnaryClientInterceptor intercepts the execution of a unary RPC on the client. inovker is the handler to complete the RPC +// and it is the responsibility of the interceptor to call it. +// This is the EXPERIMENTAL API. +type UnaryClientInterceptor func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error + +// Streamer is called by StreamClientInterceptor to create a ClientStream. +type Streamer func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) + +// StreamClientInterceptor intercepts the creation of ClientStream. It may return a custom ClientStream to intercept all I/O +// operations. streamer is the handlder to create a ClientStream and it is the responsibility of the interceptor to call it. +// This is the EXPERIMENTAL API. +type StreamClientInterceptor func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error) + // UnaryServerInfo consists of various information about a unary RPC on // server side. All per-rpc information may be mutated by the interceptor. type UnaryServerInfo struct { diff --git a/cmd/vendor/google.golang.org/grpc/metadata/metadata.go b/cmd/vendor/google.golang.org/grpc/metadata/metadata.go index 954c0f77c6c..3c0ca7a36c5 100644 --- a/cmd/vendor/google.golang.org/grpc/metadata/metadata.go +++ b/cmd/vendor/google.golang.org/grpc/metadata/metadata.go @@ -117,10 +117,17 @@ func (md MD) Len() int { // Copy returns a copy of md. func (md MD) Copy() MD { + return Join(md) +} + +// Join joins any number of MDs into a single MD. +// The order of values for each key is determined by the order in which +// the MDs containing those values are presented to Join. +func Join(mds ...MD) MD { out := MD{} - for k, v := range md { - for _, i := range v { - out[k] = append(out[k], i) + for _, md := range mds { + for k, v := range md { + out[k] = append(out[k], v...) } } return out diff --git a/cmd/vendor/google.golang.org/grpc/rpc_util.go b/cmd/vendor/google.golang.org/grpc/rpc_util.go index 35ac9cc7b00..6b60095d564 100644 --- a/cmd/vendor/google.golang.org/grpc/rpc_util.go +++ b/cmd/vendor/google.golang.org/grpc/rpc_util.go @@ -303,10 +303,10 @@ func checkRecvPayload(pf payloadFormat, recvCompress string, dc Decompressor) er case compressionNone: case compressionMade: if dc == nil || recvCompress != dc.Type() { - return transport.StreamErrorf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress) + return Errorf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress) } default: - return transport.StreamErrorf(codes.Internal, "grpc: received unexpected payload format %d", pf) + return Errorf(codes.Internal, "grpc: received unexpected payload format %d", pf) } return nil } diff --git a/cmd/vendor/google.golang.org/grpc/server.go b/cmd/vendor/google.golang.org/grpc/server.go index 1ed8aac9eb0..debbd79aede 100644 --- a/cmd/vendor/google.golang.org/grpc/server.go +++ b/cmd/vendor/google.golang.org/grpc/server.go @@ -324,7 +324,7 @@ func (s *Server) useTransportAuthenticator(rawConn net.Conn) (net.Conn, credenti // Serve accepts incoming connections on the listener lis, creating a new // ServerTransport and service goroutine for each. The service goroutines // read gRPC requests and then call the registered handlers to reply to them. -// Service returns when lis.Accept fails. lis will be closed when +// Serve returns when lis.Accept fails. lis will be closed when // this method returns. func (s *Server) Serve(lis net.Listener) error { s.mu.Lock() @@ -367,7 +367,10 @@ func (s *Server) handleRawConn(rawConn net.Conn) { s.errorf("ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err) s.mu.Unlock() grpclog.Printf("grpc: Server.Serve failed to complete security handshake from %q: %v", rawConn.RemoteAddr(), err) - rawConn.Close() + // If serverHandShake returns ErrConnDispatched, keep rawConn open. + if err != credentials.ErrConnDispatched { + rawConn.Close() + } return } @@ -544,7 +547,7 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport. return err } if err == io.ErrUnexpectedEOF { - err = transport.StreamError{Code: codes.Internal, Desc: "io.ErrUnexpectedEOF"} + err = Errorf(codes.Internal, io.ErrUnexpectedEOF.Error()) } if err != nil { switch err := err.(type) { @@ -566,8 +569,8 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport. if err := checkRecvPayload(pf, stream.RecvCompress(), s.opts.dc); err != nil { switch err := err.(type) { - case transport.StreamError: - if err := t.WriteStatus(stream, err.Code, err.Desc); err != nil { + case *rpcError: + if err := t.WriteStatus(stream, err.code, err.desc); err != nil { grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", err) } default: @@ -870,25 +873,28 @@ func SendHeader(ctx context.Context, md metadata.MD) error { } stream, ok := transport.StreamFromContext(ctx) if !ok { - return fmt.Errorf("grpc: failed to fetch the stream from the context %v", ctx) + return Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx) } t := stream.ServerTransport() if t == nil { grpclog.Fatalf("grpc: SendHeader: %v has no ServerTransport to send header metadata.", stream) } - return t.WriteHeader(stream, md) + if err := t.WriteHeader(stream, md); err != nil { + return toRPCErr(err) + } + return nil } // SetTrailer sets the trailer metadata that will be sent when an RPC returns. -// It may be called at most once from a unary RPC handler. The ctx is the RPC -// handler's Context or one derived from it. +// When called more than once, all the provided metadata will be merged. +// The ctx is the RPC handler's Context or one derived from it. func SetTrailer(ctx context.Context, md metadata.MD) error { if md.Len() == 0 { return nil } stream, ok := transport.StreamFromContext(ctx) if !ok { - return fmt.Errorf("grpc: failed to fetch the stream from the context %v", ctx) + return Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx) } return stream.SetTrailer(md) } diff --git a/cmd/vendor/google.golang.org/grpc/stream.go b/cmd/vendor/google.golang.org/grpc/stream.go index 51df3f01da8..68d777b5098 100644 --- a/cmd/vendor/google.golang.org/grpc/stream.go +++ b/cmd/vendor/google.golang.org/grpc/stream.go @@ -97,7 +97,14 @@ type ClientStream interface { // NewClientStream creates a new Stream for the client side. This is called // by generated code. -func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) { +func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) { + if cc.dopts.streamInt != nil { + return cc.dopts.streamInt(ctx, desc, cc, method, newClientStream, opts...) + } + return newClientStream(ctx, desc, cc, method, opts...) +} + +func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) { var ( t transport.ClientTransport s *transport.Stream @@ -296,7 +303,7 @@ func (cs *clientStream) SendMsg(m interface{}) (err error) { } }() if err != nil { - return transport.StreamErrorf(codes.Internal, "grpc: %v", err) + return Errorf(codes.Internal, "grpc: %v", err) } return cs.t.Write(cs.s, out, &transport.Options{Last: false}) } @@ -407,8 +414,8 @@ type ServerStream interface { // after SendProto. It fails if called multiple times or if // called after SendProto. SendHeader(metadata.MD) error - // SetTrailer sets the trailer metadata which will be sent with the - // RPC status. + // SetTrailer sets the trailer metadata which will be sent with the RPC status. + // When called more than once, all the provided metadata will be merged. SetTrailer(metadata.MD) Stream } @@ -468,10 +475,13 @@ func (ss *serverStream) SendMsg(m interface{}) (err error) { } }() if err != nil { - err = transport.StreamErrorf(codes.Internal, "grpc: %v", err) + err = Errorf(codes.Internal, "grpc: %v", err) return err } - return ss.t.Write(ss.s, out, &transport.Options{Last: false}) + if err := ss.t.Write(ss.s, out, &transport.Options{Last: false}); err != nil { + return toRPCErr(err) + } + return nil } func (ss *serverStream) RecvMsg(m interface{}) (err error) { @@ -489,5 +499,14 @@ func (ss *serverStream) RecvMsg(m interface{}) (err error) { ss.mu.Unlock() } }() - return recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxMsgSize) + if err := recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxMsgSize); err != nil { + if err == io.EOF { + return err + } + if err == io.ErrUnexpectedEOF { + err = Errorf(codes.Internal, io.ErrUnexpectedEOF.Error()) + } + return toRPCErr(err) + } + return nil } diff --git a/cmd/vendor/google.golang.org/grpc/transport/handler_server.go b/cmd/vendor/google.golang.org/grpc/transport/handler_server.go index 30e21ac0f43..114e34906a1 100644 --- a/cmd/vendor/google.golang.org/grpc/transport/handler_server.go +++ b/cmd/vendor/google.golang.org/grpc/transport/handler_server.go @@ -85,7 +85,7 @@ func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request) (ServerTr if v := r.Header.Get("grpc-timeout"); v != "" { to, err := decodeTimeout(v) if err != nil { - return nil, StreamErrorf(codes.Internal, "malformed time-out: %v", err) + return nil, streamErrorf(codes.Internal, "malformed time-out: %v", err) } st.timeoutSet = true st.timeout = to @@ -393,5 +393,5 @@ func mapRecvMsgError(err error) error { } } } - return ConnectionError{Desc: err.Error()} + return connectionErrorf(true, err, err.Error()) } diff --git a/cmd/vendor/google.golang.org/grpc/transport/http2_client.go b/cmd/vendor/google.golang.org/grpc/transport/http2_client.go index 5819cb8a43b..3c185541a54 100644 --- a/cmd/vendor/google.golang.org/grpc/transport/http2_client.go +++ b/cmd/vendor/google.golang.org/grpc/transport/http2_client.go @@ -114,14 +114,42 @@ func dial(fn func(context.Context, string) (net.Conn, error), ctx context.Contex return dialContext(ctx, "tcp", addr) } +func isTemporary(err error) bool { + switch err { + case io.EOF: + // Connection closures may be resolved upon retry, and are thus + // treated as temporary. + return true + case context.DeadlineExceeded: + // In Go 1.7, context.DeadlineExceeded implements Timeout(), and this + // special case is not needed. Until then, we need to keep this + // clause. + return true + } + + switch err := err.(type) { + case interface { + Temporary() bool + }: + return err.Temporary() + case interface { + Timeout() bool + }: + // Timeouts may be resolved upon retry, and are thus treated as + // temporary. + return err.Timeout() + } + return false +} + // newHTTP2Client constructs a connected ClientTransport to addr based on HTTP2 // and starts to receive messages on it. Non-nil error returns if construction // fails. func newHTTP2Client(ctx context.Context, addr string, opts ConnectOptions) (_ ClientTransport, err error) { scheme := "http" - conn, connErr := dial(opts.Dialer, ctx, addr) - if connErr != nil { - return nil, ConnectionErrorf(true, connErr, "transport: %v", connErr) + conn, err := dial(opts.Dialer, ctx, addr) + if err != nil { + return nil, connectionErrorf(true, err, "transport: %v", err) } // Any further errors will close the underlying connection defer func(conn net.Conn) { @@ -132,12 +160,13 @@ func newHTTP2Client(ctx context.Context, addr string, opts ConnectOptions) (_ Cl var authInfo credentials.AuthInfo if creds := opts.TransportCredentials; creds != nil { scheme = "https" - conn, authInfo, connErr = creds.ClientHandshake(ctx, addr, conn) - } - if connErr != nil { - // Credentials handshake error is not a temporary error (unless the error - // was the connection closing). - return nil, ConnectionErrorf(connErr == io.EOF, connErr, "transport: %v", connErr) + conn, authInfo, err = creds.ClientHandshake(ctx, addr, conn) + if err != nil { + // Credentials handshake errors are typically considered permanent + // to avoid retrying on e.g. bad certificates. + temp := isTemporary(err) + return nil, connectionErrorf(temp, err, "transport: %v", err) + } } ua := primaryUA if opts.UserAgent != "" { @@ -176,11 +205,11 @@ func newHTTP2Client(ctx context.Context, addr string, opts ConnectOptions) (_ Cl n, err := t.conn.Write(clientPreface) if err != nil { t.Close() - return nil, ConnectionErrorf(true, err, "transport: %v", err) + return nil, connectionErrorf(true, err, "transport: %v", err) } if n != len(clientPreface) { t.Close() - return nil, ConnectionErrorf(true, err, "transport: preface mismatch, wrote %d bytes; want %d", n, len(clientPreface)) + return nil, connectionErrorf(true, err, "transport: preface mismatch, wrote %d bytes; want %d", n, len(clientPreface)) } if initialWindowSize != defaultWindowSize { err = t.framer.writeSettings(true, http2.Setting{ @@ -192,13 +221,13 @@ func newHTTP2Client(ctx context.Context, addr string, opts ConnectOptions) (_ Cl } if err != nil { t.Close() - return nil, ConnectionErrorf(true, err, "transport: %v", err) + return nil, connectionErrorf(true, err, "transport: %v", err) } // Adjust the connection flow control window if needed. if delta := uint32(initialConnWindowSize - defaultWindowSize); delta > 0 { if err := t.framer.writeWindowUpdate(true, 0, delta); err != nil { t.Close() - return nil, ConnectionErrorf(true, err, "transport: %v", err) + return nil, connectionErrorf(true, err, "transport: %v", err) } } go t.controller() @@ -223,8 +252,10 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream { s.windowHandler = func(n int) { t.updateWindow(s, uint32(n)) } - // Make a stream be able to cancel the pending operations by itself. - s.ctx, s.cancel = context.WithCancel(ctx) + // The client side stream context should have exactly the same life cycle with the user provided context. + // That means, s.ctx should be read-only. And s.ctx is done iff ctx is done. + // So we use the original context here instead of creating a copy. + s.ctx = ctx s.dec = &recvBufferReader{ ctx: s.ctx, goAway: s.goAway, @@ -236,16 +267,6 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream { // NewStream creates a stream and register it into the transport as "active" // streams. func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) { - // Record the timeout value on the context. - var timeout time.Duration - if dl, ok := ctx.Deadline(); ok { - timeout = dl.Sub(time.Now()) - } - select { - case <-ctx.Done(): - return nil, ContextErr(ctx.Err()) - default: - } pr := &peer.Peer{ Addr: t.conn.RemoteAddr(), } @@ -266,12 +287,12 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea } pos := strings.LastIndex(callHdr.Method, "/") if pos == -1 { - return nil, StreamErrorf(codes.InvalidArgument, "transport: malformed method name: %q", callHdr.Method) + return nil, streamErrorf(codes.InvalidArgument, "transport: malformed method name: %q", callHdr.Method) } audience := "https://" + callHdr.Host + port + callHdr.Method[:pos] data, err := c.GetRequestMetadata(ctx, audience) if err != nil { - return nil, StreamErrorf(codes.InvalidArgument, "transport: %v", err) + return nil, streamErrorf(codes.InvalidArgument, "transport: %v", err) } for k, v := range data { authData[k] = v @@ -352,9 +373,12 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea if callHdr.SendCompress != "" { t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-encoding", Value: callHdr.SendCompress}) } - if timeout > 0 { + if dl, ok := ctx.Deadline(); ok { + // Send out timeout regardless its value. The server can detect timeout context by itself. + timeout := dl.Sub(time.Now()) t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-timeout", Value: encodeTimeout(timeout)}) } + for k, v := range authData { // Capital header names are illegal in HTTP/2. k = strings.ToLower(k) @@ -408,7 +432,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea } if err != nil { t.notifyError(err) - return nil, ConnectionErrorf(true, err, "transport: %v", err) + return nil, connectionErrorf(true, err, "transport: %v", err) } } t.writableChan <- 0 @@ -454,7 +478,7 @@ func (t *http2Client) CloseStream(s *Stream, err error) { } s.state = streamDone s.mu.Unlock() - if _, ok := err.(StreamError); ok { + if se, ok := err.(StreamError); ok && se.Code != codes.DeadlineExceeded { t.controlBuf.put(&resetStream{s.id, http2.ErrCodeCancel}) } } @@ -622,7 +646,7 @@ func (t *http2Client) Write(s *Stream, data []byte, opts *Options) error { // invoked. if err := t.framer.writeData(forceFlush, s.id, endStream, p); err != nil { t.notifyError(err) - return ConnectionErrorf(true, err, "transport: %v", err) + return connectionErrorf(true, err, "transport: %v", err) } if t.framer.adjustNumWriters(-1) == 0 { t.framer.flushWrite() @@ -670,7 +694,7 @@ func (t *http2Client) updateWindow(s *Stream, n uint32) { func (t *http2Client) handleData(f *http2.DataFrame) { size := len(f.Data()) if err := t.fc.onData(uint32(size)); err != nil { - t.notifyError(ConnectionErrorf(true, err, "%v", err)) + t.notifyError(connectionErrorf(true, err, "%v", err)) return } // Select the right stream to dispatch. @@ -776,7 +800,7 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) { if t.state == reachable || t.state == draining { if f.LastStreamID > 0 && f.LastStreamID%2 != 1 { t.mu.Unlock() - t.notifyError(ConnectionErrorf(true, nil, "received illegal http2 GOAWAY frame: stream ID %d is even", f.LastStreamID)) + t.notifyError(connectionErrorf(true, nil, "received illegal http2 GOAWAY frame: stream ID %d is even", f.LastStreamID)) return } select { @@ -785,7 +809,7 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) { // t.goAway has been closed (i.e.,multiple GoAways). if id < f.LastStreamID { t.mu.Unlock() - t.notifyError(ConnectionErrorf(true, nil, "received illegal http2 GOAWAY frame: previously recv GOAWAY frame with LastStramID %d, currently recv %d", id, f.LastStreamID)) + t.notifyError(connectionErrorf(true, nil, "received illegal http2 GOAWAY frame: previously recv GOAWAY frame with LastStramID %d, currently recv %d", id, f.LastStreamID)) return } t.prevGoAwayID = id @@ -823,6 +847,12 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) { state.processHeaderField(hf) } if state.err != nil { + s.mu.Lock() + if !s.headerDone { + close(s.headerChan) + s.headerDone = true + } + s.mu.Unlock() s.write(recvMsg{err: state.err}) // Something wrong. Stops reading even when there is remaining. return @@ -900,7 +930,7 @@ func (t *http2Client) reader() { t.mu.Unlock() if s != nil { // use error detail to provide better err message - handleMalformedHTTP2(s, StreamErrorf(http2ErrConvTab[se.Code], "%v", t.framer.errorDetail())) + handleMalformedHTTP2(s, streamErrorf(http2ErrConvTab[se.Code], "%v", t.framer.errorDetail())) } continue } else { diff --git a/cmd/vendor/google.golang.org/grpc/transport/http2_server.go b/cmd/vendor/google.golang.org/grpc/transport/http2_server.go index 16010d55fb2..f753c4f1ead 100644 --- a/cmd/vendor/google.golang.org/grpc/transport/http2_server.go +++ b/cmd/vendor/google.golang.org/grpc/transport/http2_server.go @@ -111,12 +111,12 @@ func newHTTP2Server(conn net.Conn, maxStreams uint32, authInfo credentials.AuthI Val: uint32(initialWindowSize)}) } if err := framer.writeSettings(true, settings...); err != nil { - return nil, ConnectionErrorf(true, err, "transport: %v", err) + return nil, connectionErrorf(true, err, "transport: %v", err) } // Adjust the connection flow control window if needed. if delta := uint32(initialConnWindowSize - defaultWindowSize); delta > 0 { if err := framer.writeWindowUpdate(true, 0, delta); err != nil { - return nil, ConnectionErrorf(true, err, "transport: %v", err) + return nil, connectionErrorf(true, err, "transport: %v", err) } } var buf bytes.Buffer @@ -448,7 +448,7 @@ func (t *http2Server) writeHeaders(s *Stream, b *bytes.Buffer, endStream bool) e } if err != nil { t.Close() - return ConnectionErrorf(true, err, "transport: %v", err) + return connectionErrorf(true, err, "transport: %v", err) } } return nil @@ -544,7 +544,7 @@ func (t *http2Server) Write(s *Stream, data []byte, opts *Options) error { s.mu.Lock() if s.state == streamDone { s.mu.Unlock() - return StreamErrorf(codes.Unknown, "the stream has been done") + return streamErrorf(codes.Unknown, "the stream has been done") } if !s.headerOk { writeHeaderFrame = true @@ -568,7 +568,7 @@ func (t *http2Server) Write(s *Stream, data []byte, opts *Options) error { } if err := t.framer.writeHeaders(false, p); err != nil { t.Close() - return ConnectionErrorf(true, err, "transport: %v", err) + return connectionErrorf(true, err, "transport: %v", err) } t.writableChan <- 0 } @@ -642,7 +642,7 @@ func (t *http2Server) Write(s *Stream, data []byte, opts *Options) error { } if err := t.framer.writeData(forceFlush, s.id, false, p); err != nil { t.Close() - return ConnectionErrorf(true, err, "transport: %v", err) + return connectionErrorf(true, err, "transport: %v", err) } if t.framer.adjustNumWriters(-1) == 0 { t.framer.flushWrite() diff --git a/cmd/vendor/google.golang.org/grpc/transport/http_util.go b/cmd/vendor/google.golang.org/grpc/transport/http_util.go index 3e16e4df42e..a3c68d4cac7 100644 --- a/cmd/vendor/google.golang.org/grpc/transport/http_util.go +++ b/cmd/vendor/google.golang.org/grpc/transport/http_util.go @@ -53,7 +53,7 @@ import ( const ( // The primary user agent - primaryUA = "grpc-go/0.11" + primaryUA = "grpc-go/1.0" // http2MaxFrameLen specifies the max length of a HTTP2 frame. http2MaxFrameLen = 16384 // 16KB frame // http://http2.github.io/http2-spec/#SettingValues @@ -162,7 +162,7 @@ func (d *decodeState) processHeaderField(f hpack.HeaderField) { switch f.Name { case "content-type": if !validContentType(f.Value) { - d.setErr(StreamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value)) + d.setErr(streamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value)) return } case "grpc-encoding": @@ -170,7 +170,7 @@ func (d *decodeState) processHeaderField(f hpack.HeaderField) { case "grpc-status": code, err := strconv.Atoi(f.Value) if err != nil { - d.setErr(StreamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)) + d.setErr(streamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)) return } d.statusCode = codes.Code(code) @@ -181,7 +181,7 @@ func (d *decodeState) processHeaderField(f hpack.HeaderField) { var err error d.timeout, err = decodeTimeout(f.Value) if err != nil { - d.setErr(StreamErrorf(codes.Internal, "transport: malformed time-out: %v", err)) + d.setErr(streamErrorf(codes.Internal, "transport: malformed time-out: %v", err)) return } case ":path": @@ -253,6 +253,9 @@ func div(d, r time.Duration) int64 { // TODO(zhaoq): It is the simplistic and not bandwidth efficient. Improve it. func encodeTimeout(t time.Duration) string { + if t <= 0 { + return "0n" + } if d := div(t, time.Nanosecond); d <= maxTimeoutValue { return strconv.FormatInt(d, 10) + "n" } @@ -349,7 +352,7 @@ func decodeGrpcMessageUnchecked(msg string) string { for i := 0; i < lenMsg; i++ { c := msg[i] if c == percentByte && i+2 < lenMsg { - parsed, err := strconv.ParseInt(msg[i+1:i+3], 16, 8) + parsed, err := strconv.ParseUint(msg[i+1:i+3], 16, 8) if err != nil { buf.WriteByte(c) } else { diff --git a/cmd/vendor/google.golang.org/grpc/transport/transport.go b/cmd/vendor/google.golang.org/grpc/transport/transport.go index d59e511372a..3d6b6a6d511 100644 --- a/cmd/vendor/google.golang.org/grpc/transport/transport.go +++ b/cmd/vendor/google.golang.org/grpc/transport/transport.go @@ -39,7 +39,6 @@ package transport // import "google.golang.org/grpc/transport" import ( "bytes" - "errors" "fmt" "io" "net" @@ -169,7 +168,8 @@ type Stream struct { // nil for client side Stream. st ServerTransport // ctx is the associated context of the stream. - ctx context.Context + ctx context.Context + // cancel is always nil for client side Stream. cancel context.CancelFunc // done is closed when the final status arrives. done chan struct{} @@ -286,19 +286,12 @@ func (s *Stream) StatusDesc() string { return s.statusDesc } -// ErrIllegalTrailerSet indicates that the trailer has already been set or it -// is too late to do so. -var ErrIllegalTrailerSet = errors.New("transport: trailer has been set") - // SetTrailer sets the trailer metadata which will be sent with the RPC status -// by the server. This can only be called at most once. Server side only. +// by the server. This can be called multiple times. Server side only. func (s *Stream) SetTrailer(md metadata.MD) error { s.mu.Lock() defer s.mu.Unlock() - if s.trailer != nil { - return ErrIllegalTrailerSet - } - s.trailer = md.Copy() + s.trailer = metadata.Join(s.trailer, md) return nil } @@ -476,16 +469,16 @@ type ServerTransport interface { Drain() } -// StreamErrorf creates an StreamError with the specified error code and description. -func StreamErrorf(c codes.Code, format string, a ...interface{}) StreamError { +// streamErrorf creates an StreamError with the specified error code and description. +func streamErrorf(c codes.Code, format string, a ...interface{}) StreamError { return StreamError{ Code: c, Desc: fmt.Sprintf(format, a...), } } -// ConnectionErrorf creates an ConnectionError with the specified error description. -func ConnectionErrorf(temp bool, e error, format string, a ...interface{}) ConnectionError { +// connectionErrorf creates an ConnectionError with the specified error description. +func connectionErrorf(temp bool, e error, format string, a ...interface{}) ConnectionError { return ConnectionError{ Desc: fmt.Sprintf(format, a...), temp: temp, @@ -522,10 +515,10 @@ func (e ConnectionError) Origin() error { var ( // ErrConnClosing indicates that the transport is closing. - ErrConnClosing = ConnectionError{Desc: "transport is closing", temp: true} + ErrConnClosing = connectionErrorf(true, nil, "transport is closing") // ErrStreamDrain indicates that the stream is rejected by the server because // the server stops accepting new RPCs. - ErrStreamDrain = StreamErrorf(codes.Unavailable, "the server stops accepting new RPCs") + ErrStreamDrain = streamErrorf(codes.Unavailable, "the server stops accepting new RPCs") ) // StreamError is an error that only affects one stream within a connection. @@ -542,9 +535,9 @@ func (e StreamError) Error() string { func ContextErr(err error) StreamError { switch err { case context.DeadlineExceeded: - return StreamErrorf(codes.DeadlineExceeded, "%v", err) + return streamErrorf(codes.DeadlineExceeded, "%v", err) case context.Canceled: - return StreamErrorf(codes.Canceled, "%v", err) + return streamErrorf(codes.Canceled, "%v", err) } panic(fmt.Sprintf("Unexpected error from context packet: %v", err)) } From 4a07bbec59aac20d597085c80420e66ee0cf70d8 Mon Sep 17 00:00:00 2001 From: Gyu-Ho Lee Date: Mon, 10 Oct 2016 11:07:55 -0700 Subject: [PATCH 5/5] clientv3: implement new grpc.Balancer interface --- clientv3/balancer.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clientv3/balancer.go b/clientv3/balancer.go index 011f6cf04a8..b484b97561c 100644 --- a/clientv3/balancer.go +++ b/clientv3/balancer.go @@ -72,7 +72,7 @@ func newSimpleBalancer(eps []string) *simpleBalancer { return sb } -func (b *simpleBalancer) Start(target string) error { return nil } +func (b *simpleBalancer) Start(target string, config grpc.BalancerConfig) error { return nil } func (b *simpleBalancer) ConnectNotify() <-chan struct{} { b.mu.Lock()