ferra_agent/
lib.rs

1//! Ferra sidecar agent.
2//!
3//! Runs alongside a service container, holds an in-memory cache of every key
4//! under one or more configured prefixes, keeps that cache fresh via SSE
5//! watches against `ferra-server`, and exposes a tiny localhost HTTP API the
6//! service uses to read config:
7//!
8//! - `GET /cfg/{key}`                     — return current value or 404
9//! - `GET /cfg/{key}?wait=30s&since=N`    — long-poll: return when key changes
10//! - `GET /cfg?prefix=...`                — list everything under a prefix
11//! - `GET /healthz`                       — process liveness
12//! - `GET /readyz`                        — initial snapshots loaded?
13
14pub mod api;
15pub mod cache;
16pub mod config;
17pub mod sse;
18pub mod upstream;
19pub mod watch;
20
21pub use config::Args;
22
23use std::sync::Arc;
24use tokio::net::TcpListener;
25use tracing::info;
26
27pub async fn run(args: Args) -> anyhow::Result<()> {
28    if args.prefix.is_empty() {
29        anyhow::bail!("at least one --prefix is required");
30    }
31
32    let upstream = upstream::UpstreamClient::new(args.server.clone());
33    let prefixes: Vec<Arc<cache::PrefixState>> = args
34        .prefix
35        .iter()
36        .map(|p| Arc::new(cache::PrefixState::new(p.clone())))
37        .collect();
38
39    // Spawn one watch task per prefix.
40    for prefix in &prefixes {
41        let prefix = prefix.clone();
42        let upstream = upstream.clone();
43        let min_backoff = args.min_backoff;
44        let max_backoff = args.max_backoff;
45        tokio::spawn(async move {
46            watch::run_loop(prefix, upstream, min_backoff, max_backoff).await;
47        });
48    }
49
50    let state = Arc::new(api::AgentState {
51        upstream,
52        prefixes: prefixes.clone(),
53    });
54
55    // Start the HTTP server right away so /healthz / /readyz are available
56    // even before snapshots have loaded. Service containers can poll /readyz
57    // to know when to start serving traffic.
58    let app = api::router(state);
59    let listener = TcpListener::bind(&args.listen).await?;
60    let actual = listener.local_addr()?;
61    info!(addr = %actual, prefixes = ?args.prefix, "ferra-agent listening");
62    axum::serve(listener, app)
63        .with_graceful_shutdown(shutdown_signal())
64        .await?;
65    Ok(())
66}
67
68async fn shutdown_signal() {
69    #[cfg(unix)]
70    {
71        use tokio::signal::unix::{signal, SignalKind};
72        let mut term = signal(SignalKind::terminate()).expect("install SIGTERM handler");
73        tokio::select! {
74            _ = tokio::signal::ctrl_c() => {},
75            _ = term.recv() => {},
76        }
77    }
78    #[cfg(not(unix))]
79    {
80        let _ = tokio::signal::ctrl_c().await;
81    }
82    tracing::info!("shutdown signal received");
83}
ferra_agent/lib.rs

ferra_agent/
lib.rs
