Skip to content

Commit 0e0fb60

Browse files
purnesh42Hpurnesh42H
committed
easwar review 2
1 parent 96e742b commit 0e0fb60

File tree

9 files changed

+399
-129
lines changed

9 files changed

+399
-129
lines changed

‎xds/internal/xdsclient/clientimpl.go‎

Lines changed: 73 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -32,11 +32,11 @@ import (
3232
"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
3333

3434
xdsbootstrap "google.golang.org/grpc/xds/bootstrap"
35-
gclients "google.golang.org/grpc/xds/internal/clients"
35+
"google.golang.org/grpc/xds/internal/clients"
3636
"google.golang.org/grpc/xds/internal/clients/grpctransport"
37-
glrsclient "google.golang.org/grpc/xds/internal/clients/lrsclient"
38-
gxdsclient "google.golang.org/grpc/xds/internal/clients/xdsclient"
39-
gxdsmetrics "google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
37+
"google.golang.org/grpc/xds/internal/clients/lrsclient"
38+
"google.golang.org/grpc/xds/internal/clients/xdsclient"
39+
"google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
4040
)
4141

4242
const (
@@ -79,18 +79,22 @@ var (
7979
})
8080
)
8181

82-
// clientImpl embed gxdsclient.XDSClient and implement internal XDSClient
82+
// clientImpl embed xdsclient.XDSClient and implement internal XDSClient
8383
// interface with ref counting so that it can be shared by the xds resolver and
8484
// balancer implementations, across multiple ClientConns and Servers.
8585
type clientImpl struct {
86-
*gxdsclient.XDSClient
86+
*xdsclient.XDSClient
8787

88-
gConfig gxdsclient.Config
89-
config *bootstrap.Config
90-
logger *grpclog.PrefixLogger
91-
target string
92-
lrsClient *glrsclient.LRSClient
93-
refCount int32 // accessed atomically
88+
// The following fields are initialized at creation time and are read-only
89+
// after that.
90+
xdsClientConfig xdsclient.Config
91+
bootstrapConfig *bootstrap.Config
92+
logger *grpclog.PrefixLogger
93+
target string
94+
lrsClient *lrsclient.LRSClient
95+
96+
// Accessed atomically
97+
refCount int32
9498
}
9599

96100
// metricsReporter implements the clients.MetricsReporter interface and uses an
@@ -109,63 +113,92 @@ func (mr *metricsReporter) ReportMetric(metric any) {
109113
}
110114

111115
switch m := metric.(type) {
112-
case *gxdsmetrics.ResourceUpdateValid:
116+
case *metrics.ResourceUpdateValid:
113117
xdsClientResourceUpdatesValidMetric.Record(mr.recorder, 1, mr.target, m.ServerURI, m.ResourceType)
114-
case *gxdsmetrics.ResourceUpdateInvalid:
118+
case *metrics.ResourceUpdateInvalid:
115119
xdsClientResourceUpdatesInvalidMetric.Record(mr.recorder, 1, mr.target, m.ServerURI, m.ResourceType)
116-
case *gxdsmetrics.ServerFailure:
120+
case *metrics.ServerFailure:
117121
xdsClientServerFailureMetric.Record(mr.recorder, 1, mr.target, m.ServerURI)
118122
}
119123
}
120124

121-
func newClientImpl(config *bootstrap.Config, metricsRecorder estats.MetricsRecorder, resourceTypes map[string]gxdsclient.ResourceType, target string) (*clientImpl, error) {
125+
func newClientImpl(config *bootstrap.Config, metricsRecorder estats.MetricsRecorder, target string) (*clientImpl, error) {
126+
gConfig, err := buildXDSClientConfig(config, metricsRecorder, target)
127+
if err != nil {
128+
return nil, err
129+
}
130+
client, err := xdsclient.New(gConfig)
131+
if err != nil {
132+
return nil, err
133+
}
134+
c := &clientImpl{XDSClient: client, xdsClientConfig: gConfig, bootstrapConfig: config, target: target, refCount: 1}
135+
c.logger = prefixLogger(c)
136+
return c, nil
137+
}
138+
139+
// BootstrapConfig returns the configuration read from the bootstrap file.
140+
// Callers must treat the return value as read-only.
141+
func (c *clientImpl) BootstrapConfig() *bootstrap.Config {
142+
return c.bootstrapConfig
143+
}
144+
145+
func (c *clientImpl) incrRef() int32 {
146+
return atomic.AddInt32(&c.refCount, 1)
147+
}
148+
149+
func (c *clientImpl) decrRef() int32 {
150+
return atomic.AddInt32(&c.refCount, -1)
151+
}
152+
153+
// buildXDSClientConfig builds the xdsclient.Config from the bootstrap.Config.
154+
func buildXDSClientConfig(config *bootstrap.Config, metricsRecorder estats.MetricsRecorder, target string) (xdsclient.Config, error) {
122155
grpcTransportConfigs := make(map[string]grpctransport.Config)
123-
gServerCfgMap := make(map[gxdsclient.ServerConfig]*bootstrap.ServerConfig)
156+
gServerCfgMap := make(map[xdsclient.ServerConfig]*bootstrap.ServerConfig)
124157

125-
gAuthorities := make(map[string]gxdsclient.Authority)
158+
gAuthorities := make(map[string]xdsclient.Authority)
126159
for name, cfg := range config.Authorities() {
127160
// If server configs are specified in the authorities map, use that.
128161
// Else, use the top-level server configs.
129162
serverCfg := config.XDSServers()
130163
if len(cfg.XDSServers) >= 1 {
131164
serverCfg = cfg.XDSServers
132165
}
133-
var gServerCfg []gxdsclient.ServerConfig
166+
var gServerCfg []xdsclient.ServerConfig
134167
for _, sc := range serverCfg {
135168
if err := populateGRPCTransportConfigsFromServerConfig(sc, grpcTransportConfigs); err != nil {
136-
return nil, err
169+
return xdsclient.Config{}, err
137170
}
138-
gsc := gxdsclient.ServerConfig{
139-
ServerIdentifier: gclients.ServerIdentifier{ServerURI: sc.ServerURI(), Extensions: grpctransport.ServerIdentifierExtension{ConfigName: sc.SelectedCreds().Type}},
171+
gsc := xdsclient.ServerConfig{
172+
ServerIdentifier: clients.ServerIdentifier{ServerURI: sc.ServerURI(), Extensions: grpctransport.ServerIdentifierExtension{ConfigName: sc.SelectedCreds().Type}},
140173
IgnoreResourceDeletion: sc.ServerFeaturesIgnoreResourceDeletion()}
141174
gServerCfg = append(gServerCfg, gsc)
142175
gServerCfgMap[gsc] = sc
143176
}
144-
gAuthorities[name] = gxdsclient.Authority{XDSServers: gServerCfg}
177+
gAuthorities[name] = xdsclient.Authority{XDSServers: gServerCfg}
145178
}
146179

147-
gServerCfgs := make([]gxdsclient.ServerConfig, 0, len(config.XDSServers()))
180+
gServerCfgs := make([]xdsclient.ServerConfig, 0, len(config.XDSServers()))
148181
for _, sc := range config.XDSServers() {
149182
if err := populateGRPCTransportConfigsFromServerConfig(sc, grpcTransportConfigs); err != nil {
150-
return nil, err
183+
return xdsclient.Config{}, err
151184
}
152-
gsc := gxdsclient.ServerConfig{
153-
ServerIdentifier: gclients.ServerIdentifier{ServerURI: sc.ServerURI(), Extensions: grpctransport.ServerIdentifierExtension{ConfigName: sc.SelectedCreds().Type}},
185+
gsc := xdsclient.ServerConfig{
186+
ServerIdentifier: clients.ServerIdentifier{ServerURI: sc.ServerURI(), Extensions: grpctransport.ServerIdentifierExtension{ConfigName: sc.SelectedCreds().Type}},
154187
IgnoreResourceDeletion: sc.ServerFeaturesIgnoreResourceDeletion()}
155188
gServerCfgs = append(gServerCfgs, gsc)
156189
gServerCfgMap[gsc] = sc
157190
}
158191

159192
node := config.Node()
160-
gNode := gclients.Node{
193+
gNode := clients.Node{
161194
ID: node.GetId(),
162195
Cluster: node.GetCluster(),
163196
Metadata: node.Metadata,
164197
UserAgentName: node.UserAgentName,
165198
UserAgentVersion: node.GetUserAgentVersion(),
166199
}
167200
if node.Locality != nil {
168-
gNode.Locality = gclients.Locality{
201+
gNode.Locality = clients.Locality{
169202
Region: node.Locality.Region,
170203
Zone: node.Locality.Zone,
171204
SubZone: node.Locality.SubZone,
@@ -174,65 +207,43 @@ func newClientImpl(config *bootstrap.Config, metricsRecorder estats.MetricsRecor
174207

175208
gTransportBuilder := grpctransport.NewBuilder(grpcTransportConfigs)
176209

177-
if resourceTypes == nil {
178-
resourceTypes = make(map[string]gxdsclient.ResourceType)
179-
resourceTypes[version.V3ListenerURL] = gxdsclient.ResourceType{
210+
resourceTypes := map[string]xdsclient.ResourceType{
211+
version.V3ListenerURL: {
180212
TypeURL: version.V3ListenerURL,
181213
TypeName: xdsresource.ListenerResourceTypeName,
182214
AllResourcesRequiredInSotW: true,
183215
Decoder: xdsresource.NewGenericListenerResourceTypeDecoder(config),
184-
}
185-
resourceTypes[version.V3RouteConfigURL] = gxdsclient.ResourceType{
216+
},
217+
version.V3RouteConfigURL: {
186218
TypeURL: version.V3RouteConfigURL,
187219
TypeName: xdsresource.RouteConfigTypeName,
188220
AllResourcesRequiredInSotW: false,
189221
Decoder: xdsresource.NewGenericRouteConfigResourceTypeDecoder(),
190-
}
191-
resourceTypes[version.V3ClusterURL] = gxdsclient.ResourceType{
222+
},
223+
version.V3ClusterURL: {
192224
TypeURL: version.V3ClusterURL,
193225
TypeName: xdsresource.ClusterResourceTypeName,
194226
AllResourcesRequiredInSotW: true,
195227
Decoder: xdsresource.NewGenericClusterResourceTypeDecoder(config, gServerCfgMap),
196-
}
197-
resourceTypes[version.V3EndpointsURL] = gxdsclient.ResourceType{
228+
},
229+
version.V3EndpointsURL: {
198230
TypeURL: version.V3EndpointsURL,
199231
TypeName: xdsresource.EndpointsResourceTypeName,
200232
AllResourcesRequiredInSotW: false,
201233
Decoder: xdsresource.NewGenericEndpointsResourceTypeDecoder(),
202-
}
234+
},
203235
}
204236

205237
mr := &metricsReporter{recorder: metricsRecorder, target: target}
206238

207-
gConfig := gxdsclient.Config{
239+
return xdsclient.Config{
208240
Authorities: gAuthorities,
209241
Servers: gServerCfgs,
210242
Node: gNode,
211243
TransportBuilder: gTransportBuilder,
212244
ResourceTypes: resourceTypes,
213245
MetricsReporter: mr,
214-
}
215-
client, err := gxdsclient.New(gConfig)
216-
if err != nil {
217-
return nil, err
218-
}
219-
c := &clientImpl{XDSClient: client, gConfig: gConfig, config: config, target: target, refCount: 1}
220-
c.logger = prefixLogger(c)
221-
return c, nil
222-
}
223-
224-
// BootstrapConfig returns the configuration read from the bootstrap file.
225-
// Callers must treat the return value as read-only.
226-
func (c *clientImpl) BootstrapConfig() *bootstrap.Config {
227-
return c.config
228-
}
229-
230-
func (c *clientImpl) incrRef() int32 {
231-
return atomic.AddInt32(&c.refCount, 1)
232-
}
233-
234-
func (c *clientImpl) decrRef() int32 {
235-
return atomic.AddInt32(&c.refCount, -1)
246+
}, nil
236247
}
237248

238249
// populateGRPCTransportConfigsFromServerConfig iterates through the channel

‎xds/internal/xdsclient/clientimpl_loadreport.go‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,16 +33,23 @@ import (
3333
// It returns a lrsclient.LoadStore for the user to report loads.
3434
func (c *clientImpl) ReportLoad(server *bootstrap.ServerConfig) (*lrsclient.LoadStore, func(context.Context)) {
3535
if c.lrsClient == nil {
36-
lrsConfig := lrsclient.Config{Node: c.gConfig.Node, TransportBuilder: c.gConfig.TransportBuilder}
37-
lrsC, err := lrsclient.New(lrsConfig)
36+
lrsC, err := lrsclient.New(lrsclient.Config{
37+
Node: c.xdsClientConfig.Node,
38+
TransportBuilder: c.xdsClientConfig.TransportBuilder,
39+
})
3840
if err != nil {
3941
c.logger.Warningf("Failed to create an lrs client to the management server to report load: %v", server, err)
4042
return nil, func(context.Context) {}
4143
}
4244
c.lrsClient = lrsC
4345
}
4446

45-
load, err := c.lrsClient.ReportLoad(clients.ServerIdentifier{ServerURI: server.ServerURI(), Extensions: grpctransport.ServerIdentifierExtension{ConfigName: server.SelectedCreds().Type}})
47+
load, err := c.lrsClient.ReportLoad(clients.ServerIdentifier{
48+
ServerURI: server.ServerURI(),
49+
Extensions: grpctransport.ServerIdentifierExtension{
50+
ConfigName: server.SelectedCreds().Type,
51+
},
52+
})
4653
if err != nil {
4754
c.logger.Warningf("Failed to create a load store to the management server to report load: %v", server, err)
4855
return nil, func(context.Context) {}

0 commit comments

Comments
 (0)