diff --git a/Cargo.lock b/Cargo.lock index 1b3c34d8dc..62fa2e4356 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3487,12 +3487,12 @@ dependencies = [ [[package]] name = "deranged" -version = "0.3.11" +version = "0.5.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b42b6fa04a440b495c8b04d0e71b707c585f83cb9cb28cf8cd0d976c315e31b4" +checksum = "ececcb659e7ba858fb4f10388c250a7252eb0a27373f1a72b8748afdd248e587" dependencies = [ "powerfmt", - "serde", + "serde_core", ] [[package]] @@ -6269,9 +6269,9 @@ dependencies = [ [[package]] name = "num-conv" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9" +checksum = "cf97ec579c3c42f953ef76dbf8d55ac91fb219dde70e49aa4a6b7d74e9919050" [[package]] name = "num-format" @@ -10339,9 +10339,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.37" +version = "0.3.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "35e7868883861bd0e56d9ac6efcaaca0d6d5d82a2a7ec8209ff492c07cf37b21" +checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" dependencies = [ "deranged", "itoa", @@ -10349,22 +10349,22 @@ dependencies = [ "num-conv", "num_threads", "powerfmt", - "serde", + "serde_core", "time-core", "time-macros", ] [[package]] name = "time-core" -version = "0.1.2" +version = "0.1.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef927ca75afb808a4d64dd374f00a2adf8d0fcff8e7b184af886c3c87ec4a3f3" +checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" [[package]] name = "time-macros" -version = "0.2.19" +version = "0.2.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2834e6017e3e5e4b9834939793b282bc03b37a3336245fa820e35e233e2a85de" +checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215" dependencies = [ "num-conv", "time-core", diff --git a/src/common/utils/auth.rs b/src/common/utils/auth.rs index 8a95ee9505..1a484aa51f 100644 --- a/src/common/utils/auth.rs +++ b/src/common/utils/auth.rs @@ -1018,6 +1018,7 @@ where || path.contains("/format_query") || path.contains("/prometheus/api/v1/series") || path.contains("/traces/latest") + || (path.contains("/traces/") && path.ends_with("/dag")) || path.contains("clusters") || path.contains("query_manager") || path.contains("/short") diff --git a/src/handler/http/request/traces/dag.rs b/src/handler/http/request/traces/dag.rs new file mode 100644 index 0000000000..b849008d02 --- /dev/null +++ b/src/handler/http/request/traces/dag.rs @@ -0,0 +1,370 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use axum::{extract::Path, http::HeaderMap, response::Response}; +use config::{get_config, meta::stream::StreamType, metrics, utils::json}; +use hashbrown::HashMap; +use serde::Serialize; +use tracing::{Instrument, Span}; + +use crate::{ + common::{ + meta::http::HttpResponse as MetaHttpResponse, + utils::{ + auth::UserEmail, + http::{get_or_create_trace_id, get_use_cache_from_request}, + }, + }, + handler::http::{ + extractors::Headers, request::search::error_utils::map_error_to_http_response, + }, + service::search as SearchService, +}; + +/// GetTraceDAG +/// +/// #{"ratelimit_module":"Traces", "ratelimit_module_operation":"list"}# +#[utoipa::path( + get, + path = "/{org_id}/{stream_name}/traces/{trace_id}/dag", + context_path = "/api", + tag = "Traces", + operation_id = "GetTraceDAG", + summary = "Get trace DAG structure", + description = "Retrieves the DAG (Directed Acyclic Graph) structure of all spans for a specific trace. Returns nodes (spans) and edges (parent-child relationships) that can be visualized as a trace execution graph. Each node contains span details like service name, operation, and status information.", + security( + ("Authorization"= []) + ), + params( + ("org_id" = String, Path, description = "Organization name"), + ("stream_name" = String, Path, description = "Stream name"), + ("trace_id" = String, Path, description = "Trace ID"), + ("start_time" = i64, Query, description = "start time in microseconds"), + ("end_time" = i64, Query, description = "end time in microseconds"), + ("timeout" = Option, Query, description = "timeout, seconds"), + ), + responses( + (status = 200, description = "Success", content_type = "application/json", body = Object, example = json!({ + "trace_id": "12345678", + "nodes": [ + { + "span_id": "span1", + "parent_span_id": null, + "service_name": "frontend", + "operation_name": "GET /api", + "span_status": "OK" + } + ], + "edges": [ + { + "from": "span1", + "to": "span2" + } + ] + })), + (status = 400, description = "Failure", content_type = "application/json", body = ()), + (status = 500, description = "Failure", content_type = "application/json", body = ()), + ) +)] +pub async fn get_trace_dag( + Path((org_id, stream_name, trace_id)): Path<(String, String, String)>, + axum::extract::Query(query): axum::extract::Query>, + headers: HeaderMap, + Headers(user_email): Headers, +) -> Response { + let start = std::time::Instant::now(); + let cfg = get_config(); + + #[cfg(feature = "enterprise")] + { + if let Err(e) = crate::service::search::check_search_allowed(&org_id, Some(&stream_name)) { + return MetaHttpResponse::too_many_requests(e.to_string()); + } + } + + let (http_span, internal_trace_id) = if cfg.common.tracing_search_enabled { + let uuid_v7_trace_id = config::ider::generate_trace_id(); + let span = tracing::info_span!( + "/api/{org_id}/{stream_name}/traces/{trace_id}/dag", + org_id = org_id.clone(), + stream_name = stream_name.clone(), + trace_id = uuid_v7_trace_id.clone() + ); + + (span, uuid_v7_trace_id) + } else { + let internal_trace_id = get_or_create_trace_id(&headers, &Span::none()); + (Span::none(), internal_trace_id) + }; + let user_id = &user_email.user_id; + + // Check permissions on stream + #[cfg(feature = "enterprise")] + { + use o2_openfga::meta::mapping::OFGA_MODELS; + + use crate::{ + common::utils::auth::{AuthExtractor, is_root_user}, + service::users::get_user, + }; + if !is_root_user(user_id) { + let user: config::meta::user::User = get_user(Some(&org_id), user_id).await.unwrap(); + let stream_type_str = StreamType::Traces.as_str(); + + if !crate::handler::http::auth::validator::check_permissions( + user_id, + AuthExtractor { + auth: "".to_string(), + method: "GET".to_string(), + o2_type: format!( + "{}:{}", + OFGA_MODELS + .get(stream_type_str) + .map_or(stream_type_str, |model| model.key), + stream_name + ), + org_id: org_id.clone(), + bypass_check: false, + parent_id: "".to_string(), + }, + user.role, + user.is_external, + ) + .await + { + return MetaHttpResponse::forbidden("Unauthorized Access"); + } + } + // Check permissions on stream ends + } + + // Parse required start_time and end_time from query parameters + let start_time = match query.get("start_time") { + Some(v) => match v.parse::() { + Ok(val) => val, + Err(_) => return MetaHttpResponse::bad_request("Invalid start_time parameter"), + }, + None => return MetaHttpResponse::bad_request("start_time parameter is required"), + }; + + let end_time = match query.get("end_time") { + Some(v) => match v.parse::() { + Ok(val) => val, + Err(_) => return MetaHttpResponse::bad_request("Invalid end_time parameter"), + }, + None => return MetaHttpResponse::bad_request("end_time parameter is required"), + }; + + // Validate time range + if start_time >= end_time { + return MetaHttpResponse::bad_request("start_time must be less than end_time"); + } + + let timeout = query + .get("timeout") + .map_or(0, |v| v.parse::().unwrap_or(0)); + + // Query all spans for this trace_id + let query_sql = format!( + "SELECT span_id, trace_id, service_name, operation_name, span_status, \ + reference_parent_span_id, start_time, end_time, _o2_llm_observation_type \ + FROM {stream_name} \ + WHERE trace_id = '{trace_id}'" + ); + + let req = config::meta::search::Request { + query: config::meta::search::Query { + sql: query_sql.to_string(), + from: 0, + size: 10000, // Large enough for most traces + start_time, + end_time, + quick_mode: false, + query_type: "".to_string(), + track_total_hits: false, + uses_zo_fn: false, + query_fn: None, + action_id: None, + skip_wal: false, + sampling_config: None, + sampling_ratio: None, + streaming_output: false, + streaming_id: None, + histogram_interval: 0, + }, + encoding: config::meta::search::RequestEncoding::Empty, + regions: vec![], + clusters: vec![], + timeout, + search_type: None, + search_event_context: None, + use_cache: get_use_cache_from_request(&query), + clear_cache: false, + local_mode: None, + }; + + let stream_type = StreamType::Traces; + let user_id_opt = Some(user_id.to_string()); + + let search_res = SearchService::cache::search( + &internal_trace_id, + &org_id, + stream_type, + user_id_opt, + &req, + "".to_string(), + false, + None, + false, + ) + .instrument(http_span) + .await; + + let resp_search = match search_res { + Ok(res) => res, + Err(err) => { + let time = start.elapsed().as_secs_f64(); + metrics::HTTP_RESPONSE_TIME + .with_label_values(&[ + "/api/org/traces/dag", + "500", + &org_id, + stream_type.as_str(), + "", + "", + ]) + .observe(time); + metrics::HTTP_INCOMING_REQUESTS + .with_label_values(&[ + "/api/org/traces/dag", + "500", + &org_id, + stream_type.as_str(), + "", + "", + ]) + .inc(); + log::error!("get trace dag data error: {err:?}"); + return map_error_to_http_response(&err, Some(internal_trace_id)); + } + }; + + if resp_search.hits.is_empty() { + return MetaHttpResponse::not_found("Trace not found"); + } + + // Build DAG structure + let mut nodes: Vec = Vec::with_capacity(resp_search.hits.len()); + let mut edges: Vec = Vec::new(); + + for item in resp_search.hits { + let span_id = item + .get("span_id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let parent_span_id = item + .get("reference_parent_span_id") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()); + + // Create node + let node = SpanNode { + span_id: span_id.clone(), + parent_span_id: parent_span_id.clone(), + service_name: item + .get("service_name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + operation_name: item + .get("operation_name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + span_status: item + .get("span_status") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + start_time: item.get("start_time").and_then(|v| v.as_i64()).unwrap_or(0), + end_time: item.get("end_time").and_then(|v| v.as_i64()).unwrap_or(0), + llm_observation_type: item + .get("_o2_llm_observation_type") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()), + }; + nodes.push(node); + + // Create edge if there's a parent + if let Some(parent_id) = parent_span_id { + edges.push(SpanEdge { + from: parent_id, + to: span_id, + }); + } + } + + let time = start.elapsed().as_secs_f64(); + metrics::HTTP_RESPONSE_TIME + .with_label_values(&[ + "/api/org/traces/dag", + "200", + &org_id, + stream_type.as_str(), + "", + "", + ]) + .observe(time); + metrics::HTTP_INCOMING_REQUESTS + .with_label_values(&[ + "/api/org/traces/dag", + "200", + &org_id, + stream_type.as_str(), + "", + "", + ]) + .inc(); + + let mut resp: HashMap<&str, json::Value> = HashMap::new(); + resp.insert("took", json::Value::from((time * 1000.0) as usize)); + resp.insert("trace_id", json::Value::from(trace_id)); + resp.insert("nodes", json::to_value(nodes).unwrap()); + resp.insert("edges", json::to_value(edges).unwrap()); + resp.insert("internal_trace_id", json::Value::from(internal_trace_id)); + + MetaHttpResponse::json(resp) +} + +#[derive(Debug, Serialize)] +struct SpanNode { + span_id: String, + parent_span_id: Option, + service_name: String, + operation_name: String, + span_status: String, + start_time: i64, + end_time: i64, + llm_observation_type: Option, +} + +#[derive(Debug, Serialize)] +struct SpanEdge { + from: String, + to: String, +} diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 51289d1d0d..20320c5c4a 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -47,6 +47,8 @@ use crate::{ service::{search as SearchService, traces}, }; +pub mod dag; + /// TracesIngest #[utoipa::path( post, diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index 691c665338..17cf3de207 100644 --- a/src/handler/http/router/mod.rs +++ b/src/handler/http/router/mod.rs @@ -571,6 +571,7 @@ pub fn service_routes() -> Router { // Traces .route("/{org_id}/{stream_name}/traces/latest", get(traces::get_latest_traces)) + .route("/{org_id}/{stream_name}/traces/{trace_id}/dag", get(traces::dag::get_trace_dag)) // Metrics .route("/{org_id}/ingest/metrics/_json", post(metrics::ingest::json)) diff --git a/src/service/traces/otel/pricing.rs b/src/service/traces/otel/pricing.rs index 1b7340d90e..2e51374305 100644 --- a/src/service/traces/otel/pricing.rs +++ b/src/service/traces/otel/pricing.rs @@ -242,4 +242,63 @@ mod tests { assert_eq!(output_cost, 0.0); assert_eq!(total_cost, 0.0); } + + #[test] + fn test_calculate_token_count_known_model() { + // Test with a known OpenAI model + let prompt = "Hello, world!"; + let token_count = calculate_token_count("gpt-4", prompt); + assert!(token_count > 0); + // "Hello, world!" should be encoded to a few tokens + assert!(token_count < 10); + } + + #[test] + fn test_calculate_token_count_unknown_model() { + // Test with an unknown model (should fallback to o200k_base) + let prompt = "This is a test prompt"; + let token_count = calculate_token_count("unknown-model-xyz", prompt); + assert!(token_count > 0); + // Should still return a valid token count using fallback encoding + } + + #[test] + fn test_calculate_token_count_empty_prompt() { + // Test with empty prompt + let token_count = calculate_token_count("gpt-4", ""); + // Empty prompt might still have some tokens (like BOS token) or be 0 + assert!(token_count >= 0); + } + + #[test] + fn test_calculate_token_count_long_prompt() { + // Test with a longer prompt + let prompt = "The quick brown fox jumps over the lazy dog. ".repeat(10); + let token_count = calculate_token_count("gpt-4", &prompt); + assert!(token_count > 0); + // Longer prompt should have more tokens + assert!(token_count > 10); + } + + #[test] + fn test_calculate_token_count_special_characters() { + // Test with special characters and unicode + let prompt = "Hello! 你好 🌍 世界"; + let token_count = calculate_token_count("gpt-4", prompt); + assert!(token_count > 0); + } + + #[test] + fn test_calculate_token_count_different_models() { + // Test that different models produce consistent results (or at least valid results) + let prompt = "Test prompt"; + let gpt4_count = calculate_token_count("gpt-4", prompt); + let gpt35_count = calculate_token_count("gpt-3.5-turbo", prompt); + let claude_count = calculate_token_count("claude-3-5-sonnet", prompt); + + // All should return valid token counts + assert!(gpt4_count > 0); + assert!(gpt35_count > 0); + assert!(claude_count > 0); + } } diff --git a/web/src/plugins/traces/TraceDAG.vue b/web/src/plugins/traces/TraceDAG.vue new file mode 100644 index 0000000000..1df860644d --- /dev/null +++ b/web/src/plugins/traces/TraceDAG.vue @@ -0,0 +1,633 @@ + + + + + + + diff --git a/web/src/plugins/traces/TraceDetails.vue b/web/src/plugins/traces/TraceDetails.vue index e4f93789fe..22b17521a8 100644 --- a/web/src/plugins/traces/TraceDetails.vue +++ b/web/src/plugins/traces/TraceDetails.vue @@ -376,13 +376,31 @@ along with this program. If not, see . ref="parentContainer" >
- -
+ + + + + + + + +
+ +
. />
+
+ + +
+
+ +
+ +
+
+
+
+ +
+
@@ -492,6 +561,7 @@ import useTraces from "@/composables/useTraces"; import { computed } from "vue"; import TraceDetailsSidebar from "./TraceDetailsSidebar.vue"; import TraceTree from "./TraceTree.vue"; +import TraceDAG from "./TraceDAG.vue"; import TraceHeader from "./TraceHeader.vue"; import { useStore } from "vuex"; import { @@ -521,6 +591,7 @@ import useNotifications from "@/composables/useNotifications"; import { parseUsageDetails, parseCostDetails, + isLLMTrace, } from "@/utils/llmUtils"; export default defineComponent({ @@ -595,6 +666,7 @@ export default defineComponent({ ShareButton, TraceDetailsSidebar, TraceTree, + TraceDAG, TraceHeader, TraceTimelineIcon, ServiceMapIcon, @@ -607,6 +679,9 @@ export default defineComponent({ setup(props, { emit }) { const traceTree: any = ref([]); const spanMap: any = ref({}); + const activeTab = ref("timeline"); + + const { searchObj, getUrlQueryParams } = useTraces(); const baseTracePosition: any = ref({}); const collapseMapping: any = ref({}); @@ -680,6 +755,12 @@ export default defineComponent({ const throttledResizing = ref(null); + // DAG panel resize state + const dagLeftWidth: Ref = ref(50); // percentage + const dagInitialX: Ref = ref(0); + const dagInitialWidth: Ref = ref(0); + const throttledDagResizing = ref(null); + // Calculate sidebar width based on leftWidth // Sidebar should take ~84% of the remaining space after left panel const sidebarWidth = computed(() => { @@ -789,6 +870,13 @@ export default defineComponent({ return searchObj.data.traceDetails.spanList; }); + // Check if the trace contains any LLM spans + const hasLLMSpans = computed(() => { + const spans = effectiveSpanList.value; + if (!spans || spans.length === 0) return false; + return spans.some((span: any) => isLLMTrace(span)); + }); + const showTraceDetails = ref(false); const currentIndex = ref(0); const searchResults = ref(0); @@ -1751,6 +1839,33 @@ export default defineComponent({ document.body.classList.remove("no-select"); }; + // DAG panel resize handlers + const startDagResize = (event: MouseEvent) => { + dagInitialX.value = event.clientX; + dagInitialWidth.value = dagLeftWidth.value; + + throttledDagResizing.value = throttle(dagResizing, 16); + window.addEventListener("mousemove", throttledDagResizing.value); + window.addEventListener("mouseup", stopDagResize); + document.body.classList.add("no-select"); + }; + + const dagResizing = (event: MouseEvent) => { + if (!parentContainer.value) return; + const containerWidth = parentContainer.value.clientWidth; + const deltaX = event.clientX - dagInitialX.value; + const deltaPercent = (deltaX / containerWidth) * 100; + const newWidth = dagInitialWidth.value + deltaPercent; + // Constrain between 20% and 80% + dagLeftWidth.value = Math.max(20, Math.min(80, newWidth)); + }; + + const stopDagResize = () => { + window.removeEventListener("mousemove", throttledDagResizing.value); + window.removeEventListener("mouseup", stopDagResize); + document.body.classList.remove("no-select"); + }; + const toggleTimeline = () => { isTimelineExpanded.value = !isTimelineExpanded.value; }; @@ -1872,6 +1987,10 @@ export default defineComponent({ } }; + const handleDAGNodeClick = (spanId: string) => { + updateSelectedSpan(spanId); + }; + const handleBackOrClose = () => { if (props.mode === "embedded") { emit("close"); @@ -1928,6 +2047,7 @@ export default defineComponent({ return { router, t, + activeTab, traceTree, collapseMapping, traceRootSpan, @@ -2010,6 +2130,12 @@ export default defineComponent({ effectiveSpanList, // New event handlers handleBackOrClose, + handleDAGNodeClick, + // DAG resize + dagLeftWidth, + startDagResize, + // LLM traces check + hasLLMSpans, }; }, }); @@ -2068,11 +2194,15 @@ $traceChartCollapseHeight: 42px; min-height: 0; position: relative; padding-bottom: 0.5rem; + display: flex; + flex-direction: column; + flex: 1; } .trace-tree-wrapper { overflow: hidden; - height: calc(100% - 2.5rem); + flex: 1; + min-height: 0; box-sizing: border-box; display: flex; flex-direction: column; @@ -2113,6 +2243,50 @@ $traceChartCollapseHeight: 42px; .toolbar-operation-name { max-width: 225px; } + +.dag-left-panel { + height: calc(100vh - 200px); + padding: 16px; + min-width: 0; + overflow: hidden; +} + +.dag-right-panel { + overflow-y: auto; + overflow-x: hidden; + min-height: 0; +} + +.dag-resizer { + width: 8px; + cursor: col-resize; + display: flex; + align-items: center; + justify-content: center; + flex-shrink: 0; + position: relative; + z-index: 10; + + &:hover .dag-resizer-line { + background-color: var(--o2-theme-color, #1976d2); + } +} + +.dag-resizer-line { + width: 3px; + height: 100%; + background-color: #e0e0e0; + border-radius: 2px; + transition: background-color 0.2s ease; +} + +body.body--dark .dag-resizer-line { + background-color: #3c3c3c; +} + +body.body--dark .dag-resizer:hover .dag-resizer-line { + background-color: #90caf9; +}