Max transaction time (#786)

grcevski · web-flow · commit 32a31c35de67 · 2025-10-22T20:52:37.000-04:00
diff --git a/bpf/common/trace_common.h b/bpf/common/trace_common.h
@@ -32,6 +32,8 @@ enum { k_bpf_traceparent_enabled = 1 };
 enum { k_bpf_traceparent_enabled = 0 };
 #endif
 
+volatile const u64 max_transaction_time;
+
 static __always_inline unsigned char *tp_char_buf() {
     int zero = 0;
     return bpf_map_lookup_elem(&tp_char_buf_mem, &zero);
@@ -109,8 +111,8 @@ static __always_inline unsigned char *bpf_strstr_tp_loop__legacy(unsigned char *
     return NULL;
 }
 
-static __always_inline const tp_info_pid_t *
-find_nginx_parent_trace(const pid_connection_info_t *p_conn, u16 orig_dport) {
+static __always_inline tp_info_pid_t *find_nginx_parent_trace(const pid_connection_info_t *p_conn,
+                                                              u16 orig_dport) {
     connection_info_part_t client_part = {};
     populate_ephemeral_info(&client_part, &p_conn->conn, orig_dport, p_conn->pid, FD_CLIENT);
     fd_info_t *fd_info = fd_info_for_conn(&client_part);
@@ -127,8 +129,8 @@ find_nginx_parent_trace(const pid_connection_info_t *p_conn, u16 orig_dport) {
     return NULL;
 }
 
-static __always_inline const tp_info_pid_t *
-find_nodejs_parent_trace(const pid_connection_info_t *p_conn, u16 orig_dport) {
+static __always_inline tp_info_pid_t *find_nodejs_parent_trace(const pid_connection_info_t *p_conn,
+                                                               u16 orig_dport) {
     connection_info_part_t client_part = {};
     populate_ephemeral_info(&client_part, &p_conn->conn, orig_dport, p_conn->pid, FD_CLIENT);
     fd_info_t *fd_info = fd_info_for_conn(&client_part);
@@ -161,12 +163,12 @@ find_nodejs_parent_trace(const pid_connection_info_t *p_conn, u16 orig_dport) {
     return trace_info_for_connection(conn, TRACE_TYPE_SERVER);
 }
 
-static __always_inline const tp_info_pid_t *find_parent_process_trace(trace_key_t *t_key) {
+static __always_inline tp_info_pid_t *find_parent_process_trace(trace_key_t *t_key) {
     // Up to 5 levels of thread nesting allowed
     enum { k_max_depth = 5 };
 
     for (u8 i = 0; i < k_max_depth; ++i) {
-        const tp_info_pid_t *server_tp = bpf_map_lookup_elem(&server_traces, t_key);
+        tp_info_pid_t *server_tp = bpf_map_lookup_elem(&server_traces, t_key);
 
         if (server_tp) {
             bpf_dbg_printk("Found parent trace for pid=%d, ns=%lx, extra_id=%llx",
@@ -191,9 +193,9 @@ static __always_inline const tp_info_pid_t *find_parent_process_trace(trace_key_
     return NULL;
 }
 
-static __always_inline const tp_info_pid_t *find_parent_trace(const pid_connection_info_t *p_conn,
-                                                              u16 orig_dport) {
-    const tp_info_pid_t *node_tp = find_nodejs_parent_trace(p_conn, orig_dport);
+static __always_inline tp_info_pid_t *find_parent_trace(const pid_connection_info_t *p_conn,
+                                                        u16 orig_dport) {
+    tp_info_pid_t *node_tp = find_nodejs_parent_trace(p_conn, orig_dport);
 
     if (node_tp) {
         return node_tp;
@@ -208,13 +210,13 @@ static __always_inline const tp_info_pid_t *find_parent_trace(const pid_connecti
                    t_key.p_key.ns,
                    t_key.extra_id);
 
-    const tp_info_pid_t *nginx_parent = find_nginx_parent_trace(p_conn, orig_dport);
+    tp_info_pid_t *nginx_parent = find_nginx_parent_trace(p_conn, orig_dport);
 
     if (nginx_parent) {
         return nginx_parent;
     }
 
-    const tp_info_pid_t *proc_parent = find_parent_process_trace(&t_key);
+    tp_info_pid_t *proc_parent = find_parent_process_trace(&t_key);
 
     if (proc_parent) {
         return proc_parent;
@@ -382,13 +384,33 @@ static __always_inline u8 find_trace_for_server_request(connection_info_t *conn,
     return found_tp;
 }
 
+static __always_inline u8 should_be_in_same_transaction(const tp_info_t *parent_tp,
+                                                        const tp_info_t *child_tp) {
+    if (child_tp->ts < parent_tp->ts) {
+        return 0;
+    }
+
+    u64 diff = child_tp->ts - parent_tp->ts;
+
+    return diff < max_transaction_time;
+}
+
 static __always_inline u8 find_trace_for_client_request(const pid_connection_info_t *p_conn,
                                                         u16 orig_dport,
                                                         tp_info_t *tp) {
-    const tp_info_pid_t *server_tp = find_parent_trace(p_conn, orig_dport);
+    tp_info_pid_t *server_tp = find_parent_trace(p_conn, orig_dport);
 
     if (server_tp && server_tp->valid && valid_trace(server_tp->tp.trace_id)) {
         bpf_dbg_printk("Found existing server tp for client call");
+
+        if (!should_be_in_same_transaction(&server_tp->tp, tp)) {
+            bpf_dbg_printk("Parent and child are too far apart, marking server trace as invalid");
+            bpf_dbg_printk(
+                "%lld >>> %lld (max: %lld)", tp->ts, server_tp->tp.ts, max_transaction_time);
+            server_tp->valid = 0;
+            return 0;
+        }
+
         __builtin_memcpy(tp->trace_id, server_tp->tp.trace_id, sizeof(tp->trace_id));
         __builtin_memcpy(tp->parent_id, server_tp->tp.span_id, sizeof(tp->parent_id));
         return 1;
diff --git a/bpf/logger/bpf_dbg.h b/bpf/logger/bpf_dbg.h
@@ -31,7 +31,7 @@ typedef struct log_info {
 
 struct {
     __uint(type, BPF_MAP_TYPE_RINGBUF);
-    __uint(max_entries, 1 << 12);
+    __uint(max_entries, 1 << 15);
     __uint(pinning, OBI_PIN_INTERNAL);
 } debug_events SEC(".maps");
 
diff --git a/pkg/config/ebpf_tracer.go b/pkg/config/ebpf_tracer.go
@@ -106,6 +106,11 @@ type EBPFTracer struct {
 
 	// Configure data extraction/parsing based on protocol
 	PayloadExtraction PayloadExtraction `yaml:"payload_extraction"`
+
+	// Maximum time allowed for two requests to be correlated as parent -> child
+	// Some programs (e.g. load generators) keep on generating requests from the same thread in perpetuity,
+	// which can generate very large traces. We want to mark the parent trace as invalid if this happens.
+	MaxTransactionTime time.Duration `yaml:"max_transaction_time" env:"OTEL_EBPF_BPF_MAX_TRANSACTION_TIME"`
 }
 
 // Per-protocol data buffer size in bytes.
diff --git a/pkg/internal/ebpf/generictracer/generictracer.go b/pkg/internal/ebpf/generictracer/generictracer.go
@@ -178,47 +178,52 @@ func (p *Tracer) SetupTailCalls() {
 	}
 }
 
-func (p *Tracer) Constants() map[string]any {
+func GenericTracerConstants(cfg *obi.Config) map[string]any {
 	m := make(map[string]any, 2)
 
-	m["wakeup_data_bytes"] = uint32(p.cfg.EBPF.WakeupLen) * uint32(unsafe.Sizeof(ebpfcommon.HTTPRequestTrace{}))
+	m["wakeup_data_bytes"] = uint32(cfg.EBPF.WakeupLen) * uint32(unsafe.Sizeof(ebpfcommon.HTTPRequestTrace{}))
 
 	// The eBPF side does some basic filtering of events that do not belong to
 	// processes which we monitor. We filter more accurately in the userspace, but
 	// for performance reasons we enable the PID based filtering in eBPF.
 	// This must match httpfltr.go, otherwise we get partial events in userspace.
-	if p.cfg.Discovery.BPFPidFilterOff {
+	if cfg.Discovery.BPFPidFilterOff {
 		m["filter_pids"] = int32(0)
 	} else {
 		m["filter_pids"] = int32(1)
 	}
 
-	if p.cfg.EBPF.TrackRequestHeaders ||
-		p.cfg.EBPF.ContextPropagation != config.ContextPropagationDisabled {
+	if cfg.EBPF.TrackRequestHeaders ||
+		cfg.EBPF.ContextPropagation != config.ContextPropagationDisabled {
 		m["capture_header_buffer"] = int32(1)
 	} else {
 		m["capture_header_buffer"] = int32(0)
 	}
 
-	if p.cfg.EBPF.HighRequestVolume {
+	if cfg.EBPF.HighRequestVolume {
 		m["high_request_volume"] = uint32(1)
 	} else {
 		m["high_request_volume"] = uint32(0)
 	}
 
-	if p.cfg.EBPF.DisableBlackBoxCP {
+	if cfg.EBPF.DisableBlackBoxCP {
 		m["disable_black_box_cp"] = uint32(1)
 	} else {
 		m["disable_black_box_cp"] = uint32(0)
 	}
 
-	m["http_buffer_size"] = p.cfg.EBPF.BufferSizes.HTTP
-	m["mysql_buffer_size"] = p.cfg.EBPF.BufferSizes.MySQL
-	m["postgres_buffer_size"] = p.cfg.EBPF.BufferSizes.Postgres
+	m["http_buffer_size"] = cfg.EBPF.BufferSizes.HTTP
+	m["mysql_buffer_size"] = cfg.EBPF.BufferSizes.MySQL
+	m["postgres_buffer_size"] = cfg.EBPF.BufferSizes.Postgres
+	m["max_transaction_time"] = uint64(cfg.EBPF.MaxTransactionTime.Nanoseconds())
 
 	return m
 }
 
+func (p *Tracer) Constants() map[string]any {
+	return GenericTracerConstants(p.cfg)
+}
+
 func (p *Tracer) RegisterOffsets(_ *exec.FileInfo, _ *goexec.Offsets) {}
 
 func (p *Tracer) ProcessBinary(_ *exec.FileInfo) {}
diff --git a/pkg/internal/ebpf/gpuevent/gpuevent.go b/pkg/internal/ebpf/gpuevent/gpuevent.go
@@ -121,6 +121,8 @@ func (p *Tracer) Constants() map[string]any {
 		m["filter_pids"] = int32(1)
 	}
 
+	m["max_transaction_time"] = uint64(p.cfg.EBPF.MaxTransactionTime.Nanoseconds())
+
 	return m
 }
 
diff --git a/pkg/internal/ebpf/tctracer/tctracer.go b/pkg/internal/ebpf/tctracer/tctracer.go
@@ -74,7 +74,9 @@ func (p *Tracer) SetupTailCalls() {
 }
 
 func (p *Tracer) Constants() map[string]any {
-	return nil
+	return map[string]any{
+		"max_transaction_time": uint64(p.cfg.EBPF.MaxTransactionTime.Nanoseconds()),
+	}
 }
 
 func (p *Tracer) RegisterOffsets(_ *exec.FileInfo, _ *goexec.Offsets) {}
diff --git a/pkg/internal/ebpf/tpinjector/tpinjector.go b/pkg/internal/ebpf/tpinjector/tpinjector.go
@@ -70,7 +70,7 @@ func (p *Tracer) SetupTailCalls() {
 }
 
 func (p *Tracer) Constants() map[string]any {
-	m := make(map[string]any, 1)
+	m := make(map[string]any, 2)
 
 	// The eBPF side does some basic filtering of events that do not belong to
 	// processes which we monitor. We filter more accurately in the userspace, but
@@ -82,6 +82,8 @@ func (p *Tracer) Constants() map[string]any {
 		m["filter_pids"] = int32(1)
 	}
 
+	m["max_transaction_time"] = uint64(p.cfg.EBPF.MaxTransactionTime.Nanoseconds())
+
 	return m
 }
 
diff --git a/pkg/obi/config.go b/pkg/obi/config.go
@@ -99,6 +99,7 @@ var DefaultConfig = Config{
 				},
 			},
 		},
+		MaxTransactionTime: 5 * time.Minute,
 	},
 	NameResolver: &transform.NameResolverConfig{
 		Sources:  []string{"k8s"},
diff --git a/pkg/obi/config_test.go b/pkg/obi/config_test.go
@@ -125,6 +125,7 @@ discovery:
 			BatchLength:               100,
 			BatchTimeout:              time.Second,
 			HTTPRequestTimeout:        0,
+			MaxTransactionTime:        5 * time.Minute,
 			TCBackend:                 config.TCBackendAuto,
 			ContextPropagationEnabled: false,
 			ContextPropagation:        config.ContextPropagationDisabled,
diff --git a/test/integration/components/pythonselfserver/main.py b/test/integration/components/pythonselfserver/main.py
@@ -1,6 +1,7 @@
 from flask import Flask, request, jsonify, Response
 import requests
 import ssl
+import time
 from threading import Thread
 
 # Create four Flask applications for different ports
@@ -48,6 +49,28 @@ def api2():
 def smoke():
     return Response(status=200)
 
+@app3.route("/smoke1")
+def smoke1():
+    return Response(status=200)
+
+# API for the first application (Port 7771)
+@app3.route('/slow', methods=['GET'])
+def slow():
+    try:
+        # Forward all incoming headers to the internal HTTPS call
+        headers = dict(request.headers)
+
+        time.sleep(2)
+
+        # Internal HTTPS call to the second API
+        response = requests.get('http://localhost:7773/smoke1', headers=headers, verify=False)
+        return jsonify({
+            "message": "Internal call to smoke succeeded",
+        }), response.status_code
+    except requests.exceptions.RequestException as e:
+        return jsonify({"error": str(e)}), 500
+
+
 # API for the first application (Port 7773)
 @app3.route('/api3', methods=['GET'])
 def api3():
diff --git a/test/integration/docker-compose-python-self.yml b/test/integration/docker-compose-python-self.yml
@@ -47,11 +47,12 @@ services:
       OTEL_EBPF_BPF_CONTEXT_PROPAGATION: "all"
       OTEL_EBPF_INTERNAL_METRICS_PROMETHEUS_PORT: 8999
       OTEL_EBPF_INTERNAL_METRICS_PROMETHEUS_PATH: /metrics
-      OTEL_EBPF_OTLP_TRACES_BATCH_TIMEOUT: "1ms"
+      OTEL_EBPF_OTLP_TRACES_BATCH_TIMEOUT: "0ms"
       OTEL_EBPF_BPF_HTTP_REQUEST_TIMEOUT: "5s"
       OTEL_EBPF_PROCESSES_INTERVAL: "100ms"
       OTEL_EBPF_METRICS_FEATURES: "application,application_process,application_span,application_service_graph"
       OTEL_EBPF_BPF_BUFFER_SIZE_HTTP: 1024
+      OTEL_EBPF_BPF_MAX_TRANSACTION_TIME: "500ms"
     depends_on:
       testserver:
         condition: service_started
diff --git a/test/integration/suites_test.go b/test/integration/suites_test.go
@@ -449,6 +449,7 @@ func TestSuite_PythonSelfReference(t *testing.T) {
 	compose.Env = append(compose.Env, `OTEL_EBPF_OPEN_PORT=7771`, `OTEL_EBPF_EXECUTABLE_PATH=`)
 	require.NoError(t, compose.Up())
 	t.Run("Python Traces with self-references", testHTTPTracesNestedSelfCalls)
+	t.Run("Python Traces transaction too long", testHTTPTracesNestedCallsTooLong)
 	require.NoError(t, compose.Close())
 }
 
diff --git a/test/integration/traces_test.go b/test/integration/traces_test.go
@@ -993,6 +993,72 @@ func testNestedHTTPSTracesKProbes(t *testing.T) {
 	assert.Empty(t, sd, sd.String())
 }
 
+func testHTTPTracesNestedCallsTooLong(t *testing.T) {
+	var parentID string
+
+	// Run a request, since we have a single app, we should see always all requests
+	doHTTPGet(t, "http://localhost:7773/slow", 200)
+
+	var trace jaeger.Trace
+	test.Eventually(t, testTimeout, func(t require.TestingT) {
+		resp, err := http.Get(jaegerQueryURL + "?service=python-self&operation=GET%20%2Fsmoke1")
+		require.NoError(t, err)
+		if resp == nil {
+			return
+		}
+		require.Equal(t, http.StatusOK, resp.StatusCode)
+		var tq jaeger.TracesQuery
+		require.NoError(t, json.NewDecoder(resp.Body).Decode(&tq))
+		traces := tq.FindBySpan(jaeger.Tag{Key: "url.path", Type: "string", Value: "/smoke1"})
+		require.Len(t, traces, 1)
+		trace = traces[0]
+
+		resp, err = http.Get(jaegerQueryURL + "?service=python-self&operation=GET%20%2Fslow")
+		require.NoError(t, err)
+		if resp == nil {
+			return
+		}
+		require.Equal(t, http.StatusOK, resp.StatusCode)
+		require.NoError(t, json.NewDecoder(resp.Body).Decode(&tq))
+		traces = tq.FindBySpan(jaeger.Tag{Key: "url.path", Type: "string", Value: "/slow"})
+		require.Len(t, traces, 1)
+		trace = traces[0]
+
+		// Check the information of the parent span
+		res := trace.FindByOperationName("GET /slow", "")
+		require.Len(t, res, 1)
+		server := res[0]
+		require.NotEmpty(t, server.TraceID)
+		require.NotEmpty(t, server.SpanID)
+		parentID = server.SpanID
+
+		// check span attributes
+		sd := server.Diff(
+			jaeger.Tag{Key: "http.request.method", Type: "string", Value: "GET"},
+			jaeger.Tag{Key: "http.response.status_code", Type: "int64", Value: float64(200)},
+			jaeger.Tag{Key: "url.path", Type: "string", Value: "/slow"},
+			jaeger.Tag{Key: "server.port", Type: "int64", Value: float64(7773)},
+			jaeger.Tag{Key: "http.route", Type: "string", Value: "/slow"},
+			jaeger.Tag{Key: "span.kind", Type: "string", Value: "server"},
+		)
+		assert.Empty(t, sd, sd.String())
+
+		children := trace.ChildrenOf(parentID)
+		clientCallChildren := 0
+
+		// We might've created the in-queue and processing spans, but
+		// none of those should have children
+		for _, c := range children {
+			children = trace.ChildrenOf(c.SpanID)
+			if len(children) > 0 {
+				clientCallChildren++
+			}
+		}
+
+		require.Equal(t, 0, clientCallChildren)
+	}, test.Interval(100*time.Millisecond))
+}
+
 func testHTTPTracesNestedSelfCalls(t *testing.T) {
 	var parentID string
 

Original file line number	Diff line number	Diff line change
`@@ -121,6 +121,8 @@ func (p *Tracer) Constants() map[string]any {`
`121`	`121`	`m["filter_pids"] = int32(1)`
`122`	`122`	`}`
`123`	`123`
	`124`	`+ m["max_transaction_time"] = uint64(p.cfg.EBPF.MaxTransactionTime.Nanoseconds())`
	`125`	`+`
`124`	`126`	`return m`
`125`	`127`	`}`
`126`	`128`
Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,9 @@ func (p *Tracer) SetupTailCalls() {`
`74`	`74`	`}`
`75`	`75`
`76`	`76`	`func (p *Tracer) Constants() map[string]any {`
`77`		`- return nil`
	`77`	`+ return map[string]any{`
	`78`	`+ "max_transaction_time": uint64(p.cfg.EBPF.MaxTransactionTime.Nanoseconds()),`
	`79`	`+ }`
`78`	`80`	`}`
`79`	`81`
`80`	`82`	`func (p Tracer) RegisterOffsets(_ exec.FileInfo, _ *goexec.Offsets) {}`
Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ func (p *Tracer) SetupTailCalls() {`
`70`	`70`	`}`
`71`	`71`
`72`	`72`	`func (p *Tracer) Constants() map[string]any {`
`73`		`- m := make(map[string]any, 1)`
	`73`	`+ m := make(map[string]any, 2)`
`74`	`74`
`75`	`75`	`// The eBPF side does some basic filtering of events that do not belong to`
`76`	`76`	`// processes which we monitor. We filter more accurately in the userspace, but`
`@@ -82,6 +82,8 @@ func (p *Tracer) Constants() map[string]any {`
`82`	`82`	`m["filter_pids"] = int32(1)`
`83`	`83`	`}`
`84`	`84`
	`85`	`+ m["max_transaction_time"] = uint64(p.cfg.EBPF.MaxTransactionTime.Nanoseconds())`
	`86`	`+`
`85`	`87`	`return m`
`86`	`88`	`}`
`87`	`89`