xds: xds_cluster_impl_balancer part 1 #4154
Conversation
menghanl
force-pushed
the
xds_cluster_impl_balancer
branch
3 times, most recently
from
3315489 to
5214205
Compare
| */ | ||
|
|
||
| // Package utils contains utility structs shared by the balancers. | ||
| package utils |
There was a problem hiding this comment.
Generic package names such a util/common/helper etc are better avoided. See go/go-utility-packages.
I would suggest a package name of loadstore and type name Wrapper.
| return | ||
| } | ||
| lsw.store = store | ||
| lsw.perCluster = nil |
There was a problem hiding this comment.
Why is this assignment to nil required?
| edsService string | ||
| // Both store and perCluster will be nil if load reporting is disabled (EDS | ||
| // response doesn't have LRS server name). Note that methods on Store and | ||
| // perCluster all handle nil, so there's no need to check nil before calling |
There was a problem hiding this comment.
The comment about nil is a little confusing, because we are checking for nil in LoadStoreWrapper.Call*() methods.
There was a problem hiding this comment.
The comment is no longer true. Updated.
| } | ||
|
|
||
| // CallStarted records a call started in the store. | ||
| func (lsw *LoadStoreWrapper) CallStarted(locality string) { |
There was a problem hiding this comment.
Is it possible that these Call* methods are called concurrently? If not, the RWMutex really worth it? Because I would assume the UpdateLoadStore() method is called really infrequently.
There was a problem hiding this comment.
Yes, they can. They are called by pickers, and each pick can record load.
| ) | ||
|
|
||
| /* | ||
| message XdsClusterImplLoadBalancingPolicyConfig { |
There was a problem hiding this comment.
Since we do have a dependency on the new protobuf library now (google.golang.org/protobuf), would it make more sense to directly use proto messages (from grpc-proto repo) corresponding to the service config, and convert from JSON directly into them.
I'm also wondering if this approach will simplify things in LB policies which generate service config for its children. Today we have strong coupling among these because the parent directly fills in the struct corresponding to the service config (and this struct is defined in the child policy).
There was a problem hiding this comment.
We cannot just use the proto message, because of the extra check we do on the sub balancer policies (internalserviceconfig.BalancerConfig).
Parent balancers shouldn't use child's policy struct directly (we do have some today, but we should change those).
Because the only validation we do is usually in parseConfig. Bypassing the JSON unmarshal also bypasses the validation, which can cause problems later.
There was a problem hiding this comment.
Would it be better to simply link to the serviceConfig proto message in grpc-proto repo rather than inlining it here in this comment? I'm only concerned about keeping this up-to-date if something changes in the actual proto.
There was a problem hiding this comment.
It's not in the proto. I deleted the comments.
| return cib.cc.NewSubConn(addresses, options) | ||
| } | ||
|
|
||
| func (cib *clusterImplBalancer) RemoveSubConn(conn balancer.SubConn) { |
There was a problem hiding this comment.
I'm not entirely convinced that it would be better to embed the balancer.ClientConn in clusterImplBalancer type, but the only advantage would be that when we change the balancer API (which we are planning to by adding UpdateAddresses() to balancer.ClientConn), we wouldn't need any changes here.
| restartLoadReport = true | ||
| cib.lrsServerName = newLRSServerName | ||
| } | ||
| if restartLoadReport { |
There was a problem hiding this comment.
Could we get rid of restartLoadReport var and do the underlying functionality under if cib.lrsServerName != newLRSServerName { ... }
There was a problem hiding this comment.
That's a good idea!
menghanl
force-pushed
the
xds_cluster_impl_balancer
branch
from
5214205 to
c6fe038
Compare
| */ | ||
|
|
||
| // Package clusterimpl implements the xds_cluster_impl balancing policy. It | ||
| // handles the features in cluster except name resolution. Features include |
There was a problem hiding this comment.
Not sure I understand your comment. Are you suggesting we don't mention the features?
Anyway, I modified the comment, to make it explicit that name resolution is done in xds_cluster_resolver.
|
|
||
| type clusterImplBB struct{} | ||
|
|
||
| func (wt *clusterImplBB) Build(cc balancer.ClientConn, _ balancer.BuildOptions) balancer.Balancer { |
| restartLoadReport = true | ||
| cib.lrsServerName = newLRSServerName | ||
| } | ||
| if restartLoadReport { |
There was a problem hiding this comment.
That's a good idea!
|
|
||
| // Compare new drop config. And update picker if it's changed. | ||
| var updatePicker bool | ||
| if cib.config == nil || !reflect.DeepEqual(cib.config.DropCategories, newConfig.DropCategories) { |
There was a problem hiding this comment.
Done.
This is not comparing the whole config, it's just comparing the drop categories.
Added a function for that.
| */ | ||
|
|
||
| // Package utils contains utility structs shared by the balancers. | ||
| package utils |
| edsService string | ||
| // Both store and perCluster will be nil if load reporting is disabled (EDS | ||
| // response doesn't have LRS server name). Note that methods on Store and | ||
| // perCluster all handle nil, so there's no need to check nil before calling |
There was a problem hiding this comment.
The comment is no longer true. Updated.
| return | ||
| } | ||
| lsw.store = store | ||
| lsw.perCluster = nil |
| } | ||
|
|
||
| // CallStarted records a call started in the store. | ||
| func (lsw *LoadStoreWrapper) CallStarted(locality string) { |
There was a problem hiding this comment.
Yes, they can. They are called by pickers, and each pick can record load.
| drops []*dropper | ||
| s balancer.State | ||
| loadStore loadReporter | ||
| counter *client.ServiceRequestsCounter |
There was a problem hiding this comment.
The counter map is global.
It's not in the client, but it's in the client package (though I kind of don't like that now, I may change it in the next PR).
This is a pointer to the global counter for this cluster.
And all circuit breaking code removed. Will add in the next PR.
menghanl
force-pushed
the
xds_cluster_impl_balancer
branch
2 times, most recently
from
271a479 to
b2fc9e1
Compare
|
|
||
| type clusterImplBB struct{} | ||
|
|
||
| func (clusterImplBB) Build(cc balancer.ClientConn, _ balancer.BuildOptions) balancer.Balancer { |
There was a problem hiding this comment.
Is it completely safe to ignore these BuildOptions here? Maybe some child policy down the tree needs it?
There was a problem hiding this comment.
No, we should keep it.
Added a field.
| b.logger = prefixLogger(b) | ||
| b.logger.Infof("Created") | ||
|
|
||
| client, err := newXDSClient() |
There was a problem hiding this comment.
Nit: It might be better to create the xdsClient as the first thing in this function for two reasons:
- We can assign to
xdsCas part of the literal struct initialization - In the case that the
xdsClientcreation fails, we wouldn't see the log line emitted byb.logger.Infof("Created").
There was a problem hiding this comment.
We do it in this order for the logger.Errorf below. And logger needs b to print the pointer.
There was a problem hiding this comment.
Got it. Then maybe, we should move the line b.logger.Infof("Created") to be after this err != nil, so that we are sure that it is not printed in the case where the xds client creation fails.
| closed *grpcsync.Event | ||
|
|
||
| logger *grpclog.PrefixLogger | ||
| xdsC xdsClientInterface | ||
|
|
||
| config *lbConfig | ||
| childLB balancer.Balancer | ||
|
|
There was a problem hiding this comment.
Nit: seems a little too sparse. Maybe get rid of some of the newlines here?
| cib.edsServiceName = newConfig.EDSServiceName | ||
| } | ||
| if updateLoadClusterAndService { | ||
| // This updates the clusterName and serviceName that will reported for the |
There was a problem hiding this comment.
s/will reported/will be reported/
| ) | ||
|
|
||
| /* | ||
| message XdsClusterImplLoadBalancingPolicyConfig { |
There was a problem hiding this comment.
Would it be better to simply link to the serviceConfig proto message in grpc-proto repo rather than inlining it here in this comment? I'm only concerned about keeping this up-to-date if something changes in the actual proto.
| ) | ||
|
|
||
| // NewLoadStoreWrapper creates a Wrapper. | ||
| func NewLoadStoreWrapper() *Wrapper { |
There was a problem hiding this comment.
s/NewLoadStoreWrapper/NewWrapper/ ?
go/go-style/decisions#repetitive-with-package
| sds := loadStore.Stats([]string{testClusterName}) | ||
| if len(sds) == 0 { | ||
| t.Fatalf("loads for cluster %v not found in store", testClusterName) | ||
| } | ||
| sd := sds[0] | ||
| if sd.Cluster != testClusterName || sd.Service != testServiceName { | ||
| t.Fatalf("got unexpected load for %q, %q, want %q, %q", sd.Cluster, sd.Service, testClusterName, testServiceName) | ||
| } | ||
| const dropCount = rpcCount * dropNumerator / dropDenominator | ||
| if diff := cmp.Diff(sd.Drops, map[string]uint64{dropReason: dropCount}); diff != "" { | ||
| t.Fatalf("got unexpected drop reports, diff (-got, +want): %v", diff) | ||
| } |
There was a problem hiding this comment.
Can we directly instantiate a wantStats variable and replace all these individual checks with the following:
wantStats := []*Data{ ... }
gotStats := loadStore.Stats([]string{testClusterName})
if !cmp.Diff(gotStats, wantStats) {
t.Fatalf(...)
}go/go-style/decisions#compare-full-structures
| } | ||
| } | ||
|
|
||
| sds2 := loadStore.Stats([]string{testClusterName}) |
There was a problem hiding this comment.
Same here. Prefer full struct comparisons.
| sds := loadStore.Stats([]string{testClusterName}) | ||
| if len(sds) == 0 { | ||
| t.Fatalf("loads for cluster %v not found in store", testClusterName) | ||
| } | ||
| sd := sds[0] | ||
| if sd.Cluster != testClusterName || sd.Service != testServiceName { | ||
| t.Fatalf("got unexpected load for %q, %q, want %q, %q", sd.Cluster, sd.Service, testClusterName, testServiceName) | ||
| } | ||
| const dropCount = rpcCount * dropNumerator / dropDenominator | ||
| if diff := cmp.Diff(sd.Drops, map[string]uint64{dropReason: dropCount}); diff != "" { | ||
| t.Fatalf("got unexpected drop reports, diff (-got, +want): %v", diff) | ||
| } |
| } | ||
| } | ||
|
|
||
| sds2 := loadStore.Stats([]string{testClusterName}) |
|
|
||
| type clusterImplBB struct{} | ||
|
|
||
| func (clusterImplBB) Build(cc balancer.ClientConn, _ balancer.BuildOptions) balancer.Balancer { |
There was a problem hiding this comment.
No, we should keep it.
Added a field.
| b.logger = prefixLogger(b) | ||
| b.logger.Infof("Created") | ||
|
|
||
| client, err := newXDSClient() |
There was a problem hiding this comment.
We do it in this order for the logger.Errorf below. And logger needs b to print the pointer.
| closed *grpcsync.Event | ||
|
|
||
| logger *grpclog.PrefixLogger | ||
| xdsC xdsClientInterface | ||
|
|
||
| config *lbConfig | ||
| childLB balancer.Balancer | ||
|
|
| cib.edsServiceName = newConfig.EDSServiceName | ||
| } | ||
| if updateLoadClusterAndService { | ||
| // This updates the clusterName and serviceName that will reported for the |
| ) | ||
|
|
||
| /* | ||
| message XdsClusterImplLoadBalancingPolicyConfig { |
There was a problem hiding this comment.
It's not in the proto. I deleted the comments.
| ) | ||
|
|
||
| // NewLoadStoreWrapper creates a Wrapper. | ||
| func NewLoadStoreWrapper() *Wrapper { |
menghanl
force-pushed
the
xds_cluster_impl_balancer
branch
from
b684f2e to
1626a79
Compare
| b.logger = prefixLogger(b) | ||
| b.logger.Infof("Created") | ||
|
|
||
| client, err := newXDSClient() |
There was a problem hiding this comment.
Got it. Then maybe, we should move the line b.logger.Infof("Created") to be after this err != nil, so that we are sure that it is not printed in the case where the xds client creation fails.
menghanl
force-pushed
the
xds_cluster_impl_balancer
branch
from
12fd642 to
3a50f6f
Compare
Part of C2P fallback. To support fallback to a DNS cluster.
This PR adds implementation of xds_cluster_impl_balancer, which will be responsible for circuit breaking and rpc dropping.
This PR only added RPC dropping. Circuit breaking will be done in a followup PR, after some necessary refactoring.