Add ELF symbol upload

cli_flags.go

@@ -10,6 +10,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"strconv"
 	"time"
 
 	cebpf "github.com/cilium/ebpf"
@@ -91,6 +92,7 @@ type arguments struct {
 	saveCPUProfile         bool
 	sendErrorFrames        bool
 	serviceName            string
+	symbolUpload           bool
 	tags                   string
 	tracers                string
 	verboseMode            bool
@@ -162,6 +164,9 @@ func parseArgs() (*arguments, error) {
 
 	args.fs = fs
 
+	symbolUpload := os.Getenv("DD_EXPERIMENTAL_LOCAL_SYMBOL_UPLOAD")
+	args.symbolUpload, _ = strconv.ParseBool(symbolUpload)
+
 	return &args, ff.Parse(fs, os.Args[1:],
 		ff.WithEnvVarPrefix("OTEL_PROFILING_AGENT"),
 		ff.WithConfigFileFlag("config"),

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/sirupsen/logrus v1.9.3
 	github.com/tklauser/numcpus v0.8.0
 	github.com/zeebo/xxh3 v1.0.2
+	golang.org/x/sync v0.7.0
 	golang.org/x/sys v0.21.0
 )
 
@@ -32,7 +33,6 @@ require (
 	golang.org/x/arch v0.8.0 // indirect
 	golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8 // indirect
 	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240513163218-0867130af1f8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240513163218-0867130af1f8 // indirect

main.go

@@ -196,6 +196,7 @@ func mainWithExitCode() exitCode {
 		IPAddress:        sourceIP,
 		SaveCPUProfile:   args.saveCPUProfile,
 		Tags:             validatedTags,
+		UploadSymbols:    args.symbolUpload,
 	})
 	if err != nil {
 		return failure("Failed to start reporting: %v", err)

reporter/config.go

@@ -43,4 +43,6 @@ type Config struct {
 	SaveCPUProfile bool
 	// Tags is a list of tags to be sent to the collection agent.
 	Tags Tags
+	// SymbolUpload defines whether the agent should upload debug symbols to the backend.
+	UploadSymbols bool
 }

reporter/datadog_reporter.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-ebpf-profiler/libpf"
 	"github.com/open-telemetry/opentelemetry-ebpf-profiler/libpf/xsync"
+	"github.com/open-telemetry/opentelemetry-ebpf-profiler/process"
 	"github.com/open-telemetry/opentelemetry-ebpf-profiler/reporter"
 	"github.com/open-telemetry/opentelemetry-ebpf-profiler/util"
 )
@@ -110,7 +111,7 @@ type DatadogReporter struct {
 	// traceEvents stores reported trace events (trace metadata with frames and counts)
 	traceEvents xsync.RWMutex[map[traceAndMetaKey]*traceFramesCounts]
 
-	// execPathes stores the last known execPath for a PID.
+	// processes stores the metadata associated to a PID.
 	processes *lru.SyncedLRU[libpf.PID, processMetadata]
 
 	// samplesPerSecond is the number of samples per second.
@@ -133,6 +134,8 @@ type DatadogReporter struct {
 	saveCPUProfile bool
 
 	tags Tags
+
+	symbolUploader *DatadogSymbolUploader
 }
 
 // ReportTraceEvent enqueues reported trace events for the Datadog reporter.
@@ -211,11 +214,15 @@ func (r *DatadogReporter) ReportFallbackSymbol(frameID libpf.FrameID, symbol str
 // ExecutableMetadata accepts a fileID with the corresponding filename
 // and caches this information.
 func (r *DatadogReporter) ExecutableMetadata(fileID libpf.FileID, fileName, buildID string,
-	interp libpf.InterpreterType, open reporter.ExecutableOpener) {
+	interp libpf.InterpreterType, opener process.FileOpener) {
 	r.executables.Add(fileID, execInfo{
 		fileName: fileName,
 		buildID:  buildID,
 	})
+
+	if interp == libpf.Native {
+		r.symbolUploader.UploadSymbols(fileID, fileName, buildID, opener)
+	}
 }
 
 // FrameMetadata accepts metadata associated with a frame and caches this information.
@@ -317,6 +324,17 @@ func Start(mainCtx context.Context, cfg *Config) (reporter.Reporter, error) {
 		return nil, err
 	}
 
+	var symbolUploader *DatadogSymbolUploader
+	if cfg.UploadSymbols {
+		log.Infof("Enabling Datadog local symbol upload")
+		symbolUploader, err = NewDatadogSymbolUploader()
+		if err != nil {
+			log.Errorf(
+				"Failed to create Datadog symbol uploader, symbol upload will be disabled: %v",
+				err)
+		}
+	}
+
 	r := &DatadogReporter{
 		name:             cfg.Name,
 		version:          cfg.Version,
@@ -335,11 +353,20 @@ func Start(mainCtx context.Context, cfg *Config) (reporter.Reporter, error) {
 		agentAddr:        cfg.CollAgentAddr,
 		saveCPUProfile:   cfg.SaveCPUProfile,
 		tags:             cfg.Tags,
+		symbolUploader:   symbolUploader,
 	}
 
 	// Create a child context for reporting features
 	ctx, cancelReporting := context.WithCancel(mainCtx)
 
+	if r.symbolUploader != nil {
+		go func() {
+			if err := r.symbolUploader.Run(ctx); err != nil {
+				log.Errorf("Failed to run symbol uploader: %v", err)
+			}
+		}()
+	}
+
 	go func() {
 		tick := time.NewTicker(cfg.ReportInterval)
 		defer tick.Stop()
@@ -360,7 +387,6 @@ func Start(mainCtx context.Context, cfg *Config) (reporter.Reporter, error) {
 
 	// When Stop() is called and a signal to 'stop' is received, then:
 	// - cancel the reporting functions currently running (using context)
-	// - close the gRPC connection with collection-agent
 	go func() {
 		<-r.stopSignal
 		cancelReporting()