From 509f7a8e966df78f54260b7b79062b75a392cce2 Mon Sep 17 00:00:00 2001 From: CodeMaster4711 Date: Wed, 1 Jul 2026 21:43:02 +0200 Subject: [PATCH 1/3] feat: add centralized log viewer with per-service tracing capture and admin retention settings --- Cargo.lock | 3 + app/src/lib/api/logs.ts | 83 ++++++ .../lib/components/sidebar/app-sidebar.svelte | 11 +- app/src/routes/(app)/logs/+page.svelte | 243 ++++++++++++++++++ control-plane/api-gateway/src/auth/rbac.rs | 4 + control-plane/api-gateway/src/init.rs | 9 + control-plane/api-gateway/src/main.rs | 7 +- control-plane/api-gateway/src/prune.rs | 55 ++++ control-plane/api-gateway/src/routes/logs.rs | 127 +++++++++ control-plane/api-gateway/src/routes/mod.rs | 4 + .../api-gateway/src/routes/settings.rs | 121 +++++++++ control-plane/api-gateway/src/telemetry.rs | 10 +- .../failover-controller/src/logger.rs | 16 +- control-plane/failover-controller/src/main.rs | 3 +- control-plane/registry/src/logger.rs | 16 +- control-plane/registry/src/main.rs | 3 +- control-plane/scheduler/src/logger.rs | 16 +- control-plane/scheduler/src/main.rs | 3 +- control-plane/sdn-controller/src/logger.rs | 16 +- control-plane/sdn-controller/src/main.rs | 3 +- .../shared/entity/src/entities/logs.rs | 65 +++++ .../shared/entity/src/entities/mod.rs | 6 +- .../{config.rs => system_settings.rs} | 9 +- control-plane/shared/migration/src/lib.rs | 2 + .../m20260701_000000_add_logs_and_settings.rs | 178 +++++++++++++ control-plane/shared/shared/Cargo.toml | 3 + .../shared/shared/src/db_log_layer.rs | 193 ++++++++++++++ control-plane/shared/shared/src/lib.rs | 2 + control-plane/volume-manager/src/logger.rs | 16 +- control-plane/volume-manager/src/main.rs | 3 +- 30 files changed, 1194 insertions(+), 36 deletions(-) create mode 100644 app/src/lib/api/logs.ts create mode 100644 app/src/routes/(app)/logs/+page.svelte create mode 100644 control-plane/api-gateway/src/prune.rs create mode 100644 control-plane/api-gateway/src/routes/logs.rs create mode 100644 control-plane/api-gateway/src/routes/settings.rs create mode 100644 control-plane/shared/entity/src/entities/logs.rs rename control-plane/shared/entity/src/entities/{config.rs => system_settings.rs} (63%) create mode 100644 control-plane/shared/migration/src/m20260701_000000_add_logs_and_settings.rs create mode 100644 control-plane/shared/shared/src/db_log_layer.rs diff --git a/Cargo.lock b/Cargo.lock index dc8abf4b..be7c5be4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4817,13 +4817,16 @@ version = "0.2.2" dependencies = [ "anyhow", "async-trait", + "chrono", "dotenvy", + "entity", "sea-orm", "serde", "thiserror 2.0.18", "tokio", "tracing", "tracing-subscriber", + "uuid", ] [[package]] diff --git a/app/src/lib/api/logs.ts b/app/src/lib/api/logs.ts new file mode 100644 index 00000000..033a215c --- /dev/null +++ b/app/src/lib/api/logs.ts @@ -0,0 +1,83 @@ +const API_BASE = '/api'; + +export interface LogEntry { + id: string; + service: string; + level: string; + classification: string; + message: string; + agent_id: string | null; + workload_id: string | null; + organization_id: string | null; + created_at: string; +} + +export interface LogsResponse { + entries: LogEntry[]; + total: number; + has_more: boolean; +} + +export interface LogsFilter { + service?: string; + level?: string; + classification?: string; + agent_id?: string; + workload_id?: string; + organization_id?: string; + from?: string; + to?: string; + q?: string; + limit?: number; + offset?: number; +} + +export interface LogsRetention { + retention_days: number; +} + +function buildQuery(filter: LogsFilter): string { + const params = new URLSearchParams(); + for (const [key, value] of Object.entries(filter)) { + if (value !== undefined && value !== null && value !== '') { + params.set(key, String(value)); + } + } + const query = params.toString(); + return query ? `?${query}` : ''; +} + +export async function listLogs(token: string, filter: LogsFilter): Promise { + const res = await fetch(`${API_BASE}/logs${buildQuery(filter)}`, { + headers: { Authorization: `Bearer ${token}` }, + }); + if (!res.ok) throw new Error(`Failed to list logs: ${res.status}`); + return res.json(); +} + +export async function getLogsRetention(token: string): Promise { + const res = await fetch(`${API_BASE}/admin/settings/logs-retention`, { + headers: { Authorization: `Bearer ${token}` }, + }); + if (!res.ok) throw new Error(`Failed to get logs retention: ${res.status}`); + return res.json(); +} + +export async function updateLogsRetention( + token: string, + retentionDays: number, +): Promise { + const res = await fetch(`${API_BASE}/admin/settings/logs-retention`, { + method: 'PUT', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ retention_days: retentionDays }), + }); + if (!res.ok) { + const err = await res.json().catch(() => ({ error: res.status })); + throw new Error(err.error ?? `Failed to update logs retention: ${res.status}`); + } + return res.json(); +} diff --git a/app/src/lib/components/sidebar/app-sidebar.svelte b/app/src/lib/components/sidebar/app-sidebar.svelte index 3376f022..2959d605 100644 --- a/app/src/lib/components/sidebar/app-sidebar.svelte +++ b/app/src/lib/components/sidebar/app-sidebar.svelte @@ -3,7 +3,14 @@ import ChartPieIcon from "@lucide/svelte/icons/chart-pie"; import MapIcon from "@lucide/svelte/icons/map"; import BookOpenIcon from "@lucide/svelte/icons/book-open"; - import { Server, Layers, ShipWheel, Container, LaptopMinimal, Settings } from "@lucide/svelte"; + import { + Server, + Layers, + ShipWheel, + Container, + LaptopMinimal, + Settings, + } from "@lucide/svelte"; import IconBucket from "./icon-bucket.svelte"; import IconDashboard from "./icon-dashboard.svelte"; import IconActivity from "./icon-activity.svelte"; @@ -25,7 +32,7 @@ }, { title: "Logs", - url: "#", + url: "/logs", icon: IconLogs, }, ], diff --git a/app/src/routes/(app)/logs/+page.svelte b/app/src/routes/(app)/logs/+page.svelte new file mode 100644 index 00000000..3aae3cc7 --- /dev/null +++ b/app/src/routes/(app)/logs/+page.svelte @@ -0,0 +1,243 @@ + + +
+ + / + Logs +
+ +
+ + +
+ {#if error} +

{error}

+ {/if} +
+ {#each visibleServices() as service (service)} + {@const serviceEntries = entriesForService(service)} +
+
+ {service} + {serviceEntries.length} +
+
+ {#each serviceEntries as entry (entry.id)} +
+ {formatTime(entry.created_at)} + {entry.level} + {entry.message} +
+ {/each} +
+
+ {/each} +
+
+
diff --git a/control-plane/api-gateway/src/auth/rbac.rs b/control-plane/api-gateway/src/auth/rbac.rs index 9d070ef1..ce94fd9b 100644 --- a/control-plane/api-gateway/src/auth/rbac.rs +++ b/control-plane/api-gateway/src/auth/rbac.rs @@ -22,6 +22,8 @@ pub struct CanManageNetworks(pub Claims); pub struct CanManageSystem(pub Claims); pub struct CanViewResourceGroups(pub Claims); pub struct CanManageResourceGroups(pub Claims); +pub struct CanViewLogs(pub Claims); +pub struct CanManageLogs(pub Claims); async fn extract_claims(parts: &mut Parts, state: &AppState) -> Result { let token = parts @@ -117,3 +119,5 @@ impl_extractor!(CanManageNetworks, "networks", "manage"); impl_extractor!(CanManageSystem, "system", "manage"); impl_extractor!(CanViewResourceGroups, "resource_groups", "view"); impl_extractor!(CanManageResourceGroups, "resource_groups", "manage"); +impl_extractor!(CanViewLogs, "logs", "view"); +impl_extractor!(CanManageLogs, "logs", "manage"); diff --git a/control-plane/api-gateway/src/init.rs b/control-plane/api-gateway/src/init.rs index 5f851917..4a70a724 100644 --- a/control-plane/api-gateway/src/init.rs +++ b/control-plane/api-gateway/src/init.rs @@ -160,6 +160,13 @@ pub async fn initialize_database( "manage", "Create, update, and delete resource groups", ), + ("logs.view", "logs", "view", "View system and service logs"), + ( + "logs.manage", + "logs", + "manage", + "Manage log retention settings", + ), ]; let mut permission_map = std::collections::HashMap::new(); @@ -262,6 +269,7 @@ pub async fn initialize_database( "members.view", "resource_groups.view", "resource_groups.manage", + "logs.view", ]; for perm_name in operator_perms { if let Some(perm_id) = permission_map.get(perm_name) { @@ -306,6 +314,7 @@ pub async fn initialize_database( "organization.view", "members.view", "resource_groups.view", + "logs.view", ]; for perm_name in viewer_perms { if let Some(perm_id) = permission_map.get(perm_name) { diff --git a/control-plane/api-gateway/src/main.rs b/control-plane/api-gateway/src/main.rs index 64a169bb..bc256352 100644 --- a/control-plane/api-gateway/src/main.rs +++ b/control-plane/api-gateway/src/main.rs @@ -8,6 +8,7 @@ mod auth_service; mod db; mod init; mod metrics; +mod prune; mod rbac_service; mod routes; mod self_monitor; @@ -137,7 +138,7 @@ async fn main() { .expect("failed to install rustls crypto provider"); metrics::init(); - telemetry::init_tracing(); + let log_receiver = telemetry::init_tracing(); let db_conn = match db::establish_connection().await { Ok(conn) => { @@ -149,6 +150,7 @@ async fn main() { std::process::exit(1); } }; + shared::spawn_log_writer(log_receiver, db_conn.clone()); let default_org_id = match init::initialize_database(&db_conn).await { Ok(id) => id, @@ -164,6 +166,9 @@ async fn main() { default_org_id: Some(default_org_id), }; + tracing::info!("starting log prune job"); + prune::spawn_log_prune_job(state.clone()); + tracing::info!("starting self-monitoring service"); self_monitor::start_self_monitoring(std::sync::Arc::new(db_conn)).await; diff --git a/control-plane/api-gateway/src/prune.rs b/control-plane/api-gateway/src/prune.rs new file mode 100644 index 00000000..18a45fd6 --- /dev/null +++ b/control-plane/api-gateway/src/prune.rs @@ -0,0 +1,55 @@ +use chrono::Utc; +use entity::{entities::logs, Logs}; +use sea_orm::{ColumnTrait, DbConn, EntityTrait, QueryFilter}; +use tokio::time::{interval, Duration}; + +use crate::routes::settings::load_retention_days; +use crate::AppState; + +const PRUNE_INTERVAL: Duration = Duration::from_secs(3600); + +pub fn spawn_log_prune_job(state: AppState) { + tokio::spawn(async move { + let mut ticker = interval(PRUNE_INTERVAL); + loop { + ticker.tick().await; + prune_once(&state).await; + } + }); +} + +async fn prune_once(state: &AppState) { + let retention_days = match load_retention_days(state).await { + Ok(days) => days, + Err(_) => { + tracing::warn!("skipping log prune, retention setting unavailable"); + return; + } + }; + + let cutoff = Utc::now() - chrono::Duration::days(retention_days); + + match delete_logs_older_than(&state.db_conn, cutoff).await { + Ok(rows_deleted) => { + tracing::info!( + rows_deleted = rows_deleted, + retention_days = retention_days, + "pruned expired logs" + ); + } + Err(error) => { + tracing::error!(error = %error, "failed to prune expired logs"); + } + } +} + +async fn delete_logs_older_than( + db: &DbConn, + cutoff: chrono::DateTime, +) -> Result { + let result = Logs::delete_many() + .filter(logs::Column::CreatedAt.lt(cutoff)) + .exec(db) + .await?; + Ok(result.rows_affected) +} diff --git a/control-plane/api-gateway/src/routes/logs.rs b/control-plane/api-gateway/src/routes/logs.rs new file mode 100644 index 00000000..e8734885 --- /dev/null +++ b/control-plane/api-gateway/src/routes/logs.rs @@ -0,0 +1,127 @@ +use axum::{ + extract::{Query, State}, + http::StatusCode, + response::{IntoResponse, Json}, + routing::get, + Router, +}; +use entity::{entities::logs, Logs}; +use sea_orm::{ + ColumnTrait, EntityTrait, Order, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, +}; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use uuid::Uuid; + +use crate::{auth::rbac::CanViewLogs, AppState}; + +const DEFAULT_LIMIT: u64 = 100; +const MAX_LIMIT: u64 = 500; + +#[derive(Debug, Deserialize)] +pub struct LogsQuery { + pub service: Option, + pub level: Option, + pub classification: Option, + pub agent_id: Option, + pub workload_id: Option, + pub organization_id: Option, + pub from: Option>, + pub to: Option>, + pub q: Option, + pub limit: Option, + pub offset: Option, +} + +#[derive(Debug, Serialize)] +pub struct LogsResponse { + pub entries: Vec, + pub total: u64, + pub has_more: bool, +} + +fn apply_filters( + mut query: sea_orm::Select, + params: &LogsQuery, +) -> sea_orm::Select { + if let Some(service) = ¶ms.service { + query = query.filter(logs::Column::Service.eq(service)); + } + if let Some(level) = ¶ms.level { + query = query.filter(logs::Column::Level.eq(level)); + } + if let Some(classification) = ¶ms.classification { + query = query.filter(logs::Column::Classification.eq(classification)); + } + if let Some(agent_id) = params.agent_id { + query = query.filter(logs::Column::AgentId.eq(agent_id)); + } + if let Some(workload_id) = params.workload_id { + query = query.filter(logs::Column::WorkloadId.eq(workload_id)); + } + if let Some(organization_id) = params.organization_id { + query = query.filter(logs::Column::OrganizationId.eq(organization_id)); + } + if let Some(from) = params.from { + query = query.filter(logs::Column::CreatedAt.gte(from)); + } + if let Some(to) = params.to { + query = query.filter(logs::Column::CreatedAt.lte(to)); + } + if let Some(q) = ¶ms.q { + query = query.filter(logs::Column::Message.contains(q)); + } + query +} + +pub async fn list_logs( + CanViewLogs(_claims): CanViewLogs, + State(state): State, + Query(params): Query, +) -> Result)> { + let limit = params.limit.unwrap_or(DEFAULT_LIMIT).min(MAX_LIMIT); + let offset = params.offset.unwrap_or(0); + + let base_query = apply_filters(Logs::find(), ¶ms); + + let total = base_query + .clone() + .count(&state.db_conn) + .await + .map_err(|e| { + tracing::error!(error = %e, "failed to count logs"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ "error": "database error" })), + ) + })?; + + let entries = base_query + .order_by(logs::Column::CreatedAt, Order::Desc) + .offset(offset) + .limit(limit) + .all(&state.db_conn) + .await + .map_err(|e| { + tracing::error!(error = %e, "failed to list logs"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ "error": "database error" })), + ) + })?; + + let has_more = offset + (entries.len() as u64) < total; + + Ok(( + StatusCode::OK, + Json(json!(LogsResponse { + entries, + total, + has_more, + })), + )) +} + +pub fn logs_routes() -> Router { + Router::new().route("/logs", get(list_logs)) +} diff --git a/control-plane/api-gateway/src/routes/mod.rs b/control-plane/api-gateway/src/routes/mod.rs index 9e7ab1a1..388d606b 100644 --- a/control-plane/api-gateway/src/routes/mod.rs +++ b/control-plane/api-gateway/src/routes/mod.rs @@ -16,11 +16,13 @@ use tracing::{info_span, Span}; pub mod agents; pub mod events; +pub mod logs; pub mod networks; pub mod organizations; pub mod registry; pub mod releases; pub mod resource_groups; +pub mod settings; pub mod ssh_keys; pub mod system; pub mod update; @@ -95,6 +97,8 @@ pub fn create_router() -> Router { .merge(workloads::workloads_routes()) .merge(events::events_routes()) .merge(resource_groups::resource_groups_routes()) + .merge(logs::logs_routes()) + .merge(settings::settings_routes()) .layer(GovernorLayer::new(governor_config)); let api_router = Router::new() diff --git a/control-plane/api-gateway/src/routes/settings.rs b/control-plane/api-gateway/src/routes/settings.rs new file mode 100644 index 00000000..5598b4bf --- /dev/null +++ b/control-plane/api-gateway/src/routes/settings.rs @@ -0,0 +1,121 @@ +use axum::{ + extract::State, + http::StatusCode, + response::{IntoResponse, Json}, + routing::get, + Router, +}; +use entity::{entities::system_settings, SystemSettings}; +use sea_orm::{ActiveModelTrait, ActiveValue::Set, EntityTrait}; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +use crate::{auth::rbac::CanManageLogs, AppState}; + +const RETENTION_KEY: &str = "logs.retention_days"; +const MIN_RETENTION_DAYS: i64 = 1; +const MAX_RETENTION_DAYS: i64 = 365; + +#[derive(Debug, Serialize)] +pub struct LogsRetentionResponse { + pub retention_days: i64, +} + +#[derive(Debug, Deserialize)] +pub struct UpdateLogsRetentionRequest { + pub retention_days: i64, +} + +pub async fn get_logs_retention( + CanManageLogs(_claims): CanManageLogs, + State(state): State, +) -> Result)> { + let retention_days = load_retention_days(&state).await?; + Ok(( + StatusCode::OK, + Json(json!(LogsRetentionResponse { retention_days })), + )) +} + +pub async fn update_logs_retention( + CanManageLogs(_claims): CanManageLogs, + State(state): State, + Json(req): Json, +) -> Result)> { + if req.retention_days < MIN_RETENTION_DAYS || req.retention_days > MAX_RETENTION_DAYS { + return Err(( + StatusCode::BAD_REQUEST, + Json(json!({ "error": format!( + "retention_days must be between {} and {}", + MIN_RETENTION_DAYS, MAX_RETENTION_DAYS + ) })), + )); + } + + let existing = SystemSettings::find_by_id(RETENTION_KEY) + .one(&state.db_conn) + .await + .map_err(|e| { + tracing::error!(error = %e, "failed to load logs retention setting"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ "error": "database error" })), + ) + })?; + + let now = chrono::Utc::now().into(); + let model = match existing { + Some(setting) => { + let mut active: system_settings::ActiveModel = setting.into(); + active.value = Set(json!(req.retention_days)); + active.updated_at = Set(now); + active + } + None => system_settings::ActiveModel { + key: Set(RETENTION_KEY.to_string()), + value: Set(json!(req.retention_days)), + updated_at: Set(now), + }, + }; + + model.save(&state.db_conn).await.map_err(|e| { + tracing::error!(error = %e, "failed to save logs retention setting"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ "error": "database error" })), + ) + })?; + + Ok(( + StatusCode::OK, + Json(json!(LogsRetentionResponse { + retention_days: req.retention_days + })), + )) +} + +pub async fn load_retention_days( + state: &AppState, +) -> Result)> { + let setting = SystemSettings::find_by_id(RETENTION_KEY) + .one(&state.db_conn) + .await + .map_err(|e| { + tracing::error!(error = %e, "failed to load logs retention setting"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ "error": "database error" })), + ) + })?; + + Ok(setting + .and_then(|s| s.value.as_i64()) + .unwrap_or(MAX_RETENTION_DAYS.min(30))) +} + +pub fn settings_routes() -> Router { + Router::new().route( + "/admin/settings/logs-retention", + get(get_logs_retention).put(update_logs_retention), + ) +} diff --git a/control-plane/api-gateway/src/telemetry.rs b/control-plane/api-gateway/src/telemetry.rs index 0614fc30..f172e1f9 100644 --- a/control-plane/api-gateway/src/telemetry.rs +++ b/control-plane/api-gateway/src/telemetry.rs @@ -28,11 +28,15 @@ fn build_otlp_provider() -> Option { Some(provider) } -pub fn init_tracing() { +pub fn init_tracing() -> tokio::sync::mpsc::UnboundedReceiver { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let fmt_layer = fmt::layer().with_target(false); + let (db_layer, db_receiver) = shared::db_log_layer::DbLogLayer::new("api-gateway"); - let registry = tracing_subscriber::registry().with(filter).with(fmt_layer); + let registry = tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(db_layer); match build_otlp_provider() { Some(provider) => { @@ -45,4 +49,6 @@ pub fn init_tracing() { registry.init(); } } + + db_receiver } diff --git a/control-plane/failover-controller/src/logger.rs b/control-plane/failover-controller/src/logger.rs index 727fe564..7d8b6529 100644 --- a/control-plane/failover-controller/src/logger.rs +++ b/control-plane/failover-controller/src/logger.rs @@ -50,15 +50,21 @@ fn build_otlp_provider(service_name: &str) -> Option { Some(provider) } -pub fn init_logger() { - init_logger_with_service(env!("CARGO_PKG_NAME")); +pub fn init_logger() -> tokio::sync::mpsc::UnboundedReceiver { + init_logger_with_service(env!("CARGO_PKG_NAME")) } -pub fn init_logger_with_service(service_name: &'static str) { +pub fn init_logger_with_service( + service_name: &'static str, +) -> tokio::sync::mpsc::UnboundedReceiver { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let fmt_layer = fmt::layer().with_target(false).with_thread_ids(true); + let (db_layer, db_receiver) = shared::db_log_layer::DbLogLayer::new(service_name); - let registry = tracing_subscriber::registry().with(filter).with(fmt_layer); + let registry = tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(db_layer); match build_otlp_provider(service_name) { Some(provider) => { @@ -71,6 +77,8 @@ pub fn init_logger_with_service(service_name: &'static str) { registry.init(); } } + + db_receiver } pub fn log_message(level: LogLevel, module: &str, location: &str, description: &str) { diff --git a/control-plane/failover-controller/src/main.rs b/control-plane/failover-controller/src/main.rs index 65c17d05..988cf090 100644 --- a/control-plane/failover-controller/src/main.rs +++ b/control-plane/failover-controller/src/main.rs @@ -16,7 +16,7 @@ async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); - logger::init_logger(); + let log_receiver = logger::init_logger(); metrics::init(); log_info!("main", "CSFX Failover Controller starting..."); @@ -27,6 +27,7 @@ async fn main() -> anyhow::Result<()> { .await .expect("Failed to connect to database"); log_info!("main", "Database connection established"); + shared::spawn_log_writer(log_receiver, db.clone()); let state = server::AppState { db: db.clone() }; diff --git a/control-plane/registry/src/logger.rs b/control-plane/registry/src/logger.rs index 727fe564..7d8b6529 100644 --- a/control-plane/registry/src/logger.rs +++ b/control-plane/registry/src/logger.rs @@ -50,15 +50,21 @@ fn build_otlp_provider(service_name: &str) -> Option { Some(provider) } -pub fn init_logger() { - init_logger_with_service(env!("CARGO_PKG_NAME")); +pub fn init_logger() -> tokio::sync::mpsc::UnboundedReceiver { + init_logger_with_service(env!("CARGO_PKG_NAME")) } -pub fn init_logger_with_service(service_name: &'static str) { +pub fn init_logger_with_service( + service_name: &'static str, +) -> tokio::sync::mpsc::UnboundedReceiver { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let fmt_layer = fmt::layer().with_target(false).with_thread_ids(true); + let (db_layer, db_receiver) = shared::db_log_layer::DbLogLayer::new(service_name); - let registry = tracing_subscriber::registry().with(filter).with(fmt_layer); + let registry = tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(db_layer); match build_otlp_provider(service_name) { Some(provider) => { @@ -71,6 +77,8 @@ pub fn init_logger_with_service(service_name: &'static str) { registry.init(); } } + + db_receiver } pub fn log_message(level: LogLevel, module: &str, location: &str, description: &str) { diff --git a/control-plane/registry/src/main.rs b/control-plane/registry/src/main.rs index 4b341cd4..e686381c 100644 --- a/control-plane/registry/src/main.rs +++ b/control-plane/registry/src/main.rs @@ -17,7 +17,7 @@ async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); - logger::init_logger(); + let log_receiver = logger::init_logger(); metrics::init(); log_info!("main", "CSFX Registry Service starting..."); @@ -28,6 +28,7 @@ async fn main() -> anyhow::Result<()> { .await .expect("Failed to connect to database"); log_info!("main", "Database connection established"); + shared::spawn_log_writer(log_receiver, db_conn.clone()); let cert_ttl_hours: i64 = std::env::var("CSFX_CERT_TTL_HOURS") .ok() diff --git a/control-plane/scheduler/src/logger.rs b/control-plane/scheduler/src/logger.rs index 727fe564..7d8b6529 100644 --- a/control-plane/scheduler/src/logger.rs +++ b/control-plane/scheduler/src/logger.rs @@ -50,15 +50,21 @@ fn build_otlp_provider(service_name: &str) -> Option { Some(provider) } -pub fn init_logger() { - init_logger_with_service(env!("CARGO_PKG_NAME")); +pub fn init_logger() -> tokio::sync::mpsc::UnboundedReceiver { + init_logger_with_service(env!("CARGO_PKG_NAME")) } -pub fn init_logger_with_service(service_name: &'static str) { +pub fn init_logger_with_service( + service_name: &'static str, +) -> tokio::sync::mpsc::UnboundedReceiver { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let fmt_layer = fmt::layer().with_target(false).with_thread_ids(true); + let (db_layer, db_receiver) = shared::db_log_layer::DbLogLayer::new(service_name); - let registry = tracing_subscriber::registry().with(filter).with(fmt_layer); + let registry = tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(db_layer); match build_otlp_provider(service_name) { Some(provider) => { @@ -71,6 +77,8 @@ pub fn init_logger_with_service(service_name: &'static str) { registry.init(); } } + + db_receiver } pub fn log_message(level: LogLevel, module: &str, location: &str, description: &str) { diff --git a/control-plane/scheduler/src/main.rs b/control-plane/scheduler/src/main.rs index 743d285f..e821ea86 100644 --- a/control-plane/scheduler/src/main.rs +++ b/control-plane/scheduler/src/main.rs @@ -14,7 +14,7 @@ mod services; async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); - logger::init_logger(); + let log_receiver = logger::init_logger(); metrics::init(); log_info!("main", "CSFX Scheduler Service starting..."); @@ -25,6 +25,7 @@ async fn main() -> anyhow::Result<()> { .await .expect("Failed to connect to database"); log_info!("main", "Database connection established"); + shared::spawn_log_writer(log_receiver, db.clone()); let etcd_endpoints = std::env::var("ETCD_ENDPOINTS").unwrap_or_else(|_| "http://localhost:2379".to_string()); diff --git a/control-plane/sdn-controller/src/logger.rs b/control-plane/sdn-controller/src/logger.rs index 727fe564..7d8b6529 100644 --- a/control-plane/sdn-controller/src/logger.rs +++ b/control-plane/sdn-controller/src/logger.rs @@ -50,15 +50,21 @@ fn build_otlp_provider(service_name: &str) -> Option { Some(provider) } -pub fn init_logger() { - init_logger_with_service(env!("CARGO_PKG_NAME")); +pub fn init_logger() -> tokio::sync::mpsc::UnboundedReceiver { + init_logger_with_service(env!("CARGO_PKG_NAME")) } -pub fn init_logger_with_service(service_name: &'static str) { +pub fn init_logger_with_service( + service_name: &'static str, +) -> tokio::sync::mpsc::UnboundedReceiver { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let fmt_layer = fmt::layer().with_target(false).with_thread_ids(true); + let (db_layer, db_receiver) = shared::db_log_layer::DbLogLayer::new(service_name); - let registry = tracing_subscriber::registry().with(filter).with(fmt_layer); + let registry = tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(db_layer); match build_otlp_provider(service_name) { Some(provider) => { @@ -71,6 +77,8 @@ pub fn init_logger_with_service(service_name: &'static str) { registry.init(); } } + + db_receiver } pub fn log_message(level: LogLevel, module: &str, location: &str, description: &str) { diff --git a/control-plane/sdn-controller/src/main.rs b/control-plane/sdn-controller/src/main.rs index 5d127850..c397a715 100644 --- a/control-plane/sdn-controller/src/main.rs +++ b/control-plane/sdn-controller/src/main.rs @@ -16,7 +16,7 @@ async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); - logger::init_logger(); + let log_receiver = logger::init_logger(); metrics::init(); log_info!("main", "CSFX SDN Controller starting..."); @@ -27,6 +27,7 @@ async fn main() -> anyhow::Result<()> { .await .expect("Failed to connect to database"); log_info!("main", "Database connection established"); + shared::spawn_log_writer(log_receiver, db.clone()); let etcd_url = std::env::var("ETCD_URL").unwrap_or_else(|_| "http://localhost:2379".to_string()); diff --git a/control-plane/shared/entity/src/entities/logs.rs b/control-plane/shared/entity/src/entities/logs.rs new file mode 100644 index 00000000..2a8987b2 --- /dev/null +++ b/control-plane/shared/entity/src/entities/logs.rs @@ -0,0 +1,65 @@ +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)] +#[sea_orm(table_name = "logs")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: Uuid, + pub service: String, + pub level: String, + pub classification: String, + pub message: String, + pub agent_id: Option, + pub workload_id: Option, + pub organization_id: Option, + pub created_at: DateTimeWithTimeZone, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::agents::Entity", + from = "Column::AgentId", + to = "super::agents::Column::Id", + on_update = "NoAction", + on_delete = "SetNull" + )] + Agent, + #[sea_orm( + belongs_to = "super::workloads::Entity", + from = "Column::WorkloadId", + to = "super::workloads::Column::Id", + on_update = "NoAction", + on_delete = "SetNull" + )] + Workload, + #[sea_orm( + belongs_to = "super::organization::Entity", + from = "Column::OrganizationId", + to = "super::organization::Column::Id", + on_update = "NoAction", + on_delete = "SetNull" + )] + Organization, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Agent.def() + } +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Workload.def() + } +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Organization.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/control-plane/shared/entity/src/entities/mod.rs b/control-plane/shared/entity/src/entities/mod.rs index 6c54b09a..31e24b11 100644 --- a/control-plane/shared/entity/src/entities/mod.rs +++ b/control-plane/shared/entity/src/entities/mod.rs @@ -4,10 +4,10 @@ pub mod agent_metrics; pub mod agents; pub mod bootstrap_tokens; pub mod certificate_revocations; -pub mod config; pub mod failover_events; pub mod invalid_jwt; pub mod key; +pub mod logs; pub mod network_members; pub mod network_policies; pub mod networks; @@ -17,6 +17,7 @@ pub mod registry_tokens; pub mod resource_groups; pub mod role; pub mod role_permission; +pub mod system_settings; pub mod user; pub mod user_organization; pub mod user_ssh_keys; @@ -30,10 +31,10 @@ pub use agent_metrics::Entity as AgentMetrics; pub use agents::Entity as Agents; pub use bootstrap_tokens::Entity as BootstrapTokens; pub use certificate_revocations::Entity as CertificateRevocations; -pub use config::Entity as Config; pub use failover_events::Entity as FailoverEvents; pub use invalid_jwt::Entity as InvalidJwt; pub use key::Entity as Key; +pub use logs::Entity as Logs; pub use network_members::Entity as NetworkMembers; pub use network_policies::Entity as NetworkPolicies; pub use networks::Entity as Networks; @@ -43,6 +44,7 @@ pub use registry_tokens::Entity as RegistryTokens; pub use resource_groups::Entity as ResourceGroups; pub use role::Entity as Role; pub use role_permission::Entity as RolePermission; +pub use system_settings::Entity as SystemSettings; pub use user::Entity as User; pub use user_organization::Entity as UserOrganization; pub use user_ssh_keys::Entity as UserSshKeys; diff --git a/control-plane/shared/entity/src/entities/config.rs b/control-plane/shared/entity/src/entities/system_settings.rs similarity index 63% rename from control-plane/shared/entity/src/entities/config.rs rename to control-plane/shared/entity/src/entities/system_settings.rs index 9233c1f2..74b56bca 100644 --- a/control-plane/shared/entity/src/entities/config.rs +++ b/control-plane/shared/entity/src/entities/system_settings.rs @@ -3,11 +3,12 @@ use sea_orm::JsonValue; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] -#[sea_orm(table_name = "config")] +#[sea_orm(table_name = "system_settings")] pub struct Model { - #[sea_orm(primary_key)] - pub id: Uuid, - pub config: JsonValue, + #[sea_orm(primary_key, auto_increment = false)] + pub key: String, + pub value: JsonValue, + pub updated_at: DateTimeWithTimeZone, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/control-plane/shared/migration/src/lib.rs b/control-plane/shared/migration/src/lib.rs index 4365d679..32a4836d 100644 --- a/control-plane/shared/migration/src/lib.rs +++ b/control-plane/shared/migration/src/lib.rs @@ -21,6 +21,7 @@ mod m20260523_000000_add_ssh_keys; mod m20260625_000000_add_resource_groups; mod m20260628_000000_add_volume_mounts; mod m20260628_100000_rg_cidr_unique_agent_wg; +mod m20260701_000000_add_logs_and_settings; pub struct Migrator; @@ -49,6 +50,7 @@ impl MigratorTrait for Migrator { Box::new(m20260625_000000_add_resource_groups::Migration), Box::new(m20260628_000000_add_volume_mounts::Migration), Box::new(m20260628_100000_rg_cidr_unique_agent_wg::Migration), + Box::new(m20260701_000000_add_logs_and_settings::Migration), ] } } diff --git a/control-plane/shared/migration/src/m20260701_000000_add_logs_and_settings.rs b/control-plane/shared/migration/src/m20260701_000000_add_logs_and_settings.rs new file mode 100644 index 00000000..84c8df68 --- /dev/null +++ b/control-plane/shared/migration/src/m20260701_000000_add_logs_and_settings.rs @@ -0,0 +1,178 @@ +use sea_orm_migration::{prelude::*, schema::*}; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(Config::Table).if_exists().to_owned()) + .await?; + + manager + .create_table( + Table::create() + .table(SystemSettings::Table) + .if_not_exists() + .col(string(SystemSettings::Key).primary_key()) + .col(json_binary(SystemSettings::Value)) + .col( + timestamp_with_time_zone(SystemSettings::UpdatedAt) + .default(Expr::current_timestamp()), + ) + .to_owned(), + ) + .await?; + + manager + .get_connection() + .execute_unprepared( + "INSERT INTO system_settings (key, value, updated_at) \ + VALUES ('logs.retention_days', '30', now())", + ) + .await?; + + manager + .create_table( + Table::create() + .table(Logs::Table) + .if_not_exists() + .col(pk_uuid(Logs::Id)) + .col(string(Logs::Service)) + .col(string(Logs::Level)) + .col(string(Logs::Classification)) + .col(text(Logs::Message)) + .col(uuid_null(Logs::AgentId)) + .col(uuid_null(Logs::WorkloadId)) + .col(uuid_null(Logs::OrganizationId)) + .col( + timestamp_with_time_zone(Logs::CreatedAt) + .default(Expr::current_timestamp()), + ) + .foreign_key( + ForeignKey::create() + .name("fk_logs_agent") + .from(Logs::Table, Logs::AgentId) + .to(Agents::Table, Agents::Id) + .on_delete(ForeignKeyAction::SetNull) + .on_update(ForeignKeyAction::NoAction), + ) + .foreign_key( + ForeignKey::create() + .name("fk_logs_workload") + .from(Logs::Table, Logs::WorkloadId) + .to(Workloads::Table, Workloads::Id) + .on_delete(ForeignKeyAction::SetNull) + .on_update(ForeignKeyAction::NoAction), + ) + .foreign_key( + ForeignKey::create() + .name("fk_logs_organization") + .from(Logs::Table, Logs::OrganizationId) + .to(Organization::Table, Organization::Id) + .on_delete(ForeignKeyAction::SetNull) + .on_update(ForeignKeyAction::NoAction), + ) + .to_owned(), + ) + .await?; + + for (index_name, column) in [ + ("idx_logs_service", Logs::Service), + ("idx_logs_level", Logs::Level), + ("idx_logs_classification", Logs::Classification), + ("idx_logs_agent_id", Logs::AgentId), + ("idx_logs_workload_id", Logs::WorkloadId), + ("idx_logs_organization_id", Logs::OrganizationId), + ] { + manager + .create_index( + Index::create() + .name(index_name) + .table(Logs::Table) + .col(column) + .to_owned(), + ) + .await?; + } + + manager + .get_connection() + .execute_unprepared( + "CREATE INDEX idx_logs_created_at ON logs (created_at DESC)", + ) + .await?; + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(Logs::Table).to_owned()) + .await?; + + manager + .drop_table(Table::drop().table(SystemSettings::Table).to_owned()) + .await?; + + manager + .create_table( + Table::create() + .table(Config::Table) + .if_not_exists() + .col(pk_uuid(Config::Id)) + .col(json_binary(Config::Config)) + .to_owned(), + ) + .await + } +} + +#[derive(DeriveIden)] +enum Config { + Table, + Id, + #[allow(clippy::enum_variant_names)] + Config, +} + +#[derive(DeriveIden)] +enum SystemSettings { + Table, + Key, + Value, + UpdatedAt, +} + +#[derive(DeriveIden)] +enum Logs { + Table, + Id, + Service, + Level, + Classification, + Message, + AgentId, + WorkloadId, + OrganizationId, + CreatedAt, +} + +#[derive(DeriveIden)] +enum Agents { + Table, + Id, +} + +#[derive(DeriveIden)] +enum Workloads { + Table, + Id, +} + +#[derive(DeriveIden)] +enum Organization { + Table, + Id, +} diff --git a/control-plane/shared/shared/Cargo.toml b/control-plane/shared/shared/Cargo.toml index 8853909c..fde950cd 100644 --- a/control-plane/shared/shared/Cargo.toml +++ b/control-plane/shared/shared/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true [dependencies] # Database sea-orm = { workspace = true } +entity = { path = "../entity" } # Async tokio = { workspace = true } @@ -20,6 +21,8 @@ tracing-subscriber = { workspace = true } dotenvy = { workspace = true } anyhow = { workspace = true } thiserror = { workspace = true } +uuid = { workspace = true } +chrono = { workspace = true } # Serialization serde = { workspace = true } diff --git a/control-plane/shared/shared/src/db_log_layer.rs b/control-plane/shared/shared/src/db_log_layer.rs new file mode 100644 index 00000000..84b03d72 --- /dev/null +++ b/control-plane/shared/shared/src/db_log_layer.rs @@ -0,0 +1,193 @@ +use entity::logs; +use sea_orm::{ActiveValue::Set, DbConn, EntityTrait}; +use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender}; +use tokio::time::{interval, Duration}; +use tracing::field::{Field, Visit}; +use tracing::{Event, Level, Subscriber}; +use tracing_subscriber::layer::Context; +use tracing_subscriber::Layer; +use uuid::Uuid; + +const FLUSH_INTERVAL: Duration = Duration::from_secs(1); +const FLUSH_BATCH_SIZE: usize = 100; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum LogClassification { + Security, + Performance, + Audit, + System, + Network, + Storage, +} + +impl LogClassification { + fn as_str(self) -> &'static str { + match self { + Self::Security => "security", + Self::Performance => "performance", + Self::Audit => "audit", + Self::System => "system", + Self::Network => "network", + Self::Storage => "storage", + } + } + + fn from_target(target: &str) -> Self { + if target.contains("auth") || target.contains("rbac") || target.contains("jwt") { + Self::Security + } else if target.contains("scheduler") || target.contains("metrics") { + Self::Performance + } else if target.contains("network") || target.contains("sdn") || target.contains("wireguard") { + Self::Network + } else if target.contains("volume") || target.contains("storage") { + Self::Storage + } else if target.contains("registry") { + Self::Audit + } else { + Self::System + } + } +} + +pub struct LogRecord { + pub service: &'static str, + pub level: &'static str, + pub classification: &'static str, + pub message: String, + pub agent_id: Option, + pub workload_id: Option, + pub organization_id: Option, +} + +#[derive(Default)] +struct EventVisitor { + message: String, + agent_id: Option, + workload_id: Option, + organization_id: Option, +} + +impl Visit for EventVisitor { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.record_field(field.name(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.record_field(field.name(), value.to_string()); + } +} + +impl EventVisitor { + fn record_field(&mut self, name: &str, value: String) { + match name { + "message" => self.message = trim_quotes(value), + "agent_id" => self.agent_id = Uuid::parse_str(&trim_quotes(value)).ok(), + "workload_id" => self.workload_id = Uuid::parse_str(&trim_quotes(value)).ok(), + "organization_id" => self.organization_id = Uuid::parse_str(&trim_quotes(value)).ok(), + _ => {} + } + } +} + +fn trim_quotes(value: String) -> String { + value.trim_matches('"').to_string() +} + +fn level_as_str(level: &Level) -> &'static str { + match *level { + Level::TRACE | Level::DEBUG => "DEBUG", + Level::INFO => "INFO", + Level::WARN => "WARN", + Level::ERROR => "ERROR", + } +} + +pub struct DbLogLayer { + service: &'static str, + sender: UnboundedSender, +} + +impl DbLogLayer { + pub fn new(service: &'static str) -> (Self, UnboundedReceiver) { + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + (Self { service, sender }, receiver) + } +} + +impl Layer for DbLogLayer { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + if *event.metadata().level() == Level::TRACE { + return; + } + + let mut visitor = EventVisitor::default(); + event.record(&mut visitor); + + let record = LogRecord { + service: self.service, + level: level_as_str(event.metadata().level()), + classification: LogClassification::from_target(event.metadata().target()).as_str(), + message: visitor.message, + agent_id: visitor.agent_id, + workload_id: visitor.workload_id, + organization_id: visitor.organization_id, + }; + + let _ = self.sender.send(record); + } +} + +pub fn spawn_log_writer(mut receiver: UnboundedReceiver, db: DbConn) { + tokio::spawn(async move { + let mut buffer = Vec::with_capacity(FLUSH_BATCH_SIZE); + let mut ticker = interval(FLUSH_INTERVAL); + + loop { + tokio::select! { + record = receiver.recv() => { + match record { + Some(record) => { + buffer.push(record); + if buffer.len() >= FLUSH_BATCH_SIZE { + flush(&db, &mut buffer).await; + } + } + None => { + flush(&db, &mut buffer).await; + break; + } + } + } + _ = ticker.tick() => { + flush(&db, &mut buffer).await; + } + } + } + }); +} + +async fn flush(db: &DbConn, buffer: &mut Vec) { + if buffer.is_empty() { + return; + } + + let models: Vec = buffer + .drain(..) + .map(|record| logs::ActiveModel { + id: Set(Uuid::new_v4()), + service: Set(record.service.to_string()), + level: Set(record.level.to_string()), + classification: Set(record.classification.to_string()), + message: Set(record.message), + agent_id: Set(record.agent_id), + workload_id: Set(record.workload_id), + organization_id: Set(record.organization_id), + created_at: Set(chrono::Utc::now().into()), + }) + .collect(); + + if let Err(error) = logs::Entity::insert_many(models).exec(db).await { + tracing::warn!(target: "shared::db_log_layer", error = %error, "failed to flush logs to database"); + } +} diff --git a/control-plane/shared/shared/src/lib.rs b/control-plane/shared/shared/src/lib.rs index 42ab76b1..cb23860c 100644 --- a/control-plane/shared/shared/src/lib.rs +++ b/control-plane/shared/shared/src/lib.rs @@ -1,5 +1,7 @@ pub mod db; +pub mod db_log_layer; pub mod logger; pub use db::establish_connection; +pub use db_log_layer::{spawn_log_writer, DbLogLayer}; pub use logger::init_logger; diff --git a/control-plane/volume-manager/src/logger.rs b/control-plane/volume-manager/src/logger.rs index 727fe564..7d8b6529 100644 --- a/control-plane/volume-manager/src/logger.rs +++ b/control-plane/volume-manager/src/logger.rs @@ -50,15 +50,21 @@ fn build_otlp_provider(service_name: &str) -> Option { Some(provider) } -pub fn init_logger() { - init_logger_with_service(env!("CARGO_PKG_NAME")); +pub fn init_logger() -> tokio::sync::mpsc::UnboundedReceiver { + init_logger_with_service(env!("CARGO_PKG_NAME")) } -pub fn init_logger_with_service(service_name: &'static str) { +pub fn init_logger_with_service( + service_name: &'static str, +) -> tokio::sync::mpsc::UnboundedReceiver { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let fmt_layer = fmt::layer().with_target(false).with_thread_ids(true); + let (db_layer, db_receiver) = shared::db_log_layer::DbLogLayer::new(service_name); - let registry = tracing_subscriber::registry().with(filter).with(fmt_layer); + let registry = tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(db_layer); match build_otlp_provider(service_name) { Some(provider) => { @@ -71,6 +77,8 @@ pub fn init_logger_with_service(service_name: &'static str) { registry.init(); } } + + db_receiver } pub fn log_message(level: LogLevel, module: &str, location: &str, description: &str) { diff --git a/control-plane/volume-manager/src/main.rs b/control-plane/volume-manager/src/main.rs index d380e0df..f26a1b87 100644 --- a/control-plane/volume-manager/src/main.rs +++ b/control-plane/volume-manager/src/main.rs @@ -20,7 +20,7 @@ async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); - logger::init_logger(); + let log_receiver = logger::init_logger(); metrics::init(); log_info!("main", "CSFX Volume Manager starting"); @@ -30,6 +30,7 @@ async fn main() -> anyhow::Result<()> { .await .expect("Failed to connect to database"); log_info!("main", "Database connection established"); + shared::spawn_log_writer(log_receiver, db.clone()); let etcd_endpoints = std::env::var("ETCD_ENDPOINTS").unwrap_or_else(|_| "http://localhost:2379".to_string()); From a6f751056c11bfbcd2dc548ebab909f49796e9b7 Mon Sep 17 00:00:00 2001 From: CodeMaster4711 Date: Thu, 2 Jul 2026 20:35:17 +0200 Subject: [PATCH 2/3] fix: filter noisy access, sql, heartbeat and performance events from log storage --- app/src/routes/(app)/logs/+page.svelte | 254 ++++++++++++------ control-plane/api-gateway/src/routes/mod.rs | 31 ++- .../shared/shared/src/db_log_layer.rs | 35 +++ 3 files changed, 236 insertions(+), 84 deletions(-) diff --git a/app/src/routes/(app)/logs/+page.svelte b/app/src/routes/(app)/logs/+page.svelte index 3aae3cc7..0c65b78d 100644 --- a/app/src/routes/(app)/logs/+page.svelte +++ b/app/src/routes/(app)/logs/+page.svelte @@ -3,7 +3,7 @@ import { auth } from "$lib/auth/store.svelte"; import { listLogs, type LogEntry, type LogsFilter } from "$lib/api/logs"; import * as Sidebar from "$lib/components/ui/sidebar/index.js"; - import { Button } from "$lib/components/ui/button/index.js"; + import * as Sheet from "$lib/components/ui/sheet/index.js"; import { Input } from "$lib/components/ui/input/index.js"; const SERVICES = [ @@ -17,14 +17,17 @@ const LEVELS = ["DEBUG", "INFO", "WARN", "ERROR"]; const CLASSIFICATIONS = ["security", "performance", "audit", "system", "network", "storage"]; const REFRESH_INTERVAL_MS = 7000; - const RECENT_LIST_LIMIT = 200; + const PAGE_LIMIT = 100; let entries = $state([]); + let total = $state(0); + let hasMore = $state(false); + let offset = $state(0); let loading = $state(true); let error = $state(null); let refreshTimer: ReturnType | null = null; - let selectedServices = $state>(new Set()); + let selectedService = $state(""); let selectedLevel = $state(""); let selectedClassification = $state(""); let searchText = $state(""); @@ -34,8 +37,12 @@ let workloadId = $state(""); let organizationId = $state(""); + let selectedEntry = $state(null); + let detailOpen = $state(false); + function currentFilter(): LogsFilter { return { + service: selectedService || undefined, level: selectedLevel || undefined, classification: selectedClassification || undefined, q: searchText || undefined, @@ -44,7 +51,8 @@ agent_id: agentId || undefined, workload_id: workloadId || undefined, organization_id: organizationId || undefined, - limit: RECENT_LIST_LIMIT, + limit: PAGE_LIMIT, + offset, }; } @@ -53,6 +61,8 @@ try { const response = await listLogs(auth.token, currentFilter()); entries = response.entries; + total = response.total; + hasMore = response.has_more; error = null; } catch (e) { error = e instanceof Error ? e.message : "Failed to load logs"; @@ -61,24 +71,6 @@ } } - function toggleService(service: string) { - const next = new Set(selectedServices); - if (next.has(service)) { - next.delete(service); - } else { - next.add(service); - } - selectedServices = next; - } - - function visibleServices(): string[] { - return selectedServices.size === 0 ? SERVICES : SERVICES.filter((s) => selectedServices.has(s)); - } - - function entriesForService(service: string): LogEntry[] { - return entries.filter((entry) => entry.service === service); - } - function levelClass(level: string): string { switch (level) { case "ERROR": return "text-red-500"; @@ -88,8 +80,38 @@ } } - function formatTime(timestamp: string): string { - return timestamp.slice(11, 19); + function severityBars(level: string): number { + switch (level) { + case "ERROR": return 3; + case "WARN": return 2; + default: return 1; + } + } + + function severityBarClass(level: string): string { + switch (level) { + case "ERROR": return "bg-red-500"; + case "WARN": return "bg-yellow-500"; + default: return "bg-green-500"; + } + } + + function formatDateTime(timestamp: string): string { + return new Date(timestamp).toLocaleString(); + } + + function openDetail(entry: LogEntry) { + selectedEntry = entry; + detailOpen = true; + } + + function nextPage() { + if (!hasMore) return; + offset += PAGE_LIMIT; + } + + function prevPage() { + offset = Math.max(0, offset - PAGE_LIMIT); } onMount(() => { @@ -102,6 +124,7 @@ }); $effect(() => { + selectedService; selectedLevel; selectedClassification; searchText; @@ -110,9 +133,22 @@ agentId; workloadId; organizationId; - selectedServices; + offset; load(); }); + + $effect(() => { + selectedService; + selectedLevel; + selectedClassification; + searchText; + fromTime; + toTime; + agentId; + workloadId; + organizationId; + offset = 0; + });
@@ -122,24 +158,18 @@
-
@@ -38,6 +47,8 @@

No general settings configured.

{:else if activeTab === "update"} + {:else if activeTab === "logs"} + {/if}
diff --git a/app/src/routes/(app)/logs/+page.svelte b/app/src/routes/(app)/logs/+page.svelte index 0c65b78d..c6c9b56e 100644 --- a/app/src/routes/(app)/logs/+page.svelte +++ b/app/src/routes/(app)/logs/+page.svelte @@ -5,6 +5,8 @@ import * as Sidebar from "$lib/components/ui/sidebar/index.js"; import * as Sheet from "$lib/components/ui/sheet/index.js"; import { Input } from "$lib/components/ui/input/index.js"; + import { Button } from "$lib/components/ui/button/index.js"; + import RefreshCwIcon from "@lucide/svelte/icons/refresh-cw"; const SERVICES = [ "api-gateway", @@ -24,6 +26,7 @@ let hasMore = $state(false); let offset = $state(0); let loading = $state(true); + let refreshing = $state(false); let error = $state(null); let refreshTimer: ReturnType | null = null; @@ -68,9 +71,15 @@ error = e instanceof Error ? e.message : "Failed to load logs"; } finally { loading = false; + refreshing = false; } } + function refreshNow() { + refreshing = true; + load(); + } + function levelClass(level: string): string { switch (level) { case "ERROR": return "text-red-500"; @@ -155,6 +164,16 @@ / Logs +
diff --git a/control-plane/api-gateway/src/auth/mod.rs b/control-plane/api-gateway/src/auth/mod.rs index 95cde4b9..29a8be56 100644 --- a/control-plane/api-gateway/src/auth/mod.rs +++ b/control-plane/api-gateway/src/auth/mod.rs @@ -2,4 +2,5 @@ pub mod agent; pub mod crypto; pub mod jwt; pub mod middleware; +pub mod rate_limit; pub mod rbac; diff --git a/control-plane/api-gateway/src/auth/rate_limit.rs b/control-plane/api-gateway/src/auth/rate_limit.rs new file mode 100644 index 00000000..3edab764 --- /dev/null +++ b/control-plane/api-gateway/src/auth/rate_limit.rs @@ -0,0 +1,45 @@ +use axum::http::Request; +use tower_governor::key_extractor::{KeyExtractor, PeerIpKeyExtractor}; +use tower_governor::GovernorError; +use uuid::Uuid; + +use super::jwt::verify_jwt; + +#[derive(Debug, Clone, Hash, PartialEq, Eq)] +pub enum RateLimitKey { + User(Uuid), + Ip(std::net::IpAddr), +} + +#[derive(Clone)] +pub struct JwtOrIpKeyExtractor { + fallback: PeerIpKeyExtractor, +} + +impl JwtOrIpKeyExtractor { + pub fn new() -> Self { + Self { + fallback: PeerIpKeyExtractor, + } + } +} + +impl KeyExtractor for JwtOrIpKeyExtractor { + type Key = RateLimitKey; + + fn extract(&self, req: &Request) -> Result { + if let Some(user_id) = extract_user_id(req) { + return Ok(RateLimitKey::User(user_id)); + } + + self.fallback.extract(req).map(RateLimitKey::Ip) + } +} + +fn extract_user_id(req: &Request) -> Option { + let header = req.headers().get(axum::http::header::AUTHORIZATION)?; + let value = header.to_str().ok()?; + let token = value.strip_prefix("Bearer ")?; + let claims = verify_jwt(token).ok()?; + Some(claims.claims.user_id) +} diff --git a/control-plane/api-gateway/src/routes/mod.rs b/control-plane/api-gateway/src/routes/mod.rs index 27348e74..e9229fd5 100644 --- a/control-plane/api-gateway/src/routes/mod.rs +++ b/control-plane/api-gateway/src/routes/mod.rs @@ -1,4 +1,5 @@ use super::AppState; +use crate::auth::rate_limit::JwtOrIpKeyExtractor; use crate::metrics; use crate::utils::router_ext::RouterExt; use axum::body::Body; @@ -8,7 +9,9 @@ use axum::http::{HeaderValue, Request, Response}; use axum::routing::get; use axum::Router; use std::sync::Arc; -use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer}; +use tower_governor::{ + governor::GovernorConfigBuilder, key_extractor::PeerIpKeyExtractor, GovernorLayer, +}; use tower_http::cors::CorsLayer; use tower_http::services::{ServeDir, ServeFile}; use tower_http::trace::TraceLayer; @@ -35,21 +38,41 @@ pub fn create_router() -> Router { let rate_limit_per_second: u64 = std::env::var("RATE_LIMIT_PER_SECOND") .ok() .and_then(|v| v.parse().ok()) - .unwrap_or(100); + .unwrap_or(500); let burst_size: u32 = std::env::var("RATE_LIMIT_BURST") .ok() .and_then(|v| v.parse().ok()) - .unwrap_or(200); + .unwrap_or(1000); let governor_config = Arc::new( GovernorConfigBuilder::default() + .key_extractor(JwtOrIpKeyExtractor::new()) .per_second(rate_limit_per_second) .burst_size(burst_size) .finish() .expect("invalid rate limit configuration"), ); + let login_rate_limit_per_second: u64 = std::env::var("LOGIN_RATE_LIMIT_PER_SECOND") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(1); + + let login_burst_size: u32 = std::env::var("LOGIN_RATE_LIMIT_BURST") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(10); + + let login_governor_config = Arc::new( + GovernorConfigBuilder::default() + .key_extractor(PeerIpKeyExtractor) + .per_second(login_rate_limit_per_second) + .burst_size(login_burst_size) + .finish() + .expect("invalid login rate limit configuration"), + ); + let frontend_url = std::env::var("FRONTEND_URL").unwrap_or_else(|_| "http://localhost:3000".to_string()); @@ -101,8 +124,13 @@ pub fn create_router() -> Router { .merge(settings::settings_routes()) .layer(GovernorLayer::new(governor_config)); + let login_rate_limited_router = Router::new() + .merge(users::public_users_routes()) + .layer(GovernorLayer::new(login_governor_config)); + let api_router = Router::new() .merge(rate_limited_router) + .merge(login_rate_limited_router) .merge(update::routes()) .merge(releases::routes()); diff --git a/control-plane/api-gateway/src/routes/users.rs b/control-plane/api-gateway/src/routes/users.rs index d65bea6a..6f86bce6 100644 --- a/control-plane/api-gateway/src/routes/users.rs +++ b/control-plane/api-gateway/src/routes/users.rs @@ -65,9 +65,9 @@ pub struct UserProfileResponse { pub force_password_change: bool, } -// Define the routes for the users module +// Define the authenticated routes for the users module pub fn users_routes() -> Router { - let auth_routes = Router::new() + Router::new() .route("/profile", get(get_user_profile)) .route("/validate-session", get(validate_session)) .route("/logout", post(logout_user)) @@ -75,13 +75,15 @@ pub fn users_routes() -> Router { .route("/2fa/enable", post(enable_2fa)) .route("/2fa/disable", post(disable_2fa)) .route("/change-password", post(change_password)) - .route("/change-email", post(change_email)); + .route("/change-email", post(change_email)) +} +// Define the public, unauthenticated routes for the users module +pub fn public_users_routes() -> Router { Router::new() .route("/register", post(register_user)) .route("/login", post(login_user)) .route("/public-key", get(get_public_key)) - .merge(auth_routes) } /// Register a new user