Merge #149315 #150379

149315: cli(debug.zip): upload profiles from node subdirectories r=aa-joshi a=aa-joshi

Previously, debug zip upload to Datadog would only considers and uploads cpu and
heap profiles which are in node sub-directories. This was inadequate because we
have cpu and heap profiles as part of `nodes/*/cpuprof` & `nodes/*/heapprof`
directories which will be ignored during upload. To address this, this patch
considers profiles of `cpuprof` and `heapprof` directories during the Datadog
upload. This patch also updates the profile explorer link generated after
upload.

Epic: [CRDB-52093](https://cockroachlabs.atlassian.net/browse/CRDB-52093)
Part of: CRDB-51119
Release note: None

-----
PFA screenshot after changes:
<img width="1330" alt="Screenshot 2025-07-09 at 11 09 58 AM" src="https://github.com/user-attachments/assets/c220e519-d55c-4a7b-9c7d-499199317e76" />


150379: kvserver: record follower write bytes in replica load r=tbg a=wenyihu6


**kvserver: takes write and ingested bytes for recordRequestWriteBytes**

Previously, recordRequestWriteBytes took a *kvadmission.StoreWriteBytes, which
was unnecessary since we're only recording write and ingested bytes, not
anything admission control specific. This commit changes the function to accept
the raw byte counts directly. This also sets up a future change where follower
replicas would not pass in StoreWriteBytes during recording.

Epic: none
Release note: none

---

**kvserver: record follower write bytes in replica load**

Previously, write bytes for range load were only tracked on leaseholder replicas
during proposal, leaving write bytes written to follower replicas unaccounted
for. This commit updates the tracking to include write bytes on follower
replicas as well.

Epic: none
Release note: none

---

**kvserver: record follower write bytes during recordStatsOnCommit**

Previously, we recorded write bytes for follower replicas in
runPostAddTriggersReplicaOnly, right before apply, which could be inaccurate
since stats should be recorded at commit time. This commit moves the recording
to recordStatsOnCommit by including the write bytes in the batch stats.

Epic: none
Release note: none

---

**kvserver: rename numWriteBytes to numWriteAndIngestedBytes**

This commit renames numWriteBytes in appBatchStats to numWriteAndIngestedBytes
for accuracy, as it accounts for both bytes written to the replica and bytes
ingested via SSTables.

Epic: none
Release note: none


Co-authored-by: Akshay Joshi <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>

Author: 3 people

pkg/cli/testdata/upload/profiles

@@ -14,7 +14,8 @@ upload-profiles
 === uploading profiles
 Upload ID: abc-20241114000000
 debug zip upload debugDir --dd-api-key=dd-api-key --dd-app-key=dd-app-key --cluster=ABC --include=profiles
-{"start":"","end":"","attachments":["cpu.pprof","heap.pprof"],"tags_profiler":"cluster:ABC,env:debug,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["cpu.pprof"],"tags_profiler":"cluster:ABC,env:debug,file_name:cpu.pprof,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["heap.pprof"],"tags_profiler":"cluster:ABC,env:debug,file_name:heap.pprof,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
 
 
 # Multi-node - both profiles
@@ -39,9 +40,10 @@ upload-profiles tags=foo:bar
 === uploading profiles
 Upload ID: abc-20241114000000
 debug zip upload debugDir --dd-api-key=dd-api-key --dd-app-key=dd-app-key --tags=foo:bar --cluster=ABC --include=profiles
-{"start":"","end":"","attachments":["cpu.pprof","heap.pprof"],"tags_profiler":"cluster:ABC,env:debug,foo:bar,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
-{"start":"","end":"","attachments":["cpu.pprof","heap.pprof"],"tags_profiler":"cluster:ABC,env:debug,foo:bar,node_id:2,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
-
+{"start":"","end":"","attachments":["cpu.pprof"],"tags_profiler":"cluster:ABC,env:debug,file_name:cpu.pprof,foo:bar,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["cpu.pprof"],"tags_profiler":"cluster:ABC,env:debug,file_name:cpu.pprof,foo:bar,node_id:2,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["heap.pprof"],"tags_profiler":"cluster:ABC,env:debug,file_name:heap.pprof,foo:bar,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["heap.pprof"],"tags_profiler":"cluster:ABC,env:debug,file_name:heap.pprof,foo:bar,node_id:2,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
 
 # Single-node - only CPU profile
 upload-profiles tags=customer:user-given-name,cluster:XYZ
@@ -58,7 +60,7 @@ upload-profiles tags=customer:user-given-name,cluster:XYZ
 === uploading profiles
 Upload ID: abc-20241114000000
 debug zip upload debugDir --dd-api-key=dd-api-key --dd-app-key=dd-app-key --tags=customer:user-given-name,cluster:XYZ --cluster=ABC --include=profiles
-{"start":"","end":"","attachments":["cpu.pprof"],"tags_profiler":"cluster:XYZ,customer:user-given-name,env:debug,foo:bar,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["cpu.pprof"],"tags_profiler":"cluster:XYZ,customer:user-given-name,env:debug,file_name:cpu.pprof,foo:bar,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
 
 
 # Single-node - no profiles found
@@ -90,10 +92,11 @@ upload-profiles tags=env:SH
 === uploading profiles
 Upload ID: abc-20241114000000
 debug zip upload debugDir --dd-api-key=dd-api-key --dd-app-key=dd-app-key --tags=env:SH --cluster=ABC --include=profiles
-{"start":"","end":"","attachments":["cpu.pprof","heap.pprof"],"tags_profiler":"cluster:ABC,env:SH,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["cpu.pprof"],"tags_profiler":"cluster:ABC,env:SH,file_name:cpu.pprof,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
+{"start":"","end":"","attachments":["heap.pprof"],"tags_profiler":"cluster:ABC,env:SH,file_name:heap.pprof,node_id:1,service:CRDB-SH,source:cockroachdb,upload_id:abc-20241114000000","family":"go","version":"4"}
 
 
-# Single-node - both profiles
+# Single-node - both profiles error
 upload-profiles tags=ERR
 {
   "nodes": {
@@ -107,7 +110,7 @@ upload-profiles tags=ERR
 }
 ----
 === uploading profiles
-Failed to upload profiles: failed to upload profiles of node 1: status: 400, body: 'runtime' is a required field
+Failed to upload profiles: failed to upload profiles to Datadog
 Upload ID: abc-20241114000000
 debug zip upload debugDir --dd-api-key=dd-api-key --dd-app-key=dd-app-key --tags=ERR --cluster=ABC --include=profiles
 

pkg/cli/zip_upload.go

@@ -71,6 +71,8 @@ const (
 
 	// the path pattern to search for specific artifacts in the debug zip directory
 	zippedProfilePattern        = "nodes/*/*.pprof"
+	zippedCPUProfilePattern     = "nodes/*/cpuprof/*.pprof"
+	zippedHeapProfilePattern    = "nodes/*/heapprof/*.pprof"
 	zippedLogsPattern           = "nodes/*/logs/*"
 	zippedNodeTableDumpsPattern = "nodes/*/*.txt"
 
@@ -85,6 +87,7 @@ const (
 	clusterTag  = "cluster"
 	ddTagsTag   = "ddtags"
 	tableTag    = "table"
+	fileNameTag = "file_name"
 
 	// datadog endpoint URLs
 	datadogProfileUploadURLTmpl = "https://intake.profile.%s/api/v2/profile"
@@ -399,61 +402,147 @@ func validateZipUploadReadiness() error {
 	return nil
 }
 
+// profilePathInfo holds the information about a profile file to be uploaded
+// in Datadog. This is used to pass the information to the upload workers
+// through upload channel.
+type profilePathInfo struct {
+	nodeID   string
+	filepath string
+}
+
 func uploadZipProfiles(ctx context.Context, uploadID string, debugDirPath string) error {
-	paths, err := expandPatterns([]string{path.Join(debugDirPath, zippedProfilePattern)})
+
+	paths, err := expandPatterns([]string{
+		path.Join(debugDirPath, zippedProfilePattern),
+		path.Join(debugDirPath, zippedCPUProfilePattern),
+		path.Join(debugDirPath, zippedHeapProfilePattern)})
+
 	if err != nil {
 		return err
 	}
 
+	if len(paths) == 0 {
+		return nil
+	}
+
+	var (
+		noOfWorkers        = min(debugZipUploadOpts.maxConcurrentUploads, len(paths))
+		uploadChan         = make(chan profilePathInfo, noOfWorkers*2) // 2x the number of workers to keep them busy
+		uploadWG           = sync.WaitGroup{}
+		profileUploadState struct {
+			syncutil.Mutex
+			isSingleUploadSucceeded bool
+		}
+		// regex to match the profile directories. This is used to extract the node ID.
+		reProfileDirectories = regexp.MustCompile(`.*(heapprof|cpuprof).*\.pprof$`)
+	)
+
+	markSuccessOnce := sync.OnceFunc(func() {
+		profileUploadState.isSingleUploadSucceeded = true
+	})
+
 	pathsByNode := make(map[string][]string)
+	maxProfilesOfNode := 0
 	for _, path := range paths {
-		nodeID := filepath.Base(filepath.Dir(path))
+		// extract the node ID from the zippedProfilePattern. If it does not match the
+		// nodeID (integer) then we assume the path is from zippedCPUProfilePattern
+		// and zippedHeapProfilePattern and try to extract the node ID from the suffix.
+		var nodeID = ""
+		if reProfileDirectories.MatchString(path) {
+			nodeID = filepath.Base(filepath.Dir(filepath.Dir(path)))
+		} else {
+			nodeID = filepath.Base(filepath.Dir(path))
+		}
+
 		if _, ok := pathsByNode[nodeID]; !ok {
 			pathsByNode[nodeID] = []string{}
 		}
 
 		pathsByNode[nodeID] = append(pathsByNode[nodeID], path)
+		maxProfilesOfNode = max(maxProfilesOfNode, len(pathsByNode[nodeID]))
 	}
 
-	retryOpts := base.DefaultRetryOptions()
-	retryOpts.MaxRetries = zipUploadRetries
-	var req *http.Request
-	for nodeID, paths := range pathsByNode {
-		for retry := retry.Start(retryOpts); retry.Next(); {
-			req, err = newProfileUploadReq(
-				ctx, paths, appendUserTags(
-					append(
-						defaultDDTags, makeDDTag(nodeIDTag, nodeID), makeDDTag(uploadIDTag, uploadID),
-						makeDDTag(clusterTag, debugZipUploadOpts.clusterName),
-					), // system generated tags
-					debugZipUploadOpts.tags..., // user provided tags
-				),
-			)
-			if err != nil {
-				continue
-			}
+	// start the upload pool
+	noOfWorkers = min(noOfWorkers, maxProfilesOfNode)
+	for i := 0; i < noOfWorkers; i++ {
+		go func() {
+			for pathInfo := range uploadChan {
+				profilePath := pathInfo.filepath
+				nodeID := pathInfo.nodeID
 
-			if _, err = doUploadReq(req); err == nil {
-				break
+				func() {
+					defer uploadWG.Done()
+					fileName, err := uploadProfile(profilePath, ctx, nodeID, uploadID)
+					if err != nil {
+						fmt.Fprintf(os.Stderr, "failed to upload profile %s of node %s: %s\n", fileName, nodeID, err)
+						return
+					}
+					markSuccessOnce()
+				}()
 			}
-		}
+		}()
+	}
 
-		if err != nil {
-			return fmt.Errorf("failed to upload profiles of node %s: %w", nodeID, err)
+	for nodeID, paths := range pathsByNode {
+		for _, path := range paths {
+			uploadWG.Add(1)
+			uploadChan <- profilePathInfo{nodeID: nodeID, filepath: path}
 		}
 
-		fmt.Fprintf(os.Stderr, "Uploaded profiles of node %s to datadog (%s)\n", nodeID, strings.Join(paths, ", "))
-		fmt.Fprintf(os.Stderr, "Explore the profiles on datadog: "+
-			"https://%s/profiling/explorer?query=%s:%s\n", ddSiteToHostMap[debugZipUploadOpts.ddSite],
-			uploadIDTag, uploadID,
-		)
+		uploadWG.Wait()
+		fmt.Fprintf(os.Stderr, "Uploaded profiles of node %s to datadog\n", nodeID)
+	}
+
+	uploadWG.Wait()
+	close(uploadChan)
+
+	if !profileUploadState.isSingleUploadSucceeded {
+		return errors.Newf("failed to upload profiles to Datadog")
 	}
 
+	toUnixTimestamp := getCurrentTime().UnixMilli()
+	//create timestamp for T-30 days.
+	fromUnixTimestamp := toUnixTimestamp - (30 * 24 * 60 * 60 * 1000)
+
+	fmt.Fprintf(os.Stderr, "Explore the profiles on datadog: "+
+		"https://%s/profiling/explorer?query=%s:%s&viz=stream&from_ts=%d&to_ts=%d&live=false\n", ddSiteToHostMap[debugZipUploadOpts.ddSite],
+		uploadIDTag, uploadID, fromUnixTimestamp, toUnixTimestamp,
+	)
+
 	return nil
 }
 
+func uploadProfile(
+	profilePath string, ctx context.Context, nodeID string, uploadID string,
+) (string, error) {
+	fileName := filepath.Base(profilePath)
+
+	req, err := newProfileUploadReq(
+		ctx, profilePath, appendUserTags(
+			append(
+				defaultDDTags, makeDDTag(nodeIDTag, nodeID), makeDDTag(uploadIDTag, uploadID),
+				makeDDTag(clusterTag, debugZipUploadOpts.clusterName), makeDDTag(fileNameTag, fileName),
+			), // system generated tags
+			debugZipUploadOpts.tags..., // user provided tags
+		),
+	)
+
+	retryOpts := base.DefaultRetryOptions()
+	retryOpts.MaxRetries = zipUploadRetries
+	for retry := retry.Start(retryOpts); retry.Next(); {
+		if err != nil {
+			continue
+		}
+
+		if _, err = doUploadReq(req); err == nil {
+			break
+		}
+	}
+	return fileName, err
+}
+
 func newProfileUploadReq(
-	ctx context.Context, profilePaths []string, tags []string,
+	ctx context.Context, profilePath string, tags []string,
 ) (*http.Request, error) {
 	var (
 		body  bytes.Buffer
@@ -473,26 +562,36 @@ func newProfileUploadReq(
 		}
 	)
 
-	for _, profilePath := range profilePaths {
-		fileName := filepath.Base(profilePath)
-		event.Attachments = append(event.Attachments, fileName)
+	fileName := filepath.Base(profilePath)
 
-		f, err := mw.CreateFormFile(fileName, fileName)
-		if err != nil {
-			return nil, err
-		}
+	// Datadog only accepts CPU and heap profiles with filename as "cpu.pprof" or "heap.pprof".
+	// The cpu profile files has "cpu" in the filename prefix and heap profile files
+	// has "memprof/heap" in the filename prefix. Hence we are renaming the files accordingly
+	// so that Datadog can recognize and accept them correctly.
+	if strings.HasPrefix(fileName, "cpu") {
+		fileName = "cpu.pprof"
+	} else {
+		// If the file is not a CPU profile, we assume it is a heap/memory profile.
+		fileName = "heap.pprof"
+	}
 
-		data, err := os.ReadFile(profilePath)
-		if err != nil {
-			return nil, err
-		}
+	event.Attachments = append(event.Attachments, fileName)
 
-		if _, err := f.Write(data); err != nil {
-			return nil, err
-		}
+	f, err := mw.CreateFormFile(fileName, fileName)
+	if err != nil {
+		return nil, err
+	}
+
+	data, err := os.ReadFile(profilePath)
+	if err != nil {
+		return nil, err
+	}
+
+	if _, err := f.Write(data); err != nil {
+		return nil, err
 	}
 
-	f, err := mw.CreatePart(textproto.MIMEHeader{
+	f, err = mw.CreatePart(textproto.MIMEHeader{
 		httputil.ContentDispositionHeader: []string{`form-data; name="event"; filename="event.json"`},
 		httputil.ContentTypeHeader:        []string{httputil.JSONContentType},
 	})

pkg/kv/kvserver/app_batch.go

@@ -33,6 +33,9 @@ type appBatchStats struct {
 	numEntriesProcessedBytes   int64
 	numEmptyEntries            int
 	numAddSST, numAddSSTCopies int
+	// numWriteAndIngestedBytes is the sum of number of bytes written to the replica and size
+	// of the sstable to ingest.
+	numWriteAndIngestedBytes int64
 
 	// NB: update `merge` when adding a new field.
 }
@@ -42,6 +45,7 @@ func (s *appBatchStats) merge(ss appBatchStats) {
 	s.numEntriesProcessed += ss.numEntriesProcessed
 	s.numEntriesProcessedBytes += ss.numEntriesProcessedBytes
 	ss.numEmptyEntries += ss.numEmptyEntries
+	s.numWriteAndIngestedBytes += ss.numWriteAndIngestedBytes
 }
 
 // appBatch is the in-progress foundation for standalone log entry

pkg/kv/kvserver/load/replica_load.go

@@ -65,7 +65,11 @@ type ReplicaLoadStats struct {
 	// follower and leaseholder reads.
 	ReadKeysPerSecond float64
 	// WriteBytesPerSecond is the replica's average bytes written per second. A
-	// "Write" is as described in WritesPerSecond.
+	// "Write" is as described in WritesPerSecond. If the replica is a leaseholder,
+	// this is recorded as the bytes that will be written by the replica during the
+	// application of the Raft command including write bytes and ingested bytes for
+	// AddSSTable requests. If the replica is a follower, this is recorded right
+	// before a command is applied to the state machine.
 	WriteBytesPerSecond float64
 	// ReadBytesPerSecond is the replica's average bytes read per second. A "Read" is as
 	// described in ReadsPerSecond.

pkg/kv/kvserver/replica_app_batch.go

@@ -253,16 +253,20 @@ func (b *replicaAppBatch) runPostAddTriggersReplicaOnly(
 	// We don't track these stats in standalone log application since they depend
 	// on whether the proposer is still waiting locally, and this concept does not
 	// apply in a standalone context.
-	//
-	// TODO(irfansharif): This code block can be removed once below-raft
-	// admission control is the only form of IO admission control. It pre-dates
-	// it -- these stats were previously used to deduct IO tokens for follower
-	// writes/ingests without waiting.
-	if !cmd.IsLocal() && !cmd.ApplyAdmissionControl() {
+	if !cmd.IsLocal() {
 		writeBytes, ingestedBytes := cmd.getStoreWriteByteSizes()
-		b.followerStoreWriteBytes.NumEntries++
-		b.followerStoreWriteBytes.WriteBytes += writeBytes
-		b.followerStoreWriteBytes.IngestedBytes += ingestedBytes
+		if writeBytes > 0 || ingestedBytes > 0 {
+			b.ab.numWriteAndIngestedBytes += writeBytes + ingestedBytes
+		}
+		// TODO(irfansharif): This code block can be removed once below-raft
+		// admission control is the only form of IO admission control. It pre-dates
+		// it -- these stats were previously used to deduct IO tokens for follower
+		// writes/ingests without waiting.
+		if !cmd.ApplyAdmissionControl() {
+			b.followerStoreWriteBytes.NumEntries++
+			b.followerStoreWriteBytes.WriteBytes += writeBytes
+			b.followerStoreWriteBytes.IngestedBytes += ingestedBytes
+		}
 	}
 
 	// MVCC history mutations violate the closed timestamp, modifying data that
@@ -721,6 +725,7 @@ func (b *replicaAppBatch) recordStatsOnCommit() {
 	b.applyStats.appBatchStats.merge(b.ab.appBatchStats)
 	b.applyStats.numBatchesProcessed++
 	b.applyStats.followerStoreWriteBytes.Merge(b.followerStoreWriteBytes)
+	b.r.recordRequestWriteBytes(b.ab.numWriteAndIngestedBytes)
 
 	if n := b.ab.numAddSST; n > 0 {
 		b.r.store.metrics.AddSSTableApplications.Inc(int64(n))

pkg/kv/kvserver/replica_raft.go

@@ -96,6 +96,8 @@ var ReplicaLeaderlessUnavailableThreshold = settings.RegisterDurationSettingWith
 //     terminate execution, although it is given no guarantee that the proposal
 //     won't still go on to commit and apply at some later time.
 //   - the proposal's ID.
+//   - the bytes that will be written by the replica during the application of
+//     the Raft command.
 //   - any error obtained during the creation or proposal of the command, in
 //     which case the other returned values are zero.
 func (r *Replica) evalAndPropose(
@@ -821,6 +823,9 @@ func (s handleRaftReadyStats) SafeFormat(p redact.SafePrinter, _ rune) {
 			p.Printf(" (copies=%d)", c)
 		}
 	}
+	if b := s.apply.numWriteAndIngestedBytes; b > 0 {
+		p.Printf(", apply-write-bytes=%s", humanizeutil.IBytes(b))
+	}
 	p.SafeString("]")
 
 	if n := s.apply.assertionsRequested; n > 0 {