Report sample counts by type

Author: gnurizen

reporter/parca_reporter.go

@@ -19,6 +19,7 @@ import (
 	"path"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 	"unicode/utf8"
 
@@ -133,6 +134,12 @@ type ParcaReporter struct {
 	debuginfoUploadRequestBytes prometheus.Counter
 	emptySamples                prometheus.Counter
 
+	// Sample counters by type (reset each reporting interval)
+	cpuSamples    atomic.Uint64
+	gpuSamples    atomic.Uint64
+	offcpuSamples atomic.Uint64
+	memorySamples atomic.Uint64
+
 	offlineModeConfig *OfflineModeConfig
 
 	// Protects the log file,
@@ -260,9 +267,11 @@ func (r *ParcaReporter) ReportTraceEvent(trace *libpf.Trace,
 	case support.TraceOriginSampling:
 		writeSample(1, time.Second.Nanoseconds(), 1e9/int64(r.samplesPerSecond), "parca_agent", "samples", "count", "cpu", "nanoseconds")
 		r.sampleWriter.Temporality.AppendString("delta")
+		r.cpuSamples.Add(1)
 	case support.TraceOriginOffCPU:
 		writeSample(meta.OffTime, time.Second.Nanoseconds(), 1e9/int64(r.samplesPerSecond), "parca_agent", "wallclock", "nanoseconds", "samples", "count")
 		r.sampleWriter.Temporality.AppendString("delta")
+		r.offcpuSamples.Add(1)
 	case support.TraceOriginMemory:
 		// This shouldn't happen too much so an info log is fine, revisit when we do continuous memory profiling.
 		log.Infof("Received memory trace event for TID %d, PID %d, comm %s", meta.TID, meta.PID, meta.Comm)
@@ -287,9 +296,11 @@ func (r *ParcaReporter) ReportTraceEvent(trace *libpf.Trace,
 			r.sampleWriter.Temporality.AppendNull()
 			writeSample(int64(meta.AllocBytes), 0, memPeriod, "memory", "alloc_space", "bytes", "space", "bytes")
 		}
+		r.memorySamples.Add(1)
 	case support.TraceOriginCuda:
 		writeSample(meta.OffTime, time.Second.Nanoseconds(), 1e9/int64(r.samplesPerSecond), "parca_agent", "cuda", "nanoseconds", "cuda", "nanoseconds")
 		r.sampleWriter.Temporality.AppendString("delta")
+		r.gpuSamples.Add(1)
 	}
 
 	return nil
@@ -917,6 +928,14 @@ func (r *ParcaReporter) logDataForOfflineMode(ctx context.Context, buf *bytes.Bu
 		return nil
 	}
 
+	// Log and reset sample counters by type
+	cpu := r.cpuSamples.Swap(0)
+	gpu := r.gpuSamples.Swap(0)
+	offcpu := r.offcpuSamples.Swap(0)
+	mem := r.memorySamples.Swap(0)
+	log.Infof("[reporter] logging %d samples to offline storage [cpu=%d, gpu=%d, offcpu=%d, memory=%d]",
+		record.NumRows(), cpu, gpu, offcpu, mem)
+
 	buf.Reset()
 
 	w := ipc.NewWriter(buf,
@@ -1067,7 +1086,13 @@ func (r *ParcaReporter) reportDataToBackend(ctx context.Context, buf *bytes.Buff
 		return err
 	}
 
-	log.Infof("[reporter] sending %d samples (%d bytes) to server", record.NumRows(), buf.Len())
+	// Log and reset sample counters by type
+	cpu := r.cpuSamples.Swap(0)
+	gpu := r.gpuSamples.Swap(0)
+	offcpu := r.offcpuSamples.Swap(0)
+	mem := r.memorySamples.Swap(0)
+	log.Infof("[reporter] sending %d samples (%d bytes) to server [cpu=%d, gpu=%d, offcpu=%d, memory=%d]",
+		record.NumRows(), buf.Len(), cpu, gpu, offcpu, mem)
 
 	client, err := r.client.Write(ctx)
 	if err != nil {