Merge pull request #240 from coroot/ebpf_traces_improvements

eBPF traces improvements

Author: def

containers/container.go

@@ -716,7 +716,17 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R
 	}
 	stats := c.l7Stats.get(r.Protocol, conn.DestinationKey)
 
-	trace := c.tracer.NewTrace(conn.DestinationKey.ActualDestinationIfKnown())
+	ebpfTracesDisabled := false
+	for _, p := range c.processes {
+		if p.Flags.EbpfTracesDisabled {
+			ebpfTracesDisabled = true
+			break
+		}
+	}
+	var trace *tracing.Trace
+	if !ebpfTracesDisabled {
+		trace = c.tracer.NewTrace(conn.DestinationKey.ActualDestinationIfKnown())
+	}
 	switch r.Protocol {
 	case l7.ProtocolHTTP:
 		method, path := l7.ParseHttp(r.Payload)
@@ -984,13 +994,10 @@ func (c *Container) runLogParser(logPath string) {
 		return
 	}
 
-	for pid := range c.processes {
-		if processFlags, err := proc.GetFlags(pid); err == nil {
-			if processFlags.LogMonitoringDisabled {
-				klog.InfoS("skipping log monitoring due to COROOT_LOG_MONITORING=disabled", "cg", c.cgroup.Id)
-				return
-			}
-			break
+	for _, p := range c.processes {
+		if p.Flags.LogMonitoringDisabled {
+			klog.InfoS("skipping log monitoring due to COROOT_LOG_MONITORING=disabled", "cg", c.cgroup.Id)
+			return
 		}
 	}
 

containers/process.go

@@ -28,6 +28,8 @@ type Process struct {
 	Pid       uint32
 	StartedAt time.Time
 
+	Flags proc.Flags
+
 	netNsId string
 
 	ctx        context.Context
@@ -46,6 +48,7 @@ type Process struct {
 
 func NewProcess(pid uint32, stats *taskstats.Stats, tracer *ebpftracer.Tracer) *Process {
 	p := &Process{Pid: pid, StartedAt: stats.BeginTime}
+	p.Flags, _ = proc.GetFlags(pid)
 	p.ctx, p.cancelFunc = context.WithCancel(context.Background())
 	go p.instrument(tracer)
 	return p

containers/registry.go

@@ -40,6 +40,7 @@ type ProcessInfo struct {
 	Pid         uint32
 	ContainerId ContainerID
 	StartedAt   time.Time
+	Flags       proc.Flags
 }
 
 type Registry struct {
@@ -236,7 +237,7 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 				if c := r.getOrCreateContainer(e.Pid); c != nil {
 					p := c.onProcessStart(e.Pid)
 					if r.processInfoCh != nil && p != nil {
-						r.processInfoCh <- ProcessInfo{Pid: p.Pid, ContainerId: c.id, StartedAt: p.StartedAt}
+						r.processInfoCh <- ProcessInfo{Pid: p.Pid, ContainerId: c.id, StartedAt: p.StartedAt, Flags: p.Flags}
 					}
 				}
 			case ebpftracer.EventTypeProcessExit:

flags/flags.go

@@ -41,6 +41,7 @@ var (
 	ApiKey             = kingpin.Flag("api-key", "Coroot API key").Envar("API_KEY").String()
 	MetricsEndpoint    = kingpin.Flag("metrics-endpoint", "The URL of the endpoint to send metrics to").Envar("METRICS_ENDPOINT").URL()
 	TracesEndpoint     = kingpin.Flag("traces-endpoint", "The URL of the endpoint to send traces to").Envar("TRACES_ENDPOINT").URL()
+	TracesSampling     = kingpin.Flag("traces-sampling", "Trace sampling rate (0.0 to 1.0)").Default("1.0").Envar("TRACES_SAMPLING").Float64()
 	LogsEndpoint       = kingpin.Flag("logs-endpoint", "The URL of the endpoint to send logs to").Envar("LOGS_ENDPOINT").URL()
 	ProfilesEndpoint   = kingpin.Flag("profiles-endpoint", "The URL of the endpoint to send profiles to").Envar("PROFILES_ENDPOINT").URL()
 	InsecureSkipVerify = kingpin.Flag("insecure-skip-verify", "whether to skip verifying the certificate or not").Envar("INSECURE_SKIP_VERIFY").Default("false").Bool()

proc/flags.go

@@ -8,6 +8,7 @@ import (
 
 type Flags struct {
 	EbpfProfilingDisabled bool
+	EbpfTracesDisabled    bool
 	LogMonitoringDisabled bool
 }
 
@@ -36,6 +37,8 @@ func GetFlags(pid uint32) (Flags, error) {
 			flags.EbpfProfilingDisabled = strings.Contains(kv[1], "disabled")
 		case "COROOT_LOG_MONITORING":
 			flags.LogMonitoringDisabled = strings.Contains(kv[1], "disabled")
+		case "COROOT_EBPF_TRACES":
+			flags.EbpfTracesDisabled = strings.Contains(kv[1], "disabled")
 		}
 	}
 	return flags, nil

profiling/profiling.go

@@ -215,6 +215,7 @@ func (tf *TargetFinder) start(processInfoCh <-chan containers.ProcessInfo) {
 				target: sd.NewTargetForTesting(cid, 0, sd.DiscoveryTarget{
 					"service_name": common.ContainerIdToOtelServiceName(cid),
 				}),
+				flags: pi.Flags,
 			}
 			tf.lock.Unlock()
 		}
@@ -234,10 +235,6 @@ func (tf *TargetFinder) FindTarget(pid uint32) *sd.Target {
 	var err error
 	if !pi.initialized {
 		pi.initialized = true
-		if pi.flags, err = proc.GetFlags(pid); err != nil {
-			delete(tf.processes, pid)
-			return nil
-		}
 		if !pi.flags.EbpfProfilingDisabled {
 			cmdline := proc.GetCmdline(pid)
 			if proc.IsJvm(cmdline) {

tracing/tracing.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"math/rand"
 	"time"
 
 	"github.com/coroot/coroot-node-agent/common"
@@ -29,6 +30,7 @@ var (
 	commonResourceAttrs []attribute.KeyValue
 	agentVersion        string
 	initialized         bool
+	samplingRate        float64
 )
 
 func Init(machineId, hostname, version string) {
@@ -37,6 +39,15 @@ func Init(machineId, hostname, version string) {
 		klog.Infoln("no OpenTelemetry traces collector endpoint configured")
 		return
 	}
+
+	samplingRate = *flags.TracesSampling
+	if samplingRate < 0.0 || samplingRate > 1.0 {
+		klog.Warningf("invalid traces-sampling value %f, must be between 0.0 and 1.0, using default 1.0", samplingRate)
+		samplingRate = 1.0
+	}
+	if samplingRate < 1.0 {
+		klog.Infof("trace sampling rate set to %f", samplingRate)
+	}
 	klog.Infoln("OpenTelemetry traces collector endpoint:", endpointUrl.String())
 	path := endpointUrl.Path
 	if path == "" {
@@ -67,6 +78,17 @@ type Tracer struct {
 	otel trace.Tracer
 }
 
+func shouldSample() bool {
+	if samplingRate >= 1.0 {
+		return true
+	}
+	if samplingRate <= 0.0 {
+		return false
+	}
+
+	return rand.Float64() < samplingRate
+}
+
 func GetContainerTracer(containerId string) *Tracer {
 	if !initialized {
 		return &Tracer{otel: nil}
@@ -103,6 +125,11 @@ func (t *Trace) createSpan(name string, duration time.Duration, error bool, attr
 		return
 	}
 	end := time.Now()
+
+	if !shouldSample() {
+		return
+	}
+
 	start := end.Add(-duration)
 	_, span := t.tracer.otel.Start(nil, name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
 	span.SetAttributes(attrs...)