Skip to content

Commit d8924ac

Browse files
authored
xds: fix support for load reporting in LOGICAL_DNS clusters (#8170)
1 parent ce2fded commit d8924ac

File tree

3 files changed

+93
-0
lines changed

3 files changed

+93
-0
lines changed

xds/internal/balancer/cdsbalancer/cdsbalancer.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -656,6 +656,7 @@ func (b *cdsBalancer) generateDMsForCluster(name string, depth int, dms []cluste
656656
Cluster: cluster.ClusterName,
657657
DNSHostname: cluster.DNSHostName,
658658
MaxConcurrentRequests: cluster.MaxRequests,
659+
LoadReportingServer: cluster.LRSServerConfig,
659660
}
660661
}
661662
odJSON := cluster.OutlierDetection

xds/internal/balancer/clusterimpl/tests/balancer_test.go

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -572,3 +572,94 @@ func hostAndPortFromAddress(t *testing.T, addr string) (string, uint32) {
572572
}
573573
return host, uint32(port)
574574
}
575+
576+
// Tests that LRS works correctly in a LOGICAL_DNS cluster.
577+
func (s) TestLRSLogicalDNS(t *testing.T) {
578+
// Create an xDS management server that serves ADS and LRS requests.
579+
mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{SupportLoadReportingService: true})
580+
581+
// Create bootstrap configuration pointing to the above management server.
582+
nodeID := uuid.New().String()
583+
bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
584+
testutils.CreateBootstrapFileForTesting(t, bc)
585+
586+
// Create an xDS resolver with the above bootstrap configuration.
587+
var resolverBuilder resolver.Builder
588+
var err error
589+
if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
590+
resolverBuilder, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
591+
if err != nil {
592+
t.Fatalf("Failed to create xDS resolver for testing: %v", err)
593+
}
594+
}
595+
596+
// Start a server backend exposing the test service.
597+
server := stubserver.StartTestService(t, nil)
598+
defer server.Stop()
599+
host, port := hostAndPortFromAddress(t, server.Address)
600+
601+
// Configure the xDS management server with default resources. Override the
602+
// default cluster to include an LRS server config pointing to self.
603+
const serviceName = "my-test-xds-service"
604+
resources := e2e.DefaultClientResources(e2e.ResourceParams{
605+
DialTarget: serviceName,
606+
NodeID: nodeID,
607+
Host: "localhost",
608+
Port: testutils.ParsePort(t, server.Address),
609+
SecLevel: e2e.SecurityLevelNone,
610+
})
611+
resources.Clusters = []*v3clusterpb.Cluster{
612+
e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
613+
ClusterName: "cluster-" + serviceName,
614+
Type: e2e.ClusterTypeLogicalDNS,
615+
DNSHostName: host,
616+
DNSPort: port,
617+
}),
618+
}
619+
resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{
620+
ConfigSourceSpecifier: &v3corepb.ConfigSource_Self{
621+
Self: &v3corepb.SelfConfigSource{},
622+
},
623+
}
624+
resources.Endpoints = nil
625+
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
626+
defer cancel()
627+
if err := mgmtServer.Update(ctx, resources); err != nil {
628+
t.Fatal(err)
629+
}
630+
631+
// Create a ClientConn and make a successful RPC.
632+
cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(resolverBuilder))
633+
if err != nil {
634+
t.Fatalf("failed to dial local test server: %v", err)
635+
}
636+
defer cc.Close()
637+
638+
client := testgrpc.NewTestServiceClient(cc)
639+
if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
640+
t.Fatalf("rpc EmptyCall() failed: %v", err)
641+
}
642+
643+
// Ensure that an LRS stream is created.
644+
if _, err := mgmtServer.LRSServer.LRSStreamOpenChan.Receive(ctx); err != nil {
645+
t.Fatalf("Failure when waiting for an LRS stream to be opened: %v", err)
646+
}
647+
648+
// Handle the initial LRS request from the xDS client.
649+
if _, err = mgmtServer.LRSServer.LRSRequestChan.Receive(ctx); err != nil {
650+
t.Fatalf("Failure waiting for initial LRS request: %v", err)
651+
}
652+
653+
resp := fakeserver.Response{
654+
Resp: &v3lrspb.LoadStatsResponse{
655+
SendAllClusters: true,
656+
LoadReportingInterval: durationpb.New(10 * time.Millisecond),
657+
},
658+
}
659+
mgmtServer.LRSServer.LRSResponseChan <- &resp
660+
661+
// Wait for load to be reported for locality of server 1.
662+
if err := waitForSuccessfulLoadReport(ctx, mgmtServer.LRSServer, ""); err != nil {
663+
t.Fatalf("Server did not receive load due to error: %v", err)
664+
}
665+
}

xds/internal/balancer/clusterresolver/configbuilder.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -153,6 +153,7 @@ func buildClusterImplConfigForDNS(g *nameGenerator, endpoints []resolver.Endpoin
153153
TelemetryLabels: mechanism.TelemetryLabels,
154154
ChildPolicy: &internalserviceconfig.BalancerConfig{Name: childPolicy},
155155
MaxConcurrentRequests: mechanism.MaxConcurrentRequests,
156+
LoadReportingServer: mechanism.LoadReportingServer,
156157
}, retEndpoints
157158
}
158159

0 commit comments

Comments
 (0)