From e4e7274ec827486366a837b2aa688c7efb9fa028 Mon Sep 17 00:00:00 2001 From: ByteBaker <42913098+ByteBaker@users.noreply.github.com> Date: Tue, 28 Oct 2025 11:48:08 +0530 Subject: [PATCH] feat: add service graph as enterprise-only feature Service graph automatically discovers service topology from distributed traces. OSS includes: - API routing endpoints (gated with enterprise feature flag) - UI components (functional only with enterprise backend) - Configuration options Enterprise includes: - 256-shard concurrent edge store with LRU eviction - Span matching and processing logic - Prometheus metrics and background workers --- .gitignore | 1 + Cargo.lock | 4 +- src/config/src/config.rs | 33 + src/handler/http/request/traces/mod.rs | 3 + src/handler/http/router/mod.rs | 4 +- src/main.rs | 7 + src/service/traces/mod.rs | 54 +- src/service/traces/service_graph/api.rs | 100 +++ src/service/traces/service_graph/mod.rs | 32 + web/src/composables/shared/router.ts | 13 + web/src/layouts/MainLayout.vue | 7 + web/src/plugins/traces/ServiceGraph.vue | 993 ++++++++++++++++++++++++ web/src/services/service_graph.ts | 34 + 13 files changed, 1278 insertions(+), 7 deletions(-) create mode 100644 src/service/traces/service_graph/api.rs create mode 100644 src/service/traces/service_graph/mod.rs create mode 100644 web/src/plugins/traces/ServiceGraph.vue create mode 100644 web/src/services/service_graph.ts diff --git a/.gitignore b/.gitignore index 07af14f72a..851b1ca5f6 100644 --- a/.gitignore +++ b/.gitignore @@ -35,3 +35,4 @@ tests/ui-testing/playwright-tests/utils/auth/user.json /tests/ui-testing/utils/test-mapping/archived-utils /tests/ui-testing/playwright-results /tests/ui-testing/testDinoPOC/MD Files +*.bak diff --git a/Cargo.lock b/Cargo.lock index cdb0c832dd..7d5c82fd08 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9491,9 +9491,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.13.2" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" dependencies = [ "serde", ] diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 97451b8492..6765773906 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -496,6 +496,7 @@ pub struct Config { pub health_check: HealthCheck, pub encryption: Encryption, pub enrichment_table: EnrichmentTable, + pub service_graph: ServiceGraph, } #[derive(EnvConfig, Default)] @@ -2115,6 +2116,38 @@ pub struct EnrichmentTable { pub merge_interval: u64, } +/// Service Graph Configuration +/// +/// Note: Worker count is fixed at 256 (one per shard) for optimal concurrency. +/// No async channels are used in the current implementation. +#[derive(EnvConfig, Default)] +pub struct ServiceGraph { + #[env_config( + name = "ZO_SGRAPH_ENABLED", + default = false, + help = "Enable service graph feature" + )] + pub enabled: bool, + #[env_config( + name = "ZO_SGRAPH_WAIT_DURATION_MS", + default = 10000, + help = "Wait duration for span pairing in milliseconds" + )] + pub wait_duration_ms: u64, + #[env_config( + name = "ZO_SGRAPH_MAX_ITEMS_PER_SHARD", + default = 100000, + help = "Maximum items per shard in edge store (256 shards total)" + )] + pub max_items_per_shard: usize, + #[env_config( + name = "ZO_SGRAPH_CLEANUP_INTERVAL_MS", + default = 2000, + help = "Cleanup interval for expired edges in milliseconds" + )] + pub cleanup_interval_ms: u64, +} + pub fn init() -> Config { if let Err(e) = load_config() { log::error!("Failed to load config {e}"); diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 5d497b328e..0d903f3a57 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -26,6 +26,9 @@ use hashbrown::HashMap; use serde::Serialize; use tracing::{Instrument, Span}; +// Re-export service graph API handlers +#[cfg(feature = "enterprise")] +pub use crate::service::traces::service_graph::{get_service_graph_metrics, get_store_stats}; use crate::{ common::{ meta::{self, http::HttpResponse as MetaHttpResponse}, diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index a94a28bc00..f66aa042cb 100644 --- a/src/handler/http/router/mod.rs +++ b/src/handler/http/router/mod.rs @@ -612,7 +612,9 @@ pub fn get_service_routes(svc: &mut web::ServiceConfig) { .service(re_pattern::delete) .service(re_pattern::test) .service(domain_management::get_domain_management_config) - .service(domain_management::set_domain_management_config); + .service(domain_management::set_domain_management_config) + .service(traces::get_service_graph_metrics) + .service(traces::get_store_stats); #[cfg(feature = "cloud")] let service = service diff --git a/src/main.rs b/src/main.rs index ae3611e9fe..ebccdf2d57 100644 --- a/src/main.rs +++ b/src/main.rs @@ -309,6 +309,13 @@ async fn main() -> Result<(), anyhow::Error> { panic!("job init failed: {e}"); } + // init service graph workers + #[cfg(feature = "enterprise")] + if cfg.service_graph.enabled { + log::info!("Initializing service graph background workers"); + openobserve::service::traces::service_graph::init_background_workers(); + } + // Register job runtime for metrics collection if let Ok(handle) = tokio::runtime::Handle::try_current() { openobserve::service::runtime_metrics::register_runtime("job".to_string(), handle); diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index d6e4c48086..8ad35782bf 100644 --- a/src/service/traces/mod.rs +++ b/src/service/traces/mod.rs @@ -41,6 +41,9 @@ use opentelemetry_proto::tonic::{ use prost::Message; use serde_json::Map; +#[cfg(feature = "enterprise")] +pub mod service_graph; + #[cfg(feature = "cloud")] use crate::service::stream::get_stream; use crate::{ @@ -343,22 +346,65 @@ pub async fn handle_otlp_request( } let local_val = Span { trace_id: trace_id.clone(), - span_id, + span_id: span_id.clone(), span_kind: span.kind.to_string(), - span_status: get_span_status(span.status), + span_status: get_span_status(span.status.clone()), operation_name: span.name.clone(), start_time, end_time, duration: (end_time - start_time) / 1000, // microseconds - reference: span_ref, + reference: span_ref.clone(), service_name: service_name.clone(), - attributes: span_att_map, + attributes: span_att_map.clone(), service: service_att_map.clone(), flags: 1, // TODO add appropriate value events: json::to_string(&events).unwrap(), links: json::to_string(&links).unwrap(), }; + // Process span for service graph if enabled + #[cfg(feature = "enterprise")] + if cfg.service_graph.enabled { + // Wrap in catch_unwind to prevent panics from crashing trace ingestion + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + log::trace!( + "[ServiceGraph] Processing span: trace_id={}, span_id={}", + trace_id, + span_id + ); + let parent_span_id_opt = + span_ref.get(PARENT_SPAN_ID).map(|s| Arc::from(s.as_str())); + // Convert HashMap to json::Map + let mut attrs_map = json::Map::new(); + for (k, v) in &span_att_map { + attrs_map.insert(k.clone(), v.clone()); + } + let graph_span = service_graph::span_to_graph_span( + Arc::from(trace_id.as_str()), + Arc::from(span_id.as_str()), + parent_span_id_opt, + Arc::from(service_name.as_str()), + &local_val.span_kind, + start_time, + end_time, + &local_val.span_status, + &attrs_map, + ); + service_graph::process_span(org_id, graph_span); + })); + + if let Err(e) = result { + log::error!( + "[ServiceGraph] Panic caught during span processing: {:?}", + e + ); + // Increment error metric + service_graph::SERVICE_GRAPH_DROPPED_SPANS + .with_label_values(&[org_id]) + .inc(); + } + } + let mut value: json::Value = json::to_value(local_val).unwrap(); // add timestamp value.as_object_mut().unwrap().insert( diff --git a/src/service/traces/service_graph/api.rs b/src/service/traces/service_graph/api.rs new file mode 100644 index 0000000000..799b2cefc6 --- /dev/null +++ b/src/service/traces/service_graph/api.rs @@ -0,0 +1,100 @@ +// Copyright 2025 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 actix_web::{HttpResponse, web}; +use config::utils::json; +use prometheus::{Encoder, TextEncoder}; +use serde::Deserialize; + +/// Query parameters for service graph API +#[derive(Debug, Deserialize)] +pub struct ServiceGraphQuery { + pub start_time: Option, + pub end_time: Option, + pub filter: Option, +} + +/// Handler for service graph metrics endpoint (Prometheus format) +/// +/// Returns Prometheus metrics filtered by org_id. +/// Metrics are filtered by the `org_id` label to ensure multi-tenant isolation. +#[actix_web::get("/{org_id}/traces/service_graph/metrics")] +pub async fn get_service_graph_metrics( + org_id: web::Path, +) -> Result { + let metric_families = prometheus::gather(); + let encoder = TextEncoder::new(); + let mut buffer = vec![]; + + // Filter metrics to only include those matching the requested org_id + let filtered_families: Vec<_> = metric_families + .into_iter() + .filter_map(|mut family| { + // Filter out metrics that don't belong to this org + let filtered_metrics: Vec<_> = family + .take_metric() + .into_iter() + .filter(|metric| { + // Check if metric has org_id label matching the requested org + metric + .get_label() + .iter() + .any(|label| label.name() == "org_id" && label.value() == org_id.as_str()) + }) + .collect(); + + if filtered_metrics.is_empty() { + None + } else { + family.set_metric(filtered_metrics); + Some(family) + } + }) + .collect(); + + encoder.encode(&filtered_families, &mut buffer).unwrap(); + + Ok(HttpResponse::Ok() + .content_type("text/plain; version=0.0.4") + .body(buffer)) +} + +/// Get service graph store statistics +/// +/// Returns statistics filtered by org_id to ensure multi-tenant isolation. +#[actix_web::get("/{org_id}/traces/service_graph/stats")] +pub async fn get_store_stats(org_id: web::Path) -> Result { + let store = o2_enterprise::enterprise::service_graph::get_store(); + let org_sizes = store.get_org_sizes(); + let org_capacities = store.get_org_capacity_utilization(); + + // Get stats for the requested org only + let org_store_size = org_sizes.get(org_id.as_str()).copied().unwrap_or(0); + let org_capacity_util = org_capacities.get(org_id.as_str()).copied().unwrap_or(0.0); + + let stats = json::json!({ + "enabled": config::get_config().service_graph.enabled, + "store_size": org_store_size, + "capacity_utilization_percent": org_capacity_util, + "shard_count": store.shard_count(), + "config": { + "max_items_per_shard": store.config().max_items_per_shard, + "total_capacity": store.config().max_items_per_shard * store.shard_count(), + "wait_duration_ms": store.config().wait_duration.as_millis(), + } + }); + + Ok(HttpResponse::Ok().json(stats)) +} diff --git a/src/service/traces/service_graph/mod.rs b/src/service/traces/service_graph/mod.rs new file mode 100644 index 0000000000..4deaac164b --- /dev/null +++ b/src/service/traces/service_graph/mod.rs @@ -0,0 +1,32 @@ +// Copyright 2025 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 . + +//! Service Graph Module +//! +//! This module provides API endpoints for the service graph feature. +//! The core logic is implemented in the enterprise repository. + +#[cfg(feature = "enterprise")] +pub mod api; + +// Re-export API handlers for routing +#[cfg(feature = "enterprise")] +pub use api::{get_service_graph_metrics, get_store_stats}; +// Re-export enterprise types and functions for internal use +#[cfg(feature = "enterprise")] +pub use o2_enterprise::enterprise::service_graph::{ + ConnectionType, SERVICE_GRAPH_DROPPED_SPANS, SpanForGraph, SpanKind, init_background_workers, + process_span, shutdown_workers, span_to_graph_span, +}; diff --git a/web/src/composables/shared/router.ts b/web/src/composables/shared/router.ts index 5895cd868e..cfd5d358cd 100644 --- a/web/src/composables/shared/router.ts +++ b/web/src/composables/shared/router.ts @@ -30,6 +30,7 @@ const AppMetrics = () => import("@/plugins/metrics/Index.vue"); const AppTraces = () => import("@/plugins/traces/Index.vue"); const TraceDetails = () => import("@/plugins/traces/TraceDetails.vue"); +const ServiceGraph = () => import("@/plugins/traces/ServiceGraph.vue"); const ViewDashboard = () => import("@/views/Dashboards/ViewDashboard.vue"); const AddPanel = () => import("@/views/Dashboards/addPanel/AddPanel.vue"); @@ -163,6 +164,18 @@ const useRoutes = () => { routeGuard(to, from, next); }, }, + { + path: "service-graph", + name: "serviceGraph", + component: ServiceGraph, + meta: { + keepAlive: true, + title: "Service Graph", + }, + beforeEnter(to: any, from: any, next: any) { + routeGuard(to, from, next); + }, + }, { name: "streamExplorer", path: "streams/stream-explore", diff --git a/web/src/layouts/MainLayout.vue b/web/src/layouts/MainLayout.vue index 6cfcf18d4a..7e5ee2bc64 100644 --- a/web/src/layouts/MainLayout.vue +++ b/web/src/layouts/MainLayout.vue @@ -631,6 +631,7 @@ import { outlinedDescription, outlinedCode, outlinedDevices, + outlinedBubbleChart, } from "@quasar/extras/material-icons-outlined"; import SlackIcon from "@/components/icons/SlackIcon.vue"; import ManagementIcon from "@/components/icons/ManagementIcon.vue"; @@ -806,6 +807,12 @@ export default defineComponent({ link: "/traces", name: "traces", }, + { + title: "Service Graph", + icon: outlinedBubbleChart, + link: "/service-graph", + name: "serviceGraph", + }, { title: t("menu.rum"), icon: outlinedDevices, diff --git a/web/src/plugins/traces/ServiceGraph.vue b/web/src/plugins/traces/ServiceGraph.vue new file mode 100644 index 0000000000..f38b1e8f33 --- /dev/null +++ b/web/src/plugins/traces/ServiceGraph.vue @@ -0,0 +1,993 @@ + + + + + diff --git a/web/src/services/service_graph.ts b/web/src/services/service_graph.ts new file mode 100644 index 0000000000..bf0bb8be75 --- /dev/null +++ b/web/src/services/service_graph.ts @@ -0,0 +1,34 @@ +// Copyright 2025 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 . + +import http from "./http"; + +const serviceGraphService = { + /** + * Get service graph metrics in Prometheus format + */ + getMetrics: (orgId: string) => { + return http().get(`/api/${orgId}/traces/service_graph/metrics`); + }, + + /** + * Get service graph store statistics + */ + getStats: (orgId: string) => { + return http().get(`/api/${orgId}/traces/service_graph/stats`); + }, +}; + +export default serviceGraphService;