fix(service-graph): run job on alert_manager nodes (#11159)

Changes

- Switch service graph job from ingester to alert_manager nodes
- Add UUID-based leader election to prevent duplicate writes across
multiple alert_manager instances
- Make query time range configurable via o2_config instead of a
hardcoded constant
- Skip processing streams with no new data since the last interval
(doc_time_max check)
- Downgrade no-trace-streams log from warn to debug

src/job/service_graph.rs

@@ -19,11 +19,11 @@ use o2_enterprise::enterprise::common::config::get_config as get_o2_config;
 pub async fn run() -> Result<(), anyhow::Error> {
     #[cfg(feature = "enterprise")]
     {
-        // Service graph processor writes data via ingestion pipeline
-        // and must run on ingester nodes only
-        if !LOCAL_NODE.is_ingester() {
+        // Only alert_manager nodes run the service graph job.
+        // (Ingester/querier/compactor/router nodes exit here.)
+        if !LOCAL_NODE.is_alert_manager() {
             log::info!(
-                "[SERVICE_GRAPH::JOB] Service graph processor disabled on non-ingester node (role: {:?})",
+                "[SERVICE_GRAPH::JOB] Service graph processor disabled on non-alert-manager node (role: {:?})",
                 LOCAL_NODE.role
             );
             return Ok(());
@@ -35,6 +35,29 @@ pub async fn run() -> Result<(), anyhow::Error> {
             "service_graph_processor",
             get_o2_config().service_graph.processing_interval_secs,
             {
+                // Leader election: only the alert_manager with the smallest UUID runs the job.
+                // This prevents duplicate writes when multiple alert_manager nodes are running.
+                let is_leader = match infra::cluster::get_cached_nodes(|node| {
+                    node.status == config::meta::cluster::NodeStatus::Online
+                        && node.is_alert_manager()
+                })
+                .await
+                {
+                    Some(mut nodes) if !nodes.is_empty() => {
+                        nodes.sort_by(|a, b| a.uuid.cmp(&b.uuid));
+                        nodes[0].uuid == LOCAL_NODE.uuid
+                    }
+                    // Cache empty or single-node — fall through and run.
+                    _ => true,
+                };
+
+                if !is_leader {
+                    log::debug!(
+                        "[SERVICE_GRAPH::JOB] Not leader alert_manager, skipping this interval"
+                    );
+                    continue;
+                }
+
                 log::debug!("[SERVICE_GRAPH::JOB] Running service graph processing");
                 if let Err(e) = crate::service::traces::service_graph::process_service_graph().await
                 {

src/job/stats.rs

@@ -59,7 +59,11 @@ fn file_list_update_stats() -> Option<tokio::task::JoinHandle<()>> {
 // For enterprise with super cluster enabled, we need to wait one round before starting this job
 // For OSS version, we will cache the stats first
 async fn cache_stream_stats() -> Option<tokio::task::JoinHandle<()>> {
-    if !LOCAL_NODE.is_ingester() && !LOCAL_NODE.is_querier() && !LOCAL_NODE.is_compactor() {
+    if !LOCAL_NODE.is_ingester()
+        && !LOCAL_NODE.is_querier()
+        && !LOCAL_NODE.is_compactor()
+        && !LOCAL_NODE.is_alert_manager()
+    {
         return None;
     }
 

src/service/traces/service_graph/processor.rs

@@ -29,9 +29,8 @@ use o2_enterprise::enterprise::common::config::get_config as get_o2_config;
 /// Called by compactor job
 #[cfg(feature = "enterprise")]
 pub async fn process_service_graph() -> Result<(), anyhow::Error> {
-    // Process last hour of traces
     let now = Utc::now().timestamp_micros();
-    let window_micros = super::DEFAULT_QUERY_WINDOW_MINUTES * 60 * 1_000_000;
+    let window_micros = get_o2_config().service_graph.query_time_range_minutes * 60 * 1_000_000;
     let start_time = now - window_micros;
 
     log::debug!(
@@ -48,18 +47,9 @@ pub async fn process_service_graph() -> Result<(), anyhow::Error> {
     );
 
     for (org_id, stream_name) in streams {
-        log::info!(
-            "[ServiceGraph] Processing stream {}/{}",
-            org_id,
-            stream_name
-        );
+        log::info!("[ServiceGraph] Processing stream {org_id}/{stream_name}");
         if let Err(e) = process_stream(&org_id, &stream_name, start_time, now).await {
-            log::error!(
-                "[ServiceGraph] Failed to process stream {}/{}: {}",
-                org_id,
-                stream_name,
-                e
-            );
+            log::error!("[ServiceGraph] Failed to process stream {org_id}/{stream_name}: {e}");
             continue; // Don't fail entire job if one stream fails
         }
     }
@@ -89,7 +79,7 @@ async fn get_trace_streams() -> Result<Vec<(String, String)>, anyhow::Error> {
         .await;
 
         if org_streams.is_empty() {
-            log::warn!(
+            log::debug!(
                 "[ServiceGraph] No trace streams found for org '{}' (identifier: {})",
                 org.name,
                 org.identifier
@@ -112,6 +102,17 @@ async fn process_stream(
     start_time: i64,
     end_time: i64,
 ) -> Result<(), anyhow::Error> {
+    // Skip if no new data was ingested since the last processing interval.
+    // Using doc_time_max against start_time avoids re-processing the same data
+    // on every tick when ingestion has stopped.
+    let stats = infra::cache::stats::get_stream_stats(org_id, stream_name, StreamType::Traces);
+    if stats.doc_time_max > 0 && stats.doc_time_max <= start_time {
+        log::info!(
+            "[ServiceGraph] Stream {org_id}/{stream_name} has no new data since last interval, skipping"
+        );
+        return Ok(());
+    }
+
     // Build SQL to aggregate service graph edges directly in DataFusion
     // Use CTE to compute client/server first, then aggregate
     log::info!(
@@ -123,9 +124,6 @@ async fn process_stream(
         (end_time - start_time) / 1_000_000
     );
 
-    // Build service graph using parent-child span relationships via reference_parent_span_id
-    // JOIN approach: match parent CLIENT/INTERNAL spans with child SERVER/INTERNAL spans
-
     let exclude_internal = get_o2_config().service_graph.exclude_internal_spans;
 
     let (server_span_kinds, client_span_kinds) = if exclude_internal {