e241b33f26
Some checks failed
codecov / upload (push) Failing after 13m59s
CodeQL / Analyze (push) Failing after 1s
Testing / vet-proto (push) Failing after 1s
Testing / tests (arm, latest) (push) Waiting to run
Testing / extras (latest) (push) Failing after 1s
Testing / static checks (latest-1) (push) Failing after 1s
Testing / tests (-race, latest) (push) Failing after 1s
Testing / tests (i386, latest) (push) Failing after 1s
Testing / tests (latest) (push) Failing after 1s
Testing / tests (latest-1) (push) Failing after 1s
Lock Threads / lock (push) Failing after 1s
govulncheck / govulncheck (.) (push) Failing after 1s
govulncheck / govulncheck (./cmd/protoc-gen-go-grpc) (push) Failing after 1s
govulncheck / govulncheck (./gcp/observability) (push) Failing after 1s
govulncheck / govulncheck (./security/advancedtls) (push) Failing after 1s
govulncheck / govulncheck (./stats/opencensus) (push) Failing after 1s
Stale bot / stale (push) Failing after 1s
822 lines
28 KiB
Go
822 lines
28 KiB
Go
/*
|
|
*
|
|
* Copyright 2025 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 xdsclient
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/go-cmp/cmp"
|
|
"github.com/google/go-cmp/cmp/cmpopts"
|
|
"github.com/google/uuid"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
"google.golang.org/grpc/internal/envconfig"
|
|
"google.golang.org/grpc/internal/testutils"
|
|
"google.golang.org/grpc/internal/xds/clients"
|
|
"google.golang.org/grpc/internal/xds/clients/grpctransport"
|
|
xdstestutils "google.golang.org/grpc/internal/xds/clients/internal/testutils"
|
|
"google.golang.org/grpc/internal/xds/clients/internal/testutils/e2e"
|
|
"google.golang.org/grpc/internal/xds/clients/internal/testutils/fakeserver"
|
|
"google.golang.org/grpc/internal/xds/clients/xdsclient/internal/xdsresource"
|
|
"google.golang.org/protobuf/testing/protocmp"
|
|
"google.golang.org/protobuf/types/known/anypb"
|
|
|
|
v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
|
|
v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
|
|
v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
|
|
v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
|
|
v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
|
|
)
|
|
|
|
// xdsChannelForTest creates an xdsChannel to the specified serverURI for
|
|
// testing purposes.
|
|
func xdsChannelForTest(t *testing.T, serverURI, nodeID string, watchExpiryTimeout time.Duration) *xdsChannel {
|
|
t.Helper()
|
|
|
|
// Create a grpc transport to the above management server.
|
|
si := clients.ServerIdentifier{
|
|
ServerURI: serverURI,
|
|
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
|
|
}
|
|
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
|
|
tr, err := (grpctransport.NewBuilder(configs)).Build(si)
|
|
if err != nil {
|
|
t.Fatalf("Failed to create a transport for server config %v: %v", si, err)
|
|
}
|
|
|
|
serverCfg := ServerConfig{
|
|
ServerIdentifier: si,
|
|
}
|
|
clientConfig := Config{
|
|
Servers: []ServerConfig{serverCfg},
|
|
Node: clients.Node{ID: nodeID},
|
|
ResourceTypes: map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
|
|
}
|
|
// Create an xdsChannel that uses everything set up above.
|
|
xc, err := newXDSChannel(xdsChannelOpts{
|
|
transport: tr,
|
|
serverConfig: &serverCfg,
|
|
clientConfig: &clientConfig,
|
|
eventHandler: newTestEventHandler(),
|
|
watchExpiryTimeout: watchExpiryTimeout,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("Failed to create xdsChannel: %v", err)
|
|
}
|
|
t.Cleanup(func() { xc.close() })
|
|
return xc
|
|
}
|
|
|
|
// verifyUpdateAndMetadata verifies that the event handler received the expected
|
|
// updates and metadata. It checks that the received resource type matches the
|
|
// expected type, and that the received updates and metadata match the expected
|
|
// values. The function ignores the timestamp fields in the metadata, as those
|
|
// are expected to be different.
|
|
func verifyUpdateAndMetadata(ctx context.Context, t *testing.T, eh *testEventHandler, wantUpdates map[string]dataAndErrTuple, wantMD xdsresource.UpdateMetadata) {
|
|
t.Helper()
|
|
|
|
gotTyp, gotUpdates, gotMD, err := eh.waitForUpdate(ctx)
|
|
if err != nil {
|
|
t.Fatalf("Timeout when waiting for update callback to be invoked on the event handler")
|
|
}
|
|
|
|
if gotTyp != listenerType {
|
|
t.Fatalf("Got resource type %v, want %v", gotTyp, listenerType)
|
|
}
|
|
opts := cmp.Options{
|
|
protocmp.Transform(),
|
|
cmpopts.EquateEmpty(),
|
|
cmpopts.EquateErrors(),
|
|
cmpopts.IgnoreFields(xdsresource.UpdateMetadata{}, "Timestamp"),
|
|
cmpopts.IgnoreFields(xdsresource.UpdateErrorMetadata{}, "Timestamp"),
|
|
}
|
|
if diff := cmp.Diff(wantUpdates, gotUpdates, opts); diff != "" {
|
|
t.Fatalf("Got unexpected diff in update (-want +got):\n%s\n want: %+v\n got: %+v", diff, wantUpdates, gotUpdates)
|
|
}
|
|
if diff := cmp.Diff(wantMD, gotMD, opts); diff != "" {
|
|
t.Fatalf("Got unexpected diff in update (-want +got):\n%s\n want: %v\n got: %v", diff, wantMD, gotMD)
|
|
}
|
|
}
|
|
|
|
// Tests different failure cases when creating a new xdsChannel. It checks that
|
|
// the xdsChannel creation fails when any of the required options (transport,
|
|
// serverConfig, bootstrapConfig, or resourceTypeGetter) are missing or nil.
|
|
func (s) TestChannel_New_FailureCases(t *testing.T) {
|
|
type fakeTransport struct {
|
|
clients.Transport
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
opts xdsChannelOpts
|
|
wantErrStr string
|
|
}{
|
|
{
|
|
name: "emptyTransport",
|
|
opts: xdsChannelOpts{},
|
|
wantErrStr: "transport is nil",
|
|
},
|
|
{
|
|
name: "emptyServerConfig",
|
|
opts: xdsChannelOpts{transport: &fakeTransport{}},
|
|
wantErrStr: "serverConfig is nil",
|
|
},
|
|
{
|
|
name: "emptyCConfig",
|
|
opts: xdsChannelOpts{
|
|
transport: &fakeTransport{},
|
|
serverConfig: &ServerConfig{},
|
|
},
|
|
wantErrStr: "clientConfig is nil",
|
|
},
|
|
{
|
|
name: "emptyEventHandler",
|
|
opts: xdsChannelOpts{
|
|
transport: &fakeTransport{},
|
|
serverConfig: &ServerConfig{},
|
|
clientConfig: &Config{},
|
|
},
|
|
wantErrStr: "eventHandler is nil",
|
|
},
|
|
}
|
|
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
if _, err := newXDSChannel(test.opts); err == nil || !strings.Contains(err.Error(), test.wantErrStr) {
|
|
t.Fatalf("newXDSChannel() = %v, want %q", err, test.wantErrStr)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// Tests different scenarios of the xdsChannel receiving a response from the
|
|
// management server. In all scenarios, the xdsChannel is expected to pass the
|
|
// received responses as-is to the resource parsing functionality specified by
|
|
// the resourceTypeGetter.
|
|
func (s) TestChannel_ADS_HandleResponseFromManagementServer(t *testing.T) {
|
|
const (
|
|
listenerName1 = "listener-name-1"
|
|
listenerName2 = "listener-name-2"
|
|
routeName = "route-name"
|
|
clusterName = "cluster-name"
|
|
)
|
|
var (
|
|
badlyMarshaledResource = &anypb.Any{
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Value: []byte{1, 2, 3, 4},
|
|
}
|
|
apiListener = &v3listenerpb.ApiListener{
|
|
ApiListener: xdstestutils.MarshalAny(t, &v3httppb.HttpConnectionManager{
|
|
RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
|
|
RouteConfig: &v3routepb.RouteConfiguration{
|
|
Name: routeName},
|
|
},
|
|
}),
|
|
}
|
|
listener1 = xdstestutils.MarshalAny(t, &v3listenerpb.Listener{
|
|
Name: listenerName1,
|
|
ApiListener: apiListener,
|
|
})
|
|
listener2 = xdstestutils.MarshalAny(t, &v3listenerpb.Listener{
|
|
Name: listenerName2,
|
|
ApiListener: apiListener,
|
|
})
|
|
)
|
|
|
|
tests := []struct {
|
|
desc string
|
|
resourceNamesToRequest []string
|
|
managementServerResponse *v3discoverypb.DiscoveryResponse
|
|
wantUpdates map[string]dataAndErrTuple
|
|
wantMD xdsresource.UpdateMetadata
|
|
wantErr error
|
|
}{
|
|
{
|
|
desc: "one bad resource - deserialization failure",
|
|
resourceNamesToRequest: []string{listenerName1},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{badlyMarshaledResource},
|
|
},
|
|
wantUpdates: nil, // No updates expected as the response runs into unmarshaling errors.
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusNACKed,
|
|
Version: "0",
|
|
ErrState: &xdsresource.UpdateErrorMetadata{
|
|
Version: "0",
|
|
Err: cmpopts.AnyError,
|
|
},
|
|
},
|
|
wantErr: cmpopts.AnyError,
|
|
},
|
|
{
|
|
desc: "one bad resource - validation failure",
|
|
resourceNamesToRequest: []string{listenerName1},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{xdstestutils.MarshalAny(t, &v3listenerpb.Listener{
|
|
Name: listenerName1,
|
|
ApiListener: &v3listenerpb.ApiListener{
|
|
ApiListener: xdstestutils.MarshalAny(t, &v3httppb.HttpConnectionManager{
|
|
RouteSpecifier: &v3httppb.HttpConnectionManager_ScopedRoutes{},
|
|
}),
|
|
},
|
|
})},
|
|
},
|
|
wantUpdates: map[string]dataAndErrTuple{
|
|
listenerName1: {
|
|
Err: cmpopts.AnyError,
|
|
},
|
|
},
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusNACKed,
|
|
Version: "0",
|
|
ErrState: &xdsresource.UpdateErrorMetadata{
|
|
Version: "0",
|
|
Err: cmpopts.AnyError,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
desc: "two bad resources",
|
|
resourceNamesToRequest: []string{listenerName1, listenerName2},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{
|
|
badlyMarshaledResource,
|
|
xdstestutils.MarshalAny(t, &v3listenerpb.Listener{
|
|
Name: listenerName2,
|
|
ApiListener: &v3listenerpb.ApiListener{
|
|
ApiListener: xdstestutils.MarshalAny(t, &v3httppb.HttpConnectionManager{
|
|
RouteSpecifier: &v3httppb.HttpConnectionManager_ScopedRoutes{},
|
|
}),
|
|
},
|
|
}),
|
|
},
|
|
},
|
|
wantUpdates: map[string]dataAndErrTuple{
|
|
listenerName2: {
|
|
Err: cmpopts.AnyError,
|
|
},
|
|
},
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusNACKed,
|
|
Version: "0",
|
|
ErrState: &xdsresource.UpdateErrorMetadata{
|
|
Version: "0",
|
|
Err: cmpopts.AnyError,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
desc: "one good resource",
|
|
resourceNamesToRequest: []string{listenerName1},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{listener1},
|
|
},
|
|
wantUpdates: map[string]dataAndErrTuple{
|
|
listenerName1: {
|
|
Resource: &listenerResourceData{Resource: listenerUpdate{
|
|
RouteConfigName: routeName,
|
|
Raw: listener1.GetValue(),
|
|
}},
|
|
},
|
|
},
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusACKed,
|
|
Version: "0",
|
|
},
|
|
},
|
|
{
|
|
desc: "one good and one bad - deserialization failure",
|
|
resourceNamesToRequest: []string{listenerName1, listenerName2},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{
|
|
badlyMarshaledResource,
|
|
listener2,
|
|
},
|
|
},
|
|
wantUpdates: map[string]dataAndErrTuple{
|
|
listenerName2: {
|
|
Resource: &listenerResourceData{Resource: listenerUpdate{
|
|
RouteConfigName: routeName,
|
|
Raw: listener2.GetValue(),
|
|
}},
|
|
},
|
|
},
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusNACKed,
|
|
Version: "0",
|
|
ErrState: &xdsresource.UpdateErrorMetadata{
|
|
Version: "0",
|
|
Err: cmpopts.AnyError,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
desc: "one good and one bad - validation failure",
|
|
resourceNamesToRequest: []string{listenerName1, listenerName2},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{
|
|
xdstestutils.MarshalAny(t, &v3listenerpb.Listener{
|
|
Name: listenerName1,
|
|
ApiListener: &v3listenerpb.ApiListener{
|
|
ApiListener: xdstestutils.MarshalAny(t, &v3httppb.HttpConnectionManager{
|
|
RouteSpecifier: &v3httppb.HttpConnectionManager_ScopedRoutes{},
|
|
}),
|
|
},
|
|
}),
|
|
listener2,
|
|
},
|
|
},
|
|
wantUpdates: map[string]dataAndErrTuple{
|
|
listenerName1: {Err: cmpopts.AnyError},
|
|
listenerName2: {
|
|
Resource: &listenerResourceData{Resource: listenerUpdate{
|
|
RouteConfigName: routeName,
|
|
Raw: listener2.GetValue(),
|
|
}},
|
|
},
|
|
},
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusNACKed,
|
|
Version: "0",
|
|
ErrState: &xdsresource.UpdateErrorMetadata{
|
|
Version: "0",
|
|
Err: cmpopts.AnyError,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
desc: "two good resources",
|
|
resourceNamesToRequest: []string{listenerName1, listenerName2},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{listener1, listener2},
|
|
},
|
|
wantUpdates: map[string]dataAndErrTuple{
|
|
listenerName1: {
|
|
Resource: &listenerResourceData{Resource: listenerUpdate{
|
|
RouteConfigName: routeName,
|
|
Raw: listener1.GetValue(),
|
|
}},
|
|
},
|
|
listenerName2: {
|
|
Resource: &listenerResourceData{Resource: listenerUpdate{
|
|
RouteConfigName: routeName,
|
|
Raw: listener2.GetValue(),
|
|
}},
|
|
},
|
|
},
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusACKed,
|
|
Version: "0",
|
|
},
|
|
},
|
|
{
|
|
desc: "two resources when we requested one",
|
|
resourceNamesToRequest: []string{listenerName1},
|
|
managementServerResponse: &v3discoverypb.DiscoveryResponse{
|
|
VersionInfo: "0",
|
|
TypeUrl: "type.googleapis.com/envoy.config.listener.v3.Listener",
|
|
Resources: []*anypb.Any{listener1, listener2},
|
|
},
|
|
wantUpdates: map[string]dataAndErrTuple{
|
|
listenerName1: {
|
|
Resource: &listenerResourceData{Resource: listenerUpdate{
|
|
RouteConfigName: routeName,
|
|
Raw: listener1.GetValue(),
|
|
}},
|
|
},
|
|
listenerName2: {
|
|
Resource: &listenerResourceData{Resource: listenerUpdate{
|
|
RouteConfigName: routeName,
|
|
Raw: listener2.GetValue(),
|
|
}},
|
|
},
|
|
},
|
|
wantMD: xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusACKed,
|
|
Version: "0",
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, test := range tests {
|
|
t.Run(test.desc, func(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
|
|
// Start a fake xDS management server and configure the response it
|
|
// would send to its client.
|
|
mgmtServer, cleanup, err := fakeserver.StartServer(nil)
|
|
if err != nil {
|
|
t.Fatalf("Failed to start fake xDS server: %v", err)
|
|
}
|
|
defer cleanup()
|
|
t.Logf("Started xDS management server on %s", mgmtServer.Address)
|
|
mgmtServer.XDSResponseChan <- &fakeserver.Response{Resp: test.managementServerResponse}
|
|
|
|
// Create an xdsChannel for the test with a long watch expiry timer
|
|
// to ensure that watches don't expire for the duration of the test.
|
|
nodeID := uuid.New().String()
|
|
xc := xdsChannelForTest(t, mgmtServer.Address, nodeID, 2*defaultTestTimeout)
|
|
defer xc.close()
|
|
|
|
// Subscribe to the resources specified in the test table.
|
|
for _, name := range test.resourceNamesToRequest {
|
|
xc.subscribe(listenerType, name)
|
|
}
|
|
|
|
// Wait for an update callback on the event handler and verify the
|
|
// contents of the update and the metadata.
|
|
verifyUpdateAndMetadata(ctx, t, xc.eventHandler.(*testEventHandler), test.wantUpdates, test.wantMD)
|
|
})
|
|
}
|
|
}
|
|
|
|
// Tests that the xdsChannel correctly handles the expiry of a watch for a
|
|
// resource by ensuring that the watch expiry callback is invoked on the event
|
|
// handler with the expected resource type and name.
|
|
func (s) TestChannel_ADS_HandleResponseWatchExpiry(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
|
|
// Start an xDS management server, but do not configure any resources on it.
|
|
// This will result in the watch for a resource to timeout.
|
|
mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
|
|
|
|
// Create an xdsChannel for the test with a short watch expiry timer to
|
|
// ensure that the test does not run very long, as it needs to wait for the
|
|
// watch to expire.
|
|
nodeID := uuid.New().String()
|
|
xc := xdsChannelForTest(t, mgmtServer.Address, nodeID, 2*defaultTestShortTimeout)
|
|
defer xc.close()
|
|
|
|
// Subscribe to a listener resource.
|
|
const listenerName = "listener-name"
|
|
xc.subscribe(listenerType, listenerName)
|
|
|
|
// Wait for the watch expiry callback on the authority to be invoked and
|
|
// verify that the watch expired for the expected resource name and type.
|
|
eventHandler := xc.eventHandler.(*testEventHandler)
|
|
gotTyp, gotName, err := eventHandler.waitForResourceDoesNotExist(ctx)
|
|
if err != nil {
|
|
t.Fatal("Timeout when waiting for the watch expiry callback to be invoked on the xDS client")
|
|
}
|
|
|
|
if gotTyp != listenerType {
|
|
t.Fatalf("Got type %v, want %v", gotTyp, listenerType)
|
|
}
|
|
if gotName != listenerName {
|
|
t.Fatalf("Got name %v, want %v", gotName, listenerName)
|
|
}
|
|
}
|
|
|
|
// Tests that the xdsChannel correctly handles stream failures by ensuring that
|
|
// the stream failure callback is invoked on the event handler.
|
|
func (s) TestChannel_ADS_StreamFailure(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20000*defaultTestTimeout)
|
|
defer cancel()
|
|
|
|
// Start an xDS management server with a restartable listener to simulate
|
|
// connection failures.
|
|
l, err := net.Listen("tcp", "localhost:0")
|
|
if err != nil {
|
|
t.Fatalf("net.Listen() failed: %v", err)
|
|
}
|
|
lis := xdstestutils.NewRestartableListener(l)
|
|
mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: lis})
|
|
|
|
// Configure a listener resource on the management server.
|
|
const listenerResourceName = "test-listener-resource"
|
|
const routeConfigurationName = "test-route-configuration-resource"
|
|
nodeID := uuid.New().String()
|
|
resources := e2e.UpdateOptions{
|
|
NodeID: nodeID,
|
|
Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerResourceName, routeConfigurationName)},
|
|
SkipValidation: true,
|
|
}
|
|
if err := mgmtServer.Update(ctx, resources); err != nil {
|
|
t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
|
|
}
|
|
|
|
// Create an xdsChannel for the test with a long watch expiry timer
|
|
// to ensure that watches don't expire for the duration of the test.
|
|
xc := xdsChannelForTest(t, mgmtServer.Address, nodeID, 2000*defaultTestTimeout)
|
|
defer xc.close()
|
|
|
|
// Subscribe to the resource created above.
|
|
xc.subscribe(listenerType, listenerResourceName)
|
|
|
|
// Wait for an update callback on the event handler and verify the
|
|
// contents of the update and the metadata.
|
|
hcm := xdstestutils.MarshalAny(t, &v3httppb.HttpConnectionManager{
|
|
RouteSpecifier: &v3httppb.HttpConnectionManager_Rds{Rds: &v3httppb.Rds{
|
|
ConfigSource: &v3corepb.ConfigSource{
|
|
ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{Ads: &v3corepb.AggregatedConfigSource{}},
|
|
},
|
|
RouteConfigName: routeConfigurationName,
|
|
}},
|
|
})
|
|
listenerResource, err := anypb.New(&v3listenerpb.Listener{
|
|
Name: listenerResourceName,
|
|
ApiListener: &v3listenerpb.ApiListener{ApiListener: hcm},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("Failed to create listener resource: %v", err)
|
|
}
|
|
|
|
wantUpdates := map[string]dataAndErrTuple{
|
|
listenerResourceName: {
|
|
Resource: &listenerResourceData{
|
|
Resource: listenerUpdate{
|
|
RouteConfigName: routeConfigurationName,
|
|
Raw: listenerResource.GetValue(),
|
|
},
|
|
},
|
|
},
|
|
}
|
|
wantMD := xdsresource.UpdateMetadata{
|
|
Status: xdsresource.ServiceStatusACKed,
|
|
Version: "1",
|
|
}
|
|
|
|
eventHandler := xc.eventHandler.(*testEventHandler)
|
|
verifyUpdateAndMetadata(ctx, t, eventHandler, wantUpdates, wantMD)
|
|
|
|
lis.Stop()
|
|
if err := eventHandler.waitForStreamFailure(ctx); err != nil {
|
|
t.Fatalf("Timeout when waiting for the stream failure callback to be invoked on the xDS client: %v", err)
|
|
}
|
|
}
|
|
|
|
// Tests the behavior of the xdsChannel when a resource is unsubscribed.
|
|
// Verifies that when a previously subscribed resource is unsubscribed, a
|
|
// request is sent without the previously subscribed resource name.
|
|
func (s) TestChannel_ADS_ResourceUnsubscribe(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
|
|
// Start an xDS management server that uses a channel to inform the test
|
|
// about the specific LDS resource names being requested.
|
|
ldsResourcesCh := make(chan []string, 1)
|
|
mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
|
|
OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
|
|
t.Logf("Received request for resources: %v of type %s", req.GetResourceNames(), req.GetTypeUrl())
|
|
|
|
if req.TypeUrl != xdsresource.V3ListenerURL {
|
|
return fmt.Errorf("unexpected resource type URL: %q", req.TypeUrl)
|
|
}
|
|
|
|
// Make the most recently requested names available to the test.
|
|
ldsResourcesCh <- req.GetResourceNames()
|
|
return nil
|
|
},
|
|
})
|
|
|
|
// Configure two listener resources on the management server.
|
|
const listenerResourceName1 = "test-listener-resource-1"
|
|
const routeConfigurationName1 = "test-route-configuration-resource-1"
|
|
const listenerResourceName2 = "test-listener-resource-2"
|
|
const routeConfigurationName2 = "test-route-configuration-resource-2"
|
|
nodeID := uuid.New().String()
|
|
resources := e2e.UpdateOptions{
|
|
NodeID: nodeID,
|
|
Listeners: []*v3listenerpb.Listener{
|
|
e2e.DefaultClientListener(listenerResourceName1, routeConfigurationName1),
|
|
e2e.DefaultClientListener(listenerResourceName2, routeConfigurationName2),
|
|
},
|
|
SkipValidation: true,
|
|
}
|
|
if err := mgmtServer.Update(ctx, resources); err != nil {
|
|
t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
|
|
}
|
|
|
|
// Create an xdsChannel for the test with a long watch expiry timer
|
|
// to ensure that watches don't expire for the duration of the test.
|
|
xc := xdsChannelForTest(t, mgmtServer.Address, nodeID, 2*defaultTestTimeout)
|
|
defer xc.close()
|
|
|
|
// Subscribe to the resources created above and verify that a request is
|
|
// sent for the same.
|
|
xc.subscribe(listenerType, listenerResourceName1)
|
|
xc.subscribe(listenerType, listenerResourceName2)
|
|
if err := waitForResourceNames(ctx, ldsResourcesCh, []string{listenerResourceName1, listenerResourceName2}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Wait for the above resources to be ACKed.
|
|
if err := waitForResourceNames(ctx, ldsResourcesCh, []string{listenerResourceName1, listenerResourceName2}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Unsubscribe to one of the resources created above, and ensure that the
|
|
// other resource is still being requested.
|
|
xc.unsubscribe(listenerType, listenerResourceName1)
|
|
if err := waitForResourceNames(ctx, ldsResourcesCh, []string{listenerResourceName2}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Since the version on the management server for the above resource is not
|
|
// changed, we will not receive an update from it for the one resource that
|
|
// we are still requesting.
|
|
|
|
// Unsubscribe to the remaining resource, and ensure that no more resources
|
|
// are being requested.
|
|
xc.unsubscribe(listenerType, listenerResourceName2)
|
|
if err := waitForResourceNames(ctx, ldsResourcesCh, []string{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// waitForResourceNames waits for the wantNames to be received on namesCh.
|
|
// Returns a non-nil error if the context expires before that.
|
|
func waitForResourceNames(ctx context.Context, namesCh chan []string, wantNames []string) error {
|
|
var lastRequestedNames []string
|
|
for ; ; <-time.After(defaultTestShortTimeout) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return fmt.Errorf("timeout waiting for resources %v to be requested from the management server. Last requested resources: %v", wantNames, lastRequestedNames)
|
|
case gotNames := <-namesCh:
|
|
if cmp.Equal(gotNames, wantNames, cmpopts.EquateEmpty(), cmpopts.SortSlices(func(s1, s2 string) bool { return s1 < s2 })) {
|
|
return nil
|
|
}
|
|
lastRequestedNames = gotNames
|
|
}
|
|
}
|
|
}
|
|
|
|
// newTestEventHandler creates a new testEventHandler instance with the
|
|
// necessary channels for testing the xdsChannel.
|
|
func newTestEventHandler() *testEventHandler {
|
|
return &testEventHandler{
|
|
typeCh: make(chan ResourceType, 1),
|
|
updateCh: make(chan map[string]dataAndErrTuple, 1),
|
|
mdCh: make(chan xdsresource.UpdateMetadata, 1),
|
|
nameCh: make(chan string, 1),
|
|
connErrCh: make(chan error, 1),
|
|
}
|
|
}
|
|
|
|
// testEventHandler is a struct that implements the xdsChannelEventhandler
|
|
// interface. It is used to receive events from an xdsChannel, and has multiple
|
|
// channels on which it makes these events available to the test.
|
|
type testEventHandler struct {
|
|
typeCh chan ResourceType // Resource type of an update or resource-does-not-exist error.
|
|
updateCh chan map[string]dataAndErrTuple // Resource updates.
|
|
mdCh chan xdsresource.UpdateMetadata // Metadata from an update.
|
|
nameCh chan string // Name of the non-existent resource.
|
|
connErrCh chan error // Connectivity error.
|
|
|
|
}
|
|
|
|
func (ta *testEventHandler) adsStreamFailure(err error) {
|
|
ta.connErrCh <- err
|
|
}
|
|
|
|
func (ta *testEventHandler) waitForStreamFailure(ctx context.Context) error {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-ta.connErrCh:
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (ta *testEventHandler) adsResourceUpdate(typ ResourceType, updates map[string]dataAndErrTuple, md xdsresource.UpdateMetadata, onDone func()) {
|
|
ta.typeCh <- typ
|
|
ta.updateCh <- updates
|
|
ta.mdCh <- md
|
|
onDone()
|
|
}
|
|
|
|
// waitForUpdate waits for the next resource update event from the xdsChannel.
|
|
// It returns the resource type, the resource updates, and the update metadata.
|
|
// If the context is canceled, it returns an error.
|
|
func (ta *testEventHandler) waitForUpdate(ctx context.Context) (ResourceType, map[string]dataAndErrTuple, xdsresource.UpdateMetadata, error) {
|
|
var typ ResourceType
|
|
var updates map[string]dataAndErrTuple
|
|
var md xdsresource.UpdateMetadata
|
|
|
|
select {
|
|
case typ = <-ta.typeCh:
|
|
case <-ctx.Done():
|
|
return ResourceType{}, nil, xdsresource.UpdateMetadata{}, ctx.Err()
|
|
}
|
|
|
|
select {
|
|
case updates = <-ta.updateCh:
|
|
case <-ctx.Done():
|
|
return ResourceType{}, nil, xdsresource.UpdateMetadata{}, ctx.Err()
|
|
}
|
|
|
|
select {
|
|
case md = <-ta.mdCh:
|
|
case <-ctx.Done():
|
|
return ResourceType{}, nil, xdsresource.UpdateMetadata{}, ctx.Err()
|
|
}
|
|
return typ, updates, md, nil
|
|
}
|
|
|
|
func (ta *testEventHandler) adsResourceDoesNotExist(typ ResourceType, name string) {
|
|
ta.typeCh <- typ
|
|
ta.nameCh <- name
|
|
}
|
|
|
|
// waitForResourceDoesNotExist waits for the next resource-does-not-exist event
|
|
// from the xdsChannel. It returns the resource type and the resource name. If
|
|
// the context is canceled, it returns an error.
|
|
func (ta *testEventHandler) waitForResourceDoesNotExist(ctx context.Context) (ResourceType, string, error) {
|
|
var typ ResourceType
|
|
var name string
|
|
|
|
select {
|
|
case typ = <-ta.typeCh:
|
|
case <-ctx.Done():
|
|
return ResourceType{}, "", ctx.Err()
|
|
}
|
|
|
|
select {
|
|
case name = <-ta.nameCh:
|
|
case <-ctx.Done():
|
|
return ResourceType{}, "", ctx.Err()
|
|
}
|
|
return typ, name, nil
|
|
}
|
|
|
|
type panicDecoder struct{}
|
|
|
|
func (panicDecoder) Decode(*AnyProto, DecodeOptions) (*DecodeResult, error) {
|
|
panic("simulate panic")
|
|
}
|
|
|
|
// TestDecodeResponse_PanicRecoveryEnabled tests the panic recovery mechanism
|
|
// in decodeResponse. It verifies that if XDSRecoverPanicInResourceParsing
|
|
// env variable is enabled, panics during unmarshaling are caught and returned
|
|
// as errors.
|
|
func (s) TestDecodeResponse_PanicRecoveryEnabled(t *testing.T) {
|
|
rType := &ResourceType{
|
|
TypeName: "resourceType",
|
|
Decoder: panicDecoder{},
|
|
}
|
|
resp := response{resources: []*anypb.Any{{Value: []byte("test")}}}
|
|
wantErr := "recovered from panic during resource parsing"
|
|
|
|
xc := &xdsChannel{}
|
|
if _, _, err := xc.decodeResponse(rType, resp); err == nil || !strings.Contains(err.Error(), wantErr) {
|
|
t.Fatalf("decodeResponse() failed with err: %v, want %q", err, wantErr)
|
|
}
|
|
}
|
|
|
|
// TestDecodeResponse_PanicRecoveryDisabled tests the panic recovery mechanism
|
|
// in decodeResponse. It verifies that when XDSRecoverPanicInResourceParsing
|
|
// env variable is disabled, panics during unmarshaling propagate.
|
|
func (s) TestDecodeResponse_PanicRecoveryDisabled(t *testing.T) {
|
|
testutils.SetEnvConfig(t, &envconfig.XDSRecoverPanicInResourceParsing, false)
|
|
rType := &ResourceType{
|
|
TypeName: "resourceType",
|
|
Decoder: panicDecoder{},
|
|
}
|
|
resp := response{resources: []*anypb.Any{{Value: []byte("test")}}}
|
|
wantErr := "simulate panic"
|
|
xc := &xdsChannel{}
|
|
|
|
defer func() {
|
|
if r := recover(); r == nil || !strings.Contains(fmt.Sprint(r), wantErr) {
|
|
t.Fatalf("Expected panic in decodeResponse, got: %v, want: %q", r, wantErr)
|
|
}
|
|
}()
|
|
xc.decodeResponse(rType, resp)
|
|
}
|