astelm/telemt
1
0
Fork 0
mirror of https://github.com/telemt/telemt.git synced 2026-09-18 08:18:31 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/main' into feat/config-api-server-listeners

Browse Source
This commit is contained in:
Dmitry
2026-07-20 11:50:30 +00:00
parent 9248076572 d851200e47
commit 723da8c34c
40 changed files with 4593 additions and 1116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/config_edit.rs
+50 -2
View File
@@ -8,19 +8,27 @@ use toml::Value as Toml;
use super::ApiShared;
use super::config_store::{
EDITABLE_SECTIONS, EDITABLE_SERVER_FIELDS, compute_revision, current_revision,
is_editable_section, save_sections_to_disk,
is_editable_section, load_config_from_disk, save_sections_to_disk,
};
use super::model::ApiFailure;
use crate::config::ProxyConfig;
use crate::config::hot_reload::classify_config_changes;
use crate::maestro::reload::{ReloadAccepted, ReloadRequest, ReloadSubmitError};
use crate::maestro::runtime_build::deferred_process_fields;
use serde::Serialize;
use std::path::Path;
use std::sync::Arc;
#[derive(Debug, Serialize)]
pub(super) struct PatchConfigResponse {
pub revision: String,
pub restart_required: bool,
pub runtime_reload_required: bool,
pub process_restart_required: bool,
pub deferred_process_fields: Vec<String>,
pub changed: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub reload: Option<ReloadAccepted>,
}
/// Shared-state wrapper around [`apply_patch_to_path`]: serializes config
@@ -29,10 +37,40 @@ pub(super) struct PatchConfigResponse {
pub(super) async fn patch_config(
patch_json: Json,
expected_revision: Option<String>,
reload_request: Option<ReloadRequest>,
shared: &ApiShared,
) -> Result<PatchConfigResponse, ApiFailure> {
let _guard = shared.mutation_lock.lock().await;
let resp = apply_patch_to_path(&shared.config_path, &patch_json, expected_revision).await?;
if reload_request.is_some()
&& let Some(reload_id) = shared.reload_control.in_progress().await
{
return Err(ApiFailure::new(
hyper::StatusCode::CONFLICT,
"reload_in_progress",
format!("Reload {} is already in progress", reload_id),
));
}
let mut resp = apply_patch_to_path(&shared.config_path, &patch_json, expected_revision).await?;
if let Some(request) = reload_request {
let config = Arc::new(load_config_from_disk(&shared.config_path).await?);
let accepted = shared
.reload_control
.submit(config, resp.revision.clone(), request)
.await
.map_err(|error| match error {
ReloadSubmitError::InProgress(reload_id) => ApiFailure::new(
hyper::StatusCode::CONFLICT,
"reload_in_progress",
format!("Reload {} is already in progress", reload_id),
),
ReloadSubmitError::MaestroUnavailable => ApiFailure::new(
hyper::StatusCode::SERVICE_UNAVAILABLE,
"maestro_unavailable",
"Maestro reload coordinator is unavailable",
),
})?;
resp.reload = Some(accepted);
}
drop(_guard);
shared
.runtime_events
@@ -119,6 +157,7 @@ pub(super) async fn apply_patch_to_path(
// 4. classify changes (Telemt's own hot/restart rule)
let class = classify_config_changes(&old_cfg, &new_cfg);
let deferred_process_fields = deferred_process_fields(&old_cfg, &new_cfg);
// 5. write only the touched top-level sections
let revision = save_sections_to_disk(config_path, &new_cfg, &touched).await?;
@@ -126,7 +165,11 @@ pub(super) async fn apply_patch_to_path(
Ok(PatchConfigResponse {
revision,
restart_required: class.restart_required,
runtime_reload_required: class.restart_required,
process_restart_required: !deferred_process_fields.is_empty(),
deferred_process_fields,
changed: class.changed,
reload: None,
})
}
@@ -335,6 +378,9 @@ mod tests {
let patch: Json = serde_json::json!({"censorship": {"tls_domain": "b.com"}});
let resp = apply_patch_to_path(&path, &patch, None).await.unwrap();
assert!(resp.restart_required);
assert!(resp.runtime_reload_required);
assert!(!resp.process_restart_required);
assert!(resp.deferred_process_fields.is_empty());
assert!(resp.changed.iter().any(|c| c == "censorship"));
let written = std::fs::read_to_string(&path).unwrap();
assert!(written.contains("tls_domain = \"b.com\""));
@@ -536,6 +582,8 @@ mod tests {
let patch: Json = serde_json::json!({"general": {"log_level": "debug"}});
let resp = apply_patch_to_path(&path, &patch, None).await.unwrap();
assert!(!resp.restart_required);
assert!(!resp.runtime_reload_required);
assert!(!resp.process_restart_required);
assert!(resp.changed.iter().any(|c| c == "general"));
}
}
src/api/config_store.rs
+15
View File
@@ -72,6 +72,13 @@ pub(super) async fn current_revision(config_path: &Path) -> Result<String, ApiFa
Ok(compute_revision(&content))
}
pub(crate) async fn current_revision_for_maestro(config_path: &Path) -> Result<String, String> {
let content = tokio::fs::read_to_string(config_path)
.await
.map_err(|error| format!("failed to read config: {}", error))?;
Ok(compute_revision(&content))
}
pub(super) fn compute_revision(content: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(content.as_bytes());
@@ -86,6 +93,14 @@ pub(super) async fn load_config_from_disk(config_path: &Path) -> Result<ProxyCon
.map_err(|e| ApiFailure::internal(format!("failed to load config: {}", e)))
}
pub(super) async fn load_config_for_reload(config_path: &Path) -> Result<ProxyConfig, ApiFailure> {
let config_path = config_path.to_path_buf();
tokio::task::spawn_blocking(move || ProxyConfig::load(config_path))
.await
.map_err(|error| ApiFailure::internal(format!("failed to join config loader: {}", error)))?
.map_err(|error| ApiFailure::bad_request(format!("invalid runtime config: {}", error)))
}
#[allow(dead_code)]
pub(super) async fn save_config_to_disk(
config_path: &Path,
src/api/mod.rs
+170 -13
View File
@@ -7,6 +7,7 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use arc_swap::ArcSwap;
use http_body_util::Full;
use hyper::body::{Bytes, Incoming};
use hyper::header::AUTHORIZATION;
@@ -19,8 +20,10 @@ use tokio::sync::{Mutex, RwLock, Semaphore, watch};
use tokio::time::timeout;
use tracing::{debug, info, warn};
use crate::config::{ApiGrayAction, ProxyConfig};
use crate::config::ApiGrayAction;
use crate::ip_tracker::UserIpTracker;
use crate::maestro::generation::{RuntimeGeneration, RuntimeWatchState};
use crate::maestro::reload::{ReloadAccepted, ReloadControl, ReloadRequest, ReloadSubmitError};
use crate::proxy::route_mode::RouteRuntimeController;
use crate::proxy::shared_state::ProxySharedState;
use crate::startup::StartupTracker;
@@ -29,11 +32,13 @@ use crate::transport::UpstreamManager;
use crate::transport::middle_proxy::MePool;
mod config_edit;
mod config_store;
pub(crate) mod config_store;
mod events;
mod http_utils;
mod model;
mod patch;
#[cfg(test)]
mod reload_tests;
mod runtime_edge;
mod runtime_init;
mod runtime_min;
@@ -44,7 +49,8 @@ mod runtime_zero;
mod users;
use config_store::{
current_revision, ensure_expected_revision, load_config_from_disk, parse_if_match,
current_revision, ensure_expected_revision, load_config_for_reload, load_config_from_disk,
parse_if_match,
};
use events::ApiEventStore;
use http_utils::{error_response, read_json, read_optional_json, success_response};
@@ -107,12 +113,15 @@ pub(super) struct ApiShared {
pub(super) minimal_cache: Arc<Mutex<Option<MinimalCacheEntry>>>,
pub(super) runtime_edge_connections_cache: Arc<Mutex<Option<EdgeConnectionsCacheEntry>>>,
pub(super) runtime_edge_recompute_lock: Arc<Mutex<()>>,
pub(super) cache_generation: Arc<AtomicU64>,
pub(super) runtime_events: Arc<ApiEventStore>,
pub(super) request_id: Arc<AtomicU64>,
pub(super) runtime_state: Arc<ApiRuntimeState>,
pub(super) startup_tracker: Arc<StartupTracker>,
pub(super) route_runtime: Arc<RouteRuntimeController>,
pub(super) proxy_shared: Arc<ProxySharedState>,
pub(super) reload_control: ReloadControl,
pub(super) active_runtime: Arc<ArcSwap<RuntimeGeneration>>,
}
impl ApiShared {
@@ -123,6 +132,31 @@ impl ApiShared {
fn detected_link_ips(&self) -> (Option<IpAddr>, Option<IpAddr>) {
*self.detected_ips_rx.borrow()
}
fn for_runtime(&self, runtime: &RuntimeGeneration) -> Self {
Self {
stats: runtime.stats.clone(),
ip_tracker: runtime.ip_tracker.clone(),
me_pool: runtime.me_pool_runtime.clone(),
upstream_manager: runtime.upstream_manager.clone(),
config_path: self.config_path.clone(),
quota_state_path: self.quota_state_path.clone(),
detected_ips_rx: self.detected_ips_rx.clone(),
mutation_lock: self.mutation_lock.clone(),
minimal_cache: self.minimal_cache.clone(),
runtime_edge_connections_cache: self.runtime_edge_connections_cache.clone(),
runtime_edge_recompute_lock: self.runtime_edge_recompute_lock.clone(),
cache_generation: self.cache_generation.clone(),
runtime_events: self.runtime_events.clone(),
request_id: self.request_id.clone(),
runtime_state: self.runtime_state.clone(),
startup_tracker: self.startup_tracker.clone(),
route_runtime: runtime.route_runtime.clone(),
proxy_shared: runtime.proxy_shared.clone(),
reload_control: self.reload_control.clone(),
active_runtime: self.active_runtime.clone(),
}
}
}
fn auth_header_matches(actual: &str, expected: &str) -> bool {
@@ -144,6 +178,41 @@ fn user_action_route_matches(path: &str, suffix: &str) -> bool {
.unwrap_or(false)
}
fn reload_status_route_id(path: &str) -> Option<u64> {
path.strip_prefix("/v1/system/reload/")
.filter(|id| !id.is_empty() && !id.contains('/'))
.and_then(|id| id.parse().ok())
}
async fn submit_reload_from_disk(
config_path: &std::path::Path,
mutation_lock: &Mutex<()>,
reload_control: &ReloadControl,
expected_revision: Option<&str>,
request: ReloadRequest,
) -> Result<(ReloadAccepted, String), ApiFailure> {
let _guard = mutation_lock.lock().await;
ensure_expected_revision(config_path, expected_revision).await?;
let revision = current_revision(config_path).await?;
let config = Arc::new(load_config_for_reload(config_path).await?);
let accepted = reload_control
.submit(config, revision.clone(), request)
.await
.map_err(|error| match error {
ReloadSubmitError::InProgress(reload_id) => ApiFailure::new(
StatusCode::CONFLICT,
"reload_in_progress",
format!("Reload {} is already in progress", reload_id),
),
ReloadSubmitError::MaestroUnavailable => ApiFailure::new(
StatusCode::SERVICE_UNAVAILABLE,
"maestro_unavailable",
"Maestro reload coordinator is unavailable",
),
})?;
Ok((accepted, revision))
}
fn allowed_methods_for_path(path: &str) -> Option<&'static str> {
match path {
"/v1/health"
@@ -175,12 +244,14 @@ fn allowed_methods_for_path(path: &str) -> Option<&'static str> {
| "/v1/stats/users/active-ips"
| "/v1/stats/users/quota"
| "/v1/stats/users" => Some(ALLOW_GET),
"/v1/system/reload" => Some(ALLOW_POST),
"/v1/users" => Some(ALLOW_GET_POST),
"/v1/config" => Some(ALLOW_GET_PATCH),
_ if user_action_route_matches(path, "/reset-quota") => Some(ALLOW_POST),
_ if user_action_route_matches(path, "/rotate-secret") => Some(ALLOW_POST),
_ if user_action_route_matches(path, "/enable") => Some(ALLOW_POST),
_ if user_action_route_matches(path, "/disable") => Some(ALLOW_POST),
_ if reload_status_route_id(path).is_some() => Some(ALLOW_GET),
_ if path
.strip_prefix("/v1/users/")
.map(|user| !user.is_empty() && !user.contains('/'))
@@ -200,14 +271,35 @@ pub async fn serve(
route_runtime: Arc<RouteRuntimeController>,
proxy_shared: Arc<ProxySharedState>,
upstream_manager: Arc<UpstreamManager>,
config_rx: watch::Receiver<Arc<ProxyConfig>>,
admission_rx: watch::Receiver<bool>,
config_path: PathBuf,
quota_state_path: PathBuf,
detected_ips_rx: watch::Receiver<(Option<IpAddr>, Option<IpAddr>)>,
process_started_at_epoch_secs: u64,
startup_tracker: Arc<StartupTracker>,
reload_control: ReloadControl,
mut active_runtime_rx: watch::Receiver<Option<Arc<ArcSwap<RuntimeGeneration>>>>,
mut runtime_watch_rx: watch::Receiver<Option<RuntimeWatchState>>,
) {
let active_runtime = loop {
if let Some(active_runtime) = active_runtime_rx.borrow().clone() {
break active_runtime;
}
if active_runtime_rx.changed().await.is_err() {
warn!("Runtime generation channel closed before API bootstrap");
return;
}
};
let initial_watch_state = loop {
if let Some(watch_state) = runtime_watch_rx.borrow().clone() {
break watch_state;
}
if runtime_watch_rx.changed().await.is_err() {
warn!("Runtime watch channel closed before API bootstrap");
return;
}
};
let config_rx = initial_watch_state.config_rx.clone();
let admission_rx = initial_watch_state.admission_rx.clone();
let listener = match TcpListener::bind(listen).await {
Ok(listener) => listener,
Err(error) => {
@@ -241,6 +333,7 @@ pub async fn serve(
minimal_cache: Arc::new(Mutex::new(None)),
runtime_edge_connections_cache: Arc::new(Mutex::new(None)),
runtime_edge_recompute_lock: Arc::new(Mutex::new(())),
cache_generation: Arc::new(AtomicU64::new(1)),
runtime_events: Arc::new(ApiEventStore::new(
config_rx.borrow().server.api.runtime_edge_events_capacity,
)),
@@ -249,11 +342,12 @@ pub async fn serve(
startup_tracker,
route_runtime,
proxy_shared,
reload_control,
active_runtime,
});
spawn_runtime_watchers(
config_rx.clone(),
admission_rx.clone(),
runtime_watch_rx,
runtime_state.clone(),
shared.runtime_events.clone(),
);
@@ -282,13 +376,11 @@ pub async fn serve(
};
let shared_conn = shared.clone();
let config_rx_conn = config_rx.clone();
tokio::spawn(async move {
let _connection_permit = connection_permit;
let svc = service_fn(move |req: Request<Incoming>| {
let shared_req = shared_conn.clone();
let config_rx_req = config_rx_conn.clone();
async move { handle(req, peer, shared_req, config_rx_req).await }
async move { handle(req, peer, shared_req).await }
});
match timeout(
API_HTTP_CONNECTION_TIMEOUT,
@@ -318,8 +410,19 @@ async fn handle(
req: Request<Incoming>,
peer: SocketAddr,
shared: Arc<ApiShared>,
config_rx: watch::Receiver<Arc<ProxyConfig>>,
) -> Result<Response<Full<Bytes>>, IoError> {
let runtime = shared.active_runtime.load_full();
let previous_cache_generation = shared.cache_generation.swap(runtime.id, Ordering::AcqRel);
if previous_cache_generation != runtime.id {
*shared.minimal_cache.lock().await = None;
*shared.runtime_edge_connections_cache.lock().await = None;
}
let shared = Arc::new(shared.for_runtime(runtime.as_ref()));
let config_rx = runtime.config_rx.clone();
shared
.runtime_state
.admission_open
.store(*runtime.admission_rx.borrow(), Ordering::Relaxed);
let request_id = shared.next_request_id();
let cfg = config_rx.borrow().clone();
let api_cfg = &cfg.server.api;
@@ -651,6 +754,33 @@ async fn handle(
config_edit::read_managed_config(&shared.config_path).await?;
Ok(success_response(StatusCode::OK, value, revision))
}
("POST", "/v1/system/reload") => {
if api_cfg.read_only {
return Ok(error_response(
request_id,
ApiFailure::new(
StatusCode::FORBIDDEN,
"read_only",
"API runs in read-only mode",
),
));
}
let expected_revision = parse_if_match(req.headers());
let request = read_optional_json::<ReloadRequest>(req.into_body(), body_limit)
.await?
.unwrap_or_default();
request.validate().map_err(ApiFailure::bad_request)?;
let (accepted, revision) = submit_reload_from_disk(
&shared.config_path,
shared.mutation_lock.as_ref(),
&shared.reload_control,
expected_revision.as_deref(),
request,
)
.await?;
Ok(success_response(StatusCode::ACCEPTED, accepted, revision))
}
("PATCH", "/v1/config") => {
if api_cfg.read_only {
return Ok(error_response(
@@ -663,11 +793,20 @@ async fn handle(
));
}
let expected_revision = parse_if_match(req.headers());
let reload_request =
ReloadRequest::from_query(query.as_deref()).map_err(ApiFailure::bad_request)?;
let body = read_json::<serde_json::Value>(req.into_body(), body_limit).await?;
match config_edit::patch_config(body, expected_revision, &shared).await {
match config_edit::patch_config(body, expected_revision, reload_request, &shared)
.await
{
Ok(resp) => {
let revision = resp.revision.clone();
Ok(success_response(StatusCode::OK, resp, revision))
let status = if resp.reload.is_some() {
StatusCode::ACCEPTED
} else {
StatusCode::OK
};
Ok(success_response(status, resp, revision))
}
Err(error) => {
shared
@@ -678,6 +817,24 @@ async fn handle(
}
}
_ => {
if method == Method::GET
&& let Some(reload_id) = reload_status_route_id(normalized_path)
{
let revision = current_revision(&shared.config_path).await?;
let status =
shared
.reload_control
.status(reload_id)
.await
.ok_or_else(|| {
ApiFailure::new(
StatusCode::NOT_FOUND,
"reload_not_found",
format!("Reload {} was not found", reload_id),
)
})?;
return Ok(success_response(StatusCode::OK, status, revision));
}
if method == Method::POST
&& let Some(base_user) = normalized_path
.strip_prefix("/v1/users/")
src/api/reload_tests.rs
+119
View File
@@ -0,0 +1,119 @@
use super::*;
use crate::config::ProxyConfig;
async fn config_file() -> (tempfile::TempDir, PathBuf, String) {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("config.toml");
let mut config = ProxyConfig::default();
config.server.max_connections = 4_242;
let body = toml::to_string_pretty(&config).unwrap();
tokio::fs::write(&path, &body).await.unwrap();
let revision = config_store::compute_revision(&body);
(directory, path, revision)
}
#[tokio::test]
async fn reload_submission_uses_matching_disk_revision_and_snapshot() {
let (_directory, path, revision) = config_file().await;
let mutation_lock = Mutex::new(());
let (control, mut commands) = ReloadControl::channel(1);
let request = ReloadRequest::default();
let (accepted, response_revision) = submit_reload_from_disk(
&path,
&mutation_lock,
&control,
Some(&revision),
request.clone(),
)
.await
.unwrap();
let command = commands.recv().await.unwrap();
assert_eq!(response_revision, revision);
assert_eq!(accepted.config_revision, revision);
assert_eq!(command.config_revision, revision);
assert_eq!(command.request, request);
assert_eq!(command.config.server.max_connections, 4_242);
}
#[tokio::test]
async fn revision_conflict_rejects_without_enqueuing_reload() {
let (_directory, path, _revision) = config_file().await;
let mutation_lock = Mutex::new(());
let (control, _commands) = ReloadControl::channel(1);
let error = submit_reload_from_disk(
&path,
&mutation_lock,
&control,
Some("stale-revision"),
ReloadRequest::default(),
)
.await
.unwrap_err();
assert_eq!(error.status, StatusCode::CONFLICT);
assert_eq!(error.code, "revision_conflict");
assert_eq!(control.in_progress().await, None);
}
#[tokio::test]
async fn reload_conflict_and_closed_coordinator_map_to_http_contract() {
let (_directory, path, _revision) = config_file().await;
let mutation_lock = Mutex::new(());
let (control, mut commands) = ReloadControl::channel(1);
let _accepted = submit_reload_from_disk(
&path,
&mutation_lock,
&control,
None,
ReloadRequest::default(),
)
.await
.unwrap();
let _command = commands.recv().await.unwrap();
let conflict = submit_reload_from_disk(
&path,
&mutation_lock,
&control,
None,
ReloadRequest::default(),
)
.await
.unwrap_err();
assert_eq!(conflict.status, StatusCode::CONFLICT);
assert_eq!(conflict.code, "reload_in_progress");
control.fail(1, "test cleanup").await;
drop(commands);
let unavailable = submit_reload_from_disk(
&path,
&mutation_lock,
&control,
None,
ReloadRequest::default(),
)
.await
.unwrap_err();
assert_eq!(unavailable.status, StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(unavailable.code, "maestro_unavailable");
}
#[test]
fn reload_routes_expose_only_documented_methods_and_ids() {
assert_eq!(
allowed_methods_for_path("/v1/system/reload"),
Some(ALLOW_POST)
);
assert_eq!(
allowed_methods_for_path("/v1/system/reload/42"),
Some(ALLOW_GET)
);
assert_eq!(reload_status_route_id("/v1/system/reload/42"), Some(42));
assert_eq!(
reload_status_route_id("/v1/system/reload/not-a-number"),
None
);
}
src/api/runtime_watch.rs
+292 -39
View File
@@ -4,58 +4,184 @@ use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::watch;
use crate::config::ProxyConfig;
use crate::maestro::generation::RuntimeWatchState;
use super::ApiRuntimeState;
use super::events::ApiEventStore;
pub(super) fn spawn_runtime_watchers(
config_rx: watch::Receiver<Arc<ProxyConfig>>,
admission_rx: watch::Receiver<bool>,
runtime_watch_rx: watch::Receiver<Option<RuntimeWatchState>>,
runtime_state: Arc<ApiRuntimeState>,
runtime_events: Arc<ApiEventStore>,
) {
let mut config_rx_reload = config_rx;
let runtime_state_reload = runtime_state.clone();
let runtime_events_reload = runtime_events.clone();
tokio::spawn(async move {
loop {
if config_rx_reload.changed().await.is_err() {
break;
}
runtime_state_reload
.config_reload_count
.fetch_add(1, Ordering::Relaxed);
runtime_state_reload
.last_config_reload_epoch_secs
.store(now_epoch_secs(), Ordering::Relaxed);
runtime_events_reload.record("config.reload.applied", "config receiver updated");
}
});
let _config_watcher = spawn_config_watcher(
runtime_watch_rx.clone(),
runtime_state.clone(),
runtime_events.clone(),
);
let _admission_watcher =
spawn_admission_watcher(runtime_watch_rx, runtime_state, runtime_events);
}
let mut admission_rx_watch = admission_rx;
fn spawn_config_watcher(
mut runtime_watch_rx: watch::Receiver<Option<RuntimeWatchState>>,
runtime_state: Arc<ApiRuntimeState>,
runtime_events: Arc<ApiEventStore>,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
runtime_state
.admission_open
.store(*admission_rx_watch.borrow(), Ordering::Relaxed);
runtime_events.record(
"admission.state",
format!("accepting_new_connections={}", *admission_rx_watch.borrow()),
);
let Some(mut current) = runtime_watch_rx.borrow().clone() else {
return;
};
loop {
if admission_rx_watch.changed().await.is_err() {
break;
tokio::select! {
biased;
changed = runtime_watch_rx.changed() => {
if changed.is_err() {
break;
}
let Some(next) = runtime_watch_rx.borrow().clone() else {
continue;
};
if next.generation_id != current.generation_id {
current = next;
record_config_reload(
&runtime_state,
&runtime_events,
format!("runtime generation {} activated", current.generation_id),
);
}
}
changed = current.config_rx.changed() => {
if changed.is_err() {
let Some(next) = wait_for_new_generation(
&mut runtime_watch_rx,
current.generation_id,
).await else {
break;
};
current = next;
record_config_reload(
&runtime_state,
&runtime_events,
format!("runtime generation {} activated", current.generation_id),
);
continue;
}
if active_generation_id(&runtime_watch_rx) != Some(current.generation_id) {
continue;
}
record_config_reload(
&runtime_state,
&runtime_events,
format!("generation {} config receiver updated", current.generation_id),
);
}
}
let admission_open = *admission_rx_watch.borrow();
runtime_state
.admission_open
.store(admission_open, Ordering::Relaxed);
runtime_events.record(
"admission.state",
format!("accepting_new_connections={}", admission_open),
);
}
});
})
}
fn spawn_admission_watcher(
mut runtime_watch_rx: watch::Receiver<Option<RuntimeWatchState>>,
runtime_state: Arc<ApiRuntimeState>,
runtime_events: Arc<ApiEventStore>,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let Some(mut current) = runtime_watch_rx.borrow().clone() else {
return;
};
record_admission_state(&runtime_state, &runtime_events, &current);
loop {
tokio::select! {
biased;
changed = runtime_watch_rx.changed() => {
if changed.is_err() {
break;
}
let Some(next) = runtime_watch_rx.borrow().clone() else {
continue;
};
if next.generation_id != current.generation_id {
current = next;
record_admission_state(&runtime_state, &runtime_events, &current);
}
}
changed = current.admission_rx.changed() => {
if changed.is_err() {
let Some(next) = wait_for_new_generation(
&mut runtime_watch_rx,
current.generation_id,
).await else {
break;
};
current = next;
record_admission_state(&runtime_state, &runtime_events, &current);
continue;
}
if active_generation_id(&runtime_watch_rx) == Some(current.generation_id) {
record_admission_state(&runtime_state, &runtime_events, &current);
}
}
}
}
})
}
fn active_generation_id(
runtime_watch_rx: &watch::Receiver<Option<RuntimeWatchState>>,
) -> Option<u64> {
runtime_watch_rx
.borrow()
.as_ref()
.map(|state| state.generation_id)
}
async fn wait_for_new_generation(
runtime_watch_rx: &mut watch::Receiver<Option<RuntimeWatchState>>,
previous_generation_id: u64,
) -> Option<RuntimeWatchState> {
loop {
if let Some(state) = runtime_watch_rx.borrow().clone()
&& state.generation_id != previous_generation_id
{
return Some(state);
}
if runtime_watch_rx.changed().await.is_err() {
return None;
}
}
}
fn record_config_reload(
runtime_state: &ApiRuntimeState,
runtime_events: &ApiEventStore,
context: String,
) {
runtime_state
.config_reload_count
.fetch_add(1, Ordering::Relaxed);
runtime_state
.last_config_reload_epoch_secs
.store(now_epoch_secs(), Ordering::Relaxed);
runtime_events.record("config.reload.applied", context);
}
fn record_admission_state(
runtime_state: &ApiRuntimeState,
runtime_events: &ApiEventStore,
current: &RuntimeWatchState,
) {
let admission_open = *current.admission_rx.borrow();
runtime_state
.admission_open
.store(admission_open, Ordering::Relaxed);
runtime_events.record(
"admission.state",
format!(
"generation={} accepting_new_connections={}",
current.generation_id, admission_open
),
);
}
fn now_epoch_secs() -> u64 {
@@ -64,3 +190,130 @@ fn now_epoch_secs() -> u64 {
.unwrap_or_default()
.as_secs()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::ProxyConfig;
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::time::Duration;
fn state(
generation_id: u64,
) -> (
RuntimeWatchState,
watch::Sender<Arc<ProxyConfig>>,
watch::Sender<bool>,
) {
let (config_tx, config_rx) = watch::channel(Arc::new(ProxyConfig::default()));
let (admission_tx, admission_rx) = watch::channel(true);
(
RuntimeWatchState {
generation_id,
config_rx,
admission_rx,
},
config_tx,
admission_tx,
)
}
fn runtime_state() -> Arc<ApiRuntimeState> {
Arc::new(ApiRuntimeState {
process_started_at_epoch_secs: 1,
config_reload_count: AtomicU64::new(0),
last_config_reload_epoch_secs: AtomicU64::new(0),
admission_open: AtomicBool::new(false),
})
}
async fn wait_for_count(runtime_state: &ApiRuntimeState, expected: u64) {
tokio::time::timeout(Duration::from_secs(1), async {
loop {
if runtime_state.config_reload_count.load(Ordering::Relaxed) == expected {
break;
}
tokio::task::yield_now().await;
}
})
.await
.unwrap();
}
#[tokio::test]
async fn watchers_follow_only_the_active_generation() {
let (initial, initial_config_tx, initial_admission_tx) = state(1);
let (runtime_watch_tx, runtime_watch_rx) = watch::channel(Some(initial));
let runtime_state = runtime_state();
let events = Arc::new(ApiEventStore::new(16));
spawn_runtime_watchers(runtime_watch_rx, runtime_state.clone(), events.clone());
tokio::task::yield_now().await;
assert_eq!(runtime_state.config_reload_count.load(Ordering::Relaxed), 0);
initial_config_tx.send_replace(Arc::new(ProxyConfig::default()));
wait_for_count(&runtime_state, 1).await;
let (next, next_config_tx, next_admission_tx) = state(2);
runtime_watch_tx.send_replace(Some(next));
wait_for_count(&runtime_state, 2).await;
initial_config_tx.send_replace(Arc::new(ProxyConfig::default()));
initial_admission_tx.send_replace(false);
tokio::task::yield_now().await;
assert_eq!(runtime_state.config_reload_count.load(Ordering::Relaxed), 2);
assert!(runtime_state.admission_open.load(Ordering::Relaxed));
next_config_tx.send_replace(Arc::new(ProxyConfig::default()));
next_admission_tx.send_replace(false);
wait_for_count(&runtime_state, 3).await;
tokio::time::timeout(Duration::from_secs(1), async {
while runtime_state.admission_open.load(Ordering::Relaxed) {
tokio::task::yield_now().await;
}
})
.await
.unwrap();
let snapshot = events.snapshot(16);
assert_eq!(
snapshot
.events
.iter()
.filter(|event| event.event_type == "config.reload.applied")
.count(),
3
);
}
#[tokio::test]
async fn watcher_recovers_from_closed_generation_and_exits_with_process_channel() {
let (initial, initial_config_tx, _initial_admission_tx) = state(1);
let (runtime_watch_tx, runtime_watch_rx) = watch::channel(Some(initial));
let runtime_state = runtime_state();
let events = Arc::new(ApiEventStore::new(16));
let watcher = spawn_config_watcher(runtime_watch_rx, runtime_state.clone(), events.clone());
drop(initial_config_tx);
tokio::task::yield_now().await;
let (next, next_config_tx, _next_admission_tx) = state(2);
runtime_watch_tx.send_replace(Some(next));
wait_for_count(&runtime_state, 1).await;
next_config_tx.send_replace(Arc::new(ProxyConfig::default()));
wait_for_count(&runtime_state, 2).await;
drop(runtime_watch_tx);
tokio::time::timeout(Duration::from_secs(1), watcher)
.await
.unwrap()
.unwrap();
assert_eq!(
events
.snapshot(16)
.events
.iter()
.filter(|event| event.event_type == "config.reload.applied")
.count(),
2
);
}
}
src/api/runtime_zero.rs
+25 -1
View File
@@ -1,4 +1,5 @@
use std::sync::atomic::Ordering;
use std::time::{SystemTime, UNIX_EPOCH};
use serde::Serialize;
@@ -162,7 +163,7 @@ pub(super) fn build_system_info_data(
build_time_utc,
rustc_version,
process_started_at_epoch_secs: shared.runtime_state.process_started_at_epoch_secs,
uptime_seconds: shared.stats.uptime_secs(),
uptime_seconds: process_uptime_seconds(shared.runtime_state.process_started_at_epoch_secs),
config_path: shared.config_path.display().to_string(),
config_hash: revision.to_string(),
config_reload_count: shared
@@ -173,6 +174,18 @@ pub(super) fn build_system_info_data(
}
}
fn process_uptime_seconds(process_started_at_epoch_secs: u64) -> f64 {
let now_epoch_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
process_uptime_seconds_at(process_started_at_epoch_secs, now_epoch_secs)
}
fn process_uptime_seconds_at(process_started_at_epoch_secs: u64, now_epoch_secs: u64) -> f64 {
now_epoch_secs.saturating_sub(process_started_at_epoch_secs) as f64
}
pub(super) async fn build_runtime_gates_data(
shared: &ApiShared,
cfg: &ProxyConfig,
@@ -339,3 +352,14 @@ fn me_writer_pick_mode_label(mode: MeWriterPickMode) -> &'static str {
MeWriterPickMode::P2c => "p2c",
}
}
#[cfg(test)]
mod tests {
use super::process_uptime_seconds_at;
#[test]
fn process_uptime_is_monotonic_and_saturating() {
assert_eq!(process_uptime_seconds_at(100, 135), 35.0);
assert_eq!(process_uptime_seconds_at(135, 100), 0.0);
}
}