Skip to content

Commit

Permalink
xdsclient: Populate total_issued_requests count in LRS load reports (#…
Browse files Browse the repository at this point in the history 
…7544) (#7565)

* Populate issued count in LRS load report

* Test success, error and issued counts

* Make pass/fail fractions unequal
  • Loading branch information
@arjan-bal
arjan-bal committed Aug 26, 2024
1 parent 8e3596c commit 973e3dc
Show file tree
Hide file tree
Showing 5 changed files with 80 additions and 21 deletions.
31 changes: 25 additions & 6 deletions xds/internal/balancer/clusterimpl/balancer_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,7 @@ func (s) TestDropByCategory(t *testing.T) {
sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
// Test pick with one backend.

const rpcCount = 20
const rpcCount = 24
if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
for i := 0; i < rpcCount; i++ {
gotSCSt, err := p.Pick(balancer.PickInfo{})
Expand All @@ -156,7 +156,13 @@ func (s) TestDropByCategory(t *testing.T) {
if err != nil || gotSCSt.SubConn != sc1 {
return fmt.Errorf("picker.Pick, got %v, %v, want SubConn=%v", gotSCSt, err, sc1)
}
if gotSCSt.Done != nil {
if gotSCSt.Done == nil {
continue
}
// Fail 1/4th of the requests that are not dropped.
if i%8 == 1 {
gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
} else {
gotSCSt.Done(balancer.DoneInfo{})
}
}
Expand All @@ -177,7 +183,11 @@ func (s) TestDropByCategory(t *testing.T) {
TotalDrops: dropCount,
Drops: map[string]uint64{dropReason: dropCount},
LocalityStats: map[string]load.LocalityData{
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{Succeeded: rpcCount - dropCount}},
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{
Succeeded: (rpcCount - dropCount) * 3 / 4,
Errored: (rpcCount - dropCount) / 4,
Issued: rpcCount - dropCount,
}},
},
}}

Expand Down Expand Up @@ -239,7 +249,10 @@ func (s) TestDropByCategory(t *testing.T) {
TotalDrops: dropCount2,
Drops: map[string]uint64{dropReason2: dropCount2},
LocalityStats: map[string]load.LocalityData{
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{Succeeded: rpcCount - dropCount2}},
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{
Succeeded: rpcCount - dropCount2,
Issued: rpcCount - dropCount2,
}},
},
}}

Expand Down Expand Up @@ -332,7 +345,9 @@ func (s) TestDropCircuitBreaking(t *testing.T) {
}
dones = append(dones, func() {
if gotSCSt.Done != nil {
gotSCSt.Done(balancer.DoneInfo{})
// Fail these requests to test error counts in the load
// report.
gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
}
})
}
Expand All @@ -356,7 +371,11 @@ func (s) TestDropCircuitBreaking(t *testing.T) {
Service: testServiceName,
TotalDrops: uint64(maxRequest),
LocalityStats: map[string]load.LocalityData{
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{Succeeded: uint64(rpcCount - maxRequest + 50)}},
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{
Succeeded: uint64(rpcCount - maxRequest),
Errored: 50,
Issued: uint64(rpcCount - maxRequest + 50),
}},
},
}}

Expand Down
17 changes: 16 additions & 1 deletion xds/internal/xdsclient/load/store.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,7 @@ func (ls *perClusterStore) CallStarted(locality string) {
p, _ = ls.localityRPCCount.LoadOrStore(locality, tp)
}
p.(*rpcCountData).incrInProgress()
p.(*rpcCountData).incrIssued()
}

// CallFinished adds one call finished record for the given locality.
Expand Down Expand Up @@ -248,6 +249,8 @@ type RequestData struct {
Errored uint64
// InProgress is the number of requests in flight.
InProgress uint64
// Issued is the total number requests that were sent.
Issued uint64
}

// ServerLoadData contains server load data.
Expand Down Expand Up @@ -296,7 +299,8 @@ func (ls *perClusterStore) stats() *Data {
succeeded := countData.loadAndClearSucceeded()
inProgress := countData.loadInProgress()
errored := countData.loadAndClearErrored()
if succeeded == 0 && inProgress == 0 && errored == 0 {
issued := countData.loadAndClearIssued()
if succeeded == 0 && inProgress == 0 && errored == 0 && issued == 0 {
return true
}

Expand All @@ -305,6 +309,7 @@ func (ls *perClusterStore) stats() *Data {
Succeeded: succeeded,
Errored: errored,
InProgress: inProgress,
Issued: issued,
},
LoadStats: make(map[string]ServerLoadData),
}
Expand Down Expand Up @@ -339,6 +344,7 @@ type rpcCountData struct {
succeeded *uint64
errored *uint64
inProgress *uint64
issued *uint64

// Map from load desc to load data (sum+count). Loading data from map is
// atomic, but updating data takes a lock, which could cause contention when
Expand All @@ -353,6 +359,7 @@ func newRPCCountData() *rpcCountData {
succeeded: new(uint64),
errored: new(uint64),
inProgress: new(uint64),
issued: new(uint64),
}
}

Expand Down Expand Up @@ -384,6 +391,14 @@ func (rcd *rpcCountData) loadInProgress() uint64 {
return atomic.LoadUint64(rcd.inProgress) // InProgress count is not clear when reading.
}

func (rcd *rpcCountData) incrIssued() {
atomic.AddUint64(rcd.issued, 1)
}

func (rcd *rpcCountData) loadAndClearIssued() uint64 {
return atomic.SwapUint64(rcd.issued, 0)
}

func (rcd *rpcCountData) addServerLoad(name string, d float64) {
loads, ok := rcd.serverLoads.Load(name)
if !ok {
Expand Down
50 changes: 36 additions & 14 deletions xds/internal/xdsclient/load/store_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,12 @@ func TestLocalityStats(t *testing.T) {
wantStoreData = &Data{
LocalityStats: map[string]LocalityData{
localities[0]: {
RequestStats: RequestData{Succeeded: 20, Errored: 10, InProgress: 10},
RequestStats: RequestData{
Succeeded: 20,
Errored: 10,
InProgress: 10,
Issued: 40,
},
LoadStats: map[string]ServerLoadData{
"net": {Count: 20, Sum: 20},
"disk": {Count: 20, Sum: 40},
Expand All @@ -108,7 +113,12 @@ func TestLocalityStats(t *testing.T) {
},
},
localities[1]: {
RequestStats: RequestData{Succeeded: 40, Errored: 20, InProgress: 20},
RequestStats: RequestData{
Succeeded: 40,
Errored: 20,
InProgress: 20,
Issued: 80,
},
LoadStats: map[string]ServerLoadData{
"net": {Count: 40, Sum: 40},
"disk": {Count: 40, Sum: 80},
Expand Down Expand Up @@ -192,7 +202,13 @@ func TestResetAfterStats(t *testing.T) {
},
LocalityStats: map[string]LocalityData{
localities[0]: {
RequestStats: RequestData{Succeeded: 20, Errored: 10, InProgress: 10},
RequestStats: RequestData{
Succeeded: 20,
Errored: 10,
InProgress: 10,
Issued: 40,
},

LoadStats: map[string]ServerLoadData{
"net": {Count: 20, Sum: 20},
"disk": {Count: 20, Sum: 40},
Expand All @@ -201,7 +217,13 @@ func TestResetAfterStats(t *testing.T) {
},
},
localities[1]: {
RequestStats: RequestData{Succeeded: 40, Errored: 20, InProgress: 20},
RequestStats: RequestData{
Succeeded: 40,
Errored: 20,
InProgress: 20,
Issued: 80,
},

LoadStats: map[string]ServerLoadData{
"net": {Count: 40, Sum: 40},
"disk": {Count: 40, Sum: 80},
Expand Down Expand Up @@ -298,7 +320,7 @@ func TestStoreStats(t *testing.T) {
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
LocalityStats: map[string]LocalityData{
"test-locality": {
RequestStats: RequestData{Succeeded: 1},
RequestStats: RequestData{Succeeded: 1, Issued: 1},
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
},
},
Expand All @@ -308,7 +330,7 @@ func TestStoreStats(t *testing.T) {
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
LocalityStats: map[string]LocalityData{
"test-locality": {
RequestStats: RequestData{Succeeded: 1},
RequestStats: RequestData{Succeeded: 1, Issued: 1},
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
},
},
Expand All @@ -327,7 +349,7 @@ func TestStoreStats(t *testing.T) {
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
LocalityStats: map[string]LocalityData{
"test-locality": {
RequestStats: RequestData{Succeeded: 1},
RequestStats: RequestData{Succeeded: 1, Issued: 1},
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
},
},
Expand All @@ -337,7 +359,7 @@ func TestStoreStats(t *testing.T) {
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
LocalityStats: map[string]LocalityData{
"test-locality": {
RequestStats: RequestData{Succeeded: 1},
RequestStats: RequestData{Succeeded: 1, Issued: 1},
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
},
},
Expand All @@ -347,7 +369,7 @@ func TestStoreStats(t *testing.T) {
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
LocalityStats: map[string]LocalityData{
"test-locality": {
RequestStats: RequestData{Succeeded: 1},
RequestStats: RequestData{Succeeded: 1, Issued: 1},
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
},
},
Expand All @@ -357,7 +379,7 @@ func TestStoreStats(t *testing.T) {
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
LocalityStats: map[string]LocalityData{
"test-locality": {
RequestStats: RequestData{Succeeded: 1},
RequestStats: RequestData{Succeeded: 1, Issued: 1},
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
},
},
Expand Down Expand Up @@ -394,25 +416,25 @@ func TestStoreStatsEmptyDataNotReported(t *testing.T) {
{
Cluster: "c0", Service: "s0",
LocalityStats: map[string]LocalityData{
"test-locality": {RequestStats: RequestData{Succeeded: 1}},
"test-locality": {RequestStats: RequestData{Succeeded: 1, Issued: 1}},
},
},
{
Cluster: "c0", Service: "s1",
LocalityStats: map[string]LocalityData{
"test-locality": {RequestStats: RequestData{Succeeded: 1}},
"test-locality": {RequestStats: RequestData{Succeeded: 1, Issued: 1}},
},
},
{
Cluster: "c1", Service: "s0",
LocalityStats: map[string]LocalityData{
"test-locality": {RequestStats: RequestData{InProgress: 1}},
"test-locality": {RequestStats: RequestData{InProgress: 1, Issued: 1}},
},
},
{
Cluster: "c1", Service: "s1",
LocalityStats: map[string]LocalityData{
"test-locality": {RequestStats: RequestData{InProgress: 1}},
"test-locality": {RequestStats: RequestData{InProgress: 1, Issued: 1}},
},
},
}
Expand Down
1 change: 1 addition & 0 deletions xds/internal/xdsclient/transport/loadreport.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,7 @@ func (t *Transport) sendLoadStatsRequest(stream lrsStream, loads []*load.Data) e
TotalSuccessfulRequests: localityData.RequestStats.Succeeded,
TotalRequestsInProgress: localityData.RequestStats.InProgress,
TotalErrorRequests: localityData.RequestStats.Errored,
TotalIssuedRequests: localityData.RequestStats.Issued,
LoadMetricStats: loadMetricStats,
UpstreamEndpointStats: nil, // TODO: populate for per endpoint loads.
})
Expand Down
2 changes: 2 additions & 0 deletions xds/internal/xdsclient/transport/loadreport_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -151,12 +151,14 @@ func (s) TestReportLoad(t *testing.T) {
// TotalMetricValue is the aggregation of 3.14 + 2.718 = 5.858
{MetricName: testKey1, NumRequestsFinishedWithMetric: 2, TotalMetricValue: 5.858}},
TotalSuccessfulRequests: 1,
TotalIssuedRequests: 1,
},
{
Locality: &v3corepb.Locality{Region: "test-region2"},
LoadMetricStats: []*v3endpointpb.EndpointLoadMetricStats{
{MetricName: testKey2, NumRequestsFinishedWithMetric: 1, TotalMetricValue: 1.618}},
TotalSuccessfulRequests: 1,
TotalIssuedRequests: 1,
},
},
}
Expand Down

0 comments on commit 973e3dc

Please sign in to comment.