[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/picker_wrapper.go b/vendor/google.golang.org/grpc/picker_wrapper.go
index e8367cb..02f9759 100644
--- a/vendor/google.golang.org/grpc/picker_wrapper.go
+++ b/vendor/google.golang.org/grpc/picker_wrapper.go
@@ -26,6 +26,7 @@
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/internal/channelz"
+ istatus "google.golang.org/grpc/internal/status"
"google.golang.org/grpc/internal/transport"
"google.golang.org/grpc/status"
)
@@ -35,6 +36,7 @@
type pickerWrapper struct {
mu sync.Mutex
done bool
+ idle bool
blockingCh chan struct{}
picker balancer.Picker
}
@@ -46,7 +48,11 @@
// updatePicker is called by UpdateBalancerState. It unblocks all blocked pick.
func (pw *pickerWrapper) updatePicker(p balancer.Picker) {
pw.mu.Lock()
- if pw.done {
+ if pw.done || pw.idle {
+ // There is a small window where a picker update from the LB policy can
+ // race with the channel going to idle mode. If the picker is idle here,
+ // it is because the channel asked it to do so, and therefore it is sage
+ // to ignore the update from the LB policy.
pw.mu.Unlock()
return
}
@@ -57,12 +63,16 @@
pw.mu.Unlock()
}
-func doneChannelzWrapper(acw *acBalancerWrapper, done func(balancer.DoneInfo)) func(balancer.DoneInfo) {
- acw.mu.Lock()
- ac := acw.ac
- acw.mu.Unlock()
+// doneChannelzWrapper performs the following:
+// - increments the calls started channelz counter
+// - wraps the done function in the passed in result to increment the calls
+// failed or calls succeeded channelz counter before invoking the actual
+// done function.
+func doneChannelzWrapper(acbw *acBalancerWrapper, result *balancer.PickResult) {
+ ac := acbw.ac
ac.incrCallsStarted()
- return func(b balancer.DoneInfo) {
+ done := result.Done
+ result.Done = func(b balancer.DoneInfo) {
if b.Err != nil && b.Err != io.EOF {
ac.incrCallsFailed()
} else {
@@ -81,7 +91,7 @@
// - the current picker returns other errors and failfast is false.
// - the subConn returned by the current picker is not READY
// When one of these situations happens, pick blocks until the picker gets updated.
-func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.PickInfo) (transport.ClientTransport, func(balancer.DoneInfo), error) {
+func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.PickInfo) (transport.ClientTransport, balancer.PickResult, error) {
var ch chan struct{}
var lastPickErr error
@@ -89,7 +99,7 @@
pw.mu.Lock()
if pw.done {
pw.mu.Unlock()
- return nil, nil, ErrClientConnClosing
+ return nil, balancer.PickResult{}, ErrClientConnClosing
}
if pw.picker == nil {
@@ -110,9 +120,9 @@
}
switch ctx.Err() {
case context.DeadlineExceeded:
- return nil, nil, status.Error(codes.DeadlineExceeded, errStr)
+ return nil, balancer.PickResult{}, status.Error(codes.DeadlineExceeded, errStr)
case context.Canceled:
- return nil, nil, status.Error(codes.Canceled, errStr)
+ return nil, balancer.PickResult{}, status.Error(codes.Canceled, errStr)
}
case <-ch:
}
@@ -124,14 +134,17 @@
pw.mu.Unlock()
pickResult, err := p.Pick(info)
-
if err != nil {
if err == balancer.ErrNoSubConnAvailable {
continue
}
- if _, ok := status.FromError(err); ok {
+ if st, ok := status.FromError(err); ok {
// Status error: end the RPC unconditionally with this status.
- return nil, nil, err
+ // First restrict the code to the list allowed by gRFC A54.
+ if istatus.IsRestrictedControlPlaneCode(st) {
+ err = status.Errorf(codes.Internal, "received picker error with illegal status: %v", err)
+ }
+ return nil, balancer.PickResult{}, dropError{error: err}
}
// For all other errors, wait for ready RPCs should block and other
// RPCs should fail with unavailable.
@@ -139,19 +152,20 @@
lastPickErr = err
continue
}
- return nil, nil, status.Error(codes.Unavailable, err.Error())
+ return nil, balancer.PickResult{}, status.Error(codes.Unavailable, err.Error())
}
- acw, ok := pickResult.SubConn.(*acBalancerWrapper)
+ acbw, ok := pickResult.SubConn.(*acBalancerWrapper)
if !ok {
logger.Errorf("subconn returned from pick is type %T, not *acBalancerWrapper", pickResult.SubConn)
continue
}
- if t := acw.getAddrConn().getReadyTransport(); t != nil {
+ if t := acbw.ac.getReadyTransport(); t != nil {
if channelz.IsOn() {
- return t, doneChannelzWrapper(acw, pickResult.Done), nil
+ doneChannelzWrapper(acbw, &pickResult)
+ return t, pickResult, nil
}
- return t, pickResult.Done, nil
+ return t, pickResult, nil
}
if pickResult.Done != nil {
// Calling done with nil error, no bytes sent and no bytes received.
@@ -175,3 +189,28 @@
pw.done = true
close(pw.blockingCh)
}
+
+func (pw *pickerWrapper) enterIdleMode() {
+ pw.mu.Lock()
+ defer pw.mu.Unlock()
+ if pw.done {
+ return
+ }
+ pw.idle = true
+}
+
+func (pw *pickerWrapper) exitIdleMode() {
+ pw.mu.Lock()
+ defer pw.mu.Unlock()
+ if pw.done {
+ return
+ }
+ pw.blockingCh = make(chan struct{})
+ pw.idle = false
+}
+
+// dropError is a wrapper error that indicates the LB policy wishes to drop the
+// RPC and not retry it.
+type dropError struct {
+ error
+}