global sort: add metrics for merge sort stage (#60971)

close #61025

Author: fzzf678

pkg/lightning/backend/external/BUILD.bazel

@@ -45,6 +45,7 @@ go_library(
         "@com_github_jfcg_sorty_v2//:sorty",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
+        "@com_github_prometheus_client_golang//prometheus",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//tikv",
         "@org_golang_x_sync//errgroup",

pkg/lightning/backend/external/byte_reader.go

@@ -19,6 +19,7 @@ import (
 	goerrors "errors"
 	"fmt"
 	"io"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/pingcap/errors"
@@ -27,6 +28,7 @@ import (
 	"github.com/pingcap/tidb/pkg/lightning/membuf"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/pingcap/tidb/pkg/util/size"
+	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
 
@@ -67,6 +69,9 @@ type byteReader struct {
 	}
 
 	logger *zap.Logger
+	// monitor the speed of reading from external storage
+	readDurHist  prometheus.Observer
+	readRateHist prometheus.Observer
 }
 
 func openStoreReaderAndSeek(
@@ -273,6 +278,18 @@ func (r *byteReader) next(n int) (int, [][]byte) {
 }
 
 func (r *byteReader) reload() error {
+	if r.readDurHist != nil && r.readRateHist != nil {
+		startTime := time.Now()
+		defer func() {
+			readSecond := time.Since(startTime).Seconds()
+			size := 0
+			for _, b := range r.curBuf {
+				size += len(b)
+			}
+			r.readDurHist.Observe(readSecond)
+			r.readRateHist.Observe(float64(size) / 1024.0 / 1024.0 / readSecond)
+		}()
+	}
 	to := r.concurrentReader.expected
 	now := r.concurrentReader.now
 	// in read only false -> true is possible

pkg/lightning/backend/external/iter.go

@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/lightning/membuf"
+	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/pingcap/tidb/pkg/util/size"
 	"go.uber.org/zap"
@@ -534,6 +535,8 @@ func NewMergeKVIter(
 			if err != nil {
 				return nil, err
 			}
+			rd.byteReader.readDurHist = metrics.GlobalSortReadFromCloudStorageDuration.WithLabelValues("merge_sort_read")
+			rd.byteReader.readRateHist = metrics.GlobalSortReadFromCloudStorageRate.WithLabelValues("merge_sort_read")
 			rd.byteReader.enableConcurrentRead(
 				exStorage,
 				paths[i],

pkg/lightning/backend/external/onefile_writer.go

@@ -19,13 +19,15 @@ import (
 	"encoding/binary"
 	"path/filepath"
 	"slices"
+	"time"
 
 	"github.com/docker/go-units"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/ingestor/engineapi"
 	tidbkv "github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/lightning/membuf"
+	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/util/intest"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"go.uber.org/zap"
@@ -212,6 +214,7 @@ func (w *OneFileWriter) doWriteRow(ctx context.Context, idxKey, idxVal []byte) e
 		return err
 	}
 	// 1. encode data and write to kvStore.
+	writeStartTime := time.Now()
 	keyLen := len(idxKey)
 	length := len(idxKey) + len(idxVal) + lengthBytes*2
 	buf, _ := w.kvBuffer.AllocBytesWithSliceLocation(length)
@@ -241,6 +244,10 @@ func (w *OneFileWriter) doWriteRow(ctx context.Context, idxKey, idxVal []byte) e
 	}
 	w.totalCnt += 1
 	w.totalSize += uint64(keyLen + len(idxVal))
+	writeDuration := time.Since(writeStartTime)
+	metrics.GlobalSortWriteToCloudStorageDuration.WithLabelValues("merge_sort_write").Observe(writeDuration.Seconds())
+	metrics.GlobalSortWriteToCloudStorageRate.WithLabelValues("merge_sort_write").
+		Observe(float64(length) / 1024.0 / 1024.0 / writeDuration.Seconds())
 	return nil
 }