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/settings/logs-settings.svelte b/app/src/lib/components/settings/logs-settings.svelte new file mode 100644 index 00000000..267a8908 --- /dev/null +++ b/app/src/lib/components/settings/logs-settings.svelte @@ -0,0 +1,98 @@ + + +{#if error} +
+ {error} +
+{/if} + +
+ + + Log retention + + Logs older than this many days are deleted automatically every hour. + + + + {#if loading} +
+ {:else} +
+
+ + +
+ + {#if saved && !saving} + Saved + {/if} +
+ {/if} + + +
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)/admin/settings/+page.svelte b/app/src/routes/(app)/admin/settings/+page.svelte index 88017e9b..7fa275f4 100644 --- a/app/src/routes/(app)/admin/settings/+page.svelte +++ b/app/src/routes/(app)/admin/settings/+page.svelte @@ -1,8 +1,9 @@ @@ -31,6 +32,14 @@ > Update +
@@ -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 new file mode 100644 index 00000000..c6c9b56e --- /dev/null +++ b/app/src/routes/(app)/logs/+page.svelte @@ -0,0 +1,360 @@ + + +
+ + / + Logs + +
+ +
+ + +
+ {#if error} +

{error}

+ {/if} + +
+ + + + + + + + + + + + + {#if loading} + + {:else if entries.length === 0} + + {:else} + {#each entries as entry (entry.id)} + openDetail(entry)} + > + + + + + + + + {/each} + {/if} + +
SeverityServiceLevelClassificationMessageDate / Time
Loading...
No log entries.
+
+ {#each [1, 2, 3] as bar (bar)} + + {/each} +
+ 		{entry.service}{entry.level}{entry.classification}{entry.message}{formatDateTime(entry.created_at)}
+
+ +
+ {entries.length === 0 ? 0 : offset + 1}-{offset + entries.length} of {total} logs +
+ + +
+
+
+
+ + + + {#if selectedEntry} + + {formatDateTime(selectedEntry.created_at)} + +
+
+ Service + {selectedEntry.service} +
+
+ Level + {selectedEntry.level} +
+
+ Classification + {selectedEntry.classification} +
+ {#if selectedEntry.agent_id} +
+ Agent ID + {selectedEntry.agent_id} +
+ {/if} + {#if selectedEntry.workload_id} +
+ Workload ID + {selectedEntry.workload_id} +
+ {/if} + {#if selectedEntry.organization_id} +
+ Organization ID + {selectedEntry.organization_id} +
+ {/if} +
+ Message +

{selectedEntry.message}

+
+
+ {/if} + + 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/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..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; @@ -16,11 +19,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; @@ -33,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()); @@ -95,10 +120,17 @@ 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 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()); @@ -123,14 +155,33 @@ pub fn create_router() -> Router { ) }) .on_request(|_request: &Request, _span: &Span| { - tracing::info!("started processing request") + tracing::info!(target: "csfx::http_access", "started processing request") }) .on_response( - |_response: &Response, latency: std::time::Duration, _span: &Span| { - tracing::info!( - latency_ms = latency.as_millis(), - "finished processing request" - ) + |response: &Response, latency: std::time::Duration, _span: &Span| { + let status = response.status(); + if status.is_server_error() { + tracing::error!( + target: "csfx::http_access", + status = status.as_u16(), + latency_ms = latency.as_millis(), + "request failed" + ); + } else if status.is_client_error() { + tracing::warn!( + target: "csfx::http_access", + status = status.as_u16(), + latency_ms = latency.as_millis(), + "request rejected" + ); + } else { + tracing::info!( + target: "csfx::http_access", + status = status.as_u16(), + latency_ms = latency.as_millis(), + "finished processing request" + ); + } }, ), ) 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/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 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..d091b1c0 --- /dev/null +++ b/control-plane/shared/shared/src/db_log_layer.rs @@ -0,0 +1,228 @@ +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) + } +} + +const NOISY_ACCESS_TARGET: &str = "csfx::http_access"; +const NOISY_TARGET_PREFIXES: [&str; 2] = ["sea_orm", "sqlx"]; +const NOISY_MESSAGE_SUBSTRINGS: [&str; 1] = ["heartbeat processed"]; + +fn is_routine_access_event(event: &Event<'_>) -> bool { + event.metadata().target() == NOISY_ACCESS_TARGET + && *event.metadata().level() == Level::INFO +} + +fn is_sql_trace_event(event: &Event<'_>) -> bool { + let target = event.metadata().target(); + NOISY_TARGET_PREFIXES + .iter() + .any(|prefix| target.starts_with(prefix)) +} + +fn is_performance_event(event: &Event<'_>) -> bool { + *event.metadata().level() == Level::INFO + && LogClassification::from_target(event.metadata().target()) == LogClassification::Performance +} + +fn is_noisy_message_event(visitor: &EventVisitor) -> bool { + NOISY_MESSAGE_SUBSTRINGS + .iter() + .any(|needle| visitor.message.contains(needle)) +} + +impl Layer for DbLogLayer { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + if *event.metadata().level() == Level::TRACE { + return; + } + + if is_routine_access_event(event) || is_sql_trace_event(event) || is_performance_event(event) { + return; + } + + let mut visitor = EventVisitor::default(); + event.record(&mut visitor); + + if is_noisy_message_event(&visitor) { + return; + } + + 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());