featherbit/
main.rs

1//! # featherbit
2//!
3//! A high-performance API gateway delivered as a single Rust binary.
4//!
5//! Featherbit routes traffic through **node-graph policies** declared in YAML:
6//! each policy is a pipeline of nodes wired together by success/error ports.
7//! Plugins come in two tiers — 13 native Rust plugins (proxying, auth,
8//! rate-limiting, CORS, logging, ...) plus scripted plugins written in Lua.
9//! A [`context::Context`] object (`request`, `response`, `message`, `errors`)
10//! flows through every node in the pipeline. Operations are handled by an
11//! admin REST API with an embedded React UI, configuration hot-reload via a
12//! file watcher, and Prometheus metrics per route and per node.
13//!
14//! # Architecture
15//!
16//! Request flow: HTTP request → `server::listener` matches a route → builds a
17//! `Context` → `CompiledGraph::execute()` walks the policy's nodes following
18//! success/error ports → the final `Context.response` is sent to the client.
19//!
20//! Configuration lives in two files: `system.yaml` (listeners, timeouts,
21//! admin API, logging) and `gateway.yaml` (routes and policies). Both support
22//! `${ENV_VAR:-default}` interpolation and the latter is hot-reloaded on change.
23
24// `PluginExecutionError` deliberately carries the whole `Context` by value so the
25// graph engine can route a failing node's context out through its `error` port
26// (see `plugins::PluginExecutionError`). That makes the `Err` variant large by
27// design; boxing it would ripple through the `Plugin` trait and every plugin.
28#![allow(clippy::result_large_err)]
29
30mod admin;
31mod balancer;
32mod batch;
33mod config;
34mod config_store;
35mod consumers;
36mod context;
37mod debug;
38mod graph;
39mod hot_reload;
40mod metrics;
41mod outbound;
42mod plugins;
43mod ratelimit;
44mod routing;
45mod server;
46mod state;
47mod stream;
48mod traffic;
49mod vars;
50
51use std::path::PathBuf;
52use std::sync::Arc;
53
54use clap::Parser;
55use tracing::{error, info};
56
57use crate::config::{ConfigSourceKind, GatewayConfig, SystemConfig};
58use crate::config_store::{ConfigStore, FileConfigStore};
59use crate::state::SharedState;
60
61/// Command-line arguments: paths to the two YAML configuration files.
62#[derive(Parser)]
63#[command(name = "featherbit", about = "A lightweight API gateway")]
64struct Cli {
65    /// Path to system.yaml
66    #[arg(long, default_value = "config/system.yaml")]
67    system_config: PathBuf,
68
69    /// Path to gateway.yaml
70    #[arg(long, default_value = "config/gateway.yaml")]
71    gateway_config: PathBuf,
72}
73
74#[tokio::main]
75async fn main() {
76    let cli = Cli::parse();
77
78    // Load system config
79    let system: SystemConfig = match config::load_yaml_with_env(&cli.system_config) {
80        Ok(c) => c,
81        Err(e) => {
82            eprintln!("Failed to load system config: {}", e);
83            std::process::exit(1);
84        }
85    };
86
87    // Initialize logging
88    init_logging(&system.logging);
89
90    info!("Starting featherbit v{}", env!("CARGO_PKG_VERSION"));
91
92    // Select the config backend and load the initial gateway config.
93    let (config_store, gateway, config_path): (
94        Arc<dyn ConfigStore>,
95        GatewayConfig,
96        Option<PathBuf>,
97    ) = match system.config.source {
98        ConfigSourceKind::File => {
99            let store = Arc::new(FileConfigStore::new(cli.gateway_config.clone()));
100            let gw = match store.load_all().await {
101                Ok(c) => c,
102                Err(e) => {
103                    error!("Failed to load gateway config: {}", e);
104                    std::process::exit(1);
105                }
106            };
107            (store, gw, Some(cli.gateway_config.clone()))
108        }
109        ConfigSourceKind::Etcd => match build_etcd_source(&system, &cli.gateway_config).await {
110            Ok(v) => v,
111            Err(e) => {
112                error!("etcd config source: {}", e);
113                std::process::exit(1);
114            }
115        },
116    };
117
118    // Build shared state
119    let state = match SharedState::new(system.clone(), gateway, config_path, config_store) {
120        Ok(s) => Arc::new(s),
121        Err(e) => {
122            error!("Failed to initialize gateway: {}", e);
123            std::process::exit(1);
124        }
125    };
126
127    {
128        let routes = state.routes.read().await;
129        for (route, _) in routes.iter() {
130            info!(
131                "Route '{}' -> policy '{}' (match: {:?})",
132                route.name, route.policy, route.match_rule.path
133            );
134        }
135    }
136
137    // Shutdown coordination: a signal task flips this to `true` on SIGTERM /
138    // Ctrl+C; every accept loop watches it, stops accepting, and drains.
139    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
140    tokio::spawn(async move {
141        shutdown_signal().await;
142        info!("Shutdown signal received; draining…");
143        let _ = shutdown_tx.send(true);
144    });
145
146    // Start the config-change watcher appropriate to the source.
147    match system.config.source {
148        ConfigSourceKind::File => {
149            let reload_state = state.clone();
150            let watch_path = cli.gateway_config.clone();
151            tokio::spawn(async move {
152                hot_reload::watch_config(reload_state, watch_path).await;
153            });
154        }
155        ConfigSourceKind::Etcd => {
156            spawn_etcd_watch(state.clone(), &system);
157        }
158    }
159
160    let drain_timeout = std::time::Duration::from_secs(system.timeouts.shutdown_timeout_seconds);
161
162    // Start admin API (if configured), keeping its handle so we can await its
163    // drain before exiting.
164    let admin_handle = system.admin.as_ref().map(|admin_config| {
165        let admin_state = state.clone();
166        let admin_cfg = admin_config.clone();
167        let admin_shutdown = shutdown_rx.clone();
168        tokio::spawn(async move {
169            if let Err(e) =
170                admin::start_admin_server(&admin_cfg, admin_state, admin_shutdown, drain_timeout)
171                    .await
172            {
173                error!("Admin API error: {}", e);
174            }
175        })
176    });
177
178    // Start L4 (TCP/UDP) stream listeners, if any. Binds fail-fast before the
179    // data plane; each listener then runs in its own detached task.
180    if !system.stream.is_empty() {
181        if let Err(e) =
182            stream::start_all(&system.stream, &system.timeouts, shutdown_rx.clone()).await
183        {
184            error!("Failed to start stream listeners: {}", e);
185            std::process::exit(1);
186        }
187    }
188
189    // Start the data-plane server. It blocks until a shutdown signal, then
190    // drains in-flight connections and returns.
191    if let Err(e) = server::start_server(&system, state, shutdown_rx).await {
192        error!("Server error: {}", e);
193        std::process::exit(1);
194    }
195
196    // Let the admin API finish draining too before the process exits.
197    if let Some(handle) = admin_handle {
198        let _ = handle.await;
199    }
200    info!("Shutdown complete");
201}
202
203/// Completes when the process receives a termination signal: Ctrl+C on any
204/// platform, or `SIGTERM` on Unix (the signal container orchestrators send).
205async fn shutdown_signal() {
206    let ctrl_c = async {
207        let _ = tokio::signal::ctrl_c().await;
208    };
209
210    #[cfg(unix)]
211    let terminate = async {
212        match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
213            Ok(mut sig) => {
214                sig.recv().await;
215            }
216            Err(_) => std::future::pending::<()>().await,
217        }
218    };
219    #[cfg(not(unix))]
220    let terminate = std::future::pending::<()>();
221
222    tokio::select! {
223        _ = ctrl_c => {},
224        _ = terminate => {},
225    }
226}
227
228/// Builds the etcd config store and the initial gateway config (seeding etcd
229/// from the local file when the prefix is empty).
230async fn build_etcd_source(
231    system: &SystemConfig,
232    seed_path: &std::path::Path,
233) -> Result<(Arc<dyn ConfigStore>, GatewayConfig, Option<PathBuf>), String> {
234    config_store::etcd::build_source(system, seed_path).await
235}
236
237/// Spawns the etcd watch task (cluster-wide config convergence).
238fn spawn_etcd_watch(state: Arc<SharedState>, system: &SystemConfig) {
239    config_store::etcd::spawn_watch(state, system);
240}
241
242/// Initializes the global `tracing` subscriber in JSON or plain-text format.
243///
244/// The `RUST_LOG` environment variable, when set, takes precedence over the
245/// level configured in `system.yaml`.
246fn init_logging(config: &config::LoggingConfig) {
247    let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
248        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(&config.level));
249
250    match config.format.as_str() {
251        "json" => {
252            tracing_subscriber::fmt()
253                .json()
254                .with_env_filter(env_filter)
255                .init();
256        }
257        _ => {
258            tracing_subscriber::fmt().with_env_filter(env_filter).init();
259        }
260    }
261}
featherbit/main.rs

featherbit/
main.rs
