summaryrefslogtreecommitdiff
path: root/vendor/google.golang.org/grpc/internal
diff options
context:
space:
mode:
authorMatthew Heon <mheon@redhat.com>2021-12-06 13:55:54 -0500
committerMatthew Heon <mheon@redhat.com>2021-12-06 13:55:54 -0500
commit8dc94f1be5fd410461f08992c9a176e95ef36169 (patch)
tree9a94ac3938e4353cc2ad78014e5e029d7cc95d67 /vendor/google.golang.org/grpc/internal
parent274a76e06864d33470152cad42b81f8501f56b8c (diff)
downloadpodman-8dc94f1be5fd410461f08992c9a176e95ef36169.tar.gz
podman-8dc94f1be5fd410461f08992c9a176e95ef36169.tar.bz2
podman-8dc94f1be5fd410461f08992c9a176e95ef36169.zip
Bump to containers/image v5.17.0
This resolves CVE-2021-41190 Signed-off-by: Matthew Heon <mheon@redhat.com>
Diffstat (limited to 'vendor/google.golang.org/grpc/internal')
-rw-r--r--vendor/google.golang.org/grpc/internal/binarylog/sink.go41
-rw-r--r--vendor/google.golang.org/grpc/internal/channelz/funcs.go2
-rw-r--r--vendor/google.golang.org/grpc/internal/channelz/types_linux.go2
-rw-r--r--vendor/google.golang.org/grpc/internal/channelz/types_nonlinux.go5
-rw-r--r--vendor/google.golang.org/grpc/internal/channelz/util_linux.go2
-rw-r--r--vendor/google.golang.org/grpc/internal/channelz/util_nonlinux.go3
-rw-r--r--vendor/google.golang.org/grpc/internal/credentials/spiffe.go2
-rw-r--r--vendor/google.golang.org/grpc/internal/credentials/spiffe_appengine.go31
-rw-r--r--vendor/google.golang.org/grpc/internal/credentials/syscallconn.go2
-rw-r--r--vendor/google.golang.org/grpc/internal/credentials/syscallconn_appengine.go30
-rw-r--r--vendor/google.golang.org/grpc/internal/credentials/util.go4
-rw-r--r--vendor/google.golang.org/grpc/internal/envconfig/envconfig.go6
-rw-r--r--vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go29
-rw-r--r--vendor/google.golang.org/grpc/internal/resolver/config_selector.go7
-rw-r--r--vendor/google.golang.org/grpc/internal/resolver/dns/dns_resolver.go9
-rw-r--r--vendor/google.golang.org/grpc/internal/resolver/dns/go113.go33
-rw-r--r--vendor/google.golang.org/grpc/internal/serviceconfig/serviceconfig.go4
-rw-r--r--vendor/google.golang.org/grpc/internal/status/status.go14
-rw-r--r--vendor/google.golang.org/grpc/internal/syscall/syscall_linux.go2
-rw-r--r--vendor/google.golang.org/grpc/internal/syscall/syscall_nonlinux.go21
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/controlbuf.go24
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/handler_server.go3
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/http2_client.go212
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/http2_server.go199
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/http_util.go224
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/networktype/networktype.go2
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/transport.go12
-rw-r--r--vendor/google.golang.org/grpc/internal/xds/env/env.go95
28 files changed, 539 insertions, 481 deletions
diff --git a/vendor/google.golang.org/grpc/internal/binarylog/sink.go b/vendor/google.golang.org/grpc/internal/binarylog/sink.go
index 7d7a3056b..c2fdd58b3 100644
--- a/vendor/google.golang.org/grpc/internal/binarylog/sink.go
+++ b/vendor/google.golang.org/grpc/internal/binarylog/sink.go
@@ -69,7 +69,8 @@ type writerSink struct {
func (ws *writerSink) Write(e *pb.GrpcLogEntry) error {
b, err := proto.Marshal(e)
if err != nil {
- grpclogLogger.Infof("binary logging: failed to marshal proto message: %v", err)
+ grpclogLogger.Errorf("binary logging: failed to marshal proto message: %v", err)
+ return err
}
hdr := make([]byte, 4)
binary.BigEndian.PutUint32(hdr, uint32(len(b)))
@@ -85,24 +86,27 @@ func (ws *writerSink) Write(e *pb.GrpcLogEntry) error {
func (ws *writerSink) Close() error { return nil }
type bufferedSink struct {
- mu sync.Mutex
- closer io.Closer
- out Sink // out is built on buf.
- buf *bufio.Writer // buf is kept for flush.
-
- writeStartOnce sync.Once
- writeTicker *time.Ticker
+ mu sync.Mutex
+ closer io.Closer
+ out Sink // out is built on buf.
+ buf *bufio.Writer // buf is kept for flush.
+ flusherStarted bool
+
+ writeTicker *time.Ticker
+ done chan struct{}
}
func (fs *bufferedSink) Write(e *pb.GrpcLogEntry) error {
- // Start the write loop when Write is called.
- fs.writeStartOnce.Do(fs.startFlushGoroutine)
fs.mu.Lock()
+ defer fs.mu.Unlock()
+ if !fs.flusherStarted {
+ // Start the write loop when Write is called.
+ fs.startFlushGoroutine()
+ fs.flusherStarted = true
+ }
if err := fs.out.Write(e); err != nil {
- fs.mu.Unlock()
return err
}
- fs.mu.Unlock()
return nil
}
@@ -113,7 +117,12 @@ const (
func (fs *bufferedSink) startFlushGoroutine() {
fs.writeTicker = time.NewTicker(bufFlushDuration)
go func() {
- for range fs.writeTicker.C {
+ for {
+ select {
+ case <-fs.done:
+ return
+ case <-fs.writeTicker.C:
+ }
fs.mu.Lock()
if err := fs.buf.Flush(); err != nil {
grpclogLogger.Warningf("failed to flush to Sink: %v", err)
@@ -124,10 +133,12 @@ func (fs *bufferedSink) startFlushGoroutine() {
}
func (fs *bufferedSink) Close() error {
+ fs.mu.Lock()
+ defer fs.mu.Unlock()
if fs.writeTicker != nil {
fs.writeTicker.Stop()
}
- fs.mu.Lock()
+ close(fs.done)
if err := fs.buf.Flush(); err != nil {
grpclogLogger.Warningf("failed to flush to Sink: %v", err)
}
@@ -137,7 +148,6 @@ func (fs *bufferedSink) Close() error {
if err := fs.out.Close(); err != nil {
grpclogLogger.Warningf("failed to close the Sink: %v", err)
}
- fs.mu.Unlock()
return nil
}
@@ -155,5 +165,6 @@ func NewBufferedSink(o io.WriteCloser) Sink {
closer: o,
out: newWriterSink(bufW),
buf: bufW,
+ done: make(chan struct{}),
}
}
diff --git a/vendor/google.golang.org/grpc/internal/channelz/funcs.go b/vendor/google.golang.org/grpc/internal/channelz/funcs.go
index f73141393..6d5760d95 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/funcs.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/funcs.go
@@ -630,7 +630,7 @@ func (c *channelMap) GetServerSockets(id int64, startID int64, maxResults int64)
if count == 0 {
end = true
}
- var s []*SocketMetric
+ s := make([]*SocketMetric, 0, len(sks))
for _, ns := range sks {
sm := &SocketMetric{}
sm.SocketData = ns.s.ChannelzMetric()
diff --git a/vendor/google.golang.org/grpc/internal/channelz/types_linux.go b/vendor/google.golang.org/grpc/internal/channelz/types_linux.go
index 692dd6181..1b1c4cce3 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/types_linux.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/types_linux.go
@@ -1,5 +1,3 @@
-// +build !appengine
-
/*
*
* Copyright 2018 gRPC authors.
diff --git a/vendor/google.golang.org/grpc/internal/channelz/types_nonlinux.go b/vendor/google.golang.org/grpc/internal/channelz/types_nonlinux.go
index 19c2fc521..8b06eed1a 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/types_nonlinux.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/types_nonlinux.go
@@ -1,4 +1,5 @@
-// +build !linux appengine
+//go:build !linux
+// +build !linux
/*
*
@@ -37,6 +38,6 @@ type SocketOptionData struct {
// Windows OS doesn't support Socket Option
func (s *SocketOptionData) Getsockopt(fd uintptr) {
once.Do(func() {
- logger.Warning("Channelz: socket options are not supported on non-linux os and appengine.")
+ logger.Warning("Channelz: socket options are not supported on non-linux environments")
})
}
diff --git a/vendor/google.golang.org/grpc/internal/channelz/util_linux.go b/vendor/google.golang.org/grpc/internal/channelz/util_linux.go
index fdf409d55..8d194e44e 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/util_linux.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/util_linux.go
@@ -1,5 +1,3 @@
-// +build linux,!appengine
-
/*
*
* Copyright 2018 gRPC authors.
diff --git a/vendor/google.golang.org/grpc/internal/channelz/util_nonlinux.go b/vendor/google.golang.org/grpc/internal/channelz/util_nonlinux.go
index 8864a0811..837ddc402 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/util_nonlinux.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/util_nonlinux.go
@@ -1,4 +1,5 @@
-// +build !linux appengine
+//go:build !linux
+// +build !linux
/*
*
diff --git a/vendor/google.golang.org/grpc/internal/credentials/spiffe.go b/vendor/google.golang.org/grpc/internal/credentials/spiffe.go
index be70b6cdf..25ade6230 100644
--- a/vendor/google.golang.org/grpc/internal/credentials/spiffe.go
+++ b/vendor/google.golang.org/grpc/internal/credentials/spiffe.go
@@ -1,5 +1,3 @@
-// +build !appengine
-
/*
*
* Copyright 2020 gRPC authors.
diff --git a/vendor/google.golang.org/grpc/internal/credentials/spiffe_appengine.go b/vendor/google.golang.org/grpc/internal/credentials/spiffe_appengine.go
deleted file mode 100644
index af6f57719..000000000
--- a/vendor/google.golang.org/grpc/internal/credentials/spiffe_appengine.go
+++ /dev/null
@@ -1,31 +0,0 @@
-// +build appengine
-
-/*
- *
- * Copyright 2020 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 credentials
-
-import (
- "crypto/tls"
- "net/url"
-)
-
-// SPIFFEIDFromState is a no-op for appengine builds.
-func SPIFFEIDFromState(state tls.ConnectionState) *url.URL {
- return nil
-}
diff --git a/vendor/google.golang.org/grpc/internal/credentials/syscallconn.go b/vendor/google.golang.org/grpc/internal/credentials/syscallconn.go
index f499a614c..2919632d6 100644
--- a/vendor/google.golang.org/grpc/internal/credentials/syscallconn.go
+++ b/vendor/google.golang.org/grpc/internal/credentials/syscallconn.go
@@ -1,5 +1,3 @@
-// +build !appengine
-
/*
*
* Copyright 2018 gRPC authors.
diff --git a/vendor/google.golang.org/grpc/internal/credentials/syscallconn_appengine.go b/vendor/google.golang.org/grpc/internal/credentials/syscallconn_appengine.go
deleted file mode 100644
index a6144cd66..000000000
--- a/vendor/google.golang.org/grpc/internal/credentials/syscallconn_appengine.go
+++ /dev/null
@@ -1,30 +0,0 @@
-// +build appengine
-
-/*
- *
- * Copyright 2018 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 credentials
-
-import (
- "net"
-)
-
-// WrapSyscallConn returns newConn on appengine.
-func WrapSyscallConn(rawConn, newConn net.Conn) net.Conn {
- return newConn
-}
diff --git a/vendor/google.golang.org/grpc/internal/credentials/util.go b/vendor/google.golang.org/grpc/internal/credentials/util.go
index 55664fa46..f792fd22c 100644
--- a/vendor/google.golang.org/grpc/internal/credentials/util.go
+++ b/vendor/google.golang.org/grpc/internal/credentials/util.go
@@ -18,7 +18,9 @@
package credentials
-import "crypto/tls"
+import (
+ "crypto/tls"
+)
const alpnProtoStrH2 = "h2"
diff --git a/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go b/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
index 73931a94b..e766ac04a 100644
--- a/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
+++ b/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
@@ -22,6 +22,8 @@ package envconfig
import (
"os"
"strings"
+
+ xdsenv "google.golang.org/grpc/internal/xds/env"
)
const (
@@ -31,8 +33,8 @@ const (
)
var (
- // Retry is set if retry is explicitly enabled via "GRPC_GO_RETRY=on".
- Retry = strings.EqualFold(os.Getenv(retryStr), "on")
+ // Retry is set if retry is explicitly enabled via "GRPC_GO_RETRY=on" or if XDS retry support is enabled.
+ Retry = strings.EqualFold(os.Getenv(retryStr), "on") || xdsenv.RetrySupport
// TXTErrIgnore is set if TXT errors should be ignored ("GRPC_GO_IGNORE_TXT_ERRORS" is not "false").
TXTErrIgnore = !strings.EqualFold(os.Getenv(txtErrIgnoreStr), "false")
)
diff --git a/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
index 200b115ca..740f83c2b 100644
--- a/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
+++ b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
@@ -31,26 +31,37 @@ var (
mu sync.Mutex
)
+// Int implements rand.Int on the grpcrand global source.
+func Int() int {
+ mu.Lock()
+ defer mu.Unlock()
+ return r.Int()
+}
+
// Int63n implements rand.Int63n on the grpcrand global source.
func Int63n(n int64) int64 {
mu.Lock()
- res := r.Int63n(n)
- mu.Unlock()
- return res
+ defer mu.Unlock()
+ return r.Int63n(n)
}
// Intn implements rand.Intn on the grpcrand global source.
func Intn(n int) int {
mu.Lock()
- res := r.Intn(n)
- mu.Unlock()
- return res
+ defer mu.Unlock()
+ return r.Intn(n)
}
// Float64 implements rand.Float64 on the grpcrand global source.
func Float64() float64 {
mu.Lock()
- res := r.Float64()
- mu.Unlock()
- return res
+ defer mu.Unlock()
+ return r.Float64()
+}
+
+// Uint64 implements rand.Uint64 on the grpcrand global source.
+func Uint64() uint64 {
+ mu.Lock()
+ defer mu.Unlock()
+ return r.Uint64()
}
diff --git a/vendor/google.golang.org/grpc/internal/resolver/config_selector.go b/vendor/google.golang.org/grpc/internal/resolver/config_selector.go
index 5e7f36703..be7e13d58 100644
--- a/vendor/google.golang.org/grpc/internal/resolver/config_selector.go
+++ b/vendor/google.golang.org/grpc/internal/resolver/config_selector.go
@@ -117,9 +117,12 @@ type ClientInterceptor interface {
NewStream(ctx context.Context, ri RPCInfo, done func(), newStream func(ctx context.Context, done func()) (ClientStream, error)) (ClientStream, error)
}
-// ServerInterceptor is unimplementable; do not use.
+// ServerInterceptor is an interceptor for incoming RPC's on gRPC server side.
type ServerInterceptor interface {
- notDefined()
+ // AllowRPC checks if an incoming RPC is allowed to proceed based on
+ // information about connection RPC was received on, and HTTP Headers. This
+ // information will be piped into context.
+ AllowRPC(ctx context.Context) error // TODO: Make this a real interceptor for filters such as rate limiting.
}
type csKeyType string
diff --git a/vendor/google.golang.org/grpc/internal/resolver/dns/dns_resolver.go b/vendor/google.golang.org/grpc/internal/resolver/dns/dns_resolver.go
index 03825bbe7..75301c514 100644
--- a/vendor/google.golang.org/grpc/internal/resolver/dns/dns_resolver.go
+++ b/vendor/google.golang.org/grpc/internal/resolver/dns/dns_resolver.go
@@ -277,18 +277,13 @@ func (d *dnsResolver) lookupSRV() ([]resolver.Address, error) {
return newAddrs, nil
}
-var filterError = func(err error) error {
+func handleDNSError(err error, lookupType string) error {
if dnsErr, ok := err.(*net.DNSError); ok && !dnsErr.IsTimeout && !dnsErr.IsTemporary {
// Timeouts and temporary errors should be communicated to gRPC to
// attempt another DNS query (with backoff). Other errors should be
// suppressed (they may represent the absence of a TXT record).
return nil
}
- return err
-}
-
-func handleDNSError(err error, lookupType string) error {
- err = filterError(err)
if err != nil {
err = fmt.Errorf("dns: %v record lookup error: %v", lookupType, err)
logger.Info(err)
@@ -323,12 +318,12 @@ func (d *dnsResolver) lookupTXT() *serviceconfig.ParseResult {
}
func (d *dnsResolver) lookupHost() ([]resolver.Address, error) {
- var newAddrs []resolver.Address
addrs, err := d.resolver.LookupHost(d.ctx, d.host)
if err != nil {
err = handleDNSError(err, "A")
return nil, err
}
+ newAddrs := make([]resolver.Address, 0, len(addrs))
for _, a := range addrs {
ip, ok := formatIP(a)
if !ok {
diff --git a/vendor/google.golang.org/grpc/internal/resolver/dns/go113.go b/vendor/google.golang.org/grpc/internal/resolver/dns/go113.go
deleted file mode 100644
index 8783a8cf8..000000000
--- a/vendor/google.golang.org/grpc/internal/resolver/dns/go113.go
+++ /dev/null
@@ -1,33 +0,0 @@
-// +build go1.13
-
-/*
- *
- * 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 dns
-
-import "net"
-
-func init() {
- filterError = func(err error) error {
- if dnsErr, ok := err.(*net.DNSError); ok && dnsErr.IsNotFound {
- // The name does not exist; not an error.
- return nil
- }
- return err
- }
-}
diff --git a/vendor/google.golang.org/grpc/internal/serviceconfig/serviceconfig.go b/vendor/google.golang.org/grpc/internal/serviceconfig/serviceconfig.go
index c0634d152..badbdbf59 100644
--- a/vendor/google.golang.org/grpc/internal/serviceconfig/serviceconfig.go
+++ b/vendor/google.golang.org/grpc/internal/serviceconfig/serviceconfig.go
@@ -78,6 +78,7 @@ func (bc *BalancerConfig) UnmarshalJSON(b []byte) error {
return err
}
+ var names []string
for i, lbcfg := range ir {
if len(lbcfg) != 1 {
return fmt.Errorf("invalid loadBalancingConfig: entry %v does not contain exactly 1 policy/config pair: %q", i, lbcfg)
@@ -92,6 +93,7 @@ func (bc *BalancerConfig) UnmarshalJSON(b []byte) error {
for name, jsonCfg = range lbcfg {
}
+ names = append(names, name)
builder := balancer.Get(name)
if builder == nil {
// If the balancer is not registered, move on to the next config.
@@ -120,7 +122,7 @@ func (bc *BalancerConfig) UnmarshalJSON(b []byte) error {
// return. This means we had a loadBalancingConfig slice but did not
// encounter a registered policy. The config is considered invalid in this
// case.
- return fmt.Errorf("invalid loadBalancingConfig: no supported policies found")
+ return fmt.Errorf("invalid loadBalancingConfig: no supported policies found in %v", names)
}
// MethodConfig defines the configuration recommended by the service providers for a
diff --git a/vendor/google.golang.org/grpc/internal/status/status.go b/vendor/google.golang.org/grpc/internal/status/status.go
index 710223b8d..e5c6513ed 100644
--- a/vendor/google.golang.org/grpc/internal/status/status.go
+++ b/vendor/google.golang.org/grpc/internal/status/status.go
@@ -97,7 +97,7 @@ func (s *Status) Err() error {
if s.Code() == codes.OK {
return nil
}
- return &Error{e: s.Proto()}
+ return &Error{s: s}
}
// WithDetails returns a new status with the provided details messages appended to the status.
@@ -136,19 +136,23 @@ func (s *Status) Details() []interface{} {
return details
}
+func (s *Status) String() string {
+ return fmt.Sprintf("rpc error: code = %s desc = %s", s.Code(), s.Message())
+}
+
// Error wraps a pointer of a status proto. It implements error and Status,
// and a nil *Error should never be returned by this package.
type Error struct {
- e *spb.Status
+ s *Status
}
func (e *Error) Error() string {
- return fmt.Sprintf("rpc error: code = %s desc = %s", codes.Code(e.e.GetCode()), e.e.GetMessage())
+ return e.s.String()
}
// GRPCStatus returns the Status represented by se.
func (e *Error) GRPCStatus() *Status {
- return FromProto(e.e)
+ return e.s
}
// Is implements future error.Is functionality.
@@ -158,5 +162,5 @@ func (e *Error) Is(target error) bool {
if !ok {
return false
}
- return proto.Equal(e.e, tse.e)
+ return proto.Equal(e.s.s, tse.s.s)
}
diff --git a/vendor/google.golang.org/grpc/internal/syscall/syscall_linux.go b/vendor/google.golang.org/grpc/internal/syscall/syscall_linux.go
index 4b2964f2a..b3a72276d 100644
--- a/vendor/google.golang.org/grpc/internal/syscall/syscall_linux.go
+++ b/vendor/google.golang.org/grpc/internal/syscall/syscall_linux.go
@@ -1,5 +1,3 @@
-// +build !appengine
-
/*
*
* Copyright 2018 gRPC authors.
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 7913ef1db..999f52cd7 100644
--- a/vendor/google.golang.org/grpc/internal/syscall/syscall_nonlinux.go
+++ b/vendor/google.golang.org/grpc/internal/syscall/syscall_nonlinux.go
@@ -1,4 +1,5 @@
-// +build !linux appengine
+//go:build !linux
+// +build !linux
/*
*
@@ -35,41 +36,41 @@ var logger = grpclog.Component("core")
func log() {
once.Do(func() {
- logger.Info("CPU time info is unavailable on non-linux or appengine environment.")
+ logger.Info("CPU time info is unavailable on non-linux environments.")
})
}
-// 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.
+// GetCPUTime returns the how much CPU time has passed since the start of this
+// process. It always returns 0 under non-linux environments.
func GetCPUTime() int64 {
log()
return 0
}
-// Rusage is an empty struct under non-linux or appengine environment.
+// Rusage is an empty struct under non-linux environments.
type Rusage struct{}
-// GetRusage is a no-op function under non-linux or appengine environment.
+// GetRusage is a no-op function under non-linux environments.
func GetRusage() *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.
+// between two Rusage structs. It a no-op function for non-linux environments.
func CPUTimeDiff(first *Rusage, latest *Rusage) (float64, float64) {
log()
return 0, 0
}
-// SetTCPUserTimeout is a no-op function under non-linux or appengine environments
+// SetTCPUserTimeout is a no-op function under non-linux 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
+// GetTCPUserTimeout is a no-op function under non-linux 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 f63a01376..45532f8ae 100644
--- a/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
+++ b/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
@@ -296,7 +296,7 @@ type controlBuffer struct {
// closed and nilled when transportResponseFrames drops below the
// threshold. Both fields are protected by mu.
transportResponseFrames int
- trfChan atomic.Value // *chan struct{}
+ trfChan atomic.Value // chan struct{}
}
func newControlBuffer(done <-chan struct{}) *controlBuffer {
@@ -310,10 +310,10 @@ func newControlBuffer(done <-chan struct{}) *controlBuffer {
// throttle blocks if there are too many incomingSettings/cleanupStreams in the
// controlbuf.
func (c *controlBuffer) throttle() {
- ch, _ := c.trfChan.Load().(*chan struct{})
+ ch, _ := c.trfChan.Load().(chan struct{})
if ch != nil {
select {
- case <-*ch:
+ case <-ch:
case <-c.done:
}
}
@@ -347,8 +347,7 @@ func (c *controlBuffer) executeAndPut(f func(it interface{}) bool, it cbItem) (b
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.trfChan.Store(make(chan struct{}))
}
}
c.mu.Unlock()
@@ -389,9 +388,9 @@ func (c *controlBuffer) get(block bool) (interface{}, error) {
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))
+ ch := c.trfChan.Load().(chan struct{})
+ close(ch)
+ c.trfChan.Store((chan struct{})(nil))
}
c.transportResponseFrames--
}
@@ -407,7 +406,6 @@ func (c *controlBuffer) get(block bool) (interface{}, error) {
select {
case <-c.ch:
case <-c.done:
- c.finish()
return nil, ErrConnClosing
}
}
@@ -432,6 +430,14 @@ func (c *controlBuffer) finish() {
hdr.onOrphaned(ErrConnClosing)
}
}
+ // In case throttle() is currently in flight, it needs to be unblocked.
+ // Otherwise, the transport may not close, since the transport is closed by
+ // the reader encountering the connection error.
+ ch, _ := c.trfChan.Load().(chan struct{})
+ if ch != nil {
+ close(ch)
+ }
+ c.trfChan.Store((chan struct{})(nil))
c.mu.Unlock()
}
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 05d3871e6..1c3459c2b 100644
--- a/vendor/google.golang.org/grpc/internal/transport/handler_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/handler_server.go
@@ -141,9 +141,8 @@ type serverHandlerTransport struct {
stats stats.Handler
}
-func (ht *serverHandlerTransport) Close() error {
+func (ht *serverHandlerTransport) Close() {
ht.closeOnce.Do(ht.closeCloseChanOnce)
- return nil
}
func (ht *serverHandlerTransport) closeCloseChanOnce() { close(ht.closedCh) }
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 48c5e52ed..755863074 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_client.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
@@ -24,6 +24,7 @@ import (
"io"
"math"
"net"
+ "net/http"
"strconv"
"strings"
"sync"
@@ -241,7 +242,15 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
// and passed to the credential handshaker. This makes it possible for
// address specific arbitrary data to reach the credential handshaker.
connectCtx = icredentials.NewClientHandshakeInfoContext(connectCtx, credentials.ClientHandshakeInfo{Attributes: addr.Attributes})
- conn, authInfo, err = transportCreds.ClientHandshake(connectCtx, addr.ServerName, conn)
+ rawConn := conn
+ // Pull the deadline from the connectCtx, which will be used for
+ // timeouts in the authentication protocol handshake. Can ignore the
+ // boolean as the deadline will return the zero value, which will make
+ // the conn not timeout on I/O operations.
+ deadline, _ := connectCtx.Deadline()
+ rawConn.SetDeadline(deadline)
+ conn, authInfo, err = transportCreds.ClientHandshake(connectCtx, addr.ServerName, rawConn)
+ rawConn.SetDeadline(time.Time{})
if err != nil {
return nil, connectionErrorf(isTemporary(err), err, "transport: authentication handshake failed: %v", err)
}
@@ -399,11 +408,10 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
logger.Errorf("transport: loopyWriter.run returning. Err: %v", err)
}
}
- // If it's a connection error, let reader goroutine handle it
- // since there might be data in the buffers.
- if _, ok := err.(net.Error); !ok {
- t.conn.Close()
- }
+ // Do not close the transport. Let reader goroutine handle it since
+ // there might be data in the buffers.
+ t.conn.Close()
+ t.controlBuf.finish()
close(t.writerDone)
}()
return t, nil
@@ -608,26 +616,35 @@ func (t *http2Client) getCallAuthData(ctx context.Context, audience string, call
return callAuthData, nil
}
-// PerformedIOError wraps an error to indicate IO may have been performed
-// before the error occurred.
-type PerformedIOError struct {
+// NewStreamError wraps an error and reports additional information. Typically
+// NewStream errors result in transparent retry, as they mean nothing went onto
+// the wire. However, there are two notable exceptions:
+//
+// 1. If the stream headers violate the max header list size allowed by the
+// server. In this case there is no reason to retry at all, as it is
+// assumed the RPC would continue to fail on subsequent attempts.
+// 2. If the credentials errored when requesting their headers. In this case,
+// it's possible a retry can fix the problem, but indefinitely transparently
+// retrying is not appropriate as it is likely the credentials, if they can
+// eventually succeed, would need I/O to do so.
+type NewStreamError struct {
Err error
+
+ DoNotRetry bool
+ DoNotTransparentRetry bool
}
-// Error implements error.
-func (p PerformedIOError) Error() string {
- return p.Err.Error()
+func (e NewStreamError) Error() string {
+ return e.Err.Error()
}
// NewStream creates a stream and registers it into the transport as "active"
-// streams.
+// streams. All non-nil errors returned will be *NewStreamError.
func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {
ctx = peer.NewContext(ctx, t.getPeer())
headerFields, err := t.createHeaderFields(ctx, callHdr)
if err != nil {
- // We may have performed I/O in the per-RPC creds callback, so do not
- // allow transparent retry.
- return nil, PerformedIOError{err}
+ return nil, &NewStreamError{Err: err, DoNotTransparentRetry: true}
}
s := t.newStream(ctx, callHdr)
cleanup := func(err error) {
@@ -727,23 +744,23 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
return true
}, hdr)
if err != nil {
- return nil, err
+ return nil, &NewStreamError{Err: err}
}
if success {
break
}
if hdrListSizeErr != nil {
- return nil, hdrListSizeErr
+ return nil, &NewStreamError{Err: hdrListSizeErr, DoNotRetry: true}
}
firstTry = false
select {
case <-ch:
- case <-s.ctx.Done():
- return nil, ContextErr(s.ctx.Err())
+ case <-ctx.Done():
+ return nil, &NewStreamError{Err: ContextErr(ctx.Err())}
case <-t.goAway:
- return nil, errStreamDrain
+ return nil, &NewStreamError{Err: errStreamDrain}
case <-t.ctx.Done():
- return nil, ErrConnClosing
+ return nil, &NewStreamError{Err: ErrConnClosing}
}
}
if t.statsHandler != nil {
@@ -878,12 +895,18 @@ func (t *http2Client) Close(err error) {
// Append info about previous goaways if there were any, since this may be important
// for understanding the root cause for this connection to be closed.
_, goAwayDebugMessage := t.GetGoAwayReason()
+
+ var st *status.Status
if len(goAwayDebugMessage) > 0 {
- err = fmt.Errorf("closing transport due to: %v, received prior goaway: %v", err, goAwayDebugMessage)
+ st = status.Newf(codes.Unavailable, "closing transport due to: %v, received prior goaway: %v", err, goAwayDebugMessage)
+ err = st.Err()
+ } else {
+ st = status.New(codes.Unavailable, err.Error())
}
+
// Notify all active streams.
for _, s := range streams {
- t.closeStream(s, err, false, http2.ErrCodeNo, status.New(codes.Unavailable, err.Error()), nil, false)
+ t.closeStream(s, err, false, http2.ErrCodeNo, st, nil, false)
}
if t.statsHandler != nil {
connEnd := &stats.ConnEnd{
@@ -1221,7 +1244,11 @@ func (t *http2Client) setGoAwayReason(f *http2.GoAwayFrame) {
t.goAwayReason = GoAwayTooManyPings
}
}
- t.goAwayDebugMessage = fmt.Sprintf("code: %s, debug data: %v", f.ErrCode, string(f.DebugData()))
+ if len(f.DebugData()) == 0 {
+ t.goAwayDebugMessage = fmt.Sprintf("code: %s", f.ErrCode)
+ } else {
+ t.goAwayDebugMessage = fmt.Sprintf("code: %s, debug data: %q", f.ErrCode, string(f.DebugData()))
+ }
}
func (t *http2Client) GetGoAwayReason() (GoAwayReason, string) {
@@ -1254,11 +1281,124 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
return
}
- 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 h2code, err := state.decodeHeader(frame); err != nil {
- t.closeStream(s, err, true, h2code, status.Convert(err), nil, endStream)
+ // frame.Truncated is set to true when framer detects that the current header
+ // list size hits MaxHeaderListSize limit.
+ if frame.Truncated {
+ se := status.New(codes.Internal, "peer header list size exceeded limit")
+ t.closeStream(s, se.Err(), true, http2.ErrCodeFrameSize, se, nil, endStream)
+ return
+ }
+
+ var (
+ // If a gRPC Response-Headers has already been received, then it means
+ // that the peer is speaking gRPC and we are in gRPC mode.
+ isGRPC = !initialHeader
+ mdata = make(map[string][]string)
+ contentTypeErr = "malformed header: missing HTTP content-type"
+ grpcMessage string
+ statusGen *status.Status
+ recvCompress string
+ httpStatusCode *int
+ httpStatusErr string
+ rawStatusCode = codes.Unknown
+ // headerError is set if an error is encountered while parsing the headers
+ headerError string
+ )
+
+ if initialHeader {
+ httpStatusErr = "malformed header: missing HTTP status"
+ }
+
+ for _, hf := range frame.Fields {
+ switch hf.Name {
+ case "content-type":
+ if _, validContentType := grpcutil.ContentSubtype(hf.Value); !validContentType {
+ contentTypeErr = fmt.Sprintf("transport: received unexpected content-type %q", hf.Value)
+ break
+ }
+ contentTypeErr = ""
+ mdata[hf.Name] = append(mdata[hf.Name], hf.Value)
+ isGRPC = true
+ case "grpc-encoding":
+ recvCompress = hf.Value
+ case "grpc-status":
+ code, err := strconv.ParseInt(hf.Value, 10, 32)
+ if err != nil {
+ se := status.New(codes.Internal, fmt.Sprintf("transport: malformed grpc-status: %v", err))
+ t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
+ return
+ }
+ rawStatusCode = codes.Code(uint32(code))
+ case "grpc-message":
+ grpcMessage = decodeGrpcMessage(hf.Value)
+ case "grpc-status-details-bin":
+ var err error
+ statusGen, err = decodeGRPCStatusDetails(hf.Value)
+ if err != nil {
+ headerError = fmt.Sprintf("transport: malformed grpc-status-details-bin: %v", err)
+ }
+ case ":status":
+ if hf.Value == "200" {
+ httpStatusErr = ""
+ statusCode := 200
+ httpStatusCode = &statusCode
+ break
+ }
+
+ c, err := strconv.ParseInt(hf.Value, 10, 32)
+ if err != nil {
+ se := status.New(codes.Internal, fmt.Sprintf("transport: malformed http-status: %v", err))
+ t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
+ return
+ }
+ statusCode := int(c)
+ httpStatusCode = &statusCode
+
+ httpStatusErr = fmt.Sprintf(
+ "unexpected HTTP status code received from server: %d (%s)",
+ statusCode,
+ http.StatusText(statusCode),
+ )
+ default:
+ if isReservedHeader(hf.Name) && !isWhitelistedHeader(hf.Name) {
+ break
+ }
+ v, err := decodeMetadataHeader(hf.Name, hf.Value)
+ if err != nil {
+ headerError = fmt.Sprintf("transport: malformed %s: %v", hf.Name, err)
+ logger.Warningf("Failed to decode metadata header (%q, %q): %v", hf.Name, hf.Value, err)
+ break
+ }
+ mdata[hf.Name] = append(mdata[hf.Name], v)
+ }
+ }
+
+ if !isGRPC || httpStatusErr != "" {
+ var code = codes.Internal // when header does not include HTTP status, return INTERNAL
+
+ if httpStatusCode != nil {
+ var ok bool
+ code, ok = HTTPStatusConvTab[*httpStatusCode]
+ if !ok {
+ code = codes.Unknown
+ }
+ }
+ var errs []string
+ if httpStatusErr != "" {
+ errs = append(errs, httpStatusErr)
+ }
+ if contentTypeErr != "" {
+ errs = append(errs, contentTypeErr)
+ }
+ // Verify the HTTP response is a 200.
+ se := status.New(code, strings.Join(errs, "; "))
+ t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
+ return
+ }
+
+ if headerError != "" {
+ se := status.New(codes.Internal, headerError)
+ t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
return
}
@@ -1293,9 +1433,9 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
// 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.data.encoding
- if len(state.data.mdata) > 0 {
- s.header = state.data.mdata
+ s.recvCompress = recvCompress
+ if len(mdata) > 0 {
+ s.header = mdata
}
} else {
// HEADERS frame block carries a Trailers-Only.
@@ -1308,9 +1448,13 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
return
}
+ if statusGen == nil {
+ statusGen = status.New(rawStatusCode, grpcMessage)
+ }
+
// 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.data.mdata, true)
+ t.closeStream(s, io.EOF, rst, http2.ErrCodeNo, statusGen, mdata, true)
}
// reader runs as a separate goroutine in charge of reading data from network
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 11be5599c..19c13e041 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_server.go
@@ -102,11 +102,11 @@ type http2Server struct {
mu sync.Mutex // guard the following
- // drainChan is initialized when drain(...) is called the first time.
+ // drainChan is initialized when Drain() is called the first time.
// After which the server writes out the first GoAway(with ID 2^31-1) frame.
// Then an independent goroutine will be launched to later send the second GoAway.
// During this time we don't want to write another first GoAway(with ID 2^31 -1) frame.
- // Thus call to drain(...) will be a no-op if drainChan is already initialized since draining is
+ // Thus call to Drain() will be a no-op if drainChan is already initialized since draining is
// already underway.
drainChan chan struct{}
state transportState
@@ -125,9 +125,30 @@ type http2Server struct {
connectionID uint64
}
-// newHTTP2Server constructs a ServerTransport based on HTTP2. ConnectionError is
-// returned if something goes wrong.
-func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) {
+// NewServerTransport creates a http2 transport with conn and configuration
+// options from config.
+//
+// It returns a non-nil transport and a nil error on success. On failure, it
+// returns a non-nil transport and a nil-error. For a special case where the
+// underlying conn gets closed before the client preface could be read, it
+// returns a nil transport and a nil error.
+func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) {
+ var authInfo credentials.AuthInfo
+ rawConn := conn
+ if config.Credentials != nil {
+ var err error
+ conn, authInfo, err = config.Credentials.ServerHandshake(rawConn)
+ if err != nil {
+ // ErrConnDispatched means that the connection was dispatched away
+ // from gRPC; those connections should be left open. io.EOF means
+ // the connection was closed before handshaking completed, which can
+ // happen naturally from probers. Return these errors directly.
+ if err == credentials.ErrConnDispatched || err == io.EOF {
+ return nil, err
+ }
+ return nil, connectionErrorf(false, err, "ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err)
+ }
+ }
writeBufSize := config.WriteBufferSize
readBufSize := config.ReadBufferSize
maxHeaderListSize := defaultServerMaxHeaderListSize
@@ -210,14 +231,15 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
if kep.MinTime == 0 {
kep.MinTime = defaultKeepalivePolicyMinTime
}
+
done := make(chan struct{})
t := &http2Server{
- ctx: context.Background(),
+ ctx: setConnection(context.Background(), rawConn),
done: done,
conn: conn,
remoteAddr: conn.RemoteAddr(),
localAddr: conn.LocalAddr(),
- authInfo: config.AuthInfo,
+ authInfo: authInfo,
framer: framer,
readerDone: make(chan struct{}),
writerDone: make(chan struct{}),
@@ -266,6 +288,13 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
// Check the validity of client preface.
preface := make([]byte, len(clientPreface))
if _, err := io.ReadFull(t.conn, preface); err != nil {
+ // In deployments where a gRPC server runs behind a cloud load balancer
+ // which performs regular TCP level health checks, the connection is
+ // closed immediately by the latter. Skipping the error here will help
+ // reduce log clutter.
+ if err == io.EOF {
+ return nil, nil
+ }
return nil, connectionErrorf(false, err, "transport: http2Server.HandleStreams failed to receive the preface from client: %v", err)
}
if !bytes.Equal(preface, clientPreface) {
@@ -295,6 +324,7 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
}
}
t.conn.Close()
+ t.controlBuf.finish()
close(t.writerDone)
}()
go t.keepalive()
@@ -304,37 +334,92 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
// 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,
- }
- if h2code, err := state.decodeHeader(frame); err != nil {
- if _, ok := status.FromError(err); ok {
- t.controlBuf.put(&cleanupStream{
- streamID: streamID,
- rst: true,
- rstCode: h2code,
- onWrite: func() {},
- })
- }
+
+ // frame.Truncated is set to true when framer detects that the current header
+ // list size hits MaxHeaderListSize limit.
+ if frame.Truncated {
+ t.controlBuf.put(&cleanupStream{
+ streamID: streamID,
+ rst: true,
+ rstCode: http2.ErrCodeFrameSize,
+ onWrite: func() {},
+ })
return false
}
buf := newRecvBuffer()
s := &Stream{
- id: streamID,
- st: t,
- buf: buf,
- fc: &inFlow{limit: uint32(t.initialWindowSize)},
- recvCompress: state.data.encoding,
- method: state.data.method,
- contentSubtype: state.data.contentSubtype,
+ id: streamID,
+ st: t,
+ buf: buf,
+ fc: &inFlow{limit: uint32(t.initialWindowSize)},
+ }
+
+ var (
+ // If a gRPC Response-Headers has already been received, then it means
+ // that the peer is speaking gRPC and we are in gRPC mode.
+ isGRPC = false
+ mdata = make(map[string][]string)
+ httpMethod string
+ // headerError is set if an error is encountered while parsing the headers
+ headerError bool
+
+ timeoutSet bool
+ timeout time.Duration
+ )
+
+ for _, hf := range frame.Fields {
+ switch hf.Name {
+ case "content-type":
+ contentSubtype, validContentType := grpcutil.ContentSubtype(hf.Value)
+ if !validContentType {
+ break
+ }
+ mdata[hf.Name] = append(mdata[hf.Name], hf.Value)
+ s.contentSubtype = contentSubtype
+ isGRPC = true
+ case "grpc-encoding":
+ s.recvCompress = hf.Value
+ case ":method":
+ httpMethod = hf.Value
+ case ":path":
+ s.method = hf.Value
+ case "grpc-timeout":
+ timeoutSet = true
+ var err error
+ if timeout, err = decodeTimeout(hf.Value); err != nil {
+ headerError = true
+ }
+ default:
+ if isReservedHeader(hf.Name) && !isWhitelistedHeader(hf.Name) {
+ break
+ }
+ v, err := decodeMetadataHeader(hf.Name, hf.Value)
+ if err != nil {
+ headerError = true
+ logger.Warningf("Failed to decode metadata header (%q, %q): %v", hf.Name, hf.Value, err)
+ break
+ }
+ mdata[hf.Name] = append(mdata[hf.Name], v)
+ }
}
+
+ if !isGRPC || headerError {
+ t.controlBuf.put(&cleanupStream{
+ streamID: streamID,
+ rst: true,
+ rstCode: http2.ErrCodeProtocol,
+ onWrite: func() {},
+ })
+ return false
+ }
+
if frame.StreamEnded() {
// s is just created by the caller. No lock needed.
s.state = streamReadDone
}
- if state.data.timeoutSet {
- s.ctx, s.cancel = context.WithTimeout(t.ctx, state.data.timeout)
+ if timeoutSet {
+ s.ctx, s.cancel = context.WithTimeout(t.ctx, timeout)
} else {
s.ctx, s.cancel = context.WithCancel(t.ctx)
}
@@ -347,14 +432,14 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
}
s.ctx = peer.NewContext(s.ctx, pr)
// Attach the received metadata to the context.
- if len(state.data.mdata) > 0 {
- s.ctx = metadata.NewIncomingContext(s.ctx, state.data.mdata)
- }
- if state.data.statsTags != nil {
- s.ctx = stats.SetIncomingTags(s.ctx, state.data.statsTags)
- }
- if state.data.statsTrace != nil {
- s.ctx = stats.SetIncomingTrace(s.ctx, state.data.statsTrace)
+ if len(mdata) > 0 {
+ s.ctx = metadata.NewIncomingContext(s.ctx, mdata)
+ if statsTags := mdata["grpc-tags-bin"]; len(statsTags) > 0 {
+ s.ctx = stats.SetIncomingTags(s.ctx, []byte(statsTags[len(statsTags)-1]))
+ }
+ if statsTrace := mdata["grpc-trace-bin"]; len(statsTrace) > 0 {
+ s.ctx = stats.SetIncomingTrace(s.ctx, []byte(statsTrace[len(statsTrace)-1]))
+ }
}
t.mu.Lock()
if t.state != reachable {
@@ -383,10 +468,10 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
return true
}
t.maxStreamID = streamID
- if state.data.httpMethod != http.MethodPost {
+ if httpMethod != http.MethodPost {
t.mu.Unlock()
if logger.V(logLevel) {
- logger.Warningf("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", state.data.httpMethod)
+ logger.Infof("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", httpMethod)
}
t.controlBuf.put(&cleanupStream{
streamID: streamID,
@@ -399,7 +484,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
}
if t.inTapHandle != nil {
var err error
- if s.ctx, err = t.inTapHandle(s.ctx, &tap.Info{FullMethodName: state.data.method}); err != nil {
+ if s.ctx, err = t.inTapHandle(s.ctx, &tap.Info{FullMethodName: s.method}); err != nil {
t.mu.Unlock()
if logger.V(logLevel) {
logger.Infof("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
@@ -437,7 +522,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
LocalAddr: t.localAddr,
Compression: s.recvCompress,
WireLength: int(frame.Header().Length),
- Header: metadata.MD(state.data.mdata).Copy(),
+ Header: metadata.MD(mdata).Copy(),
}
t.stats.HandleRPC(s.ctx, inHeader)
}
@@ -1004,12 +1089,12 @@ func (t *http2Server) keepalive() {
if val <= 0 {
// The connection has been idle for a duration of keepalive.MaxConnectionIdle or more.
// Gracefully close the connection.
- t.drain(http2.ErrCodeNo, []byte{})
+ t.Drain()
return
}
idleTimer.Reset(val)
case <-ageTimer.C:
- t.drain(http2.ErrCodeNo, []byte{})
+ t.Drain()
ageTimer.Reset(t.kp.MaxConnectionAgeGrace)
select {
case <-ageTimer.C:
@@ -1063,11 +1148,11 @@ func (t *http2Server) keepalive() {
// Close starts shutting down the http2Server transport.
// TODO(zhaoq): Now the destruction is not blocked on any pending streams. This
// could cause some resource issue. Revisit this later.
-func (t *http2Server) Close() error {
+func (t *http2Server) Close() {
t.mu.Lock()
if t.state == closing {
t.mu.Unlock()
- return errors.New("transport: Close() was already called")
+ return
}
t.state = closing
streams := t.activeStreams
@@ -1075,7 +1160,9 @@ func (t *http2Server) Close() error {
t.mu.Unlock()
t.controlBuf.finish()
close(t.done)
- err := t.conn.Close()
+ if err := t.conn.Close(); err != nil && logger.V(logLevel) {
+ logger.Infof("transport: error closing conn during Close: %v", err)
+ }
if channelz.IsOn() {
channelz.RemoveEntry(t.channelzID)
}
@@ -1087,7 +1174,6 @@ func (t *http2Server) Close() error {
connEnd := &stats.ConnEnd{}
t.stats.HandleConn(t.ctx, connEnd)
}
- return err
}
// deleteStream deletes the stream s from transport's active streams.
@@ -1152,17 +1238,13 @@ func (t *http2Server) RemoteAddr() net.Addr {
}
func (t *http2Server) Drain() {
- t.drain(http2.ErrCodeNo, []byte{})
-}
-
-func (t *http2Server) drain(code http2.ErrCode, debugData []byte) {
t.mu.Lock()
defer t.mu.Unlock()
if t.drainChan != nil {
return
}
t.drainChan = make(chan struct{})
- t.controlBuf.put(&goAway{code: code, debugData: debugData, headsUp: true})
+ t.controlBuf.put(&goAway{code: http2.ErrCodeNo, debugData: []byte{}, headsUp: true})
}
var goAwayPing = &ping{data: [8]byte{1, 6, 1, 8, 0, 3, 3, 9}}
@@ -1280,3 +1362,18 @@ func getJitter(v time.Duration) time.Duration {
j := grpcrand.Int63n(2*r) - r
return time.Duration(j)
}
+
+type connectionKey struct{}
+
+// GetConnection gets the connection from the context.
+func GetConnection(ctx context.Context) net.Conn {
+ conn, _ := ctx.Value(connectionKey{}).(net.Conn)
+ return conn
+}
+
+// SetConnection adds the connection to the context to be able to get
+// information about the destination ip and port for an incoming RPC. This also
+// allows any unary or streaming interceptors to see the connection.
+func setConnection(ctx context.Context, conn net.Conn) context.Context {
+ return context.WithValue(ctx, connectionKey{}, conn)
+}
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 c7dee140c..d8247bcdf 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http_util.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http_util.go
@@ -39,7 +39,6 @@ import (
spb "google.golang.org/genproto/googleapis/rpc/status"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/grpclog"
- "google.golang.org/grpc/internal/grpcutil"
"google.golang.org/grpc/status"
)
@@ -96,53 +95,6 @@ var (
logger = grpclog.Component("transport")
)
-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
- // parsed, use the status method to retrieve the status.
- statusGen *status.Status
- // rawStatusCode and rawStatusMsg are set from the raw trailer fields and are not
- // intended for direct access outside of parsing.
- rawStatusCode *int
- rawStatusMsg string
- httpStatus *int
- // Server side only fields.
- timeoutSet bool
- timeout time.Duration
- method string
- httpMethod string
- // key-value metadata map from the peer.
- mdata map[string][]string
- 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
// reserved by gRPC protocol. Any other headers are classified as the
// user-specified metadata.
@@ -180,14 +132,6 @@ func isWhitelistedHeader(hdr string) bool {
}
}
-func (d *decodeState) status() *status.Status {
- if d.data.statusGen == nil {
- // No status-details were provided; generate status using code/msg.
- d.data.statusGen = status.New(codes.Code(int32(*(d.data.rawStatusCode))), d.data.rawStatusMsg)
- }
- return d.data.statusGen
-}
-
const binHdrSuffix = "-bin"
func encodeBinHeader(v []byte) string {
@@ -217,168 +161,16 @@ func decodeMetadataHeader(k, v string) (string, error) {
return v, nil
}
-func (d *decodeState) decodeHeader(frame *http2.MetaHeadersFrame) (http2.ErrCode, error) {
- // frame.Truncated is set to true when framer detects that the current header
- // list size hits MaxHeaderListSize limit.
- if frame.Truncated {
- return http2.ErrCodeFrameSize, status.Error(codes.Internal, "peer header list size exceeded limit")
- }
-
- for _, hf := range frame.Fields {
- d.processHeaderField(hf)
- }
-
- if d.data.isGRPC {
- if d.data.grpcErr != nil {
- return http2.ErrCodeProtocol, d.data.grpcErr
- }
- if d.serverSide {
- return http2.ErrCodeNo, 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 http2.ErrCodeNo, nil
- }
-
- // HTTP fallback mode
- if d.data.httpErr != nil {
- return http2.ErrCodeProtocol, d.data.httpErr
- }
-
- var (
- code = codes.Internal // when header does not include HTTP status, return INTERNAL
- ok bool
- )
-
- if d.data.httpStatus != nil {
- code, ok = HTTPStatusConvTab[*(d.data.httpStatus)]
- if !ok {
- code = codes.Unknown
- }
- }
-
- return http2.ErrCodeProtocol, 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.data.mdata == nil {
- d.data.mdata = make(map[string][]string)
+func decodeGRPCStatusDetails(rawDetails string) (*status.Status, error) {
+ v, err := decodeBinHeader(rawDetails)
+ if err != nil {
+ return nil, err
}
- d.data.mdata[k] = append(d.data.mdata[k], v)
-}
-
-func (d *decodeState) processHeaderField(f hpack.HeaderField) {
- switch f.Name {
- case "content-type":
- contentSubtype, validContentType := grpcutil.ContentSubtype(f.Value)
- if !validContentType {
- d.data.contentTypeErr = fmt.Sprintf("transport: received the unexpected content-type %q", f.Value)
- return
- }
- 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.data.encoding = f.Value
- case "grpc-status":
- code, err := strconv.Atoi(f.Value)
- if err != nil {
- d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status: %v", err)
- return
- }
- d.data.rawStatusCode = &code
- case "grpc-message":
- d.data.rawStatusMsg = decodeGrpcMessage(f.Value)
- case "grpc-status-details-bin":
- v, err := decodeBinHeader(f.Value)
- if err != nil {
- 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 {
- d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
- return
- }
- d.data.statusGen = status.FromProto(s)
- case "grpc-timeout":
- d.data.timeoutSet = true
- var err error
- 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.data.method = f.Value
- case ":status":
- code, err := strconv.Atoi(f.Value)
- if err != nil {
- d.data.httpErr = status.Errorf(codes.Internal, "transport: malformed http-status: %v", err)
- return
- }
- d.data.httpStatus = &code
- case "grpc-tags-bin":
- v, err := decodeBinHeader(f.Value)
- if err != nil {
- d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-tags-bin: %v", err)
- return
- }
- d.data.statsTags = v
- d.addMetadata(f.Name, string(v))
- case "grpc-trace-bin":
- v, err := decodeBinHeader(f.Value)
- if err != nil {
- d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-trace-bin: %v", err)
- return
- }
- d.data.statsTrace = v
- d.addMetadata(f.Name, string(v))
- case ":method":
- d.data.httpMethod = f.Value
- default:
- if isReservedHeader(f.Name) && !isWhitelistedHeader(f.Name) {
- break
- }
- v, err := decodeMetadataHeader(f.Name, f.Value)
- if err != nil {
- if logger.V(logLevel) {
- logger.Errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
- }
- return
- }
- d.addMetadata(f.Name, v)
+ st := &spb.Status{}
+ if err = proto.Unmarshal(v, st); err != nil {
+ return nil, err
}
+ return status.FromProto(st), nil
}
type timeoutUnit uint8
diff --git a/vendor/google.golang.org/grpc/internal/transport/networktype/networktype.go b/vendor/google.golang.org/grpc/internal/transport/networktype/networktype.go
index 96967428b..7bb53cff1 100644
--- a/vendor/google.golang.org/grpc/internal/transport/networktype/networktype.go
+++ b/vendor/google.golang.org/grpc/internal/transport/networktype/networktype.go
@@ -17,7 +17,7 @@
*/
// Package networktype declares the network type to be used in the default
-// dailer. Attribute of a resolver.Address.
+// dialer. Attribute of a resolver.Address.
package networktype
import (
diff --git a/vendor/google.golang.org/grpc/internal/transport/transport.go b/vendor/google.golang.org/grpc/internal/transport/transport.go
index 6cc1031fd..d3bf65b2b 100644
--- a/vendor/google.golang.org/grpc/internal/transport/transport.go
+++ b/vendor/google.golang.org/grpc/internal/transport/transport.go
@@ -30,6 +30,7 @@ import (
"net"
"sync"
"sync/atomic"
+ "time"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
@@ -518,7 +519,8 @@ const (
// ServerConfig consists of all the configurations to establish a server transport.
type ServerConfig struct {
MaxStreams uint32
- AuthInfo credentials.AuthInfo
+ ConnectionTimeout time.Duration
+ Credentials credentials.TransportCredentials
InTapHandle tap.ServerInHandle
StatsHandler stats.Handler
KeepaliveParams keepalive.ServerParameters
@@ -532,12 +534,6 @@ type ServerConfig struct {
HeaderTableSize *uint32
}
-// NewServerTransport creates a ServerTransport with conn or non-nil error
-// if it fails.
-func NewServerTransport(protocol string, conn net.Conn, config *ServerConfig) (ServerTransport, error) {
- return newHTTP2Server(conn, config)
-}
-
// ConnectOptions covers all relevant options for communicating with the server.
type ConnectOptions struct {
// UserAgent is the application user agent.
@@ -694,7 +690,7 @@ type ServerTransport interface {
// Close tears down the transport. Once it is called, the transport
// should not be accessed any more. All the pending streams and their
// handlers will be terminated asynchronously.
- Close() error
+ Close()
// RemoteAddr returns the remote network address.
RemoteAddr() net.Addr
diff --git a/vendor/google.golang.org/grpc/internal/xds/env/env.go b/vendor/google.golang.org/grpc/internal/xds/env/env.go
new file mode 100644
index 000000000..b171ac91f
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/xds/env/env.go
@@ -0,0 +1,95 @@
+/*
+ *
+ * Copyright 2020 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 env acts a single source of definition for all environment variables
+// related to the xDS implementation in gRPC.
+package env
+
+import (
+ "os"
+ "strings"
+)
+
+const (
+ // BootstrapFileNameEnv is the env variable to set bootstrap file name.
+ // Do not use this and read from env directly. Its value is read and kept in
+ // variable BootstrapFileName.
+ //
+ // When both bootstrap FileName and FileContent are set, FileName is used.
+ BootstrapFileNameEnv = "GRPC_XDS_BOOTSTRAP"
+ // BootstrapFileContentEnv is the env variable to set bootstrapp file
+ // content. Do not use this and read from env directly. Its value is read
+ // and kept in variable BootstrapFileName.
+ //
+ // When both bootstrap FileName and FileContent are set, FileName is used.
+ BootstrapFileContentEnv = "GRPC_XDS_BOOTSTRAP_CONFIG"
+
+ ringHashSupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH"
+ clientSideSecuritySupportEnv = "GRPC_XDS_EXPERIMENTAL_SECURITY_SUPPORT"
+ aggregateAndDNSSupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER"
+ retrySupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_RETRY"
+ rbacSupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_RBAC"
+
+ c2pResolverSupportEnv = "GRPC_EXPERIMENTAL_GOOGLE_C2P_RESOLVER"
+ c2pResolverTestOnlyTrafficDirectorURIEnv = "GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI"
+)
+
+var (
+ // BootstrapFileName holds the name of the file which contains xDS bootstrap
+ // configuration. Users can specify the location of the bootstrap file by
+ // setting the environment variable "GRPC_XDS_BOOTSTRAP".
+ //
+ // When both bootstrap FileName and FileContent are set, FileName is used.
+ BootstrapFileName = os.Getenv(BootstrapFileNameEnv)
+ // BootstrapFileContent holds the content of the xDS bootstrap
+ // configuration. Users can specify the bootstrap config by
+ // setting the environment variable "GRPC_XDS_BOOTSTRAP_CONFIG".
+ //
+ // When both bootstrap FileName and FileContent are set, FileName is used.
+ BootstrapFileContent = os.Getenv(BootstrapFileContentEnv)
+ // RingHashSupport indicates whether ring hash support is enabled, which can
+ // be disabled by setting the environment variable
+ // "GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH" to "false".
+ RingHashSupport = !strings.EqualFold(os.Getenv(ringHashSupportEnv), "false")
+ // ClientSideSecuritySupport is used to control processing of security
+ // configuration on the client-side.
+ //
+ // Note that there is no env var protection for the server-side because we
+ // have a brand new API on the server-side and users explicitly need to use
+ // the new API to get security integration on the server.
+ ClientSideSecuritySupport = !strings.EqualFold(os.Getenv(clientSideSecuritySupportEnv), "false")
+ // AggregateAndDNSSupportEnv indicates whether processing of aggregated
+ // cluster and DNS cluster is enabled, which can be enabled by setting the
+ // environment variable
+ // "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER" to
+ // "true".
+ AggregateAndDNSSupportEnv = strings.EqualFold(os.Getenv(aggregateAndDNSSupportEnv), "true")
+
+ // RetrySupport indicates whether xDS retry is enabled.
+ RetrySupport = !strings.EqualFold(os.Getenv(retrySupportEnv), "false")
+
+ // RBACSupport indicates whether xDS configured RBAC HTTP Filter is enabled.
+ RBACSupport = strings.EqualFold(os.Getenv(rbacSupportEnv), "true")
+
+ // C2PResolverSupport indicates whether support for C2P resolver is enabled.
+ // This can be enabled by setting the environment variable
+ // "GRPC_EXPERIMENTAL_GOOGLE_C2P_RESOLVER" to "true".
+ C2PResolverSupport = strings.EqualFold(os.Getenv(c2pResolverSupportEnv), "true")
+ // C2PResolverTestOnlyTrafficDirectorURI is the TD URI for testing.
+ C2PResolverTestOnlyTrafficDirectorURI = os.Getenv(c2pResolverTestOnlyTrafficDirectorURIEnv)
+)