From 3de5fabff2873017c854f11abe8dfa73a4fa9e6e Mon Sep 17 00:00:00 2001 From: CodeMaster4711 Date: Thu, 2 Jul 2026 22:24:52 +0200 Subject: [PATCH 01/16] feat: add container status detail, log streaming, and exec shell for workloads --- Cargo.lock | 176 +++++++---- Cargo.toml | 1 + agent/Cargo.toml | 2 + agent/src/client.rs | 11 +- agent/src/docker.rs | 85 +++++- agent/src/main.rs | 95 +++++- agent/src/server.rs | 276 +++++++++++++++++ app/package-lock.json | 17 ++ app/package.json | 2 + app/src/lib/api/resource-groups.ts | 32 ++ .../(app)/resource-groups/[id]/+page.svelte | 192 +++++++++++- control-plane/api-gateway/Cargo.toml | 6 +- control-plane/api-gateway/src/auth/jwt.rs | 53 ++++ .../api-gateway/src/routes/agent_proxy.rs | 288 ++++++++++++++++++ .../api-gateway/src/routes/agents.rs | 1 + control-plane/api-gateway/src/routes/mod.rs | 2 + control-plane/api-gateway/src/self_monitor.rs | 1 + control-plane/registry/Cargo.toml | 1 + control-plane/registry/src/db/agents.rs | 5 + control-plane/registry/src/handlers/agent.rs | 7 +- control-plane/registry/src/handlers/pki.rs | 60 +++- control-plane/registry/src/main.rs | 7 +- control-plane/registry/src/models/agent.rs | 1 + control-plane/registry/src/server.rs | 1 + control-plane/registry/src/services/pki.rs | 27 +- .../registry/src/services/registry.rs | 2 + control-plane/scheduler/src/db/workloads.rs | 4 +- .../scheduler/src/models/workload.rs | 9 + .../shared/entity/src/entities/agents.rs | 1 + control-plane/shared/migration/src/lib.rs | 2 + ...m20260702_000000_add_agent_wg_tunnel_ip.rs | 31 ++ 31 files changed, 1312 insertions(+), 86 deletions(-) create mode 100644 agent/src/server.rs create mode 100644 control-plane/api-gateway/src/routes/agent_proxy.rs create mode 100644 control-plane/shared/migration/src/m20260702_000000_add_agent_wg_tunnel_ip.rs diff --git a/Cargo.lock b/Cargo.lock index be7c5be4..795c5026 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -154,9 +154,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "api-gateway" @@ -174,6 +174,7 @@ dependencies = [ "dotenvy", "entity", "etcd-client", + "futures-util", "http", "hyper", "hyper-util", @@ -186,7 +187,7 @@ dependencies = [ "opentelemetry_sdk", "prometheus", "qrcode", - "rand 0.10.1", + "rand 0.10.2", "rcgen", "reqwest", "ring", @@ -201,6 +202,7 @@ dependencies = [ "sysinfo", "thiserror 2.0.18", "tokio", + "tokio-tungstenite", "totp-rs", "tower-http 0.7.0", "tower_governor", @@ -223,9 +225,9 @@ dependencies = [ [[package]] name = "arc-swap" -version = "1.9.1" +version = "1.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207" +checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b" dependencies = [ "rustversion", ] @@ -243,9 +245,9 @@ dependencies = [ [[package]] name = "arrayvec" -version = "0.7.6" +version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" +checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" [[package]] name = "as-slice" @@ -548,6 +550,7 @@ checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", "axum-macros", + "base64", "bytes", "form_urlencoded", "futures-util", @@ -566,8 +569,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1 0.10.6", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -712,9 +717,9 @@ dependencies = [ [[package]] name = "bitvec" -version = "1.0.1" +version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bc2832c24239b0141d5674bb9174f9d68a8b5b3f2753311927c172ca46f7e9c" +checksum = "ddcec3d12c579d40898fe0a9a358a803c23e9c52ca3c425707f81c9436211837" dependencies = [ "funty", "radium", @@ -808,9 +813,9 @@ dependencies = [ [[package]] name = "borsh" -version = "1.6.1" +version = "1.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfd1e3f8955a5d7de9fab72fc8373fade9fb8a703968cb200ae3dc6cf08e185a" +checksum = "2f3f6da4992df95bbcd9af42a6c7dcb994498fc9048230405f3b36ff7cd3f145" dependencies = [ "borsh-derive", "bytes", @@ -819,9 +824,9 @@ dependencies = [ [[package]] name = "borsh-derive" -version = "1.6.1" +version = "1.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bfcfdc083699101d5a7965e49925975f2f55060f94f9a05e7187be95d530ca59" +checksum = "3ae8fb4fb5740e4b2c4884ff95f5f32f5e8479db1e8fd8eb49ddbe09eb09bb7c" dependencies = [ "once_cell", "proc-macro-crate", @@ -884,15 +889,15 @@ checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" [[package]] name = "bytes" -version = "1.11.1" +version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593" [[package]] name = "cc" -version = "1.2.64" +version = "1.2.65" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dad887fd958be91b5098c0248def011f4523ab786cd411be668777e55063501f" +checksum = "e228eec9be7c17ccb640b59b36a5cd805ea2a564a4c5e162c2f659fea30d3b96" dependencies = [ "find-msvc-tools", "jobserver", @@ -920,9 +925,9 @@ checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" [[package]] name = "chacha20" -version = "0.10.0" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1198,6 +1203,7 @@ name = "csfx-agent" version = "0.2.2" dependencies = [ "anyhow", + "axum", "base64", "bollard", "chrono", @@ -1214,6 +1220,7 @@ dependencies = [ "tracing-subscriber", "uuid", "x25519-dalek", + "x509-parser", "zbus", ] @@ -2233,9 +2240,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hybrid-array" -version = "0.4.12" +version = "0.4.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9155a582abd142abc056962c29e3ce5ff2ad5469f4246b537ed42c5deba857da" +checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" dependencies = [ "typenum", ] @@ -2651,9 +2658,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.102" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "03d04c30968dffe80775bd4d7fb676131cd04a1fb46d2686dbffbaec2d9dfd31" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" dependencies = [ "cfg-if", "futures-util", @@ -2732,14 +2739,14 @@ checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" [[package]] name = "libredox" -version = "0.1.17" +version = "0.1.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f02ab6bace2054fb888a3c16f990117b579d14a3088e472d63c6011fa185c9d3" +checksum = "c943259e342f1e06ff2da7a83eabdfe7f92ce10262688dbf1895ff0b3e6e4652" dependencies = [ "bitflags", "libc", "plain", - "redox_syscall 0.8.1", + "redox_syscall 0.9.0", ] [[package]] @@ -2781,9 +2788,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.32" +version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" dependencies = [ "value-bag", ] @@ -3171,9 +3178,9 @@ dependencies = [ [[package]] name = "opentelemetry-semantic-conventions" -version = "0.32.0" +version = "0.32.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ca2f98a0437b427b4b08f19f1caa3c44db885a202bc12cfea13d6c702243d68" +checksum = "c913ac17a6c451661ee255f4625d143e51647ae78ebd969b75e41c4442f4fe47" [[package]] name = "opentelemetry_sdk" @@ -3798,9 +3805,9 @@ checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3" [[package]] name = "quote" -version = "1.0.45" +version = "1.0.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" dependencies = [ "proc-macro2", ] @@ -3846,9 +3853,9 @@ dependencies = [ [[package]] name = "rand" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ "chacha20", "getrandom 0.4.3", @@ -4003,9 +4010,9 @@ dependencies = [ [[package]] name = "redox_syscall" -version = "0.8.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b44b894f2a6e36457d665d1e08c3866add6ed5e70050c1b4ba8a8ddedb02ce7" +checksum = "c5102a6aaa05aa011a238e178e6bca86d2cb56fc9f586d37cb80f5bca6e07759" dependencies = [ "bitflags", ] @@ -4045,6 +4052,7 @@ version = "0.2.2" dependencies = [ "anyhow", "axum", + "base64", "chrono", "dotenvy", "entity", @@ -4090,6 +4098,7 @@ dependencies = [ "base64", "bytes", "futures-core", + "futures-util", "http", "http-body", "http-body-util", @@ -4108,12 +4117,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", + "tokio-util", "tower", "tower-http 0.6.11", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", "webpki-roots 1.0.8", ] @@ -4295,9 +4306,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.40" +version = "0.23.41" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" dependencies = [ "log", "once_cell", @@ -4331,9 +4342,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.14.1" +version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" +checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" dependencies = [ "zeroize", ] @@ -5331,9 +5342,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.49" +version = "0.3.53" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "711a53c2d47bbd818258c498c8dbfe186a2526c631495cfe7e078567f86b8469" +checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50" dependencies = [ "deranged", "num-conv", @@ -5351,9 +5362,9 @@ checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" [[package]] name = "time-macros" -version = "0.2.29" +version = "0.2.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71c652a3727a9cbb9a02f707f530b618ce00d0ccd762009c8c23bd191df3c17d" +checksum = "c431b87111666e491a90baa837f914fb45cd5dc3c268591b0220ff5057f2085f" dependencies = [ "num-conv", "time-core", @@ -5434,6 +5445,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -5560,9 +5583,9 @@ dependencies = [ [[package]] name = "totp-rs" -version = "5.7.1" +version = "5.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2b36a9dd327e9f401320a2cb4572cc76ff43742bcfc3291f871691050f140ba" +checksum = "50e69a15e21b2ff22c415446983978bded3244195f17d59cb113551c1e806f91" dependencies = [ "base32", "constant_time_eq", @@ -5750,6 +5773,22 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.4", + "sha1 0.10.6", + "thiserror 2.0.18", +] + [[package]] name = "typenum" version = "1.20.1" @@ -5897,9 +5936,9 @@ dependencies = [ [[package]] name = "uuid" -version = "1.23.3" +version = "1.23.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "144d6b123cef80b301b8f72a9e2ca4370ddec21950d0a103dd22c437006d2db7" +checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -6015,9 +6054,9 @@ checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b" [[package]] name = "wasm-bindgen" -version = "0.2.125" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ddb3f79143bced6de84270411622a2699cee572fc0875aeaf1e7867cf9fca1a" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" dependencies = [ "cfg-if", "once_cell", @@ -6029,9 +6068,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.75" +version = "0.4.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "503b14d284f2c8dac03b819967e155ea753f573586193b2b2c95990cb5d69280" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" dependencies = [ "js-sys", "wasm-bindgen", @@ -6039,9 +6078,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.125" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e21a184b13fb19e157296e2c46056aec9092264fab83e4ba59e68c61b323c3d" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -6049,9 +6088,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.125" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fecefd9c35bd935a20fc3fc344b5f29138961e4f47fb03297d88f2587afb5ebd" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" dependencies = [ "bumpalo", "proc-macro2", @@ -6062,18 +6101,31 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.125" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23939e44bb9a5d7576fa2b563dc2e136628f1224e88a8deed09e04858b77871f" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "web-sys" -version = "0.3.102" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6430a72df5eb332242960fe84b3002a241163998241eb596d4f739b9757061d" +checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141" dependencies = [ "js-sys", "wasm-bindgen", @@ -6769,9 +6821,9 @@ dependencies = [ [[package]] name = "zlib-rs" -version = "0.6.3" +version = "0.6.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3be3d40e40a133f9c916ee3f9f4fa2d9d63435b5fbe1bfc6d9dae0aa0ada1513" +checksum = "5431d5661c32445236631278f27946e444ddafe4684cac70b185272d4f9c52d5" [[package]] name = "zmij" diff --git a/Cargo.toml b/Cargo.toml index a96b7b0e..27858853 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,6 +38,7 @@ tower-http = { version = "0.7.0", features = ["cors", "trace", "fs"] } hyper = "1.5" hyper-util = { version = "0.1", features = ["client", "client-legacy", "http1", "http2", "tokio"] } http = "1.1" +tokio-tungstenite = "0.29" # Serialization serde = { version = "1.0", features = ["derive"] } diff --git a/agent/Cargo.toml b/agent/Cargo.toml index f4321e38..85ecbafc 100644 --- a/agent/Cargo.toml +++ b/agent/Cargo.toml @@ -27,3 +27,5 @@ futures-util = { workspace = true } zbus = { workspace = true } x25519-dalek = { workspace = true } base64 = { workspace = true } +axum = { workspace = true, features = ["ws"] } +x509-parser = { workspace = true } diff --git a/agent/src/client.rs b/agent/src/client.rs index d9b79271..ad5c67b1 100644 --- a/agent/src/client.rs +++ b/agent/src/client.rs @@ -52,6 +52,7 @@ struct HeartbeatRequest { uptime_seconds: Option, wg_public_key: Option, wg_endpoint: Option, + wg_tunnel_ip: Option, } #[derive(Debug, Serialize, Deserialize, Clone)] @@ -94,6 +95,7 @@ pub struct ApiClient { cert_pem: Option, wg_public_key: String, wg_endpoint: Option, + wg_tunnel_ip: Option, } pub fn generate_wg_keypair() -> (String, String) { @@ -109,7 +111,12 @@ pub fn generate_wg_keypair() -> (String, String) { } impl ApiClient { - pub fn new(gateway_url: String, wg_public_key: String, wg_endpoint: Option) -> Result { + pub fn new( + gateway_url: String, + wg_public_key: String, + wg_endpoint: Option, + wg_tunnel_ip: Option, + ) -> Result { let client = Client::builder() .timeout(std::time::Duration::from_secs(30)) .danger_accept_invalid_certs(true) @@ -122,6 +129,7 @@ impl ApiClient { cert_pem: None, wg_public_key, wg_endpoint, + wg_tunnel_ip, }) } @@ -243,6 +251,7 @@ impl ApiClient { uptime_seconds, wg_public_key: Some(self.wg_public_key.clone()), wg_endpoint: self.wg_endpoint.clone(), + wg_tunnel_ip: self.wg_tunnel_ip.clone(), }); if let Some(ref cert_pem) = self.cert_pem { diff --git a/agent/src/docker.rs b/agent/src/docker.rs index 5d883000..8057960c 100644 --- a/agent/src/docker.rs +++ b/agent/src/docker.rs @@ -1,10 +1,13 @@ use anyhow::{Context, Result}; -use bollard::models::{ContainerCreateBody, HostConfig}; +use axum::body::Bytes; +use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults}; +use bollard::models::{ContainerCreateBody, ContainerStateStatusEnum, HostConfig}; use bollard::query_parameters::{ - CreateContainerOptionsBuilder, CreateImageOptionsBuilder, StartContainerOptionsBuilder, + CreateContainerOptionsBuilder, CreateImageOptionsBuilder, InspectContainerOptionsBuilder, + LogsOptionsBuilder, StartContainerOptionsBuilder, }; use bollard::Docker; -use futures_util::StreamExt; +use futures_util::{Stream, StreamExt}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::path::Path; @@ -164,6 +167,82 @@ impl DockerManager { Ok(container.id) } + pub async fn inspect_status(&self, container_id: &str) -> Result { + let options = InspectContainerOptionsBuilder::default().build(); + + let inspect = self + .docker + .inspect_container(container_id, Some(options)) + .await + .context("Failed to inspect container")?; + + let state = inspect.state.unwrap_or_default(); + let status = state.status.unwrap_or(ContainerStateStatusEnum::EMPTY); + let exit_code = state.exit_code.unwrap_or(0); + + Ok(match status { + ContainerStateStatusEnum::CREATED => "creating".to_string(), + ContainerStateStatusEnum::RUNNING => "running".to_string(), + ContainerStateStatusEnum::RESTARTING | ContainerStateStatusEnum::PAUSED => { + "running".to_string() + } + ContainerStateStatusEnum::EXITED if exit_code == 0 => "stopped".to_string(), + ContainerStateStatusEnum::EXITED | ContainerStateStatusEnum::DEAD => { + "failed".to_string() + } + ContainerStateStatusEnum::REMOVING | ContainerStateStatusEnum::STOPPING => { + "stopped".to_string() + } + ContainerStateStatusEnum::EMPTY => "failed".to_string(), + }) + } + + pub fn logs( + &self, + container_id: &str, + ) -> impl Stream> + Send + 'static { + let options = LogsOptionsBuilder::default() + .stdout(true) + .stderr(true) + .follow(true) + .tail("200") + .build(); + + self.docker + .logs(container_id, Some(options)) + .map(|item| match item { + Ok(log_output) => Ok(Bytes::from(log_output.into_bytes())), + Err(e) => Err(std::io::Error::other(e.to_string())), + }) + } + + pub async fn exec(&self, container_id: &str) -> Result { + let create_options = CreateExecOptions { + attach_stdin: Some(true), + attach_stdout: Some(true), + attach_stderr: Some(true), + tty: Some(true), + cmd: Some(vec!["/bin/sh".to_string()]), + ..Default::default() + }; + + let created = self + .docker + .create_exec(container_id, create_options) + .await + .context("Failed to create exec session")?; + + let start_options = StartExecOptions { + tty: true, + ..Default::default() + }; + + self.docker + .start_exec(&created.id, Some(start_options)) + .await + .context("Failed to start exec session") + } + pub async fn stop_container(&self, container_id: &str) -> Result<()> { self.docker .stop_container(container_id, None) diff --git a/agent/src/main.rs b/agent/src/main.rs index 003bfea1..3e59e87b 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -3,6 +3,7 @@ mod config; mod docker; mod pki; mod rbd; +mod server; mod ssh_keys; mod system; mod update_watch; @@ -52,9 +53,10 @@ async fn main() -> Result<()> { .unwrap_or(60); let wg_endpoint = std::env::var("CSFX_WG_ENDPOINT").ok(); + let wg_tunnel_ip = std::env::var("CSFX_WG_TUNNEL_IP").ok(); let (_wg_private_key, wg_public_key) = client::generate_wg_keypair(); - let api_client = client::ApiClient::new(gateway_url.clone(), wg_public_key, wg_endpoint) + let api_client = client::ApiClient::new(gateway_url.clone(), wg_public_key, wg_endpoint, wg_tunnel_ip) .context("Failed to initialize API client")?; let agent_pki = pki::AgentPki::load_or_generate().context("Failed to initialize PKI")?; @@ -97,8 +99,28 @@ async fn main() -> Result<()> { let running_containers: Arc>> = Arc::new(Mutex::new(HashMap::new())); + let workload_phases: Arc>> = Arc::new(Mutex::new(HashMap::new())); + let mounted_volumes: Arc>> = Arc::new(Mutex::new(HashMap::new())); + if let Some(port) = std::env::var("CSFX_AGENT_PORT") + .ok() + .and_then(|v| v.parse::().ok()) + { + let server_state = server::ServerState { + docker: docker_manager.clone(), + running_containers: running_containers.clone(), + agent_id, + }; + tokio::spawn(async move { + if let Err(e) = server::run(server_state, port).await { + error!(error = %e, "agent inbound server stopped"); + } + }); + } else { + info!("CSFX_AGENT_PORT not set, agent inbound server disabled"); + } + run_heartbeat_loop( &api_client, agent_id, @@ -106,6 +128,7 @@ async fn main() -> Result<()> { heartbeat_interval_secs, docker_manager, running_containers, + workload_phases, mounted_volumes, ) .await; @@ -183,6 +206,7 @@ async fn run_heartbeat_loop( interval_secs: u64, docker: Arc>>, running_containers: Arc>>, + workload_phases: Arc>>, mounted_volumes: Arc>>, ) { let mut interval = tokio::time::interval(Duration::from_secs(interval_secs)); @@ -193,9 +217,9 @@ async fn run_heartbeat_loop( tokio::select! { _ = interval.tick() => { process_volumes(client, agent_id, api_key, &mounted_volumes).await; - process_workloads(client, api_key, &docker, &running_containers, &mounted_volumes).await; + process_workloads(client, api_key, &docker, &running_containers, &workload_phases, &mounted_volumes).await; - let statuses = build_container_statuses(&running_containers).await; + let statuses = build_container_statuses(&docker, &running_containers, &workload_phases).await; let metrics = system::collect_metrics(); match client.heartbeat(agent_id, api_key, Some(statuses), Some(metrics)).await { @@ -315,6 +339,7 @@ async fn process_workloads( api_key: &str, docker: &Arc>>, running_containers: &Arc>>, + workload_phases: &Arc>>, mounted_volumes: &Arc>>, ) { let workloads = match client.fetch_assigned_workloads(api_key).await { @@ -353,8 +378,14 @@ async fn process_workloads( info!(workload_id = %workload.id, image = %workload.image, "Starting workload"); + workload_phases + .lock() + .await + .insert(workload.id.clone(), "pulling".to_string()); + if let Err(e) = docker.pull_image(&workload.image).await { warn!(workload_id = %workload.id, error = %e, "Failed to pull image"); + workload_phases.lock().await.remove(&workload.id); continue; } @@ -368,6 +399,11 @@ async fn process_workloads( } } + workload_phases + .lock() + .await + .insert(workload.id.clone(), "creating".to_string()); + let spec = docker::WorkloadSpec { workload_id: workload.id.clone(), name: workload.name.clone(), @@ -387,6 +423,10 @@ async fn process_workloads( match docker.start_container(&spec).await { Ok(container_id) => { + workload_phases + .lock() + .await + .insert(workload.id.clone(), "starting".to_string()); running_containers .lock() .await @@ -398,6 +438,7 @@ async fn process_workloads( ); } Err(e) => { + workload_phases.lock().await.remove(&workload.id); warn!(workload_id = %workload.id, error = %e, "Failed to start container"); } } @@ -405,16 +446,48 @@ async fn process_workloads( } async fn build_container_statuses( + docker: &Arc>>, running_containers: &Arc>>, + workload_phases: &Arc>>, ) -> Vec { - running_containers - .lock() - .await - .iter() - .map(|(workload_id, container_id)| client::ContainerStatus { + let containers = running_containers.lock().await.clone(); + let mut statuses = Vec::with_capacity(containers.len()); + + let docker_guard = docker.lock().await; + for (workload_id, container_id) in containers.iter() { + let status = match docker_guard.as_ref() { + Some(dm) => match dm.inspect_status(container_id).await { + Ok(s) => s, + Err(e) => { + warn!(workload_id = %workload_id, container_id = %container_id, error = %e, "Failed to inspect container"); + "failed".to_string() + } + }, + None => "failed".to_string(), + }; + + if status != "creating" { + workload_phases.lock().await.remove(workload_id); + } + + statuses.push(client::ContainerStatus { workload_id: workload_id.clone(), container_id: container_id.clone(), - status: "running".to_string(), - }) - .collect() + status, + }); + } + drop(docker_guard); + + let phases = workload_phases.lock().await; + for (workload_id, phase) in phases.iter() { + if !containers.contains_key(workload_id) { + statuses.push(client::ContainerStatus { + workload_id: workload_id.clone(), + container_id: String::new(), + status: phase.clone(), + }); + } + } + + statuses } diff --git a/agent/src/server.rs b/agent/src/server.rs new file mode 100644 index 00000000..ad5e55d4 --- /dev/null +++ b/agent/src/server.rs @@ -0,0 +1,276 @@ +use anyhow::{anyhow, Context, Result}; +use axum::{ + extract::ws::{Message, WebSocket, WebSocketUpgrade}, + extract::{ConnectInfo, Path, State}, + http::{HeaderMap, StatusCode}, + routing::get, + Router, +}; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; +use bollard::exec::StartExecResults; +use futures_util::{SinkExt, StreamExt}; +use ring::signature::{UnparsedPublicKey, ECDSA_P256_SHA256_ASN1}; +use std::collections::HashMap; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::Arc; +use tokio::io::AsyncWriteExt; +use tokio::sync::Mutex; +use tracing::{info, warn}; + +use crate::docker::DockerManager; +use crate::pki::AgentPki; + +const TICKET_HEADER: &str = "X-Csfx-Proxy-Ticket"; + +#[derive(Clone)] +pub struct ServerState { + pub docker: Arc>>, + pub running_containers: Arc>>, + pub agent_id: uuid::Uuid, +} + +pub async fn run(state: ServerState, port: u16) -> Result<()> { + let app = Router::new() + .route("/logs/{workload_id}", get(logs_handler)) + .route("/exec/{workload_id}", get(exec_handler)) + .with_state(state); + + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port); + let listener = tokio::net::TcpListener::bind(addr) + .await + .context("Failed to bind agent inbound server")?; + + info!(port = port, "csfx-agent inbound server listening"); + + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .context("Agent inbound server stopped unexpectedly") +} + +fn is_internal_source(addr: &SocketAddr) -> bool { + let ip = addr.ip(); + ip.is_loopback() + || match ip { + IpAddr::V4(v4) => { + let octets = v4.octets(); + octets[0] == 10 + || (octets[0] == 172 && octets[1] >= 16 && octets[1] <= 31) + || (octets[0] == 192 && octets[1] == 168) + } + IpAddr::V6(_) => false, + } +} + +struct TicketClaims { + agent_id: String, + workload_id: String, + expires_at: i64, +} + +fn verify_ticket(headers: &HeaderMap, workload_id: &str, expected_agent_id: &str) -> Result<()> { + let raw = headers + .get(TICKET_HEADER) + .and_then(|v| v.to_str().ok()) + .ok_or_else(|| anyhow!("missing {} header", TICKET_HEADER))?; + + let (payload_b64, signature_b64) = raw + .split_once('.') + .ok_or_else(|| anyhow!("malformed proxy ticket"))?; + + let payload_bytes = URL_SAFE_NO_PAD + .decode(payload_b64) + .context("invalid ticket payload encoding")?; + let signature = URL_SAFE_NO_PAD + .decode(signature_b64) + .context("invalid ticket signature encoding")?; + + let ca_pem = AgentPki::load_ca_pem().context("failed to load trusted CA certificate")?; + let public_key_bytes = extract_ca_public_key(&ca_pem)?; + + let verifier = UnparsedPublicKey::new(&ECDSA_P256_SHA256_ASN1, &public_key_bytes); + verifier + .verify(&payload_bytes, &signature) + .map_err(|_| anyhow!("proxy ticket signature verification failed"))?; + + let claims = parse_claims(&payload_bytes)?; + + if claims.workload_id != workload_id { + return Err(anyhow!("proxy ticket workload_id mismatch")); + } + if claims.agent_id != expected_agent_id { + return Err(anyhow!("proxy ticket agent_id mismatch")); + } + if claims.expires_at < chrono::Utc::now().timestamp() { + return Err(anyhow!("proxy ticket expired")); + } + + Ok(()) +} + +fn parse_claims(payload_bytes: &[u8]) -> Result { + let payload = std::str::from_utf8(payload_bytes).context("proxy ticket payload not utf8")?; + let mut parts = payload.splitn(3, '.'); + + let agent_id = parts + .next() + .ok_or_else(|| anyhow!("proxy ticket missing agent_id"))? + .to_string(); + let workload_id = parts + .next() + .ok_or_else(|| anyhow!("proxy ticket missing workload_id"))? + .to_string(); + let expires_at = parts + .next() + .ok_or_else(|| anyhow!("proxy ticket missing expiry"))? + .parse::() + .context("proxy ticket expiry not a valid timestamp")?; + + Ok(TicketClaims { + agent_id, + workload_id, + expires_at, + }) +} + +fn extract_ca_public_key(ca_pem: &str) -> Result> { + let (_, pem) = + x509_parser::pem::parse_x509_pem(ca_pem.as_bytes()).context("failed to parse CA PEM")?; + let cert = pem.parse_x509().context("failed to parse CA certificate")?; + Ok(cert + .tbs_certificate + .subject_pki + .subject_public_key + .data + .to_vec()) +} + +async fn logs_handler( + State(state): State, + Path(workload_id): Path, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, +) -> Result { + if !is_internal_source(&addr) { + warn!(source = %addr, "rejected agent inbound request from non-internal source"); + return Err((StatusCode::FORBIDDEN, "source not allowed".to_string())); + } + + verify_ticket(&headers, &workload_id, &state.agent_id.to_string()) + .map_err(|e| (StatusCode::UNAUTHORIZED, e.to_string()))?; + + let container_id = state + .running_containers + .lock() + .await + .get(&workload_id) + .cloned() + .ok_or((StatusCode::NOT_FOUND, "workload not running here".to_string()))?; + + let docker_guard = state.docker.lock().await; + let docker = docker_guard + .as_ref() + .ok_or((StatusCode::SERVICE_UNAVAILABLE, "docker unavailable".to_string()))?; + + let stream = docker.logs(&container_id); + Ok(axum::body::Body::from_stream(stream)) +} + +async fn exec_handler( + State(state): State, + Path(workload_id): Path, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + ws: WebSocketUpgrade, +) -> Result { + if !is_internal_source(&addr) { + warn!(source = %addr, "rejected agent inbound request from non-internal source"); + return Err((StatusCode::FORBIDDEN, "source not allowed".to_string())); + } + + verify_ticket(&headers, &workload_id, &state.agent_id.to_string()) + .map_err(|e| (StatusCode::UNAUTHORIZED, e.to_string()))?; + + let container_id = state + .running_containers + .lock() + .await + .get(&workload_id) + .cloned() + .ok_or((StatusCode::NOT_FOUND, "workload not running here".to_string()))?; + + Ok(ws.on_upgrade(move |socket| handle_exec_socket(socket, state, container_id))) +} + +async fn handle_exec_socket(socket: WebSocket, state: ServerState, container_id: String) { + let docker_guard = state.docker.lock().await; + let docker = match docker_guard.as_ref() { + Some(d) => d, + None => { + warn!("docker unavailable for exec session"); + return; + } + }; + + let exec_result = match docker.exec(&container_id).await { + Ok(r) => r, + Err(e) => { + warn!(container_id = %container_id, error = %e, "failed to start exec session"); + return; + } + }; + drop(docker_guard); + + let StartExecResults::Attached { + mut output, + mut input, + } = exec_result + else { + warn!(container_id = %container_id, "exec session detached unexpectedly"); + return; + }; + + let (mut ws_sink, mut ws_stream) = socket.split(); + + let output_task = tokio::spawn(async move { + while let Some(item) = output.next().await { + match item { + Ok(log_output) => { + let bytes = log_output.into_bytes(); + if ws_sink.send(Message::Binary(bytes)).await.is_err() { + break; + } + } + Err(_) => break, + } + } + }); + + let input_task = tokio::spawn(async move { + while let Some(Ok(msg)) = ws_stream.next().await { + match msg { + Message::Binary(data) => { + if input.write_all(&data).await.is_err() { + break; + } + } + Message::Text(text) => { + if input.write_all(text.as_bytes()).await.is_err() { + break; + } + } + Message::Close(_) => break, + _ => {} + } + } + }); + + tokio::select! { + _ = output_task => {} + _ = input_task => {} + } + + info!(container_id = %container_id, "exec session closed"); +} diff --git a/app/package-lock.json b/app/package-lock.json index 9a5881d8..0dba6c85 100644 --- a/app/package-lock.json +++ b/app/package-lock.json @@ -10,6 +10,8 @@ "dependencies": { "@iconify/svelte": "^5.2.1", "@jis3r/icons": "^2.7.0", + "@xterm/addon-fit": "^0.11.0", + "@xterm/xterm": "^6.0.0", "geist": "^1.7.0" }, "devDependencies": { @@ -1546,6 +1548,21 @@ "integrity": "sha512-ScaPdn1dQczgbl0QFTeTOmVHFULt394XJgOQNoyVhZ6r2vLnMLJfBPd53SB52T/3G36VI1/g2MZaX0cwDuXsfw==", "license": "MIT" }, + "node_modules/@xterm/addon-fit": { + "version": "0.11.0", + "resolved": "https://registry.npmjs.org/@xterm/addon-fit/-/addon-fit-0.11.0.tgz", + "integrity": "sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g==", + "license": "MIT" + }, + "node_modules/@xterm/xterm": { + "version": "6.0.0", + "resolved": "https://registry.npmjs.org/@xterm/xterm/-/xterm-6.0.0.tgz", + "integrity": "sha512-TQwDdQGtwwDt+2cgKDLn0IRaSxYu1tSUjgKarSDkUM0ZNiSRXFpjxEsvc/Zgc5kq5omJ+V0a8/kIM2WD3sMOYg==", + "license": "MIT", + "workspaces": [ + "addons/*" + ] + }, "node_modules/acorn": { "version": "8.16.0", "resolved": "https://registry.npmjs.org/acorn/-/acorn-8.16.0.tgz", diff --git a/app/package.json b/app/package.json index ea80f909..330b4695 100644 --- a/app/package.json +++ b/app/package.json @@ -37,6 +37,8 @@ "dependencies": { "@iconify/svelte": "^5.2.1", "@jis3r/icons": "^2.7.0", + "@xterm/addon-fit": "^0.11.0", + "@xterm/xterm": "^6.0.0", "geist": "^1.7.0" } } diff --git a/app/src/lib/api/resource-groups.ts b/app/src/lib/api/resource-groups.ts index 9ef94e8a..87b4dc2e 100644 --- a/app/src/lib/api/resource-groups.ts +++ b/app/src/lib/api/resource-groups.ts @@ -145,6 +145,38 @@ export async function deleteWorkload(token: string, id: string): Promise { if (!res.ok) throw new Error(`Failed to delete workload: ${res.status}`); } +export async function streamWorkloadLogs( + token: string, + id: string, + signal: AbortSignal, +): Promise> { + const res = await fetch(`${API_BASE}/workloads/${id}/logs`, { + headers: { Authorization: `Bearer ${token}` }, + signal, + }); + if (!res.ok || !res.body) { + const err = await res.json().catch(() => ({ error: res.status })); + throw new Error(err.error ?? `Failed to stream logs: ${res.status}`); + } + return res.body; +} + +export async function openWorkloadExecSocket(token: string, id: string): Promise { + const res = await fetch(`${API_BASE}/workloads/${id}/exec/ticket`, { + method: 'POST', + headers: { Authorization: `Bearer ${token}` }, + }); + if (!res.ok) { + const err = await res.json().catch(() => ({ error: res.status })); + throw new Error(err.error ?? `Failed to issue exec ticket: ${res.status}`); + } + const { ticket } = await res.json(); + + const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const url = `${wsProtocol}//${window.location.host}${API_BASE}/workloads/${id}/exec?ticket=${encodeURIComponent(ticket)}`; + return new WebSocket(url); +} + export async function listResourceGroupVolumes(token: string, rgId: string): Promise { const res = await fetch(`${API_BASE}/resource-groups/${rgId}/volumes`, { headers: { Authorization: `Bearer ${token}` }, diff --git a/app/src/routes/(app)/resource-groups/[id]/+page.svelte b/app/src/routes/(app)/resource-groups/[id]/+page.svelte index 51da30d1..cfae95c8 100644 --- a/app/src/routes/(app)/resource-groups/[id]/+page.svelte +++ b/app/src/routes/(app)/resource-groups/[id]/+page.svelte @@ -1,5 +1,5 @@ Date: Sat, 4 Jul 2026 13:25:45 +0200 Subject: [PATCH 04/16] fix: sync missing permissions to existing Admin role on startup --- control-plane/api-gateway/src/init.rs | 40 ++++++++++++++++++--------- 1 file changed, 27 insertions(+), 13 deletions(-) diff --git a/control-plane/api-gateway/src/init.rs b/control-plane/api-gateway/src/init.rs index 4a70a724..dd6a1577 100644 --- a/control-plane/api-gateway/src/init.rs +++ b/control-plane/api-gateway/src/init.rs @@ -220,22 +220,36 @@ pub async fn initialize_database( created_at: ActiveValue::Set(now), }; Role::insert(new_role).exec_without_returning(db).await?; - - // Assign all permissions to Admin role - for perm_id in permission_map.values() { - let role_perm = role_permission::ActiveModel { - role_id: ActiveValue::Set(role_id), - permission_id: ActiveValue::Set(*perm_id), - }; - RolePermission::insert(role_perm) - .exec_without_returning(db) - .await?; - } - - tracing::info!("Admin role created with all permissions"); + tracing::info!("Admin role created"); role_id }; + let admin_existing_perm_ids: std::collections::HashSet = RolePermission::find() + .filter(role_permission::Column::RoleId.eq(admin_role_id)) + .all(db) + .await? + .into_iter() + .map(|rp| rp.permission_id) + .collect(); + + let mut admin_granted = 0; + for perm_id in permission_map.values() { + if admin_existing_perm_ids.contains(perm_id) { + continue; + } + let role_perm = role_permission::ActiveModel { + role_id: ActiveValue::Set(admin_role_id), + permission_id: ActiveValue::Set(*perm_id), + }; + RolePermission::insert(role_perm) + .exec_without_returning(db) + .await?; + admin_granted += 1; + } + if admin_granted > 0 { + tracing::info!(count = admin_granted, "granted missing permissions to Admin role"); + } + // 4b. Create Operator role let operator_role_exists = Role::find() .filter(role::Column::Name.eq("Operator")) From 9ff3735b34d241669772152ae034715a7fabcc27 Mon Sep 17 00:00:00 2001 From: CodeMaster4711 Date: Sat, 4 Jul 2026 13:28:22 +0200 Subject: [PATCH 05/16] fix: wait for auth token before loading data on nodes, resource-groups and logs pages --- app/src/routes/(app)/logs/+page.svelte | 2 +- app/src/routes/(app)/nodes/+page.svelte | 17 ++++++++++++++--- .../routes/(app)/resource-groups/+page.svelte | 10 ++++++++-- 3 files changed, 23 insertions(+), 6 deletions(-) diff --git a/app/src/routes/(app)/logs/+page.svelte b/app/src/routes/(app)/logs/+page.svelte index c6c9b56e..07d86c91 100644 --- a/app/src/routes/(app)/logs/+page.svelte +++ b/app/src/routes/(app)/logs/+page.svelte @@ -124,7 +124,6 @@ } onMount(() => { - load(); refreshTimer = setInterval(load, REFRESH_INTERVAL_MS); }); @@ -133,6 +132,7 @@ }); $effect(() => { + auth.token; selectedService; selectedLevel; selectedClassification; diff --git a/app/src/routes/(app)/nodes/+page.svelte b/app/src/routes/(app)/nodes/+page.svelte index 2f429736..ab6a4794 100644 --- a/app/src/routes/(app)/nodes/+page.svelte +++ b/app/src/routes/(app)/nodes/+page.svelte @@ -81,11 +81,12 @@ } } - onMount(async () => { - if (!auth.token) return; + let loadStarted = false; + + async function loadInitial() { try { [nodes] = await Promise.all([ - listNodes(auth.token), + listNodes(auth.token!), fetchStats(), ]); } catch (e) { @@ -93,6 +94,16 @@ } finally { loading = false; } + } + + $effect(() => { + if (auth.token && !loadStarted) { + loadStarted = true; + loadInitial(); + } + }); + + onMount(() => { pollInterval = setInterval(fetchStats, 10_000); }); diff --git a/app/src/routes/(app)/resource-groups/+page.svelte b/app/src/routes/(app)/resource-groups/+page.svelte index 21a6c3bf..42d37fcc 100644 --- a/app/src/routes/(app)/resource-groups/+page.svelte +++ b/app/src/routes/(app)/resource-groups/+page.svelte @@ -1,5 +1,4 @@ Date: Sat, 4 Jul 2026 14:13:18 +0200 Subject: [PATCH 06/16] feat: add docker compose stack deployment with service discovery and smart workload icons --- Cargo.lock | 21 ++ Cargo.toml | 1 + agent/src/client.rs | 2 + agent/src/docker.rs | 62 +++- agent/src/main.rs | 2 + app/src/lib/api/resource-groups.ts | 39 ++ app/src/lib/utils/image-icon.ts | 27 ++ .../(app)/resource-groups/[id]/+page.svelte | 339 +++++++++++++++--- .../api-gateway/src/routes/workloads.rs | 19 + control-plane/scheduler/Cargo.toml | 2 + control-plane/scheduler/src/db/mod.rs | 1 + .../scheduler/src/db/workload_stacks.rs | 41 +++ control-plane/scheduler/src/db/workloads.rs | 4 + control-plane/scheduler/src/handlers/mod.rs | 1 + .../scheduler/src/handlers/stacks.rs | 17 + control-plane/scheduler/src/models/compose.rs | 28 ++ control-plane/scheduler/src/models/mod.rs | 1 + .../scheduler/src/models/workload.rs | 6 + control-plane/scheduler/src/server.rs | 6 +- .../scheduler/src/services/compose_parser.rs | 107 ++++++ control-plane/scheduler/src/services/etcd.rs | 4 + control-plane/scheduler/src/services/mod.rs | 1 + .../scheduler/src/services/scheduler.rs | 172 ++++++++- .../shared/entity/src/entities/mod.rs | 2 + .../entity/src/entities/resource_groups.rs | 8 + .../entity/src/entities/workload_stacks.rs | 42 +++ .../shared/entity/src/entities/workloads.rs | 16 + control-plane/shared/migration/src/lib.rs | 2 + .../m20260704_000000_add_workload_stacks.rs | 137 +++++++ 29 files changed, 1054 insertions(+), 56 deletions(-) create mode 100644 app/src/lib/utils/image-icon.ts create mode 100644 control-plane/scheduler/src/db/workload_stacks.rs create mode 100644 control-plane/scheduler/src/handlers/stacks.rs create mode 100644 control-plane/scheduler/src/models/compose.rs create mode 100644 control-plane/scheduler/src/services/compose_parser.rs create mode 100644 control-plane/shared/entity/src/entities/workload_stacks.rs create mode 100644 control-plane/shared/migration/src/m20260704_000000_add_workload_stacks.rs diff --git a/Cargo.lock b/Cargo.lock index 795c5026..6ecc057e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4435,7 +4435,9 @@ dependencies = [ "sea-orm", "serde", "serde_json", + "serde_yaml", "shared", + "thiserror 2.0.18", "tokio", "tracing", "tracing-opentelemetry", @@ -4769,6 +4771,19 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_yaml" +version = "0.9.34+deprecated" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" +dependencies = [ + "indexmap", + "itoa", + "ryu", + "serde", + "unsafe-libyaml", +] + [[package]] name = "sha1" version = "0.10.6" @@ -5855,6 +5870,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "unsafe-libyaml" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" + [[package]] name = "untrusted" version = "0.9.0" diff --git a/Cargo.toml b/Cargo.toml index 27858853..8c1be5b9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,6 +43,7 @@ tokio-tungstenite = "0.29" # Serialization serde = { version = "1.0", features = ["derive"] } serde_json = "1.0.145" +serde_yaml = "0.9" # Logging & Tracing tracing = "0.1" diff --git a/agent/src/client.rs b/agent/src/client.rs index ad5c67b1..ab0f1897 100644 --- a/agent/src/client.rs +++ b/agent/src/client.rs @@ -87,6 +87,8 @@ pub struct AssignedWorkload { pub volume_mounts: Option>, pub status: String, pub container_id: Option, + pub stack_id: Option, + pub service_name: Option, } pub struct ApiClient { diff --git a/agent/src/docker.rs b/agent/src/docker.rs index 8057960c..dda0432e 100644 --- a/agent/src/docker.rs +++ b/agent/src/docker.rs @@ -1,10 +1,13 @@ use anyhow::{Context, Result}; use axum::body::Bytes; use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults}; -use bollard::models::{ContainerCreateBody, ContainerStateStatusEnum, HostConfig}; +use bollard::models::{ + ContainerCreateBody, ContainerStateStatusEnum, EndpointSettings, HostConfig, + NetworkCreateRequest, NetworkingConfig, +}; use bollard::query_parameters::{ CreateContainerOptionsBuilder, CreateImageOptionsBuilder, InspectContainerOptionsBuilder, - LogsOptionsBuilder, StartContainerOptionsBuilder, + ListNetworksOptionsBuilder, LogsOptionsBuilder, StartContainerOptionsBuilder, }; use bollard::Docker; use futures_util::{Stream, StreamExt}; @@ -41,6 +44,8 @@ pub struct WorkloadSpec { pub env_vars: Option>, pub ports: Option>, pub volume_mounts: Option>, + pub stack_id: Option, + pub service_name: Option, } pub struct DockerManager { @@ -92,6 +97,41 @@ impl DockerManager { Ok(()) } + pub async fn ensure_stack_network(&self, stack_id: &str) -> Result { + let network_name = format!("csfx-stack-{}", stack_id); + + let filters = HashMap::from([("name".to_string(), vec![network_name.clone()])]); + let list_options = ListNetworksOptionsBuilder::default() + .filters(&filters) + .build(); + + let existing = self + .docker + .list_networks(Some(list_options)) + .await + .context("Failed to list networks")?; + + if let Some(network) = existing.into_iter().find(|n| n.name.as_deref() == Some(&network_name)) { + return Ok(network.id.unwrap_or(network_name)); + } + + let config = NetworkCreateRequest { + name: network_name.clone(), + driver: Some("bridge".to_string()), + ..Default::default() + }; + + let response = self + .docker + .create_network(config) + .await + .context("Failed to create stack network")?; + + info!(stack_id = %stack_id, network = %network_name, "Stack network ready"); + + Ok(response.id) + } + pub async fn start_container(&self, spec: &WorkloadSpec) -> Result { let container_name = format!("csfx-{}", spec.workload_id); @@ -125,6 +165,23 @@ impl DockerManager { ..Default::default() }; + let networking_config = match &spec.stack_id { + Some(stack_id) => { + let network_name = self.ensure_stack_network(stack_id).await?; + let aliases = spec.service_name.clone().map(|name| vec![name]); + Some(NetworkingConfig { + endpoints_config: Some(HashMap::from([( + network_name, + EndpointSettings { + aliases, + ..Default::default() + }, + )])), + }) + } + None => None, + }; + let config = ContainerCreateBody { image: Some(spec.image.clone()), env, @@ -134,6 +191,7 @@ impl DockerManager { Some(exposed_ports) }, host_config: Some(host_config), + networking_config, labels: Some(HashMap::from([ ("csfx.workload_id".to_string(), spec.workload_id.clone()), ("csfx.managed".to_string(), "true".to_string()), diff --git a/agent/src/main.rs b/agent/src/main.rs index 3e59e87b..7f672cb0 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -419,6 +419,8 @@ async fn process_workloads( }) .collect() }), + stack_id: workload.stack_id, + service_name: workload.service_name, }; match docker.start_container(&spec).await { diff --git a/app/src/lib/api/resource-groups.ts b/app/src/lib/api/resource-groups.ts index 87b4dc2e..8d79a55f 100644 --- a/app/src/lib/api/resource-groups.ts +++ b/app/src/lib/api/resource-groups.ts @@ -51,6 +51,8 @@ export interface Workload { container_id: string | null; volume_mounts: VolumeMount[] | null; resource_group_id: string | null; + stack_id: string | null; + service_name: string | null; created_at: string; updated_at: string | null; } @@ -73,6 +75,24 @@ export interface CreateVolumeRequest { resource_group_id: string; } +export interface CreateStackRequest { + name: string; + resource_group_id: string; + compose_yaml: string; +} + +export interface CreateStackWorkloadResult { + workload_id: string; + status: string; + assigned_agent_id: string | null; + message: string; +} + +export interface CreateStackResponse { + stack_id: string; + workloads: CreateStackWorkloadResult[]; +} + export async function listResourceGroups(token: string): Promise { const res = await fetch(`${API_BASE}/resource-groups`, { headers: { Authorization: `Bearer ${token}` }, @@ -137,6 +157,25 @@ export async function createWorkload(token: string, req: CreateWorkloadRequest): return res.json(); } +export async function createWorkloadStack( + token: string, + req: CreateStackRequest, +): Promise { + const res = await fetch(`${API_BASE}/workload-stacks`, { + method: 'POST', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify(req), + }); + if (!res.ok) { + const err = await res.json().catch(() => ({ error: res.status })); + throw new Error(err.error ?? `Failed to create stack: ${res.status}`); + } + return res.json(); +} + export async function deleteWorkload(token: string, id: string): Promise { const res = await fetch(`${API_BASE}/workloads/${id}`, { method: 'DELETE', diff --git a/app/src/lib/utils/image-icon.ts b/app/src/lib/utils/image-icon.ts new file mode 100644 index 00000000..6a3e95b3 --- /dev/null +++ b/app/src/lib/utils/image-icon.ts @@ -0,0 +1,27 @@ +const KNOWN_IMAGE_ICONS: Record = { + redis: 'logos:redis', + postgres: 'logos:postgresql', + postgresql: 'logos:postgresql', + nginx: 'logos:nginx', + mysql: 'logos:mysql', + mariadb: 'logos:mariadb', + mongo: 'logos:mongodb', + mongodb: 'logos:mongodb', + rabbitmq: 'logos:rabbitmq-icon', + memcached: 'logos:memcached', + traefik: 'logos:traefikproxy-icon', + caddy: 'logos:caddy-icon', + elasticsearch: 'logos:elasticsearch', + grafana: 'logos:grafana', + prometheus: 'logos:prometheus', +}; + +const DOCKER_FALLBACK_ICON = 'logos:docker-icon'; + +export function resolveImageIcon(image: string): string { + const withoutTag = image.split('@')[0].split(':')[0]; + const segments = withoutTag.split('/'); + const bareName = segments[segments.length - 1].toLowerCase(); + + return KNOWN_IMAGE_ICONS[bareName] ?? DOCKER_FALLBACK_ICON; +} diff --git a/app/src/routes/(app)/resource-groups/[id]/+page.svelte b/app/src/routes/(app)/resource-groups/[id]/+page.svelte index 983316ef..501bedfe 100644 --- a/app/src/routes/(app)/resource-groups/[id]/+page.svelte +++ b/app/src/routes/(app)/resource-groups/[id]/+page.svelte @@ -8,6 +8,7 @@ listResourceGroupWorkloads, listResourceGroupVolumes, createWorkload, + createWorkloadStack, deleteWorkload, createVolume, deleteVolume, @@ -19,9 +20,11 @@ type PortMapping, type VolumeMount, } from "$lib/api/resource-groups"; + import { resolveImageIcon } from "$lib/utils/image-icon"; import type { Terminal } from "@xterm/xterm"; import type { FitAddon } from "@xterm/addon-fit"; import "@xterm/xterm/css/xterm.css"; + import Icon from "@iconify/svelte"; import * as Sidebar from "$lib/components/ui/sidebar/index.js"; import { Button } from "$lib/components/ui/button/index.js"; @@ -38,11 +41,21 @@ let deployDialog = $state(null); let volumeDialog = $state(null); + let resourcePickerDialog = $state(null); + let composeDialog = $state(null); let deploying = $state(false); let creatingVolume = $state(false); + let deployingStack = $state(false); let deployError = $state(null); let volumeError = $state(null); + let composeError = $state(null); let downloadingVpn = $state(false); + let expandedStacks = $state>(new Set()); + + const RESOURCE_TYPES = [ + { key: "docker-container", label: "Docker Container", description: "Deploy a single container", icon: "mdi:docker" }, + { key: "docker-compose", label: "Docker Compose", description: "Deploy multiple related containers as one stack", icon: "mdi:file-code-outline" }, + ] as const; let formImage = $state(""); let formName = $state(""); @@ -53,6 +66,9 @@ let formPorts = $state(""); let formVolumeMounts = $state(""); + let composeStackName = $state(""); + let composeYaml = $state(""); + let volFormName = $state(""); let volFormSize = $state("10"); @@ -163,6 +179,60 @@ } } + async function handleDeployStack() { + if (!auth.token || !composeYaml.trim() || !composeStackName) return; + deployingStack = true; + composeError = null; + try { + await createWorkloadStack(auth.token, { + name: composeStackName, + resource_group_id: rgId, + compose_yaml: composeYaml, + }); + composeDialog?.close(); + resetComposeForm(); + workloads = await listResourceGroupWorkloads(auth.token, rgId); + } catch (e) { + composeError = e instanceof Error ? e.message : "Failed to deploy stack"; + } finally { + deployingStack = false; + } + } + + function resetComposeForm() { + composeStackName = ""; + composeYaml = ""; + composeError = null; + } + + function handleComposeFileUpload(event: Event) { + const input = event.target as HTMLInputElement; + const file = input.files?.[0]; + if (!file) return; + file.text().then((text) => { + composeYaml = text; + }); + } + + function openResourceType(key: (typeof RESOURCE_TYPES)[number]["key"]) { + resourcePickerDialog?.close(); + if (key === "docker-container") { + deployDialog?.showModal(); + } else { + composeDialog?.showModal(); + } + } + + function toggleStack(stackId: string) { + const next = new Set(expandedStacks); + if (next.has(stackId)) { + next.delete(stackId); + } else { + next.add(stackId); + } + expandedStacks = next; + } + async function handleCreateVolume() { if (!auth.token || !volFormName) return; creatingVolume = true; @@ -336,24 +406,68 @@ deployError = null; } + type WorkloadStack = { + stack_id: string; + stack_name: string; + children: Workload[]; + }; + type ResourceItem = | { kind: "container"; data: Workload } + | { kind: "stack"; data: WorkloadStack } | { kind: "volume"; data: Volume }; + function groupWorkloadsByStack(items: Workload[]): ResourceItem[] { + const standalone: Workload[] = []; + const stacks = new Map(); + + for (const w of items) { + if (!w.stack_id) { + standalone.push(w); + continue; + } + const group = stacks.get(w.stack_id) ?? []; + group.push(w); + stacks.set(w.stack_id, group); + } + + const stackItems: ResourceItem[] = Array.from(stacks.entries()).map( + ([stackId, children]): ResourceItem => ({ + kind: "stack", + data: { + stack_id: stackId, + stack_name: `Stack ${stackId.slice(0, 8)}`, + children, + }, + }), + ); + + return [ + ...standalone.map((w): ResourceItem => ({ kind: "container", data: w })), + ...stackItems, + ]; + } + let allResources = $derived([ - ...workloads.map((w): ResourceItem => ({ kind: "container", data: w })), + ...groupWorkloadsByStack(workloads), ...volumes.map((v): ResourceItem => ({ kind: "volume", data: v })), ]); let filteredResources = $derived( allResources.filter((r) => { - if (activeTab === "container" && r.kind !== "container") return false; + if (activeTab === "container" && r.kind === "volume") return false; if (activeTab === "volume" && r.kind !== "volume") return false; if (!filterText) return true; const q = filterText.toLowerCase(); if (r.kind === "container") { return r.data.name.toLowerCase().includes(q) || r.data.image.toLowerCase().includes(q); } + if (r.kind === "stack") { + return ( + r.data.stack_name.toLowerCase().includes(q) || + r.data.children.some((c) => c.name.toLowerCase().includes(q) || c.image.toLowerCase().includes(q)) + ); + } return r.data.name.toLowerCase().includes(q); }) ); @@ -485,6 +599,88 @@ + +
+
+

Add Resource

+ +
+
+ {#each RESOURCE_TYPES as type} + + {/each} +
+
+
+ + resetComposeForm()} +> +
+
+

Deploy Docker Compose Stack

+ +
+
+ + +
+
+
+ + +
+ +
+ {#if composeError} +

{composeError}

+ {/if} +
+ + +
+
+
+ volumeDialog?.showModal()}> Add Volume - @@ -685,68 +881,111 @@ {:else} - {#each filteredResources as item (item.kind + item.data.id)} + {#snippet workloadActions(w: Workload)} + + + + {/snippet} + {#snippet workloadRow(w: Workload, indented: boolean)} + + +
+
+ +
+
+

{w.service_name ?? w.name}

+

{w.image}

+
+
+ + + Container + + + {w.assigned_agent_id ? w.assigned_agent_id.slice(0, 8) : "-"} + + +
+ {w.cpu_millicores}m +
+
+
+
+ + + + {statusLabel(w.status)} + + + + {@render workloadActions(w)} + + + {/snippet} + {#each filteredResources as item (item.kind + (item.kind === "stack" ? item.data.stack_id : item.data.id))} {#if item.kind === "container"} - {@const w = item.data} + {@render workloadRow(item.data, false)} + {:else if item.kind === "stack"} + {@const stack = item.data} + {@const expanded = expandedStacks.has(stack.stack_id)} -
+
+ + - Container + Compose Stack - {w.assigned_agent_id ? w.assigned_agent_id.slice(0, 8) : "-"} - - -
- {w.cpu_millicores}m -
-
-
-
+ {stack.children[0]?.assigned_agent_id ? stack.children[0].assigned_agent_id!.slice(0, 8) : "-"} + - - - {statusLabel(w.status)} + + {stack.children.filter((c) => c.status === 'running').length}/{stack.children.length} running - - - - - + + {#if expanded} + {#each stack.children as child (child.id)} + {@render workloadRow(child, true)} + {/each} + {/if} {:else} {@const v = item.data} diff --git a/control-plane/api-gateway/src/routes/workloads.rs b/control-plane/api-gateway/src/routes/workloads.rs index df1e0a81..80d61e73 100644 --- a/control-plane/api-gateway/src/routes/workloads.rs +++ b/control-plane/api-gateway/src/routes/workloads.rs @@ -96,6 +96,24 @@ pub async fn delete_workload( .await } +pub async fn create_stack( + CanManageWorkloads(_claims): CanManageWorkloads, + State(state): State, + headers: HeaderMap, + body: String, +) -> Result)> { + let body_json: Option = serde_json::from_str(&body).ok(); + let header_map = header_vec(&headers); + proxy_to_scheduler( + &state, + reqwest::Method::POST, + "/workload-stacks", + body_json, + Some(header_map), + ) + .await +} + fn header_vec(headers: &HeaderMap) -> Vec<(String, String)> { headers .iter() @@ -108,4 +126,5 @@ pub fn workloads_routes() -> Router { .route("/workloads", post(create_workload)) .route("/workloads", get(list_workloads)) .route("/workloads/{id}", delete(delete_workload)) + .route("/workload-stacks", post(create_stack)) } diff --git a/control-plane/scheduler/Cargo.toml b/control-plane/scheduler/Cargo.toml index 1a8ef34d..e867f558 100644 --- a/control-plane/scheduler/Cargo.toml +++ b/control-plane/scheduler/Cargo.toml @@ -20,8 +20,10 @@ tracing = { workspace = true } tracing-subscriber = { workspace = true, features = ["env-filter"] } serde = { workspace = true } serde_json = { workspace = true } +serde_yaml = { workspace = true } dotenvy = { workspace = true } anyhow = { workspace = true } +thiserror = { workspace = true } uuid = { workspace = true, features = ["v4", "serde"] } chrono = { workspace = true, features = ["serde"] } sea-orm = { workspace = true } diff --git a/control-plane/scheduler/src/db/mod.rs b/control-plane/scheduler/src/db/mod.rs index c94899ea..1e7c9165 100644 --- a/control-plane/scheduler/src/db/mod.rs +++ b/control-plane/scheduler/src/db/mod.rs @@ -1,2 +1,3 @@ pub mod agents; +pub mod workload_stacks; pub mod workloads; diff --git a/control-plane/scheduler/src/db/workload_stacks.rs b/control-plane/scheduler/src/db/workload_stacks.rs new file mode 100644 index 00000000..1fade151 --- /dev/null +++ b/control-plane/scheduler/src/db/workload_stacks.rs @@ -0,0 +1,41 @@ +use chrono::Utc; +use entity::entities::workload_stacks; +use sea_orm::{ActiveModelTrait, ActiveValue::Set, DatabaseConnection, EntityTrait, ModelTrait}; +use uuid::Uuid; + +pub async fn create( + db: &DatabaseConnection, + resource_group_id: Uuid, + name: &str, + compose_source: &str, +) -> Result { + let model = workload_stacks::ActiveModel { + id: Set(Uuid::new_v4()), + resource_group_id: Set(resource_group_id), + name: Set(name.to_string()), + compose_source: Set(Some(compose_source.to_string())), + status: Set("pending".to_string()), + created_at: Set(Utc::now().naive_utc()), + updated_at: Set(None), + }; + + model.insert(db).await +} + +pub async fn update_status( + db: &DatabaseConnection, + stack_id: Uuid, + status: &str, +) -> Result<(), sea_orm::DbErr> { + let stack = workload_stacks::Entity::find_by_id(stack_id) + .one(db) + .await? + .ok_or(sea_orm::DbErr::RecordNotFound(stack_id.to_string()))?; + + let mut active: workload_stacks::ActiveModel = stack.into(); + active.status = Set(status.to_string()); + active.updated_at = Set(Some(Utc::now().naive_utc())); + + active.update(db).await?; + Ok(()) +} diff --git a/control-plane/scheduler/src/db/workloads.rs b/control-plane/scheduler/src/db/workloads.rs index df90fb3a..91141337 100644 --- a/control-plane/scheduler/src/db/workloads.rs +++ b/control-plane/scheduler/src/db/workloads.rs @@ -38,6 +38,8 @@ pub async fn create( created_by: Set(None), organization_id: Set(None), resource_group_id: Set(req.resource_group_id), + stack_id: Set(req.stack_id), + service_name: Set(req.service_name.clone()), created_at: Set(Utc::now().naive_utc()), updated_at: Set(None), }; @@ -118,6 +120,8 @@ fn into_response(m: workloads::Model) -> WorkloadResponse { container_id: m.container_id, volume_mounts, resource_group_id: m.resource_group_id, + stack_id: m.stack_id, + service_name: m.service_name, created_at: m.created_at.and_utc(), updated_at: m.updated_at.map(|dt| dt.and_utc()), } diff --git a/control-plane/scheduler/src/handlers/mod.rs b/control-plane/scheduler/src/handlers/mod.rs index 474d2676..3f038ecf 100644 --- a/control-plane/scheduler/src/handlers/mod.rs +++ b/control-plane/scheduler/src/handlers/mod.rs @@ -1,2 +1,3 @@ pub mod internal; +pub mod stacks; pub mod workloads; diff --git a/control-plane/scheduler/src/handlers/stacks.rs b/control-plane/scheduler/src/handlers/stacks.rs new file mode 100644 index 00000000..7fefacfe --- /dev/null +++ b/control-plane/scheduler/src/handlers/stacks.rs @@ -0,0 +1,17 @@ +use axum::{extract::State, http::StatusCode, response::IntoResponse, Json}; + +use crate::{models::compose::CreateStackRequest, server::AppState}; + +pub async fn create_stack( + State(state): State, + Json(req): Json, +) -> impl IntoResponse { + match state.scheduler.schedule_stack(req).await { + Ok(resp) => (StatusCode::CREATED, Json(serde_json::json!(resp))).into_response(), + Err(e) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": e })), + ) + .into_response(), + } +} diff --git a/control-plane/scheduler/src/models/compose.rs b/control-plane/scheduler/src/models/compose.rs new file mode 100644 index 00000000..abd47fe3 --- /dev/null +++ b/control-plane/scheduler/src/models/compose.rs @@ -0,0 +1,28 @@ +use std::collections::HashMap; +use uuid::Uuid; + +use crate::models::workload::PortMapping; + +#[derive(Debug, Clone)] +pub struct ComposeServiceSpec { + pub service_name: String, + pub image: String, + pub env_vars: Option>, + pub ports: Option>, + pub cpu_millicores: i32, + pub memory_bytes: i64, + pub disk_bytes: i64, +} + +#[derive(Debug, Clone, serde::Deserialize)] +pub struct CreateStackRequest { + pub name: String, + pub resource_group_id: Uuid, + pub compose_yaml: String, +} + +#[derive(Debug, Clone, serde::Serialize)] +pub struct CreateStackResponse { + pub stack_id: Uuid, + pub workloads: Vec, +} diff --git a/control-plane/scheduler/src/models/mod.rs b/control-plane/scheduler/src/models/mod.rs index 8580f7df..748da2a6 100644 --- a/control-plane/scheduler/src/models/mod.rs +++ b/control-plane/scheduler/src/models/mod.rs @@ -1 +1,2 @@ +pub mod compose; pub mod workload; diff --git a/control-plane/scheduler/src/models/workload.rs b/control-plane/scheduler/src/models/workload.rs index 8b6a4340..fdc5888b 100644 --- a/control-plane/scheduler/src/models/workload.rs +++ b/control-plane/scheduler/src/models/workload.rs @@ -61,6 +61,10 @@ pub struct CreateWorkloadRequest { pub ports: Option>, pub volume_mounts: Option>, pub resource_group_id: Option, + #[serde(default)] + pub stack_id: Option, + #[serde(default)] + pub service_name: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -91,6 +95,8 @@ pub struct WorkloadResponse { pub container_id: Option, pub volume_mounts: Option>, pub resource_group_id: Option, + pub stack_id: Option, + pub service_name: Option, pub created_at: DateTime, pub updated_at: Option>, } diff --git a/control-plane/scheduler/src/server.rs b/control-plane/scheduler/src/server.rs index 0b85206e..b03e0795 100644 --- a/control-plane/scheduler/src/server.rs +++ b/control-plane/scheduler/src/server.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use tokio::sync::Mutex; use crate::{ - handlers::{internal, workloads}, + handlers::{internal, stacks, workloads}, metrics, services::scheduler::SchedulerService, }; @@ -34,6 +34,10 @@ pub fn create_router(state: AppState) -> Router { "/workloads/{id}", axum::routing::delete(workloads::delete_workload), ) + .route( + "/workload-stacks", + axum::routing::post(stacks::create_stack), + ) .route( "/internal/workloads/status", axum::routing::post(internal::update_container_statuses), diff --git a/control-plane/scheduler/src/services/compose_parser.rs b/control-plane/scheduler/src/services/compose_parser.rs new file mode 100644 index 00000000..0bb70f7b --- /dev/null +++ b/control-plane/scheduler/src/services/compose_parser.rs @@ -0,0 +1,107 @@ +use std::collections::HashMap; + +use crate::models::compose::ComposeServiceSpec; +use crate::models::workload::PortMapping; + +const DEFAULT_CPU_MILLICORES: i32 = 500; +const DEFAULT_MEMORY_BYTES: i64 = 512 * 1024 * 1024; +const DEFAULT_DISK_BYTES: i64 = 1024 * 1024 * 1024; + +#[derive(Debug, thiserror::Error)] +pub enum ComposeParseError { + #[error("invalid compose yaml: {0}")] + InvalidYaml(String), + #[error("compose file defines no services")] + NoServicesDefined, + #[error("service '{0}' uses build, only image is supported")] + UnsupportedBuildDirective(String), + #[error("service '{0}' has no image")] + MissingImage(String), +} + +#[derive(Debug, serde::Deserialize)] +struct RawCompose { + services: HashMap, +} + +#[derive(Debug, serde::Deserialize)] +struct RawService { + image: Option, + build: Option, + ports: Option>, + environment: Option, +} + +#[derive(Debug, serde::Deserialize)] +#[serde(untagged)] +enum RawEnvironment { + List(Vec), + Map(HashMap), +} + +pub fn parse_compose(yaml: &str) -> Result, ComposeParseError> { + let raw: RawCompose = + serde_yaml::from_str(yaml).map_err(|e| ComposeParseError::InvalidYaml(e.to_string()))?; + + if raw.services.is_empty() { + return Err(ComposeParseError::NoServicesDefined); + } + + raw.services + .into_iter() + .map(|(service_name, service)| build_service_spec(service_name, service)) + .collect() +} + +fn build_service_spec( + service_name: String, + service: RawService, +) -> Result { + if service.build.is_some() { + return Err(ComposeParseError::UnsupportedBuildDirective(service_name)); + } + let image = service + .image + .ok_or_else(|| ComposeParseError::MissingImage(service_name.clone()))?; + + Ok(ComposeServiceSpec { + service_name, + image, + env_vars: normalize_environment(service.environment), + ports: service.ports.map(|raw| parse_compose_ports(&raw)), + cpu_millicores: DEFAULT_CPU_MILLICORES, + memory_bytes: DEFAULT_MEMORY_BYTES, + disk_bytes: DEFAULT_DISK_BYTES, + }) +} + +fn normalize_environment(raw: Option) -> Option> { + match raw? { + RawEnvironment::Map(map) => Some(map), + RawEnvironment::List(list) => Some( + list.into_iter() + .filter_map(|entry| { + let (key, value) = entry.split_once('=')?; + Some((key.to_string(), value.to_string())) + }) + .collect(), + ), + } +} + +fn parse_compose_ports(raw: &[String]) -> Vec { + raw.iter().filter_map(|entry| parse_compose_port(entry)).collect() +} + +fn parse_compose_port(entry: &str) -> Option { + let (node_port, container_port) = match entry.split_once(':') { + Some((host, container)) => (host.parse().ok(), container), + None => (None, entry), + }; + + Some(PortMapping { + container_port: container_port.parse().ok()?, + protocol: None, + node_port, + }) +} diff --git a/control-plane/scheduler/src/services/etcd.rs b/control-plane/scheduler/src/services/etcd.rs index ec826a65..1eefdeaa 100644 --- a/control-plane/scheduler/src/services/etcd.rs +++ b/control-plane/scheduler/src/services/etcd.rs @@ -13,6 +13,10 @@ pub struct PlacementRecord { pub memory_bytes: i64, pub disk_bytes: i64, pub scheduled_at: String, + #[serde(default)] + pub stack_id: Option, + #[serde(default)] + pub service_name: Option, } pub async fn put_placement( diff --git a/control-plane/scheduler/src/services/mod.rs b/control-plane/scheduler/src/services/mod.rs index d5016dca..9655cca1 100644 --- a/control-plane/scheduler/src/services/mod.rs +++ b/control-plane/scheduler/src/services/mod.rs @@ -1,2 +1,3 @@ +pub mod compose_parser; pub mod etcd; pub mod scheduler; diff --git a/control-plane/scheduler/src/services/scheduler.rs b/control-plane/scheduler/src/services/scheduler.rs index 8ae77ebc..b6bc2dd3 100644 --- a/control-plane/scheduler/src/services/scheduler.rs +++ b/control-plane/scheduler/src/services/scheduler.rs @@ -5,9 +5,11 @@ use std::sync::Arc; use tokio::sync::Mutex; use uuid::Uuid; +use crate::models::compose::{ComposeServiceSpec, CreateStackRequest, CreateStackResponse}; use crate::models::workload::{ AgentResources, CreateWorkloadRequest, CreateWorkloadResponse, WorkloadStatus, }; +use crate::services::compose_parser::{self, ComposeParseError}; use crate::services::etcd::{delete_placement, put_placement, PlacementRecord}; pub struct SchedulerService { @@ -63,6 +65,8 @@ impl SchedulerService { memory_bytes: req.memory_bytes, disk_bytes: req.disk_bytes, scheduled_at: Utc::now().to_rfc3339(), + stack_id: req.stack_id, + service_name: req.service_name.clone(), }; put_placement(&self.etcd, &record).await?; @@ -98,6 +102,139 @@ impl SchedulerService { } } + pub async fn schedule_stack( + &self, + req: CreateStackRequest, + ) -> Result { + let services = compose_parser::parse_compose(&req.compose_yaml) + .map_err(|e| format_compose_error(&e))?; + + let stack = crate::db::workload_stacks::create( + &self.db, + req.resource_group_id, + &req.name, + &req.compose_yaml, + ) + .await + .map_err(|e| format!("Failed to persist stack: {}", e))?; + + let workloads = self + .place_stack_workloads(stack.id, req.resource_group_id, &services) + .await?; + + crate::db::workload_stacks::update_status( + &self.db, + stack.id, + if workloads.iter().all(|w| w.assigned_agent_id.is_some()) { + "active" + } else { + "pending" + }, + ) + .await + .map_err(|e| format!("Failed to update stack status: {}", e))?; + + Ok(CreateStackResponse { + stack_id: stack.id, + workloads, + }) + } + + async fn place_stack_workloads( + &self, + stack_id: Uuid, + resource_group_id: Uuid, + services: &[ComposeServiceSpec], + ) -> Result, String> { + let mut agents = crate::db::agents::get_online_agents_with_resources(&self.db) + .await + .map_err(|e| format!("Failed to fetch agent resources: {}", e))?; + + for agent in agents.iter_mut() { + let (reserved_cpu, reserved_mem, reserved_disk) = + crate::db::agents::get_assigned_workload_resources(&self.db, agent.agent_id) + .await + .map_err(|e| format!("Failed to fetch reserved resources: {}", e))?; + + agent.free_cpu_millicores -= reserved_cpu; + agent.free_memory_bytes -= reserved_mem; + agent.free_disk_bytes -= reserved_disk; + } + + let total_cpu: i32 = services.iter().map(|s| s.cpu_millicores).sum(); + let total_memory: i64 = services.iter().map(|s| s.memory_bytes).sum(); + let total_disk: i64 = services.iter().map(|s| s.disk_bytes).sum(); + + let agent_id = self.first_fit_resources(total_cpu, total_memory, total_disk, &agents); + + let mut responses = Vec::with_capacity(services.len()); + for service in services { + let req = CreateWorkloadRequest { + name: format!("{}-{}", stack_id, service.service_name), + image: service.image.clone(), + cpu_millicores: service.cpu_millicores, + memory_bytes: service.memory_bytes, + disk_bytes: service.disk_bytes, + env_vars: service.env_vars.clone(), + ports: service.ports.clone(), + volume_mounts: None, + resource_group_id: Some(resource_group_id), + stack_id: Some(stack_id), + service_name: Some(service.service_name.clone()), + }; + + let workload = crate::db::workloads::create(&self.db, &req) + .await + .map_err(|e| format!("Failed to persist workload: {}", e))?; + + responses.push(match agent_id { + Some(agent_id) => { + crate::db::workloads::assign(&self.db, workload.id, agent_id) + .await + .map_err(|e| format!("Failed to assign workload: {}", e))?; + + let record = PlacementRecord { + workload_id: workload.id, + agent_id, + image: req.image.clone(), + cpu_millicores: req.cpu_millicores, + memory_bytes: req.memory_bytes, + disk_bytes: req.disk_bytes, + scheduled_at: Utc::now().to_rfc3339(), + stack_id: Some(stack_id), + service_name: Some(service.service_name.clone()), + }; + put_placement(&self.etcd, &record).await?; + + CreateWorkloadResponse { + workload_id: workload.id, + status: WorkloadStatus::Scheduled, + assigned_agent_id: Some(agent_id), + message: format!("Workload assigned to agent {}", agent_id), + } + } + None => CreateWorkloadResponse { + workload_id: workload.id, + status: WorkloadStatus::Pending, + assigned_agent_id: None, + message: "No agent with sufficient resources available".to_string(), + }, + }); + } + + crate::log_info!( + "scheduler", + &format!( + "Stack scheduled stack_id={} agent_id={:?} services={}", + stack_id, + agent_id, + services.len() + ) + ); + + Ok(responses) + } + async fn resolve_volume_affinity( &self, req: &CreateWorkloadRequest, @@ -133,6 +270,21 @@ impl SchedulerService { } fn first_fit(&self, req: &CreateWorkloadRequest, agents: &[AgentResources]) -> Option { + self.first_fit_resources( + req.cpu_millicores, + req.memory_bytes, + req.disk_bytes, + agents, + ) + } + + fn first_fit_resources( + &self, + cpu_millicores: i32, + memory_bytes: i64, + disk_bytes: i64, + agents: &[AgentResources], + ) -> Option { let mut sorted: Vec<&AgentResources> = agents.iter().collect(); sorted.sort_by(|a, b| { let score_a = a.free_cpu_millicores as i64 + a.free_memory_bytes / (1024 * 1024); @@ -143,9 +295,9 @@ impl SchedulerService { sorted .into_iter() .find(|a| { - a.free_cpu_millicores >= req.cpu_millicores - && a.free_memory_bytes >= req.memory_bytes - && a.free_disk_bytes >= req.disk_bytes + a.free_cpu_millicores >= cpu_millicores + && a.free_memory_bytes >= memory_bytes + && a.free_disk_bytes >= disk_bytes }) .map(|a| a.agent_id) } @@ -173,3 +325,17 @@ impl SchedulerService { Ok(()) } } + +fn format_compose_error(error: &ComposeParseError) -> String { + match error { + ComposeParseError::InvalidYaml(detail) => format!("invalid compose yaml: {}", detail), + ComposeParseError::NoServicesDefined => "compose file defines no services".to_string(), + ComposeParseError::UnsupportedBuildDirective(service) => format!( + "service '{}' uses build, only image is supported", + service + ), + ComposeParseError::MissingImage(service) => { + format!("service '{}' has no image", service) + } + } +} diff --git a/control-plane/shared/entity/src/entities/mod.rs b/control-plane/shared/entity/src/entities/mod.rs index 31e24b11..8b1b761b 100644 --- a/control-plane/shared/entity/src/entities/mod.rs +++ b/control-plane/shared/entity/src/entities/mod.rs @@ -23,6 +23,7 @@ pub mod user_organization; pub mod user_ssh_keys; pub mod volume_snapshots; pub mod volumes; +pub mod workload_stacks; pub mod workloads; pub use agent_api_keys::Entity as AgentApiKeys; @@ -50,4 +51,5 @@ pub use user_organization::Entity as UserOrganization; pub use user_ssh_keys::Entity as UserSshKeys; pub use volume_snapshots::Entity as VolumeSnapshots; pub use volumes::Entity as Volumes; +pub use workload_stacks::Entity as WorkloadStacks; pub use workloads::Entity as Workloads; diff --git a/control-plane/shared/entity/src/entities/resource_groups.rs b/control-plane/shared/entity/src/entities/resource_groups.rs index cc63a440..1b8febac 100644 --- a/control-plane/shared/entity/src/entities/resource_groups.rs +++ b/control-plane/shared/entity/src/entities/resource_groups.rs @@ -30,6 +30,8 @@ pub enum Relation { Volumes, #[sea_orm(has_many = "super::networks::Entity")] Networks, + #[sea_orm(has_many = "super::workload_stacks::Entity")] + WorkloadStacks, } impl Related for Entity { @@ -56,4 +58,10 @@ impl Related for Entity { } } +impl Related for Entity { + fn to() -> RelationDef { + Relation::WorkloadStacks.def() + } +} + impl ActiveModelBehavior for ActiveModel {} diff --git a/control-plane/shared/entity/src/entities/workload_stacks.rs b/control-plane/shared/entity/src/entities/workload_stacks.rs new file mode 100644 index 00000000..74b87305 --- /dev/null +++ b/control-plane/shared/entity/src/entities/workload_stacks.rs @@ -0,0 +1,42 @@ +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)] +#[sea_orm(table_name = "workload_stacks")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: Uuid, + pub resource_group_id: Uuid, + pub name: String, + pub compose_source: Option, + pub status: String, + pub created_at: chrono::NaiveDateTime, + pub updated_at: Option, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::resource_groups::Entity", + from = "Column::ResourceGroupId", + to = "super::resource_groups::Column::Id", + on_delete = "Cascade" + )] + ResourceGroup, + #[sea_orm(has_many = "super::workloads::Entity")] + Workloads, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::ResourceGroup.def() + } +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Workloads.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/control-plane/shared/entity/src/entities/workloads.rs b/control-plane/shared/entity/src/entities/workloads.rs index b4c662d9..ee82536f 100644 --- a/control-plane/shared/entity/src/entities/workloads.rs +++ b/control-plane/shared/entity/src/entities/workloads.rs @@ -20,6 +20,8 @@ pub struct Model { pub created_by: Option, pub organization_id: Option, pub resource_group_id: Option, + pub stack_id: Option, + pub service_name: Option, pub created_at: DateTime, pub updated_at: Option, } @@ -42,6 +44,14 @@ pub enum Relation { on_delete = "SetNull" )] ResourceGroup, + #[sea_orm( + belongs_to = "super::workload_stacks::Entity", + from = "Column::StackId", + to = "super::workload_stacks::Column::Id", + on_update = "NoAction", + on_delete = "Cascade" + )] + Stack, } impl Related for Entity { @@ -56,4 +66,10 @@ impl Related for Entity { } } +impl Related for Entity { + fn to() -> RelationDef { + Relation::Stack.def() + } +} + impl ActiveModelBehavior for ActiveModel {} diff --git a/control-plane/shared/migration/src/lib.rs b/control-plane/shared/migration/src/lib.rs index 98df39ee..111dbc3a 100644 --- a/control-plane/shared/migration/src/lib.rs +++ b/control-plane/shared/migration/src/lib.rs @@ -23,6 +23,7 @@ mod m20260628_000000_add_volume_mounts; mod m20260628_100000_rg_cidr_unique_agent_wg; mod m20260701_000000_add_logs_and_settings; mod m20260702_000000_add_agent_wg_tunnel_ip; +mod m20260704_000000_add_workload_stacks; pub struct Migrator; @@ -53,6 +54,7 @@ impl MigratorTrait for Migrator { Box::new(m20260628_100000_rg_cidr_unique_agent_wg::Migration), Box::new(m20260701_000000_add_logs_and_settings::Migration), Box::new(m20260702_000000_add_agent_wg_tunnel_ip::Migration), + Box::new(m20260704_000000_add_workload_stacks::Migration), ] } } diff --git a/control-plane/shared/migration/src/m20260704_000000_add_workload_stacks.rs b/control-plane/shared/migration/src/m20260704_000000_add_workload_stacks.rs new file mode 100644 index 00000000..1d28afe4 --- /dev/null +++ b/control-plane/shared/migration/src/m20260704_000000_add_workload_stacks.rs @@ -0,0 +1,137 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .create_table( + Table::create() + .table(WorkloadStacks::Table) + .if_not_exists() + .col(ColumnDef::new(WorkloadStacks::Id).uuid().not_null().primary_key()) + .col(ColumnDef::new(WorkloadStacks::ResourceGroupId).uuid().not_null()) + .col(ColumnDef::new(WorkloadStacks::Name).string().not_null()) + .col(ColumnDef::new(WorkloadStacks::ComposeSource).text().null()) + .col(ColumnDef::new(WorkloadStacks::Status).string().not_null().default("pending")) + .col(ColumnDef::new(WorkloadStacks::CreatedAt).date_time().not_null()) + .col(ColumnDef::new(WorkloadStacks::UpdatedAt).date_time().null()) + .foreign_key( + ForeignKey::create() + .from(WorkloadStacks::Table, WorkloadStacks::ResourceGroupId) + .to(ResourceGroups::Table, ResourceGroups::Id) + .on_delete(ForeignKeyAction::Cascade), + ) + .to_owned(), + ) + .await?; + + manager + .alter_table( + Table::alter() + .table(Workloads::Table) + .add_column(ColumnDef::new(Workloads::StackId).uuid().null()) + .add_column(ColumnDef::new(Workloads::ServiceName).string().null()) + .to_owned(), + ) + .await?; + + manager + .create_foreign_key( + ForeignKey::create() + .name("fk_workloads_stack_id") + .from(Workloads::Table, Workloads::StackId) + .to(WorkloadStacks::Table, WorkloadStacks::Id) + .on_delete(ForeignKeyAction::Cascade) + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .table(WorkloadStacks::Table) + .col(WorkloadStacks::ResourceGroupId) + .name("idx_workload_stacks_resource_group_id") + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .table(Workloads::Table) + .col(Workloads::StackId) + .name("idx_workloads_stack_id") + .to_owned(), + ) + .await?; + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_index(Index::drop().name("idx_workloads_stack_id").to_owned()) + .await?; + manager + .drop_index( + Index::drop() + .name("idx_workload_stacks_resource_group_id") + .to_owned(), + ) + .await?; + + manager + .drop_foreign_key( + ForeignKey::drop() + .name("fk_workloads_stack_id") + .table(Workloads::Table) + .to_owned(), + ) + .await?; + + manager + .alter_table( + Table::alter() + .table(Workloads::Table) + .drop_column(Workloads::ServiceName) + .drop_column(Workloads::StackId) + .to_owned(), + ) + .await?; + + manager + .drop_table(Table::drop().table(WorkloadStacks::Table).to_owned()) + .await?; + + Ok(()) + } +} + +#[derive(DeriveIden)] +enum WorkloadStacks { + Table, + Id, + ResourceGroupId, + Name, + ComposeSource, + Status, + CreatedAt, + UpdatedAt, +} + +#[derive(DeriveIden)] +enum ResourceGroups { + Table, + Id, +} + +#[derive(DeriveIden)] +enum Workloads { + Table, + StackId, + ServiceName, +} From caf3acb92b7f6e48fe6fa6f0498db6f6f0dfd39b Mon Sep 17 00:00:00 2001 From: CodeMaster4711 Date: Sat, 4 Jul 2026 14:34:40 +0200 Subject: [PATCH 07/16] feat: add resource picker with docker compose stacks and dynamic service icons --- .../routes/(app)/resource-groups/[id]/+page.svelte | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/app/src/routes/(app)/resource-groups/[id]/+page.svelte b/app/src/routes/(app)/resource-groups/[id]/+page.svelte index 501bedfe..3985de64 100644 --- a/app/src/routes/(app)/resource-groups/[id]/+page.svelte +++ b/app/src/routes/(app)/resource-groups/[id]/+page.svelte @@ -620,8 +620,8 @@ class="flex items-center gap-3 rounded-lg border p-3 text-left hover:bg-muted/40 transition-colors" onclick={() => openResourceType(type.key)} > -
- +
+

{type.label}

@@ -912,8 +912,8 @@
-
- +
+

{w.service_name ?? w.name}

@@ -957,8 +957,8 @@ class="flex items-center gap-2.5 text-left" onclick={() => toggleStack(stack.stack_id)} > -
- +
+

{stack.stack_name}

From 46266c7801cebac8a26931f3834276c3be402049 Mon Sep 17 00:00:00 2001 From: CodeMaster4711 Date: Sat, 4 Jul 2026 14:59:34 +0200 Subject: [PATCH 08/16] fix: correct pre-release version comparison and let forced updates bypass pause --- app/src/lib/utils/compose-preview.ts | 80 ++++++++++++++ app/src/lib/utils/yaml-highlight.ts | 55 ++++++++++ .../(app)/resource-groups/[id]/+page.svelte | 101 ++++++++++++++---- .../api-gateway/src/routes/releases.rs | 14 ++- .../api-gateway/src/routes/update.rs | 9 ++ control-plane/csfx-updater/src/etcd.rs | 6 ++ control-plane/csfx-updater/src/main.rs | 11 +- 7 files changed, 254 insertions(+), 22 deletions(-) create mode 100644 app/src/lib/utils/compose-preview.ts create mode 100644 app/src/lib/utils/yaml-highlight.ts diff --git a/app/src/lib/utils/compose-preview.ts b/app/src/lib/utils/compose-preview.ts new file mode 100644 index 00000000..a44328d1 --- /dev/null +++ b/app/src/lib/utils/compose-preview.ts @@ -0,0 +1,80 @@ +export interface ComposePreviewService { + serviceName: string; + image: string | null; + ports: string[]; +} + +export interface ComposePreviewResult { + services: ComposePreviewService[]; + volumes: string[]; + error: string | null; +} + +function stripInlineComment(line: string): string { + const hashIndex = line.indexOf('#'); + return hashIndex === -1 ? line : line.slice(0, hashIndex); +} + +function indentOf(line: string): number { + return line.length - line.trimStart().length; +} + +function parseTopLevelBlockKeys(lines: string[], blockName: string): string[] { + const blockLineIndex = lines.findIndex((line) => line.trim() === `${blockName}:`); + if (blockLineIndex === -1) return []; + + const keys: string[] = []; + for (let i = blockLineIndex + 1; i < lines.length; i++) { + const line = lines[i]; + if (!line.trim()) continue; + const indent = indentOf(line); + if (indent === 0) break; + if (indent === 2 && line.trim().endsWith(':')) { + keys.push(line.trim().slice(0, -1)); + } + } + return keys; +} + +export function parseComposePreview(yaml: string): ComposePreviewResult { + const lines = yaml.split('\n').map(stripInlineComment); + const volumes = parseTopLevelBlockKeys(lines, 'volumes'); + const servicesLineIndex = lines.findIndex((line) => line.trim() === 'services:'); + if (servicesLineIndex === -1) { + return { services: [], volumes, error: null }; + } + + const services: ComposePreviewService[] = []; + let current: ComposePreviewService | null = null; + let inPorts = false; + + for (let i = servicesLineIndex + 1; i < lines.length; i++) { + const line = lines[i]; + if (!line.trim()) continue; + const indent = indentOf(line); + const trimmed = line.trim(); + + if (indent === 0) break; + + if (indent === 2 && trimmed.endsWith(':')) { + current = { serviceName: trimmed.slice(0, -1), image: null, ports: [] }; + services.push(current); + inPorts = false; + continue; + } + if (indent <= 0 || !current) continue; + + if (trimmed.startsWith('image:')) { + current.image = trimmed.slice('image:'.length).trim().replace(/^["']|["']$/g, ''); + inPorts = false; + } else if (trimmed === 'ports:') { + inPorts = true; + } else if (inPorts && trimmed.startsWith('-')) { + current.ports.push(trimmed.slice(1).trim().replace(/^["']|["']$/g, '')); + } else if (trimmed.endsWith(':')) { + inPorts = false; + } + } + + return { services, volumes, error: null }; +} diff --git a/app/src/lib/utils/yaml-highlight.ts b/app/src/lib/utils/yaml-highlight.ts new file mode 100644 index 00000000..62a315e6 --- /dev/null +++ b/app/src/lib/utils/yaml-highlight.ts @@ -0,0 +1,55 @@ +function escapeHtml(text: string): string { + return text + .replace(/&/g, '&') + .replace(//g, '>'); +} + +function highlightLine(line: string): string { + const commentIndex = line.indexOf('#'); + const code = commentIndex === -1 ? line : line.slice(0, commentIndex); + const comment = commentIndex === -1 ? '' : line.slice(commentIndex); + + const leadingSpaces = code.match(/^\s*/)?.[0] ?? ''; + const rest = code.slice(leadingSpaces.length); + + const dashMatch = rest.match(/^-\s*/); + const dashPrefix = dashMatch ? dashMatch[0] : ''; + const afterDash = rest.slice(dashPrefix.length); + + const keyMatch = afterDash.match(/^([A-Za-z0-9_.-]+)(:)(\s|$)/); + let highlighted: string; + + if (keyMatch) { + const key = keyMatch[1]; + const colon = keyMatch[2]; + const separator = keyMatch[3]; + const value = afterDash.slice(keyMatch[0].length); + highlighted = `${escapeHtml(key)}${colon}${separator}${highlightValue(value)}`; + } else { + highlighted = highlightValue(afterDash); + } + + const dashHtml = dashPrefix ? `${escapeHtml(dashPrefix)}` : ''; + const commentHtml = comment ? `${escapeHtml(comment)}` : ''; + + return `${escapeHtml(leadingSpaces)}${dashHtml}${highlighted}${commentHtml}`; +} + +function highlightValue(value: string): string { + if (!value) return ''; + if (/^["'].*["']$/.test(value)) { + return `${escapeHtml(value)}`; + } + if (/^-?\d+(\.\d+)?$/.test(value)) { + return `${escapeHtml(value)}`; + } + if (value === 'true' || value === 'false' || value === 'null') { + return `${escapeHtml(value)}`; + } + return `${escapeHtml(value)}`; +} + +export function highlightYaml(yaml: string): string { + return yaml.split('\n').map(highlightLine).join('\n'); +} diff --git a/app/src/routes/(app)/resource-groups/[id]/+page.svelte b/app/src/routes/(app)/resource-groups/[id]/+page.svelte index 3985de64..708e5c3b 100644 --- a/app/src/routes/(app)/resource-groups/[id]/+page.svelte +++ b/app/src/routes/(app)/resource-groups/[id]/+page.svelte @@ -21,6 +21,8 @@ type VolumeMount, } from "$lib/api/resource-groups"; import { resolveImageIcon } from "$lib/utils/image-icon"; + import { parseComposePreview } from "$lib/utils/compose-preview"; + import { highlightYaml } from "$lib/utils/yaml-highlight"; import type { Terminal } from "@xterm/xterm"; import type { FitAddon } from "@xterm/addon-fit"; import "@xterm/xterm/css/xterm.css"; @@ -53,8 +55,8 @@ let expandedStacks = $state>(new Set()); const RESOURCE_TYPES = [ - { key: "docker-container", label: "Docker Container", description: "Deploy a single container", icon: "mdi:docker" }, - { key: "docker-compose", label: "Docker Compose", description: "Deploy multiple related containers as one stack", icon: "mdi:file-code-outline" }, + { key: "docker-container", label: "Docker Container", description: "Deploy a single container", icon: "logos:docker-icon" }, + { key: "docker-compose", label: "Docker Compose", description: "Deploy multiple related containers as one stack", icon: "logos:docker-icon" }, ] as const; let formImage = $state(""); @@ -68,6 +70,21 @@ let composeStackName = $state(""); let composeYaml = $state(""); + let composePreview = $derived(parseComposePreview(composeYaml)); + let composeLineCount = $derived(Math.max(composeYaml.split("\n").length, 1)); + let composeGutter = $state(null); + let composeTextarea = $state(null); + let composeHighlightLayer = $state(null); + let composeHighlighted = $derived(highlightYaml(composeYaml)); + function syncComposeScroll() { + if (composeGutter && composeTextarea) { + composeGutter.scrollTop = composeTextarea.scrollTop; + } + if (composeHighlightLayer && composeTextarea) { + composeHighlightLayer.scrollTop = composeTextarea.scrollTop; + composeHighlightLayer.scrollLeft = composeTextarea.scrollLeft; + } + } let volFormName = $state(""); let volFormSize = $state("10"); @@ -635,10 +652,10 @@ resetComposeForm()} > -
+

Deploy Docker Compose Stack

-
-
- - +
+
+ +
+ {#if composePreview.services.length === 0 && composePreview.volumes.length === 0} +

No services detected yet.

+ {:else} + {#each composePreview.services as service (service.serviceName)} +
+ +
+

{service.serviceName}

+

{service.image ?? "no image"}

+ {#if service.ports.length > 0} +

{service.ports.join(", ")}

+ {/if} +
+
+ {/each} + {#if composePreview.volumes.length > 0} +

Volumes

+ {#each composePreview.volumes as volumeName (volumeName)} +
+ +

{volumeName}

+
+ {/each} + {/if} + {/if} +
+
+
+
+ + +
+
+
+ {#each Array(composeLineCount) as _, i} +
{i + 1}
+ {/each} +
+
{@html composeHighlighted}
+ +
-
{#if composeError}

{composeError}

diff --git a/control-plane/api-gateway/src/routes/releases.rs b/control-plane/api-gateway/src/routes/releases.rs index 1f6e2dc1..c0453ae6 100644 --- a/control-plane/api-gateway/src/routes/releases.rs +++ b/control-plane/api-gateway/src/routes/releases.rs @@ -116,16 +116,26 @@ fn semver_is_newer(candidate: &str, current: &str) -> bool { .unwrap_or(false) } -fn parse_semver(v: &str) -> Option<(u32, u32, u32)> { +fn parse_semver(v: &str) -> Option<(u32, u32, u32, u32)> { let v = v.trim_start_matches('v'); - let base = v.split('-').next().unwrap_or(v); + let mut segments = v.splitn(2, '-'); + let base = segments.next().unwrap_or(v); + let pre_release = segments.next(); + let parts: Vec<&str> = base.split('.').collect(); if parts.len() != 3 { return None; } + + let pre_release_number = pre_release + .and_then(|p| p.rsplit('.').next()) + .and_then(|n| n.parse().ok()) + .unwrap_or(u32::MAX); + Some(( parts[0].parse().ok()?, parts[1].parse().ok()?, parts[2].parse().ok()?, + pre_release_number, )) } diff --git a/control-plane/api-gateway/src/routes/update.rs b/control-plane/api-gateway/src/routes/update.rs index 124928b1..9cf6a899 100644 --- a/control-plane/api-gateway/src/routes/update.rs +++ b/control-plane/api-gateway/src/routes/update.rs @@ -18,6 +18,7 @@ const ETCD_DESIRED_FLAKE_REV_KEY: &str = "/csfx/config/desired_flake_rev"; const ETCD_BUILD_STATUS_KEY: &str = "/csfx/config/cp_build_status"; const ETCD_RESULT_KEY: &str = "/csfx/config/last_build_result"; const ETCD_PAUSED_KEY: &str = "/csfx/config/update_paused"; +const ETCD_FORCE_UPDATE_KEY: &str = "/csfx/config/update_force"; #[derive(Debug, Deserialize)] pub struct UpdateRequest { @@ -79,6 +80,14 @@ async fn trigger_update( StatusCode::INTERNAL_SERVER_ERROR })?; + client + .put(ETCD_FORCE_UPDATE_KEY, b"true", None) + .await + .map_err(|e| { + tracing::error!(error = %e, "failed to write force update flag to etcd"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + tracing::info!(version = %req.version, "update requested"); Ok(Json(UpdateResponse { diff --git a/control-plane/csfx-updater/src/etcd.rs b/control-plane/csfx-updater/src/etcd.rs index 7052752e..0cf40f64 100644 --- a/control-plane/csfx-updater/src/etcd.rs +++ b/control-plane/csfx-updater/src/etcd.rs @@ -8,6 +8,7 @@ pub const DESIRED_FLAKE_REV_KEY: &str = "/csfx/config/desired_flake_rev"; pub const BUILD_STATUS_KEY: &str = "/csfx/config/cp_build_status"; pub const RESULT_KEY: &str = "/csfx/config/last_build_result"; pub const PAUSED_KEY: &str = "/csfx/config/update_paused"; +pub const FORCE_UPDATE_KEY: &str = "/csfx/config/update_force"; pub const NODE_HEARTBEAT_PREFIX: &str = "/csfx/nodes/"; pub struct Client { @@ -35,6 +36,11 @@ impl Client { Ok(()) } + pub async fn delete(&mut self, key: &str) -> Result<()> { + self.inner.delete(key, None).await?; + Ok(()) + } + pub async fn delete_prefix(&mut self, prefix: &str) -> Result<()> { use etcd_client::DeleteOptions; self.inner diff --git a/control-plane/csfx-updater/src/main.rs b/control-plane/csfx-updater/src/main.rs index 8f57875d..224cc40d 100644 --- a/control-plane/csfx-updater/src/main.rs +++ b/control-plane/csfx-updater/src/main.rs @@ -102,8 +102,10 @@ async fn run_executor_loop(cfg: &config::Config) { async fn execute_once(cfg: &config::Config, last_applied: &str) -> anyhow::Result> { let mut etcd = etcd::Client::connect(cfg).await?; - if etcd.get(etcd::PAUSED_KEY).await?.as_deref() == Some("true") { - tracing::debug!("updater paused, skipping cycle"); + let forced = etcd.get(etcd::FORCE_UPDATE_KEY).await?.as_deref() == Some("true"); + + if !forced && etcd.get(etcd::PAUSED_KEY).await?.as_deref() == Some("true") { + tracing::warn!("updater paused, skipping cycle"); return Ok(None); } @@ -125,6 +127,11 @@ async fn execute_once(cfg: &config::Config, last_applied: &str) -> anyhow::Resul return Ok(None); } + if forced { + etcd.delete(etcd::FORCE_UPDATE_KEY).await?; + info!(flake_rev = %desired, "forced update requested, bypassing pause"); + } + if !is_valid_sha(&desired) { tracing::warn!(flake_rev = %desired, "rejected invalid flake rev"); etcd.put(etcd::RESULT_KEY, "failed").await?; From 72c46e3ad47056116611a9c7140442e17223f86b Mon Sep 17 00:00:00 2001 From: CodeMaster4711 Date: Sat, 4 Jul 2026 15:13:08 +0200 Subject: [PATCH 09/16] feat: propagate agent version on every heartbeat and add live node metrics tunnel --- agent/src/client.rs | 2 + agent/src/server.rs | 51 +++++++ agent/src/system.rs | 51 +++++++ app/src/lib/api/nodes.ts | 28 ++++ .../components/nodes/NodeDetailSheet.svelte | 71 +++++++++- control-plane/api-gateway/src/auth/jwt.rs | 53 +++++++ .../api-gateway/src/routes/agent_proxy.rs | 133 +++++++++++++++++- control-plane/registry/src/db/agents.rs | 4 + control-plane/registry/src/handlers/agent.rs | 1 + control-plane/registry/src/handlers/pki.rs | 4 +- control-plane/registry/src/models/agent.rs | 1 + control-plane/registry/src/services/pki.rs | 2 +- .../registry/src/services/registry.rs | 2 + 13 files changed, 393 insertions(+), 10 deletions(-) diff --git a/agent/src/client.rs b/agent/src/client.rs index ab0f1897..6a26473c 100644 --- a/agent/src/client.rs +++ b/agent/src/client.rs @@ -53,6 +53,7 @@ struct HeartbeatRequest { wg_public_key: Option, wg_endpoint: Option, wg_tunnel_ip: Option, + agent_version: Option, } #[derive(Debug, Serialize, Deserialize, Clone)] @@ -254,6 +255,7 @@ impl ApiClient { wg_public_key: Some(self.wg_public_key.clone()), wg_endpoint: self.wg_endpoint.clone(), wg_tunnel_ip: self.wg_tunnel_ip.clone(), + agent_version: Some(env!("CARGO_PKG_VERSION").to_string()), }); if let Some(ref cert_pem) = self.cert_pem { diff --git a/agent/src/server.rs b/agent/src/server.rs index ad5e55d4..02d7dc83 100644 --- a/agent/src/server.rs +++ b/agent/src/server.rs @@ -19,8 +19,10 @@ use tracing::{info, warn}; use crate::docker::DockerManager; use crate::pki::AgentPki; +use crate::system::LiveMetricsCollector; const TICKET_HEADER: &str = "X-Csfx-Proxy-Ticket"; +pub const METRICS_TICKET_SCOPE: &str = "__node_metrics__"; #[derive(Clone)] pub struct ServerState { @@ -33,6 +35,7 @@ pub async fn run(state: ServerState, port: u16) -> Result<()> { let app = Router::new() .route("/logs/{workload_id}", get(logs_handler)) .route("/exec/{workload_id}", get(exec_handler)) + .route("/metrics/stream", get(metrics_stream_handler)) .with_state(state); let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port); @@ -274,3 +277,51 @@ async fn handle_exec_socket(socket: WebSocket, state: ServerState, container_id: info!(container_id = %container_id, "exec session closed"); } + +async fn metrics_stream_handler( + State(state): State, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + ws: WebSocketUpgrade, +) -> Result { + if !is_internal_source(&addr) { + warn!(source = %addr, "rejected agent inbound request from non-internal source"); + return Err((StatusCode::FORBIDDEN, "source not allowed".to_string())); + } + + verify_ticket(&headers, METRICS_TICKET_SCOPE, &state.agent_id.to_string()) + .map_err(|e| (StatusCode::UNAUTHORIZED, e.to_string()))?; + + Ok(ws.on_upgrade(handle_metrics_socket)) +} + +async fn handle_metrics_socket(socket: WebSocket) { + let (mut sink, mut stream) = socket.split(); + let mut collector = LiveMetricsCollector::new(); + let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); + + loop { + tokio::select! { + _ = interval.tick() => { + let metrics = collector.sample(); + let payload = match serde_json::to_string(&metrics) { + Ok(p) => p, + Err(e) => { + warn!(error = %e, "failed to serialize live metrics"); + continue; + } + }; + if sink.send(Message::Text(payload.into())).await.is_err() { + break; + } + } + msg = stream.next() => match msg { + Some(Ok(Message::Close(_))) | None => break, + Some(Err(_)) => break, + _ => {} + } + } + } + + info!("live metrics session closed"); +} diff --git a/agent/src/system.rs b/agent/src/system.rs index 2ae27cae..b3d188ef 100644 --- a/agent/src/system.rs +++ b/agent/src/system.rs @@ -1,3 +1,4 @@ +use serde::Serialize; use sysinfo::{Disks, Networks, System}; pub struct SystemInfo { @@ -7,6 +8,7 @@ pub struct SystemInfo { pub architecture: String, } +#[derive(Serialize)] pub struct SystemMetrics { pub cpu_usage_percent: f32, pub cpu_cores: u32, @@ -19,6 +21,55 @@ pub struct SystemMetrics { pub uptime_seconds: u64, } +pub struct LiveMetricsCollector { + sys: System, +} + +impl LiveMetricsCollector { + pub fn new() -> Self { + let mut sys = System::new_all(); + sys.refresh_cpu_usage(); + Self { sys } + } + + pub fn sample(&mut self) -> SystemMetrics { + self.sys.refresh_cpu_usage(); + self.sys.refresh_memory(); + + let cpu_usage_percent = self.sys.global_cpu_usage(); + let cpu_cores = self.sys.cpus().len() as u32; + let memory_total_bytes = self.sys.total_memory(); + let memory_used_bytes = self.sys.used_memory(); + + let disks = Disks::new_with_refreshed_list(); + let (disk_total_bytes, disk_used_bytes) = + disks.iter().fold((0u64, 0u64), |(total, used), d| { + ( + total + d.total_space(), + used + (d.total_space() - d.available_space()), + ) + }); + + let networks = Networks::new_with_refreshed_list(); + let (network_rx_bytes, network_tx_bytes) = + networks.iter().fold((0u64, 0u64), |(rx, tx), (_, data)| { + (rx + data.total_received(), tx + data.total_transmitted()) + }); + + SystemMetrics { + cpu_usage_percent, + cpu_cores, + memory_total_bytes, + memory_used_bytes, + disk_total_bytes, + disk_used_bytes, + network_rx_bytes, + network_tx_bytes, + uptime_seconds: System::uptime(), + } + } +} + fn parse_os_release_field(content: &str, field: &str) -> Option { content .lines() diff --git a/app/src/lib/api/nodes.ts b/app/src/lib/api/nodes.ts index d8073408..131174ef 100644 --- a/app/src/lib/api/nodes.ts +++ b/app/src/lib/api/nodes.ts @@ -92,6 +92,34 @@ export async function getNodeMetricsLatest(token: string, id: string): Promise { + const res = await fetch(`${API_BASE}/agents/${agentId}/metrics/ticket`, { + method: 'POST', + headers: { Authorization: `Bearer ${token}` }, + }); + if (!res.ok) { + const err = await res.json().catch(() => ({ error: res.status })); + throw new Error(err.error ?? `Failed to issue metrics ticket: ${res.status}`); + } + const { ticket } = await res.json(); + + const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const url = `${wsProtocol}//${window.location.host}${API_BASE}/agents/${agentId}/metrics/stream?ticket=${encodeURIComponent(ticket)}`; + return new WebSocket(url); +} + export interface HealthHistoryPoint { bucket: string; online_count: number; diff --git a/app/src/lib/components/nodes/NodeDetailSheet.svelte b/app/src/lib/components/nodes/NodeDetailSheet.svelte index 85f6d9ee..9d72aed9 100644 --- a/app/src/lib/components/nodes/NodeDetailSheet.svelte +++ b/app/src/lib/components/nodes/NodeDetailSheet.svelte @@ -1,6 +1,6 @@