devrig 0.30.1

Local development orchestrator
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

pub mod query;
pub mod receiver_grpc;
pub mod receiver_http;
pub mod storage;
pub mod types;

use std::sync::Arc;
use std::time::Duration;

use tokio::sync::{broadcast, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, warn};

use crate::config::model::OtelConfig;
use types::TelemetryEvent;

use self::storage::TelemetryStore;

/// Coordinator for OTLP gRPC + HTTP receivers and telemetry storage.
pub struct OtelCollector {
    store: Arc<RwLock<TelemetryStore>>,
    events_tx: broadcast::Sender<TelemetryEvent>,
    grpc_port: u16,
    http_port: u16,
}

impl OtelCollector {
    pub fn new(otel_config: &OtelConfig) -> Self {
        let retention = humantime::parse_duration(&otel_config.retention)
            .unwrap_or_else(|_| Duration::from_secs(3600));

        let store = Arc::new(RwLock::new(TelemetryStore::new(
            otel_config.trace_buffer,
            otel_config.log_buffer,
            otel_config.metric_buffer,
            retention,
        )));

        let (events_tx, _) = broadcast::channel(1024);

        Self {
            store,
            events_tx,
            grpc_port: otel_config.grpc_port.as_fixed().expect("otel grpc_port must be resolved before creating collector"),
            http_port: otel_config.http_port.as_fixed().expect("otel http_port must be resolved before creating collector"),
        }
    }

    pub fn store(&self) -> Arc<RwLock<TelemetryStore>> {
        Arc::clone(&self.store)
    }

    pub fn events_tx(&self) -> broadcast::Sender<TelemetryEvent> {
        self.events_tx.clone()
    }

    /// Start the OTLP gRPC and HTTP receivers as background tasks.
    /// Also starts a background sweeper for expired telemetry.
    pub async fn start(&self, cancel: CancellationToken) -> anyhow::Result<()> {
        // Start gRPC OTLP receiver
        let grpc_store = Arc::clone(&self.store);
        let grpc_tx = self.events_tx.clone();
        let grpc_port = self.grpc_port;
        let grpc_cancel = cancel.clone();
        tokio::spawn(async move {
            if let Err(e) =
                receiver_grpc::start_grpc_server(grpc_port, grpc_store, grpc_tx, grpc_cancel).await
            {
                warn!(error = %e, "OTLP gRPC server failed");
            }
        });
        debug!(port = self.grpc_port, "OTLP gRPC receiver started");

        // Start HTTP OTLP receiver
        let http_store = Arc::clone(&self.store);
        let http_tx = self.events_tx.clone();
        let http_port = self.http_port;
        let http_cancel = cancel.clone();
        tokio::spawn(async move {
            if let Err(e) =
                receiver_http::start_http_otlp_server(http_port, http_store, http_tx, http_cancel)
                    .await
            {
                warn!(error = %e, "OTLP HTTP server failed");
            }
        });
        debug!(port = self.http_port, "OTLP HTTP receiver started");

        // Background sweeper for expired telemetry
        let sweep_store = Arc::clone(&self.store);
        let sweep_cancel = cancel.clone();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(30)) => {
                        let mut store = sweep_store.write().await;
                        store.sweep_expired();
                    }
                    _ = sweep_cancel.cancelled() => break,
                }
            }
        });

        Ok(())
    }
}