grpc
/
grpc-go
mirror of https://github.com/grpc/grpc-go.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

xds: modify generic clients grpctransport to accept optional custom grpc new client function (#8301)

This commit is contained in:
Purnesh Dixit 2025-05-08 14:54:25 +05:30 committed by GitHub
parent c84fab05de
commit 0e656b20dd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 214 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
xds/internal/clients/grpctransport/examples_test.go
View File

@ -30,10 +30,10 @@ import (
// its extensions.
//
// This example is creating clients.ServerIdentifier to connect to server at
// localhost:5678 using the credentials named "local". Note that "local" must
// exist as an entry in the provided credentials to grpctransport.Builder.
// localhost:5678 using the config named "local". Note that "local" must
// exist as an entry in the provided configs to grpctransport.Builder.
func ExampleServerIdentifierExtension() {
// Note the Extensions field is set by value and not by pointer.
fmt.Printf("%+v", clients.ServerIdentifier{ServerURI: "localhost:5678", Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"}})
// Output: {ServerURI:localhost:5678 Extensions:{Credentials:local}}
fmt.Printf("%+v", clients.ServerIdentifier{ServerURI: "localhost:5678", Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"}})
// Output: {ServerURI:localhost:5678 Extensions:{ConfigName:local}}
}

42
xds/internal/clients/grpctransport/grpc_transport.go
View File

@ -43,31 +43,41 @@ var (
// It must be set by value (not pointer) in the
// clients.ServerIdentifier.Extensions field (See Example).
type ServerIdentifierExtension struct {
// Credentials is name of the credentials to use for this transport to the
// server. It must be present in the map passed to NewBuilder.
Credentials string
// ConfigName is the name of the configuration to use for this transport.
// It must be present as a key in the map of configs passed to NewBuilder.
ConfigName string
}
// Builder creates gRPC-based Transports. It must be paired with ServerIdentifiers
// that contain an Extension field of type ServerIdentifierExtension.
type Builder struct {
// credentials is a map of credentials names to credentials.Bundle which
// can be used to connect to the server.
credentials map[string]credentials.Bundle
// configs is a map of configuration names to their respective Config.
configs map[string]Config
mu sync.Mutex
// connections is a map of clients.ServerIdentifiers in use by the Builder
// to connect to different servers.
connections map[clients.ServerIdentifier]*grpc.ClientConn
refs map[clients.ServerIdentifier]int
// refs tracks the number of active references to each connection.
refs map[clients.ServerIdentifier]int
}
// Config defines the configuration for connecting to a gRPC server, including
// credentials and an optional custom new client function.
type Config struct {
// Credentials is the credentials bundle to be used for the connection.
Credentials credentials.Bundle
// GRPCNewClient is an optional custom function to establish a gRPC connection.
// If nil, grpc.NewClient will be used as the default.
GRPCNewClient func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error)
}
// NewBuilder provides a builder for creating gRPC-based Transports using
// the credentials from provided map of credentials names to
// credentials.Bundle.
func NewBuilder(credentials map[string]credentials.Bundle) *Builder {
func NewBuilder(configs map[string]Config) *Builder {
return &Builder{
credentials: credentials,
configs: configs,
connections: make(map[clients.ServerIdentifier]*grpc.ClientConn),
refs: make(map[clients.ServerIdentifier]int),
}
@ -88,9 +98,12 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
return nil, fmt.Errorf("grpctransport: Extensions field is %T, but must be %T in ServerIdentifier", si.Extensions, ServerIdentifierExtension{})
}
creds, ok := b.credentials[sce.Credentials]
config, ok := b.configs[sce.ConfigName]
if !ok {
return nil, fmt.Errorf("grpctransport: unknown credentials type %q specified in extensions", sce.Credentials)
return nil, fmt.Errorf("grpctransport: unknown config name %q specified in ServerIdentifierExtension", sce.ConfigName)
}
if config.Credentials == nil {
return nil, fmt.Errorf("grpctransport: config %q has nil credentials bundle", sce.ConfigName)
}
b.mu.Lock()
@ -114,7 +127,12 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
Time: 5 * time.Minute,
Timeout: 20 * time.Second,
})
cc, err := grpc.NewClient(si.ServerURI, kpCfg, grpc.WithCredentialsBundle(creds), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
dopts := []grpc.DialOption{kpCfg, grpc.WithCredentialsBundle(config.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{}))}
newClientFunc := grpc.NewClient
if config.GRPCNewClient != nil {
newClientFunc = config.GRPCNewClient
}
cc, err := newClientFunc(si.ServerURI, dopts...)
if err != nil {
return nil, fmt.Errorf("grpctransport: failed to create connection to server %q: %v", si.ServerURI, err)
}

20
xds/internal/clients/grpctransport/grpc_transport_ext_test.go
View File

@ -72,14 +72,14 @@ func (s) TestBuild_Single(t *testing.T) {
serverID := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"},
}
credentials := map[string]credentials.Bundle{
"local": &testCredentials{transportCredentials: local.NewCredentials()},
configs := map[string]grpctransport.Config{
"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
}
// Calling Build() first time should create new gRPC transport.
builder := grpctransport.NewBuilder(credentials)
builder := grpctransport.NewBuilder(configs)
tr, err := builder.Build(serverID)
if err != nil {
t.Fatalf("Failed to build transport: %v", err)
@ -173,19 +173,19 @@ func (s) TestBuild_Multiple(t *testing.T) {
serverID1 := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"},
}
serverID2 := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{
"local": &testCredentials{transportCredentials: local.NewCredentials()},
"insecure": insecure.NewBundle(),
configs := map[string]grpctransport.Config{
"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
"insecure": {Credentials: insecure.NewBundle()},
}
// Create two gRPC transports.
builder := grpctransport.NewBuilder(credentials)
builder := grpctransport.NewBuilder(configs)
tr1, err := builder.Build(serverID1)
if err != nil {

116
xds/internal/clients/grpctransport/grpc_transport_test.go
View File

@ -137,15 +137,15 @@ func (tc *testCredentials) PerRPCCredentials() credentials.PerRPCCredentials {
// Transport in both cases when provided clients.ServerIdentifer is same
// one of the existing transport or a new one.
func (s) TestBuild_Success(t *testing.T) {
credentials := map[string]credentials.Bundle{
"local": &testCredentials{transportCredentials: local.NewCredentials()},
"insecure": insecure.NewBundle(),
configs := map[string]Config{
"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
"insecure": {Credentials: insecure.NewBundle()},
}
b := NewBuilder(credentials)
b := NewBuilder(configs)
serverID1 := clients.ServerIdentifier{
ServerURI: "server-address",
Extensions: ServerIdentifierExtension{Credentials: "local"},
Extensions: ServerIdentifierExtension{ConfigName: "local"},
}
tr1, err := b.Build(serverID1)
if err != nil {
@ -155,7 +155,7 @@ func (s) TestBuild_Success(t *testing.T) {
serverID2 := clients.ServerIdentifier{
ServerURI: "server-address",
Extensions: ServerIdentifierExtension{Credentials: "local"},
Extensions: ServerIdentifierExtension{ConfigName: "local"},
}
tr2, err := b.Build(serverID2)
if err != nil {
@ -165,7 +165,7 @@ func (s) TestBuild_Success(t *testing.T) {
serverID3 := clients.ServerIdentifier{
ServerURI: "server-address",
Extensions: ServerIdentifierExtension{Credentials: "insecure"},
Extensions: ServerIdentifierExtension{ConfigName: "insecure"},
}
tr3, err := b.Build(serverID3)
if err != nil {
@ -191,7 +191,7 @@ func (s) TestBuild_Failure(t *testing.T) {
name: "ServerURI is empty",
serverID: clients.ServerIdentifier{
ServerURI: "",
Extensions: ServerIdentifierExtension{Credentials: "local"},
Extensions: ServerIdentifierExtension{ConfigName: "local"},
},
},
{
@ -206,26 +206,41 @@ func (s) TestBuild_Failure(t *testing.T) {
},
},
{
name: "ServerIdentifierExtension Credentials is nil",
name: "ServerIdentifierExtension without ConfigName",
serverID: clients.ServerIdentifier{
ServerURI: "server-address",
Extensions: ServerIdentifierExtension{},
},
},
{
name: "ServerIdentifierExtension ConfigName is not present",
serverID: clients.ServerIdentifier{
ServerURI: "server-address",
Extensions: ServerIdentifierExtension{ConfigName: "unknown"},
},
},
{
name: "ServerIdentifierExtension ConfigName maps to nil credentials",
serverID: clients.ServerIdentifier{
ServerURI: "server-address",
Extensions: ServerIdentifierExtension{ConfigName: "nil-credentials"},
},
},
{
name: "ServerIdentifierExtension is added as pointer",
serverID: clients.ServerIdentifier{
ServerURI: "server-address",
Extensions: &ServerIdentifierExtension{Credentials: "local"},
Extensions: &ServerIdentifierExtension{ConfigName: "local"},
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
credentials := map[string]credentials.Bundle{
"local": &testCredentials{transportCredentials: local.NewCredentials()},
configs := map[string]Config{
"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
"nil-credentials": {Credentials: nil},
}
b := NewBuilder(credentials)
b := NewBuilder(configs)
tr, err := b.Build(test.serverID)
if err == nil {
t.Fatalf("Build() succeeded, want error")
@ -238,18 +253,19 @@ func (s) TestBuild_Failure(t *testing.T) {
}
// TestNewStream_Success verifies that NewStream() successfully creates a new
// client stream for the server when provided a valid server URI.
// client stream for the server when provided a valid server URI and a config
// with valid credentials.
func (s) TestNewStream_Success(t *testing.T) {
ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
serverCfg := clients.ServerIdentifier{
ServerURI: ts.address,
Extensions: ServerIdentifierExtension{Credentials: "local"},
Extensions: ServerIdentifierExtension{ConfigName: "local"},
}
credentials := map[string]credentials.Bundle{
"local": &testCredentials{transportCredentials: local.NewCredentials()},
configs := map[string]Config{
"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
}
builder := NewBuilder(credentials)
builder := NewBuilder(configs)
transport, err := builder.Build(serverCfg)
if err != nil {
t.Fatalf("Failed to build transport: %v", err)
@ -263,17 +279,65 @@ func (s) TestNewStream_Success(t *testing.T) {
}
}
// TestNewStream_Success_WithCustomGRPCNewClient verifies that NewStream()
// successfully creates a new client stream for the server when provided a
// valid server URI and a config with valid credentials and a custom gRPC
// NewClient function.
func (s) TestNewStream_Success_WithCustomGRPCNewClient(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
// Create a custom dialer function that will be used by the gRPC client.
customDialerCalled := make(chan struct{}, 1)
customGRPCNewClient := func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
customDialerCalled <- struct{}{}
return grpc.NewClient(target, opts...)
}
configs := map[string]Config{
"custom-dialer-config": {
Credentials: &testCredentials{transportCredentials: local.NewCredentials()},
GRPCNewClient: customGRPCNewClient,
},
}
builder := NewBuilder(configs)
serverID := clients.ServerIdentifier{
ServerURI: ts.address,
Extensions: ServerIdentifierExtension{ConfigName: "custom-dialer-config"},
}
transport, err := builder.Build(serverID)
if err != nil {
t.Fatalf("builder.Build(%+v) failed: %v", serverID, err)
}
defer transport.Close()
select {
case <-customDialerCalled:
case <-ctx.Done():
t.Fatalf("Timeout waiting for custom dialer to be called: %v", ctx.Err())
}
// Verify that the transport works by creating a stream.
if _, err = transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources"); err != nil {
t.Fatalf("transport.NewStream() failed with custom dialer: %v", err)
}
}
// TestNewStream_Error verifies that NewStream() returns an error
// when attempting to create a stream with an invalid server URI.
func (s) TestNewStream_Error(t *testing.T) {
serverCfg := clients.ServerIdentifier{
ServerURI: "invalid-server-uri",
Extensions: ServerIdentifierExtension{Credentials: "local"},
Extensions: ServerIdentifierExtension{ConfigName: "local"},
}
credentials := map[string]credentials.Bundle{
"local": &testCredentials{transportCredentials: local.NewCredentials()},
configs := map[string]Config{
"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
}
builder := NewBuilder(credentials)
builder := NewBuilder(configs)
transport, err := builder.Build(serverCfg)
if err != nil {
t.Fatalf("Failed to build transport: %v", err)
@ -304,12 +368,12 @@ func (s) TestStream_SendAndRecv(t *testing.T) {
// Build a grpc-based transport to the above server.
serverCfg := clients.ServerIdentifier{
ServerURI: ts.address,
Extensions: ServerIdentifierExtension{Credentials: "local"},
Extensions: ServerIdentifierExtension{ConfigName: "local"},
}
credentials := map[string]credentials.Bundle{
"local": &testCredentials{transportCredentials: local.NewCredentials()},
configs := map[string]Config{
"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
}
builder := NewBuilder(credentials)
builder := NewBuilder(configs)
transport, err := builder.Build(serverCfg)
if err != nil {
t.Fatalf("Failed to build transport: %v", err)

21
xds/internal/clients/lrsclient/loadreport_test.go
View File

@ -28,7 +28,6 @@ import (
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/internal/grpctest"
"google.golang.org/grpc/status"
@ -132,17 +131,17 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
// reporting is per-server and not per-authority.
nodeID := uuid.New().String()
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
config := lrsclient.Config{
Node: clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
}
client, err := lrsclient.New(config)
if err != nil {
t.Fatalf("lrsclient.New() failed: %v", err)
}
serverIdentifier1 := clients.ServerIdentifier{ServerURI: mgmtServer1.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
serverIdentifier1 := clients.ServerIdentifier{ServerURI: mgmtServer1.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
loadStore1, err := client.ReportLoad(serverIdentifier1)
if err != nil {
t.Fatalf("client.ReportLoad() failed: %v", err)
@ -162,7 +161,7 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
// Call the load reporting API to report load to the first management
// server, and ensure that a connection to the server is created.
serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
loadStore2, err := client.ReportLoad(serverIdentifier2)
if err != nil {
t.Fatalf("client.ReportLoad() failed: %v", err)
@ -276,10 +275,10 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
// Create an LRS client with configuration pointing to the above server.
nodeID := uuid.New().String()
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
config := lrsclient.Config{
Node: clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
}
client, err := lrsclient.New(config)
if err != nil {
@ -287,7 +286,7 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
}
// Call the load reporting API, and ensure that an LRS stream is created.
serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
loadStore1, err := client.ReportLoad(serverIdentifier)
if err != nil {
t.Fatalf("client.ReportLoad() failed: %v", err)
@ -471,10 +470,10 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
// Create an LRS client with configuration pointing to the above server.
nodeID := uuid.New().String()
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
config := lrsclient.Config{
Node: clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
}
client, err := lrsclient.New(config)
if err != nil {
@ -482,7 +481,7 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
}
// Call the load reporting API, and ensure that an LRS stream is created.
serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
loadStore, err := client.ReportLoad(serverIdentifier)
if err != nil {
t.Fatalf("client.ReportLoad() failed: %v", err)

7
xds/internal/clients/xdsclient/channel_test.go
View File

@ -29,7 +29,6 @@ import (
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/xds/internal/clients"
"google.golang.org/grpc/xds/internal/clients/grpctransport"
@ -55,10 +54,10 @@ func xdsChannelForTest(t *testing.T, serverURI, nodeID string, watchExpiryTimeou
// Create a grpc transport to the above management server.
si := clients.ServerIdentifier{
ServerURI: serverURI,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
tr, err := (grpctransport.NewBuilder(credentials)).Build(si)
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)
}

15
xds/internal/clients/xdsclient/test/authority_test.go
View File

@ -24,7 +24,6 @@ import (
"testing"
"github.com/google/uuid"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/xds/internal/clients"
"google.golang.org/grpc/xds/internal/clients/grpctransport"
@ -81,7 +80,7 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
nodeID := uuid.New().String()
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
ext := grpctransport.ServerIdentifierExtension{Credentials: "insecure"}
ext := grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}
siDefault := clients.ServerIdentifier{
ServerURI: defaultAuthorityServer.Address,
Extensions: ext,
@ -91,11 +90,11 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
Extensions: ext,
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: siDefault}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp style resource names used in this test use a slash removed
// version of t.Name as their authority, and the empty config
@ -242,20 +241,20 @@ func (s) TestAuthority_Fallback(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
psi := clients.ServerIdentifier{
ServerURI: primaryMgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
ssi := clients.ServerIdentifier{
ServerURI: secondaryMgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
// Create config with the above primary and fallback management servers,
// and an xDS client with that configuration.
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: psi}, {ServerIdentifier: ssi}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp resource names used in this test do not specify an
// authority. These will end up looking up an entry with the

15
xds/internal/clients/xdsclient/test/dump_test.go
View File

@ -28,7 +28,6 @@ import (
"github.com/google/go-cmp/cmp"
"github.com/google/uuid"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/xds/internal/clients"
"google.golang.org/grpc/xds/internal/clients/grpctransport"
@ -124,14 +123,14 @@ func (s) TestDumpResources_ManyToOne(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp resource names used in this test do not specify an
// authority. These will end up looking up an entry with the
@ -308,18 +307,18 @@ func (s) TestDumpResources_ManyToMany(t *testing.T) {
resourceTypes[xdsresource.V3ListenerURL] = listenerType
si1 := clients.ServerIdentifier{
ServerURI: mgmtServer1.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
si2 := clients.ServerIdentifier{
ServerURI: mgmtServer2.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si1}},
Node: clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp style resource names used in this test use a slash removed
// version of t.Name as their authority, and the empty config

75
xds/internal/clients/xdsclient/test/lds_watchers_test.go
View File

@ -28,7 +28,6 @@ import (
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/xds/internal/clients"
"google.golang.org/grpc/xds/internal/clients/grpctransport"
@ -252,14 +251,14 @@ func (s) TestLDSWatch(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp resource names used in this test do not specify an
// authority. These will end up looking up an entry with the
@ -398,14 +397,14 @@ func (s) TestLDSWatch_TwoWatchesForSameResourceName(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp resource names used in this test do not specify an
// authority. These will end up looking up an entry with the
@ -503,14 +502,14 @@ func (s) TestLDSWatch_ThreeWatchesForDifferentResourceNames(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp style resource names used in this test use a slash removed
// version of t.Name as their authority, and the empty config
@ -610,14 +609,14 @@ func (s) TestLDSWatch_ResourceCaching(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
}
@ -684,7 +683,7 @@ func (s) TestLDSWatch_ResourceCaching(t *testing.T) {
// does not receive an LDS response for the request that it sends. The test
// verifies that the watch callback is invoked with an error once the
// watchExpiryTimer fires.
func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
func TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
nodeID := uuid.New().String()
@ -692,14 +691,14 @@ func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
}
@ -743,14 +742,14 @@ func (s) TestLDSWatch_ValidResponseCancelsExpiryTimerBehavior(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
}
@ -820,14 +819,14 @@ func (s) TestLDSWatch_ResourceRemoved(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp style resource names used in this test use a slash removed
// version of t.Name as their authority, and the empty config
@ -949,14 +948,14 @@ func (s) TestLDSWatch_NewWatcherForRemovedResource(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
}
@ -1035,14 +1034,14 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
}
@ -1103,14 +1102,14 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
}
@ -1196,14 +1195,14 @@ func (s) TestLDSWatch_PartialValid(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp style resource names used in this test use a slash removed
// version of t.Name as their authority, and the empty config
@ -1286,14 +1285,14 @@ func (s) TestLDSWatch_PartialResponse(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp style resource names used in this test use a slash removed
// version of t.Name as their authority, and the empty config

21
xds/internal/clients/xdsclient/test/misc_watchers_test.go
View File

@ -26,7 +26,6 @@ import (
"testing"
"github.com/google/uuid"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/xds/internal/clients"
"google.golang.org/grpc/xds/internal/clients/grpctransport"
@ -109,14 +108,14 @@ func (s) TestWatchCallAnotherWatch(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp style resource names used in this test use a slash removed
// version of t.Name as their authority, and the empty config
@ -220,14 +219,14 @@ func (s) TestNodeProtoSentOnlyInFirstRequest(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp resource names used in this test do not specify an
// authority. These will end up looking up an entry with the
@ -385,14 +384,14 @@ func (s) TestWatchErrorsContainNodeID(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
xdsClientConfig := xdsclient.Config{
Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}},
Node: clients.Node{ID: nodeID},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
ResourceTypes: resourceTypes,
// Xdstp resource names used in this test do not specify an
// authority. These will end up looking up an entry with the
@ -483,7 +482,7 @@ func (s) TestWatchErrorsContainNodeID_ChannelCreationFailure(t *testing.T) {
resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
si := clients.ServerIdentifier{
ServerURI: mgmtServer.Address,
Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
}
xdsClientConfig := xdsclient.Config{

13
xds/internal/clients/xdsclient/xdsclient_test.go
View File

@ -22,7 +22,6 @@ import (
"strings"
"testing"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/xds/internal/clients"
"google.golang.org/grpc/xds/internal/clients/grpctransport"
@ -30,7 +29,7 @@ import (
)
func (s) TestXDSClient_New(t *testing.T) {
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
tests := []struct {
name string
@ -62,7 +61,7 @@ func (s) TestXDSClient_New(t *testing.T) {
config: Config{
Node: clients.Node{ID: "node-id"},
ResourceTypes: map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
},
wantErr: "no servers or authorities specified",
},
@ -71,7 +70,7 @@ func (s) TestXDSClient_New(t *testing.T) {
config: Config{
Node: clients.Node{ID: "node-id"},
ResourceTypes: map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
Servers: []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}},
},
wantErr: "",
@ -81,7 +80,7 @@ func (s) TestXDSClient_New(t *testing.T) {
config: Config{
Node: clients.Node{ID: "node-id"},
ResourceTypes: map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
Authorities: map[string]Authority{"authority-name": {XDSServers: []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}}}},
},
wantErr: "",
@ -107,11 +106,11 @@ func (s) TestXDSClient_New(t *testing.T) {
}
func (s) TestXDSClient_Close(t *testing.T) {
credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
config := Config{
Node: clients.Node{ID: "node-id"},
ResourceTypes: map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
TransportBuilder: grpctransport.NewBuilder(credentials),
TransportBuilder: grpctransport.NewBuilder(configs),
Servers: []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}},
}
c, err := New(config)