Skip to content

GitLab

"docs/vscode:/vscode.git/clone" did not exist on "234a89c0c22d3ce29e52426e931a9c4ac31c62b7"
Unverified Commit 1f92dd54 authored by mohammedabdulwahhab's avatar mohammedabdulwahhab Committed by GitHub
Browse files

feat: OTEL Exporter and Tempo Visualization (#3307) 


Signed-off-by: default avatarmohammedabdulwahhab <furkhan324@berkeley.edu>
parent 037cc35d
Show whitespace changes
Inline Side-by-side
Showing with 1184 additions and 133 deletions
Cargo.lock
View file @ 1f92dd54
......@@ -1252,7 +1252,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8030735ecb0d128428b64cd379809817e620a40e5001c54465b99ec5feec2857"
dependencies = [
"futures-core",
"prost",
"prost 0.13.5",
"prost-types",
"tonic 0.12.3",
"tracing-core",
......@@ -1271,7 +1271,7 @@ dependencies = [
"hdrhistogram",
"humantime",
"hyper-util",
"prost",
"prost 0.13.5",
"prost-types",
"serde",
"serde_json",
......@@ -2176,7 +2176,7 @@ dependencies = [
"parking_lot",
"prometheus",
"proptest",
"prost",
"prost 0.13.5",
"rand 0.9.2",
"rayon",
"regex",
......@@ -2295,6 +2295,9 @@ dependencies = [
"nix 0.29.0",
"nuid",
"once_cell",
"opentelemetry",
"opentelemetry-otlp",
"opentelemetry_sdk",
"prometheus",
"rand 0.9.2",
"rayon",
......@@ -2314,6 +2317,7 @@ dependencies = [
"tokio-util",
"tower-http",
"tracing",
"tracing-opentelemetry",
"tracing-subscriber",
"url",
"uuid 1.18.1",
......@@ -2519,7 +2523,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "88365f1a5671eb2f7fc240adb216786bc6494b38ce15f1d26ad6eaa303d5e822"
dependencies = [
"http 1.3.1",
"prost",
"prost 0.13.5",
"tokio",
"tokio-stream",
"tonic 0.13.1",
......@@ -4881,7 +4885,7 @@ dependencies = [
"colored",
"futures",
"modelexpress-common",
"prost",
"prost 0.13.5",
"serde",
"serde_json",
"thiserror 2.0.16",
......@@ -4904,7 +4908,7 @@ dependencies = [
"config",
"hf-hub",
"jiff",
"prost",
"prost 0.13.5",
"serde",
"serde_json",
"serde_yaml",
......@@ -5485,6 +5489,82 @@ dependencies = [
"vcpkg",
]
[[package]]
name = "opentelemetry"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0"
dependencies = [
"futures-core",
"futures-sink",
"js-sys",
"pin-project-lite",
"thiserror 2.0.16",
"tracing",
]
[[package]]
name = "opentelemetry-http"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d"
dependencies = [
"async-trait",
"bytes",
"http 1.3.1",
"opentelemetry",
"reqwest 0.12.23",
]
[[package]]
name = "opentelemetry-otlp"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a2366db2dca4d2ad033cad11e6ee42844fd727007af5ad04a1730f4cb8163bf"
dependencies = [
"http 1.3.1",
"opentelemetry",
"opentelemetry-http",
"opentelemetry-proto",
"opentelemetry_sdk",
"prost 0.14.1",
"reqwest 0.12.23",
"thiserror 2.0.16",
"tokio",
"tonic 0.14.2",
"tracing",
]
[[package]]
name = "opentelemetry-proto"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f"
dependencies = [
"opentelemetry",
"opentelemetry_sdk",
"prost 0.14.1",
"tonic 0.14.2",
"tonic-prost",
]
[[package]]
name = "opentelemetry_sdk"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd"
dependencies = [
"futures-channel",
"futures-executor",
"futures-util",
"opentelemetry",
"percent-encoding",
"rand 0.9.2",
"thiserror 2.0.16",
"tokio",
"tokio-stream",
]
[[package]]
name = "option-ext"
version = "0.2.0"
......@@ -5991,7 +6071,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5"
dependencies = [
"bytes",
"prost-derive",
"prost-derive 0.13.5",
]
[[package]]
name = "prost"
version = "0.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7231bd9b3d3d33c86b58adbac74b5ec0ad9f496b19d22801d773636feaa95f3d"
dependencies = [
"bytes",
"prost-derive 0.14.1",
]
[[package]]
......@@ -6007,7 +6097,7 @@ dependencies = [
"once_cell",
"petgraph",
"prettyplease",
"prost",
"prost 0.13.5",
"prost-types",
"regex",
"syn 2.0.106",
......@@ -6027,13 +6117,26 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "prost-derive"
version = "0.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9120690fafc389a67ba3803df527d0ec9cbbc9cc45e4cc20b332996dfb672425"
dependencies = [
"anyhow",
"itertools 0.14.0",
"proc-macro2",
"quote",
"syn 2.0.106",
]
[[package]]
name = "prost-types"
version = "0.13.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16"
dependencies = [
"prost",
"prost 0.13.5",
]
[[package]]
......@@ -8622,7 +8725,7 @@ dependencies = [
"hyper-util",
"percent-encoding",
"pin-project",
"prost",
"prost 0.13.5",
"socket2 0.5.10",
"tokio",
"tokio-stream",
......@@ -8651,7 +8754,7 @@ dependencies = [
"hyper-util",
"percent-encoding",
"pin-project",
"prost",
"prost 0.13.5",
"socket2 0.5.10",
"tokio",
"tokio-rustls",
......@@ -8662,6 +8765,32 @@ dependencies = [
"tracing",
]
[[package]]
name = "tonic"
version = "0.14.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eb7613188ce9f7df5bfe185db26c5814347d110db17920415cf2fbcad85e7203"
dependencies = [
"async-trait",
"base64 0.22.1",
"bytes",
"http 1.3.1",
"http-body 1.0.1",
"http-body-util",
"hyper 1.7.0",
"hyper-timeout",
"hyper-util",
"percent-encoding",
"pin-project",
"sync_wrapper 1.0.2",
"tokio",
"tokio-stream",
"tower 0.5.2",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "tonic-build"
version = "0.13.1"
......@@ -8676,6 +8805,17 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "tonic-prost"
version = "0.14.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "66bd50ad6ce1252d87ef024b3d64fe4c3cf54a86fb9ef4c631fdd0ded7aeaa67"
dependencies = [
"bytes",
"prost 0.14.1",
"tonic 0.14.2",
]
[[package]]
name = "tower"
version = "0.4.13"
......@@ -8801,6 +8941,25 @@ dependencies = [
"tracing-core",
]
[[package]]
name = "tracing-opentelemetry"
version = "0.32.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e6e5658463dd88089aba75c7791e1d3120633b1bfde22478b28f625a9bb1b8e"
dependencies = [
"js-sys",
"opentelemetry",
"opentelemetry_sdk",
"rustversion",
"smallvec",
"thiserror 2.0.16",
"tracing",
"tracing-core",
"tracing-log",
"tracing-subscriber",
"web-time",
]
[[package]]
name = "tracing-serde"
version = "0.2.0"
......
Cargo.toml
View file @ 1f92dd54
......@@ -105,6 +105,10 @@ tracing-subscriber = { version = "0.3", features = [
"local-time",
"json",
] }
tracing-opentelemetry = { version = "0.32.0" }
opentelemetry = { version = "0.31.0", features = ["trace"] }
opentelemetry_sdk = { version = "0.31.0", features = ["trace", "rt-tokio"] }
opentelemetry-otlp = { version = "0.31.0", features = ["trace", "grpc-tonic"] }
validator = { version = "0.20.0", features = ["derive"] }
uuid = { version = "1.17", features = ["v4", "serde"] }
url = { version = "2.5", features = ["serde"] }
......
deploy/tracing/README.md 0 → 100644
View file @ 1f92dd54
<!--
SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
SPDX-License-Identifier: Apache-2.0
-->
# Distributed Tracing with Tempo
This guide explains how to set up and view distributed traces in Grafana Tempo for Dynamo workloads.
## Overview
Dynamo supports OpenTelemetry-based distributed tracing, allowing you to visualize request flows across Frontend and Worker components. Traces are exported to Tempo via OTLP (OpenTelemetry Protocol) and visualized in Grafana.
## Prerequisites
- Docker and Docker Compose (for local deployment)
- Kubernetes cluster with kubectl access (for Kubernetes deployment)
- Dynamo runtime with tracing support
## Environment Variables
Dynamo's tracing is configured via environment variables. For complete logging documentation, see [docs/guides/logging.md](../../docs/guides/logging.md).
### Required Environment Variables
| Variable | Description | Example Value |
|----------|-------------|---------------|
| `DYN_LOGGING_JSONL` | Enable JSONL logging format (required for tracing) | `true` |
| `OTEL_EXPORT_ENABLED` | Enable OTLP trace export | `1` |
| `OTEL_EXPORT_ENDPOINT` | OTLP gRPC endpoint for Tempo | `http://localhost:4317` (local) or `http://tempo:4317` (docker) |
| `OTEL_SERVICE_NAME` | Service name for identifying components | `dynamo-frontend`, `dynamo-worker-prefill`, `dynamo-worker-decode` |
### Example Configuration
```bash
# Enable JSONL logging and tracing
export DYN_LOGGING_JSONL=true
# Enable trace export to Tempo
export OTEL_EXPORT_ENABLED=1
# Set the Tempo endpoint (docker-compose network)
export OTEL_EXPORT_ENDPOINT=http://tempo:4317
# Set service name to identify this component
export OTEL_SERVICE_NAME=dynamo-frontend
```
---
## Local Deployment with Docker Compose
### 1. Start Tempo and Grafana
From the `deploy/tracing` directory, start the observability stack:
```bash
cd deploy/tracing
docker-compose up -d
```
This will start:
- **Tempo** on `http://localhost:3200` (HTTP API) and `localhost:4317` (OTLP gRPC)
- **Grafana** on `http://localhost:3000` (username: `admin`, password: `admin`)
Verify services are running:
```bash
docker-compose ps
```
### 2. Set Environment Variables
Configure Dynamo components to export traces:
```bash
# Enable JSONL logging and tracing
export DYN_LOGGING_JSONL=true
export OTEL_EXPORT_ENABLED=1
export OTEL_EXPORT_ENDPOINT=http://localhost:4317
# Set service names for each component
export OTEL_SERVICE_NAME=dynamo-frontend
```
### 3. Run vLLM Disaggregated Deployment
Run the vLLM disaggregated script with tracing enabled:
```bash
# Navigate to vLLM launch directory
cd components/backends/vllm/launch
# Run disaggregated deployment (modify the script to export env vars first)
./disagg.sh
```
**Note:** You may need to modify `disagg.sh` to export the tracing environment variables before starting each component:
```bash
#!/bin/bash
set -e
trap 'echo Cleaning up...; kill 0' EXIT
# Enable tracing
export DYN_LOGGING_JSONL=true
export OTEL_EXPORT_ENABLED=1
export OTEL_EXPORT_ENDPOINT=http://localhost:4317
# Run frontend
export OTEL_SERVICE_NAME=dynamo-frontend
python -m dynamo.frontend --router-mode kv --http-port=8000 &
# Run decode worker
export OTEL_SERVICE_NAME=dynamo-worker-decode
CUDA_VISIBLE_DEVICES=0 python3 -m dynamo.vllm --model Qwen/Qwen3-0.6B --enforce-eager &
# Run prefill worker
export OTEL_SERVICE_NAME=dynamo-worker-prefill
CUDA_VISIBLE_DEVICES=1 python3 -m dynamo.vllm \
--model Qwen/Qwen3-0.6B \
--enforce-eager \
--is-prefill-worker &
wait
```
### 4. Generate Traces
Send requests to the frontend to generate traces:
```bash
curl -d '{
"model": "Qwen/Qwen3-0.6B",
"max_completion_tokens": 100,
"messages": [
{"role": "user", "content": "What is the capital of France?"}
]
}' \
-H 'Content-Type: application/json' \
-H 'x-request-id: test-trace-001' \
http://localhost:8000/v1/chat/completions
```
### 5. View Traces in Grafana Tempo
1. Open Grafana at `http://localhost:3000`
2. Login with username `admin` and password `admin`
3. Navigate to **Explore** (compass icon in the left sidebar)
4. Select **Tempo** as the data source (should be selected by default)
5. Use the **Search** tab to find traces:
- Search by **Service Name** (e.g., `dynamo-frontend`)
- Search by **Span Name** (e.g., `http-request`, `handle_payload`)
- Search by **Tags** (e.g., `x_request_id=test-trace-001`)
6. Click on a trace to view the detailed flame graph
#### Example Trace View
Below is an example of what a trace looks like in Grafana Tempo:
![Trace Example](./trace.png)
### 6. Stop Services
When done, stop the Tempo and Grafana stack:
```bash
cd deploy/tracing
docker-compose down
```
---
## Kubernetes Deployment
For Kubernetes deployments, ensure you have a Tempo instance deployed and accessible (e.g., `http://tempo.observability.svc.cluster.local:4317`).
### Modify DynamoGraphDeployment for Tracing
Add common tracing environment variables at the top level and service-specific names in each component in your `DynamoGraphDeployment` (e.g., `components/backends/vllm/deploy/disagg.yaml`):
```yaml
apiVersion: nvidia.com/v1alpha1
kind: DynamoGraphDeployment
metadata:
name: vllm-disagg
spec:
# Common environment variables for all services
env:
- name: DYN_LOGGING_JSONL
value: "true"
- name: OTEL_EXPORT_ENABLED
value: "1"
- name: OTEL_EXPORT_ENDPOINT
value: "http://tempo.observability.svc.cluster.local:4317"
services:
Frontend:
# ... existing configuration ...
extraPodSpec:
mainContainer:
# ... existing configuration ...
env:
- name: OTEL_SERVICE_NAME
value: "dynamo-frontend"
VllmDecodeWorker:
# ... existing configuration ...
extraPodSpec:
mainContainer:
# ... existing configuration ...
env:
- name: OTEL_SERVICE_NAME
value: "dynamo-worker-decode"
VllmPrefillWorker:
# ... existing configuration ...
extraPodSpec:
mainContainer:
# ... existing configuration ...
env:
- name: OTEL_SERVICE_NAME
value: "dynamo-worker-prefill"
```
Apply the updated DynamoGraphDeployment:
```bash
kubectl apply -f components/backends/vllm/deploy/disagg.yaml
```
Traces will now be exported to Tempo and can be viewed in Grafana.
deploy/tracing/docker-compose.yml 0 → 100644
View file @ 1f92dd54
# SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
version: '3.8'
services:
# Tempo - Distributed tracing backend
tempo:
image: grafana/tempo:2.8.2
command: [ "-config.file=/etc/tempo.yaml" ]
volumes:
- ./tempo.yaml:/etc/tempo.yaml
- tempo-data:/tmp/tempo
ports:
- "3200:3200" # Tempo HTTP
- "4317:4317" # OTLP gRPC receiver (accessible from host)
- "4318:4318" # OTLP HTTP receiver (accessible from host)
# Grafana - Visualization and dashboards
grafana:
image: grafana/grafana:12.2.0
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
- GF_FEATURE_TOGGLES_ENABLE=traceqlEditor
volumes:
- grafana-data:/var/lib/grafana
- ./grafana/provisioning:/etc/grafana/provisioning
depends_on:
- tempo
volumes:
tempo-data:
grafana-data:
deploy/tracing/grafana/provisioning/datasources/tempo.yaml 0 → 100644
View file @ 1f92dd54
# SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
apiVersion: 1
datasources:
- name: Tempo
type: tempo
access: proxy
url: http://tempo:3200
uid: tempo
isDefault: true
editable: true
jsonData:
httpMethod: GET
serviceMap:
datasourceUid: tempo
search:
hide: false
nodeGraph:
enabled: true
traceQuery:
timeShiftEnabled: true
spanStartTimeShift: 1h
spanEndTimeShift: 1h
spanBar:
type: Tag
deploy/tracing/tempo.yaml 0 → 100644
View file @ 1f92dd54
# SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
server:
http_listen_port: 3200
distributor:
receivers:
otlp:
protocols:
grpc:
endpoint: 0.0.0.0:4317
http:
endpoint: 0.0.0.0:4318
ingester:
trace_idle_period: 10s
max_block_bytes: 1_000_000
max_block_duration: 5m
storage:
trace:
backend: local
block:
bloom_filter_false_positive: .05
wal:
path: /tmp/tempo/wal
local:
path: /tmp/tempo/blocks
compactor:
compaction:
compaction_window: 1h
max_compaction_objects: 1000000
block_retention: 1h
compacted_block_retention: 10m
deploy/tracing/trace.png 0 → 100644
View file @ 1f92dd54
File suppressed by a .gitattributes entry or the file's encoding is unsupported.
docs/guides/logging.md
View file @ 1f92dd54
......@@ -32,6 +32,9 @@ distributed tracing.
| `DYN_LOG_USE_LOCAL_TZ` | Use local timezone for logging timestamps (default: UTC) | `DYN_LOG_USE_LOCAL_TZ=1` |
| `DYN_LOG` | Log levels per target `<default_level>,<module_path>=<level>,<module_path>=<level>` | `DYN_LOG=info,dynamo_runtime::system_status_server:trace` |
| `DYN_LOGGING_CONFIG_PATH` | Path to custom TOML logging configuration file | `DYN_LOGGING_CONFIG_PATH=/path/to/config.toml`|
| `OTEL_SERVICE_NAME` | Service name for OpenTelemetry traces (default: `dynamo`) | `OTEL_SERVICE_NAME=dynamo-frontend` |
| `OTEL_EXPORT_ENABLED` | Enable OTLP trace exporting (set to `1` to enable) | `OTEL_EXPORT_ENABLED=1` |
| `OTEL_EXPORT_ENDPOINT` | OTLP exporter endpoint (default: http://localhost:4317) | `OTEL_EXPORT_ENDPOINT=http://tempo:4317` |
## Available Logging Levels
......@@ -82,10 +85,49 @@ Resulting Log format:
{"time":"2025-09-02T15:53:31.943747Z","level":"INFO","target":"log","message":"Scheduler config values: {'max_num_seqs': 256, 'max_num_batched_tokens': 2048}","log.file":"/opt/dynamo/venv/lib/python3.12/site-packages/dynamo/vllm/main.py","log.line":268,"log.target":"main.get_engine_cache_info"}
```
## OpenTelemetry Distributed Tracing
When `DYN_LOGGING_JSONL` is enabled, Dynamo uses OpenTelemetry for distributed tracing. All logs include `trace_id` and `span_id` fields, and spans are automatically created for requests. By default, traces are **not exported**. To export traces to an observability backend (like Tempo, Jaeger, or Zipkin), set `OTEL_EXPORT_ENABLED=1`.
### Behavior
- **With `DYN_LOGGING_JSONL=true` only**: OpenTelemetry layer is active, generating trace context and span IDs for all requests. Traces appear in logs but are not exported anywhere.
- **With `OTEL_EXPORT_ENABLED=1` and `DYN_LOGGING_JSONL=true`**: Same as above, plus traces are exported to an OTLP collector for visualization.
### Configuration
To enable OTLP trace exporting:
1. Set `OTEL_EXPORT_ENABLED=1` to enable trace export
2. Optionally configure the endpoint using `OTEL_EXPORT_ENDPOINT` (default: `http://localhost:4317`)
3. Optionally set `OTEL_SERVICE_NAME` to identify the service (useful in Kubernetes, default: `dynamo`)
**Export Settings:**
- **Protocol**: gRPC (Tonic)
- **Service Name**: Value of `OTEL_SERVICE_NAME` env var, or `dynamo` if not set
- **Endpoint**: Value of `OTEL_EXPORT_ENDPOINT` env var, or `http://localhost:4317` if not set
### Example: JSONL Logging Only (No Export)
```bash
export DYN_LOGGING_JSONL=true
# OpenTelemetry is active, traces appear in logs, but nothing is exported
```
### Example: JSONL Logging + Trace Export to Tempo
```bash
export DYN_LOGGING_JSONL=true
export OTEL_EXPORT_ENABLED=1
export OTEL_EXPORT_ENDPOINT=http://tempo:4317
export OTEL_SERVICE_NAME=dynamo-frontend
# OpenTelemetry is active, traces appear in logs AND are exported to Tempo
```
## Trace and Span information
When `DYN_LOGGING_JSONL` is enabled with `DYN_LOG` set to greater than or equal to
`info` level trace information is added to all logged spans along with
`info` level, trace information is added to all logged spans along with
`SPAN_CREATED` and `SPAN_CLOSED` events.
### Example Request
......
lib/bindings/python/Cargo.lock
View file @ 1f92dd54
......@@ -1502,7 +1502,7 @@ dependencies = [
"oneshot",
"parking_lot",
"prometheus",
"prost",
"prost 0.13.5",
"rand 0.9.2",
"rayon",
"regex",
......@@ -1520,7 +1520,7 @@ dependencies = [
"tokio-util",
"toktrie",
"toktrie_hf_tokenizers",
"tonic",
"tonic 0.13.1",
"tonic-build",
"tower-http",
"tracing",
......@@ -1619,6 +1619,9 @@ dependencies = [
"nix 0.29.0",
"nuid",
"once_cell",
"opentelemetry",
"opentelemetry-otlp",
"opentelemetry_sdk",
"prometheus",
"rand 0.9.2",
"rayon",
......@@ -1633,6 +1636,7 @@ dependencies = [
"tokio-util",
"tower-http",
"tracing",
"tracing-opentelemetry",
"tracing-subscriber",
"url",
"uuid",
......@@ -1792,10 +1796,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "88365f1a5671eb2f7fc240adb216786bc6494b38ce15f1d26ad6eaa303d5e822"
dependencies = [
"http",
"prost",
"prost 0.13.5",
"tokio",
"tokio-stream",
"tonic",
"tonic 0.13.1",
"tonic-build",
"tower",
"tower-service",
......@@ -3482,12 +3486,12 @@ dependencies = [
"colored",
"futures",
"modelexpress-common",
"prost",
"prost 0.13.5",
"serde",
"serde_json",
"thiserror 2.0.16",
"tokio",
"tonic",
"tonic 0.13.1",
"tracing",
"tracing-subscriber",
"uuid",
......@@ -3505,13 +3509,13 @@ dependencies = [
"config",
"hf-hub",
"jiff",
"prost",
"prost 0.13.5",
"serde",
"serde_json",
"serde_yaml",
"thiserror 2.0.16",
"tokio",
"tonic",
"tonic 0.13.1",
"tonic-build",
"tracing",
]
......@@ -3936,6 +3940,82 @@ version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e"
[[package]]
name = "opentelemetry"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0"
dependencies = [
"futures-core",
"futures-sink",
"js-sys",
"pin-project-lite",
"thiserror 2.0.16",
"tracing",
]
[[package]]
name = "opentelemetry-http"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d"
dependencies = [
"async-trait",
"bytes",
"http",
"opentelemetry",
"reqwest",
]
[[package]]
name = "opentelemetry-otlp"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a2366db2dca4d2ad033cad11e6ee42844fd727007af5ad04a1730f4cb8163bf"
dependencies = [
"http",
"opentelemetry",
"opentelemetry-http",
"opentelemetry-proto",
"opentelemetry_sdk",
"prost 0.14.1",
"reqwest",
"thiserror 2.0.16",
"tokio",
"tonic 0.14.2",
"tracing",
]
[[package]]
name = "opentelemetry-proto"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f"
dependencies = [
"opentelemetry",
"opentelemetry_sdk",
"prost 0.14.1",
"tonic 0.14.2",
"tonic-prost",
]
[[package]]
name = "opentelemetry_sdk"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd"
dependencies = [
"futures-channel",
"futures-executor",
"futures-util",
"opentelemetry",
"percent-encoding",
"rand 0.9.2",
"thiserror 2.0.16",
"tokio",
"tokio-stream",
]
[[package]]
name = "option-ext"
version = "0.2.0"
......@@ -4358,7 +4438,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5"
dependencies = [
"bytes",
"prost-derive",
"prost-derive 0.13.5",
]
[[package]]
name = "prost"
version = "0.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7231bd9b3d3d33c86b58adbac74b5ec0ad9f496b19d22801d773636feaa95f3d"
dependencies = [
"bytes",
"prost-derive 0.14.1",
]
[[package]]
......@@ -4374,7 +4464,7 @@ dependencies = [
"once_cell",
"petgraph",
"prettyplease",
"prost",
"prost 0.13.5",
"prost-types",
"regex",
"syn 2.0.106",
......@@ -4394,13 +4484,26 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "prost-derive"
version = "0.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9120690fafc389a67ba3803df527d0ec9cbbc9cc45e4cc20b332996dfb672425"
dependencies = [
"anyhow",
"itertools 0.14.0",
"proc-macro2",
"quote",
"syn 2.0.106",
]
[[package]]
name = "prost-types"
version = "0.13.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16"
dependencies = [
"prost",
"prost 0.13.5",
]
[[package]]
......@@ -6299,7 +6402,7 @@ dependencies = [
"hyper-util",
"percent-encoding",
"pin-project",
"prost",
"prost 0.13.5",
"socket2 0.5.10",
"tokio",
"tokio-rustls",
......@@ -6310,6 +6413,32 @@ dependencies = [
"tracing",
]
[[package]]
name = "tonic"
version = "0.14.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eb7613188ce9f7df5bfe185db26c5814347d110db17920415cf2fbcad85e7203"
dependencies = [
"async-trait",
"base64 0.22.1",
"bytes",
"http",
"http-body",
"http-body-util",
"hyper",
"hyper-timeout",
"hyper-util",
"percent-encoding",
"pin-project",
"sync_wrapper",
"tokio",
"tokio-stream",
"tower",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "tonic-build"
version = "0.13.1"
......@@ -6324,6 +6453,17 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "tonic-prost"
version = "0.14.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "66bd50ad6ce1252d87ef024b3d64fe4c3cf54a86fb9ef4c631fdd0ded7aeaa67"
dependencies = [
"bytes",
"prost 0.14.1",
"tonic 0.14.2",
]
[[package]]
name = "tower"
version = "0.5.2"
......@@ -6418,6 +6558,25 @@ dependencies = [
"tracing-core",
]
[[package]]
name = "tracing-opentelemetry"
version = "0.32.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e6e5658463dd88089aba75c7791e1d3120633b1bfde22478b28f625a9bb1b8e"
dependencies = [
"js-sys",
"opentelemetry",
"opentelemetry_sdk",
"rustversion",
"smallvec",
"thiserror 2.0.16",
"tracing",
"tracing-core",
"tracing-log",
"tracing-subscriber",
"web-time",
]
[[package]]
name = "tracing-serde"
version = "0.2.0"
......
lib/bindings/python/rust/lib.rs
View file @ 1f92dd54
......@@ -14,7 +14,7 @@ use std::path::PathBuf;
use std::time::Duration;
use std::{fmt::Display, sync::Arc};
use tokio::sync::Mutex;
use tracing::{Instrument, info_span};
use tracing::Instrument;
use dynamo_runtime::{
self as rs, logging,
......@@ -64,35 +64,14 @@ static INIT: OnceCell<()> = OnceCell::new();
const DEFAULT_ANNOTATED_SETTING: Option<bool> = Some(true);
// Helper to create client span - always emit spans for consistency
fn create_client_span(
operation: &str,
request_id: &str,
trace_context: Option<&dynamo_runtime::logging::DistributedTraceContext>,
) -> tracing::Span {
if let Some(ctx) = trace_context {
info_span!(
"client_request",
operation = operation,
request_id = request_id,
trace_id = ctx.trace_id.as_str(),
parent_id = ctx.span_id.as_str(),
x_request_id = ctx.x_request_id.as_deref().unwrap_or(""),
x_dynamo_request_id = ctx.x_dynamo_request_id.as_deref().unwrap_or(""),
tracestate = ctx.tracestate.as_deref().unwrap_or("")
)
} else {
info_span!(
"client_request",
operation = operation,
request_id = request_id,
)
}
}
// Helper to get appropriate span for instrumentation - always emit spans
fn get_span_for_context(context: &context::Context, operation: &str) -> tracing::Span {
create_client_span(operation, context.inner().id(), context.trace_context())
logging::make_client_request_span(
operation,
context.inner().id(),
context.trace_context(),
None,
)
}
// Helper to create span for direct method with instance_id
......@@ -101,26 +80,12 @@ fn get_span_for_direct_context(
operation: &str,
instance_id: &str,
) -> tracing::Span {
if let Some(trace_ctx) = context.trace_context() {
info_span!(
"client_request",
operation = operation,
request_id = context.inner().id(),
instance_id = instance_id,
trace_id = trace_ctx.trace_id.as_str(),
parent_id = trace_ctx.span_id.as_str(),
x_request_id = trace_ctx.x_request_id.as_deref().unwrap_or(""),
x_dynamo_request_id = trace_ctx.x_dynamo_request_id.as_deref().unwrap_or(""),
tracestate = trace_ctx.tracestate.as_deref().unwrap_or("")
)
} else {
info_span!(
"client_request",
operation = operation,
request_id = context.inner().id(),
instance_id = instance_id,
logging::make_client_request_span(
operation,
context.inner().id(),
context.trace_context(),
Some(instance_id),
)
}
}
/// A Python module implemented in Rust. The name of this function must match
......@@ -128,7 +93,6 @@ fn get_span_for_direct_context(
/// import the module.
#[pymodule]
fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> {
logging::init();
m.add_function(wrap_pyfunction!(llm::kv::compute_block_hash_for_seq_py, m)?)?;
m.add_function(wrap_pyfunction!(log_message, m)?)?;
m.add_function(wrap_pyfunction!(register_llm, m)?)?;
......@@ -392,6 +356,12 @@ impl DistributedRuntime {
let runtime = worker.runtime().clone();
// Initialize logging in context where tokio runtime is available
// otel exporter requires it
runtime.secondary().block_on(async {
rs::logging::init();
});
let inner =
if is_static {
runtime.secondary().block_on(
......
lib/runtime/Cargo.toml
View file @ 1f92dd54
......@@ -46,6 +46,10 @@ tokio-util = { workspace = true }
tower-http = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
tracing-opentelemetry = { workspace = true }
opentelemetry = { workspace = true }
opentelemetry_sdk = { workspace = true }
opentelemetry-otlp = { workspace = true }
thiserror = { workspace = true }
uuid = { workspace = true }
url = { workspace = true }
......
lib/runtime/src/logging.rs
View file @ 1f92dd54
......@@ -69,6 +69,24 @@ use tracing_subscriber::layer::Context;
use tracing_subscriber::registry::SpanData;
use uuid::Uuid;
use opentelemetry::propagation::{Extractor, Injector, TextMapPropagator};
use opentelemetry::trace::TraceContextExt;
use opentelemetry::{global, trace::Tracer};
use opentelemetry_otlp::WithExportConfig;
use opentelemetry::trace::TracerProvider as _;
use opentelemetry::{Key, KeyValue};
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::trace::SdkTracerProvider;
use tracing::error;
use tracing_subscriber::layer::SubscriberExt;
// use tracing_subscriber::Registry;
use std::time::Duration;
use tracing::{info, instrument};
use tracing_opentelemetry::OpenTelemetrySpanExt;
use tracing_subscriber::util::SubscriberInitExt;
/// ENV used to set the log level
const FILTER_ENV: &str = "DYN_LOG";
......@@ -78,6 +96,21 @@ const DEFAULT_FILTER_LEVEL: &str = "info";
/// ENV used to set the path to the logging configuration file
const CONFIG_PATH_ENV: &str = "DYN_LOGGING_CONFIG_PATH";
/// Enable OTLP trace exporting
const OTEL_EXPORT_ENABLED_ENV: &str = "OTEL_EXPORT_ENABLED";
/// OTEL exporter endpoint
const OTEL_EXPORT_ENDPOINT_ENV: &str = "OTEL_EXPORT_ENDPOINT";
/// Default OTLP endpoint
const DEFAULT_OTLP_ENDPOINT: &str = "http://localhost:4317";
/// Service name environment variable
const OTEL_SERVICE_NAME_ENV: &str = "OTEL_SERVICE_NAME";
/// Default service name
const DEFAULT_OTEL_SERVICE_NAME: &str = "dynamo";
/// Once instance to ensure the logger is only initialized once
static INIT: Once = Once::new();
......@@ -102,22 +135,24 @@ impl Default for LoggingConfig {
("tonic".to_string(), "error".to_string()),
("mistralrs_core".to_string(), "error".to_string()),
("hf_hub".to_string(), "error".to_string()),
("opentelemetry".to_string(), "error".to_string()),
("opentelemetry-otlp".to_string(), "error".to_string()),
("opentelemetry_sdk".to_string(), "error".to_string()),
]),
}
}
}
/// Generate a 32-character, lowercase hex trace ID (W3C-compliant)
fn generate_trace_id() -> String {
Uuid::new_v4().simple().to_string()
/// Check if OTLP trace exporting is enabled (set OTEL_EXPORT_ENABLED=1 to enable)
fn otlp_exporter_enabled() -> bool {
std::env::var(OTEL_EXPORT_ENABLED_ENV)
.map(|v| v == "1")
.unwrap_or(false)
}
/// Generate a 16-character, lowercase hex span ID (W3C-compliant)
fn generate_span_id() -> String {
// Use the first 8 bytes (16 hex chars) of a UUID v4
let uuid = Uuid::new_v4();
let bytes = uuid.as_bytes();
bytes[..8].iter().map(|b| format!("{:02x}", b)).collect()
/// Get the service name from environment or use default
fn get_service_name() -> String {
std::env::var(OTEL_SERVICE_NAME_ENV).unwrap_or_else(|_| DEFAULT_OTEL_SERVICE_NAME.to_string())
}
/// Validate a given trace ID according to W3C Trace Context specifications.
......@@ -152,6 +187,17 @@ pub struct DistributedTraceContext {
pub x_dynamo_request_id: Option<String>,
}
/// Pending context data collected in on_new_span, to be finalized in on_enter
#[derive(Debug, Clone)]
struct PendingDistributedTraceContext {
trace_id: Option<String>,
span_id: Option<String>,
parent_id: Option<String>,
tracestate: Option<String>,
x_request_id: Option<String>,
x_dynamo_request_id: Option<String>,
}
impl DistributedTraceContext {
/// Create a traceparent string from the context
pub fn create_traceparent(&self) -> String {
......@@ -242,10 +288,9 @@ pub fn make_request_span<B>(req: &Request<B>) -> Span {
let method = req.method();
let uri = req.uri();
let version = format!("{:?}", req.version());
let trace_parent = TraceParent::from_headers(req.headers());
tracing::info_span!(
let span = tracing::info_span!(
"http-request",
method = %method,
uri = %uri,
......@@ -254,8 +299,184 @@ pub fn make_request_span<B>(req: &Request<B>) -> Span {
parent_id = trace_parent.parent_id,
x_request_id = trace_parent.x_request_id,
x_dynamo_request_id = trace_parent.x_dynamo_request_id,
);
span
}
/// Create a handle_payload span from NATS headers with component context
pub fn make_handle_payload_span(
headers: &async_nats::HeaderMap,
component: &str,
endpoint: &str,
namespace: &str,
instance_id: i64,
) -> Span {
let (otel_context, trace_id, parent_span_id) = extract_otel_context_from_nats_headers(headers);
let trace_parent = TraceParent::from_headers(headers);
if let (Some(trace_id), Some(parent_id)) = (trace_id.as_ref(), parent_span_id.as_ref()) {
let span = tracing::info_span!(
"handle_payload",
trace_id = trace_id.as_str(),
parent_id = parent_id.as_str(),
x_request_id = trace_parent.x_request_id,
x_dynamo_request_id = trace_parent.x_dynamo_request_id,
tracestate = trace_parent.tracestate,
component = component,
endpoint = endpoint,
namespace = namespace,
instance_id = instance_id,
);
if let Some(context) = otel_context {
let _ = span.set_parent(context);
}
span
} else {
tracing::info_span!(
"handle_payload",
x_request_id = trace_parent.x_request_id,
x_dynamo_request_id = trace_parent.x_dynamo_request_id,
tracestate = trace_parent.tracestate,
component = component,
endpoint = endpoint,
namespace = namespace,
instance_id = instance_id,
)
}
}
/// Extract OpenTelemetry trace context from NATS headers for distributed tracing
pub fn extract_otel_context_from_nats_headers(
headers: &async_nats::HeaderMap,
) -> (
Option<opentelemetry::Context>,
Option<String>,
Option<String>,
) {
let traceparent_value = match headers.get("traceparent") {
Some(value) => value.as_str(),
None => return (None, None, None),
};
let (trace_id, parent_span_id) = parse_traceparent(traceparent_value);
struct NatsHeaderExtractor<'a>(&'a async_nats::HeaderMap);
impl<'a> Extractor for NatsHeaderExtractor<'a> {
fn get(&self, key: &str) -> Option<&str> {
self.0.get(key).map(|value| value.as_str())
}
fn keys(&self) -> Vec<&str> {
vec!["traceparent", "tracestate"]
.into_iter()
.filter(|&key| self.0.get(key).is_some())
.collect()
}
}
let extractor = NatsHeaderExtractor(headers);
let propagator = opentelemetry_sdk::propagation::TraceContextPropagator::new();
let otel_context = propagator.extract(&extractor);
let context_with_trace = if otel_context.span().span_context().is_valid() {
Some(otel_context)
} else {
None
};
(context_with_trace, trace_id, parent_span_id)
}
/// Inject OpenTelemetry trace context into NATS headers using W3C Trace Context propagation
pub fn inject_otel_context_into_nats_headers(
headers: &mut async_nats::HeaderMap,
context: Option<opentelemetry::Context>,
) {
let otel_context = context.unwrap_or_else(|| Span::current().context());
struct NatsHeaderInjector<'a>(&'a mut async_nats::HeaderMap);
impl<'a> Injector for NatsHeaderInjector<'a> {
fn set(&mut self, key: &str, value: String) {
self.0.insert(key, value);
}
}
let mut injector = NatsHeaderInjector(headers);
let propagator = opentelemetry_sdk::propagation::TraceContextPropagator::new();
propagator.inject_context(&otel_context, &mut injector);
}
/// Inject trace context from current span into NATS headers
pub fn inject_current_trace_into_nats_headers(headers: &mut async_nats::HeaderMap) {
inject_otel_context_into_nats_headers(headers, None);
}
/// Create a client_request span linked to the parent trace context
pub fn make_client_request_span(
operation: &str,
request_id: &str,
trace_context: Option<&DistributedTraceContext>,
instance_id: Option<&str>,
) -> Span {
if let Some(ctx) = trace_context {
let mut headers = async_nats::HeaderMap::new();
headers.insert("traceparent", ctx.create_traceparent());
if let Some(ref tracestate) = ctx.tracestate {
headers.insert("tracestate", tracestate.as_str());
}
let (otel_context, _extracted_trace_id, _extracted_parent_span_id) =
extract_otel_context_from_nats_headers(&headers);
let span = if let Some(inst_id) = instance_id {
tracing::info_span!(
"client_request",
operation = operation,
request_id = request_id,
instance_id = inst_id,
trace_id = ctx.trace_id.as_str(),
parent_id = ctx.span_id.as_str(),
x_request_id = ctx.x_request_id.as_deref(),
x_dynamo_request_id = ctx.x_dynamo_request_id.as_deref(),
// tracestate = ctx.tracestate.as_deref(),
)
} else {
tracing::info_span!(
"client_request",
operation = operation,
request_id = request_id,
trace_id = ctx.trace_id.as_str(),
parent_id = ctx.span_id.as_str(),
x_request_id = ctx.x_request_id.as_deref(),
x_dynamo_request_id = ctx.x_dynamo_request_id.as_deref(),
// tracestate = ctx.tracestate.as_deref(),
)
};
if let Some(context) = otel_context {
let _ = span.set_parent(context);
}
span
} else if let Some(inst_id) = instance_id {
tracing::info_span!(
"client_request",
operation = operation,
request_id = request_id,
instance_id = inst_id,
)
} else {
tracing::info_span!(
"client_request",
operation = operation,
request_id = request_id,
)
}
}
#[derive(Debug, Default)]
......@@ -292,7 +513,8 @@ where
}
}
// Adds W3C compliant span_id, trace_id, and parent_id if not already present
// Collects span attributes and metadata in on_new_span
// Final initialization deferred to on_enter when OtelData is available
fn on_new_span(&self, attrs: &span::Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
if let Some(span) = ctx.span(id) {
let mut trace_id: Option<String> = None;
......@@ -304,6 +526,7 @@ where
let mut visitor = FieldVisitor::default();
attrs.record(&mut visitor);
// Extract trace_id from span attributes
if let Some(trace_id_input) = visitor.fields.get("trace_id") {
if !is_valid_trace_id(trace_id_input) {
tracing::trace!("trace id '{}' is not valid! Ignoring.", trace_id_input);
......@@ -312,6 +535,7 @@ where
}
}
// Extract span_id from span attributes
if let Some(span_id_input) = visitor.fields.get("span_id") {
if !is_valid_span_id(span_id_input) {
tracing::trace!("span id '{}' is not valid! Ignoring.", span_id_input);
......@@ -320,6 +544,7 @@ where
}
}
// Extract parent_id from span attributes
if let Some(parent_id_input) = visitor.fields.get("parent_id") {
if !is_valid_span_id(parent_id_input) {
tracing::trace!("parent id '{}' is not valid! Ignoring.", parent_id_input);
......@@ -328,18 +553,22 @@ where
}
}
// Extract tracestate
if let Some(tracestate_input) = visitor.fields.get("tracestate") {
tracestate = Some(tracestate_input.to_string());
}
// Extract x_request_id
if let Some(x_request_id_input) = visitor.fields.get("x_request_id") {
x_request_id = Some(x_request_id_input.to_string());
}
// Extract x_dynamo_request_id
if let Some(x_request_id_input) = visitor.fields.get("x_dynamo_request_id") {
x_dynamo_request_id = Some(x_request_id_input.to_string());
}
// Inherit trace context from parent span if available
if parent_id.is_none()
&& let Some(parent_span_id) = ctx.current_span().id()
&& let Some(parent_span) = ctx.span(parent_span_id)
......@@ -352,6 +581,7 @@ where
}
}
// Validate consistency
if (parent_id.is_some() || span_id.is_some()) && trace_id.is_none() {
tracing::error!("parent id or span id are set but trace id is not set!");
// Clear inconsistent IDs to maintain trace integrity
......@@ -359,13 +589,90 @@ where
span_id = None;
}
// Store pending context - will be finalized in on_enter
let mut extensions = span.extensions_mut();
extensions.insert(PendingDistributedTraceContext {
trace_id,
span_id,
parent_id,
tracestate,
x_request_id,
x_dynamo_request_id,
});
}
}
// Finalizes the DistributedTraceContext when span is entered
// At this point, OtelData should have valid trace_id and span_id
fn on_enter(&self, id: &Id, ctx: Context<'_, S>) {
if let Some(span) = ctx.span(id) {
// Check if already initialized (e.g., span re-entered)
{
let extensions = span.extensions();
if extensions.get::<DistributedTraceContext>().is_some() {
return;
}
}
// Get the pending context and extract OtelData IDs
let mut extensions = span.extensions_mut();
let pending = match extensions.remove::<PendingDistributedTraceContext>() {
Some(p) => p,
None => {
// This shouldn't happen - on_new_span should have created it
tracing::error!("PendingDistributedTraceContext not found in on_enter");
return;
}
};
let mut trace_id = pending.trace_id;
let mut span_id = pending.span_id;
let parent_id = pending.parent_id;
let tracestate = pending.tracestate;
let x_request_id = pending.x_request_id;
let x_dynamo_request_id = pending.x_dynamo_request_id;
// Try to extract from OtelData if not already set
// Need to drop extensions_mut to get immutable borrow for OtelData
drop(extensions);
if trace_id.is_none() || span_id.is_none() {
let extensions = span.extensions();
if let Some(otel_data) = extensions.get::<tracing_opentelemetry::OtelData>() {
// Extract trace_id from OTEL data if not already set
if trace_id.is_none()
&& let Some(otel_trace_id) = otel_data.trace_id()
{
let trace_id_str = format!("{}", otel_trace_id);
if is_valid_trace_id(&trace_id_str) {
trace_id = Some(trace_id_str);
}
}
// Extract span_id from OTEL data if not already set
if span_id.is_none()
&& let Some(otel_span_id) = otel_data.span_id()
{
let span_id_str = format!("{}", otel_span_id);
if is_valid_span_id(&span_id_str) {
span_id = Some(span_id_str);
}
}
}
}
// Panic if we still don't have required IDs
if trace_id.is_none() {
trace_id = Some(generate_trace_id());
panic!(
"trace_id is not set in on_enter - OtelData may not be properly initialized"
);
}
if span_id.is_none() {
span_id = Some(generate_span_id());
panic!("span_id is not set in on_enter - OtelData may not be properly initialized");
}
// Re-acquire mutable borrow to insert the finalized context
let mut extensions = span.extensions_mut();
extensions.insert(DistributedTraceContext {
trace_id: trace_id.expect("Trace ID must be set"),
......@@ -397,9 +704,14 @@ pub fn get_distributed_tracing_context() -> Option<DistributedTraceContext> {
.flatten()
}
/// Initialize the logger
/// Initialize the logger - must be called when Tokio runtime is available
pub fn init() {
INIT.call_once(setup_logging);
INIT.call_once(|| {
if let Err(e) = setup_logging() {
eprintln!("Failed to initialize logging: {}", e);
std::process::exit(1);
}
});
}
#[cfg(feature = "tokio-console")]
......@@ -424,9 +736,11 @@ fn setup_logging() {
}
#[cfg(not(feature = "tokio-console"))]
fn setup_logging() {
fn setup_logging() -> Result<(), Box<dyn std::error::Error>> {
let fmt_filter_layer = filters(load_config());
let trace_filter_layer = filters(load_config());
let otel_filter_layer = filters(load_config());
if jsonl_logging_enabled() {
let l = fmt::layer()
.with_ansi(false)
......@@ -434,7 +748,62 @@ fn setup_logging() {
.event_format(CustomJsonFormatter::new())
.with_writer(std::io::stderr)
.with_filter(fmt_filter_layer);
// Create OpenTelemetry tracer - conditionally export to OTLP based on env var
let service_name = get_service_name();
// Build tracer provider - with or without OTLP export
let tracer_provider = if otlp_exporter_enabled() {
// Export enabled: create OTLP exporter with batch processor
let endpoint = std::env::var(OTEL_EXPORT_ENDPOINT_ENV)
.unwrap_or_else(|_| DEFAULT_OTLP_ENDPOINT.to_string());
tracing::info!(
"OpenTelemetry OTLP export enabled, endpoint: {}, service: {}",
endpoint,
service_name
);
// Initialize OTLP exporter using gRPC (Tonic)
let otlp_exporter = opentelemetry_otlp::SpanExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.build()?;
// Create tracer provider with batch exporter and service name
opentelemetry_sdk::trace::SdkTracerProvider::builder()
.with_batch_exporter(otlp_exporter)
.with_resource(
opentelemetry_sdk::Resource::builder_empty()
.with_service_name(service_name.clone())
.build(),
)
.build()
} else {
// No export - traces generated locally only (for logging/trace IDs)
tracing::info!(
"OpenTelemetry OTLP export disabled, traces local only, service: {}",
service_name
);
opentelemetry_sdk::trace::SdkTracerProvider::builder()
.with_resource(
opentelemetry_sdk::Resource::builder_empty()
.with_service_name(service_name.clone())
.build(),
)
.build()
};
// Get a tracer from the provider
let tracer = tracer_provider.tracer(service_name);
tracing_subscriber::registry()
.with(
tracing_opentelemetry::layer()
.with_tracer(tracer)
.with_filter(otel_filter_layer),
)
.with(DistributedTraceIdLayer.with_filter(trace_filter_layer))
.with(l)
.init();
......@@ -444,8 +813,11 @@ fn setup_logging() {
.event_format(fmt::format().compact().with_timer(TimeFormatter::new()))
.with_writer(std::io::stderr)
.with_filter(fmt_filter_layer);
tracing_subscriber::registry().with(l).init();
}
Ok(())
}
fn filters(config: LoggingConfig) -> EnvFilter {
......@@ -826,17 +1198,8 @@ pub mod tests {
}
"#;
#[tracing::instrument(
skip_all,
fields(
span_id = "abd16e319329445f",
trace_id = "2adfd24468724599bb9a4990dc342288"
)
)]
#[tracing::instrument(skip_all)]
async fn parent() {
tracing::Span::current().record("trace_id", "invalid");
tracing::Span::current().record("span_id", "invalid");
tracing::Span::current().record("span_name", "invalid");
tracing::trace!(message = "parent!");
if let Some(my_ctx) = get_distributed_tracing_context() {
tracing::info!(my_trace_id = my_ctx.trace_id);
......@@ -907,16 +1270,43 @@ pub mod tests {
let lines = load_log(file_name)?;
// 1. Validate my_trace_id matches parent's trace ID
let parent_trace_id = Uuid::parse_str("2adfd24468724599bb9a4990dc342288")
.unwrap()
.simple()
// 1. Extract the dynamically generated trace ID and validate consistency
// All logs should have the same trace_id since they're part of the same trace
let trace_id = lines
.first()
.and_then(|log_line| log_line.get("trace_id"))
.and_then(|v| v.as_str())
.expect("First log line should have a trace_id")
.to_string();
// Verify trace_id is not a zero/invalid ID
assert_ne!(
trace_id, "00000000000000000000000000000000",
"trace_id should not be a zero/invalid ID"
);
assert!(
!trace_id.chars().all(|c| c == '0'),
"trace_id should not be all zeros"
);
// Verify all logs have the same trace_id
for log_line in &lines {
if let Some(line_trace_id) = log_line.get("trace_id") {
assert_eq!(
line_trace_id.as_str().unwrap(),
&trace_id,
"All logs should have the same trace_id"
);
}
}
// Validate my_trace_id matches the actual trace ID
for log_line in &lines {
if let Some(my_trace_id) = log_line.get("my_trace_id") {
assert_eq!(
my_trace_id,
&serde_json::Value::String(parent_trace_id.clone())
&serde_json::Value::String(trace_id.clone()),
"my_trace_id should match the trace_id from distributed tracing context"
);
}
}
......
lib/runtime/src/pipeline/network/egress/addressed_router.rs
View file @ 1f92dd54
......@@ -8,6 +8,7 @@ use tracing as log;
use super::*;
use crate::logging::DistributedTraceContext;
use crate::logging::get_distributed_tracing_context;
use crate::logging::inject_otel_context_into_nats_headers;
use crate::{Result, protocols::maybe_error::MaybeError};
use tokio_stream::{StreamExt, StreamNotifyClose, wrappers::ReceiverStream};
use tracing::Instrument;
......@@ -145,12 +146,13 @@ where
// Enables span to be created in push_endpoint before
// payload is parsed
// Prepare trace headers using the OpenTelemetry injector pattern
// This handles traceparent and tracestate headers according to W3C Trace Context standard
let mut headers = HeaderMap::new();
inject_otel_context_into_nats_headers(&mut headers, None);
// Add additional custom headers that aren't handled by the OpenTelemetry propagator
if let Some(trace_context) = get_distributed_tracing_context() {
headers.insert("traceparent", trace_context.create_traceparent());
if let Some(tracestate) = trace_context.tracestate {
headers.insert("tracestate", tracestate);
}
if let Some(x_request_id) = trace_context.x_request_id {
headers.insert("x-request-id", x_request_id);
}
......
lib/runtime/src/pipeline/network/ingress/push_endpoint.rs
View file @ 1f92dd54
......@@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use super::*;
use crate::SystemHealth;
use crate::config::HealthStatus;
use crate::logging::TraceParent;
use crate::logging::make_handle_payload_span;
use crate::protocols::LeaseId;
use anyhow::Result;
use async_nats::service::endpoint::Endpoint;
......@@ -94,33 +94,23 @@ impl PushEndpoint {
let notify_clone = notify.clone();
// Handle headers here for tracing
let mut traceparent = TraceParent::default();
if let Some(headers) = req.message.headers.as_ref() {
traceparent = TraceParent::from_headers(headers);
}
let span = if let Some(headers) = req.message.headers.as_ref() {
make_handle_payload_span(
headers,
component_name.as_ref(),
endpoint_name.as_ref(),
namespace.as_ref(),
instance_id,
)
} else {
tracing::info_span!("handle_payload")
};
tokio::spawn(async move {
tracing::trace!(instance_id, "handling new request");
let result = ingress
.handle_payload(req.message.payload)
.instrument(
// Create span with trace ids as set
// in headers.
tracing::info_span!(
"handle_payload",
component = component_name.as_ref(),
endpoint = endpoint_name.as_ref(),
namespace = namespace.as_ref(),
instance_id = instance_id,
trace_id = traceparent.trace_id,
parent_id = traceparent.parent_id,
x_request_id = traceparent.x_request_id,
x_dynamo_request_id = traceparent.x_dynamo_request_id,
tracestate = traceparent.tracestate
),
)
.instrument(span)
.await;
match result {
Ok(_) => {
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment