client: pay back WRU when the request fails (#5941)

ref #5851

When a write request fails, pay back the WRU cost we count before.

Signed-off-by: JmPotato <ghzpotato@gmail.com>

Co-authored-by: Ti Chi Robot <ti-community-prow-bot@tidb.io>

client/resource_manager/client/client.go

@@ -32,11 +32,11 @@ const (
 	maxNotificationChanLen = 200
 )
 
-// ResourceGroupKVInterceptor is used as quato limit controller for resource group using kv store.
+// ResourceGroupKVInterceptor is used as quota limit controller for resource group using kv store.
 type ResourceGroupKVInterceptor interface {
 	// OnRequestWait is used to check whether resource group has enough tokens. It maybe needs wait some time.
 	OnRequestWait(ctx context.Context, resourceGroupName string, info RequestInfo) error
-	// OnResponse is used to consume tokens atfer receiving response
+	// OnResponse is used to consume tokens after receiving response
 	OnResponse(ctx context.Context, resourceGroupName string, req RequestInfo, resp ResponseInfo) error
 }
 

client/resource_manager/client/model.go

@@ -37,6 +37,9 @@ type RequestInfo interface {
 type ResponseInfo interface {
 	ReadBytes() uint64
 	KVCPUMs() uint64
+	// Succeed is used to tell whether the request is successfully returned.
+	// If not, we need to pay back the WRU cost of the request.
+	Succeed() bool
 }
 
 // ResourceCalculator is used to calculate the resource consumption of a request.
@@ -72,9 +75,7 @@ func (kc *KVCalculator) BeforeKVRequest(consumption *rmpb.Consumption, req Reque
 	if req.IsWrite() {
 		consumption.KvWriteRpcCount += 1
 		// Write bytes are knowable in advance, so we can calculate the WRU cost here.
-		writeBytes := float64(req.WriteBytes())
-		consumption.WriteBytes += writeBytes
-		consumption.WRU += float64(kc.WriteBaseCost) + float64(kc.WriteBytesCost)*writeBytes
+		kc.calculateWriteCost(consumption, req)
 	} else {
 		consumption.KvReadRpcCount += 1
 		// Read bytes could not be known before the request is executed,
@@ -83,20 +84,44 @@ func (kc *KVCalculator) BeforeKVRequest(consumption *rmpb.Consumption, req Reque
 	}
 }
 
+func (kc *KVCalculator) calculateWriteCost(consumption *rmpb.Consumption, req RequestInfo) {
+	writeBytes := float64(req.WriteBytes())
+	consumption.WriteBytes += writeBytes
+	consumption.WRU += float64(kc.WriteBaseCost) + float64(kc.WriteBytesCost)*writeBytes
+}
+
 // AfterKVRequest ...
 func (kc *KVCalculator) AfterKVRequest(consumption *rmpb.Consumption, req RequestInfo, res ResponseInfo) {
-	// For now, we can only collect the KV CPU cost for a read request.
 	if !req.IsWrite() {
-		kvCPUMs := float64(res.KVCPUMs())
-		consumption.TotalCpuTimeMs += kvCPUMs
-		consumption.RRU += float64(kc.CPUMsCost) * kvCPUMs
+		// For now, we can only collect the KV CPU cost for a read request.
+		kc.calculateCPUCost(consumption, res)
+	} else if !res.Succeed() {
+		// If the write request is not successfully returned, we need to pay back the WRU cost.
+		kc.payBackWriteCost(consumption, req)
 	}
 	// A write request may also read data, which should be counted into the RRU cost.
+	// This part should be counted even if the request does not succeed.
+	kc.calculateReadCost(consumption, res)
+}
+
+func (kc *KVCalculator) calculateReadCost(consumption *rmpb.Consumption, res ResponseInfo) {
 	readBytes := float64(res.ReadBytes())
 	consumption.ReadBytes += readBytes
 	consumption.RRU += float64(kc.ReadBytesCost) * readBytes
 }
 
+func (kc *KVCalculator) calculateCPUCost(consumption *rmpb.Consumption, res ResponseInfo) {
+	kvCPUMs := float64(res.KVCPUMs())
+	consumption.TotalCpuTimeMs += kvCPUMs
+	consumption.RRU += float64(kc.CPUMsCost) * kvCPUMs
+}
+
+func (kc *KVCalculator) payBackWriteCost(consumption *rmpb.Consumption, req RequestInfo) {
+	writeBytes := float64(req.WriteBytes())
+	consumption.WriteBytes -= writeBytes
+	consumption.WRU -= float64(kc.WriteBaseCost) + float64(kc.WriteBytesCost)*writeBytes
+}
+
 // SQLCalculator is used to calculate the SQL-side consumption.
 type SQLCalculator struct {
 	*Config

tests/mcs/resource_manager/resource_manager_test.go

@@ -234,6 +234,7 @@ func (ti *testRequestInfo) WriteBytes() uint64 {
 type testResponseInfo struct {
 	cpuMs     uint64
 	readBytes uint64
+	succeed   bool
 }
 
 func (tri *testResponseInfo) ReadBytes() uint64 {
@@ -244,6 +245,10 @@ func (tri *testResponseInfo) KVCPUMs() uint64 {
 	return tri.cpuMs
 }
 
+func (tri *testResponseInfo) Succeed() bool {
+	return tri.succeed
+}
+
 type tokenConsumptionPerSecond struct {
 	rruTokensAtATime float64
 	wruTokensAtATime float64
@@ -276,6 +281,7 @@ func (t tokenConsumptionPerSecond) makeWriteResponse() *testResponseInfo {
 	return &testResponseInfo{
 		readBytes: 0,
 		cpuMs:     0,
+		succeed:   true,
 	}
 }