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

Runtime Stats in API

Browse Source 
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
This commit is contained in:
Alexey
2026-03-04 02:46:47 +03:00
parent 5876623bb0
commit 716b4adef2
2 changed files with 216 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
src/api/mod.rs
View File

@@ -3,7 +3,6 @@ use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use http_body_util::{BodyExt, Full};
use hyper::body::{Bytes, Incoming};
@@ -24,13 +23,17 @@ use crate::transport::middle_proxy::MePool;
mod config_store;
mod model;
mod runtime_stats;
mod users;
use config_store::{current_revision, parse_if_match};
use model::{
ApiFailure, CreateUserRequest, DcStatus, DcStatusData, ErrorBody, ErrorResponse, HealthData,
MeWriterStatus, MeWritersData, MeWritersSummary, PatchUserRequest, RotateSecretRequest,
SuccessResponse, SummaryData,
ApiFailure, CreateUserRequest, ErrorBody, ErrorResponse, HealthData, PatchUserRequest,
RotateSecretRequest, SuccessResponse, SummaryData,
};
use runtime_stats::{
MinimalCacheEntry, build_dcs_data, build_me_writers_data, build_minimal_all_data,
build_zero_all_data,
};
use users::{create_user, delete_user, patch_user, rotate_secret, users_from_config};
@@ -41,6 +44,7 @@ pub(super) struct ApiShared {
pub(super) me_pool: Option<Arc<MePool>>,
pub(super) config_path: PathBuf,
pub(super) mutation_lock: Arc<Mutex<()>>,
pub(super) minimal_cache: Arc<Mutex<Option<MinimalCacheEntry>>>,
pub(super) request_id: Arc<AtomicU64>,
}
@@ -78,6 +82,7 @@ pub async fn serve(
me_pool,
config_path,
mutation_lock: Arc::new(Mutex::new(())),
minimal_cache: Arc::new(Mutex::new(None)),
request_id: Arc::new(AtomicU64::new(1)),
});
@@ -185,91 +190,24 @@ async fn handle(
};
Ok(success_response(StatusCode::OK, data, revision))
}
("GET", "/v1/stats/zero/all") => {
let revision = current_revision(&shared.config_path).await?;
let data = build_zero_all_data(&shared.stats, cfg.access.users.len());
Ok(success_response(StatusCode::OK, data, revision))
}
("GET", "/v1/stats/minimal/all") => {
let revision = current_revision(&shared.config_path).await?;
let data = build_minimal_all_data(shared.as_ref(), api_cfg).await;
Ok(success_response(StatusCode::OK, data, revision))
}
("GET", "/v1/stats/me-writers") => {
let revision = current_revision(&shared.config_path).await?;
let data = match &shared.me_pool {
Some(pool) => {
let snapshot = pool.api_status_snapshot().await;
let writers = snapshot
.writers
.into_iter()
.map(|entry| MeWriterStatus {
writer_id: entry.writer_id,
dc: entry.dc,
endpoint: entry.endpoint.to_string(),
generation: entry.generation,
state: entry.state,
draining: entry.draining,
degraded: entry.degraded,
bound_clients: entry.bound_clients,
idle_for_secs: entry.idle_for_secs,
rtt_ema_ms: entry.rtt_ema_ms,
})
.collect();
MeWritersData {
middle_proxy_enabled: true,
generated_at_epoch_secs: snapshot.generated_at_epoch_secs,
summary: MeWritersSummary {
configured_dc_groups: snapshot.configured_dc_groups,
configured_endpoints: snapshot.configured_endpoints,
available_endpoints: snapshot.available_endpoints,
available_pct: snapshot.available_pct,
required_writers: snapshot.required_writers,
alive_writers: snapshot.alive_writers,
coverage_pct: snapshot.coverage_pct,
},
writers,
}
}
None => MeWritersData {
middle_proxy_enabled: false,
generated_at_epoch_secs: now_epoch_secs(),
summary: MeWritersSummary {
configured_dc_groups: 0,
configured_endpoints: 0,
available_endpoints: 0,
available_pct: 0.0,
required_writers: 0,
alive_writers: 0,
coverage_pct: 0.0,
},
writers: Vec::new(),
},
};
let data = build_me_writers_data(shared.as_ref(), api_cfg).await;
Ok(success_response(StatusCode::OK, data, revision))
}
("GET", "/v1/stats/dcs") => {
let revision = current_revision(&shared.config_path).await?;
let data = match &shared.me_pool {
Some(pool) => {
let snapshot = pool.api_status_snapshot().await;
let dcs = snapshot
.dcs
.into_iter()
.map(|entry| DcStatus {
dc: entry.dc,
endpoints: entry.endpoints.into_iter().map(|value| value.to_string()).collect(),
available_endpoints: entry.available_endpoints,
available_pct: entry.available_pct,
required_writers: entry.required_writers,
alive_writers: entry.alive_writers,
coverage_pct: entry.coverage_pct,
rtt_ms: entry.rtt_ms,
load: entry.load,
})
.collect();
DcStatusData {
middle_proxy_enabled: true,
generated_at_epoch_secs: snapshot.generated_at_epoch_secs,
dcs,
}
}
None => DcStatusData {
middle_proxy_enabled: false,
generated_at_epoch_secs: now_epoch_secs(),
dcs: Vec::new(),
},
};
let data = build_dcs_data(shared.as_ref(), api_cfg).await;
Ok(success_response(StatusCode::OK, data, revision))
}
("GET", "/v1/stats/users") | ("GET", "/v1/users") => {
@@ -397,13 +335,6 @@ async fn handle(
}
}
fn now_epoch_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
fn success_response<T: Serialize>(
status: StatusCode,
data: T,