New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
rls: propagate headers received in RLS response to backends #5883
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -20,16 +20,22 @@ package rls | |
|
||
import ( | ||
"context" | ||
"fmt" | ||
"testing" | ||
"time" | ||
|
||
"google.golang.org/grpc" | ||
"google.golang.org/grpc/codes" | ||
"google.golang.org/grpc/credentials/insecure" | ||
rlspb "google.golang.org/grpc/internal/proto/grpc_lookup_v1" | ||
"google.golang.org/grpc/internal/stubserver" | ||
rlstest "google.golang.org/grpc/internal/testutils/rls" | ||
"google.golang.org/grpc/metadata" | ||
"google.golang.org/grpc/status" | ||
"google.golang.org/protobuf/types/known/durationpb" | ||
|
||
rlspb "google.golang.org/grpc/internal/proto/grpc_lookup_v1" | ||
testgrpc "google.golang.org/grpc/test/grpc_testing" | ||
testpb "google.golang.org/grpc/test/grpc_testing" | ||
) | ||
|
||
// Test verifies the scenario where there is no matching entry in the data cache | ||
|
@@ -241,6 +247,72 @@ func (s) TestPick_DataCacheHit_NoPendingEntry_ValidEntry(t *testing.T) { | |
verifyRLSRequest(t, rlsReqCh, false) | ||
} | ||
|
||
// Test verifies the scenario where there is a matching entry in the data cache | ||
// which is valid and there is no pending request. The pick is expected to be | ||
// delegated to the child policy. | ||
func (s) TestPick_DataCacheHit_NoPendingEntry_ValidEntry_WithHeaderData(t *testing.T) { | ||
// Start an RLS server and set the throttler to never throttle requests. | ||
rlsServer, _ := rlstest.SetupFakeRLSServer(t, nil) | ||
overrideAdaptiveThrottler(t, neverThrottlingThrottler()) | ||
|
||
// Build the RLS config without a default target. | ||
rlsConfig := buildBasicRLSConfigWithChildPolicy(t, t.Name(), rlsServer.Address) | ||
|
||
// Start a test backend which expects the header data contents sent from the | ||
// RLS server to be part of RPC metadata as X-Google-RLS-Data header. | ||
const headerDataContents = "foo,bar,baz" | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I find this test case particularly interesting since header values are comma separated, aren't they? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The actual value does not matter to the LB policy. So, I just put something in :) Do you have a recommendation for a better value? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. No, I just thought it was interesting. Just to confirm: this feature allows us to set exactly one header to exactly one value, right? I.e. if the config has commas as the value, we aren't expected to add multiple values to the same header key, right? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
That's right. |
||
rpcErrCh := make(chan error, 1) | ||
backend := &stubserver.StubServer{ | ||
EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) { | ||
gotHeaderData := metadata.ValueFromIncomingContext(ctx, "x-google-rls-data") | ||
if len(gotHeaderData) != 1 || gotHeaderData[0] != headerDataContents { | ||
rpcErrCh <- fmt.Errorf("got metadata in `X-Google-RLS-Data` is %v, want %s", gotHeaderData, headerDataContents) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It seems like you could simplify this by just returning the error directly from the RPC? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done. |
||
return &testpb.Empty{}, nil | ||
} | ||
rpcErrCh <- nil | ||
return &testpb.Empty{}, nil | ||
}, | ||
} | ||
if err := backend.StartServer(); err != nil { | ||
t.Fatalf("Failed to start backend: %v", err) | ||
} | ||
t.Logf("Started TestService backend at: %q", backend.Address) | ||
t.Cleanup(func() { backend.Stop() }) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done. |
||
|
||
// Setup the fake RLS server to return the above backend as a target in the | ||
// RLS response. Also, populate the header data field in the response. | ||
rlsServer.SetResponseCallback(func(_ context.Context, req *rlspb.RouteLookupRequest) *rlstest.RouteLookupResponse { | ||
return &rlstest.RouteLookupResponse{Resp: &rlspb.RouteLookupResponse{ | ||
Targets: []string{backend.Address}, | ||
HeaderData: headerDataContents, | ||
}} | ||
}) | ||
|
||
// Register a manual resolver and push the RLS service config through it. | ||
r := startManualResolverWithConfig(t, rlsConfig) | ||
|
||
// Dial the backend. | ||
cc, err := grpc.Dial(r.Scheme()+":///", grpc.WithResolvers(r), grpc.WithTransportCredentials(insecure.NewCredentials())) | ||
if err != nil { | ||
t.Fatalf("grpc.Dial() failed: %v", err) | ||
} | ||
defer cc.Close() | ||
|
||
// Make an RPC and ensure it gets routed to the test backend with the header | ||
// data sent by the RLS server. | ||
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) | ||
defer cancel() | ||
testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}) | ||
select { | ||
case err := <-rpcErrCh: | ||
if err != nil { | ||
t.Fatal(err) | ||
} | ||
case <-ctx.Done(): | ||
t.Fatalf("Timeout when waiting for RPC to reach backend: %v", err) | ||
} | ||
} | ||
|
||
// Test verifies the scenario where there is a matching entry in the data cache | ||
// which is stale and there is no pending request. The pick is expected to be | ||
// delegated to the child policy with a proactive cache refresh. | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
const
?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done.