Merge pull request #14 from florianl/off-cpu

Off cpu

cli_flags.go

@@ -12,6 +12,7 @@ import (
 	"github.com/peterbourgon/ff/v3"
 
 	"go.opentelemetry.io/ebpf-profiler/internal/controller"
+	"go.opentelemetry.io/ebpf-profiler/support"
 	"go.opentelemetry.io/ebpf-profiler/tracer"
 )
 
@@ -24,6 +25,7 @@ const (
 	defaultProbabilisticThreshold = tracer.ProbabilisticThresholdMax
 	defaultProbabilisticInterval  = 1 * time.Minute
 	defaultArgSendErrorFrames     = false
+	defaultOffCPUThreshold        = support.OffCPUThresholdMax
 
 	// This is the X in 2^(n + x) where n is the default hardcoded map size value
 	defaultArgMapScaleFactor = 0
@@ -61,6 +63,11 @@ var (
 		"If zero, monotonic-realtime clock sync will be performed once, " +
 		"on agent startup, but not periodically."
 	sendErrorFramesHelp = "Send error frames (devfiler only, breaks Kibana)"
+	offCPUThresholdHelp = fmt.Sprintf("If set to a value between 1 and %d will enable "+
+		"off-cpu profiling: Every time an off-cpu entry point is hit, a random number between "+
+		"0 and %d is chosen. If the given threshold is greater than this random number, the "+
+		"off-cpu trace is collected and reported.",
+		support.OffCPUThresholdMax-1, support.OffCPUThresholdMax-1)
 )
 
 // Package-scope variable, so that conditionally compiled other components can refer
@@ -114,6 +121,9 @@ func parseArgs() (*controller.Config, error) {
 	fs.BoolVar(&args.VerboseMode, "verbose", false, verboseModeHelp)
 	fs.BoolVar(&args.Version, "version", false, versionHelp)
 
+	fs.UintVar(&args.OffCPUThreshold, "off-cpu-threshold",
+		defaultOffCPUThreshold, offCPUThresholdHelp)
+
 	fs.Usage = func() {
 		fs.PrintDefaults()
 	}

host/host.go

@@ -54,6 +54,8 @@ type Trace struct {
 	KTime            times.KTime
 	PID              libpf.PID
 	TID              libpf.PID
+	Origin           libpf.Origin
+	OffTime          uint64 // Time a task was off-cpu in nanoseconds.
 	APMTraceID       libpf.APMTraceID
 	APMTransactionID libpf.APMTransactionID
 	CPU              int

internal/controller/config.go

@@ -33,7 +33,8 @@ type Config struct {
 	// HostName is the name of the host.
 	HostName string
 	// IPAddress is the IP address of the host that sends data to CollAgentAddr.
-	IPAddress string
+	IPAddress       string
+	OffCPUThreshold uint
 
 	Reporter reporter.Reporter
 

internal/controller/controller.go

@@ -13,6 +13,7 @@ import (
 	"go.opentelemetry.io/ebpf-profiler/internal/helpers"
 	"go.opentelemetry.io/ebpf-profiler/metrics"
 	"go.opentelemetry.io/ebpf-profiler/reporter"
+	"go.opentelemetry.io/ebpf-profiler/support"
 	"go.opentelemetry.io/ebpf-profiler/times"
 	"go.opentelemetry.io/ebpf-profiler/tracehandler"
 	"go.opentelemetry.io/ebpf-profiler/tracer"
@@ -124,6 +125,13 @@ func (c *Controller) Start(ctx context.Context) error {
 	}
 	log.Info("Attached tracer program")
 
+	if c.config.OffCPUThreshold < support.OffCPUThresholdMax {
+		if err := trc.StartOffCPUProfiling(); err != nil {
+			return fmt.Errorf("failed to start off-cpu profiling: %v", err)
+		}
+		log.Printf("Enabled off-cpu profiling")
+	}
+
 	if c.config.ProbabilisticThreshold < tracer.ProbabilisticThresholdMax {
 		trc.StartProbabilisticProfiling(ctx)
 		log.Printf("Enabled probabilistic profiling")

libpf/libpf.go

@@ -60,3 +60,6 @@ type Void struct{}
 // source line numbers associated with offsets in native code, or for source line numbers in
 // interpreted code.
 type SourceLineno uint64
+
+// Origin determines the source of a trace.
+type Origin int

libpf/symbol.go

@@ -6,6 +6,7 @@ package libpf // import "go.opentelemetry.io/ebpf-profiler/libpf"
 import (
 	"fmt"
 	"sort"
+	"strings"
 )
 
 // SymbolValue represents the value associated with a symbol, e.g. either an
@@ -81,6 +82,17 @@ func (symmap *SymbolMap) LookupSymbol(symbolName SymbolName) (*Symbol, error) {
 	return nil, fmt.Errorf("symbol %v not present in map", symbolName)
 }
 
+// LookupSymbolByPrefix loops over all known symbols and returns the first symbol
+// that starts with the given prefix.
+func (symmap *SymbolMap) LookupSymbolByPrefix(prefix string) (*Symbol, error) {
+	for name, sym := range symmap.nameToSymbol {
+		if strings.HasPrefix(string(name), prefix) {
+			return sym, nil
+		}
+	}
+	return nil, fmt.Errorf("no symbol present that starts with '%s'", prefix)
+}
+
 // LookupSymbolAddress returns the address of a symbol.
 // Returns SymbolValueInvalid and error if not found.
 func (symmap *SymbolMap) LookupSymbolAddress(symbolName SymbolName) (SymbolValue, error) {

reporter/base_reporter.go

@@ -13,6 +13,7 @@ import (
 	"go.opentelemetry.io/ebpf-profiler/libpf/xsync"
 	"go.opentelemetry.io/ebpf-profiler/reporter/internal/pdata"
 	"go.opentelemetry.io/ebpf-profiler/reporter/internal/samples"
+	"go.opentelemetry.io/ebpf-profiler/support"
 )
 
 // baseReporter encapsulates shared behavior between all the available reporters.
@@ -35,7 +36,7 @@ type baseReporter struct {
 	cgroupv2ID *lru.SyncedLRU[libpf.PID, string]
 
 	// traceEvents stores reported trace events (trace metadata with frames and counts)
-	traceEvents xsync.RWMutex[map[samples.TraceAndMetaKey]*samples.TraceEvents]
+	traceEvents xsync.RWMutex[map[libpf.Origin]samples.KeyToEventMapping]
 
 	// hostmetadata stores metadata that is sent out with every request.
 	hostmetadata *lru.SyncedLRU[string, string]
@@ -97,8 +98,11 @@ func (*baseReporter) ReportMetrics(_ uint32, _ []uint32, _ []int64) {}
 func (*baseReporter) SupportsReportTraceEvent() bool { return true }
 
 func (b *baseReporter) ReportTraceEvent(trace *libpf.Trace, meta *TraceEventMeta) {
-	traceEventsMap := b.traceEvents.WLock()
-	defer b.traceEvents.WUnlock(&traceEventsMap)
+	if meta.Origin != support.TraceOriginSampling && meta.Origin != support.TraceOriginOffCPU {
+		// At the moment only on-CPU and off-CPU traces are reported.
+		log.Errorf("Skip reporting trace for unexpected %d origin", meta.Origin)
+		return
+	}
 
 	var extraMeta any
 	if b.cfg.ExtraSampleAttrProd != nil {
@@ -121,20 +125,25 @@ func (b *baseReporter) ReportTraceEvent(trace *libpf.Trace, meta *TraceEventMeta
 		ExtraMeta:      extraMeta,
 	}
 
-	if events, exists := (*traceEventsMap)[key]; exists {
+	traceEventsMap := b.traceEvents.WLock()
+	defer b.traceEvents.WUnlock(&traceEventsMap)
+
+	if events, exists := (*traceEventsMap)[meta.Origin][key]; exists {
 		events.Timestamps = append(events.Timestamps, uint64(meta.Timestamp))
-		(*traceEventsMap)[key] = events
+		events.OffTimes = append(events.OffTimes, meta.OffTime)
+		(*traceEventsMap)[meta.Origin][key] = events
 		return
 	}
 
-	(*traceEventsMap)[key] = &samples.TraceEvents{
+	(*traceEventsMap)[meta.Origin][key] = &samples.TraceEvents{
 		Files:              trace.Files,
 		Linenos:            trace.Linenos,
 		FrameTypes:         trace.FrameTypes,
 		MappingStarts:      trace.MappingStart,
 		MappingEnds:        trace.MappingEnd,
 		MappingFileOffsets: trace.MappingFileOffsets,
 		Timestamps:         []uint64{uint64(meta.Timestamp)},
+		OffTimes:           []uint64{meta.OffTime},
 	}
 }
 

reporter/collector_reporter.go

@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/ebpf-profiler/libpf/xsync"
 	"go.opentelemetry.io/ebpf-profiler/reporter/internal/pdata"
 	"go.opentelemetry.io/ebpf-profiler/reporter/internal/samples"
+	"go.opentelemetry.io/ebpf-profiler/support"
 )
 
 // Assert that we implement the full Reporter interface.
@@ -59,16 +60,20 @@ func NewCollector(cfg *Config, nextConsumer consumerprofiles.Profiles) (*Collect
 		return nil, err
 	}
 
+	originsMap := make(map[libpf.Origin]samples.KeyToEventMapping, 2)
+	for _, origin := range []libpf.Origin{support.TraceOriginSampling,
+		support.TraceOriginOffCPU} {
+		originsMap[origin] = make(samples.KeyToEventMapping)
+	}
+
 	return &CollectorReporter{
 		baseReporter: &baseReporter{
-			cfg:        cfg,
-			name:       cfg.Name,
-			version:    cfg.Version,
-			pdata:      data,
-			cgroupv2ID: cgroupv2ID,
-			traceEvents: xsync.NewRWMutex(
-				map[samples.TraceAndMetaKey]*samples.TraceEvents{},
-			),
+			cfg:          cfg,
+			name:         cfg.Name,
+			version:      cfg.Version,
+			pdata:        data,
+			cgroupv2ID:   cgroupv2ID,
+			traceEvents:  xsync.NewRWMutex(originsMap),
 			hostmetadata: hostmetadata,
 			runLoop: &runLoop{
 				stopSignal: make(chan libpf.Void),

reporter/internal/pdata/generate.go

@@ -15,6 +15,7 @@ import (
 
 	"go.opentelemetry.io/ebpf-profiler/libpf"
 	"go.opentelemetry.io/ebpf-profiler/reporter/internal/samples"
+	"go.opentelemetry.io/ebpf-profiler/support"
 )
 
 const (
@@ -24,14 +25,16 @@ const (
 
 // Generate generates a pdata request out of internal profiles data, to be
 // exported.
-func (p Pdata) Generate(events map[samples.TraceAndMetaKey]*samples.TraceEvents) pprofile.Profiles {
+func (p Pdata) Generate(events map[libpf.Origin]samples.KeyToEventMapping) pprofile.Profiles {
 	profiles := pprofile.NewProfiles()
 	rp := profiles.ResourceProfiles().AppendEmpty()
 	sp := rp.ScopeProfiles().AppendEmpty()
-	prof := sp.Profiles().AppendEmpty()
-	prof.SetProfileID(pprofile.ProfileID(mkProfileID()))
-	p.setProfile(events, prof)
-
+	for _, origin := range []libpf.Origin{support.TraceOriginSampling,
+		support.TraceOriginOffCPU} {
+		prof := sp.Profiles().AppendEmpty()
+		prof.SetProfileID(pprofile.ProfileID(mkProfileID()))
+		p.setProfile(origin, events[origin], prof)
+	}
 	return profiles
 }
 
@@ -48,6 +51,7 @@ func mkProfileID() []byte {
 // setProfile sets the data an OTLP profile with all collected samples up to
 // this moment.
 func (p *Pdata) setProfile(
+	origin libpf.Origin,
 	events map[samples.TraceAndMetaKey]*samples.TraceEvents,
 	profile pprofile.Profile,
 ) {
@@ -62,13 +66,23 @@ func (p *Pdata) setProfile(
 	funcMap[samples.FuncInfo{Name: "", FileName: ""}] = 0
 
 	st := profile.SampleType().AppendEmpty()
-	st.SetTypeStrindex(getStringMapIndex(stringMap, "samples"))
-	st.SetUnitStrindex(getStringMapIndex(stringMap, "count"))
-
-	pt := profile.PeriodType()
-	pt.SetTypeStrindex(getStringMapIndex(stringMap, "cpu"))
-	pt.SetUnitStrindex(getStringMapIndex(stringMap, "nanoseconds"))
-	profile.SetPeriod(1e9 / int64(p.samplesPerSecond))
+	switch origin {
+	case support.TraceOriginSampling:
+		st.SetTypeStrindex(getStringMapIndex(stringMap, "samples"))
+		st.SetUnitStrindex(getStringMapIndex(stringMap, "count"))
+
+		pt := profile.PeriodType()
+		pt.SetTypeStrindex(getStringMapIndex(stringMap, "cpu"))
+		pt.SetUnitStrindex(getStringMapIndex(stringMap, "nanoseconds"))
+
+		profile.SetPeriod(1e9 / int64(p.samplesPerSecond))
+	case support.TraceOriginOffCPU:
+		st.SetTypeStrindex(getStringMapIndex(stringMap, "events"))
+		st.SetUnitStrindex(getStringMapIndex(stringMap, "nanoseconds"))
+	default:
+		log.Errorf("Generating profile for unsupported origin %d", origin)
+		return
+	}
 
 	// Temporary lookup to reference existing Mappings.
 	fileIDtoMapping := make(map[libpf.FileID]int32)
@@ -85,7 +99,15 @@ func (p *Pdata) setProfile(
 		endTS = pcommon.Timestamp(traceInfo.Timestamps[len(traceInfo.Timestamps)-1])
 
 		sample.TimestampsUnixNano().FromRaw(traceInfo.Timestamps)
-		sample.Value().Append(1)
+
+		switch origin {
+		case support.TraceOriginSampling:
+			sample.Value().Append(1)
+		case support.TraceOriginOffCPU:
+			for _, offTime := range traceInfo.OffTimes {
+				sample.Value().Append(int64(offTime))
+			}
+		}
 
 		// Walk every frame of the trace.
 		for i := range traceInfo.FrameTypes {

reporter/internal/samples/samples.go

@@ -12,6 +12,8 @@ type TraceEventMeta struct {
 	APMServiceName string
 	PID, TID       libpf.PID
 	CPU            int
+	Origin         libpf.Origin
+	OffTime        uint64
 }
 
 // TraceEvents holds known information about a trace.
@@ -23,6 +25,7 @@ type TraceEvents struct {
 	MappingEnds        []libpf.Address
 	MappingFileOffsets []uint64
 	Timestamps         []uint64 // in nanoseconds
+	OffTimes           []uint64 // in nanoseconds
 }
 
 // TraceAndMetaKey is the deduplication key for samples. This **must always**
@@ -43,6 +46,9 @@ type TraceAndMetaKey struct {
 	ExtraMeta any
 }
 
+// KeyToEventMapping supports temporary mapping traces to additional information.
+type KeyToEventMapping map[TraceAndMetaKey]*TraceEvents
+
 // AttrKeyValue is a helper to populate Profile.attribute_table.
 type AttrKeyValue[T string | int64] struct {
 	Key string

reporter/otlp_reporter.go

@@ -23,6 +23,7 @@ import (
 	"go.opentelemetry.io/ebpf-profiler/libpf/xsync"
 	"go.opentelemetry.io/ebpf-profiler/reporter/internal/pdata"
 	"go.opentelemetry.io/ebpf-profiler/reporter/internal/samples"
+	"go.opentelemetry.io/ebpf-profiler/support"
 )
 
 // Assert that we implement the full Reporter interface.
@@ -86,16 +87,20 @@ func NewOTLP(cfg *Config) (*OTLPReporter, error) {
 		return nil, err
 	}
 
+	originsMap := make(map[libpf.Origin]samples.KeyToEventMapping, 2)
+	for _, origin := range []libpf.Origin{support.TraceOriginSampling,
+		support.TraceOriginOffCPU} {
+		originsMap[origin] = make(samples.KeyToEventMapping)
+	}
+
 	return &OTLPReporter{
 		baseReporter: &baseReporter{
-			cfg:        cfg,
-			name:       cfg.Name,
-			version:    cfg.Version,
-			pdata:      data,
-			cgroupv2ID: cgroupv2ID,
-			traceEvents: xsync.NewRWMutex(
-				map[samples.TraceAndMetaKey]*samples.TraceEvents{},
-			),
+			cfg:          cfg,
+			name:         cfg.Name,
+			version:      cfg.Version,
+			pdata:        data,
+			cgroupv2ID:   cgroupv2ID,
+			traceEvents:  xsync.NewRWMutex(originsMap),
 			hostmetadata: hostmetadata,
 			runLoop: &runLoop{
 				stopSignal: make(chan libpf.Void),
@@ -165,7 +170,13 @@ func (r *OTLPReporter) Start(ctx context.Context) error {
 func (r *OTLPReporter) reportOTLPProfile(ctx context.Context) error {
 	traceEvents := r.traceEvents.WLock()
 	events := maps.Clone(*traceEvents)
+	originsMap := make(map[libpf.Origin]samples.KeyToEventMapping, 2)
 	clear(*traceEvents)
+	for _, origin := range []libpf.Origin{support.TraceOriginSampling,
+		support.TraceOriginOffCPU} {
+		originsMap[origin] = make(samples.KeyToEventMapping)
+	}
+	*traceEvents = originsMap
 	r.traceEvents.WUnlock(&traceEvents)
 
 	profiles := r.pdata.Generate(events)

support/ebpf/bpfdefs.h

@@ -83,6 +83,8 @@ static int (*bpf_perf_event_output)(void *ctx, void *map, unsigned long long fla
     (void *)BPF_FUNC_perf_event_output;
 static int (*bpf_get_stackid)(void *ctx, void *map, u64 flags) =
     (void *)BPF_FUNC_get_stackid;
+static unsigned long long (*bpf_get_prandom_u32)(void) =
+	(void *) BPF_FUNC_get_prandom_u32;
 
 __attribute__ ((format (printf, 1, 3)))
 static int (*bpf_trace_printk)(const char *fmt, int fmt_size, ...) =

support/ebpf/dotnet_tracer.ebpf.c

@@ -244,7 +244,7 @@ ErrorCode unwind_one_dotnet_frame(PerCPURecord *record, DotnetProcInfo *vi, bool
 // unwind_dotnet is the entry point for tracing when invoked from the native tracer
 // or interpreter dispatcher. It does not reset the trace object and will append the
 // dotnet stack frames to the trace object for the current CPU.
-SEC("perf_event/unwind_dotnet")
+static inline __attribute__((__always_inline__))
 int unwind_dotnet(struct pt_regs *ctx) {
   PerCPURecord *record = get_per_cpu_record();
   if (!record) {
@@ -289,3 +289,4 @@ int unwind_dotnet(struct pt_regs *ctx) {
   DEBUG_PRINT("dotnet: tail call for next frame unwinder (%d) failed", unwinder);
   return -1;
 }
+MULTI_USE_FUNC(unwind_dotnet)