Flush the data before updating the report

Author: Haoning-Sun

cli_flags.go

@@ -17,15 +17,16 @@ import (
 
 const (
 	// Default values for CLI flags
-	defaultArgSamplesPerSecond    = 20
-	defaultArgReporterInterval    = 5.0 * time.Second
-	defaultArgMonitorInterval     = 5.0 * time.Second
-	defaultClockSyncInterval      = 3 * time.Minute
-	defaultProbabilisticThreshold = tracer.ProbabilisticThresholdMax
-	defaultProbabilisticInterval  = 1 * time.Minute
-	defaultArgSendErrorFrames     = false
-	defaultOffCPUThreshold        = 0
-	defaultEnvVarsValue           = ""
+	defaultArgSamplesPerSecond     = 20
+	defaultArgMaxSamplesPerSecond  = 0 // 0 means disabled
+	defaultArgReporterInterval     = 5.0 * time.Second
+	defaultArgMonitorInterval      = 5.0 * time.Second
+	defaultClockSyncInterval       = 3 * time.Minute
+	defaultProbabilisticThreshold  = tracer.ProbabilisticThresholdMax
+	defaultProbabilisticInterval   = 1 * time.Minute
+	defaultArgSendErrorFrames      = false
+	defaultOffCPUThreshold         = 0
+	defaultEnvVarsValue            = ""
 
 	// This is the X in 2^(n + x) where n is the default hardcoded map size value
 	defaultArgMapScaleFactor = 0

internal/controller/controller.go

@@ -168,15 +168,14 @@ func (c *Controller) UpdateSamplingFrequency(newSamplesPerSecond int) error {
 		return fmt.Errorf("tracer is not initialized")
 	}
 
-	// Update the tracer's sampling frequency
-	if err := c.tracer.UpdateSamplingFrequency(newSamplesPerSecond); err != nil {
-		return fmt.Errorf("failed to update tracer sampling frequency: %w", err)
+	if c.reporter != nil {
+		if err := c.reporter.UpdateSamplingFrequency(newSamplesPerSecond); err != nil {
+			log.Warnf("Failed to update reporter sampling frequency: %v", err)
+		}
 	}
 
-	// Update the reporter's sampling frequency for profile metadata
-	if c.reporter != nil {
-		c.reporter.UpdateSamplingFrequency(newSamplesPerSecond)
-		log.Infof("Updated reporter sampling frequency to %d Hz", newSamplesPerSecond)
+	if err := c.tracer.UpdateSamplingFrequency(newSamplesPerSecond); err != nil {
+		return fmt.Errorf("failed to update tracer sampling frequency: %w", err)
 	}
 
 	return nil

reporter/base_reporter.go

@@ -41,8 +41,10 @@ func (b *baseReporter) Stop() {
 	b.runLoop.Stop()
 }
 
-func (b *baseReporter) UpdateSamplingFrequency(samplesPerSecond int) {
+// UpdateSamplingFrequency updates the sampling frequency used in profile generation.
+func (b *baseReporter) UpdateSamplingFrequency(samplesPerSecond int) error {
 	b.pdata.UpdateSamplingFrequency(samplesPerSecond)
+	return nil
 }
 
 func (b *baseReporter) ReportTraceEvent(trace *libpf.Trace, meta *samples.TraceEventMeta) error {

reporter/collector_reporter.go

@@ -5,6 +5,7 @@ package reporter // import "go.opentelemetry.io/ebpf-profiler/reporter"
 
 import (
 	"context"
+	"fmt"
 
 	log "github.com/sirupsen/logrus"
 	"go.opentelemetry.io/collector/consumer/xconsumer"
@@ -97,3 +98,16 @@ func (r *CollectorReporter) reportProfile(ctx context.Context) error {
 
 	return r.nextConsumer.ConsumeProfiles(ctx, profiles)
 }
+
+// Flush immediately reports all currently accumulated trace events.
+func (r *CollectorReporter) Flush(ctx context.Context) error {
+	return r.reportProfile(ctx)
+}
+
+// UpdateSamplingFrequency updates the sampling frequency used in profile generation.
+func (r *CollectorReporter) UpdateSamplingFrequency(samplesPerSecond int) error {
+	if err := r.Flush(context.Background()); err != nil {
+		return fmt.Errorf("failed to flush reporter before updating sampling frequency: %w", err)
+	}
+	return r.baseReporter.UpdateSamplingFrequency(samplesPerSecond)
+}

reporter/iface.go

@@ -26,7 +26,10 @@ type Reporter interface {
 	Stop()
 
 	// UpdateSamplingFrequency updates the sampling frequency used in profile generation.
-	UpdateSamplingFrequency(samplesPerSecond int)
+	UpdateSamplingFrequency(samplesPerSecond int) error
+
+	// Flush immediately reports all currently accumulated trace events.
+	Flush(ctx context.Context) error
 }
 
 type TraceReporter interface {

reporter/otlp_reporter.go

@@ -6,6 +6,7 @@ package reporter // import "go.opentelemetry.io/ebpf-profiler/reporter"
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -134,6 +135,20 @@ func (r *OTLPReporter) reportOTLPProfile(ctx context.Context) error {
 	return err
 }
 
+// Flush immediately reports all currently accumulated trace events.
+func (r *OTLPReporter) Flush(ctx context.Context) error {
+	return r.reportOTLPProfile(ctx)
+}
+
+// UpdateSamplingFrequency updates the sampling frequency used in profile generation.
+// It first flushes accumulated data to ensure data isolation, then updates the frequency.
+func (r *OTLPReporter) UpdateSamplingFrequency(samplesPerSecond int) error {
+	if err := r.Flush(context.Background()); err != nil {
+		return fmt.Errorf("failed to flush reporter before updating sampling frequency: %w", err)
+	}
+	return r.baseReporter.UpdateSamplingFrequency(samplesPerSecond)
+}
+
 // waitGrpcEndpoint waits until the gRPC connection is established.
 func waitGrpcEndpoint(ctx context.Context, cfg *Config) (*grpc.ClientConn, error) {
 	// Sleep with a fixed backoff time added of +/- 20% jitter

tracer/events.go

@@ -140,7 +140,6 @@ func (t *Tracer) startTraceEventMonitor(ctx context.Context,
 	if t.maxSamplesPerSecond > t.samplesPerSecond {
 		effectiveMaxSPS = t.maxSamplesPerSecond
 	}
-	}
 	eventReader, err := perf.NewReader(eventsMap,
 		effectiveMaxSPS*support.Sizeof_Trace)
 	if err != nil {