Failed to generate code suggestions for PR

Greptile Overview

Greptile Summary

Replaces in-memory service graph processing with SQL-based daemon architecture that runs in compactor every hour, completely decoupling service graph from trace ingestion.

Major Changes:

• Added processor.rs (202 lines) - SQL daemon that queries trace streams using DataFusion with CTE for client/server edge extraction and percentile calculations
• Added aggregator.rs (155 lines) - Writes SQL-aggregated edges to _o2_service_graph internal stream via bulk ingestion
• Modified api.rs - Queries _o2_service_graph stream for last 60 minutes, returns availableStreams for filtering
• Modified compactor.rs - Integrated service_graph_processor job with configurable interval
• Modified traces/mod.rs - Removed ALL inline processing during trace ingestion (lines 398-399)
• Frontend updated to support stream filtering via query params

Architecture Benefits:

• Zero impact on trace ingestion performance
• SQL aggregation leverages DataFusion for efficient processing
• Configurable processing interval (default: 1 hour) and query window (default: 60 min)
• Graceful error handling per stream (failures don't block other streams)

Configuration:

• O2_SERVICE_GRAPH_ENABLED=true
• O2_SERVICE_GRAPH_PROCESSING_INTERVAL_SECS=3600
• O2_SERVICE_GRAPH_QUERY_TIME_RANGE_MINUTES=60

Confidence Score: 4/5

• Safe to merge with attention to SQL query assumptions and companion PR requirement
• Clean architectural refactor with proper error handling and feature gating. Score reflects: (1) SQL query assumes peer_service field exists in trace schema, (2) requires companion o2-enterprise PR for business logic, (3) no validation on stream_filter query parameter for SQL injection prevention
• Pay close attention to src/service/traces/service_graph/processor.rs - SQL query assumes peer_service field exists

Important Files Changed

File Analysis

Sequence Diagram

sequenceDiagram
    participant Client as Frontend Client
    participant API as Service Graph API
    participant Compactor as Compactor Job
    participant Processor as SQL Processor
    participant DataFusion as DataFusion Query Engine
    participant TraceStream as Trace Streams
    participant ServiceGraphStream as _o2_service_graph Stream
    
    Note over Compactor,Processor: Daemon runs every processing_interval_secs (default: 3600s)
    
    Compactor->>Processor: process_service_graph()
    Processor->>Processor: get_trace_streams() - List all orgs & trace streams
    
    loop For each trace stream
        Processor->>DataFusion: Execute SQL with CTE
        Note right of DataFusion: WITH edges AS (<br/>  SELECT client, server, duration, span_status<br/>  WHERE span_kind IN ('2','3')<br/>  AND peer_service IS NOT NULL<br/>)<br/>SELECT COUNT(*), percentiles<br/>GROUP BY client, server
        DataFusion->>TraceStream: Query last 60min of traces
        TraceStream-->>DataFusion: Span records
        DataFusion-->>Processor: Aggregated edges (p50/p95/p99)
        Processor->>ServiceGraphStream: write_sql_aggregated_edges()
        Note right of ServiceGraphStream: Bulk ingest via logs API<br/>to _o2_service_graph
    end
    
    Note over Client,API: User queries topology (independent of daemon)
    
    Client->>API: GET /topology/current?stream_name=X
    API->>ServiceGraphStream: SQL query last 60min
    ServiceGraphStream-->>API: Edge records
    API->>API: build_topology() - Convert to nodes/edges
    API-->>Client: {nodes, edges, availableStreams}