From c21374f2a2181ae95540f374c2f2748a707b7053 Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Tue, 25 Aug 2026 19:29:53 +0200 Subject: [PATCH] Full rewrite of Kubernetes integration. --- Cargo.lock | 661 ++++++++++++++++++- Cargo.toml | 4 + etc/kubernetes/aralez-certificate-secret.yml | 9 + etc/kubernetes/aralez-example-deployment.yml | 43 ++ etc/kubernetes/aralez-example-main.yml | 16 + etc/kubernetes/aralez-example-upstreams.yml | 14 + etc/kubernetes/nginx-example-deployment.yml | 21 + etc/kubernetes/nginx-example-ingress.yml | 24 + etc/kubernetes/nginx-example-service.yml | 11 + src/utils.rs | 1 + src/utils/healthcheck.rs | 10 +- src/utils/kuberconsul.rs | 208 +++--- src/utils/kubewatcher.rs | 271 ++++++++ 13 files changed, 1204 insertions(+), 89 deletions(-) create mode 100644 etc/kubernetes/aralez-certificate-secret.yml create mode 100644 etc/kubernetes/aralez-example-deployment.yml create mode 100644 etc/kubernetes/aralez-example-main.yml create mode 100644 etc/kubernetes/aralez-example-upstreams.yml create mode 100644 etc/kubernetes/nginx-example-deployment.yml create mode 100644 etc/kubernetes/nginx-example-ingress.yml create mode 100644 etc/kubernetes/nginx-example-service.yml create mode 100644 src/utils/kubewatcher.rs diff --git a/Cargo.lock b/Cargo.lock index cddf4db..0d7dd44 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -69,6 +69,17 @@ dependencies = [ "libc", ] +[[package]] +name = "annotate-snippets" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f211a51805bc641f3ad5b7664c77d2547af685cc33b4cd8d31964027a46f13f1" +dependencies = [ + "anstyle", + "memchr", + "unicode-width", +] + [[package]] name = "anstream" version = "1.0.0" @@ -137,11 +148,14 @@ dependencies = [ "base64 0.23.1", "bytes", "dashmap", + "futures", "http-body-util", "hyper", "hyper-util", "instant-acme", "jsonwebtoken", + "k8s-openapi", + "kube", "libc", "log", "log4rs", @@ -159,6 +173,7 @@ dependencies = [ "prometheus 0.14.0", "rand 0.10.2", "rcgen", + "rustls", "rustls-pemfile", "sd-notify", "serde", @@ -168,7 +183,7 @@ dependencies = [ "subtle", "tokio", "tokio-native-tls", - "tower-http", + "tower-http 0.7.0", "urlencoding", "x509-parser", ] @@ -182,6 +197,12 @@ dependencies = [ "rustversion", ] +[[package]] +name = "arraydeque" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236" + [[package]] name = "arrayvec" version = "0.7.8" @@ -227,6 +248,40 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "async-broadcast" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532" +dependencies = [ + "event-listener", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "async-trait" version = "0.1.91" @@ -326,6 +381,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "backon" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" +dependencies = [ + "fastrand", + "gloo-timers", + "tokio", +] + [[package]] name = "backtrace" version = "0.3.76" @@ -772,8 +838,18 @@ version = "0.20.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc7f46116c46ff9ab3eb1597a45688b6715c6e628b5c133e288e709a29bcb4ee" dependencies = [ - "darling_core", - "darling_macro", + "darling_core 0.20.11", + "darling_macro 0.20.11", +] + +[[package]] +name = "darling" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +dependencies = [ + "darling_core 0.23.0", + "darling_macro 0.23.0", ] [[package]] @@ -790,13 +866,37 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "darling_core" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +dependencies = [ + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn 2.0.118", +] + [[package]] name = "darling_macro" version = "0.20.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ - "darling_core", + "darling_core 0.20.11", + "quote", + "syn 2.0.118", +] + +[[package]] +name = "darling_macro" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +dependencies = [ + "darling_core 0.23.0", "quote", "syn 2.0.118", ] @@ -821,6 +921,37 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" +[[package]] +name = "defmt" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1" +dependencies = [ + "bitflags 1.3.2", + "defmt-macros", +] + +[[package]] +name = "defmt-macros" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8" +dependencies = [ + "defmt-parser", + "proc-macro2", + "quote", + "syn 2.0.118", +] + +[[package]] +name = "defmt-parser" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e" +dependencies = [ + "thiserror 2.0.18", +] + [[package]] name = "der" version = "0.7.10" @@ -878,7 +1009,7 @@ version = "0.20.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2d5bcf7b024d6835cfb3d473887cd966994907effbe9227e8c8219824d06c4e8" dependencies = [ - "darling", + "darling 0.20.11", "proc-macro2", "quote", "syn 2.0.118", @@ -961,6 +1092,12 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "ecdsa" version = "0.16.9" @@ -999,6 +1136,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "educe" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d7bc049e1bd8cdeb31b68bbd586a9464ecf9f3944af3958a7a9d0f8b9799417" +dependencies = [ + "enum-ordinalize", + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "either" version = "1.16.0" @@ -1026,6 +1175,44 @@ dependencies = [ "zeroize", ] +[[package]] +name = "encoding_rs" +version = "0.8.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "encoding_rs_io" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fba3fe847045ecff794b9c138293a80db914678c453ad63fbf0c6a9eb6e00b22" +dependencies = [ + "encoding_rs", +] + +[[package]] +name = "enum-ordinalize" +version = "4.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89dd01549b09589510cf0647475075d12071456586d70f5c75c98ae2a5537677" +dependencies = [ + "enum-ordinalize-derive", +] + +[[package]] +name = "enum-ordinalize-derive" +version = "4.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a65863d15a4ce2888bd2f0f543cc963d3879c3a022c8ee43f6141d479a3ac815" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -1042,6 +1229,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "fastrand" version = "2.5.0" @@ -1246,9 +1453,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 5.3.0", "wasip2", + "wasm-bindgen", ] [[package]] @@ -1280,6 +1489,28 @@ version = "0.32.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7" +[[package]] +name = "gloo-timers" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "granit-parser" +version = "0.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d03f81ad4732830d85cfd417a9f62cde6dadda4354d37d078a6084a19560aa2d" +dependencies = [ + "arraydeque", + "smallvec", +] + [[package]] name = "group" version = "0.13.0" @@ -1482,6 +1713,7 @@ dependencies = [ "http", "hyper", "hyper-util", + "log", "rustls", "rustls-native-certs", "rustls-platform-verifier", @@ -1490,6 +1722,19 @@ dependencies = [ "tower-service", ] +[[package]] +name = "hyper-timeout" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0" +dependencies = [ + "hyper", + "hyper-util", + "pin-project-lite", + "tokio", + "tower-service", +] + [[package]] name = "hyper-util" version = "0.1.20" @@ -1620,6 +1865,42 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jiff" +version = "0.2.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "668b7183bd07af9a4885f5c35b0cc5c83c4607a913c16b7e17291832910d2dcc" +dependencies = [ + "defmt", + "jiff-core", + "jiff-static", + "log", + "portable-atomic", + "portable-atomic-util", + "serde_core", +] + +[[package]] +name = "jiff-core" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7feca88439efe53da3754500c1851dedf3cb36c524dd5cf8225cc0794de95d09" +dependencies = [ + "defmt", +] + +[[package]] +name = "jiff-static" +version = "0.2.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a69dcb3a21cfb32ce1cd056169337ca284af0766dd766e7878819b251a49204" +dependencies = [ + "jiff-core", + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "jni" version = "0.22.4" @@ -1690,6 +1971,41 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "json-patch" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7421438de105a0827e44fadd05377727847d717c80ce29a229f85fd04c427b72" +dependencies = [ + "jsonptr", + "serde", + "serde_json", + "thiserror 2.0.18", +] + +[[package]] +name = "jsonpath-rust" +version = "1.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e07021eefc09f897611b067ba7897b5e9266edfc902768418968772e5bb06a7" +dependencies = [ + "pest", + "pest_derive", + "regex", + "serde_json", + "thiserror 2.0.18", +] + +[[package]] +name = "jsonptr" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5a3cc660ba5d72bce0b3bb295bf20847ccbb40fd423f3f05b61273672e561fe" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "jsonwebtoken" version = "11.0.0" @@ -1714,6 +2030,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "k8s-openapi" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c6922f6afe80418dd6019818af5d0d34584c371780ff09b9752370c25b4abb" +dependencies = [ + "base64 0.22.1", + "jiff", + "serde", + "serde_json", +] + [[package]] name = "kqueue" version = "1.2.0" @@ -1734,6 +2062,115 @@ dependencies = [ "libc", ] +[[package]] +name = "kube" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "208d7fe1380066abb194812a8a8e303cb896a33bb3d7b3177b71bd03ff39bf18" +dependencies = [ + "k8s-openapi", + "kube-client", + "kube-core", + "kube-derive", + "kube-runtime", +] + +[[package]] +name = "kube-client" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31e940a73033a7c5c7918b5ece7851d84571b58be25e937198da37fa13f116d3" +dependencies = [ + "base64 0.22.1", + "bytes", + "either", + "futures", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-timeout", + "hyper-util", + "jiff", + "jsonpath-rust", + "k8s-openapi", + "kube-core", + "pem", + "rustls", + "rustls-platform-verifier", + "secrecy", + "serde", + "serde-saphyr", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tower", + "tower-http 0.6.11", + "tracing", +] + +[[package]] +name = "kube-core" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9d2353c118cf3462c352ee0b5bd5b0cf17990af456dfd8662d136ff9812eeb4" +dependencies = [ + "derive_more", + "form_urlencoded", + "http", + "jiff", + "json-patch", + "k8s-openapi", + "schemars", + "serde", + "serde-value", + "serde_json", + "thiserror 2.0.18", +] + +[[package]] +name = "kube-derive" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92141c19e1fa83bf91633c1234c66b03375c29aa8ca5486331ddb47e3a3da9c0" +dependencies = [ + "darling 0.23.0", + "proc-macro2", + "quote", + "serde", + "serde_json", + "syn 2.0.118", +] + +[[package]] +name = "kube-runtime" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eb064ef71c7bea55e934a443960da9e9ec31fbb2ba63e47e4aeca32603d5cc7" +dependencies = [ + "ahash", + "async-broadcast", + "async-stream", + "backon", + "educe", + "futures", + "hashbrown 0.16.1", + "hostname", + "json-patch", + "k8s-openapi", + "kube-client", + "parking_lot", + "pin-project", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -2021,6 +2458,12 @@ dependencies = [ "libc", ] +[[package]] +name = "nohash-hasher" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2bf50223579dc7cdcfb3bfcacf7069ff68243f8c363f62ffa99cf000a6b9c451" + [[package]] name = "nom" version = "7.1.3" @@ -2278,6 +2721,12 @@ dependencies = [ "sha2 0.10.9", ] +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -2326,6 +2775,48 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pest" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a07a60cc7a4d00c91f95c685609d1d2f79050e6804b70ebedd7650f0b839bcf" +dependencies = [ + "memchr", + "ucd-trie", +] + +[[package]] +name = "pest_derive" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3a83744a5c8455b8b3e0dc5031362780a347c878bdd11584d1a8984228cc88d" +dependencies = [ + "pest", + "pest_generator", +] + +[[package]] +name = "pest_generator" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0cd3451aa3de60d4b9a1e736885e4dea6b31617598026f12256ad566d63304a" +dependencies = [ + "pest", + "pest_meta", + "proc-macro2", + "quote", + "syn 2.0.118", +] + +[[package]] +name = "pest_meta" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e04d3a0849e241d7dfce834c83b1c5edc8622009e8dd51a12ba1927c32f05496" +dependencies = [ + "pest", +] + [[package]] name = "petgraph" version = "0.7.1" @@ -2336,6 +2827,26 @@ dependencies = [ "indexmap 2.14.0", ] +[[package]] +name = "pin-project" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -2639,6 +3150,15 @@ version = "1.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" +[[package]] +name = "portable-atomic-util" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618" +dependencies = [ + "portable-atomic", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -2859,6 +3379,26 @@ dependencies = [ "bitflags 2.13.0", ] +[[package]] +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "regex" version = "1.13.0" @@ -3011,7 +3551,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" dependencies = [ "aws-lc-rs", + "log", "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", @@ -3117,6 +3659,31 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "schemars" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a" +dependencies = [ + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98c67716b46af2f0b8cf752abc930f6f9aecfbf671ecfb531db8a31dbe4e2ba" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 3.0.3", +] + [[package]] name = "scopeguard" version = "1.2.0" @@ -3146,6 +3713,15 @@ dependencies = [ "zeroize", ] +[[package]] +name = "secrecy" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e891af845473308773346dc847b2c23ee78fe442e0472ac50e22a18a93d3ae5a" +dependencies = [ + "zeroize", +] + [[package]] name = "security-framework" version = "3.7.0" @@ -3185,6 +3761,25 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde-saphyr" +version = "0.0.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7bd22781911de0ca6debda95f073c8f18bec65d1a94f1fa9573f3102e514cea4" +dependencies = [ + "ahash", + "annotate-snippets", + "base64 0.22.1", + "encoding_rs_io", + "getrandom 0.3.4", + "granit-parser", + "nohash-hasher", + "num-traits", + "serde_core", + "smallvec", + "zmij", +] + [[package]] name = "serde-value" version = "0.7.0" @@ -3215,6 +3810,17 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "serde_ignored" version = "0.1.14" @@ -3708,6 +4314,7 @@ dependencies = [ "futures-core", "futures-sink", "pin-project-lite", + "slab", "tokio", ] @@ -3722,6 +4329,26 @@ dependencies = [ "pin-project-lite", "sync_wrapper", "tokio", + "tokio-util", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "base64 0.22.1", + "bitflags 2.13.0", + "bytes", + "http", + "http-body", + "mime", + "pin-project-lite", + "tower", "tower-layer", "tower-service", "tracing", @@ -3772,9 +4399,21 @@ checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ "log", "pin-project-lite", + "tracing-attributes", "tracing-core", ] +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "tracing-core" version = "0.1.36" @@ -3824,6 +4463,12 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "ucd-trie" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" + [[package]] name = "unicase" version = "2.9.0" @@ -3842,6 +4487,12 @@ version = "1.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" +[[package]] +name = "unicode-width" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" + [[package]] name = "unicode-xid" version = "0.2.6" diff --git a/Cargo.toml b/Cargo.toml index 4bdec3b..f65ca99 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -58,3 +58,7 @@ hyper-util = "0.1.20" http-body-util = "0.1.4" bytes = "1.12.1" +rustls = { version = "0.23", default-features = false, features = ["ring"] } +kube = { version = "4.2.0", features = ["runtime", "client", "derive"] } +k8s-openapi = { version = "0.28.0", features = ["v1_32"] } +futures = "0.3" diff --git a/etc/kubernetes/aralez-certificate-secret.yml b/etc/kubernetes/aralez-certificate-secret.yml new file mode 100644 index 0000000..d9a8a3d --- /dev/null +++ b/etc/kubernetes/aralez-certificate-secret.yml @@ -0,0 +1,9 @@ +apiVersion: v1 +kind: Secret +metadata: + name: aralez-tls +type: kubernetes.io/tls +data: + tls.crt: + tls.key: + diff --git a/etc/kubernetes/aralez-example-deployment.yml b/etc/kubernetes/aralez-example-deployment.yml new file mode 100644 index 0000000..70f12b8 --- /dev/null +++ b/etc/kubernetes/aralez-example-deployment.yml @@ -0,0 +1,43 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: aralez +spec: + replicas: 1 + selector: + matchLabels: + app: aralez + template: + metadata: + labels: + app: aralez + spec: + serviceAccountName: aralez-sa + containers: + - name: aralez + image: sadoyan/aralez:latest + ports: + - containerPort: 80 + - containerPort: 443 + volumeMounts: + - name: main-config + mountPath: /etc/aralez/main.yaml + subPath: main.yaml + readOnly: true + - name: upstreams-config + mountPath: /etc/aralez/upstreams.yml + subPath: upstreams.yml + readOnly: true + - name: tls-certs + mountPath: /etc/aralez/certificates + readOnly: true + volumes: + - name: main-config + configMap: + name: aralez-main-config + - name: upstreams-config + configMap: + name: aralez-upstreams-config + - name: tls-certs + secret: + secretName: aralez-tls diff --git a/etc/kubernetes/aralez-example-main.yml b/etc/kubernetes/aralez-example-main.yml new file mode 100644 index 0000000..026e20d --- /dev/null +++ b/etc/kubernetes/aralez-example-main.yml @@ -0,0 +1,16 @@ +threads: 12 +daemon: false +upstream_keepalive_pool_size: 500 +pid_file: /tmp/aralez.pid +error_log: /tmp/aralez_err.log +upgrade_sock: /tmp/aralez.sock +config_api_enabled: false +config_address: 127.0.0.1:3000 +proxy_address_http: 0.0.0.0:80 +proxy_address_tls: 0.0.0.0:443 +proxy_configs: /etc/aralez +proxy_tls_grade: high +upstreams_conf: /etc/aralez/upstreams.yml +log_level: info +hc_method: HEAD +hc_interval: 2 diff --git a/etc/kubernetes/aralez-example-upstreams.yml b/etc/kubernetes/aralez-example-upstreams.yml new file mode 100644 index 0000000..0990d8b --- /dev/null +++ b/etc/kubernetes/aralez-example-upstreams.yml @@ -0,0 +1,14 @@ +# The file under watch and hot reload, changes are applied immediately, no need to restart or reload. +provider: "kubernetes" # "file" "consul" "kubernetes" +sticky_sessions: 172000 +to_https: false +rate_limit: 500000 +x4xx_limit: 100000 +server_headers: + - "X-Forwarded-Proto:https" + - "X-Forwarded-Port:443" +client_headers: + - "X-Global-Client:Yooooooo" +kubernetes: + servers: + - "127.0.0.1:6443" # Gets KUBERNETES_SERVICE_HOST : KUBERNETES_SERVICE_PORT_HTTPS env variables. diff --git a/etc/kubernetes/nginx-example-deployment.yml b/etc/kubernetes/nginx-example-deployment.yml new file mode 100644 index 0000000..54c3d58 --- /dev/null +++ b/etc/kubernetes/nginx-example-deployment.yml @@ -0,0 +1,21 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: svc-nginx + labels: + app: svc-nginx +spec: + selector: + matchLabels: + app: svc-nginx + replicas: 3 + template: + metadata: + labels: + app: svc-nginx + spec: + containers: + - name: svc-nginx + image: nginx:latest + ports: + - containerPort: 80 diff --git a/etc/kubernetes/nginx-example-ingress.yml b/etc/kubernetes/nginx-example-ingress.yml new file mode 100644 index 0000000..f16f017 --- /dev/null +++ b/etc/kubernetes/nginx-example-ingress.yml @@ -0,0 +1,24 @@ +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: wss-service-v2 + namespace: default + annotations: + # These custom annotations are parsed by Aralez + aralez.rs/rate_limit: "50" + aralez.rs/x4xx_limit: "10" + aralez.rs/client_headers: '["X-Some-Client:Some Custom Header", "X-Example-Client:An Example Header"]' + aralez.rs/server_headers: '["X-Some-Server:Some Custom Header", "X-Example-Server:An Example Header"]' +spec: + ingressClassName: aralez # Matches the IngressClass metadata + rules: + - host: wss-service-v2.blabla.com + http: + paths: + - path: /ws + pathType: Prefix + backend: + service: + name: svc-nginx + port: + number: 80 diff --git a/etc/kubernetes/nginx-example-service.yml b/etc/kubernetes/nginx-example-service.yml new file mode 100644 index 0000000..e2228e0 --- /dev/null +++ b/etc/kubernetes/nginx-example-service.yml @@ -0,0 +1,11 @@ +apiVersion: v1 +kind: Service +metadata: + name: svc-nginx +spec: + selector: + app: svc-nginx + ports: + - protocol: TCP + port: 80 + targetPort: 80 diff --git a/src/utils.rs b/src/utils.rs index 8c290cb..35b3bb4 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -8,6 +8,7 @@ pub mod healthcheck; pub mod httpclient; pub mod jwt; pub mod kuberconsul; +pub mod kubewatcher; pub mod lazylock; pub mod metrics; pub mod parceyaml; diff --git a/src/utils/healthcheck.rs b/src/utils/healthcheck.rs index 8b86c0d..58f38e4 100644 --- a/src/utils/healthcheck.rs +++ b/src/utils/healthcheck.rs @@ -7,7 +7,7 @@ use std::sync::atomic::AtomicUsize; use std::sync::Arc; use std::time::Duration; use tokio::net::TcpStream; -use tokio::time::interval; +use tokio::time::{interval, timeout}; use tokio_native_tls::native_tls::TlsConnector; pub async fn hc2(upslist: Arc, fullist: Arc, idlist: Arc, params: (&str, u64)) { @@ -86,12 +86,12 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD async fn detect_tls(ip: &str, port: &u16, sni: &str) -> bool { let addr = format!("{ip}:{port}"); - - let stream = match TcpStream::connect(&addr).await { - Ok(s) => s, + let timeout_duration = Duration::from_secs(2); // THINK !!! + let stream = match timeout(timeout_duration, TcpStream::connect(&addr)).await { + Ok(Ok(s)) => s, + Ok(Err(_)) => return false, Err(_) => return false, }; - let connector = tokio_native_tls::TlsConnector::from( TlsConnector::builder() .danger_accept_invalid_certs(true) diff --git a/src/utils/kuberconsul.rs b/src/utils/kuberconsul.rs index e523ca3..028abc0 100644 --- a/src/utils/kuberconsul.rs +++ b/src/utils/kuberconsul.rs @@ -1,9 +1,11 @@ use crate::utils::httpclient; +use crate::utils::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate}; use crate::utils::parceyaml::build_headers; use crate::utils::structs::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap}; use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams}; use async_trait::async_trait; use dashmap::DashMap; +use log::error; use pingora::prelude::sleep; use rand::RngExt; use serde::Deserialize; @@ -16,12 +18,16 @@ use std::sync::Arc; use std::time::Duration; use tokio::fs::File; use tokio::io::AsyncReadExt; + +use rustls::crypto::ring::default_provider; +use rustls::crypto::CryptoProvider; use tokio::sync::mpsc::Sender; #[derive(Debug, serde::Deserialize)] pub struct KubeEndpoints { pub subsets: Option>, } + #[derive(Debug, serde::Deserialize)] pub struct KubeSubset { pub addresses: Option>, @@ -51,31 +57,27 @@ pub struct ConsulTaggedAddress { #[serde(rename = "Port")] pub port: u16, } + #[allow(clippy::type_complexity)] pub fn list_to_upstreams(lt: Option, (Vec>, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &GlobalServiceMapping) { if let Some(list) = lt { - match upstreams.get(&*i.hostname.clone()) { + let key: Arc = Arc::from(i.hostname.as_str()); + match upstreams.get(&key) { Some(upstr) => { for (k, v) in list { - upstr.value().insert(k.to_owned(), v); + upstr.value().insert(k, v); } } None => { - upstreams.insert(Arc::from(i.hostname.clone()), list); + upstreams.insert(key, list); } }; } } pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap, (Vec>, AtomicUsize)>, values: Vec>) { - match conf.path { - Some(ref p) => { - upstreams.insert(Arc::from(p.clone()), (values, AtomicUsize::new(0))); - } - None => { - upstreams.insert(Arc::from("/"), (values, AtomicUsize::new(0))); - } - } + let path = conf.path.as_deref().unwrap_or("/"); + upstreams.insert(Arc::from(path), (values, AtomicUsize::new(0))); } async fn read_token(path: &str) -> String { @@ -84,6 +86,7 @@ async fn read_token(path: &str) -> String { file.read_to_string(&mut contents).await.unwrap(); contents.trim().to_string() } + #[async_trait] pub trait ServiceDiscovery { async fn fetch_upstreams(&self, config: Arc, toreturn: Sender); @@ -95,55 +98,104 @@ pub struct ConsulDiscovery; #[async_trait] impl ServiceDiscovery for KubernetesDiscovery { async fn fetch_upstreams(&self, config: Arc, toreturn: Sender) { - let prev_upstreams = UpstreamsDashMap::new(); - if let Some(kuber) = config.kubernetes.clone() { - let servers = kuber.servers.unwrap_or(vec![format!( - "{}:{}", - env::var("KUBERNETES_SERVICE_HOST").unwrap_or("0.0.0.0".to_string()), - env::var("KUBERNETES_SERVICE_PORT_HTTPS").unwrap_or("0".to_string()) - )]); + let servers = kuber.servers.unwrap_or_else(|| { + vec![format!( + "{}:{}", + env::var("KUBERNETES_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()), + env::var("KUBERNETES_SERVICE_PORT_HTTPS").unwrap_or_else(|_| "0".to_string()) + )] + }); let end = servers.len().saturating_sub(1); let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 }; let server = servers.get(num).unwrap().to_string(); - let path = kuber.tokenpath.unwrap_or("/var/run/secrets/kubernetes.io/serviceaccount/token".to_string()); + let path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string()); let namespace = get_current_namespace().unwrap_or_else(|| "default".to_string()); let token = read_token(path.as_str()).await; - loop { - let upstreams = UpstreamsDashMap::new(); - if let Some(kuber) = config.kubernetes.clone() { - if let Some(svc) = kuber.services { - for service in svc { - let header_list: DashMap, Vec<(String, Arc)>> = DashMap::new(); - let mut hl = Vec::new(); - build_headers(&service.client_headers, config.as_ref(), &mut hl); - if !hl.is_empty() { - match service.path.clone() { - Some(path) => { - header_list.insert(Arc::from(path.as_str()), hl); - } - None => { - header_list.insert(Arc::from("/"), hl); - } - } - // header_list.insert(Arc::from(path.as_str()), hl); - // header_list.insert(Arc::from(i.path).unwrap_or(Arc::from("/")).as_str(), hl); - config.client_headers.insert(Arc::from(service.hostname.clone()), header_list); - } - let url = format!("https://{}/api/v1/namespaces/{}/endpoints/{}", server, namespace, service.hostname); - // let url = format!("https://{}/api/v1/namespaces/{}/endpoints?labelSelector=app", server, namespace); - let list = httpclient::for_kuber(&url, &token, &service).await; - // println!("{:?}", list); - list_to_upstreams(list, &upstreams, &service); - } - } - if let Some(lt) = clone_compare(&upstreams, &prev_upstreams, &config).await { - toreturn.send(lt).await.unwrap(); - } + if CryptoProvider::get_default().is_none() { + default_provider().install_default().expect("Failed to install rustls crypto provider"); + } + + let (tx, mut rx) = tokio::sync::mpsc::channel::(100); + let srv = server.clone(); + let tok = token.clone(); + tokio::spawn(async move { + if let Err(e) = start_ingress_watcher_with_config(srv.as_str(), tok.as_str(), tx, "aralez").await { + error!("Fatal watcher error: {:?}", e); + } + }); + + let upstreams = UpstreamsDashMap::new(); + + while let Some(update) = rx.recv().await { + let host_key: Arc = Arc::from(update.host.as_str()); + + if update.is_deleted { + upstreams.remove(&host_key); + config.client_headers.remove(&host_key); + } else { + let service = GlobalServiceMapping { + upstream: update.service_name.clone(), + hostname: update.host.clone(), + path: Some(update.path.clone()), + to_https: None, + sticky_sessions: update.sticky_sessions, + rate_limit: update.rate_limit, + x4xx_limit: update.x4xx_limit, + client_headers: None, + server_headers: None, + }; + + let cheader_list: DashMap, Vec<(String, Arc)>> = DashMap::new(); + let sheader_list: DashMap, Vec<(String, Arc)>> = DashMap::new(); + + let mut client_headers_list = Vec::new(); + let mut server_headers_list = Vec::new(); + + let mut chl: Option> = None; + let mut shl: Option> = None; + + if let Some(hdr) = service.client_headers.as_ref() { + chl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned()); + } + if let Some(hdr) = service.server_headers.as_ref() { + shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned()); + } + + if let Some(ch) = update.client_headers { + chl.get_or_insert_with(Vec::new).extend(ch); + } + if let Some(hdr) = update.server_headers.as_ref() { + shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned()); + } + + build_headers(&chl, config.as_ref(), &mut client_headers_list); + if !client_headers_list.is_empty() { + let path_key = service.path.as_deref().unwrap_or("/"); + cheader_list.insert(Arc::from(path_key), client_headers_list); + config.client_headers.insert(host_key.clone(), cheader_list); + } + + build_headers(&shl, config.as_ref(), &mut server_headers_list); + if !server_headers_list.is_empty() { + let path_key = service.path.as_deref().unwrap_or("/"); + sheader_list.insert(Arc::from(path_key), server_headers_list); + config.server_headers.insert(host_key.clone(), sheader_list); + } + let url = format!("https://{}/api/v1/namespaces/{}/endpoints/{}", server, namespace, update.service_name); + let list = httpclient::for_kuber(&url, &token, &service).await; + + if list.is_none() { + error!("Endpoint URL: {} returned empty/invalid response", url); + } + + list_to_upstreams(list, &upstreams, &service); + } + if let Some(lt) = clone_compare(&upstreams, &config).await { + let _ = toreturn.send(lt).await; } - sleep(Duration::from_secs(5)).await; } } } @@ -156,61 +208,58 @@ fn get_current_namespace() -> Option { return Some(contents.trim().to_string()); } } - std::env::var("POD_NAMESPACE").ok() + env::var("POD_NAMESPACE").ok() } #[async_trait] impl ServiceDiscovery for ConsulDiscovery { async fn fetch_upstreams(&self, config: Arc, toreturn: Sender) { - let prev_upstreams = UpstreamsDashMap::new(); loop { let upstreams = UpstreamsDashMap::new(); if let Some(consul) = config.consul.clone() { - let servers = consul.servers.unwrap_or(vec![format!( - "{}:{}", - env::var("CONSUL_SERVICE_HOST").unwrap_or("0.0.0.0".to_string()), - env::var("CONSUL_SERVICE_PORT").unwrap_or("0".to_string()) - )]); + let servers = consul.servers.unwrap_or_else(|| { + vec![format!( + "{}:{}", + env::var("CONSUL_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()), + env::var("CONSUL_SERVICE_PORT").unwrap_or_else(|_| "0".to_string()) + )] + }); let end = servers.len().saturating_sub(1); let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 }; let consul_data = servers.get(num).unwrap().to_string(); - let ss = consul_data + "/v1/catalog/service/"; + let ss = format!("{}/v1/catalog/service/", consul_data); if let Some(svc) = consul.services { - for i in svc { + for i in &svc { let header_list = DashMap::new(); let mut hl = Vec::new(); build_headers(&i.client_headers, config.as_ref(), &mut hl); if !hl.is_empty() { - match i.path.clone() { - Some(path) => { - header_list.insert(Arc::from(path.as_str()), hl); - } - None => { - header_list.insert(Arc::from("/"), hl); - } - } - // header_list.insert(i.path.clone().unwrap_or("/".to_string()), hl); + let path_key = i.path.as_deref().unwrap_or("/"); + header_list.insert(Arc::from(path_key), hl); config.client_headers.insert(Arc::from(i.hostname.clone()), header_list); } - let pref = ss.clone() + &i.upstream; - let list = httpclient::for_consul(pref.as_str(), consul.token.clone(), &i).await; - list_to_upstreams(list, &upstreams, &i); + let pref = format!("{}{}", ss, i.upstream); + let list = httpclient::for_consul(pref.as_str(), consul.token.clone(), i).await; + list_to_upstreams(list, &upstreams, i); } } } - if let Some(lt) = clone_compare(&upstreams, &prev_upstreams, &config).await { - toreturn.send(lt).await.unwrap(); + + if let Some(lt) = clone_compare(&upstreams, &config).await { + let _ = toreturn.send(lt).await; } + sleep(Duration::from_secs(5)).await; } } } -async fn clone_compare(upstreams: &UpstreamsDashMap, prev_upstreams: &UpstreamsDashMap, config: &Arc) -> Option { - if !compare_dashmaps(upstreams, prev_upstreams) { - let tosend: Configuration = Configuration { + +async fn clone_compare(upstreams: &UpstreamsDashMap, config: &Arc) -> Option { + if !compare_dashmaps(upstreams, &config.upstreams) { + let tosend = Configuration { upstreams: Default::default(), client_headers: config.client_headers.clone(), server_headers: config.server_headers.clone(), @@ -219,10 +268,11 @@ async fn clone_compare(upstreams: &UpstreamsDashMap, prev_upstreams: &UpstreamsD typecfg: config.typecfg.clone(), extraparams: config.extraparams.clone(), }; - clone_dashmap_into(upstreams, prev_upstreams); + clone_dashmap_into(upstreams, &config.upstreams); clone_dashmap_into(upstreams, &tosend.upstreams); + print_upstreams(&tosend.upstreams, &tosend.extraparams); return Some(tosend); - }; + } None } diff --git a/src/utils/kubewatcher.rs b/src/utils/kubewatcher.rs new file mode 100644 index 0000000..b45db3e --- /dev/null +++ b/src/utils/kubewatcher.rs @@ -0,0 +1,271 @@ +use futures::StreamExt; +use hyper::Uri; +use k8s_openapi::api::core::v1::{Node, Service}; +use k8s_openapi::api::networking::v1::Ingress; +use kube::api::{Patch, PatchParams}; +use kube::{ + api::Api, + client::Client, + config::{AuthInfo, Config}, + runtime::watcher::{self, Event}, +}; +use log::{error, info}; +use serde_json::json; +use std::time::Duration; +use tokio::sync::mpsc; + +pub struct AralezRouteUpdate { + pub host: String, + pub path: String, + pub service_name: String, + pub is_deleted: bool, + pub rate_limit: Option, + pub x4xx_limit: Option, + pub sticky_sessions: Option, + pub client_headers: Option>, + pub server_headers: Option>, +} + +#[derive(Debug, Clone)] +struct AralezAnnotation { + rate_limit: Option, + x4xx_limit: Option, + sticky_sessions: Option, + pub client_headers: Option>, + pub server_headers: Option>, +} + +pub async fn start_ingress_watcher_with_config( + api_server: &str, + token: &str, + tx: mpsc::Sender, + target_class: &'static str, +) -> Result<(), Box> { + let client = create_custom_k8s_client(api_server, token, true).await?; + let ingresses: Api = Api::all(client.clone()); + + let watcher_config = watcher::Config::default().any_semantic(); + let mut stream = watcher::watcher(ingresses, watcher_config).boxed(); + + while let Some(event) = stream.next().await { + match event { + Ok(Event::Apply(ing)) => { + process_ingress_change(&client, &ing, target_class, false, &tx).await; + } + Ok(Event::Delete(ing)) => { + process_ingress_change(&client, &ing, target_class, true, &tx).await; + } + Ok(Event::InitApply(ing)) => { + process_ingress_change(&client, &ing, target_class, false, &tx).await; + } + Ok(Event::InitDone) => { + info!("Ingress watcher initial sync complete"); + } + Err(err) => { + error!("Kubernetes Ingress watch error: {:?}", err); + } + _ => {} + } + } + Ok(()) +} + +async fn process_ingress_change(client: &Client, ing: &Ingress, target_class: &str, is_deleted: bool, tx: &mpsc::Sender) { + let spec = match &ing.spec { + Some(s) => s, + None => return, + }; + let class_name = spec.ingress_class_name.as_deref().unwrap_or_default(); + if class_name != target_class { + return; + } + + if !is_deleted { + if let (Some(name), Some(ns)) = (ing.metadata.name.as_deref(), ing.metadata.namespace.as_deref()) { + let svc_name = std::env::var("ARALEZ_SERVICE_NAME").unwrap_or_else(|_| "aralez-service".to_string()); + let external_ip = match discover_controller_ip(&client, ns, &svc_name).await { + Some(ip) => ip, + None => discover_node_ip(&client) + .await + .unwrap_or_else(|| std::env::var("ARALEZ_EXTERNAL_IP").unwrap_or_else(|_| "127.0.0.1".to_string())), + }; + let client_clone = client.clone(); + let name_clone = name.to_string(); + let ns_clone = ns.to_string(); + let status = ing.status.clone(); + tokio::spawn(async move { + patch_ingress_status_address(&client_clone, &ns_clone, &name_clone, status.as_ref(), &external_ip).await; + }); + } + } + + if let Some(rules) = &spec.rules { + for rule in rules { + let host = match &rule.host { + Some(h) => h.clone(), + None => "*".to_string(), + }; + + if let Some(http) = &rule.http { + for path_entry in &http.paths { + let path = path_entry.path.clone().unwrap_or_else(|| "/".to_string()); + if let Some(service) = &path_entry.backend.service { + let service_name = service.name.clone(); + let mut update = AralezRouteUpdate { + host: host.clone(), + path, + service_name, + is_deleted, + rate_limit: None, + x4xx_limit: None, + sticky_sessions: None, + client_headers: None, + server_headers: None, + }; + if let Some(annot) = parse_ingress_config(ing) { + update.rate_limit = annot.rate_limit; + update.x4xx_limit = annot.x4xx_limit; + update.sticky_sessions = annot.sticky_sessions; + update.client_headers = annot.client_headers; + update.server_headers = annot.server_headers; + } + if let Err(e) = tx.send(update).await { + error!("Failed to send route update to channel: {:?}", e); + } + } + } + } + } + } +} + +pub async fn create_custom_k8s_client(api_server_url: &str, bearer_token: &str, insecure_tls: bool) -> Result> { + let full_url = if !api_server_url.starts_with("http://") && !api_server_url.starts_with("https://") { + format!("https://{}", api_server_url) + } else { + api_server_url.to_string() + }; + let uri: Uri = full_url.parse()?; + let mut config = Config::new(uri); + config.auth_info = AuthInfo { + token: Some(bearer_token.trim().to_string().into()), + ..Default::default() + }; + config.connect_timeout = Some(Duration::from_secs(10)); + config.read_timeout = None; + if insecure_tls { + config.accept_invalid_certs = true; + } + let client = Client::try_from(config)?; + Ok(client) +} + +fn parse_ingress_config(ing: &Ingress) -> Option { + if let Some(annotations) = &ing.metadata.annotations { + let mut clh: Option> = None; + let mut srh: Option> = None; + + if let Some(ch) = annotations.get("aralez.rs/client_headers") { + let data: Vec = serde_json::from_str(ch).unwrap_or_default(); + clh = Some(data); + } + if let Some(sh) = annotations.get("aralez.rs/server_headers") { + let data: Vec = serde_json::from_str(sh).unwrap_or_default(); + srh = Some(data); + }; + + let annots = AralezAnnotation { + rate_limit: annotations.get("aralez.rs/rate_limit").and_then(|v| v.parse().ok()), + x4xx_limit: annotations.get("aralez.rs/x4xx_limit").and_then(|v| v.parse().ok()), + sticky_sessions: annotations.get("aralez.rs/sticky_sessions").and_then(|v| v.parse().ok()), + client_headers: clh, + server_headers: srh, + }; + return Some(annots); + } + None +} + +pub async fn patch_ingress_status_address( + client: &Client, + namespace: &str, + name: &str, + current_status: Option<&k8s_openapi::api::networking::v1::IngressStatus>, + external_ip: &str, +) { + if let Some(status) = current_status { + if let Some(lb) = &status.load_balancer { + if let Some(ing_list) = &lb.ingress { + if ing_list.iter().any(|i| i.ip.as_deref() == Some(external_ip)) { + return; + } + } + } + } + + let ingresses: Api = Api::namespaced(client.clone(), namespace); + + let patch = json!({ + "apiVersion": "networking.k8s.io/v1", + "kind": "Ingress", + "status": { + "loadBalancer": { + "ingress": [ + { + "ip": external_ip + } + ] + } + } + }); + + let params = PatchParams::apply("aralez-ingress-controller").force(); + + match ingresses.patch_status(name, ¶ms, &Patch::Apply(patch)).await { + Ok(_) => info!("Patched Ingress {}/{} ADDRESS to {}", namespace, name, external_ip), + Err(e) => error!("Failed to patch status for Ingress {}/{}: {:?}", namespace, name, e), + } +} + +pub async fn discover_node_ip(client: &Client) -> Option { + let nodes: Api = Api::all(client.clone()); + if let Ok(node_list) = nodes.list(&Default::default()).await { + for node in node_list { + if let Some(status) = node.status { + if let Some(addresses) = status.addresses { + for addr in &addresses { + if addr.type_ == "ExternalIP" { + return Some(addr.address.clone()); + } + } + for addr in &addresses { + if addr.type_ == "InternalIP" { + return Some(addr.address.clone()); + } + } + } + } + } + } + None +} +pub async fn discover_controller_ip(client: &Client, namespace: &str, service_name: &str) -> Option { + let services: Api = Api::namespaced(client.clone(), namespace); + if let Ok(svc) = services.get(service_name).await { + if let Some(status) = svc.status { + if let Some(lb) = status.load_balancer { + if let Some(ingress_list) = lb.ingress { + for ing in ingress_list { + if let Some(ip) = ing.ip { + return Some(ip); + } + if let Some(hostname) = ing.hostname { + return Some(hostname); + } + } + } + } + } + } + None +}