[VOL-5291] On demand statistics for ONU and OLT
Change-Id: I4850bb0f0d2235122cb0c1bcf835b3672bb34436
Signed-off-by: Akash Reddy Kankanala <akash.kankanala@radisys.com>
diff --git a/vendor/google.golang.org/grpc/internal/balancer/gracefulswitch/gracefulswitch.go b/vendor/google.golang.org/grpc/internal/balancer/gracefulswitch/gracefulswitch.go
new file mode 100644
index 0000000..08666f6
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/balancer/gracefulswitch/gracefulswitch.go
@@ -0,0 +1,384 @@
+/*
+ *
+ * Copyright 2022 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 gracefulswitch implements a graceful switch load balancer.
+package gracefulswitch
+
+import (
+ "errors"
+ "fmt"
+ "sync"
+
+ "google.golang.org/grpc/balancer"
+ "google.golang.org/grpc/balancer/base"
+ "google.golang.org/grpc/connectivity"
+ "google.golang.org/grpc/resolver"
+)
+
+var errBalancerClosed = errors.New("gracefulSwitchBalancer is closed")
+var _ balancer.Balancer = (*Balancer)(nil)
+
+// NewBalancer returns a graceful switch Balancer.
+func NewBalancer(cc balancer.ClientConn, opts balancer.BuildOptions) *Balancer {
+ return &Balancer{
+ cc: cc,
+ bOpts: opts,
+ }
+}
+
+// Balancer is a utility to gracefully switch from one balancer to
+// a new balancer. It implements the balancer.Balancer interface.
+type Balancer struct {
+ bOpts balancer.BuildOptions
+ cc balancer.ClientConn
+
+ // mu protects the following fields and all fields within balancerCurrent
+ // and balancerPending. mu does not need to be held when calling into the
+ // child balancers, as all calls into these children happen only as a direct
+ // result of a call into the gracefulSwitchBalancer, which are also
+ // guaranteed to be synchronous. There is one exception: an UpdateState call
+ // from a child balancer when current and pending are populated can lead to
+ // calling Close() on the current. To prevent that racing with an
+ // UpdateSubConnState from the channel, we hold currentMu during Close and
+ // UpdateSubConnState calls.
+ mu sync.Mutex
+ balancerCurrent *balancerWrapper
+ balancerPending *balancerWrapper
+ closed bool // set to true when this balancer is closed
+
+ // currentMu must be locked before mu. This mutex guards against this
+ // sequence of events: UpdateSubConnState() called, finds the
+ // balancerCurrent, gives up lock, updateState comes in, causes Close() on
+ // balancerCurrent before the UpdateSubConnState is called on the
+ // balancerCurrent.
+ currentMu sync.Mutex
+}
+
+// swap swaps out the current lb with the pending lb and updates the ClientConn.
+// The caller must hold gsb.mu.
+func (gsb *Balancer) swap() {
+ gsb.cc.UpdateState(gsb.balancerPending.lastState)
+ cur := gsb.balancerCurrent
+ gsb.balancerCurrent = gsb.balancerPending
+ gsb.balancerPending = nil
+ go func() {
+ gsb.currentMu.Lock()
+ defer gsb.currentMu.Unlock()
+ cur.Close()
+ }()
+}
+
+// Helper function that checks if the balancer passed in is current or pending.
+// The caller must hold gsb.mu.
+func (gsb *Balancer) balancerCurrentOrPending(bw *balancerWrapper) bool {
+ return bw == gsb.balancerCurrent || bw == gsb.balancerPending
+}
+
+// SwitchTo initializes the graceful switch process, which completes based on
+// connectivity state changes on the current/pending balancer. Thus, the switch
+// process is not complete when this method returns. This method must be called
+// synchronously alongside the rest of the balancer.Balancer methods this
+// Graceful Switch Balancer implements.
+func (gsb *Balancer) SwitchTo(builder balancer.Builder) error {
+ gsb.mu.Lock()
+ if gsb.closed {
+ gsb.mu.Unlock()
+ return errBalancerClosed
+ }
+ bw := &balancerWrapper{
+ gsb: gsb,
+ lastState: balancer.State{
+ ConnectivityState: connectivity.Connecting,
+ Picker: base.NewErrPicker(balancer.ErrNoSubConnAvailable),
+ },
+ subconns: make(map[balancer.SubConn]bool),
+ }
+ balToClose := gsb.balancerPending // nil if there is no pending balancer
+ if gsb.balancerCurrent == nil {
+ gsb.balancerCurrent = bw
+ } else {
+ gsb.balancerPending = bw
+ }
+ gsb.mu.Unlock()
+ balToClose.Close()
+ // This function takes a builder instead of a balancer because builder.Build
+ // can call back inline, and this utility needs to handle the callbacks.
+ newBalancer := builder.Build(bw, gsb.bOpts)
+ if newBalancer == nil {
+ // This is illegal and should never happen; we clear the balancerWrapper
+ // we were constructing if it happens to avoid a potential panic.
+ gsb.mu.Lock()
+ if gsb.balancerPending != nil {
+ gsb.balancerPending = nil
+ } else {
+ gsb.balancerCurrent = nil
+ }
+ gsb.mu.Unlock()
+ return balancer.ErrBadResolverState
+ }
+
+ // This write doesn't need to take gsb.mu because this field never gets read
+ // or written to on any calls from the current or pending. Calls from grpc
+ // to this balancer are guaranteed to be called synchronously, so this
+ // bw.Balancer field will never be forwarded to until this SwitchTo()
+ // function returns.
+ bw.Balancer = newBalancer
+ return nil
+}
+
+// Returns nil if the graceful switch balancer is closed.
+func (gsb *Balancer) latestBalancer() *balancerWrapper {
+ gsb.mu.Lock()
+ defer gsb.mu.Unlock()
+ if gsb.balancerPending != nil {
+ return gsb.balancerPending
+ }
+ return gsb.balancerCurrent
+}
+
+// UpdateClientConnState forwards the update to the latest balancer created.
+func (gsb *Balancer) UpdateClientConnState(state balancer.ClientConnState) error {
+ // The resolver data is only relevant to the most recent LB Policy.
+ balToUpdate := gsb.latestBalancer()
+ if balToUpdate == nil {
+ return errBalancerClosed
+ }
+ // Perform this call without gsb.mu to prevent deadlocks if the child calls
+ // back into the channel. The latest balancer can never be closed during a
+ // call from the channel, even without gsb.mu held.
+ return balToUpdate.UpdateClientConnState(state)
+}
+
+// ResolverError forwards the error to the latest balancer created.
+func (gsb *Balancer) ResolverError(err error) {
+ // The resolver data is only relevant to the most recent LB Policy.
+ balToUpdate := gsb.latestBalancer()
+ if balToUpdate == nil {
+ return
+ }
+ // Perform this call without gsb.mu to prevent deadlocks if the child calls
+ // back into the channel. The latest balancer can never be closed during a
+ // call from the channel, even without gsb.mu held.
+ balToUpdate.ResolverError(err)
+}
+
+// ExitIdle forwards the call to the latest balancer created.
+//
+// If the latest balancer does not support ExitIdle, the subConns are
+// re-connected to manually.
+func (gsb *Balancer) ExitIdle() {
+ balToUpdate := gsb.latestBalancer()
+ if balToUpdate == nil {
+ return
+ }
+ // There is no need to protect this read with a mutex, as the write to the
+ // Balancer field happens in SwitchTo, which completes before this can be
+ // called.
+ if ei, ok := balToUpdate.Balancer.(balancer.ExitIdler); ok {
+ ei.ExitIdle()
+ return
+ }
+ gsb.mu.Lock()
+ defer gsb.mu.Unlock()
+ for sc := range balToUpdate.subconns {
+ sc.Connect()
+ }
+}
+
+// UpdateSubConnState forwards the update to the appropriate child.
+func (gsb *Balancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
+ gsb.currentMu.Lock()
+ defer gsb.currentMu.Unlock()
+ gsb.mu.Lock()
+ // Forward update to the appropriate child. Even if there is a pending
+ // balancer, the current balancer should continue to get SubConn updates to
+ // maintain the proper state while the pending is still connecting.
+ var balToUpdate *balancerWrapper
+ if gsb.balancerCurrent != nil && gsb.balancerCurrent.subconns[sc] {
+ balToUpdate = gsb.balancerCurrent
+ } else if gsb.balancerPending != nil && gsb.balancerPending.subconns[sc] {
+ balToUpdate = gsb.balancerPending
+ }
+ gsb.mu.Unlock()
+ if balToUpdate == nil {
+ // SubConn belonged to a stale lb policy that has not yet fully closed,
+ // or the balancer was already closed.
+ return
+ }
+ balToUpdate.UpdateSubConnState(sc, state)
+}
+
+// Close closes any active child balancers.
+func (gsb *Balancer) Close() {
+ gsb.mu.Lock()
+ gsb.closed = true
+ currentBalancerToClose := gsb.balancerCurrent
+ gsb.balancerCurrent = nil
+ pendingBalancerToClose := gsb.balancerPending
+ gsb.balancerPending = nil
+ gsb.mu.Unlock()
+
+ currentBalancerToClose.Close()
+ pendingBalancerToClose.Close()
+}
+
+// balancerWrapper wraps a balancer.Balancer, and overrides some Balancer
+// methods to help cleanup SubConns created by the wrapped balancer.
+//
+// It implements the balancer.ClientConn interface and is passed down in that
+// capacity to the wrapped balancer. It maintains a set of subConns created by
+// the wrapped balancer and calls from the latter to create/update/remove
+// SubConns update this set before being forwarded to the parent ClientConn.
+// State updates from the wrapped balancer can result in invocation of the
+// graceful switch logic.
+type balancerWrapper struct {
+ balancer.Balancer
+ gsb *Balancer
+
+ lastState balancer.State
+ subconns map[balancer.SubConn]bool // subconns created by this balancer
+}
+
+func (bw *balancerWrapper) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
+ if state.ConnectivityState == connectivity.Shutdown {
+ bw.gsb.mu.Lock()
+ delete(bw.subconns, sc)
+ bw.gsb.mu.Unlock()
+ }
+ // There is no need to protect this read with a mutex, as the write to the
+ // Balancer field happens in SwitchTo, which completes before this can be
+ // called.
+ bw.Balancer.UpdateSubConnState(sc, state)
+}
+
+// Close closes the underlying LB policy and removes the subconns it created. bw
+// must not be referenced via balancerCurrent or balancerPending in gsb when
+// called. gsb.mu must not be held. Does not panic with a nil receiver.
+func (bw *balancerWrapper) Close() {
+ // before Close is called.
+ if bw == nil {
+ return
+ }
+ // There is no need to protect this read with a mutex, as Close() is
+ // impossible to be called concurrently with the write in SwitchTo(). The
+ // callsites of Close() for this balancer in Graceful Switch Balancer will
+ // never be called until SwitchTo() returns.
+ bw.Balancer.Close()
+ bw.gsb.mu.Lock()
+ for sc := range bw.subconns {
+ bw.gsb.cc.RemoveSubConn(sc)
+ }
+ bw.gsb.mu.Unlock()
+}
+
+func (bw *balancerWrapper) UpdateState(state balancer.State) {
+ // Hold the mutex for this entire call to ensure it cannot occur
+ // concurrently with other updateState() calls. This causes updates to
+ // lastState and calls to cc.UpdateState to happen atomically.
+ bw.gsb.mu.Lock()
+ defer bw.gsb.mu.Unlock()
+ bw.lastState = state
+
+ if !bw.gsb.balancerCurrentOrPending(bw) {
+ return
+ }
+
+ if bw == bw.gsb.balancerCurrent {
+ // In the case that the current balancer exits READY, and there is a pending
+ // balancer, you can forward the pending balancer's cached State up to
+ // ClientConn and swap the pending into the current. This is because there
+ // is no reason to gracefully switch from and keep using the old policy as
+ // the ClientConn is not connected to any backends.
+ if state.ConnectivityState != connectivity.Ready && bw.gsb.balancerPending != nil {
+ bw.gsb.swap()
+ return
+ }
+ // Even if there is a pending balancer waiting to be gracefully switched to,
+ // continue to forward current balancer updates to the Client Conn. Ignoring
+ // state + picker from the current would cause undefined behavior/cause the
+ // system to behave incorrectly from the current LB policies perspective.
+ // Also, the current LB is still being used by grpc to choose SubConns per
+ // RPC, and thus should use the most updated form of the current balancer.
+ bw.gsb.cc.UpdateState(state)
+ return
+ }
+ // This method is now dealing with a state update from the pending balancer.
+ // If the current balancer is currently in a state other than READY, the new
+ // policy can be swapped into place immediately. This is because there is no
+ // reason to gracefully switch from and keep using the old policy as the
+ // ClientConn is not connected to any backends.
+ if state.ConnectivityState != connectivity.Connecting || bw.gsb.balancerCurrent.lastState.ConnectivityState != connectivity.Ready {
+ bw.gsb.swap()
+ }
+}
+
+func (bw *balancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
+ bw.gsb.mu.Lock()
+ if !bw.gsb.balancerCurrentOrPending(bw) {
+ bw.gsb.mu.Unlock()
+ return nil, fmt.Errorf("%T at address %p that called NewSubConn is deleted", bw, bw)
+ }
+ bw.gsb.mu.Unlock()
+
+ sc, err := bw.gsb.cc.NewSubConn(addrs, opts)
+ if err != nil {
+ return nil, err
+ }
+ bw.gsb.mu.Lock()
+ if !bw.gsb.balancerCurrentOrPending(bw) { // balancer was closed during this call
+ bw.gsb.cc.RemoveSubConn(sc)
+ bw.gsb.mu.Unlock()
+ return nil, fmt.Errorf("%T at address %p that called NewSubConn is deleted", bw, bw)
+ }
+ bw.subconns[sc] = true
+ bw.gsb.mu.Unlock()
+ return sc, nil
+}
+
+func (bw *balancerWrapper) ResolveNow(opts resolver.ResolveNowOptions) {
+ // Ignore ResolveNow requests from anything other than the most recent
+ // balancer, because older balancers were already removed from the config.
+ if bw != bw.gsb.latestBalancer() {
+ return
+ }
+ bw.gsb.cc.ResolveNow(opts)
+}
+
+func (bw *balancerWrapper) RemoveSubConn(sc balancer.SubConn) {
+ bw.gsb.mu.Lock()
+ if !bw.gsb.balancerCurrentOrPending(bw) {
+ bw.gsb.mu.Unlock()
+ return
+ }
+ bw.gsb.mu.Unlock()
+ bw.gsb.cc.RemoveSubConn(sc)
+}
+
+func (bw *balancerWrapper) UpdateAddresses(sc balancer.SubConn, addrs []resolver.Address) {
+ bw.gsb.mu.Lock()
+ if !bw.gsb.balancerCurrentOrPending(bw) {
+ bw.gsb.mu.Unlock()
+ return
+ }
+ bw.gsb.mu.Unlock()
+ bw.gsb.cc.UpdateAddresses(sc, addrs)
+}
+
+func (bw *balancerWrapper) Target() string {
+ return bw.gsb.cc.Target()
+}
diff --git a/vendor/google.golang.org/grpc/internal/binarylog/binarylog.go b/vendor/google.golang.org/grpc/internal/binarylog/binarylog.go
index 5cc3aed..755fdeb 100644
--- a/vendor/google.golang.org/grpc/internal/binarylog/binarylog.go
+++ b/vendor/google.golang.org/grpc/internal/binarylog/binarylog.go
@@ -28,38 +28,48 @@
"google.golang.org/grpc/internal/grpcutil"
)
-// Logger is the global binary logger. It can be used to get binary logger for
-// each method.
+var grpclogLogger = grpclog.Component("binarylog")
+
+// Logger specifies MethodLoggers for method names with a Log call that
+// takes a context.
+//
+// This is used in the 1.0 release of gcp/observability, and thus must not be
+// deleted or changed.
type Logger interface {
- getMethodLogger(methodName string) *MethodLogger
+ GetMethodLogger(methodName string) MethodLogger
}
// binLogger is the global binary logger for the binary. One of this should be
// built at init time from the configuration (environment variable or flags).
//
-// It is used to get a methodLogger for each individual method.
+// It is used to get a MethodLogger for each individual method.
var binLogger Logger
-var grpclogLogger = grpclog.Component("binarylog")
-
-// SetLogger sets the binarg logger.
+// SetLogger sets the binary logger.
//
// Only call this at init time.
func SetLogger(l Logger) {
binLogger = l
}
-// GetMethodLogger returns the methodLogger for the given methodName.
+// GetLogger gets the binary logger.
+//
+// Only call this at init time.
+func GetLogger() Logger {
+ return binLogger
+}
+
+// GetMethodLogger returns the MethodLogger for the given methodName.
//
// methodName should be in the format of "/service/method".
//
-// Each methodLogger returned by this method is a new instance. This is to
+// Each MethodLogger returned by this method is a new instance. This is to
// generate sequence id within the call.
-func GetMethodLogger(methodName string) *MethodLogger {
+func GetMethodLogger(methodName string) MethodLogger {
if binLogger == nil {
return nil
}
- return binLogger.getMethodLogger(methodName)
+ return binLogger.GetMethodLogger(methodName)
}
func init() {
@@ -68,17 +78,29 @@
binLogger = NewLoggerFromConfigString(configStr)
}
-type methodLoggerConfig struct {
+// MethodLoggerConfig contains the setting for logging behavior of a method
+// logger. Currently, it contains the max length of header and message.
+type MethodLoggerConfig struct {
// Max length of header and message.
- hdr, msg uint64
+ Header, Message uint64
+}
+
+// LoggerConfig contains the config for loggers to create method loggers.
+type LoggerConfig struct {
+ All *MethodLoggerConfig
+ Services map[string]*MethodLoggerConfig
+ Methods map[string]*MethodLoggerConfig
+
+ Blacklist map[string]struct{}
}
type logger struct {
- all *methodLoggerConfig
- services map[string]*methodLoggerConfig
- methods map[string]*methodLoggerConfig
+ config LoggerConfig
+}
- blacklist map[string]struct{}
+// NewLoggerFromConfig builds a logger with the given LoggerConfig.
+func NewLoggerFromConfig(config LoggerConfig) Logger {
+ return &logger{config: config}
}
// newEmptyLogger creates an empty logger. The map fields need to be filled in
@@ -88,83 +110,83 @@
}
// Set method logger for "*".
-func (l *logger) setDefaultMethodLogger(ml *methodLoggerConfig) error {
- if l.all != nil {
+func (l *logger) setDefaultMethodLogger(ml *MethodLoggerConfig) error {
+ if l.config.All != nil {
return fmt.Errorf("conflicting global rules found")
}
- l.all = ml
+ l.config.All = ml
return nil
}
// Set method logger for "service/*".
//
-// New methodLogger with same service overrides the old one.
-func (l *logger) setServiceMethodLogger(service string, ml *methodLoggerConfig) error {
- if _, ok := l.services[service]; ok {
+// New MethodLogger with same service overrides the old one.
+func (l *logger) setServiceMethodLogger(service string, ml *MethodLoggerConfig) error {
+ if _, ok := l.config.Services[service]; ok {
return fmt.Errorf("conflicting service rules for service %v found", service)
}
- if l.services == nil {
- l.services = make(map[string]*methodLoggerConfig)
+ if l.config.Services == nil {
+ l.config.Services = make(map[string]*MethodLoggerConfig)
}
- l.services[service] = ml
+ l.config.Services[service] = ml
return nil
}
// Set method logger for "service/method".
//
-// New methodLogger with same method overrides the old one.
-func (l *logger) setMethodMethodLogger(method string, ml *methodLoggerConfig) error {
- if _, ok := l.blacklist[method]; ok {
+// New MethodLogger with same method overrides the old one.
+func (l *logger) setMethodMethodLogger(method string, ml *MethodLoggerConfig) error {
+ if _, ok := l.config.Blacklist[method]; ok {
return fmt.Errorf("conflicting blacklist rules for method %v found", method)
}
- if _, ok := l.methods[method]; ok {
+ if _, ok := l.config.Methods[method]; ok {
return fmt.Errorf("conflicting method rules for method %v found", method)
}
- if l.methods == nil {
- l.methods = make(map[string]*methodLoggerConfig)
+ if l.config.Methods == nil {
+ l.config.Methods = make(map[string]*MethodLoggerConfig)
}
- l.methods[method] = ml
+ l.config.Methods[method] = ml
return nil
}
// Set blacklist method for "-service/method".
func (l *logger) setBlacklist(method string) error {
- if _, ok := l.blacklist[method]; ok {
+ if _, ok := l.config.Blacklist[method]; ok {
return fmt.Errorf("conflicting blacklist rules for method %v found", method)
}
- if _, ok := l.methods[method]; ok {
+ if _, ok := l.config.Methods[method]; ok {
return fmt.Errorf("conflicting method rules for method %v found", method)
}
- if l.blacklist == nil {
- l.blacklist = make(map[string]struct{})
+ if l.config.Blacklist == nil {
+ l.config.Blacklist = make(map[string]struct{})
}
- l.blacklist[method] = struct{}{}
+ l.config.Blacklist[method] = struct{}{}
return nil
}
-// getMethodLogger returns the methodLogger for the given methodName.
+// getMethodLogger returns the MethodLogger for the given methodName.
//
// methodName should be in the format of "/service/method".
//
-// Each methodLogger returned by this method is a new instance. This is to
+// Each MethodLogger returned by this method is a new instance. This is to
// generate sequence id within the call.
-func (l *logger) getMethodLogger(methodName string) *MethodLogger {
+func (l *logger) GetMethodLogger(methodName string) MethodLogger {
s, m, err := grpcutil.ParseMethod(methodName)
if err != nil {
grpclogLogger.Infof("binarylogging: failed to parse %q: %v", methodName, err)
return nil
}
- if ml, ok := l.methods[s+"/"+m]; ok {
- return newMethodLogger(ml.hdr, ml.msg)
+ if ml, ok := l.config.Methods[s+"/"+m]; ok {
+ return NewTruncatingMethodLogger(ml.Header, ml.Message)
}
- if _, ok := l.blacklist[s+"/"+m]; ok {
+ if _, ok := l.config.Blacklist[s+"/"+m]; ok {
return nil
}
- if ml, ok := l.services[s]; ok {
- return newMethodLogger(ml.hdr, ml.msg)
+ if ml, ok := l.config.Services[s]; ok {
+ return NewTruncatingMethodLogger(ml.Header, ml.Message)
}
- if l.all == nil {
+ if l.config.All == nil {
return nil
}
- return newMethodLogger(l.all.hdr, l.all.msg)
+ return NewTruncatingMethodLogger(l.config.All.Header, l.config.All.Message)
}
diff --git a/vendor/google.golang.org/grpc/internal/binarylog/env_config.go b/vendor/google.golang.org/grpc/internal/binarylog/env_config.go
index 129aa2d..f9e80e2 100644
--- a/vendor/google.golang.org/grpc/internal/binarylog/env_config.go
+++ b/vendor/google.golang.org/grpc/internal/binarylog/env_config.go
@@ -57,7 +57,7 @@
return l
}
-// fillMethodLoggerWithConfigString parses config, creates methodLogger and adds
+// fillMethodLoggerWithConfigString parses config, creates TruncatingMethodLogger and adds
// it to the right map in the logger.
func (l *logger) fillMethodLoggerWithConfigString(config string) error {
// "" is invalid.
@@ -89,7 +89,7 @@
if err != nil {
return fmt.Errorf("invalid config: %q, %v", config, err)
}
- if err := l.setDefaultMethodLogger(&methodLoggerConfig{hdr: hdr, msg: msg}); err != nil {
+ if err := l.setDefaultMethodLogger(&MethodLoggerConfig{Header: hdr, Message: msg}); err != nil {
return fmt.Errorf("invalid config: %v", err)
}
return nil
@@ -104,11 +104,11 @@
return fmt.Errorf("invalid header/message length config: %q, %v", suffix, err)
}
if m == "*" {
- if err := l.setServiceMethodLogger(s, &methodLoggerConfig{hdr: hdr, msg: msg}); err != nil {
+ if err := l.setServiceMethodLogger(s, &MethodLoggerConfig{Header: hdr, Message: msg}); err != nil {
return fmt.Errorf("invalid config: %v", err)
}
} else {
- if err := l.setMethodMethodLogger(s+"/"+m, &methodLoggerConfig{hdr: hdr, msg: msg}); err != nil {
+ if err := l.setMethodMethodLogger(s+"/"+m, &MethodLoggerConfig{Header: hdr, Message: msg}); err != nil {
return fmt.Errorf("invalid config: %v", err)
}
}
diff --git a/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go b/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go
index 0cdb418..6c3f632 100644
--- a/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go
+++ b/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go
@@ -19,6 +19,7 @@
package binarylog
import (
+ "context"
"net"
"strings"
"sync/atomic"
@@ -26,7 +27,7 @@
"github.com/golang/protobuf/proto"
"github.com/golang/protobuf/ptypes"
- pb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
+ binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
@@ -48,7 +49,16 @@
var idGen callIDGenerator
// MethodLogger is the sub-logger for each method.
-type MethodLogger struct {
+//
+// This is used in the 1.0 release of gcp/observability, and thus must not be
+// deleted or changed.
+type MethodLogger interface {
+ Log(context.Context, LogEntryConfig)
+}
+
+// TruncatingMethodLogger is a method logger that truncates headers and messages
+// based on configured fields.
+type TruncatingMethodLogger struct {
headerMaxLen, messageMaxLen uint64
callID uint64
@@ -57,8 +67,12 @@
sink Sink // TODO(blog): make this plugable.
}
-func newMethodLogger(h, m uint64) *MethodLogger {
- return &MethodLogger{
+// NewTruncatingMethodLogger returns a new truncating method logger.
+//
+// This is used in the 1.0 release of gcp/observability, and thus must not be
+// deleted or changed.
+func NewTruncatingMethodLogger(h, m uint64) *TruncatingMethodLogger {
+ return &TruncatingMethodLogger{
headerMaxLen: h,
messageMaxLen: m,
@@ -69,8 +83,10 @@
}
}
-// Log creates a proto binary log entry, and logs it to the sink.
-func (ml *MethodLogger) Log(c LogEntryConfig) {
+// Build is an internal only method for building the proto message out of the
+// input event. It's made public to enable other library to reuse as much logic
+// in TruncatingMethodLogger as possible.
+func (ml *TruncatingMethodLogger) Build(c LogEntryConfig) *binlogpb.GrpcLogEntry {
m := c.toProto()
timestamp, _ := ptypes.TimestampProto(time.Now())
m.Timestamp = timestamp
@@ -78,18 +94,22 @@
m.SequenceIdWithinCall = ml.idWithinCallGen.next()
switch pay := m.Payload.(type) {
- case *pb.GrpcLogEntry_ClientHeader:
+ case *binlogpb.GrpcLogEntry_ClientHeader:
m.PayloadTruncated = ml.truncateMetadata(pay.ClientHeader.GetMetadata())
- case *pb.GrpcLogEntry_ServerHeader:
+ case *binlogpb.GrpcLogEntry_ServerHeader:
m.PayloadTruncated = ml.truncateMetadata(pay.ServerHeader.GetMetadata())
- case *pb.GrpcLogEntry_Message:
+ case *binlogpb.GrpcLogEntry_Message:
m.PayloadTruncated = ml.truncateMessage(pay.Message)
}
-
- ml.sink.Write(m)
+ return m
}
-func (ml *MethodLogger) truncateMetadata(mdPb *pb.Metadata) (truncated bool) {
+// Log creates a proto binary log entry, and logs it to the sink.
+func (ml *TruncatingMethodLogger) Log(ctx context.Context, c LogEntryConfig) {
+ ml.sink.Write(ml.Build(c))
+}
+
+func (ml *TruncatingMethodLogger) truncateMetadata(mdPb *binlogpb.Metadata) (truncated bool) {
if ml.headerMaxLen == maxUInt {
return false
}
@@ -108,7 +128,7 @@
// but not counted towards the size limit.
continue
}
- currentEntryLen := uint64(len(entry.Value))
+ currentEntryLen := uint64(len(entry.GetKey())) + uint64(len(entry.GetValue()))
if currentEntryLen > bytesLimit {
break
}
@@ -119,7 +139,7 @@
return truncated
}
-func (ml *MethodLogger) truncateMessage(msgPb *pb.Message) (truncated bool) {
+func (ml *TruncatingMethodLogger) truncateMessage(msgPb *binlogpb.Message) (truncated bool) {
if ml.messageMaxLen == maxUInt {
return false
}
@@ -131,8 +151,11 @@
}
// LogEntryConfig represents the configuration for binary log entry.
+//
+// This is used in the 1.0 release of gcp/observability, and thus must not be
+// deleted or changed.
type LogEntryConfig interface {
- toProto() *pb.GrpcLogEntry
+ toProto() *binlogpb.GrpcLogEntry
}
// ClientHeader configs the binary log entry to be a ClientHeader entry.
@@ -146,10 +169,10 @@
PeerAddr net.Addr
}
-func (c *ClientHeader) toProto() *pb.GrpcLogEntry {
+func (c *ClientHeader) toProto() *binlogpb.GrpcLogEntry {
// This function doesn't need to set all the fields (e.g. seq ID). The Log
// function will set the fields when necessary.
- clientHeader := &pb.ClientHeader{
+ clientHeader := &binlogpb.ClientHeader{
Metadata: mdToMetadataProto(c.Header),
MethodName: c.MethodName,
Authority: c.Authority,
@@ -157,16 +180,16 @@
if c.Timeout > 0 {
clientHeader.Timeout = ptypes.DurationProto(c.Timeout)
}
- ret := &pb.GrpcLogEntry{
- Type: pb.GrpcLogEntry_EVENT_TYPE_CLIENT_HEADER,
- Payload: &pb.GrpcLogEntry_ClientHeader{
+ ret := &binlogpb.GrpcLogEntry{
+ Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_HEADER,
+ Payload: &binlogpb.GrpcLogEntry_ClientHeader{
ClientHeader: clientHeader,
},
}
if c.OnClientSide {
- ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
} else {
- ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
}
if c.PeerAddr != nil {
ret.Peer = addrToProto(c.PeerAddr)
@@ -182,19 +205,19 @@
PeerAddr net.Addr
}
-func (c *ServerHeader) toProto() *pb.GrpcLogEntry {
- ret := &pb.GrpcLogEntry{
- Type: pb.GrpcLogEntry_EVENT_TYPE_SERVER_HEADER,
- Payload: &pb.GrpcLogEntry_ServerHeader{
- ServerHeader: &pb.ServerHeader{
+func (c *ServerHeader) toProto() *binlogpb.GrpcLogEntry {
+ ret := &binlogpb.GrpcLogEntry{
+ Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_HEADER,
+ Payload: &binlogpb.GrpcLogEntry_ServerHeader{
+ ServerHeader: &binlogpb.ServerHeader{
Metadata: mdToMetadataProto(c.Header),
},
},
}
if c.OnClientSide {
- ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
} else {
- ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
}
if c.PeerAddr != nil {
ret.Peer = addrToProto(c.PeerAddr)
@@ -210,7 +233,7 @@
Message interface{}
}
-func (c *ClientMessage) toProto() *pb.GrpcLogEntry {
+func (c *ClientMessage) toProto() *binlogpb.GrpcLogEntry {
var (
data []byte
err error
@@ -225,19 +248,19 @@
} else {
grpclogLogger.Infof("binarylogging: message to log is neither proto.message nor []byte")
}
- ret := &pb.GrpcLogEntry{
- Type: pb.GrpcLogEntry_EVENT_TYPE_CLIENT_MESSAGE,
- Payload: &pb.GrpcLogEntry_Message{
- Message: &pb.Message{
+ ret := &binlogpb.GrpcLogEntry{
+ Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_MESSAGE,
+ Payload: &binlogpb.GrpcLogEntry_Message{
+ Message: &binlogpb.Message{
Length: uint32(len(data)),
Data: data,
},
},
}
if c.OnClientSide {
- ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
} else {
- ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
}
return ret
}
@@ -250,7 +273,7 @@
Message interface{}
}
-func (c *ServerMessage) toProto() *pb.GrpcLogEntry {
+func (c *ServerMessage) toProto() *binlogpb.GrpcLogEntry {
var (
data []byte
err error
@@ -265,19 +288,19 @@
} else {
grpclogLogger.Infof("binarylogging: message to log is neither proto.message nor []byte")
}
- ret := &pb.GrpcLogEntry{
- Type: pb.GrpcLogEntry_EVENT_TYPE_SERVER_MESSAGE,
- Payload: &pb.GrpcLogEntry_Message{
- Message: &pb.Message{
+ ret := &binlogpb.GrpcLogEntry{
+ Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_MESSAGE,
+ Payload: &binlogpb.GrpcLogEntry_Message{
+ Message: &binlogpb.Message{
Length: uint32(len(data)),
Data: data,
},
},
}
if c.OnClientSide {
- ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
} else {
- ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
}
return ret
}
@@ -287,15 +310,15 @@
OnClientSide bool
}
-func (c *ClientHalfClose) toProto() *pb.GrpcLogEntry {
- ret := &pb.GrpcLogEntry{
- Type: pb.GrpcLogEntry_EVENT_TYPE_CLIENT_HALF_CLOSE,
+func (c *ClientHalfClose) toProto() *binlogpb.GrpcLogEntry {
+ ret := &binlogpb.GrpcLogEntry{
+ Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_HALF_CLOSE,
Payload: nil, // No payload here.
}
if c.OnClientSide {
- ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
} else {
- ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
}
return ret
}
@@ -311,7 +334,7 @@
PeerAddr net.Addr
}
-func (c *ServerTrailer) toProto() *pb.GrpcLogEntry {
+func (c *ServerTrailer) toProto() *binlogpb.GrpcLogEntry {
st, ok := status.FromError(c.Err)
if !ok {
grpclogLogger.Info("binarylogging: error in trailer is not a status error")
@@ -327,10 +350,10 @@
grpclogLogger.Infof("binarylogging: failed to marshal status proto: %v", err)
}
}
- ret := &pb.GrpcLogEntry{
- Type: pb.GrpcLogEntry_EVENT_TYPE_SERVER_TRAILER,
- Payload: &pb.GrpcLogEntry_Trailer{
- Trailer: &pb.Trailer{
+ ret := &binlogpb.GrpcLogEntry{
+ Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_TRAILER,
+ Payload: &binlogpb.GrpcLogEntry_Trailer{
+ Trailer: &binlogpb.Trailer{
Metadata: mdToMetadataProto(c.Trailer),
StatusCode: uint32(st.Code()),
StatusMessage: st.Message(),
@@ -339,9 +362,9 @@
},
}
if c.OnClientSide {
- ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
} else {
- ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
}
if c.PeerAddr != nil {
ret.Peer = addrToProto(c.PeerAddr)
@@ -354,15 +377,15 @@
OnClientSide bool
}
-func (c *Cancel) toProto() *pb.GrpcLogEntry {
- ret := &pb.GrpcLogEntry{
- Type: pb.GrpcLogEntry_EVENT_TYPE_CANCEL,
+func (c *Cancel) toProto() *binlogpb.GrpcLogEntry {
+ ret := &binlogpb.GrpcLogEntry{
+ Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CANCEL,
Payload: nil,
}
if c.OnClientSide {
- ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
} else {
- ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+ ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
}
return ret
}
@@ -379,15 +402,15 @@
return strings.HasPrefix(key, "grpc-")
}
-func mdToMetadataProto(md metadata.MD) *pb.Metadata {
- ret := &pb.Metadata{}
+func mdToMetadataProto(md metadata.MD) *binlogpb.Metadata {
+ ret := &binlogpb.Metadata{}
for k, vv := range md {
if metadataKeyOmit(k) {
continue
}
for _, v := range vv {
ret.Entry = append(ret.Entry,
- &pb.MetadataEntry{
+ &binlogpb.MetadataEntry{
Key: k,
Value: []byte(v),
},
@@ -397,26 +420,26 @@
return ret
}
-func addrToProto(addr net.Addr) *pb.Address {
- ret := &pb.Address{}
+func addrToProto(addr net.Addr) *binlogpb.Address {
+ ret := &binlogpb.Address{}
switch a := addr.(type) {
case *net.TCPAddr:
if a.IP.To4() != nil {
- ret.Type = pb.Address_TYPE_IPV4
+ ret.Type = binlogpb.Address_TYPE_IPV4
} else if a.IP.To16() != nil {
- ret.Type = pb.Address_TYPE_IPV6
+ ret.Type = binlogpb.Address_TYPE_IPV6
} else {
- ret.Type = pb.Address_TYPE_UNKNOWN
+ ret.Type = binlogpb.Address_TYPE_UNKNOWN
// Do not set address and port fields.
break
}
ret.Address = a.IP.String()
ret.IpPort = uint32(a.Port)
case *net.UnixAddr:
- ret.Type = pb.Address_TYPE_UNIX
+ ret.Type = binlogpb.Address_TYPE_UNIX
ret.Address = a.String()
default:
- ret.Type = pb.Address_TYPE_UNKNOWN
+ ret.Type = binlogpb.Address_TYPE_UNKNOWN
}
return ret
}
diff --git a/vendor/google.golang.org/grpc/internal/binarylog/sink.go b/vendor/google.golang.org/grpc/internal/binarylog/sink.go
index c2fdd58..264de38 100644
--- a/vendor/google.golang.org/grpc/internal/binarylog/sink.go
+++ b/vendor/google.golang.org/grpc/internal/binarylog/sink.go
@@ -26,7 +26,7 @@
"time"
"github.com/golang/protobuf/proto"
- pb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
+ binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
)
var (
@@ -42,15 +42,15 @@
// Write will be called to write the log entry into the sink.
//
// It should be thread-safe so it can be called in parallel.
- Write(*pb.GrpcLogEntry) error
+ Write(*binlogpb.GrpcLogEntry) error
// Close will be called when the Sink is replaced by a new Sink.
Close() error
}
type noopSink struct{}
-func (ns *noopSink) Write(*pb.GrpcLogEntry) error { return nil }
-func (ns *noopSink) Close() error { return nil }
+func (ns *noopSink) Write(*binlogpb.GrpcLogEntry) error { return nil }
+func (ns *noopSink) Close() error { return nil }
// newWriterSink creates a binary log sink with the given writer.
//
@@ -66,7 +66,7 @@
out io.Writer
}
-func (ws *writerSink) Write(e *pb.GrpcLogEntry) error {
+func (ws *writerSink) Write(e *binlogpb.GrpcLogEntry) error {
b, err := proto.Marshal(e)
if err != nil {
grpclogLogger.Errorf("binary logging: failed to marshal proto message: %v", err)
@@ -96,7 +96,7 @@
done chan struct{}
}
-func (fs *bufferedSink) Write(e *pb.GrpcLogEntry) error {
+func (fs *bufferedSink) Write(e *binlogpb.GrpcLogEntry) error {
fs.mu.Lock()
defer fs.mu.Unlock()
if !fs.flusherStarted {
diff --git a/vendor/google.golang.org/grpc/internal/buffer/unbounded.go b/vendor/google.golang.org/grpc/internal/buffer/unbounded.go
index 9f6a0c1..81c2f5f 100644
--- a/vendor/google.golang.org/grpc/internal/buffer/unbounded.go
+++ b/vendor/google.golang.org/grpc/internal/buffer/unbounded.go
@@ -35,6 +35,7 @@
// internal/transport/transport.go for an example of this.
type Unbounded struct {
c chan interface{}
+ closed bool
mu sync.Mutex
backlog []interface{}
}
@@ -47,16 +48,18 @@
// Put adds t to the unbounded buffer.
func (b *Unbounded) Put(t interface{}) {
b.mu.Lock()
+ defer b.mu.Unlock()
+ if b.closed {
+ return
+ }
if len(b.backlog) == 0 {
select {
case b.c <- t:
- b.mu.Unlock()
return
default:
}
}
b.backlog = append(b.backlog, t)
- b.mu.Unlock()
}
// Load sends the earliest buffered data, if any, onto the read channel
@@ -64,6 +67,10 @@
// value from the read channel.
func (b *Unbounded) Load() {
b.mu.Lock()
+ defer b.mu.Unlock()
+ if b.closed {
+ return
+ }
if len(b.backlog) > 0 {
select {
case b.c <- b.backlog[0]:
@@ -72,7 +79,6 @@
default:
}
}
- b.mu.Unlock()
}
// Get returns a read channel on which values added to the buffer, via Put(),
@@ -80,6 +86,20 @@
//
// Upon reading a value from this channel, users are expected to call Load() to
// send the next buffered value onto the channel if there is any.
+//
+// If the unbounded buffer is closed, the read channel returned by this method
+// is closed.
func (b *Unbounded) Get() <-chan interface{} {
return b.c
}
+
+// Close closes the unbounded buffer.
+func (b *Unbounded) Close() {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ if b.closed {
+ return
+ }
+ b.closed = true
+ close(b.c)
+}
diff --git a/vendor/google.golang.org/grpc/internal/channelz/funcs.go b/vendor/google.golang.org/grpc/internal/channelz/funcs.go
index cd18075..777cbcd 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/funcs.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/funcs.go
@@ -24,6 +24,8 @@
package channelz
import (
+ "context"
+ "errors"
"fmt"
"sort"
"sync"
@@ -49,7 +51,8 @@
// TurnOn turns on channelz data collection.
func TurnOn() {
if !IsOn() {
- NewChannelzStorage()
+ db.set(newChannelMap())
+ idGen.reset()
atomic.StoreInt32(&curState, 1)
}
}
@@ -94,46 +97,40 @@
return d.DB
}
-// NewChannelzStorage initializes channelz data storage and id generator.
+// NewChannelzStorageForTesting initializes channelz data storage and id
+// generator for testing purposes.
//
-// This function returns a cleanup function to wait for all channelz state to be reset by the
-// grpc goroutines when those entities get closed. By using this cleanup function, we make sure tests
-// don't mess up each other, i.e. lingering goroutine from previous test doing entity removal happen
-// to remove some entity just register by the new test, since the id space is the same.
-//
-// Note: This function is exported for testing purpose only. User should not call
-// it in most cases.
-func NewChannelzStorage() (cleanup func() error) {
- db.set(&channelMap{
- topLevelChannels: make(map[int64]struct{}),
- channels: make(map[int64]*channel),
- listenSockets: make(map[int64]*listenSocket),
- normalSockets: make(map[int64]*normalSocket),
- servers: make(map[int64]*server),
- subChannels: make(map[int64]*subChannel),
- })
+// Returns a cleanup function to be invoked by the test, which waits for up to
+// 10s for all channelz state to be reset by the grpc goroutines when those
+// entities get closed. This cleanup function helps with ensuring that tests
+// don't mess up each other.
+func NewChannelzStorageForTesting() (cleanup func() error) {
+ db.set(newChannelMap())
idGen.reset()
+
return func() error {
- var err error
cm := db.get()
if cm == nil {
return nil
}
- for i := 0; i < 1000; i++ {
- cm.mu.Lock()
- if len(cm.topLevelChannels) == 0 && len(cm.servers) == 0 && len(cm.channels) == 0 && len(cm.subChannels) == 0 && len(cm.listenSockets) == 0 && len(cm.normalSockets) == 0 {
- cm.mu.Unlock()
- // all things stored in the channelz map have been cleared.
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ ticker := time.NewTicker(10 * time.Millisecond)
+ defer ticker.Stop()
+ for {
+ cm.mu.RLock()
+ topLevelChannels, servers, channels, subChannels, listenSockets, normalSockets := len(cm.topLevelChannels), len(cm.servers), len(cm.channels), len(cm.subChannels), len(cm.listenSockets), len(cm.normalSockets)
+ cm.mu.RUnlock()
+
+ if err := ctx.Err(); err != nil {
+ return fmt.Errorf("after 10s the channelz map has not been cleaned up yet, topchannels: %d, servers: %d, channels: %d, subchannels: %d, listen sockets: %d, normal sockets: %d", topLevelChannels, servers, channels, subChannels, listenSockets, normalSockets)
+ }
+ if topLevelChannels == 0 && servers == 0 && channels == 0 && subChannels == 0 && listenSockets == 0 && normalSockets == 0 {
return nil
}
- cm.mu.Unlock()
- time.Sleep(10 * time.Millisecond)
+ <-ticker.C
}
-
- cm.mu.Lock()
- err = fmt.Errorf("after 10s the channelz map has not been cleaned up yet, topchannels: %d, servers: %d, channels: %d, subchannels: %d, listen sockets: %d, normal sockets: %d", len(cm.topLevelChannels), len(cm.servers), len(cm.channels), len(cm.subChannels), len(cm.listenSockets), len(cm.normalSockets))
- cm.mu.Unlock()
- return err
}
}
@@ -188,54 +185,77 @@
return db.get().GetServer(id)
}
-// RegisterChannel registers the given channel c in channelz database with ref
-// as its reference name, and add it to the child list of its parent (identified
-// by pid). pid = 0 means no parent. It returns the unique channelz tracking id
-// assigned to this channel.
-func RegisterChannel(c Channel, pid int64, ref string) int64 {
+// RegisterChannel registers the given channel c in the channelz database with
+// ref as its reference name, and adds it to the child list of its parent
+// (identified by pid). pid == nil means no parent.
+//
+// Returns a unique channelz identifier assigned to this channel.
+//
+// If channelz is not turned ON, the channelz database is not mutated.
+func RegisterChannel(c Channel, pid *Identifier, ref string) *Identifier {
id := idGen.genID()
+ var parent int64
+ isTopChannel := true
+ if pid != nil {
+ isTopChannel = false
+ parent = pid.Int()
+ }
+
+ if !IsOn() {
+ return newIdentifer(RefChannel, id, pid)
+ }
+
cn := &channel{
refName: ref,
c: c,
subChans: make(map[int64]string),
nestedChans: make(map[int64]string),
id: id,
- pid: pid,
+ pid: parent,
trace: &channelTrace{createdTime: time.Now(), events: make([]*TraceEvent, 0, getMaxTraceEntry())},
}
- if pid == 0 {
- db.get().addChannel(id, cn, true, pid)
- } else {
- db.get().addChannel(id, cn, false, pid)
- }
- return id
+ db.get().addChannel(id, cn, isTopChannel, parent)
+ return newIdentifer(RefChannel, id, pid)
}
-// RegisterSubChannel registers the given channel c in channelz database with ref
-// as its reference name, and add it to the child list of its parent (identified
-// by pid). It returns the unique channelz tracking id assigned to this subchannel.
-func RegisterSubChannel(c Channel, pid int64, ref string) int64 {
- if pid == 0 {
- logger.Error("a SubChannel's parent id cannot be 0")
- return 0
+// RegisterSubChannel registers the given subChannel c in the channelz database
+// with ref as its reference name, and adds it to the child list of its parent
+// (identified by pid).
+//
+// Returns a unique channelz identifier assigned to this subChannel.
+//
+// If channelz is not turned ON, the channelz database is not mutated.
+func RegisterSubChannel(c Channel, pid *Identifier, ref string) (*Identifier, error) {
+ if pid == nil {
+ return nil, errors.New("a SubChannel's parent id cannot be nil")
}
id := idGen.genID()
+ if !IsOn() {
+ return newIdentifer(RefSubChannel, id, pid), nil
+ }
+
sc := &subChannel{
refName: ref,
c: c,
sockets: make(map[int64]string),
id: id,
- pid: pid,
+ pid: pid.Int(),
trace: &channelTrace{createdTime: time.Now(), events: make([]*TraceEvent, 0, getMaxTraceEntry())},
}
- db.get().addSubChannel(id, sc, pid)
- return id
+ db.get().addSubChannel(id, sc, pid.Int())
+ return newIdentifer(RefSubChannel, id, pid), nil
}
// RegisterServer registers the given server s in channelz database. It returns
// the unique channelz tracking id assigned to this server.
-func RegisterServer(s Server, ref string) int64 {
+//
+// If channelz is not turned ON, the channelz database is not mutated.
+func RegisterServer(s Server, ref string) *Identifier {
id := idGen.genID()
+ if !IsOn() {
+ return newIdentifer(RefServer, id, nil)
+ }
+
svr := &server{
refName: ref,
s: s,
@@ -244,71 +264,92 @@
id: id,
}
db.get().addServer(id, svr)
- return id
+ return newIdentifer(RefServer, id, nil)
}
// RegisterListenSocket registers the given listen socket s in channelz database
// with ref as its reference name, and add it to the child list of its parent
// (identified by pid). It returns the unique channelz tracking id assigned to
// this listen socket.
-func RegisterListenSocket(s Socket, pid int64, ref string) int64 {
- if pid == 0 {
- logger.Error("a ListenSocket's parent id cannot be 0")
- return 0
+//
+// If channelz is not turned ON, the channelz database is not mutated.
+func RegisterListenSocket(s Socket, pid *Identifier, ref string) (*Identifier, error) {
+ if pid == nil {
+ return nil, errors.New("a ListenSocket's parent id cannot be 0")
}
id := idGen.genID()
- ls := &listenSocket{refName: ref, s: s, id: id, pid: pid}
- db.get().addListenSocket(id, ls, pid)
- return id
+ if !IsOn() {
+ return newIdentifer(RefListenSocket, id, pid), nil
+ }
+
+ ls := &listenSocket{refName: ref, s: s, id: id, pid: pid.Int()}
+ db.get().addListenSocket(id, ls, pid.Int())
+ return newIdentifer(RefListenSocket, id, pid), nil
}
// RegisterNormalSocket registers the given normal socket s in channelz database
-// with ref as its reference name, and add it to the child list of its parent
+// with ref as its reference name, and adds it to the child list of its parent
// (identified by pid). It returns the unique channelz tracking id assigned to
// this normal socket.
-func RegisterNormalSocket(s Socket, pid int64, ref string) int64 {
- if pid == 0 {
- logger.Error("a NormalSocket's parent id cannot be 0")
- return 0
+//
+// If channelz is not turned ON, the channelz database is not mutated.
+func RegisterNormalSocket(s Socket, pid *Identifier, ref string) (*Identifier, error) {
+ if pid == nil {
+ return nil, errors.New("a NormalSocket's parent id cannot be 0")
}
id := idGen.genID()
- ns := &normalSocket{refName: ref, s: s, id: id, pid: pid}
- db.get().addNormalSocket(id, ns, pid)
- return id
+ if !IsOn() {
+ return newIdentifer(RefNormalSocket, id, pid), nil
+ }
+
+ ns := &normalSocket{refName: ref, s: s, id: id, pid: pid.Int()}
+ db.get().addNormalSocket(id, ns, pid.Int())
+ return newIdentifer(RefNormalSocket, id, pid), nil
}
// RemoveEntry removes an entry with unique channelz tracking id to be id from
// channelz database.
-func RemoveEntry(id int64) {
- db.get().removeEntry(id)
+//
+// If channelz is not turned ON, this function is a no-op.
+func RemoveEntry(id *Identifier) {
+ if !IsOn() {
+ return
+ }
+ db.get().removeEntry(id.Int())
}
-// TraceEventDesc is what the caller of AddTraceEvent should provide to describe the event to be added
-// to the channel trace.
-// The Parent field is optional. It is used for event that will be recorded in the entity's parent
-// trace also.
+// TraceEventDesc is what the caller of AddTraceEvent should provide to describe
+// the event to be added to the channel trace.
+//
+// The Parent field is optional. It is used for an event that will be recorded
+// in the entity's parent trace.
type TraceEventDesc struct {
Desc string
Severity Severity
Parent *TraceEventDesc
}
-// AddTraceEvent adds trace related to the entity with specified id, using the provided TraceEventDesc.
-func AddTraceEvent(l grpclog.DepthLoggerV2, id int64, depth int, desc *TraceEventDesc) {
- for d := desc; d != nil; d = d.Parent {
- switch d.Severity {
- case CtUnknown, CtInfo:
- l.InfoDepth(depth+1, d.Desc)
- case CtWarning:
- l.WarningDepth(depth+1, d.Desc)
- case CtError:
- l.ErrorDepth(depth+1, d.Desc)
- }
+// AddTraceEvent adds trace related to the entity with specified id, using the
+// provided TraceEventDesc.
+//
+// If channelz is not turned ON, this will simply log the event descriptions.
+func AddTraceEvent(l grpclog.DepthLoggerV2, id *Identifier, depth int, desc *TraceEventDesc) {
+ // Log only the trace description associated with the bottom most entity.
+ switch desc.Severity {
+ case CtUnknown, CtInfo:
+ l.InfoDepth(depth+1, withParens(id)+desc.Desc)
+ case CtWarning:
+ l.WarningDepth(depth+1, withParens(id)+desc.Desc)
+ case CtError:
+ l.ErrorDepth(depth+1, withParens(id)+desc.Desc)
}
+
if getMaxTraceEntry() == 0 {
return
}
- db.get().traceEvent(id, desc)
+ if IsOn() {
+ db.get().traceEvent(id.Int(), desc)
+ }
}
// channelMap is the storage data structure for channelz.
@@ -326,6 +367,17 @@
normalSockets map[int64]*normalSocket
}
+func newChannelMap() *channelMap {
+ return &channelMap{
+ topLevelChannels: make(map[int64]struct{}),
+ channels: make(map[int64]*channel),
+ listenSockets: make(map[int64]*listenSocket),
+ normalSockets: make(map[int64]*normalSocket),
+ servers: make(map[int64]*server),
+ subChannels: make(map[int64]*subChannel),
+ }
+}
+
func (c *channelMap) addServer(id int64, s *server) {
c.mu.Lock()
s.cm = c
diff --git a/vendor/google.golang.org/grpc/internal/channelz/id.go b/vendor/google.golang.org/grpc/internal/channelz/id.go
new file mode 100644
index 0000000..c9a27ac
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/channelz/id.go
@@ -0,0 +1,75 @@
+/*
+ *
+ * Copyright 2022 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 channelz
+
+import "fmt"
+
+// Identifier is an opaque identifier which uniquely identifies an entity in the
+// channelz database.
+type Identifier struct {
+ typ RefChannelType
+ id int64
+ str string
+ pid *Identifier
+}
+
+// Type returns the entity type corresponding to id.
+func (id *Identifier) Type() RefChannelType {
+ return id.typ
+}
+
+// Int returns the integer identifier corresponding to id.
+func (id *Identifier) Int() int64 {
+ return id.id
+}
+
+// String returns a string representation of the entity corresponding to id.
+//
+// This includes some information about the parent as well. Examples:
+// Top-level channel: [Channel #channel-number]
+// Nested channel: [Channel #parent-channel-number Channel #channel-number]
+// Sub channel: [Channel #parent-channel SubChannel #subchannel-number]
+func (id *Identifier) String() string {
+ return id.str
+}
+
+// Equal returns true if other is the same as id.
+func (id *Identifier) Equal(other *Identifier) bool {
+ if (id != nil) != (other != nil) {
+ return false
+ }
+ if id == nil && other == nil {
+ return true
+ }
+ return id.typ == other.typ && id.id == other.id && id.pid == other.pid
+}
+
+// NewIdentifierForTesting returns a new opaque identifier to be used only for
+// testing purposes.
+func NewIdentifierForTesting(typ RefChannelType, id int64, pid *Identifier) *Identifier {
+ return newIdentifer(typ, id, pid)
+}
+
+func newIdentifer(typ RefChannelType, id int64, pid *Identifier) *Identifier {
+ str := fmt.Sprintf("%s #%d", typ, id)
+ if pid != nil {
+ str = fmt.Sprintf("%s %s", pid, str)
+ }
+ return &Identifier{typ: typ, id: id, str: str, pid: pid}
+}
diff --git a/vendor/google.golang.org/grpc/internal/channelz/logging.go b/vendor/google.golang.org/grpc/internal/channelz/logging.go
index b0013f9..8e13a3d 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/logging.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/logging.go
@@ -26,77 +26,54 @@
var logger = grpclog.Component("channelz")
+func withParens(id *Identifier) string {
+ return "[" + id.String() + "] "
+}
+
// Info logs and adds a trace event if channelz is on.
-func Info(l grpclog.DepthLoggerV2, id int64, args ...interface{}) {
- if IsOn() {
- AddTraceEvent(l, id, 1, &TraceEventDesc{
- Desc: fmt.Sprint(args...),
- Severity: CtInfo,
- })
- } else {
- l.InfoDepth(1, args...)
- }
+func Info(l grpclog.DepthLoggerV2, id *Identifier, args ...interface{}) {
+ AddTraceEvent(l, id, 1, &TraceEventDesc{
+ Desc: fmt.Sprint(args...),
+ Severity: CtInfo,
+ })
}
// Infof logs and adds a trace event if channelz is on.
-func Infof(l grpclog.DepthLoggerV2, id int64, format string, args ...interface{}) {
- msg := fmt.Sprintf(format, args...)
- if IsOn() {
- AddTraceEvent(l, id, 1, &TraceEventDesc{
- Desc: msg,
- Severity: CtInfo,
- })
- } else {
- l.InfoDepth(1, msg)
- }
+func Infof(l grpclog.DepthLoggerV2, id *Identifier, format string, args ...interface{}) {
+ AddTraceEvent(l, id, 1, &TraceEventDesc{
+ Desc: fmt.Sprintf(format, args...),
+ Severity: CtInfo,
+ })
}
// Warning logs and adds a trace event if channelz is on.
-func Warning(l grpclog.DepthLoggerV2, id int64, args ...interface{}) {
- if IsOn() {
- AddTraceEvent(l, id, 1, &TraceEventDesc{
- Desc: fmt.Sprint(args...),
- Severity: CtWarning,
- })
- } else {
- l.WarningDepth(1, args...)
- }
+func Warning(l grpclog.DepthLoggerV2, id *Identifier, args ...interface{}) {
+ AddTraceEvent(l, id, 1, &TraceEventDesc{
+ Desc: fmt.Sprint(args...),
+ Severity: CtWarning,
+ })
}
// Warningf logs and adds a trace event if channelz is on.
-func Warningf(l grpclog.DepthLoggerV2, id int64, format string, args ...interface{}) {
- msg := fmt.Sprintf(format, args...)
- if IsOn() {
- AddTraceEvent(l, id, 1, &TraceEventDesc{
- Desc: msg,
- Severity: CtWarning,
- })
- } else {
- l.WarningDepth(1, msg)
- }
+func Warningf(l grpclog.DepthLoggerV2, id *Identifier, format string, args ...interface{}) {
+ AddTraceEvent(l, id, 1, &TraceEventDesc{
+ Desc: fmt.Sprintf(format, args...),
+ Severity: CtWarning,
+ })
}
// Error logs and adds a trace event if channelz is on.
-func Error(l grpclog.DepthLoggerV2, id int64, args ...interface{}) {
- if IsOn() {
- AddTraceEvent(l, id, 1, &TraceEventDesc{
- Desc: fmt.Sprint(args...),
- Severity: CtError,
- })
- } else {
- l.ErrorDepth(1, args...)
- }
+func Error(l grpclog.DepthLoggerV2, id *Identifier, args ...interface{}) {
+ AddTraceEvent(l, id, 1, &TraceEventDesc{
+ Desc: fmt.Sprint(args...),
+ Severity: CtError,
+ })
}
// Errorf logs and adds a trace event if channelz is on.
-func Errorf(l grpclog.DepthLoggerV2, id int64, format string, args ...interface{}) {
- msg := fmt.Sprintf(format, args...)
- if IsOn() {
- AddTraceEvent(l, id, 1, &TraceEventDesc{
- Desc: msg,
- Severity: CtError,
- })
- } else {
- l.ErrorDepth(1, msg)
- }
+func Errorf(l grpclog.DepthLoggerV2, id *Identifier, format string, args ...interface{}) {
+ AddTraceEvent(l, id, 1, &TraceEventDesc{
+ Desc: fmt.Sprintf(format, args...),
+ Severity: CtError,
+ })
}
diff --git a/vendor/google.golang.org/grpc/internal/channelz/types.go b/vendor/google.golang.org/grpc/internal/channelz/types.go
index 0b541d2..7b2f350 100644
--- a/vendor/google.golang.org/grpc/internal/channelz/types.go
+++ b/vendor/google.golang.org/grpc/internal/channelz/types.go
@@ -686,12 +686,33 @@
type RefChannelType int
const (
+ // RefUnknown indicates an unknown entity type, the zero value for this type.
+ RefUnknown RefChannelType = iota
// RefChannel indicates the referenced entity is a Channel.
- RefChannel RefChannelType = iota
+ RefChannel
// RefSubChannel indicates the referenced entity is a SubChannel.
RefSubChannel
+ // RefServer indicates the referenced entity is a Server.
+ RefServer
+ // RefListenSocket indicates the referenced entity is a ListenSocket.
+ RefListenSocket
+ // RefNormalSocket indicates the referenced entity is a NormalSocket.
+ RefNormalSocket
)
+var refChannelTypeToString = map[RefChannelType]string{
+ RefUnknown: "Unknown",
+ RefChannel: "Channel",
+ RefSubChannel: "SubChannel",
+ RefServer: "Server",
+ RefListenSocket: "ListenSocket",
+ RefNormalSocket: "NormalSocket",
+}
+
+func (r RefChannelType) String() string {
+ return refChannelTypeToString[r]
+}
+
func (c *channelTrace) dumpData() *ChannelTrace {
c.mu.Lock()
ct := &ChannelTrace{EventNum: c.eventCount, CreationTime: c.createdTime}
diff --git a/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go b/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
index 6f02725..80fd5c7 100644
--- a/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
+++ b/vendor/google.golang.org/grpc/internal/envconfig/envconfig.go
@@ -21,15 +21,46 @@
import (
"os"
+ "strconv"
"strings"
)
-const (
- prefix = "GRPC_GO_"
- txtErrIgnoreStr = prefix + "IGNORE_TXT_ERRORS"
-)
-
var (
// TXTErrIgnore is set if TXT errors should be ignored ("GRPC_GO_IGNORE_TXT_ERRORS" is not "false").
- TXTErrIgnore = !strings.EqualFold(os.Getenv(txtErrIgnoreStr), "false")
+ TXTErrIgnore = boolFromEnv("GRPC_GO_IGNORE_TXT_ERRORS", true)
+ // AdvertiseCompressors is set if registered compressor should be advertised
+ // ("GRPC_GO_ADVERTISE_COMPRESSORS" is not "false").
+ AdvertiseCompressors = boolFromEnv("GRPC_GO_ADVERTISE_COMPRESSORS", true)
+ // RingHashCap indicates the maximum ring size which defaults to 4096
+ // entries but may be overridden by setting the environment variable
+ // "GRPC_RING_HASH_CAP". This does not override the default bounds
+ // checking which NACKs configs specifying ring sizes > 8*1024*1024 (~8M).
+ RingHashCap = uint64FromEnv("GRPC_RING_HASH_CAP", 4096, 1, 8*1024*1024)
+ // PickFirstLBConfig is set if we should support configuration of the
+ // pick_first LB policy, which can be enabled by setting the environment
+ // variable "GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG" to "true".
+ PickFirstLBConfig = boolFromEnv("GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG", false)
)
+
+func boolFromEnv(envVar string, def bool) bool {
+ if def {
+ // The default is true; return true unless the variable is "false".
+ return !strings.EqualFold(os.Getenv(envVar), "false")
+ }
+ // The default is false; return false unless the variable is "true".
+ return strings.EqualFold(os.Getenv(envVar), "true")
+}
+
+func uint64FromEnv(envVar string, def, min, max uint64) uint64 {
+ v, err := strconv.ParseUint(os.Getenv(envVar), 10, 64)
+ if err != nil {
+ return def
+ }
+ if v < min {
+ return min
+ }
+ if v > max {
+ return max
+ }
+ return v
+}
diff --git a/vendor/google.golang.org/grpc/internal/envconfig/observability.go b/vendor/google.golang.org/grpc/internal/envconfig/observability.go
new file mode 100644
index 0000000..dd314cf
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/envconfig/observability.go
@@ -0,0 +1,42 @@
+/*
+ *
+ * Copyright 2022 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 envconfig
+
+import "os"
+
+const (
+ envObservabilityConfig = "GRPC_GCP_OBSERVABILITY_CONFIG"
+ envObservabilityConfigFile = "GRPC_GCP_OBSERVABILITY_CONFIG_FILE"
+)
+
+var (
+ // ObservabilityConfig is the json configuration for the gcp/observability
+ // package specified directly in the envObservabilityConfig env var.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ ObservabilityConfig = os.Getenv(envObservabilityConfig)
+ // ObservabilityConfigFile is the json configuration for the
+ // gcp/observability specified in a file with the location specified in
+ // envObservabilityConfigFile env var.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ ObservabilityConfigFile = os.Getenv(envObservabilityConfigFile)
+)
diff --git a/vendor/google.golang.org/grpc/internal/envconfig/xds.go b/vendor/google.golang.org/grpc/internal/envconfig/xds.go
index 9bad03c..02b4b6a 100644
--- a/vendor/google.golang.org/grpc/internal/envconfig/xds.go
+++ b/vendor/google.golang.org/grpc/internal/envconfig/xds.go
@@ -20,31 +20,21 @@
import (
"os"
- "strings"
)
const (
// XDSBootstrapFileNameEnv 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.
+ // variable XDSBootstrapFileName.
//
// When both bootstrap FileName and FileContent are set, FileName is used.
XDSBootstrapFileNameEnv = "GRPC_XDS_BOOTSTRAP"
- // XDSBootstrapFileContentEnv is the env variable to set bootstrapp file
+ // XDSBootstrapFileContentEnv is the env variable to set bootstrap file
// content. Do not use this and read from env directly. Its value is read
- // and kept in variable BootstrapFileName.
+ // and kept in variable XDSBootstrapFileContent.
//
// When both bootstrap FileName and FileContent are set, FileName is used.
XDSBootstrapFileContentEnv = "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"
- rbacSupportEnv = "GRPC_XDS_EXPERIMENTAL_RBAC"
- federationEnv = "GRPC_EXPERIMENTAL_XDS_FEDERATION"
- rlsInXDSEnv = "GRPC_EXPERIMENTAL_XDS_RLS_LB"
-
- c2pResolverTestOnlyTrafficDirectorURIEnv = "GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI"
)
var (
@@ -63,35 +53,43 @@
// XDSRingHash indicates whether ring hash support is enabled, which can be
// disabled by setting the environment variable
// "GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH" to "false".
- XDSRingHash = !strings.EqualFold(os.Getenv(ringHashSupportEnv), "false")
+ XDSRingHash = boolFromEnv("GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH", true)
// XDSClientSideSecurity 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.
- XDSClientSideSecurity = !strings.EqualFold(os.Getenv(clientSideSecuritySupportEnv), "false")
- // XDSAggregateAndDNS 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".
- XDSAggregateAndDNS = strings.EqualFold(os.Getenv(aggregateAndDNSSupportEnv), "true")
+ XDSClientSideSecurity = boolFromEnv("GRPC_XDS_EXPERIMENTAL_SECURITY_SUPPORT", true)
+ // XDSAggregateAndDNS indicates whether processing of aggregated cluster and
+ // DNS cluster is enabled, which can be disabled by setting the environment
+ // variable "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER"
+ // to "false".
+ XDSAggregateAndDNS = boolFromEnv("GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER", true)
// XDSRBAC indicates whether xDS configured RBAC HTTP Filter is enabled,
// which can be disabled by setting the environment variable
// "GRPC_XDS_EXPERIMENTAL_RBAC" to "false".
- XDSRBAC = !strings.EqualFold(os.Getenv(rbacSupportEnv), "false")
-
- // XDSFederation indicates whether federation support is enabled.
- XDSFederation = strings.EqualFold(os.Getenv(federationEnv), "true")
+ XDSRBAC = boolFromEnv("GRPC_XDS_EXPERIMENTAL_RBAC", true)
+ // XDSOutlierDetection indicates whether outlier detection support is
+ // enabled, which can be disabled by setting the environment variable
+ // "GRPC_EXPERIMENTAL_ENABLE_OUTLIER_DETECTION" to "false".
+ XDSOutlierDetection = boolFromEnv("GRPC_EXPERIMENTAL_ENABLE_OUTLIER_DETECTION", true)
+ // XDSFederation indicates whether federation support is enabled, which can
+ // be enabled by setting the environment variable
+ // "GRPC_EXPERIMENTAL_XDS_FEDERATION" to "true".
+ XDSFederation = boolFromEnv("GRPC_EXPERIMENTAL_XDS_FEDERATION", true)
// XDSRLS indicates whether processing of Cluster Specifier plugins and
- // support for the RLS CLuster Specifier is enabled, which can be enabled by
+ // support for the RLS CLuster Specifier is enabled, which can be disabled by
// setting the environment variable "GRPC_EXPERIMENTAL_XDS_RLS_LB" to
- // "true".
- XDSRLS = strings.EqualFold(os.Getenv(rlsInXDSEnv), "true")
+ // "false".
+ XDSRLS = boolFromEnv("GRPC_EXPERIMENTAL_XDS_RLS_LB", true)
// C2PResolverTestOnlyTrafficDirectorURI is the TD URI for testing.
- C2PResolverTestOnlyTrafficDirectorURI = os.Getenv(c2pResolverTestOnlyTrafficDirectorURIEnv)
+ C2PResolverTestOnlyTrafficDirectorURI = os.Getenv("GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI")
+ // XDSCustomLBPolicy indicates whether Custom LB Policies are enabled, which
+ // can be disabled by setting the environment variable
+ // "GRPC_EXPERIMENTAL_XDS_CUSTOM_LB_CONFIG" to "false".
+ XDSCustomLBPolicy = boolFromEnv("GRPC_EXPERIMENTAL_XDS_CUSTOM_LB_CONFIG", true)
)
diff --git a/vendor/google.golang.org/grpc/internal/grpclog/prefixLogger.go b/vendor/google.golang.org/grpc/internal/grpclog/prefixLogger.go
index 82af70e..02224b4 100644
--- a/vendor/google.golang.org/grpc/internal/grpclog/prefixLogger.go
+++ b/vendor/google.golang.org/grpc/internal/grpclog/prefixLogger.go
@@ -63,6 +63,9 @@
// Debugf does info logging at verbose level 2.
func (pl *PrefixLogger) Debugf(format string, args ...interface{}) {
+ // TODO(6044): Refactor interfaces LoggerV2 and DepthLogger, and maybe
+ // rewrite PrefixLogger a little to ensure that we don't use the global
+ // `Logger` here, and instead use the `logger` field.
if !Logger.V(2) {
return
}
@@ -73,6 +76,15 @@
return
}
InfoDepth(1, fmt.Sprintf(format, args...))
+
+}
+
+// V reports whether verbosity level l is at least the requested verbose level.
+func (pl *PrefixLogger) V(l int) bool {
+ // TODO(6044): Refactor interfaces LoggerV2 and DepthLogger, and maybe
+ // rewrite PrefixLogger a little to ensure that we don't use the global
+ // `Logger` here, and instead use the `logger` field.
+ return Logger.V(l)
}
// NewPrefixLogger creates a prefix logger with the given prefix.
diff --git a/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
index 740f83c..d08e3e9 100644
--- a/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
+++ b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
@@ -52,6 +52,13 @@
return r.Intn(n)
}
+// Int31n implements rand.Int31n on the grpcrand global source.
+func Int31n(n int32) int32 {
+ mu.Lock()
+ defer mu.Unlock()
+ return r.Int31n(n)
+}
+
// Float64 implements rand.Float64 on the grpcrand global source.
func Float64() float64 {
mu.Lock()
@@ -65,3 +72,17 @@
defer mu.Unlock()
return r.Uint64()
}
+
+// Uint32 implements rand.Uint32 on the grpcrand global source.
+func Uint32() uint32 {
+ mu.Lock()
+ defer mu.Unlock()
+ return r.Uint32()
+}
+
+// Shuffle implements rand.Shuffle on the grpcrand global source.
+var Shuffle = func(n int, f func(int, int)) {
+ mu.Lock()
+ defer mu.Unlock()
+ r.Shuffle(n, f)
+}
diff --git a/vendor/google.golang.org/grpc/internal/grpcsync/callback_serializer.go b/vendor/google.golang.org/grpc/internal/grpcsync/callback_serializer.go
new file mode 100644
index 0000000..37b8d41
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/grpcsync/callback_serializer.go
@@ -0,0 +1,119 @@
+/*
+ *
+ * Copyright 2022 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 grpcsync
+
+import (
+ "context"
+ "sync"
+
+ "google.golang.org/grpc/internal/buffer"
+)
+
+// CallbackSerializer provides a mechanism to schedule callbacks in a
+// synchronized manner. It provides a FIFO guarantee on the order of execution
+// of scheduled callbacks. New callbacks can be scheduled by invoking the
+// Schedule() method.
+//
+// This type is safe for concurrent access.
+type CallbackSerializer struct {
+ // Done is closed once the serializer is shut down completely, i.e all
+ // scheduled callbacks are executed and the serializer has deallocated all
+ // its resources.
+ Done chan struct{}
+
+ callbacks *buffer.Unbounded
+ closedMu sync.Mutex
+ closed bool
+}
+
+// NewCallbackSerializer returns a new CallbackSerializer instance. The provided
+// context will be passed to the scheduled callbacks. Users should cancel the
+// provided context to shutdown the CallbackSerializer. It is guaranteed that no
+// callbacks will be added once this context is canceled, and any pending un-run
+// callbacks will be executed before the serializer is shut down.
+func NewCallbackSerializer(ctx context.Context) *CallbackSerializer {
+ t := &CallbackSerializer{
+ Done: make(chan struct{}),
+ callbacks: buffer.NewUnbounded(),
+ }
+ go t.run(ctx)
+ return t
+}
+
+// Schedule adds a callback to be scheduled after existing callbacks are run.
+//
+// Callbacks are expected to honor the context when performing any blocking
+// operations, and should return early when the context is canceled.
+//
+// Return value indicates if the callback was successfully added to the list of
+// callbacks to be executed by the serializer. It is not possible to add
+// callbacks once the context passed to NewCallbackSerializer is cancelled.
+func (t *CallbackSerializer) Schedule(f func(ctx context.Context)) bool {
+ t.closedMu.Lock()
+ defer t.closedMu.Unlock()
+
+ if t.closed {
+ return false
+ }
+ t.callbacks.Put(f)
+ return true
+}
+
+func (t *CallbackSerializer) run(ctx context.Context) {
+ var backlog []func(context.Context)
+
+ defer close(t.Done)
+ for ctx.Err() == nil {
+ select {
+ case <-ctx.Done():
+ // Do nothing here. Next iteration of the for loop will not happen,
+ // since ctx.Err() would be non-nil.
+ case callback, ok := <-t.callbacks.Get():
+ if !ok {
+ return
+ }
+ t.callbacks.Load()
+ callback.(func(ctx context.Context))(ctx)
+ }
+ }
+
+ // Fetch pending callbacks if any, and execute them before returning from
+ // this method and closing t.Done.
+ t.closedMu.Lock()
+ t.closed = true
+ backlog = t.fetchPendingCallbacks()
+ t.callbacks.Close()
+ t.closedMu.Unlock()
+ for _, b := range backlog {
+ b(ctx)
+ }
+}
+
+func (t *CallbackSerializer) fetchPendingCallbacks() []func(context.Context) {
+ var backlog []func(context.Context)
+ for {
+ select {
+ case b := <-t.callbacks.Get():
+ backlog = append(backlog, b.(func(context.Context)))
+ t.callbacks.Load()
+ default:
+ return backlog
+ }
+ }
+}
diff --git a/vendor/google.golang.org/grpc/internal/grpcsync/oncefunc.go b/vendor/google.golang.org/grpc/internal/grpcsync/oncefunc.go
new file mode 100644
index 0000000..6635f7b
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/grpcsync/oncefunc.go
@@ -0,0 +1,32 @@
+/*
+ *
+ * Copyright 2022 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 grpcsync
+
+import (
+ "sync"
+)
+
+// OnceFunc returns a function wrapping f which ensures f is only executed
+// once even if the returned function is executed multiple times.
+func OnceFunc(f func()) func() {
+ var once sync.Once
+ return func() {
+ once.Do(f)
+ }
+}
diff --git a/vendor/google.golang.org/grpc/internal/grpcutil/compressor.go b/vendor/google.golang.org/grpc/internal/grpcutil/compressor.go
new file mode 100644
index 0000000..9f40909
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/grpcutil/compressor.go
@@ -0,0 +1,47 @@
+/*
+ *
+ * Copyright 2022 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 grpcutil
+
+import (
+ "strings"
+
+ "google.golang.org/grpc/internal/envconfig"
+)
+
+// RegisteredCompressorNames holds names of the registered compressors.
+var RegisteredCompressorNames []string
+
+// IsCompressorNameRegistered returns true when name is available in registry.
+func IsCompressorNameRegistered(name string) bool {
+ for _, compressor := range RegisteredCompressorNames {
+ if compressor == name {
+ return true
+ }
+ }
+ return false
+}
+
+// RegisteredCompressors returns a string of registered compressor names
+// separated by comma.
+func RegisteredCompressors() string {
+ if !envconfig.AdvertiseCompressors {
+ return ""
+ }
+ return strings.Join(RegisteredCompressorNames, ",")
+}
diff --git a/vendor/google.golang.org/grpc/internal/grpcutil/method.go b/vendor/google.golang.org/grpc/internal/grpcutil/method.go
index c6f2012..ec62b47 100644
--- a/vendor/google.golang.org/grpc/internal/grpcutil/method.go
+++ b/vendor/google.golang.org/grpc/internal/grpcutil/method.go
@@ -38,6 +38,11 @@
return methodName[:pos], methodName[pos+1:], nil
}
+// baseContentType is the base content-type for gRPC. This is a valid
+// content-type on it's own, but can also include a content-subtype such as
+// "proto" as a suffix after "+" or ";". See
+// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
+// for more details.
const baseContentType = "application/grpc"
// ContentSubtype returns the content-subtype for the given content-type. The
diff --git a/vendor/google.golang.org/grpc/internal/internal.go b/vendor/google.golang.org/grpc/internal/internal.go
index 1b596bf..42ff39c 100644
--- a/vendor/google.golang.org/grpc/internal/internal.go
+++ b/vendor/google.golang.org/grpc/internal/internal.go
@@ -38,11 +38,10 @@
// KeepaliveMinPingTime is the minimum ping interval. This must be 10s by
// default, but tests may wish to set it lower for convenience.
KeepaliveMinPingTime = 10 * time.Second
- // ParseServiceConfigForTesting is for creating a fake
- // ClientConn for resolver testing only
- ParseServiceConfigForTesting interface{} // func(string) *serviceconfig.ParseResult
+ // ParseServiceConfig parses a JSON representation of the service config.
+ ParseServiceConfig interface{} // func(string) *serviceconfig.ParseResult
// EqualServiceConfigForTesting is for testing service config generation and
- // parsing. Both a and b should be returned by ParseServiceConfigForTesting.
+ // parsing. Both a and b should be returned by ParseServiceConfig.
// This function compares the config without rawJSON stripped, in case the
// there's difference in white space.
EqualServiceConfigForTesting func(a, b serviceconfig.Config) bool
@@ -59,11 +58,112 @@
// gRPC server. An xDS-enabled server needs to know what type of credentials
// is configured on the underlying gRPC server. This is set by server.go.
GetServerCredentials interface{} // func (*grpc.Server) credentials.TransportCredentials
+ // CanonicalString returns the canonical string of the code defined here:
+ // https://github.com/grpc/grpc/blob/master/doc/statuscodes.md.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ CanonicalString interface{} // func (codes.Code) string
// DrainServerTransports initiates a graceful close of existing connections
// on a gRPC server accepted on the provided listener address. An
// xDS-enabled server invokes this method on a grpc.Server when a particular
// listener moves to "not-serving" mode.
DrainServerTransports interface{} // func(*grpc.Server, string)
+ // AddGlobalServerOptions adds an array of ServerOption that will be
+ // effective globally for newly created servers. The priority will be: 1.
+ // user-provided; 2. this method; 3. default values.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ AddGlobalServerOptions interface{} // func(opt ...ServerOption)
+ // ClearGlobalServerOptions clears the array of extra ServerOption. This
+ // method is useful in testing and benchmarking.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ ClearGlobalServerOptions func()
+ // AddGlobalDialOptions adds an array of DialOption that will be effective
+ // globally for newly created client channels. The priority will be: 1.
+ // user-provided; 2. this method; 3. default values.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ AddGlobalDialOptions interface{} // func(opt ...DialOption)
+ // DisableGlobalDialOptions returns a DialOption that prevents the
+ // ClientConn from applying the global DialOptions (set via
+ // AddGlobalDialOptions).
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ DisableGlobalDialOptions interface{} // func() grpc.DialOption
+ // ClearGlobalDialOptions clears the array of extra DialOption. This
+ // method is useful in testing and benchmarking.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ ClearGlobalDialOptions func()
+ // JoinDialOptions combines the dial options passed as arguments into a
+ // single dial option.
+ JoinDialOptions interface{} // func(...grpc.DialOption) grpc.DialOption
+ // JoinServerOptions combines the server options passed as arguments into a
+ // single server option.
+ JoinServerOptions interface{} // func(...grpc.ServerOption) grpc.ServerOption
+
+ // WithBinaryLogger returns a DialOption that specifies the binary logger
+ // for a ClientConn.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ WithBinaryLogger interface{} // func(binarylog.Logger) grpc.DialOption
+ // BinaryLogger returns a ServerOption that can set the binary logger for a
+ // server.
+ //
+ // This is used in the 1.0 release of gcp/observability, and thus must not be
+ // deleted or changed.
+ BinaryLogger interface{} // func(binarylog.Logger) grpc.ServerOption
+
+ // NewXDSResolverWithConfigForTesting creates a new xds resolver builder using
+ // the provided xds bootstrap config instead of the global configuration from
+ // the supported environment variables. The resolver.Builder is meant to be
+ // used in conjunction with the grpc.WithResolvers DialOption.
+ //
+ // Testing Only
+ //
+ // This function should ONLY be used for testing and may not work with some
+ // other features, including the CSDS service.
+ NewXDSResolverWithConfigForTesting interface{} // func([]byte) (resolver.Builder, error)
+
+ // RegisterRLSClusterSpecifierPluginForTesting registers the RLS Cluster
+ // Specifier Plugin for testing purposes, regardless of the XDSRLS environment
+ // variable.
+ //
+ // TODO: Remove this function once the RLS env var is removed.
+ RegisterRLSClusterSpecifierPluginForTesting func()
+
+ // UnregisterRLSClusterSpecifierPluginForTesting unregisters the RLS Cluster
+ // Specifier Plugin for testing purposes. This is needed because there is no way
+ // to unregister the RLS Cluster Specifier Plugin after registering it solely
+ // for testing purposes using RegisterRLSClusterSpecifierPluginForTesting().
+ //
+ // TODO: Remove this function once the RLS env var is removed.
+ UnregisterRLSClusterSpecifierPluginForTesting func()
+
+ // RegisterRBACHTTPFilterForTesting registers the RBAC HTTP Filter for testing
+ // purposes, regardless of the RBAC environment variable.
+ //
+ // TODO: Remove this function once the RBAC env var is removed.
+ RegisterRBACHTTPFilterForTesting func()
+
+ // UnregisterRBACHTTPFilterForTesting unregisters the RBAC HTTP Filter for
+ // testing purposes. This is needed because there is no way to unregister the
+ // HTTP Filter after registering it solely for testing purposes using
+ // RegisterRBACHTTPFilterForTesting().
+ //
+ // TODO: Remove this function once the RBAC env var is removed.
+ UnregisterRBACHTTPFilterForTesting func()
+
+ // ORCAAllowAnyMinReportingInterval is for examples/orca use ONLY.
+ ORCAAllowAnyMinReportingInterval interface{} // func(so *orca.ServiceOptions)
)
// HealthChecker defines the signature of the client-side LB channel health checking function.
@@ -86,3 +186,9 @@
// that supports backend returned by grpclb balancer.
CredsBundleModeBackendFromBalancer = "backend-from-balancer"
)
+
+// RLSLoadBalancingPolicyName is the name of the RLS LB policy.
+//
+// It currently has an experimental suffix which would be removed once
+// end-to-end testing of the policy is completed.
+const RLSLoadBalancingPolicyName = "rls_experimental"
diff --git a/vendor/google.golang.org/grpc/internal/metadata/metadata.go b/vendor/google.golang.org/grpc/internal/metadata/metadata.go
index b8733db..c82e608 100644
--- a/vendor/google.golang.org/grpc/internal/metadata/metadata.go
+++ b/vendor/google.golang.org/grpc/internal/metadata/metadata.go
@@ -22,6 +22,9 @@
package metadata
import (
+ "fmt"
+ "strings"
+
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/resolver"
)
@@ -72,3 +75,58 @@
addr.Attributes = addr.Attributes.WithValue(mdKey, mdValue(md))
return addr
}
+
+// Validate validates every pair in md with ValidatePair.
+func Validate(md metadata.MD) error {
+ for k, vals := range md {
+ if err := ValidatePair(k, vals...); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// hasNotPrintable return true if msg contains any characters which are not in %x20-%x7E
+func hasNotPrintable(msg string) bool {
+ // for i that saving a conversion if not using for range
+ for i := 0; i < len(msg); i++ {
+ if msg[i] < 0x20 || msg[i] > 0x7E {
+ return true
+ }
+ }
+ return false
+}
+
+// ValidatePair validate a key-value pair with the following rules (the pseudo-header will be skipped) :
+//
+// - key must contain one or more characters.
+// - the characters in the key must be contained in [0-9 a-z _ - .].
+// - if the key ends with a "-bin" suffix, no validation of the corresponding value is performed.
+// - the characters in the every value must be printable (in [%x20-%x7E]).
+func ValidatePair(key string, vals ...string) error {
+ // key should not be empty
+ if key == "" {
+ return fmt.Errorf("there is an empty key in the header")
+ }
+ // pseudo-header will be ignored
+ if key[0] == ':' {
+ return nil
+ }
+ // check key, for i that saving a conversion if not using for range
+ for i := 0; i < len(key); i++ {
+ r := key[i]
+ if !(r >= 'a' && r <= 'z') && !(r >= '0' && r <= '9') && r != '.' && r != '-' && r != '_' {
+ return fmt.Errorf("header key %q contains illegal characters not in [0-9a-z-_.]", key)
+ }
+ }
+ if strings.HasSuffix(key, "-bin") {
+ return nil
+ }
+ // check value
+ for _, val := range vals {
+ if hasNotPrintable(val) {
+ return fmt.Errorf("header key %q contains value with non-printable ASCII characters", key)
+ }
+ }
+ return nil
+}
diff --git a/vendor/google.golang.org/grpc/internal/pretty/pretty.go b/vendor/google.golang.org/grpc/internal/pretty/pretty.go
new file mode 100644
index 0000000..0177af4
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/pretty/pretty.go
@@ -0,0 +1,82 @@
+/*
+ *
+ * Copyright 2021 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 pretty defines helper functions to pretty-print structs for logging.
+package pretty
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+
+ "github.com/golang/protobuf/jsonpb"
+ protov1 "github.com/golang/protobuf/proto"
+ "google.golang.org/protobuf/encoding/protojson"
+ protov2 "google.golang.org/protobuf/proto"
+)
+
+const jsonIndent = " "
+
+// ToJSON marshals the input into a json string.
+//
+// If marshal fails, it falls back to fmt.Sprintf("%+v").
+func ToJSON(e interface{}) string {
+ switch ee := e.(type) {
+ case protov1.Message:
+ mm := jsonpb.Marshaler{Indent: jsonIndent}
+ ret, err := mm.MarshalToString(ee)
+ if err != nil {
+ // This may fail for proto.Anys, e.g. for xDS v2, LDS, the v2
+ // messages are not imported, and this will fail because the message
+ // is not found.
+ return fmt.Sprintf("%+v", ee)
+ }
+ return ret
+ case protov2.Message:
+ mm := protojson.MarshalOptions{
+ Multiline: true,
+ Indent: jsonIndent,
+ }
+ ret, err := mm.Marshal(ee)
+ if err != nil {
+ // This may fail for proto.Anys, e.g. for xDS v2, LDS, the v2
+ // messages are not imported, and this will fail because the message
+ // is not found.
+ return fmt.Sprintf("%+v", ee)
+ }
+ return string(ret)
+ default:
+ ret, err := json.MarshalIndent(ee, "", jsonIndent)
+ if err != nil {
+ return fmt.Sprintf("%+v", ee)
+ }
+ return string(ret)
+ }
+}
+
+// FormatJSON formats the input json bytes with indentation.
+//
+// If Indent fails, it returns the unchanged input as string.
+func FormatJSON(b []byte) string {
+ var out bytes.Buffer
+ err := json.Indent(&out, b, "", jsonIndent)
+ if err != nil {
+ return string(b)
+ }
+ return out.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 75301c5..09a667f 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
@@ -116,7 +116,7 @@
// Build creates and starts a DNS resolver that watches the name resolution of the target.
func (b *dnsBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
- host, port, err := parseTarget(target.Endpoint, defaultPort)
+ host, port, err := parseTarget(target.Endpoint(), defaultPort)
if err != nil {
return nil, err
}
@@ -140,10 +140,10 @@
disableServiceConfig: opts.DisableServiceConfig,
}
- if target.Authority == "" {
+ if target.URL.Host == "" {
d.resolver = defaultResolver
} else {
- d.resolver, err = customAuthorityResolver(target.Authority)
+ d.resolver, err = customAuthorityResolver(target.URL.Host)
if err != nil {
return nil, err
}
diff --git a/vendor/google.golang.org/grpc/internal/resolver/passthrough/passthrough.go b/vendor/google.golang.org/grpc/internal/resolver/passthrough/passthrough.go
index 520d922..afac565 100644
--- a/vendor/google.golang.org/grpc/internal/resolver/passthrough/passthrough.go
+++ b/vendor/google.golang.org/grpc/internal/resolver/passthrough/passthrough.go
@@ -20,13 +20,20 @@
// name without scheme back to gRPC as resolved address.
package passthrough
-import "google.golang.org/grpc/resolver"
+import (
+ "errors"
+
+ "google.golang.org/grpc/resolver"
+)
const scheme = "passthrough"
type passthroughBuilder struct{}
func (*passthroughBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+ if target.Endpoint() == "" && opts.Dialer == nil {
+ return nil, errors.New("passthrough: received empty target in Build()")
+ }
r := &passthroughResolver{
target: target,
cc: cc,
@@ -45,7 +52,7 @@
}
func (r *passthroughResolver) start() {
- r.cc.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: r.target.Endpoint}}})
+ r.cc.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: r.target.Endpoint()}}})
}
func (*passthroughResolver) ResolveNow(o resolver.ResolveNowOptions) {}
diff --git a/vendor/google.golang.org/grpc/internal/resolver/unix/unix.go b/vendor/google.golang.org/grpc/internal/resolver/unix/unix.go
index 20852e5..1609116 100644
--- a/vendor/google.golang.org/grpc/internal/resolver/unix/unix.go
+++ b/vendor/google.golang.org/grpc/internal/resolver/unix/unix.go
@@ -34,8 +34,8 @@
}
func (b *builder) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) (resolver.Resolver, error) {
- if target.Authority != "" {
- return nil, fmt.Errorf("invalid (non-empty) authority: %v", target.Authority)
+ if target.URL.Host != "" {
+ return nil, fmt.Errorf("invalid (non-empty) authority: %v", target.URL.Host)
}
// gRPC was parsing the dial target manually before PR #4817, and we
@@ -49,8 +49,9 @@
}
addr := resolver.Address{Addr: endpoint}
if b.scheme == unixAbstractScheme {
- // prepend "\x00" to address for unix-abstract
- addr.Addr = "\x00" + addr.Addr
+ // We can not prepend \0 as c++ gRPC does, as in Golang '@' is used to signify we do
+ // not want trailing \0 in address.
+ addr.Addr = "@" + addr.Addr
}
cc.UpdateState(resolver.State{Addresses: []resolver.Address{networktype.Set(addr, "unix")}})
return &nopResolver{}, nil
diff --git a/vendor/google.golang.org/grpc/internal/serviceconfig/duration.go b/vendor/google.golang.org/grpc/internal/serviceconfig/duration.go
new file mode 100644
index 0000000..11d82af
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/serviceconfig/duration.go
@@ -0,0 +1,130 @@
+/*
+ *
+ * Copyright 2023 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package serviceconfig
+
+import (
+ "encoding/json"
+ "fmt"
+ "math"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// Duration defines JSON marshal and unmarshal methods to conform to the
+// protobuf JSON spec defined [here].
+//
+// [here]: https://protobuf.dev/reference/protobuf/google.protobuf/#duration
+type Duration time.Duration
+
+func (d Duration) String() string {
+ return fmt.Sprint(time.Duration(d))
+}
+
+// MarshalJSON converts from d to a JSON string output.
+func (d Duration) MarshalJSON() ([]byte, error) {
+ ns := time.Duration(d).Nanoseconds()
+ sec := ns / int64(time.Second)
+ ns = ns % int64(time.Second)
+
+ var sign string
+ if sec < 0 || ns < 0 {
+ sign, sec, ns = "-", -1*sec, -1*ns
+ }
+
+ // Generated output always contains 0, 3, 6, or 9 fractional digits,
+ // depending on required precision.
+ str := fmt.Sprintf("%s%d.%09d", sign, sec, ns)
+ str = strings.TrimSuffix(str, "000")
+ str = strings.TrimSuffix(str, "000")
+ str = strings.TrimSuffix(str, ".000")
+ return []byte(fmt.Sprintf("\"%ss\"", str)), nil
+}
+
+// UnmarshalJSON unmarshals b as a duration JSON string into d.
+func (d *Duration) UnmarshalJSON(b []byte) error {
+ var s string
+ if err := json.Unmarshal(b, &s); err != nil {
+ return err
+ }
+ if !strings.HasSuffix(s, "s") {
+ return fmt.Errorf("malformed duration %q: missing seconds unit", s)
+ }
+ neg := false
+ if s[0] == '-' {
+ neg = true
+ s = s[1:]
+ }
+ ss := strings.SplitN(s[:len(s)-1], ".", 3)
+ if len(ss) > 2 {
+ return fmt.Errorf("malformed duration %q: too many decimals", s)
+ }
+ // hasDigits is set if either the whole or fractional part of the number is
+ // present, since both are optional but one is required.
+ hasDigits := false
+ var sec, ns int64
+ if len(ss[0]) > 0 {
+ var err error
+ if sec, err = strconv.ParseInt(ss[0], 10, 64); err != nil {
+ return fmt.Errorf("malformed duration %q: %v", s, err)
+ }
+ // Maximum seconds value per the durationpb spec.
+ const maxProtoSeconds = 315_576_000_000
+ if sec > maxProtoSeconds {
+ return fmt.Errorf("out of range: %q", s)
+ }
+ hasDigits = true
+ }
+ if len(ss) == 2 && len(ss[1]) > 0 {
+ if len(ss[1]) > 9 {
+ return fmt.Errorf("malformed duration %q: too many digits after decimal", s)
+ }
+ var err error
+ if ns, err = strconv.ParseInt(ss[1], 10, 64); err != nil {
+ return fmt.Errorf("malformed duration %q: %v", s, err)
+ }
+ for i := 9; i > len(ss[1]); i-- {
+ ns *= 10
+ }
+ hasDigits = true
+ }
+ if !hasDigits {
+ return fmt.Errorf("malformed duration %q: contains no numbers", s)
+ }
+
+ if neg {
+ sec *= -1
+ ns *= -1
+ }
+
+ // Maximum/minimum seconds/nanoseconds representable by Go's time.Duration.
+ const maxSeconds = math.MaxInt64 / int64(time.Second)
+ const maxNanosAtMaxSeconds = math.MaxInt64 % int64(time.Second)
+ const minSeconds = math.MinInt64 / int64(time.Second)
+ const minNanosAtMinSeconds = math.MinInt64 % int64(time.Second)
+
+ if sec > maxSeconds || (sec == maxSeconds && ns >= maxNanosAtMaxSeconds) {
+ *d = Duration(math.MaxInt64)
+ } else if sec < minSeconds || (sec == minSeconds && ns <= minNanosAtMinSeconds) {
+ *d = Duration(math.MinInt64)
+ } else {
+ *d = Duration(sec*int64(time.Second) + ns)
+ }
+ return nil
+}
diff --git a/vendor/google.golang.org/grpc/internal/status/status.go b/vendor/google.golang.org/grpc/internal/status/status.go
index e5c6513..b0ead4f 100644
--- a/vendor/google.golang.org/grpc/internal/status/status.go
+++ b/vendor/google.golang.org/grpc/internal/status/status.go
@@ -164,3 +164,13 @@
}
return proto.Equal(e.s.s, tse.s.s)
}
+
+// IsRestrictedControlPlaneCode returns whether the status includes a code
+// restricted for control plane usage as defined by gRFC A54.
+func IsRestrictedControlPlaneCode(s *Status) bool {
+ switch s.Code() {
+ case codes.InvalidArgument, codes.NotFound, codes.AlreadyExists, codes.FailedPrecondition, codes.Aborted, codes.OutOfRange, codes.DataLoss:
+ return true
+ }
+ return false
+}
diff --git a/vendor/google.golang.org/grpc/internal/transport/controlbuf.go b/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
index 8394d25..be5a9c8 100644
--- a/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
+++ b/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
@@ -22,6 +22,7 @@
"bytes"
"errors"
"fmt"
+ "net"
"runtime"
"strconv"
"sync"
@@ -29,6 +30,7 @@
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
+ "google.golang.org/grpc/internal/grpclog"
"google.golang.org/grpc/internal/grpcutil"
"google.golang.org/grpc/status"
)
@@ -137,6 +139,7 @@
streamID uint32
contentSubtype string
status *status.Status
+ rst bool
}
func (*earlyAbortStream) isTransportResponseFrame() bool { return false }
@@ -190,7 +193,7 @@
code http2.ErrCode
debugData []byte
headsUp bool
- closeConn bool
+ closeConn error // if set, loopyWriter will exit, resulting in conn closure
}
func (*goAway) isTransportResponseFrame() bool { return false }
@@ -208,6 +211,14 @@
func (*outFlowControlSizeRequest) isTransportResponseFrame() bool { return false }
+// closeConnection is an instruction to tell the loopy writer to flush the
+// framer and exit, which will cause the transport's connection to be closed
+// (by the client or server). The transport itself will close after the reader
+// encounters the EOF caused by the connection closure.
+type closeConnection struct{}
+
+func (closeConnection) isTransportResponseFrame() bool { return false }
+
type outStreamState int
const (
@@ -407,7 +418,7 @@
select {
case <-c.ch:
case <-c.done:
- return nil, ErrConnClosing
+ return nil, errors.New("transport closed by client")
}
}
}
@@ -477,12 +488,14 @@
hEnc *hpack.Encoder // HPACK encoder.
bdpEst *bdpEstimator
draining bool
+ conn net.Conn
+ logger *grpclog.PrefixLogger
// Side-specific handlers
ssGoAwayHandler func(*goAway) (bool, error)
}
-func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimator) *loopyWriter {
+func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimator, conn net.Conn, logger *grpclog.PrefixLogger) *loopyWriter {
var buf bytes.Buffer
l := &loopyWriter{
side: s,
@@ -495,6 +508,8 @@
hBuf: &buf,
hEnc: hpack.NewEncoder(&buf),
bdpEst: bdpEst,
+ conn: conn,
+ logger: logger,
}
return l
}
@@ -512,23 +527,26 @@
// 2. Stream level flow control quota available.
//
// In each iteration of run loop, other than processing the incoming control
-// frame, loopy calls processData, which processes one node from the activeStreams linked-list.
-// This results in writing of HTTP2 frames into an underlying write buffer.
-// When there's no more control frames to read from controlBuf, loopy flushes the write buffer.
-// As an optimization, to increase the batch size for each flush, loopy yields the processor, once
-// if the batch size is too low to give stream goroutines a chance to fill it up.
+// frame, loopy calls processData, which processes one node from the
+// activeStreams linked-list. This results in writing of HTTP2 frames into an
+// underlying write buffer. When there's no more control frames to read from
+// controlBuf, loopy flushes the write buffer. As an optimization, to increase
+// the batch size for each flush, loopy yields the processor, once if the batch
+// size is too low to give stream goroutines a chance to fill it up.
+//
+// Upon exiting, if the error causing the exit is not an I/O error, run()
+// flushes and closes the underlying connection. Otherwise, the connection is
+// left open to allow the I/O error to be encountered by the reader instead.
func (l *loopyWriter) run() (err error) {
defer func() {
- if err == ErrConnClosing {
- // Don't log ErrConnClosing as error since it happens
- // 1. When the connection is closed by some other known issue.
- // 2. User closed the connection.
- // 3. A graceful close of connection.
- if logger.V(logLevel) {
- logger.Infof("transport: loopyWriter.run returning. %v", err)
- }
- err = nil
+ if l.logger.V(logLevel) {
+ l.logger.Infof("loopyWriter exiting with error: %v", err)
}
+ if !isIOError(err) {
+ l.framer.writer.Flush()
+ l.conn.Close()
+ }
+ l.cbuf.finish()
}()
for {
it, err := l.cbuf.get(true)
@@ -573,7 +591,6 @@
}
l.framer.writer.Flush()
break hasdata
-
}
}
}
@@ -582,11 +599,11 @@
return l.framer.fr.WriteWindowUpdate(w.streamID, w.increment)
}
-func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) error {
+func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) {
// Otherwise update the quota.
if w.streamID == 0 {
l.sendQuota += w.increment
- return nil
+ return
}
// Find the stream and update it.
if str, ok := l.estdStreams[w.streamID]; ok {
@@ -594,10 +611,9 @@
if strQuota := int(l.oiws) - str.bytesOutStanding; strQuota > 0 && str.state == waitingOnStreamQuota {
str.state = active
l.activeStreams.enqueue(str)
- return nil
+ return
}
}
- return nil
}
func (l *loopyWriter) outgoingSettingsHandler(s *outgoingSettings) error {
@@ -605,13 +621,11 @@
}
func (l *loopyWriter) incomingSettingsHandler(s *incomingSettings) error {
- if err := l.applySettings(s.ss); err != nil {
- return err
- }
+ l.applySettings(s.ss)
return l.framer.fr.WriteSettingsAck()
}
-func (l *loopyWriter) registerStreamHandler(h *registerStream) error {
+func (l *loopyWriter) registerStreamHandler(h *registerStream) {
str := &outStream{
id: h.streamID,
state: empty,
@@ -619,15 +633,14 @@
wq: h.wq,
}
l.estdStreams[h.streamID] = str
- return nil
}
func (l *loopyWriter) headerHandler(h *headerFrame) error {
if l.side == serverSide {
str, ok := l.estdStreams[h.streamID]
if !ok {
- if logger.V(logLevel) {
- logger.Warningf("transport: loopy doesn't recognize the stream: %d", h.streamID)
+ if l.logger.V(logLevel) {
+ l.logger.Infof("Unrecognized streamID %d in loopyWriter", h.streamID)
}
return nil
}
@@ -654,19 +667,20 @@
itl: &itemList{},
wq: h.wq,
}
- str.itl.enqueue(h)
- return l.originateStream(str)
+ return l.originateStream(str, h)
}
-func (l *loopyWriter) originateStream(str *outStream) error {
- hdr := str.itl.dequeue().(*headerFrame)
- if err := hdr.initStream(str.id); err != nil {
- if err == ErrConnClosing {
- return err
- }
- // Other errors(errStreamDrain) need not close transport.
+func (l *loopyWriter) originateStream(str *outStream, hdr *headerFrame) error {
+ // l.draining is set when handling GoAway. In which case, we want to avoid
+ // creating new streams.
+ if l.draining {
+ // TODO: provide a better error with the reason we are in draining.
+ hdr.onOrphaned(errStreamDrain)
return nil
}
+ if err := hdr.initStream(str.id); err != nil {
+ return err
+ }
if err := l.writeHeader(str.id, hdr.endStream, hdr.hf, hdr.onWrite); err != nil {
return err
}
@@ -681,8 +695,8 @@
l.hBuf.Reset()
for _, f := range hf {
if err := l.hEnc.WriteField(f); err != nil {
- if logger.V(logLevel) {
- logger.Warningf("transport: loopyWriter.writeHeader encountered error while encoding headers: %v", err)
+ if l.logger.V(logLevel) {
+ l.logger.Warningf("Encountered error while encoding headers: %v", err)
}
}
}
@@ -720,10 +734,10 @@
return nil
}
-func (l *loopyWriter) preprocessData(df *dataFrame) error {
+func (l *loopyWriter) preprocessData(df *dataFrame) {
str, ok := l.estdStreams[df.streamID]
if !ok {
- return nil
+ return
}
// If we got data for a stream it means that
// stream was originated and the headers were sent out.
@@ -732,7 +746,6 @@
str.state = active
l.activeStreams.enqueue(str)
}
- return nil
}
func (l *loopyWriter) pingHandler(p *ping) error {
@@ -743,9 +756,8 @@
}
-func (l *loopyWriter) outFlowControlSizeRequestHandler(o *outFlowControlSizeRequest) error {
+func (l *loopyWriter) outFlowControlSizeRequestHandler(o *outFlowControlSizeRequest) {
o.resp <- l.sendQuota
- return nil
}
func (l *loopyWriter) cleanupStreamHandler(c *cleanupStream) error {
@@ -762,8 +774,9 @@
return err
}
}
- if l.side == clientSide && l.draining && len(l.estdStreams) == 0 {
- return ErrConnClosing
+ if l.draining && len(l.estdStreams) == 0 {
+ // Flush and close the connection; we are done with it.
+ return errors.New("finished processing active streams while in draining mode")
}
return nil
}
@@ -786,6 +799,11 @@
if err := l.writeHeader(eas.streamID, true, headerFields, nil); err != nil {
return err
}
+ if eas.rst {
+ if err := l.framer.fr.WriteRSTStream(eas.streamID, http2.ErrCodeNo); err != nil {
+ return err
+ }
+ }
return nil
}
@@ -793,7 +811,8 @@
if l.side == clientSide {
l.draining = true
if len(l.estdStreams) == 0 {
- return ErrConnClosing
+ // Flush and close the connection; we are done with it.
+ return errors.New("received GOAWAY with no active streams")
}
}
return nil
@@ -814,7 +833,7 @@
func (l *loopyWriter) handle(i interface{}) error {
switch i := i.(type) {
case *incomingWindowUpdate:
- return l.incomingWindowUpdateHandler(i)
+ l.incomingWindowUpdateHandler(i)
case *outgoingWindowUpdate:
return l.outgoingWindowUpdateHandler(i)
case *incomingSettings:
@@ -824,7 +843,7 @@
case *headerFrame:
return l.headerHandler(i)
case *registerStream:
- return l.registerStreamHandler(i)
+ l.registerStreamHandler(i)
case *cleanupStream:
return l.cleanupStreamHandler(i)
case *earlyAbortStream:
@@ -832,19 +851,24 @@
case *incomingGoAway:
return l.incomingGoAwayHandler(i)
case *dataFrame:
- return l.preprocessData(i)
+ l.preprocessData(i)
case *ping:
return l.pingHandler(i)
case *goAway:
return l.goAwayHandler(i)
case *outFlowControlSizeRequest:
- return l.outFlowControlSizeRequestHandler(i)
+ l.outFlowControlSizeRequestHandler(i)
+ case closeConnection:
+ // Just return a non-I/O error and run() will flush and close the
+ // connection.
+ return ErrConnClosing
default:
return fmt.Errorf("transport: unknown control message type %T", i)
}
+ return nil
}
-func (l *loopyWriter) applySettings(ss []http2.Setting) error {
+func (l *loopyWriter) applySettings(ss []http2.Setting) {
for _, s := range ss {
switch s.ID {
case http2.SettingInitialWindowSize:
@@ -863,7 +887,6 @@
updateHeaderTblSize(l.hEnc, s.Val)
}
}
- return nil
}
// processData removes the first stream from active streams, writes out at most 16KB
@@ -880,9 +903,9 @@
dataItem := str.itl.peek().(*dataFrame) // Peek at the first data item this stream.
// A data item is represented by a dataFrame, since it later translates into
// multiple HTTP2 data frames.
- // Every dataFrame has two buffers; h that keeps grpc-message header and d that is acutal data.
+ // Every dataFrame has two buffers; h that keeps grpc-message header and d that is actual data.
// As an optimization to keep wire traffic low, data from d is copied to h to make as big as the
- // maximum possilbe HTTP2 frame size.
+ // maximum possible HTTP2 frame size.
if len(dataItem.h) == 0 && len(dataItem.d) == 0 { // Empty data frame
// Client sends out empty data frame with endStream = true
@@ -897,7 +920,7 @@
return false, err
}
if err := l.cleanupStreamHandler(trailer.cleanup); err != nil {
- return false, nil
+ return false, err
}
} else {
l.activeStreams.enqueue(str)
diff --git a/vendor/google.golang.org/grpc/internal/transport/defaults.go b/vendor/google.golang.org/grpc/internal/transport/defaults.go
index 9fa306b..bc8ee07 100644
--- a/vendor/google.golang.org/grpc/internal/transport/defaults.go
+++ b/vendor/google.golang.org/grpc/internal/transport/defaults.go
@@ -47,3 +47,9 @@
defaultClientMaxHeaderListSize = uint32(16 << 20)
defaultServerMaxHeaderListSize = uint32(16 << 20)
)
+
+// MaxStreamID is the upper bound for the stream ID before the current
+// transport gracefully closes and new transport is created for subsequent RPCs.
+// This is set to 75% of 2^31-1. Streams are identified with an unsigned 31-bit
+// integer. It's exported so that tests can override it.
+var MaxStreamID = uint32(math.MaxInt32 * 3 / 4)
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 b897cba..98f80e3 100644
--- a/vendor/google.golang.org/grpc/internal/transport/handler_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/handler_server.go
@@ -39,6 +39,7 @@
"golang.org/x/net/http2"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/internal/grpclog"
"google.golang.org/grpc/internal/grpcutil"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/peer"
@@ -46,24 +47,32 @@
"google.golang.org/grpc/status"
)
-// NewServerHandlerTransport returns a ServerTransport handling gRPC
-// from inside an http.Handler. It requires that the http Server
-// supports HTTP/2.
-func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request, stats stats.Handler) (ServerTransport, error) {
+// NewServerHandlerTransport returns a ServerTransport handling gRPC from
+// inside an http.Handler, or writes an HTTP error to w and returns an error.
+// It requires that the http Server supports HTTP/2.
+func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request, stats []stats.Handler) (ServerTransport, error) {
if r.ProtoMajor != 2 {
- return nil, errors.New("gRPC requires HTTP/2")
+ msg := "gRPC requires HTTP/2"
+ http.Error(w, msg, http.StatusBadRequest)
+ return nil, errors.New(msg)
}
if r.Method != "POST" {
- return nil, errors.New("invalid gRPC request method")
+ msg := fmt.Sprintf("invalid gRPC request method %q", r.Method)
+ http.Error(w, msg, http.StatusBadRequest)
+ return nil, errors.New(msg)
}
contentType := r.Header.Get("Content-Type")
// TODO: do we assume contentType is lowercase? we did before
contentSubtype, validContentType := grpcutil.ContentSubtype(contentType)
if !validContentType {
- return nil, errors.New("invalid gRPC request content-type")
+ msg := fmt.Sprintf("invalid gRPC request content-type %q", contentType)
+ http.Error(w, msg, http.StatusUnsupportedMediaType)
+ return nil, errors.New(msg)
}
if _, ok := w.(http.Flusher); !ok {
- return nil, errors.New("gRPC requires a ResponseWriter supporting http.Flusher")
+ msg := "gRPC requires a ResponseWriter supporting http.Flusher"
+ http.Error(w, msg, http.StatusInternalServerError)
+ return nil, errors.New(msg)
}
st := &serverHandlerTransport{
@@ -75,11 +84,14 @@
contentSubtype: contentSubtype,
stats: stats,
}
+ st.logger = prefixLoggerForServerHandlerTransport(st)
if v := r.Header.Get("grpc-timeout"); v != "" {
to, err := decodeTimeout(v)
if err != nil {
- return nil, status.Errorf(codes.Internal, "malformed time-out: %v", err)
+ msg := fmt.Sprintf("malformed grpc-timeout: %v", err)
+ http.Error(w, msg, http.StatusBadRequest)
+ return nil, status.Error(codes.Internal, msg)
}
st.timeoutSet = true
st.timeout = to
@@ -97,7 +109,9 @@
for _, v := range vv {
v, err := decodeMetadataHeader(k, v)
if err != nil {
- return nil, status.Errorf(codes.Internal, "malformed binary metadata: %v", err)
+ msg := fmt.Sprintf("malformed binary metadata %q in header %q: %v", v, k, err)
+ http.Error(w, msg, http.StatusBadRequest)
+ return nil, status.Error(codes.Internal, msg)
}
metakv = append(metakv, k, v)
}
@@ -138,15 +152,19 @@
// TODO make sure this is consistent across handler_server and http2_server
contentSubtype string
- stats stats.Handler
+ stats []stats.Handler
+ logger *grpclog.PrefixLogger
}
-func (ht *serverHandlerTransport) Close() {
- ht.closeOnce.Do(ht.closeCloseChanOnce)
+func (ht *serverHandlerTransport) Close(err error) {
+ ht.closeOnce.Do(func() {
+ if ht.logger.V(logLevel) {
+ ht.logger.Infof("Closing: %v", err)
+ }
+ close(ht.closedCh)
+ })
}
-func (ht *serverHandlerTransport) closeCloseChanOnce() { close(ht.closedCh) }
-
func (ht *serverHandlerTransport) RemoteAddr() net.Addr { return strAddr(ht.req.RemoteAddr) }
// strAddr is a net.Addr backed by either a TCP "ip:port" string, or
@@ -228,15 +246,15 @@
})
if err == nil { // transport has not been closed
- if ht.stats != nil {
- // Note: The trailer fields are compressed with hpack after this call returns.
- // No WireLength field is set here.
- ht.stats.HandleRPC(s.Context(), &stats.OutTrailer{
+ // Note: The trailer fields are compressed with hpack after this call returns.
+ // No WireLength field is set here.
+ for _, sh := range ht.stats {
+ sh.HandleRPC(s.Context(), &stats.OutTrailer{
Trailer: s.trailer.Copy(),
})
}
}
- ht.Close()
+ ht.Close(errors.New("finished writing status"))
return err
}
@@ -314,10 +332,10 @@
})
if err == nil {
- if ht.stats != nil {
+ for _, sh := range ht.stats {
// Note: The header fields are compressed with hpack after this call returns.
// No WireLength field is set here.
- ht.stats.HandleRPC(s.Context(), &stats.OutHeader{
+ sh.HandleRPC(s.Context(), &stats.OutHeader{
Header: md.Copy(),
Compression: s.sendCompress,
})
@@ -346,7 +364,7 @@
case <-ht.req.Context().Done():
}
cancel()
- ht.Close()
+ ht.Close(errors.New("request is done processing"))
}()
req := ht.req
@@ -369,14 +387,14 @@
}
ctx = metadata.NewIncomingContext(ctx, ht.headerMD)
s.ctx = peer.NewContext(ctx, pr)
- if ht.stats != nil {
- s.ctx = ht.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
+ for _, sh := range ht.stats {
+ s.ctx = sh.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
inHeader := &stats.InHeader{
FullMethod: s.method,
RemoteAddr: ht.RemoteAddr(),
Compression: s.recvCompress,
}
- ht.stats.HandleRPC(s.ctx, inHeader)
+ sh.HandleRPC(s.ctx, inHeader)
}
s.trReader = &transportReader{
reader: &recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: s.buf, freeBuffer: func(*bytes.Buffer) {}},
@@ -435,7 +453,7 @@
func (ht *serverHandlerTransport) IncrMsgRecv() {}
-func (ht *serverHandlerTransport) Drain() {
+func (ht *serverHandlerTransport) Drain(debugData string) {
panic("Drain() is not implemented")
}
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 10b5461..326bf08 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_client.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
@@ -38,8 +38,11 @@
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/internal/channelz"
icredentials "google.golang.org/grpc/internal/credentials"
+ "google.golang.org/grpc/internal/grpclog"
+ "google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/internal/grpcutil"
imetadata "google.golang.org/grpc/internal/metadata"
+ istatus "google.golang.org/grpc/internal/status"
"google.golang.org/grpc/internal/syscall"
"google.golang.org/grpc/internal/transport/networktype"
"google.golang.org/grpc/keepalive"
@@ -57,11 +60,15 @@
// http2Client implements the ClientTransport interface with HTTP2.
type http2Client struct {
- lastRead int64 // Keep this field 64-bit aligned. Accessed atomically.
- ctx context.Context
- cancel context.CancelFunc
- ctxDone <-chan struct{} // Cache the ctx.Done() chan.
- userAgent string
+ lastRead int64 // Keep this field 64-bit aligned. Accessed atomically.
+ ctx context.Context
+ cancel context.CancelFunc
+ ctxDone <-chan struct{} // Cache the ctx.Done() chan.
+ userAgent string
+ // address contains the resolver returned address for this transport.
+ // If the `ServerName` field is set, it takes precedence over `CallHdr.Host`
+ // passed to `NewStream`, when determining the :authority header.
+ address resolver.Address
md metadata.MD
conn net.Conn // underlying communication channel
loopy *loopyWriter
@@ -78,6 +85,7 @@
framer *framer
// controlBuf delivers all the control related tasks (e.g., window
// updates, reset streams, and various settings) to the controller.
+ // Do not access controlBuf with mu held.
controlBuf *controlBuffer
fc *trInFlow
// The scheme used: https if TLS is on, http otherwise.
@@ -90,7 +98,7 @@
kp keepalive.ClientParameters
keepaliveEnabled bool
- statsHandler stats.Handler
+ statsHandlers []stats.Handler
initialWindowSize int32
@@ -98,17 +106,15 @@
maxSendHeaderListSize *uint32
bdpEst *bdpEstimator
- // onPrefaceReceipt is a callback that client transport calls upon
- // receiving server preface to signal that a succefull HTTP2
- // connection was established.
- onPrefaceReceipt func()
maxConcurrentStreams uint32
streamQuota int64
streamsQuotaAvailable chan struct{}
waitingStreams uint32
nextID uint32
+ registeredCompressors string
+ // Do not access controlBuf with mu held.
mu sync.Mutex // guard the following variables
state transportState
activeStreams map[uint32]*Stream
@@ -132,15 +138,15 @@
kpDormant bool
// Fields below are for channelz metric collection.
- channelzID int64 // channelz unique identification number
+ channelzID *channelz.Identifier
czData *channelzData
- onGoAway func(GoAwayReason)
- onClose func()
+ onClose func(GoAwayReason)
bufferPool *bufferPool
connectionID uint64
+ logger *grpclog.PrefixLogger
}
func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error), addr resolver.Address, useProxy bool, grpcUA string) (net.Conn, error) {
@@ -192,7 +198,7 @@
// 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(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (_ *http2Client, err error) {
+func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onClose func(GoAwayReason)) (_ *http2Client, err error) {
scheme := "http"
ctx, cancel := context.WithCancel(ctx)
defer func() {
@@ -212,14 +218,40 @@
if opts.FailOnNonTempDialError {
return nil, connectionErrorf(isTemporary(err), err, "transport: error while dialing: %v", err)
}
- return nil, connectionErrorf(true, err, "transport: Error while dialing %v", err)
+ return nil, connectionErrorf(true, err, "transport: Error while dialing: %v", err)
}
+
// Any further errors will close the underlying connection
defer func(conn net.Conn) {
if err != nil {
conn.Close()
}
}(conn)
+
+ // The following defer and goroutine monitor the connectCtx for cancelation
+ // and deadline. On context expiration, the connection is hard closed and
+ // this function will naturally fail as a result. Otherwise, the defer
+ // waits for the goroutine to exit to prevent the context from being
+ // monitored (and to prevent the connection from ever being closed) after
+ // returning from this function.
+ ctxMonitorDone := grpcsync.NewEvent()
+ newClientCtx, newClientDone := context.WithCancel(connectCtx)
+ defer func() {
+ newClientDone() // Awaken the goroutine below if connectCtx hasn't expired.
+ <-ctxMonitorDone.Done() // Wait for the goroutine below to exit.
+ }()
+ go func(conn net.Conn) {
+ defer ctxMonitorDone.Fire() // Signal this goroutine has exited.
+ <-newClientCtx.Done() // Block until connectCtx expires or the defer above executes.
+ if err := connectCtx.Err(); err != nil {
+ // connectCtx expired before exiting the function. Hard close the connection.
+ if logger.V(logLevel) {
+ logger.Infof("Aborting due to connect deadline expiring: %v", err)
+ }
+ conn.Close()
+ }
+ }(conn)
+
kp := opts.KeepaliveParams
// Validate keepalive parameters.
if kp.Time == 0 {
@@ -251,15 +283,7 @@
}
}
if transportCreds != nil {
- 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{})
+ conn, authInfo, err = transportCreds.ClientHandshake(connectCtx, addr.ServerName, conn)
if err != nil {
return nil, connectionErrorf(isTemporary(err), err, "transport: authentication handshake failed: %v", err)
}
@@ -297,6 +321,8 @@
ctxDone: ctx.Done(), // Cache Done chan.
cancel: cancel,
userAgent: opts.UserAgent,
+ registeredCompressors: grpcutil.RegisteredCompressors(),
+ address: addr,
conn: conn,
remoteAddr: conn.RemoteAddr(),
localAddr: conn.LocalAddr(),
@@ -311,19 +337,20 @@
isSecure: isSecure,
perRPCCreds: perRPCCreds,
kp: kp,
- statsHandler: opts.StatsHandler,
+ statsHandlers: opts.StatsHandlers,
initialWindowSize: initialWindowSize,
- onPrefaceReceipt: onPrefaceReceipt,
nextID: 1,
maxConcurrentStreams: defaultMaxStreamsClient,
streamQuota: defaultMaxStreamsClient,
streamsQuotaAvailable: make(chan struct{}, 1),
czData: new(channelzData),
- onGoAway: onGoAway,
- onClose: onClose,
keepaliveEnabled: keepaliveEnabled,
bufferPool: newBufferPool(),
+ onClose: onClose,
}
+ t.logger = prefixLoggerForClientTransport(t)
+ // Add peer information to the http2client context.
+ t.ctx = peer.NewContext(t.ctx, t.getPeer())
if md, ok := addr.Metadata.(*metadata.MD); ok {
t.md = *md
@@ -341,38 +368,50 @@
updateFlowControl: t.updateFlowControl,
}
}
- if t.statsHandler != nil {
- t.ctx = t.statsHandler.TagConn(t.ctx, &stats.ConnTagInfo{
+ for _, sh := range t.statsHandlers {
+ t.ctx = sh.TagConn(t.ctx, &stats.ConnTagInfo{
RemoteAddr: t.remoteAddr,
LocalAddr: t.localAddr,
})
connBegin := &stats.ConnBegin{
Client: true,
}
- t.statsHandler.HandleConn(t.ctx, connBegin)
+ sh.HandleConn(t.ctx, connBegin)
}
- if channelz.IsOn() {
- t.channelzID = channelz.RegisterNormalSocket(t, opts.ChannelzParentID, fmt.Sprintf("%s -> %s", t.localAddr, t.remoteAddr))
+ t.channelzID, err = channelz.RegisterNormalSocket(t, opts.ChannelzParentID, fmt.Sprintf("%s -> %s", t.localAddr, t.remoteAddr))
+ if err != nil {
+ return nil, err
}
if t.keepaliveEnabled {
t.kpDormancyCond = sync.NewCond(&t.mu)
go t.keepalive()
}
- // Start the reader goroutine for incoming message. Each transport has
- // a dedicated goroutine which reads HTTP2 frame from network. Then it
- // dispatches the frame to the corresponding stream entity.
- go t.reader()
+
+ // Start the reader goroutine for incoming messages. Each transport has a
+ // dedicated goroutine which reads HTTP2 frames from the network. Then it
+ // dispatches the frame to the corresponding stream entity. When the
+ // server preface is received, readerErrCh is closed. If an error occurs
+ // first, an error is pushed to the channel. This must be checked before
+ // returning from this function.
+ readerErrCh := make(chan error, 1)
+ go t.reader(readerErrCh)
+ defer func() {
+ if err == nil {
+ err = <-readerErrCh
+ }
+ if err != nil {
+ t.Close(err)
+ }
+ }()
// Send connection preface to server.
n, err := t.conn.Write(clientPreface)
if err != nil {
err = connectionErrorf(true, err, "transport: failed to write client preface: %v", err)
- t.Close(err)
return nil, err
}
if n != len(clientPreface) {
err = connectionErrorf(true, nil, "transport: preface mismatch, wrote %d bytes; want %d", n, len(clientPreface))
- t.Close(err)
return nil, err
}
var ss []http2.Setting
@@ -392,14 +431,12 @@
err = t.framer.fr.WriteSettings(ss...)
if err != nil {
err = connectionErrorf(true, err, "transport: failed to write initial settings frame: %v", err)
- t.Close(err)
return nil, err
}
// Adjust the connection flow control window if needed.
if delta := uint32(icwz - defaultWindowSize); delta > 0 {
if err := t.framer.fr.WriteWindowUpdate(0, delta); err != nil {
err = connectionErrorf(true, err, "transport: failed to write window update: %v", err)
- t.Close(err)
return nil, err
}
}
@@ -410,17 +447,8 @@
return nil, err
}
go func() {
- t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst)
- err := t.loopy.run()
- if err != nil {
- if logger.V(logLevel) {
- logger.Errorf("transport: loopyWriter.run returning. Err: %v", err)
- }
- }
- // Do not close the transport. Let reader goroutine handle it since
- // there might be data in the buffers.
- t.conn.Close()
- t.controlBuf.finish()
+ t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst, t.conn, t.logger)
+ t.loopy.run()
close(t.writerDone)
}()
return t, nil
@@ -466,7 +494,7 @@
func (t *http2Client) getPeer() *peer.Peer {
return &peer.Peer{
Addr: t.remoteAddr,
- AuthInfo: t.authInfo,
+ AuthInfo: t.authInfo, // Can be nil
}
}
@@ -502,9 +530,22 @@
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-previous-rpc-attempts", Value: strconv.Itoa(callHdr.PreviousAttempts)})
}
+ registeredCompressors := t.registeredCompressors
if callHdr.SendCompress != "" {
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-encoding", Value: callHdr.SendCompress})
- headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-accept-encoding", Value: callHdr.SendCompress})
+ // Include the outgoing compressor name when compressor is not registered
+ // via encoding.RegisterCompressor. This is possible when client uses
+ // WithCompressor dial option.
+ if !grpcutil.IsCompressorNameRegistered(callHdr.SendCompress) {
+ if registeredCompressors != "" {
+ registeredCompressors += ","
+ }
+ registeredCompressors += callHdr.SendCompress
+ }
+ }
+
+ if registeredCompressors != "" {
+ headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-accept-encoding", Value: registeredCompressors})
}
if dl, ok := ctx.Deadline(); ok {
// Send out timeout regardless its value. The server can detect timeout context by itself.
@@ -584,7 +625,11 @@
for _, c := range t.perRPCCreds {
data, err := c.GetRequestMetadata(ctx, audience)
if err != nil {
- if _, ok := status.FromError(err); ok {
+ if st, ok := status.FromError(err); ok {
+ // Restrict the code to the list allowed by gRFC A54.
+ if istatus.IsRestrictedControlPlaneCode(st) {
+ err = status.Errorf(codes.Internal, "transport: received per-RPC creds error with illegal status: %v", err)
+ }
return nil, err
}
@@ -613,7 +658,14 @@
}
data, err := callCreds.GetRequestMetadata(ctx, audience)
if err != nil {
- return nil, status.Errorf(codes.Internal, "transport: %v", err)
+ if st, ok := status.FromError(err); ok {
+ // Restrict the code to the list allowed by gRFC A54.
+ if istatus.IsRestrictedControlPlaneCode(st) {
+ err = status.Errorf(codes.Internal, "transport: received per-RPC creds error with illegal status: %v", err)
+ }
+ return nil, err
+ }
+ return nil, status.Errorf(codes.Internal, "transport: per-RPC creds failed due to error: %v", err)
}
callAuthData = make(map[string]string, len(data))
for k, v := range data {
@@ -630,8 +682,8 @@
// 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.
+// server. It's possible this could succeed on another transport, even if
+// it's unlikely, but do not transparently retry.
// 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
@@ -639,8 +691,7 @@
type NewStreamError struct {
Err error
- DoNotRetry bool
- DoNotTransparentRetry bool
+ AllowTransparentRetry bool
}
func (e NewStreamError) Error() string {
@@ -649,11 +700,23 @@
// NewStream creates a stream and registers it into the transport as "active"
// streams. All non-nil errors returned will be *NewStreamError.
-func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {
+func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error) {
ctx = peer.NewContext(ctx, t.getPeer())
+
+ // ServerName field of the resolver returned address takes precedence over
+ // Host field of CallHdr to determine the :authority header. This is because,
+ // the ServerName field takes precedence for server authentication during
+ // TLS handshake, and the :authority header should match the value used
+ // for server authentication.
+ if t.address.ServerName != "" {
+ newCallHdr := *callHdr
+ newCallHdr.Host = t.address.ServerName
+ callHdr = &newCallHdr
+ }
+
headerFields, err := t.createHeaderFields(ctx, callHdr)
if err != nil {
- return nil, &NewStreamError{Err: err, DoNotTransparentRetry: true}
+ return nil, &NewStreamError{Err: err, AllowTransparentRetry: false}
}
s := t.newStream(ctx, callHdr)
cleanup := func(err error) {
@@ -675,17 +738,13 @@
endStream: false,
initStream: func(id uint32) error {
t.mu.Lock()
- if state := t.state; state != reachable {
+ // TODO: handle transport closure in loopy instead and remove this
+ // initStream is never called when transport is draining.
+ if t.state == closing {
t.mu.Unlock()
- // Do a quick cleanup.
- err := error(errStreamDrain)
- if state == closing {
- err = ErrConnClosing
- }
- cleanup(err)
- return err
+ cleanup(ErrConnClosing)
+ return ErrConnClosing
}
- t.activeStreams[id] = s
if channelz.IsOn() {
atomic.AddInt64(&t.czData.streamsStarted, 1)
atomic.StoreInt64(&t.czData.lastStreamCreatedTime, time.Now().UnixNano())
@@ -702,6 +761,7 @@
}
firstTry := true
var ch chan struct{}
+ transportDrainRequired := false
checkForStreamQuota := func(it interface{}) bool {
if t.streamQuota <= 0 { // Can go negative if server decreases it.
if firstTry {
@@ -717,8 +777,20 @@
h := it.(*headerFrame)
h.streamID = t.nextID
t.nextID += 2
+
+ // Drain client transport if nextID > MaxStreamID which signals gRPC that
+ // the connection is closed and a new one must be created for subsequent RPCs.
+ transportDrainRequired = t.nextID > MaxStreamID
+
s.id = h.streamID
s.fc = &inFlow{limit: uint32(t.initialWindowSize)}
+ t.mu.Lock()
+ if t.state == draining || t.activeStreams == nil { // Can be niled from Close().
+ t.mu.Unlock()
+ return false // Don't create a stream if the transport is already closed.
+ }
+ t.activeStreams[s.id] = s
+ t.mu.Unlock()
if t.streamQuota > 0 && t.waitingStreams > 0 {
select {
case t.streamsQuotaAvailable <- struct{}{}:
@@ -744,22 +816,17 @@
}
for {
success, err := t.controlBuf.executeAndPut(func(it interface{}) bool {
- if !checkForStreamQuota(it) {
- return false
- }
- if !checkForHeaderListSize(it) {
- return false
- }
- return true
+ return checkForHeaderListSize(it) && checkForStreamQuota(it)
}, hdr)
if err != nil {
- return nil, &NewStreamError{Err: err}
+ // Connection closed.
+ return nil, &NewStreamError{Err: err, AllowTransparentRetry: true}
}
if success {
break
}
if hdrListSizeErr != nil {
- return nil, &NewStreamError{Err: hdrListSizeErr, DoNotRetry: true}
+ return nil, &NewStreamError{Err: hdrListSizeErr}
}
firstTry = false
select {
@@ -767,29 +834,38 @@
case <-ctx.Done():
return nil, &NewStreamError{Err: ContextErr(ctx.Err())}
case <-t.goAway:
- return nil, &NewStreamError{Err: errStreamDrain}
+ return nil, &NewStreamError{Err: errStreamDrain, AllowTransparentRetry: true}
case <-t.ctx.Done():
- return nil, &NewStreamError{Err: ErrConnClosing}
+ return nil, &NewStreamError{Err: ErrConnClosing, AllowTransparentRetry: true}
}
}
- if t.statsHandler != nil {
+ if len(t.statsHandlers) != 0 {
header, ok := metadata.FromOutgoingContext(ctx)
if ok {
header.Set("user-agent", t.userAgent)
} else {
header = metadata.Pairs("user-agent", t.userAgent)
}
- // Note: The header fields are compressed with hpack after this call returns.
- // No WireLength field is set here.
- outHeader := &stats.OutHeader{
- Client: true,
- FullMethod: callHdr.Method,
- RemoteAddr: t.remoteAddr,
- LocalAddr: t.localAddr,
- Compression: callHdr.SendCompress,
- Header: header,
+ for _, sh := range t.statsHandlers {
+ // Note: The header fields are compressed with hpack after this call returns.
+ // No WireLength field is set here.
+ // Note: Creating a new stats object to prevent pollution.
+ outHeader := &stats.OutHeader{
+ Client: true,
+ FullMethod: callHdr.Method,
+ RemoteAddr: t.remoteAddr,
+ LocalAddr: t.localAddr,
+ Compression: callHdr.SendCompress,
+ Header: header,
+ }
+ sh.HandleRPC(s.ctx, outHeader)
}
- t.statsHandler.HandleRPC(s.ctx, outHeader)
+ }
+ if transportDrainRequired {
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Draining transport: t.nextID > MaxStreamID")
+ }
+ t.GracefulClose()
}
return s, nil
}
@@ -872,20 +948,21 @@
// Close kicks off the shutdown process of the transport. This should be called
// only once on a transport. Once it is called, the transport should not be
// accessed any more.
-//
-// This method blocks until the addrConn that initiated this transport is
-// re-connected. This happens because t.onClose() begins reconnect logic at the
-// addrConn level and blocks until the addrConn is successfully connected.
func (t *http2Client) Close(err error) {
t.mu.Lock()
- // Make sure we only Close once.
+ // Make sure we only close once.
if t.state == closing {
t.mu.Unlock()
return
}
- // Call t.onClose before setting the state to closing to prevent the client
- // from attempting to create new streams ASAP.
- t.onClose()
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Closing: %v", err)
+ }
+ // Call t.onClose ASAP to prevent the client from attempting to create new
+ // streams.
+ if t.state != draining {
+ t.onClose(GoAwayInvalid)
+ }
t.state = closing
streams := t.activeStreams
t.activeStreams = nil
@@ -898,9 +975,7 @@
t.controlBuf.finish()
t.cancel()
t.conn.Close()
- if channelz.IsOn() {
- channelz.RemoveEntry(t.channelzID)
- }
+ channelz.RemoveEntry(t.channelzID)
// 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()
@@ -917,11 +992,11 @@
for _, s := range streams {
t.closeStream(s, err, false, http2.ErrCodeNo, st, nil, false)
}
- if t.statsHandler != nil {
+ for _, sh := range t.statsHandlers {
connEnd := &stats.ConnEnd{
Client: true,
}
- t.statsHandler.HandleConn(t.ctx, connEnd)
+ sh.HandleConn(t.ctx, connEnd)
}
}
@@ -937,11 +1012,15 @@
t.mu.Unlock()
return
}
+ if t.logger.V(logLevel) {
+ t.logger.Infof("GracefulClose called")
+ }
+ t.onClose(GoAwayInvalid)
t.state = draining
active := len(t.activeStreams)
t.mu.Unlock()
if active == 0 {
- t.Close(ErrConnClosing)
+ t.Close(connectionErrorf(true, nil, "no active streams left to process while draining"))
return
}
t.controlBuf.put(&incomingGoAway{})
@@ -1001,13 +1080,13 @@
// for the transport and the stream based on the current bdp
// estimation.
func (t *http2Client) updateFlowControl(n uint32) {
- t.mu.Lock()
- for _, s := range t.activeStreams {
- s.fc.newLimit(n)
- }
- t.mu.Unlock()
updateIWS := func(interface{}) bool {
t.initialWindowSize = int32(n)
+ t.mu.Lock()
+ for _, s := range t.activeStreams {
+ s.fc.newLimit(n)
+ }
+ t.mu.Unlock()
return true
}
t.controlBuf.executeAndPut(updateIWS, &outgoingWindowUpdate{streamID: 0, increment: t.fc.newLimit(n)})
@@ -1098,8 +1177,8 @@
}
statusCode, ok := http2ErrConvTab[f.ErrCode]
if !ok {
- if logger.V(logLevel) {
- logger.Warningf("transport: http2Client.handleRSTStream found no mapped gRPC status for the received http2 error %v", f.ErrCode)
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Received a RST_STREAM frame with code %q, but found no mapped gRPC status", f.ErrCode)
}
statusCode = codes.Unknown
}
@@ -1181,10 +1260,12 @@
t.mu.Unlock()
return
}
- if f.ErrCode == http2.ErrCodeEnhanceYourCalm {
- if logger.V(logLevel) {
- logger.Infof("Client received GoAway with http2.ErrCodeEnhanceYourCalm.")
- }
+ if f.ErrCode == http2.ErrCodeEnhanceYourCalm && string(f.DebugData()) == "too_many_pings" {
+ // When a client receives a GOAWAY with error code ENHANCE_YOUR_CALM and debug
+ // data equal to ASCII "too_many_pings", it should log the occurrence at a log level that is
+ // enabled by default and double the configure KEEPALIVE_TIME used for new connections
+ // on that channel.
+ logger.Errorf("Client received GoAway with error code ENHANCE_YOUR_CALM and debug data equal to ASCII \"too_many_pings\".")
}
id := f.LastStreamID
if id > 0 && id%2 == 0 {
@@ -1213,12 +1294,14 @@
default:
t.setGoAwayReason(f)
close(t.goAway)
- t.controlBuf.put(&incomingGoAway{})
+ defer t.controlBuf.put(&incomingGoAway{}) // Defer as t.mu is currently held.
// Notify the clientconn about the GOAWAY before we set the state to
// draining, to allow the client to stop attempting to create streams
// before disallowing new streams on this connection.
- t.onGoAway(t.goAwayReason)
- t.state = draining
+ if t.state != draining {
+ t.onClose(t.goAwayReason)
+ t.state = draining
+ }
}
// All streams with IDs greater than the GoAwayId
// and smaller than the previous GoAway ID should be killed.
@@ -1226,24 +1309,35 @@
if upperLimit == 0 { // This is the first GoAway Frame.
upperLimit = math.MaxUint32 // Kill all streams after the GoAway ID.
}
+
+ t.prevGoAwayID = id
+ if len(t.activeStreams) == 0 {
+ t.mu.Unlock()
+ t.Close(connectionErrorf(true, nil, "received goaway and there are no active streams"))
+ return
+ }
+
+ streamsToClose := make([]*Stream, 0)
for streamID, stream := range t.activeStreams {
if streamID > id && streamID <= upperLimit {
// The stream was unprocessed by the server.
- atomic.StoreUint32(&stream.unprocessed, 1)
- t.closeStream(stream, errStreamDrain, false, http2.ErrCodeNo, statusGoAway, nil, false)
+ if streamID > id && streamID <= upperLimit {
+ atomic.StoreUint32(&stream.unprocessed, 1)
+ streamsToClose = append(streamsToClose, stream)
+ }
}
}
- t.prevGoAwayID = id
- active := len(t.activeStreams)
t.mu.Unlock()
- if active == 0 {
- t.Close(connectionErrorf(true, nil, "received goaway and there are no active streams"))
+ // Called outside t.mu because closeStream can take controlBuf's mu, which
+ // could induce deadlock and is not allowed.
+ for _, stream := range streamsToClose {
+ t.closeStream(stream, errStreamDrain, false, http2.ErrCodeNo, statusGoAway, nil, false)
}
}
// setGoAwayReason sets the value of t.goAwayReason based
// on the GoAway frame received.
-// It expects a lock on transport's mutext to be held by
+// It expects a lock on transport's mutex to be held by
// the caller.
func (t *http2Client) setGoAwayReason(f *http2.GoAwayFrame) {
t.goAwayReason = GoAwayNoReason
@@ -1433,7 +1527,7 @@
close(s.headerChan)
}
- if t.statsHandler != nil {
+ for _, sh := range t.statsHandlers {
if isHeader {
inHeader := &stats.InHeader{
Client: true,
@@ -1441,14 +1535,14 @@
Header: metadata.MD(mdata).Copy(),
Compression: s.recvCompress,
}
- t.statsHandler.HandleRPC(s.ctx, inHeader)
+ sh.HandleRPC(s.ctx, inHeader)
} else {
inTrailer := &stats.InTrailer{
Client: true,
WireLength: int(frame.Header().Length),
Trailer: metadata.MD(mdata).Copy(),
}
- t.statsHandler.HandleRPC(s.ctx, inTrailer)
+ sh.HandleRPC(s.ctx, inTrailer)
}
}
@@ -1465,33 +1559,35 @@
t.closeStream(s, io.EOF, rst, http2.ErrCodeNo, statusGen, mdata, true)
}
-// reader runs as a separate goroutine in charge of reading data from network
-// connection.
-//
-// TODO(zhaoq): currently one reader per transport. Investigate whether this is
-// optimal.
-// TODO(zhaoq): Check the validity of the incoming frame sequence.
-func (t *http2Client) reader() {
- defer close(t.readerDone)
- // Check the validity of server preface.
+// readServerPreface reads and handles the initial settings frame from the
+// server.
+func (t *http2Client) readServerPreface() error {
frame, err := t.framer.fr.ReadFrame()
if err != nil {
- err = connectionErrorf(true, err, "error reading server preface: %v", err)
- t.Close(err) // this kicks off resetTransport, so must be last before return
- return
- }
- t.conn.SetReadDeadline(time.Time{}) // reset deadline once we get the settings frame (we didn't time out, yay!)
- if t.keepaliveEnabled {
- atomic.StoreInt64(&t.lastRead, time.Now().UnixNano())
+ return connectionErrorf(true, err, "error reading server preface: %v", err)
}
sf, ok := frame.(*http2.SettingsFrame)
if !ok {
- // this kicks off resetTransport, so must be last before return
- t.Close(connectionErrorf(true, nil, "initial http2 frame from server is not a settings frame: %T", frame))
+ return connectionErrorf(true, nil, "initial http2 frame from server is not a settings frame: %T", frame)
+ }
+ t.handleSettings(sf, true)
+ return nil
+}
+
+// reader verifies the server preface and reads all subsequent data from
+// network connection. If the server preface is not read successfully, an
+// error is pushed to errCh; otherwise errCh is closed with no error.
+func (t *http2Client) reader(errCh chan<- error) {
+ defer close(t.readerDone)
+
+ if err := t.readServerPreface(); err != nil {
+ errCh <- err
return
}
- t.onPrefaceReceipt()
- t.handleSettings(sf, true)
+ close(errCh)
+ if t.keepaliveEnabled {
+ atomic.StoreInt64(&t.lastRead, time.Now().UnixNano())
+ }
// loop to keep reading incoming messages on this transport.
for {
@@ -1694,3 +1790,9 @@
return -2
}
}
+
+func (t *http2Client) stateForTesting() transportState {
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ return t.state
+}
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 2c6eaf0..79e86ba 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_server.go
@@ -35,12 +35,16 @@
"github.com/golang/protobuf/proto"
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
+ "google.golang.org/grpc/internal/grpclog"
"google.golang.org/grpc/internal/grpcutil"
+ "google.golang.org/grpc/internal/pretty"
+ "google.golang.org/grpc/internal/syscall"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/internal/channelz"
"google.golang.org/grpc/internal/grpcrand"
+ "google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/peer"
@@ -52,10 +56,10 @@
var (
// ErrIllegalHeaderWrite indicates that setting header is illegal because of
// the stream's state.
- ErrIllegalHeaderWrite = errors.New("transport: the stream is done or WriteHeader was already called")
+ ErrIllegalHeaderWrite = status.Error(codes.Internal, "transport: SendHeader called multiple times")
// ErrHeaderListSizeLimitViolation indicates that the header list size is larger
// than the limit set by peer.
- ErrHeaderListSizeLimitViolation = errors.New("transport: trying to send header list size larger than the limit set by peer")
+ ErrHeaderListSizeLimitViolation = status.Error(codes.Internal, "transport: trying to send header list size larger than the limit set by peer")
)
// serverConnectionCounter counts the number of connections a server has seen
@@ -82,7 +86,7 @@
// updates, reset streams, and various settings) to the controller.
controlBuf *controlBuffer
fc *trInFlow
- stats stats.Handler
+ stats []stats.Handler
// Keepalive and max-age parameters for the server.
kp keepalive.ServerParameters
// Keepalive enforcement policy.
@@ -101,13 +105,13 @@
mu sync.Mutex // guard the following
- // 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
- // already underway.
- drainChan chan struct{}
+ // drainEvent 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 drainEvent is
+ // already initialized since draining is already underway.
+ drainEvent *grpcsync.Event
state transportState
activeStreams map[uint32]*Stream
// idle is the time instant when the connection went idle.
@@ -117,7 +121,7 @@
idle time.Time
// Fields below are for channelz metric collection.
- channelzID int64 // channelz unique identification number
+ channelzID *channelz.Identifier
czData *channelzData
bufferPool *bufferPool
@@ -127,6 +131,8 @@
// This lock may not be taken if mu is already held.
maxStreamMu sync.Mutex
maxStreamID uint32 // max stream ID ever seen
+
+ logger *grpclog.PrefixLogger
}
// NewServerTransport creates a http2 transport with conn and configuration
@@ -231,6 +237,11 @@
if kp.Timeout == 0 {
kp.Timeout = defaultServerKeepaliveTimeout
}
+ if kp.Time != infinity {
+ if err = syscall.SetTCPUserTimeout(conn, kp.Timeout); err != nil {
+ return nil, connectionErrorf(false, err, "transport: failed to set TCP_USER_TIMEOUT: %v", err)
+ }
+ }
kep := config.KeepalivePolicy
if kep.MinTime == 0 {
kep.MinTime = defaultKeepalivePolicyMinTime
@@ -252,7 +263,7 @@
fc: &trInFlow{limit: uint32(icwz)},
state: reachable,
activeStreams: make(map[uint32]*Stream),
- stats: config.StatsHandler,
+ stats: config.StatsHandlers,
kp: kp,
idle: time.Now(),
kep: kep,
@@ -260,6 +271,10 @@
czData: new(channelzData),
bufferPool: newBufferPool(),
}
+ t.logger = prefixLoggerForServerTransport(t)
+ // Add peer information to the http2server context.
+ t.ctx = peer.NewContext(t.ctx, t.getPeer())
+
t.controlBuf = newControlBuffer(t.done)
if dynamicWindow {
t.bdpEst = &bdpEstimator{
@@ -267,25 +282,25 @@
updateFlowControl: t.updateFlowControl,
}
}
- if t.stats != nil {
- t.ctx = t.stats.TagConn(t.ctx, &stats.ConnTagInfo{
+ for _, sh := range t.stats {
+ t.ctx = sh.TagConn(t.ctx, &stats.ConnTagInfo{
RemoteAddr: t.remoteAddr,
LocalAddr: t.localAddr,
})
connBegin := &stats.ConnBegin{}
- t.stats.HandleConn(t.ctx, connBegin)
+ sh.HandleConn(t.ctx, connBegin)
}
- if channelz.IsOn() {
- t.channelzID = channelz.RegisterNormalSocket(t, config.ChannelzParentID, fmt.Sprintf("%s -> %s", t.remoteAddr, t.localAddr))
+ t.channelzID, err = channelz.RegisterNormalSocket(t, config.ChannelzParentID, fmt.Sprintf("%s -> %s", t.remoteAddr, t.localAddr))
+ if err != nil {
+ return nil, err
}
t.connectionID = atomic.AddUint64(&serverConnectionCounter, 1)
-
t.framer.writer.Flush()
defer func() {
if err != nil {
- t.Close()
+ t.Close(err)
}
}()
@@ -321,23 +336,18 @@
t.handleSettings(sf)
go func() {
- t.loopy = newLoopyWriter(serverSide, t.framer, t.controlBuf, t.bdpEst)
+ t.loopy = newLoopyWriter(serverSide, t.framer, t.controlBuf, t.bdpEst, t.conn, t.logger)
t.loopy.ssGoAwayHandler = t.outgoingGoAwayHandler
- if err := t.loopy.run(); err != nil {
- if logger.V(logLevel) {
- logger.Errorf("transport: loopyWriter.run returning. Err: %v", err)
- }
- }
- t.conn.Close()
- t.controlBuf.finish()
+ t.loopy.run()
close(t.writerDone)
}()
go t.keepalive()
return t, nil
}
-// 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) {
+// operateHeaders takes action on the decoded headers. Returns an error if fatal
+// error encountered and transport needs to close, otherwise returns nil.
+func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) error {
// Acquire max stream ID lock for entire duration
t.maxStreamMu.Lock()
defer t.maxStreamMu.Unlock()
@@ -353,15 +363,12 @@
rstCode: http2.ErrCodeFrameSize,
onWrite: func() {},
})
- return false
+ return nil
}
if streamID%2 != 1 || streamID <= t.maxStreamID {
// illegal gRPC stream id.
- if logger.V(logLevel) {
- logger.Errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", streamID)
- }
- return true
+ return fmt.Errorf("received an illegal stream id: %v. headers frame: %+v", streamID, frame)
}
t.maxStreamID = streamID
@@ -373,13 +380,14 @@
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
+ // if false, content-type was missing or invalid
+ isGRPC = false
+ contentType = ""
+ mdata = make(metadata.MD, len(frame.Fields))
+ httpMethod string
+ // these are set if an error is encountered while parsing the headers
+ protocolError bool
+ headerError *status.Status
timeoutSet bool
timeout time.Duration
@@ -390,11 +398,23 @@
case "content-type":
contentSubtype, validContentType := grpcutil.ContentSubtype(hf.Value)
if !validContentType {
+ contentType = hf.Value
break
}
mdata[hf.Name] = append(mdata[hf.Name], hf.Value)
s.contentSubtype = contentSubtype
isGRPC = true
+
+ case "grpc-accept-encoding":
+ mdata[hf.Name] = append(mdata[hf.Name], hf.Value)
+ if hf.Value == "" {
+ continue
+ }
+ compressors := hf.Value
+ if s.clientAdvertisedCompressors != "" {
+ compressors = s.clientAdvertisedCompressors + "," + compressors
+ }
+ s.clientAdvertisedCompressors = compressors
case "grpc-encoding":
s.recvCompress = hf.Value
case ":method":
@@ -405,23 +425,23 @@
timeoutSet = true
var err error
if timeout, err = decodeTimeout(hf.Value); err != nil {
- headerError = true
+ headerError = status.Newf(codes.Internal, "malformed grpc-timeout: %v", err)
}
// "Transports must consider requests containing the Connection header
// as malformed." - A41
case "connection":
- if logger.V(logLevel) {
- logger.Errorf("transport: http2Server.operateHeaders parsed a :connection header which makes a request malformed as per the HTTP/2 spec")
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Received a HEADERS frame with a :connection header which makes the request malformed, as per the HTTP/2 spec")
}
- headerError = true
+ protocolError = 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)
+ headerError = status.Newf(codes.Internal, "malformed binary metadata %q in header %q: %v", hf.Value, hf.Name, err)
+ t.logger.Warningf("Failed to decode metadata header (%q, %q): %v", hf.Name, hf.Value, err)
break
}
mdata[hf.Name] = append(mdata[hf.Name], v)
@@ -435,26 +455,47 @@
// error, this takes precedence over a client not speaking gRPC.
if len(mdata[":authority"]) > 1 || len(mdata["host"]) > 1 {
errMsg := fmt.Sprintf("num values of :authority: %v, num values of host: %v, both must only have 1 value as per HTTP/2 spec", len(mdata[":authority"]), len(mdata["host"]))
- if logger.V(logLevel) {
- logger.Errorf("transport: %v", errMsg)
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Aborting the stream early: %v", errMsg)
}
t.controlBuf.put(&earlyAbortStream{
- httpStatus: 400,
+ httpStatus: http.StatusBadRequest,
streamID: streamID,
contentSubtype: s.contentSubtype,
status: status.New(codes.Internal, errMsg),
+ rst: !frame.StreamEnded(),
})
- return false
+ return nil
}
- if !isGRPC || headerError {
+ if protocolError {
t.controlBuf.put(&cleanupStream{
streamID: streamID,
rst: true,
rstCode: http2.ErrCodeProtocol,
onWrite: func() {},
})
- return false
+ return nil
+ }
+ if !isGRPC {
+ t.controlBuf.put(&earlyAbortStream{
+ httpStatus: http.StatusUnsupportedMediaType,
+ streamID: streamID,
+ contentSubtype: s.contentSubtype,
+ status: status.Newf(codes.InvalidArgument, "invalid gRPC request content-type %q", contentType),
+ rst: !frame.StreamEnded(),
+ })
+ return nil
+ }
+ if headerError != nil {
+ t.controlBuf.put(&earlyAbortStream{
+ httpStatus: http.StatusBadRequest,
+ streamID: streamID,
+ contentSubtype: s.contentSubtype,
+ status: headerError,
+ rst: !frame.StreamEnded(),
+ })
+ return nil
}
// "If :authority is missing, Host must be renamed to :authority." - A41
@@ -479,14 +520,7 @@
} else {
s.ctx, s.cancel = context.WithCancel(t.ctx)
}
- pr := &peer.Peer{
- Addr: t.remoteAddr,
- }
- // Attach Auth info if there is any.
- if t.authInfo != nil {
- pr.AuthInfo = t.authInfo
- }
- s.ctx = peer.NewContext(s.ctx, pr)
+
// Attach the received metadata to the context.
if len(mdata) > 0 {
s.ctx = metadata.NewIncomingContext(s.ctx, mdata)
@@ -501,7 +535,7 @@
if t.state != reachable {
t.mu.Unlock()
s.cancel()
- return false
+ return nil
}
if uint32(len(t.activeStreams)) >= t.maxStreams {
t.mu.Unlock()
@@ -512,28 +546,30 @@
onWrite: func() {},
})
s.cancel()
- return false
+ return nil
}
if httpMethod != http.MethodPost {
t.mu.Unlock()
- if logger.V(logLevel) {
- logger.Infof("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", httpMethod)
+ errMsg := fmt.Sprintf("Received a HEADERS frame with :method %q which should be POST", httpMethod)
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Aborting the stream early: %v", errMsg)
}
- t.controlBuf.put(&cleanupStream{
- streamID: streamID,
- rst: true,
- rstCode: http2.ErrCodeProtocol,
- onWrite: func() {},
+ t.controlBuf.put(&earlyAbortStream{
+ httpStatus: 405,
+ streamID: streamID,
+ contentSubtype: s.contentSubtype,
+ status: status.New(codes.Internal, errMsg),
+ rst: !frame.StreamEnded(),
})
s.cancel()
- return false
+ return nil
}
if t.inTapHandle != nil {
var err error
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)
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Aborting the stream early due to InTapHandle failure: %v", err)
}
stat, ok := status.FromError(err)
if !ok {
@@ -544,8 +580,9 @@
streamID: s.id,
contentSubtype: s.contentSubtype,
status: stat,
+ rst: !frame.StreamEnded(),
})
- return false
+ return nil
}
}
t.activeStreams[streamID] = s
@@ -561,17 +598,17 @@
t.adjustWindow(s, uint32(n))
}
s.ctx = traceCtx(s.ctx, s.method)
- if t.stats != nil {
- s.ctx = t.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
+ for _, sh := range t.stats {
+ s.ctx = sh.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
inHeader := &stats.InHeader{
FullMethod: s.method,
RemoteAddr: t.remoteAddr,
LocalAddr: t.localAddr,
Compression: s.recvCompress,
WireLength: int(frame.Header().Length),
- Header: metadata.MD(mdata).Copy(),
+ Header: mdata.Copy(),
}
- t.stats.HandleRPC(s.ctx, inHeader)
+ sh.HandleRPC(s.ctx, inHeader)
}
s.ctxDone = s.ctx.Done()
s.wq = newWriteQuota(defaultWriteQuota, s.ctxDone)
@@ -592,7 +629,7 @@
wq: s.wq,
})
handle(s)
- return false
+ return nil
}
// HandleStreams receives incoming streams using the given handler. This is
@@ -606,8 +643,8 @@
atomic.StoreInt64(&t.lastRead, time.Now().UnixNano())
if err != nil {
if se, ok := err.(http2.StreamError); ok {
- if logger.V(logLevel) {
- logger.Warningf("transport: http2Server.HandleStreams encountered http2.StreamError: %v", se)
+ if t.logger.V(logLevel) {
+ t.logger.Warningf("Encountered http2.StreamError: %v", se)
}
t.mu.Lock()
s := t.activeStreams[se.StreamID]
@@ -625,19 +662,16 @@
continue
}
if err == io.EOF || err == io.ErrUnexpectedEOF {
- t.Close()
+ t.Close(err)
return
}
- if logger.V(logLevel) {
- logger.Warningf("transport: http2Server.HandleStreams failed to read frame: %v", err)
- }
- t.Close()
+ t.Close(err)
return
}
switch frame := frame.(type) {
case *http2.MetaHeadersFrame:
- if t.operateHeaders(frame, handle, traceCtx) {
- t.Close()
+ if err := t.operateHeaders(frame, handle, traceCtx); err != nil {
+ t.Close(err)
break
}
case *http2.DataFrame:
@@ -653,8 +687,8 @@
case *http2.GoAwayFrame:
// TODO: Handle GoAway from the client appropriately.
default:
- if logger.V(logLevel) {
- logger.Errorf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame)
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Received unsupported frame type %T", frame)
}
}
}
@@ -838,8 +872,8 @@
func (t *http2Server) handlePing(f *http2.PingFrame) {
if f.IsAck() {
- if f.Data == goAwayPing.data && t.drainChan != nil {
- close(t.drainChan)
+ if f.Data == goAwayPing.data && t.drainEvent != nil {
+ t.drainEvent.Fire()
return
}
// Maybe it's a BDP ping.
@@ -881,10 +915,7 @@
if t.pingStrikes > maxPingStrikes {
// Send goaway and close the connection.
- if logger.V(logLevel) {
- logger.Errorf("transport: Got too many pings from the client, closing the connection.")
- }
- t.controlBuf.put(&goAway{code: http2.ErrCodeEnhanceYourCalm, debugData: []byte("too_many_pings"), closeConn: true})
+ t.controlBuf.put(&goAway{code: http2.ErrCodeEnhanceYourCalm, debugData: []byte("too_many_pings"), closeConn: errors.New("got too many pings from the client")})
}
}
@@ -916,8 +947,8 @@
var sz int64
for _, f := range hdrFrame.hf {
if sz += int64(f.Size()); sz > int64(*t.maxSendHeaderListSize) {
- if logger.V(logLevel) {
- logger.Errorf("header list size to send violates the maximum size (%d bytes) set by client", *t.maxSendHeaderListSize)
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Header list size to send violates the maximum size (%d bytes) set by client", *t.maxSendHeaderListSize)
}
return false
}
@@ -925,12 +956,27 @@
return true
}
+func (t *http2Server) streamContextErr(s *Stream) error {
+ select {
+ case <-t.done:
+ return ErrConnClosing
+ default:
+ }
+ return ContextErr(s.ctx.Err())
+}
+
// WriteHeader sends the header metadata md back to the client.
func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error {
- if s.updateHeaderSent() || s.getState() == streamDone {
+ s.hdrMu.Lock()
+ defer s.hdrMu.Unlock()
+ if s.getState() == streamDone {
+ return t.streamContextErr(s)
+ }
+
+ if s.updateHeaderSent() {
return ErrIllegalHeaderWrite
}
- s.hdrMu.Lock()
+
if md.Len() > 0 {
if s.header.Len() > 0 {
s.header = metadata.Join(s.header, md)
@@ -939,10 +985,8 @@
}
}
if err := t.writeHeaderLocked(s); err != nil {
- s.hdrMu.Unlock()
- return err
+ return status.Convert(err).Err()
}
- s.hdrMu.Unlock()
return nil
}
@@ -973,14 +1017,14 @@
t.closeStream(s, true, http2.ErrCodeInternal, false)
return ErrHeaderListSizeLimitViolation
}
- if t.stats != nil {
+ for _, sh := range t.stats {
// Note: Headers are compressed with hpack after this call returns.
// No WireLength field is set here.
outHeader := &stats.OutHeader{
Header: s.header.Copy(),
Compression: s.sendCompress,
}
- t.stats.HandleRPC(s.Context(), outHeader)
+ sh.HandleRPC(s.Context(), outHeader)
}
return nil
}
@@ -990,17 +1034,19 @@
// TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early
// OK is adopted.
func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error {
+ s.hdrMu.Lock()
+ defer s.hdrMu.Unlock()
+
if s.getState() == streamDone {
return nil
}
- s.hdrMu.Lock()
+
// TODO(mmukhi): Benchmark if the performance gets better if count the metadata and other header fields
// first and create a slice of that exact size.
headerFields := make([]hpack.HeaderField, 0, 2) // grpc-status and grpc-message will be there if none else.
if !s.updateHeaderSent() { // No headers have been sent.
if len(s.header) > 0 { // Send a separate header frame.
if err := t.writeHeaderLocked(s); err != nil {
- s.hdrMu.Unlock()
return err
}
} else { // Send a trailer only response.
@@ -1015,7 +1061,7 @@
stBytes, err := proto.Marshal(p)
if err != nil {
// TODO: return error instead, when callers are able to handle it.
- logger.Errorf("transport: failed to marshal rpc status: %v, error: %v", p, err)
+ t.logger.Errorf("Failed to marshal rpc status: %s, error: %v", pretty.ToJSON(p), err)
} else {
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-status-details-bin", Value: encodeBinHeader(stBytes)})
}
@@ -1029,7 +1075,7 @@
endStream: true,
onWrite: t.setResetPingStrikes,
}
- s.hdrMu.Unlock()
+
success, err := t.controlBuf.execute(t.checkForHeaderListSize, trailingHeader)
if !success {
if err != nil {
@@ -1041,10 +1087,10 @@
// Send a RST_STREAM after the trailers if the client has not already half-closed.
rst := s.getState() == streamActive
t.finishStream(s, rst, http2.ErrCodeNo, trailingHeader, true)
- if t.stats != nil {
+ for _, sh := range t.stats {
// Note: The trailer fields are compressed with hpack after this call returns.
// No WireLength field is set here.
- t.stats.HandleRPC(s.Context(), &stats.OutTrailer{
+ sh.HandleRPC(s.Context(), &stats.OutTrailer{
Trailer: s.trailer.Copy(),
})
}
@@ -1056,23 +1102,12 @@
func (t *http2Server) Write(s *Stream, hdr []byte, data []byte, opts *Options) error {
if !s.isHeaderSent() { // Headers haven't been written yet.
if err := t.WriteHeader(s, nil); err != nil {
- if _, ok := err.(ConnectionError); ok {
- return err
- }
- // TODO(mmukhi, dfawley): Make sure this is the right code to return.
- return status.Errorf(codes.Internal, "transport: %v", err)
+ return err
}
} else {
// Writing headers checks for this condition.
if s.getState() == streamDone {
- // TODO(mmukhi, dfawley): Should the server write also return io.EOF?
- s.cancel()
- select {
- case <-t.done:
- return ErrConnClosing
- default:
- }
- return ContextErr(s.ctx.Err())
+ return t.streamContextErr(s)
}
}
df := &dataFrame{
@@ -1082,12 +1117,7 @@
onEachWrite: t.setResetPingStrikes,
}
if err := s.wq.get(int32(len(hdr) + len(data))); err != nil {
- select {
- case <-t.done:
- return ErrConnClosing
- default:
- }
- return ContextErr(s.ctx.Err())
+ return t.streamContextErr(s)
}
return t.controlBuf.put(df)
}
@@ -1136,20 +1166,20 @@
if val <= 0 {
// The connection has been idle for a duration of keepalive.MaxConnectionIdle or more.
// Gracefully close the connection.
- t.Drain()
+ t.Drain("max_idle")
return
}
idleTimer.Reset(val)
case <-ageTimer.C:
- t.Drain()
+ t.Drain("max_age")
ageTimer.Reset(t.kp.MaxConnectionAgeGrace)
select {
case <-ageTimer.C:
// Close the connection after grace period.
- if logger.V(logLevel) {
- logger.Infof("transport: closing server transport due to maximum connection age.")
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Closing server transport due to maximum connection age")
}
- t.Close()
+ t.controlBuf.put(closeConnection{})
case <-t.done:
}
return
@@ -1165,10 +1195,7 @@
continue
}
if outstandingPing && kpTimeoutLeft <= 0 {
- if logger.V(logLevel) {
- logger.Infof("transport: closing server transport due to idleness.")
- }
- t.Close()
+ t.Close(fmt.Errorf("keepalive ping not acked within timeout %s", t.kp.Time))
return
}
if !outstandingPing {
@@ -1195,40 +1222,37 @@
// 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() {
+func (t *http2Server) Close(err error) {
t.mu.Lock()
if t.state == closing {
t.mu.Unlock()
return
}
+ if t.logger.V(logLevel) {
+ t.logger.Infof("Closing: %v", err)
+ }
t.state = closing
streams := t.activeStreams
t.activeStreams = nil
t.mu.Unlock()
t.controlBuf.finish()
close(t.done)
- if err := t.conn.Close(); err != nil && logger.V(logLevel) {
- logger.Infof("transport: error closing conn during Close: %v", err)
+ if err := t.conn.Close(); err != nil && t.logger.V(logLevel) {
+ t.logger.Infof("Error closing underlying net.Conn during Close: %v", err)
}
- if channelz.IsOn() {
- channelz.RemoveEntry(t.channelzID)
- }
+ channelz.RemoveEntry(t.channelzID)
// Cancel all active streams.
for _, s := range streams {
s.cancel()
}
- if t.stats != nil {
+ for _, sh := range t.stats {
connEnd := &stats.ConnEnd{}
- t.stats.HandleConn(t.ctx, connEnd)
+ sh.HandleConn(t.ctx, connEnd)
}
}
// deleteStream deletes the stream s from transport's active streams.
func (t *http2Server) deleteStream(s *Stream, eosReceived bool) {
- // In case stream sending and receiving are invoked in separate
- // goroutines (e.g., bi-directional streaming), cancel needs to be
- // called to interrupt the potential blocking on other goroutines.
- s.cancel()
t.mu.Lock()
if _, ok := t.activeStreams[s.id]; ok {
@@ -1250,6 +1274,11 @@
// finishStream closes the stream and puts the trailing headerFrame into controlbuf.
func (t *http2Server) finishStream(s *Stream, rst bool, rstCode http2.ErrCode, hdr *headerFrame, eosReceived bool) {
+ // In case stream sending and receiving are invoked in separate
+ // goroutines (e.g., bi-directional streaming), cancel needs to be
+ // called to interrupt the potential blocking on other goroutines.
+ s.cancel()
+
oldState := s.swapState(streamDone)
if oldState == streamDone {
// If the stream was already done, return.
@@ -1269,6 +1298,11 @@
// closeStream clears the footprint of a stream when the stream is not needed any more.
func (t *http2Server) closeStream(s *Stream, rst bool, rstCode http2.ErrCode, eosReceived bool) {
+ // In case stream sending and receiving are invoked in separate
+ // goroutines (e.g., bi-directional streaming), cancel needs to be
+ // called to interrupt the potential blocking on other goroutines.
+ s.cancel()
+
s.swapState(streamDone)
t.deleteStream(s, eosReceived)
@@ -1284,14 +1318,14 @@
return t.remoteAddr
}
-func (t *http2Server) Drain() {
+func (t *http2Server) Drain(debugData string) {
t.mu.Lock()
defer t.mu.Unlock()
- if t.drainChan != nil {
+ if t.drainEvent != nil {
return
}
- t.drainChan = make(chan struct{})
- t.controlBuf.put(&goAway{code: http2.ErrCodeNo, debugData: []byte{}, headsUp: true})
+ t.drainEvent = grpcsync.NewEvent()
+ t.controlBuf.put(&goAway{code: http2.ErrCodeNo, debugData: []byte(debugData), headsUp: true})
}
var goAwayPing = &ping{data: [8]byte{1, 6, 1, 8, 0, 3, 3, 9}}
@@ -1311,19 +1345,17 @@
// Stop accepting more streams now.
t.state = draining
sid := t.maxStreamID
+ retErr := g.closeConn
if len(t.activeStreams) == 0 {
- g.closeConn = true
+ retErr = errors.New("second GOAWAY written and no active streams left to process")
}
t.mu.Unlock()
t.maxStreamMu.Unlock()
if err := t.framer.fr.WriteGoAway(sid, g.code, g.debugData); err != nil {
return false, err
}
- if g.closeConn {
- // Abruptly close the connection following the GoAway (via
- // loopywriter). But flush out what's inside the buffer first.
- t.framer.writer.Flush()
- return false, fmt.Errorf("transport: Connection closing")
+ if retErr != nil {
+ return false, retErr
}
return true, nil
}
@@ -1335,7 +1367,7 @@
// originated before the GoAway reaches the client.
// After getting the ack or timer expiration send out another GoAway this
// time with an ID of the max stream server intends to process.
- if err := t.framer.fr.WriteGoAway(math.MaxUint32, http2.ErrCodeNo, []byte{}); err != nil {
+ if err := t.framer.fr.WriteGoAway(math.MaxUint32, http2.ErrCodeNo, g.debugData); err != nil {
return false, err
}
if err := t.framer.fr.WritePing(false, goAwayPing.data); err != nil {
@@ -1345,7 +1377,7 @@
timer := time.NewTimer(time.Minute)
defer timer.Stop()
select {
- case <-t.drainChan:
+ case <-t.drainEvent.Done():
case <-timer.C:
case <-t.done:
return
@@ -1404,6 +1436,13 @@
}
}
+func (t *http2Server) getPeer() *peer.Peer {
+ return &peer.Peer{
+ Addr: t.remoteAddr,
+ AuthInfo: t.authInfo, // Can be nil
+ }
+}
+
func getJitter(v time.Duration) time.Duration {
if v == infinity {
return 0
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 d8247bc..19cbb18 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http_util.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http_util.go
@@ -20,8 +20,8 @@
import (
"bufio"
- "bytes"
"encoding/base64"
+ "errors"
"fmt"
"io"
"math"
@@ -38,21 +38,14 @@
"golang.org/x/net/http2/hpack"
spb "google.golang.org/genproto/googleapis/rpc/status"
"google.golang.org/grpc/codes"
- "google.golang.org/grpc/grpclog"
"google.golang.org/grpc/status"
)
const (
// http2MaxFrameLen specifies the max length of a HTTP2 frame.
http2MaxFrameLen = 16384 // 16KB frame
- // http://http2.github.io/http2-spec/#SettingValues
+ // https://httpwg.org/specs/rfc7540.html#SettingValues
http2InitHeaderTableSize = 4096
- // baseContentType is the base content-type for gRPC. This is a valid
- // content-type on it's own, but can also include a content-subtype such as
- // "proto" as a suffix after "+" or ";". See
- // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
- // for more details.
-
)
var (
@@ -92,7 +85,6 @@
// 504 Gateway timeout - UNAVAILABLE.
http.StatusGatewayTimeout: codes.Unavailable,
}
- logger = grpclog.Component("transport")
)
// isReservedHeader checks whether hdr belongs to HTTP2 headers
@@ -257,13 +249,13 @@
}
func encodeGrpcMessageUnchecked(msg string) string {
- var buf bytes.Buffer
+ var sb strings.Builder
for len(msg) > 0 {
r, size := utf8.DecodeRuneInString(msg)
for _, b := range []byte(string(r)) {
if size > 1 {
// If size > 1, r is not ascii. Always do percent encoding.
- buf.WriteString(fmt.Sprintf("%%%02X", b))
+ fmt.Fprintf(&sb, "%%%02X", b)
continue
}
@@ -272,14 +264,14 @@
//
// fmt.Sprintf("%%%02X", utf8.RuneError) gives "%FFFD".
if b >= spaceByte && b <= tildeByte && b != percentByte {
- buf.WriteByte(b)
+ sb.WriteByte(b)
} else {
- buf.WriteString(fmt.Sprintf("%%%02X", b))
+ fmt.Fprintf(&sb, "%%%02X", b)
}
}
msg = msg[size:]
}
- return buf.String()
+ return sb.String()
}
// decodeGrpcMessage decodes the msg encoded by encodeGrpcMessage.
@@ -297,23 +289,23 @@
}
func decodeGrpcMessageUnchecked(msg string) string {
- var buf bytes.Buffer
+ var sb strings.Builder
lenMsg := len(msg)
for i := 0; i < lenMsg; i++ {
c := msg[i]
if c == percentByte && i+2 < lenMsg {
parsed, err := strconv.ParseUint(msg[i+1:i+3], 16, 8)
if err != nil {
- buf.WriteByte(c)
+ sb.WriteByte(c)
} else {
- buf.WriteByte(byte(parsed))
+ sb.WriteByte(byte(parsed))
i += 2
}
} else {
- buf.WriteByte(c)
+ sb.WriteByte(c)
}
}
- return buf.String()
+ return sb.String()
}
type bufWriter struct {
@@ -322,8 +314,6 @@
batchSize int
conn net.Conn
err error
-
- onFlush func()
}
func newBufWriter(conn net.Conn, batchSize int) *bufWriter {
@@ -339,7 +329,8 @@
return 0, w.err
}
if w.batchSize == 0 { // Buffer has been disabled.
- return w.conn.Write(b)
+ n, err = w.conn.Write(b)
+ return n, toIOError(err)
}
for len(b) > 0 {
nn := copy(w.buf[w.offset:], b)
@@ -360,14 +351,31 @@
if w.offset == 0 {
return nil
}
- if w.onFlush != nil {
- w.onFlush()
- }
_, w.err = w.conn.Write(w.buf[:w.offset])
+ w.err = toIOError(w.err)
w.offset = 0
return w.err
}
+type ioError struct {
+ error
+}
+
+func (i ioError) Unwrap() error {
+ return i.error
+}
+
+func isIOError(err error) bool {
+ return errors.As(err, &ioError{})
+}
+
+func toIOError(err error) error {
+ if err == nil {
+ return nil
+ }
+ return ioError{error: err}
+}
+
type framer struct {
writer *bufWriter
fr *http2.Framer
diff --git a/vendor/google.golang.org/grpc/internal/transport/logging.go b/vendor/google.golang.org/grpc/internal/transport/logging.go
new file mode 100644
index 0000000..42ed2b0
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/transport/logging.go
@@ -0,0 +1,40 @@
+/*
+ *
+ * Copyright 2023 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 transport
+
+import (
+ "fmt"
+
+ "google.golang.org/grpc/grpclog"
+ internalgrpclog "google.golang.org/grpc/internal/grpclog"
+)
+
+var logger = grpclog.Component("transport")
+
+func prefixLoggerForServerTransport(p *http2Server) *internalgrpclog.PrefixLogger {
+ return internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf("[server-transport %p] ", p))
+}
+
+func prefixLoggerForServerHandlerTransport(p *serverHandlerTransport) *internalgrpclog.PrefixLogger {
+ return internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf("[server-handler-transport %p] ", p))
+}
+
+func prefixLoggerForClientTransport(p *http2Client) *internalgrpclog.PrefixLogger {
+ return internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf("[client-transport %p] ", p))
+}
diff --git a/vendor/google.golang.org/grpc/internal/transport/transport.go b/vendor/google.golang.org/grpc/internal/transport/transport.go
index d3bf65b..aa1c896 100644
--- a/vendor/google.golang.org/grpc/internal/transport/transport.go
+++ b/vendor/google.golang.org/grpc/internal/transport/transport.go
@@ -34,6 +34,7 @@
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/internal/channelz"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/resolver"
@@ -42,6 +43,10 @@
"google.golang.org/grpc/tap"
)
+// ErrNoHeaders is used as a signal that a trailers only response was received,
+// and is not a real error.
+var ErrNoHeaders = errors.New("stream has no headers")
+
const logLevel = 2
type bufferPool struct {
@@ -252,6 +257,9 @@
fc *inFlow
wq *writeQuota
+ // Holds compressor names passed in grpc-accept-encoding metadata from the
+ // client. This is empty for the client side stream.
+ clientAdvertisedCompressors string
// Callback to state application's intentions to read data. This
// is used to adjust flow control, if needed.
requestRead func(int)
@@ -340,8 +348,24 @@
}
// SetSendCompress sets the compression algorithm to the stream.
-func (s *Stream) SetSendCompress(str string) {
- s.sendCompress = str
+func (s *Stream) SetSendCompress(name string) error {
+ if s.isHeaderSent() || s.getState() == streamDone {
+ return errors.New("transport: set send compressor called after headers sent or stream done")
+ }
+
+ s.sendCompress = name
+ return nil
+}
+
+// SendCompress returns the send compressor name.
+func (s *Stream) SendCompress() string {
+ return s.sendCompress
+}
+
+// ClientAdvertisedCompressors returns the compressor names advertised by the
+// client via grpc-accept-encoding header.
+func (s *Stream) ClientAdvertisedCompressors() string {
+ return s.clientAdvertisedCompressors
}
// Done returns a channel which is closed when it receives the final status
@@ -365,9 +389,15 @@
return s.header.Copy(), nil
}
s.waitOnHeader()
+
if !s.headerValid {
return nil, s.status.Err()
}
+
+ if s.noHeaders {
+ return nil, ErrNoHeaders
+ }
+
return s.header.Copy(), nil
}
@@ -522,14 +552,14 @@
ConnectionTimeout time.Duration
Credentials credentials.TransportCredentials
InTapHandle tap.ServerInHandle
- StatsHandler stats.Handler
+ StatsHandlers []stats.Handler
KeepaliveParams keepalive.ServerParameters
KeepalivePolicy keepalive.EnforcementPolicy
InitialWindowSize int32
InitialConnWindowSize int32
WriteBufferSize int
ReadBufferSize int
- ChannelzParentID int64
+ ChannelzParentID *channelz.Identifier
MaxHeaderListSize *uint32
HeaderTableSize *uint32
}
@@ -552,8 +582,8 @@
CredsBundle credentials.Bundle
// KeepaliveParams stores the keepalive parameters.
KeepaliveParams keepalive.ClientParameters
- // StatsHandler stores the handler for stats.
- StatsHandler stats.Handler
+ // StatsHandlers stores the handler for stats.
+ StatsHandlers []stats.Handler
// InitialWindowSize sets the initial window size for a stream.
InitialWindowSize int32
// InitialConnWindowSize sets the initial window size for a connection.
@@ -563,7 +593,7 @@
// ReadBufferSize sets the size of read buffer, which in turn determines how much data can be read at most for one read syscall.
ReadBufferSize int
// ChannelzParentID sets the addrConn id which initiate the creation of this client transport.
- ChannelzParentID int64
+ ChannelzParentID *channelz.Identifier
// MaxHeaderListSize sets the max (uncompressed) size of header list that is prepared to be received.
MaxHeaderListSize *uint32
// UseProxy specifies if a proxy should be used.
@@ -572,8 +602,8 @@
// NewClientTransport establishes the transport with the required ConnectOptions
// and returns it to the caller.
-func NewClientTransport(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (ClientTransport, error) {
- return newHTTP2Client(connectCtx, ctx, addr, opts, onPrefaceReceipt, onGoAway, onClose)
+func NewClientTransport(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onClose func(GoAwayReason)) (ClientTransport, error) {
+ return newHTTP2Client(connectCtx, ctx, addr, opts, onClose)
}
// Options provides additional hints and information for message
@@ -690,13 +720,13 @@
// 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()
+ Close(err error)
// RemoteAddr returns the remote network address.
RemoteAddr() net.Addr
// Drain notifies the client this ServerTransport stops accepting new RPCs.
- Drain()
+ Drain(debugData string)
// IncrMsgSent increments the number of message sent through this transport.
IncrMsgSent()
@@ -741,6 +771,12 @@
return e.err
}
+// Unwrap returns the original error of this connection error or nil when the
+// origin is nil.
+func (e ConnectionError) Unwrap() error {
+ return e.err
+}
+
var (
// ErrConnClosing indicates that the transport is closing.
ErrConnClosing = connectionErrorf(true, nil, "transport is closing")