opensource/grpc-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

rls: Implementation of the picker (#3423)

Browse Source
This commit is contained in:
Easwar Swaminathan
2020-03-16 10:43:00 -07:00
committed by GitHub
parent 399ae78064
commit 804ff443fc
4 changed files with 767 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
balancer/rls/internal/cache/cache.go vendored
View File

@ -25,6 +25,7 @@ import (
"sync"
"time"
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/grpclog"
"google.golang.org/grpc/internal/backoff"
)
@ -83,8 +84,11 @@ type Entry struct {
// HeaderData is received in an RLS response and is to be sent in the
// X-Google-RLS-Data header for matching RPCs.
HeaderData string
// TODO(easwars): Add support to store the ChildPolicy here. Need a
// balancerWrapper type to be implemented for this.
// ChildPicker is a very thin wrapper around the child policy wrapper.
// The type is declared as a V2Picker interface since the users of
// the cache only care about the picker provided by the child policy, and
// this makes it easy for testing.
ChildPicker balancer.V2Picker
// size stores the size of this cache entry. Uses only a subset of the
// fields. See `entrySize` for this is computed.

1
balancer/rls/internal/client_test.go
View File

@ -130,7 +130,6 @@ func TestLookupSuccess(t *testing.T) {
defer cleanup()
const (
defaultTestTimeout = 1 * time.Second
rlsReqPath = "/service/method"
rlsRespTarget = "us_east_1.firestore.googleapis.com"
rlsHeaderData = "headerData"

173
balancer/rls/internal/picker.go Normal file
View File

@ -0,0 +1,173 @@
// +build go1.10
/*
*
* Copyright 2020 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package rls
import (
"errors"
"time"
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/balancer/rls/internal/cache"
"google.golang.org/grpc/balancer/rls/internal/keys"
rlspb "google.golang.org/grpc/balancer/rls/internal/proto/grpc_lookup_v1"
"google.golang.org/grpc/metadata"
)
var errRLSThrottled = balancer.TransientFailureError(errors.New("RLS call throttled at client side"))
// Compile time assert to ensure we implement V2Picker.
var _ balancer.V2Picker = (*picker)(nil)
// RLS picker selects the subConn to be used for a particular RPC. It does not
// manage subConns directly and usually deletegates to pickers provided by
// child policies.
//
// The RLS LB policy creates a new picker object whenever its ServiceConfig is
// updated and provides a bunch of hooks for the picker to get the latest state
// that it can used to make its decision.
type picker struct {
// The keyBuilder map used to generate RLS keys for the RPC. This is built
// by the LB policy based on the received ServiceConfig.
kbm keys.BuilderMap
// This is the request processing strategy as indicated by the LB policy's
// ServiceConfig. This controls how to process a RPC when the data required
// to make the pick decision is not in the cache.
strategy rlspb.RouteLookupConfig_RequestProcessingStrategy
// The following hooks are setup by the LB policy to enable the picker to
// access state stored in the policy. This approach has the following
// advantages:
// 1. The picker is loosely coupled with the LB policy in the sense that
// updates happening on the LB policy like the receipt of an RLS
// response, or an update to the default picker etc are not explicitly
// pushed to the picker, but are readily available to the picker when it
// invokes these hooks. And the LB policy takes care of synchronizing
// access to these shared state.
// 2. It makes unit testing the picker easy since any number of these hooks
// could be overridden.
// readCache is used to read from the data cache and the pending request
// map in an atomic fashion. The first return parameter is the entry in the
// data cache, and the second indicates whether an entry for the same key
// is present in the pending cache.
readCache func(cache.Key) (*cache.Entry, bool)
// shouldThrottle decides if the current RPC should be throttled at the
// client side. It uses an adaptive throttling algorithm.
shouldThrottle func() bool
// startRLS kicks off an RLS request in the background for the provided RPC
// path and keyMap. An entry in the pending request map is created before
// sending out the request and an entry in the data cache is created or
// updated upon receipt of a response. See implementation in the LB policy
// for details.
startRLS func(string, keys.KeyMap)
// defaultPick enables the picker to delegate the pick decision to the
// picker returned by the child LB policy pointing to the default target
// specified in the service config.
defaultPick func(balancer.PickInfo) (balancer.PickResult, error)
}
// This helper function decides if the pick should delegate to the default
// picker based on the request processing strategy. This is used when the data
// cache does not have a valid entry for the current RPC and the RLS request is
// throttled, or if the current data cache entry is in backoff.
func (p *picker) shouldDelegateToDefault() bool {
return p.strategy == rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR ||
p.strategy == rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS
}
// Pick makes the routing decision for every outbound RPC.
func (p *picker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
// For every incoming request, we first build the RLS keys using the
// keyBuilder we received from the LB policy. If no metadata is present in
// the context, we end up using an empty key.
km := keys.KeyMap{}
md, ok := metadata.FromOutgoingContext(info.Ctx)
if ok {
km = p.kbm.RLSKey(md, info.FullMethodName)
}
// We use the LB policy hook to read the data cache and the pending request
// map (whether or not an entry exists) for the RPC path and the generated
// RLS keys. We will end up kicking off an RLS request only if there is no
// pending request for the current RPC path and keys, and either we didn't
// find an entry in the data cache or the entry was stale and it wasn't in
// backoff.
startRequest := false
now := time.Now()
entry, pending := p.readCache(cache.Key{Path: info.FullMethodName, KeyMap: km.Str})
if entry == nil {
startRequest = true
} else {
entry.Mu.Lock()
defer entry.Mu.Unlock()
if entry.StaleTime.Before(now) && entry.BackoffTime.Before(now) {
// This is the proactive cache refresh.
startRequest = true
}
}
if startRequest && !pending {
if p.shouldThrottle() {
// The entry doesn't exist or has expired and the new RLS request
// has been throttled. Treat it as an error and delegate to default
// pick or fail the pick, based on the request processing strategy.
if entry == nil || entry.ExpiryTime.Before(now) {
if p.shouldDelegateToDefault() {
return p.defaultPick(info)
}
return balancer.PickResult{}, errRLSThrottled
}
// The proactive refresh has been throttled. Nothing to worry, just
// keep using the existing entry.
} else {
p.startRLS(info.FullMethodName, km)
}
}
if entry != nil {
if entry.ExpiryTime.After(now) {
// This is the jolly good case where we have found a valid entry in
// the data cache. We delegate to the LB policy associated with
// this cache entry.
return entry.ChildPicker.Pick(info)
} else if entry.BackoffTime.After(now) {
// The entry has expired, but is in backoff. We either delegate to
// the default picker or return the error from the last failed RLS
// request for this entry.
if p.shouldDelegateToDefault() {
return p.defaultPick(info)
}
return balancer.PickResult{}, entry.CallStatus
}
}
// Either we didn't find an entry or found an entry which had expired and
// was not in backoff (which is also essentially equivalent to not finding
// an entry), and we started an RLS request in the background. We either
// queue the pick or delegate to the default pick. In the former case, upon
// receipt of an RLS response, the LB policy will send a new picker to the
// channel, and the pick will be retried.
if p.strategy == rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR ||
p.strategy == rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR {
return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
}
return p.defaultPick(info)
}

585
balancer/rls/internal/picker_test.go Normal file
View File

@ -0,0 +1,585 @@
// +build go1.10
/*
*
* Copyright 2020 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package rls
import (
"context"
"errors"
"fmt"
"math"
"testing"
"time"
"google.golang.org/grpc/internal/grpcrand"
"github.com/google/go-cmp/cmp"
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/balancer/rls/internal/cache"
"google.golang.org/grpc/balancer/rls/internal/keys"
rlspb "google.golang.org/grpc/balancer/rls/internal/proto/grpc_lookup_v1"
"google.golang.org/grpc/metadata"
)
const defaultTestMaxAge = 5 * time.Second
func initKeyBuilderMap() (keys.BuilderMap, error) {
kb1 := &rlspb.GrpcKeyBuilder{
Names: []*rlspb.GrpcKeyBuilder_Name{{Service: "gFoo"}},
Headers: []*rlspb.NameMatcher{{Key: "k1", Names: []string{"n1"}}},
}
kb2 := &rlspb.GrpcKeyBuilder{
Names: []*rlspb.GrpcKeyBuilder_Name{{Service: "gBar", Method: "method1"}},
Headers: []*rlspb.NameMatcher{{Key: "k2", Names: []string{"n21", "n22"}}},
}
kb3 := &rlspb.GrpcKeyBuilder{
Names: []*rlspb.GrpcKeyBuilder_Name{{Service: "gFoobar"}},
Headers: []*rlspb.NameMatcher{{Key: "k3", Names: []string{"n3"}}},
}
return keys.MakeBuilderMap(&rlspb.RouteLookupConfig{
GrpcKeybuilders: []*rlspb.GrpcKeyBuilder{kb1, kb2, kb3},
})
}
// fakeSubConn embeds the balancer.SubConn interface and contains an id which
// helps verify that the expected subConn was returned by the picker.
type fakeSubConn struct {
balancer.SubConn
id int
}
// fakeChildPicker sends a PickResult with a fakeSubConn with the configured id.
type fakeChildPicker struct {
id int
}
func (p *fakeChildPicker) Pick(_ balancer.PickInfo) (balancer.PickResult, error) {
return balancer.PickResult{SubConn: &fakeSubConn{id: p.id}}, nil
}
// TestPickKeyBuilder verifies the different possible scenarios for
// forming an RLS key for an incoming RPC.
func TestPickKeyBuilder(t *testing.T) {
kbm, err := initKeyBuilderMap()
if err != nil {
t.Fatalf("Failed to create keyBuilderMap: %v", err)
}
tests := []struct {
desc string
rpcPath string
md metadata.MD
wantKey cache.Key
}{
{
desc: "non existent service in keyBuilder map",
rpcPath: "/gNonExistentService/method",
md: metadata.New(map[string]string{"n1": "v1", "n3": "v3"}),
wantKey: cache.Key{Path: "/gNonExistentService/method", KeyMap: ""},
},
{
desc: "no metadata in incoming context",
rpcPath: "/gFoo/method",
md: metadata.MD{},
wantKey: cache.Key{Path: "/gFoo/method", KeyMap: ""},
},
{
desc: "keyBuilderMatch",
rpcPath: "/gFoo/method",
md: metadata.New(map[string]string{"n1": "v1", "n3": "v3"}),
wantKey: cache.Key{Path: "/gFoo/method", KeyMap: "k1=v1"},
},
}
for _, test := range tests {
t.Run(test.desc, func(t *testing.T) {
randID := grpcrand.Intn(math.MaxInt32)
p := picker{
kbm: kbm,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
readCache: func(key cache.Key) (*cache.Entry, bool) {
if !cmp.Equal(key, test.wantKey) {
t.Fatalf("picker using cacheKey %v, want %v", key, test.wantKey)
}
now := time.Now()
return &cache.Entry{
ExpiryTime: now.Add(defaultTestMaxAge),
StaleTime: now.Add(defaultTestMaxAge),
// Cache entry is configured with a child policy whose
// picker always returns an empty PickResult and nil
// error.
ChildPicker: &fakeChildPicker{id: randID},
}, false
},
// The other hooks are not set here because they are not expected to be
// invoked for these cases and if they get invoked, they will panic.
}
gotResult, err := p.Pick(balancer.PickInfo{
FullMethodName: test.rpcPath,
Ctx: metadata.NewOutgoingContext(context.Background(), test.md),
})
if err != nil {
t.Fatalf("Pick() failed with error: %v", err)
}
sc, ok := gotResult.SubConn.(*fakeSubConn)
if !ok {
t.Fatalf("Pick() returned a SubConn of type %T, want %T", gotResult.SubConn, &fakeSubConn{})
}
if sc.id != randID {
t.Fatalf("Pick() returned SubConn %d, want %d", sc.id, randID)
}
})
}
}
func TestPick(t *testing.T) {
const (
rpcPath = "/gFoo/method"
wantKeyMapStr = "k1=v1"
)
kbm, err := initKeyBuilderMap()
if err != nil {
t.Fatalf("Failed to create keyBuilderMap: %v", err)
}
md := metadata.New(map[string]string{"n1": "v1", "n3": "v3"})
wantKey := cache.Key{Path: rpcPath, KeyMap: wantKeyMapStr}
rlsLastErr := errors.New("last RLS request failed")
tests := []struct {
desc string
// The cache entry, as returned by the overridden readCache hook.
cacheEntry *cache.Entry
// Whether or not a pending entry exists, as returned by the overridden
// readCache hook.
pending bool
// Whether or not the RLS request should be throttled.
throttle bool
// Whether or not the test is expected to make a new RLS request.
newRLSRequest bool
// Whether or not the test ends up delegating to the default pick.
useDefaultPick bool
// Whether or not the test ends up delegating to the child policy in
// the cache entry.
useChildPick bool
// Request processing strategy as used by the picker.
strategy rlspb.RouteLookupConfig_RequestProcessingStrategy
// Expected error returned by the picker under test.
wantErr error
}{
{
desc: "cacheMiss_pending_defaultTargetOnError",
pending: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheMiss_pending_clientSeesError",
pending: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheMiss_pending_defaultTargetOnMiss",
pending: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheMiss_noPending_notThrottled_defaultTargetOnError",
newRLSRequest: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheMiss_noPending_notThrottled_clientSeesError",
newRLSRequest: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheMiss_noPending_notThrottled_defaultTargetOnMiss",
newRLSRequest: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheMiss_noPending_throttled_defaultTargetOnError",
throttle: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheMiss_noPending_throttled_clientSeesError",
throttle: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: errRLSThrottled,
},
{
desc: "cacheMiss_noPending_throttled_defaultTargetOnMiss",
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
throttle: true,
useDefaultPick: true,
wantErr: nil,
},
{
desc: "cacheHit_noPending_boExpired_dataExpired_throttled_defaultTargetOnError",
cacheEntry: &cache.Entry{}, // Everything is expired in this entry
throttle: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_boExpired_dataExpired_throttled_clientSeesError",
cacheEntry: &cache.Entry{}, // Everything is expired in this entry
throttle: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: errRLSThrottled,
},
{
desc: "cacheHit_noPending_boExpired_dataExpired_throttled_defaultTargetOnMiss",
cacheEntry: &cache.Entry{}, // Everything is expired in this entry
throttle: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boExpired_dataNotExpired_throttled_defaultTargetOnMiss",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
throttle: true, // Proactive refresh is throttled.
useChildPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boExpired_dataNotExpired_throttled_clientSeesError",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
throttle: true, // Proactive refresh is throttled.
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boExpired_dataNotExpired_throttled_defaultTargetOnError",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
throttle: true, // Proactive refresh is throttled.
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_boExpired_dataExpired_notThrottled_defaultTargetOnError",
cacheEntry: &cache.Entry{}, // Everything is expired in this entry
newRLSRequest: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheHit_noPending_boExpired_dataExpired_notThrottled_clientSeesError",
cacheEntry: &cache.Entry{}, // Everything is expired in this entry
newRLSRequest: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheHit_noPending_boExpired_dataExpired_notThrottled_defaultTargetOnMiss",
cacheEntry: &cache.Entry{}, // Everything is expired in this entry
newRLSRequest: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boExpired_dataNotExpired_notThrottled_defaultTargetOnMiss",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
newRLSRequest: true, // Proactive refresh.
useChildPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boExpired_dataNotExpired_notThrottled_clientSeesError",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
newRLSRequest: true, // Proactive refresh.
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boExpired_dataNotExpired_notThrottled_defaultTargetOnError",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
newRLSRequest: true, // Proactive refresh.
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boNotExpired_dataExpired_defaultTargetOnError",
cacheEntry: &cache.Entry{BackoffTime: time.Now().Add(defaultTestMaxAge)},
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boNotExpired_dataExpired_defaultTargetOnMiss",
cacheEntry: &cache.Entry{BackoffTime: time.Now().Add(defaultTestMaxAge)},
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boNotExpired_dataExpired_clientSeesError",
cacheEntry: &cache.Entry{
BackoffTime: time.Now().Add(defaultTestMaxAge),
CallStatus: rlsLastErr,
},
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: rlsLastErr,
},
{
desc: "cacheHit_noPending_stale_boNotExpired_dataNotExpired_defaultTargetOnError",
cacheEntry: &cache.Entry{
ExpiryTime: time.Now().Add(defaultTestMaxAge),
BackoffTime: time.Now().Add(defaultTestMaxAge),
CallStatus: rlsLastErr,
},
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boNotExpired_dataNotExpired_defaultTargetOnMiss",
cacheEntry: &cache.Entry{
ExpiryTime: time.Now().Add(defaultTestMaxAge),
BackoffTime: time.Now().Add(defaultTestMaxAge),
CallStatus: rlsLastErr,
},
useChildPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_noPending_stale_boNotExpired_dataNotExpired_clientSeesError",
cacheEntry: &cache.Entry{
ExpiryTime: time.Now().Add(defaultTestMaxAge),
BackoffTime: time.Now().Add(defaultTestMaxAge),
CallStatus: rlsLastErr,
},
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_notStale_dataNotExpired_defaultTargetOnError",
cacheEntry: &cache.Entry{
ExpiryTime: time.Now().Add(defaultTestMaxAge),
StaleTime: time.Now().Add(defaultTestMaxAge),
},
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_noPending_notStale_dataNotExpired_defaultTargetOnMiss",
cacheEntry: &cache.Entry{
ExpiryTime: time.Now().Add(defaultTestMaxAge),
StaleTime: time.Now().Add(defaultTestMaxAge),
},
useChildPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_noPending_notStale_dataNotExpired_clientSeesError",
cacheEntry: &cache.Entry{
ExpiryTime: time.Now().Add(defaultTestMaxAge),
StaleTime: time.Now().Add(defaultTestMaxAge),
},
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_pending_dataExpired_boExpired_defaultTargetOnError",
cacheEntry: &cache.Entry{},
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheHit_pending_dataExpired_boExpired_defaultTargetOnMiss",
cacheEntry: &cache.Entry{},
pending: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_pending_dataExpired_boExpired_clientSeesError",
cacheEntry: &cache.Entry{},
pending: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: balancer.ErrNoSubConnAvailable,
},
{
desc: "cacheHit_pending_dataExpired_boNotExpired_defaultTargetOnError",
cacheEntry: &cache.Entry{
BackoffTime: time.Now().Add(defaultTestMaxAge),
CallStatus: rlsLastErr,
},
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_pending_dataExpired_boNotExpired_defaultTargetOnMiss",
cacheEntry: &cache.Entry{
BackoffTime: time.Now().Add(defaultTestMaxAge),
CallStatus: rlsLastErr,
},
pending: true,
useDefaultPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_pending_dataExpired_boNotExpired_clientSeesError",
cacheEntry: &cache.Entry{
BackoffTime: time.Now().Add(defaultTestMaxAge),
CallStatus: rlsLastErr,
},
pending: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: rlsLastErr,
},
{
desc: "cacheHit_pending_dataNotExpired_defaultTargetOnError",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
pending: true,
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_DEFAULT_TARGET_ON_ERROR,
wantErr: nil,
},
{
desc: "cacheHit_pending_dataNotExpired_defaultTargetOnMiss",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
pending: true,
useChildPick: true,
strategy: rlspb.RouteLookupConfig_ASYNC_LOOKUP_DEFAULT_TARGET_ON_MISS,
wantErr: nil,
},
{
desc: "cacheHit_pending_dataNotExpired_clientSeesError",
cacheEntry: &cache.Entry{ExpiryTime: time.Now().Add(defaultTestMaxAge)},
pending: true,
useChildPick: true,
strategy: rlspb.RouteLookupConfig_SYNC_LOOKUP_CLIENT_SEES_ERROR,
wantErr: nil,
},
}
for _, test := range tests {
t.Run(test.desc, func(t *testing.T) {
rlsCh := make(chan error, 1)
randID := grpcrand.Intn(math.MaxInt32)
// We instantiate a fakeChildPicker which will return a fakeSubConn
// with configured id. Either the childPicker or the defaultPicker
// is configured to use this fakePicker based on whether
// useChidlPick or useDefaultPick is set in the test.
childPicker := &fakeChildPicker{id: randID}
p := picker{
kbm: kbm,
strategy: test.strategy,
readCache: func(key cache.Key) (*cache.Entry, bool) {
if !cmp.Equal(key, wantKey) {
t.Fatalf("cache lookup using cacheKey %v, want %v", key, wantKey)
}
if test.useChildPick {
test.cacheEntry.ChildPicker = childPicker
}
return test.cacheEntry, test.pending
},
shouldThrottle: func() bool { return test.throttle },
startRLS: func(path string, km keys.KeyMap) {
if !test.newRLSRequest {
rlsCh <- errors.New("RLS request attempted when none was expected")
return
}
if path != rpcPath {
rlsCh <- fmt.Errorf("RLS request initiated for rpcPath %s, want %s", path, rpcPath)
return
}
if km.Str != wantKeyMapStr {
rlsCh <- fmt.Errorf("RLS request initiated with keys %v, want %v", km.Str, wantKeyMapStr)
return
}
rlsCh <- nil
},
defaultPick: func(info balancer.PickInfo) (balancer.PickResult, error) {
if !test.useDefaultPick {
return balancer.PickResult{}, errors.New("Using default pick when the test doesn't want to use default pick")
}
return childPicker.Pick(info)
},
}
gotResult, err := p.Pick(balancer.PickInfo{
FullMethodName: rpcPath,
Ctx: metadata.NewOutgoingContext(context.Background(), md),
})
if err != test.wantErr {
t.Fatalf("Pick() returned error {%v}, want {%v}", err, test.wantErr)
}
if test.useChildPick || test.useDefaultPick {
// For cases where the pick is not queued, but is delegated to
// either the child picker or the default picker, we verify that
// the expected fakeSubConn is returned.
sc, ok := gotResult.SubConn.(*fakeSubConn)
if !ok {
t.Fatalf("Pick() returned a SubConn of type %T, want %T", gotResult.SubConn, &fakeSubConn{})
}
if sc.id != randID {
t.Fatalf("Pick() returned SubConn %d, want %d", sc.id, randID)
}
}
// If the test specified that a new RLS request should be made,
// verify it.
if test.newRLSRequest {
timer := time.NewTimer(defaultTestTimeout)
select {
case err := <-rlsCh:
timer.Stop()
if err != nil {
t.Fatal(err)
}
case <-timer.C:
t.Fatal("Timeout waiting for RLS request to be sent out")
}
}
})
}
}