Skip to content

Commit 973e3dc

Browse files
arjan-balarjan-bal
authored
xdsclient: Populate total_issued_requests count in LRS load reports (#7544) (#7565)
* Populate issued count in LRS load report * Test success, error and issued counts * Make pass/fail fractions unequal
1 parent 8e3596c commit 973e3dc

File tree

5 files changed

+80
-21
lines changed

5 files changed

+80
-21
lines changed

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

Lines changed: 25 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -142,7 +142,7 @@ func (s) TestDropByCategory(t *testing.T) {
142142
sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
143143
// Test pick with one backend.
144144

145-
const rpcCount = 20
145+
const rpcCount = 24
146146
if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
147147
for i := 0; i < rpcCount; i++ {
148148
gotSCSt, err := p.Pick(balancer.PickInfo{})
@@ -156,7 +156,13 @@ func (s) TestDropByCategory(t *testing.T) {
156156
if err != nil || gotSCSt.SubConn != sc1 {
157157
return fmt.Errorf("picker.Pick, got %v, %v, want SubConn=%v", gotSCSt, err, sc1)
158158
}
159-
if gotSCSt.Done != nil {
159+
if gotSCSt.Done == nil {
160+
continue
161+
}
162+
// Fail 1/4th of the requests that are not dropped.
163+
if i%8 == 1 {
164+
gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
165+
} else {
160166
gotSCSt.Done(balancer.DoneInfo{})
161167
}
162168
}
@@ -177,7 +183,11 @@ func (s) TestDropByCategory(t *testing.T) {
177183
TotalDrops: dropCount,
178184
Drops: map[string]uint64{dropReason: dropCount},
179185
LocalityStats: map[string]load.LocalityData{
180-
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{Succeeded: rpcCount - dropCount}},
186+
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{
187+
Succeeded: (rpcCount - dropCount) * 3 / 4,
188+
Errored: (rpcCount - dropCount) / 4,
189+
Issued: rpcCount - dropCount,
190+
}},
181191
},
182192
}}
183193

@@ -239,7 +249,10 @@ func (s) TestDropByCategory(t *testing.T) {
239249
TotalDrops: dropCount2,
240250
Drops: map[string]uint64{dropReason2: dropCount2},
241251
LocalityStats: map[string]load.LocalityData{
242-
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{Succeeded: rpcCount - dropCount2}},
252+
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{
253+
Succeeded: rpcCount - dropCount2,
254+
Issued: rpcCount - dropCount2,
255+
}},
243256
},
244257
}}
245258

@@ -332,7 +345,9 @@ func (s) TestDropCircuitBreaking(t *testing.T) {
332345
}
333346
dones = append(dones, func() {
334347
if gotSCSt.Done != nil {
335-
gotSCSt.Done(balancer.DoneInfo{})
348+
// Fail these requests to test error counts in the load
349+
// report.
350+
gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
336351
}
337352
})
338353
}
@@ -356,7 +371,11 @@ func (s) TestDropCircuitBreaking(t *testing.T) {
356371
Service: testServiceName,
357372
TotalDrops: uint64(maxRequest),
358373
LocalityStats: map[string]load.LocalityData{
359-
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{Succeeded: uint64(rpcCount - maxRequest + 50)}},
374+
assertString(xdsinternal.LocalityID{}.ToString): {RequestStats: load.RequestData{
375+
Succeeded: uint64(rpcCount - maxRequest),
376+
Errored: 50,
377+
Issued: uint64(rpcCount - maxRequest + 50),
378+
}},
360379
},
361380
}}
362381

‎xds/internal/xdsclient/load/store.go

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,7 @@ func (ls *perClusterStore) CallStarted(locality string) {
174174
p, _ = ls.localityRPCCount.LoadOrStore(locality, tp)
175175
}
176176
p.(*rpcCountData).incrInProgress()
177+
p.(*rpcCountData).incrIssued()
177178
}
178179

179180
// CallFinished adds one call finished record for the given locality.
@@ -248,6 +249,8 @@ type RequestData struct {
248249
Errored uint64
249250
// InProgress is the number of requests in flight.
250251
InProgress uint64
252+
// Issued is the total number requests that were sent.
253+
Issued uint64
251254
}
252255

253256
// ServerLoadData contains server load data.
@@ -296,7 +299,8 @@ func (ls *perClusterStore) stats() *Data {
296299
succeeded := countData.loadAndClearSucceeded()
297300
inProgress := countData.loadInProgress()
298301
errored := countData.loadAndClearErrored()
299-
if succeeded == 0 && inProgress == 0 && errored == 0 {
302+
issued := countData.loadAndClearIssued()
303+
if succeeded == 0 && inProgress == 0 && errored == 0 && issued == 0 {
300304
return true
301305
}
302306

@@ -305,6 +309,7 @@ func (ls *perClusterStore) stats() *Data {
305309
Succeeded: succeeded,
306310
Errored: errored,
307311
InProgress: inProgress,
312+
Issued: issued,
308313
},
309314
LoadStats: make(map[string]ServerLoadData),
310315
}
@@ -339,6 +344,7 @@ type rpcCountData struct {
339344
succeeded *uint64
340345
errored *uint64
341346
inProgress *uint64
347+
issued *uint64
342348

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

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

394+
func (rcd *rpcCountData) incrIssued() {
395+
atomic.AddUint64(rcd.issued, 1)
396+
}
397+
398+
func (rcd *rpcCountData) loadAndClearIssued() uint64 {
399+
return atomic.SwapUint64(rcd.issued, 0)
400+
}
401+
387402
func (rcd *rpcCountData) addServerLoad(name string, d float64) {
388403
loads, ok := rcd.serverLoads.Load(name)
389404
if !ok {

‎xds/internal/xdsclient/load/store_test.go

Lines changed: 36 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,12 @@ func TestLocalityStats(t *testing.T) {
9999
wantStoreData = &Data{
100100
LocalityStats: map[string]LocalityData{
101101
localities[0]: {
102-
RequestStats: RequestData{Succeeded: 20, Errored: 10, InProgress: 10},
102+
RequestStats: RequestData{
103+
Succeeded: 20,
104+
Errored: 10,
105+
InProgress: 10,
106+
Issued: 40,
107+
},
103108
LoadStats: map[string]ServerLoadData{
104109
"net": {Count: 20, Sum: 20},
105110
"disk": {Count: 20, Sum: 40},
@@ -108,7 +113,12 @@ func TestLocalityStats(t *testing.T) {
108113
},
109114
},
110115
localities[1]: {
111-
RequestStats: RequestData{Succeeded: 40, Errored: 20, InProgress: 20},
116+
RequestStats: RequestData{
117+
Succeeded: 40,
118+
Errored: 20,
119+
InProgress: 20,
120+
Issued: 80,
121+
},
112122
LoadStats: map[string]ServerLoadData{
113123
"net": {Count: 40, Sum: 40},
114124
"disk": {Count: 40, Sum: 80},
@@ -192,7 +202,13 @@ func TestResetAfterStats(t *testing.T) {
192202
},
193203
LocalityStats: map[string]LocalityData{
194204
localities[0]: {
195-
RequestStats: RequestData{Succeeded: 20, Errored: 10, InProgress: 10},
205+
RequestStats: RequestData{
206+
Succeeded: 20,
207+
Errored: 10,
208+
InProgress: 10,
209+
Issued: 40,
210+
},
211+
196212
LoadStats: map[string]ServerLoadData{
197213
"net": {Count: 20, Sum: 20},
198214
"disk": {Count: 20, Sum: 40},
@@ -201,7 +217,13 @@ func TestResetAfterStats(t *testing.T) {
201217
},
202218
},
203219
localities[1]: {
204-
RequestStats: RequestData{Succeeded: 40, Errored: 20, InProgress: 20},
220+
RequestStats: RequestData{
221+
Succeeded: 40,
222+
Errored: 20,
223+
InProgress: 20,
224+
Issued: 80,
225+
},
226+
205227
LoadStats: map[string]ServerLoadData{
206228
"net": {Count: 40, Sum: 40},
207229
"disk": {Count: 40, Sum: 80},
@@ -298,7 +320,7 @@ func TestStoreStats(t *testing.T) {
298320
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
299321
LocalityStats: map[string]LocalityData{
300322
"test-locality": {
301-
RequestStats: RequestData{Succeeded: 1},
323+
RequestStats: RequestData{Succeeded: 1, Issued: 1},
302324
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
303325
},
304326
},
@@ -308,7 +330,7 @@ func TestStoreStats(t *testing.T) {
308330
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
309331
LocalityStats: map[string]LocalityData{
310332
"test-locality": {
311-
RequestStats: RequestData{Succeeded: 1},
333+
RequestStats: RequestData{Succeeded: 1, Issued: 1},
312334
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
313335
},
314336
},
@@ -327,7 +349,7 @@ func TestStoreStats(t *testing.T) {
327349
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
328350
LocalityStats: map[string]LocalityData{
329351
"test-locality": {
330-
RequestStats: RequestData{Succeeded: 1},
352+
RequestStats: RequestData{Succeeded: 1, Issued: 1},
331353
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
332354
},
333355
},
@@ -337,7 +359,7 @@ func TestStoreStats(t *testing.T) {
337359
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
338360
LocalityStats: map[string]LocalityData{
339361
"test-locality": {
340-
RequestStats: RequestData{Succeeded: 1},
362+
RequestStats: RequestData{Succeeded: 1, Issued: 1},
341363
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
342364
},
343365
},
@@ -347,7 +369,7 @@ func TestStoreStats(t *testing.T) {
347369
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
348370
LocalityStats: map[string]LocalityData{
349371
"test-locality": {
350-
RequestStats: RequestData{Succeeded: 1},
372+
RequestStats: RequestData{Succeeded: 1, Issued: 1},
351373
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
352374
},
353375
},
@@ -357,7 +379,7 @@ func TestStoreStats(t *testing.T) {
357379
TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
358380
LocalityStats: map[string]LocalityData{
359381
"test-locality": {
360-
RequestStats: RequestData{Succeeded: 1},
382+
RequestStats: RequestData{Succeeded: 1, Issued: 1},
361383
LoadStats: map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
362384
},
363385
},
@@ -394,25 +416,25 @@ func TestStoreStatsEmptyDataNotReported(t *testing.T) {
394416
{
395417
Cluster: "c0", Service: "s0",
396418
LocalityStats: map[string]LocalityData{
397-
"test-locality": {RequestStats: RequestData{Succeeded: 1}},
419+
"test-locality": {RequestStats: RequestData{Succeeded: 1, Issued: 1}},
398420
},
399421
},
400422
{
401423
Cluster: "c0", Service: "s1",
402424
LocalityStats: map[string]LocalityData{
403-
"test-locality": {RequestStats: RequestData{Succeeded: 1}},
425+
"test-locality": {RequestStats: RequestData{Succeeded: 1, Issued: 1}},
404426
},
405427
},
406428
{
407429
Cluster: "c1", Service: "s0",
408430
LocalityStats: map[string]LocalityData{
409-
"test-locality": {RequestStats: RequestData{InProgress: 1}},
431+
"test-locality": {RequestStats: RequestData{InProgress: 1, Issued: 1}},
410432
},
411433
},
412434
{
413435
Cluster: "c1", Service: "s1",
414436
LocalityStats: map[string]LocalityData{
415-
"test-locality": {RequestStats: RequestData{InProgress: 1}},
437+
"test-locality": {RequestStats: RequestData{InProgress: 1, Issued: 1}},
416438
},
417439
},
418440
}

‎xds/internal/xdsclient/transport/loadreport.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -223,6 +223,7 @@ func (t *Transport) sendLoadStatsRequest(stream lrsStream, loads []*load.Data) e
223223
TotalSuccessfulRequests: localityData.RequestStats.Succeeded,
224224
TotalRequestsInProgress: localityData.RequestStats.InProgress,
225225
TotalErrorRequests: localityData.RequestStats.Errored,
226+
TotalIssuedRequests: localityData.RequestStats.Issued,
226227
LoadMetricStats: loadMetricStats,
227228
UpstreamEndpointStats: nil, // TODO: populate for per endpoint loads.
228229
})

‎xds/internal/xdsclient/transport/loadreport_test.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,12 +151,14 @@ func (s) TestReportLoad(t *testing.T) {
151151
// TotalMetricValue is the aggregation of 3.14 + 2.718 = 5.858
152152
{MetricName: testKey1, NumRequestsFinishedWithMetric: 2, TotalMetricValue: 5.858}},
153153
TotalSuccessfulRequests: 1,
154+
TotalIssuedRequests: 1,
154155
},
155156
{
156157
Locality: &v3corepb.Locality{Region: "test-region2"},
157158
LoadMetricStats: []*v3endpointpb.EndpointLoadMetricStats{
158159
{MetricName: testKey2, NumRequestsFinishedWithMetric: 1, TotalMetricValue: 1.618}},
159160
TotalSuccessfulRequests: 1,
161+
TotalIssuedRequests: 1,
160162
},
161163
},
162164
}

0 commit comments

Comments
 (0)