Add Linux user_events receiver

.github/workflows/rust-ci.yml

@@ -196,6 +196,24 @@ jobs:
           retention-days: 30
           if-no-files-found: ignore
 
+  userevents_linux_smoke:
+    runs-on: ubuntu-24.04
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          submodules: true
+      - uses: dtolnay/rust-toolchain@3c5f7ea28cd621ae0bf5283f0e981fb97b8a7af9
+        with:
+          toolchain: stable
+      - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
+        with:
+          workspaces: ./rust/otap-dataflow
+      - name: Run user_events Linux smoke test when supported
+        env:
+          OTAP_DF_RUN_USEREVENTS_E2E: "1"
+        run: cargo test -p otap-df-contrib-nodes user_events_linux_e2e_smoke_when_available --features userevents-receiver,otap-df-otap/crypto-ring
+        working-directory: ./rust/otap-dataflow
+
   fmt:
     strategy:
       fail-fast: false

rust/otap-dataflow/Cargo.toml

@@ -231,6 +231,7 @@ anyhow = "1.0.98"
 crossterm = "0.29"
 data_engine_recordset = { path = '../experimental/query_engine/engine-recordset' }
 data_engine_recordset_otlp_bridge = { path = '../experimental/query_engine/engine-recordset-otlp-bridge' }
+eventheader_dynamic = "0.5.0"
 include_dir = "0.7.4"
 mime_guess = "2.0.5"
 minijinja = "2"
@@ -247,6 +248,8 @@ ratatui = "0.30"
 testcontainers = "0.27"
 toml = "1.0.0"
 tonic-prost-build = "0.14"
+tracepoint = { version = "0.5.0", features = ["user_events"] }
+tracepoint_decode = "0.5.0"
 urlencoding = "2"
 # Pinned to open-telemetry/opentelemetry-rust-contrib main on 2026-04-25.
 geneva-uploader = { git = "https://github.com/open-telemetry/opentelemetry-rust-contrib.git", rev = "ce866b48a1645f1fb222bdea19c42821a2b48d5a" }
@@ -290,6 +293,10 @@ crypto-openssl = ["otap-df-otap/crypto-openssl"]
 contrib-exporters = ["otap-df-contrib-nodes/contrib-exporters"]
 geneva-exporter = ["otap-df-contrib-nodes/geneva-exporter"]
 azure-monitor-exporter = ["otap-df-contrib-nodes/azure-monitor-exporter"]
+# Contrib receivers (opt-in) - now in contrib-nodes
+contrib-receivers = ["otap-df-contrib-nodes/contrib-receivers"]
+userevents-receiver = ["otap-df-contrib-nodes/userevents-receiver"]
+userevents-eventheader = ["otap-df-contrib-nodes/userevents-eventheader"]
 # Contrib processors (opt-in) - now in contrib-nodes
 contrib-processors = ["otap-df-contrib-nodes/contrib-processors"]
 condense-attributes-processor = ["otap-df-contrib-nodes/condense-attributes-processor"]

rust/otap-dataflow/configs/user-events-eventheader-debug.yaml

@@ -0,0 +1,61 @@
+version: otel_dataflow/v1
+engine: {}
+groups:
+  default:
+    pipelines:
+      main:
+        # Linux user_events EventHeader -> debug processor pipeline for manual
+        # E2E validation.
+        #
+        # Build:    cargo build --features userevents-eventheader
+        # Run:      ./target/debug/df_engine --config configs/user-events-eventheader-debug.yaml --num-cores 1
+        # Produce:  cargo run -p otap-df-contrib-nodes --features userevents-eventheader --example user_events_eventheader_producer
+        # Verify:   cat /tmp/user_events_eventheader_debug.log
+
+        policies:
+          channel_capacity:
+            control:
+              node: 100
+              pipeline: 100
+            pdata: 128
+
+        nodes:
+          receiver:
+            type: urn:otel:receiver:user_events
+            config:
+              subscriptions:
+                - tracepoint: "otap_df_eventheader_demo_L4K1"
+                  format:
+                    type: event_header
+              session:
+                per_cpu_buffer_size: 1048576
+                late_registration:
+                  enabled: true
+                  poll_interval_ms: 100
+              drain:
+                max_records_per_turn: 128
+                max_bytes_per_turn: 1048576
+                max_drain_ns: 2ms
+              batching:
+                max_size: 1
+                max_duration: 100ms
+              overflow:
+                on_downstream_full: drop
+
+          debug:
+            type: processor:debug
+            config:
+              verbosity: detailed
+              mode: signal
+              signals:
+                - logs
+              output: /tmp/user_events_eventheader_debug.log
+
+          noop:
+            type: exporter:noop
+
+        connections:
+          - from: receiver
+            to: debug
+          - from: debug
+            to: noop

rust/otap-dataflow/configs/user-events-tracefs-debug.yaml

@@ -0,0 +1,61 @@
+version: otel_dataflow/v1
+engine: {}
+groups:
+  default:
+    pipelines:
+      main:
+        # Linux user_events -> debug processor pipeline for manual tracefs E2E
+        # validation.
+        #
+        # Build:    cargo build --features userevents-receiver
+        # Run:      ./target/debug/df_engine --config configs/user-events-tracefs-debug.yaml --num-cores 1
+        # Produce:  cargo run -p otap-df-contrib-nodes --features userevents-receiver --example user_events_tracefs_producer
+        # Verify:   cat /tmp/user_events_tracefs_debug.log
+
+        policies:
+          channel_capacity:
+            control:
+              node: 100
+              pipeline: 100
+            pdata: 128
+
+        nodes:
+          receiver:
+            type: urn:otel:receiver:user_events
+            config:
+              subscriptions:
+                - tracepoint: "otap_df_tracefs_demo"
+                  format:
+                    type: tracefs
+              session:
+                per_cpu_buffer_size: 1048576
+                late_registration:
+                  enabled: true
+                  poll_interval_ms: 100
+              drain:
+                max_records_per_turn: 128
+                max_bytes_per_turn: 1048576
+                max_drain_ns: 2ms
+              batching:
+                max_size: 1
+                max_duration: 100ms
+              overflow:
+                on_downstream_full: drop
+
+          debug:
+            type: processor:debug
+            config:
+              verbosity: detailed
+              mode: signal
+              signals:
+                - logs
+              output: /tmp/user_events_tracefs_debug.log
+
+          noop:
+            type: exporter:noop
+
+        connections:
+          - from: receiver
+            to: debug
+          - from: debug
+            to: noop

rust/otap-dataflow/crates/contrib-nodes/Cargo.toml

@@ -14,29 +14,31 @@ workspace = true
 
 [dependencies]
 # Workspace crates (likely used by most/all features)
+otap-df-channel = { workspace = true }
 otap-df-config = { workspace = true }
 otap-df-engine = { workspace = true }
 otap-df-otap = { workspace = true }
 otap-df-pdata = { workspace = true }
 otap-df-pdata-views = { workspace = true }
 otap-df-telemetry = { workspace = true }
-otap-df-channel = { workspace = true }
 otap-df-telemetry-macros = { workspace = true }
 
 arrow.workspace = true
 async-trait.workspace = true
 futures.workspace = true
-tokio.workspace = true
-tracing.workspace = true
 bytes.workspace = true
 prost.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
 linkme.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+base64 = { workspace = true, optional = true }
 chrono = { workspace = true, optional = true }
-geneva-uploader = { workspace = true, optional = true }
 humantime-serde = { workspace = true, optional = true }
+itoa = { workspace = true, optional = true }
+geneva-uploader = { workspace = true, optional = true }
 opentelemetry-proto = { workspace = true, optional = true }
 data_engine_recordset = { workspace = true, optional = true }
 data_engine_recordset_otlp_bridge = { workspace = true, optional = true }
@@ -45,14 +47,33 @@ azure_core = { workspace = true, optional = true, features = ["reqwest"] }
 azure_identity = { workspace = true, optional = true }
 flate2 = { workspace = true, optional = true }
 http = { workspace = true, optional = true }
-itoa = { workspace = true, optional = true }
 rand = { workspace = true, optional = true }
 ryu = { workspace = true, optional = true }
 reqwest = { workspace = true, optional = true, features = ["rustls-no-provider"] }
 sysinfo = { workspace = true, optional = true }
 urlencoding = { workspace = true, optional = true }
 
+[target.'cfg(target_os = "linux")'.dependencies]
+# TODO: Remove this pinned git dependency once one-collect is upstream in the
+# normal dependency graph and no longer needs to be pulled directly here. The
+# pinned commit is from 2026-04-10.
+one_collect = { git = "https://github.com/microsoft/one-collect.git", rev = "2d33e674348168f4a19bd3ab40d92f1c556407ec", features = ["scripting"], optional = true }
+tracepoint_decode = { workspace = true, optional = true }
+
 [features]
+contrib-receivers = [
+    "userevents-receiver",
+]
+userevents-receiver = [
+    "dep:base64",
+    "dep:chrono",
+    "dep:humantime-serde",
+    "dep:one_collect",
+]
+userevents-eventheader = [
+    "userevents-receiver",
+    "dep:tracepoint_decode",
+]
 contrib-exporters = [
     "geneva-exporter",
     "azure-monitor-exporter",
@@ -96,13 +117,21 @@ axum.workspace = true
 clap.workspace = true
 criterion = { workspace = true, features = ["html_reports"] }
 humantime.workspace = true
+core_affinity.workspace = true
+otap-df-channel = { workspace = true }
 otap-df-engine = { workspace = true, features = ["test-utils"] }
 otap-df-otap = { workspace = true, features = ["test-utils"] }
+otap-df-pdata = { workspace = true, features = ["testing"] }
 pretty_assertions.workspace = true
 rand.workspace = true
 serde_yaml.workspace = true
+tracing-subscriber = { workspace = true, features = ["fmt", "std", "env-filter"] }
 wiremock.workspace = true
 
+[target.'cfg(target_os = "linux")'.dev-dependencies]
+eventheader_dynamic.workspace = true
+tracepoint.workspace = true
+
 [[bench]]
 name = "azure_monitor_transformer"
 path = "benches/exporters/azure_monitor_exporter/transformer.rs"
@@ -119,3 +148,13 @@ required-features = ["azure-monitor-exporter"]
 name = "mock_la_server"
 path = "examples/mock_la_server.rs"
 required-features = ["azure-monitor-exporter"]
+
+[[example]]
+name = "user_events_tracefs_producer"
+path = "src/receivers/user_events_receiver/examples/tracefs_producer.rs"
+required-features = ["userevents-receiver"]
+
+[[example]]
+name = "user_events_eventheader_producer"
+path = "src/receivers/user_events_receiver/examples/eventheader_producer.rs"
+required-features = ["userevents-eventheader"]

rust/otap-dataflow/crates/contrib-nodes/README.md

@@ -1,11 +1,15 @@
+<!-- markdownlint-disable MD013 -->
+
 # Contrib Nodes
 
-This crate contains optional (feature-gated) contrib processors and exporters.
+This crate contains contrib receivers, processors, and exporters.
 
 ## Folder Layout
 
 - `src/exporters/`
   - Contrib exporters
+- `src/receivers/`
+  - Contrib receivers
 - `src/processors/`
   - Contrib processors
 
@@ -14,6 +18,20 @@ This crate contains optional (feature-gated) contrib processors and exporters.
 Feature flags are grouped into aggregate categories and individual node flags.
 Aggregate flags enable all nodes in their category.
 
+### Receivers
+
+| Node | URN | Module |
+| ---- | --- | ------ |
+| user_events_receiver | `urn:otel:receiver:user_events` | `src/receivers/user_events_receiver/` |
+
+#### user_events_receiver
+
+- Reads Linux `user_events` tracepoints through per-CPU perf sessions
+- Supports single-tracepoint and multi-tracepoint configuration
+- Supports tracefs structural decoding by default
+- Supports EventHeader decoding when the `userevents-eventheader` feature is
+  enabled
+
 ### Exporters
 
 - `contrib-exporters` (enables all contrib exporters)

rust/otap-dataflow/crates/contrib-nodes/src/lib.rs

@@ -6,5 +6,8 @@
 /// Exporter implementations for contrib nodes.
 pub mod exporters;
 
+/// Receiver implementations for contrib nodes.
+pub mod receivers;
+
 /// Processor implementations for contrib nodes.
 pub mod processors;

rust/otap-dataflow/crates/contrib-nodes/src/receivers/mod.rs

@@ -0,0 +1,6 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+/// Linux user_events receiver.
+#[cfg(all(feature = "userevents-receiver", target_os = "linux"))]
+pub mod user_events_receiver;