diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index c270ed8..afcb177 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -49,6 +49,7 @@ jobs: uv run test/release/host_sessions.py /tmp/duckflight.duckdb_extension uv run test/release/oidc_health.py /tmp/duckflight.duckdb_extension uv run test/release/observability.py /tmp/duckflight.duckdb_extension + uv run test/release/client_regressions.py /tmp/duckflight.duckdb_extension --native-sigpipe - name: Push tested architecture if: github.ref == 'refs/heads/main' && github.event_name != 'pull_request' env: diff --git a/.github/workflows/helm.yml b/.github/workflows/helm.yml index f46b940..b22e921 100644 --- a/.github/workflows/helm.yml +++ b/.github/workflows/helm.yml @@ -38,7 +38,7 @@ jobs: strategy: fail-fast: false matrix: - catalog: [iceberg, ducklake] + catalog: [iceberg, ducklake, idle] steps: - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 with: @@ -64,10 +64,17 @@ jobs: docker build --tag duckflight:e2e . docker build --file test/kubernetes/Dockerfile --tag duckflight-e2e-client:local . - name: Helm install and real lake queries on k3s + if: matrix.catalog != 'idle' run: >- uv run --no-project test/kubernetes/run.py --catalog ${{ matrix.catalog }} --image duckflight:e2e --artifacts "$RUNNER_TEMP/e2e-artifacts" + - name: PostgreSQL and Flight idle sleep and wake on k3s + if: matrix.catalog == 'idle' + run: >- + uv run --no-project test/kubernetes/run_idle.py + --image duckflight:e2e --client-image duckflight-e2e-client:local + --artifacts "$RUNNER_TEMP/e2e-artifacts" - name: Upload failure diagnostics if: failure() uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 diff --git a/Cargo.lock b/Cargo.lock index a519de3..cbb7702 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -40,6 +40,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + [[package]] name = "arbitrary" version = "1.4.2" @@ -221,6 +227,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.5.1" @@ -279,6 +291,23 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + [[package]] name = "chrono" version = "0.4.45" @@ -321,12 +350,31 @@ dependencies = [ "tiny-keccak", ] +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "core-foundation-sys" version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + [[package]] name = "crc32fast" version = "1.5.1" @@ -399,6 +447,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" + [[package]] name = "derive_arbitrary" version = "1.4.2" @@ -410,6 +464,17 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "displaydoc" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "duckdb" version = "1.10506.0" @@ -456,6 +521,21 @@ dependencies = [ name = "duckflight-extension-abi" version = "0.1.0" +[[package]] +name = "duckflight-idle-proxy" +version = "0.1.0" +dependencies = [ + "anyhow", + "rcgen", + "reqwest", + "serde", + "serde_json", + "tempfile", + "tokio", + "tokio-rustls", + "uuid", +] + [[package]] name = "duckflight-mock-core" version = "0.1.0" @@ -536,6 +616,24 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + [[package]] name = "futures-core" version = "0.3.34" @@ -567,8 +665,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -579,10 +679,24 @@ checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 5.3.0", "wasip2", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "r-efi 6.0.0", + "rand_core", + "wasm-bindgen", +] + [[package]] name = "half" version = "2.7.1" @@ -635,12 +749,94 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + [[package]] name = "httparse" version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "hyper" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c3e324da4c95177d6291d4c8730197c0d1822f8a9766814a4a44fa5ab797c9c" +dependencies = [ + "atomic-waker", + "bytes", + "futures-core", + "http", + "http-body", + "httparse", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfa8e654703247911e29c23fbeaa261834bd9bb74efba2f9acddc37bfb127f53" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "rustls-native-certs", + "tokio", + "tokio-rustls", + "tower-service", +] + +[[package]] +name = "hyper-util" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" +dependencies = [ + "base64 0.23.1", + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "httparse", + "hyper", + "ipnet", + "libc", + "percent-encoding", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -665,12 +861,116 @@ dependencies = [ "cc", ] +[[package]] +name = "icu_collections" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0" + +[[package]] +name = "icu_properties" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" + +[[package]] +name = "icu_provider" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + [[package]] name = "ident_case" version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + [[package]] name = "indexmap" version = "2.14.1" @@ -681,6 +981,12 @@ dependencies = [ "hashbrown 0.17.1", ] +[[package]] +name = "ipnet" +version = "2.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" + [[package]] name = "itoa" version = "1.0.18" @@ -808,6 +1114,12 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + [[package]] name = "lock_api" version = "0.4.14" @@ -823,6 +1135,12 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "lru-slab" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" + [[package]] name = "memchr" version = "2.8.3" @@ -839,6 +1157,17 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + [[package]] name = "num" version = "0.4.3" @@ -872,6 +1201,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + [[package]] name = "num-integer" version = "0.1.47" @@ -918,6 +1253,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "openssl-probe" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" + [[package]] name = "parking_lot" version = "0.12.5" @@ -941,6 +1282,16 @@ dependencies = [ "windows-link", ] +[[package]] +name = "pem" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" +dependencies = [ + "base64 0.22.1", + "serde_core", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -959,6 +1310,21 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + +[[package]] +name = "powerfmt" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a6394b9e965e73d0a289ee54f589087e2c676aedf60885baf52c76b771e4958" + [[package]] name = "prettyplease" version = "0.2.37" @@ -978,6 +1344,62 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quinn" +version = "0.11.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4051e23e9185c255a7e33ef59cdbca87a22d359052eecd22fc6b901fb37d9d11" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e750cca55fe4f0439a15d0bb529da9651e79993e8e72c61a899a36d462befbe" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af66907df18639dcf4db56ca65490cabc4b27a97dbadd96f2926cca73298f016" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.47" @@ -994,29 +1416,74 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" [[package]] -name = "redox_syscall" -version = "0.5.18" +name = "r-efi" +version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" dependencies = [ - "bitflags", + "chacha20", + "getrandom 0.4.3", + "rand_core", ] [[package]] -name = "regex" -version = "1.13.1" +name = "rand_core" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" dependencies = [ - "aho-corasick", - "memchr", - "regex-automata", - "regex-syntax", + "rand_core", ] [[package]] -name = "regex-automata" -version = "0.4.18" +name = "rcgen" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "yasna", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "regex" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" dependencies = [ @@ -1031,6 +1498,44 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-native-certs", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "ring" version = "0.17.14" @@ -1045,6 +1550,12 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + [[package]] name = "rustix" version = "0.38.44" @@ -1086,12 +1597,25 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + [[package]] name = "rustls-pki-types" version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ + "web-time", "zeroize", ] @@ -1118,12 +1642,44 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "schannel" +version = "0.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "security-framework" +version = "3.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "serde" version = "1.0.229" @@ -1167,12 +1723,34 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "shlex" version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "simd-adler32" version = "0.3.10" @@ -1191,6 +1769,22 @@ version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "strsim" version = "0.11.1" @@ -1246,6 +1840,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] + +[[package]] +name = "synstructure" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "901704edd0dfe137f1987838ee4f259e4e063c31371bdb423f7ae38ec6f77f02" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "tar" version = "0.4.46" @@ -1264,12 +1878,51 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix 1.1.4", "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "time" +version = "0.3.55" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde_core", + "time-core", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + [[package]] name = "tiny-keccak" version = "2.0.2" @@ -1279,6 +1932,130 @@ dependencies = [ "crunchy", ] +[[package]] +name = "tinystr" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" +dependencies = [ + "displaydoc", + "zerovec", +] + +[[package]] +name = "tinyvec" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" + +[[package]] +name = "tokio" +version = "1.53.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044" +dependencies = [ + "bytes", + "libc", + "mio", + "parking_lot", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", +] + +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -1331,12 +2108,41 @@ dependencies = [ "log", ] +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + [[package]] name = "utf8-zero" version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e" +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + +[[package]] +name = "uuid" +version = "1.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cc1186384beb7dd8eedea376413fd654937285ea6c9cfbb928dc3043ea4b606" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "vcpkg" version = "0.2.15" @@ -1349,6 +2155,15 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "want" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec4cdd0dd910afe868b7ef477227d8d538b46b3075031afee8a9f2acb0a2ed0b" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -1377,6 +2192,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.77" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b7777d5cc23d0e91404e53ce2d5e8ec7acae3026b16233dba62cd3246457950" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.127" @@ -1409,6 +2234,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "web-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c435338968042f4f59a557f690a253676d47ce13ceb55d70100e7facf6620a30" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-roots" version = "1.0.9" @@ -1596,6 +2441,12 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + [[package]] name = "xattr" version = "1.6.1" @@ -1606,6 +2457,38 @@ dependencies = [ "rustix 1.1.4", ] +[[package]] +name = "yasna" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd" +dependencies = [ + "time", +] + +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", + "synstructure", +] + [[package]] name = "zerocopy" version = "0.8.56" @@ -1626,12 +2509,66 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f75b4683f6c7f45248d4d64056a24298c6281e0993356d7d1b4a1a962ef10d4a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", + "synstructure", +] + [[package]] name = "zeroize" version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0464e17806c1d976d5cba29399c7f08e516e279e2ba493f63123b5fca67dd8" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "zip" version = "6.0.0" diff --git a/Cargo.toml b/Cargo.toml index d6ad635..d638d91 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ publish = false crate-type = ["cdylib"] [workspace] -members = ["crates/duckflight-extension-abi", "test/mock_core"] +members = ["crates/duckflight-extension-abi", "crates/duckflight-idle-proxy", "test/mock_core"] default-members = ["."] resolver = "2" diff --git a/Dockerfile b/Dockerfile index 8a92e58..17a8878 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,6 @@ # syntax=docker/dockerfile:1 +ARG DUCKFLIGHT_CORE_TEST_IMAGE=scratch +FROM ${DUCKFLIGHT_CORE_TEST_IMAGE} AS test_core FROM ghcr.io/astral-sh/uv:0.8.22@sha256:9874eb7afe5ca16c363fe80b294fe700e460df29a55532bbfea234a0f12eddb1 AS uv FROM rust:1.90-bookworm@sha256:3914072ca0c3b8aad871db9169a651ccfce30cf58303e5d6f2db16d1d8a7e58f AS build COPY --from=uv /uv /usr/local/bin/uv @@ -16,9 +18,14 @@ RUN uv venv --python python3 configure/venv \ && uv pip install --python configure/venv/bin/python duckdb==1.5.6 packaging \ && configure/venv/bin/python extension-ci-tools/scripts/configure_helper.py --duckdb-platform \ && configure/venv/bin/python -c "import tomllib; from pathlib import Path; Path('configure/extension_version.txt').write_text(tomllib.loads(Path('Cargo.toml').read_text())['package']['version'])" +ARG DUCKFLIGHT_CORE_TEST_SHA256 RUN --mount=type=cache,target=/usr/local/cargo/registry \ --mount=type=cache,target=/usr/local/cargo/git \ - DUCKFLIGHT_CORE_BUNDLE_PATH="$(./scripts/fetch-core-release.sh)" \ + --mount=type=bind,from=test_core,target=/test-core \ + DUCKFLIGHT_CORE_BUNDLE_PATH="$(if test -f /test-core/libduckflight_core_ffi.so; then \ + test -n "$DUCKFLIGHT_CORE_TEST_SHA256" && \ + printf '%s /test-core/libduckflight_core_ffi.so\n' "$DUCKFLIGHT_CORE_TEST_SHA256" | sha256sum --check --status && \ + printf '%s' /test-core/libduckflight_core_ffi.so; else ./scripts/fetch-core-release.sh; fi)" \ DUCKDB_EXTENSION_NAME=duckflight DUCKDB_EXTENSION_MIN_DUCKDB_VERSION=v1.5.6 \ cargo build --locked --release \ && strip --strip-unneeded target/release/libduckflight.so \ @@ -26,6 +33,9 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \ -l target/release/libduckflight.so -o /build/duckflight.duckdb_extension \ -n duckflight -dv v1.5.6 -evf configure/extension_version.txt \ -pf configure/platform.txt --abi-type C_STRUCT_UNSTABLE +RUN --mount=type=cache,target=/usr/local/cargo/registry \ + --mount=type=cache,target=/usr/local/cargo/git \ + cargo build --locked --release -p duckflight-idle-proxy RUN set -eu; platform="$(cat configure/platform.txt)"; \ record="$(awk -v platform="$platform" '$1 == platform {print $2, $3}' docker/cli-assets.lock)"; \ test -n "$record"; set -- $record; \ @@ -34,9 +44,9 @@ RUN set -eu; platform="$(cat configure/platform.txt)"; \ unzip /tmp/duckdb.zip -d /build/cli \ && strip --strip-unneeded /build/cli/duckdb RUN curl --fail --location --retry 3 \ - https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.10/DUCKFLIGHT_CORE_BINARY_LICENSE.txt \ + https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.13/DUCKFLIGHT_CORE_BINARY_LICENSE.txt \ -o /build/DUCKFLIGHT_CORE_BINARY_LICENSE.txt \ - && echo 'facf1a4cbd9343500854cc5e617f502f4c3da24f76b3fcf7dfdbbd80d021cca4 /build/DUCKFLIGHT_CORE_BINARY_LICENSE.txt' | sha256sum --check - + && echo '006fb1abf934cb4acf0db7ef911befb299df0faf69e8a8e363d3b1619e4e545e /build/DUCKFLIGHT_CORE_BINARY_LICENSE.txt' | sha256sum --check - RUN curl --fail --location --retry 3 https://raw.githubusercontent.com/duckdb/duckdb/v1.5.6/LICENSE \ -o /build/DUCKDB_LICENSE.txt \ && echo '7e17fd31249fa875cb3b1c5e05c6c3e99b75509f6a2804ca176c217834de1dcb /build/DUCKDB_LICENSE.txt' | sha256sum --check - @@ -48,12 +58,13 @@ RUN mkdir -p /runtime/usr/local/bin /runtime/opt/duckflight /runtime/etc/ssl/cer /runtime/data /runtime/tmp \ && cc -Os -s -Wall -Wextra -Werror docker/server.c -o /runtime/usr/local/bin/duckflight-server \ && cp /build/cli/duckdb /runtime/usr/local/bin/ \ + && cp /build/target/release/duckflight-idle-proxy /runtime/usr/local/bin/ \ && cp /build/duckflight.duckdb_extension LICENSE DUCKFLIGHT_CORE_BINARY_LICENSE.txt DUCKDB_LICENSE.txt /runtime/opt/duckflight/ \ && cp /usr/share/doc/libc6/copyright /runtime/opt/duckflight/LIBC_COPYRIGHT \ && cp /usr/share/doc/libstdc++6/copyright /runtime/opt/duckflight/LIBSTDCXX_COPYRIGHT \ && cp /usr/share/doc/libgcc-s1/copyright /runtime/opt/duckflight/LIBGCC_COPYRIGHT \ && cp /etc/ssl/certs/ca-certificates.crt /runtime/etc/ssl/certs/ \ - && { ldd /build/cli/duckdb; ldd /build/duckflight.duckdb_extension; ldd /runtime/usr/local/bin/duckflight-server; } \ + && { ldd /build/cli/duckdb; ldd /build/duckflight.duckdb_extension; ldd /runtime/usr/local/bin/duckflight-server; ldd /runtime/usr/local/bin/duckflight-idle-proxy; } \ | awk '/=> \// {print $3} /^\t\// {print $1}' | sort -u \ | xargs -I '{}' cp --parents -L '{}' /runtime \ && cp --parents -L /lib/"$(cc -dumpmachine)"/librt.so.1 /runtime \ diff --git a/charts/duckflight/Chart.yaml b/charts/duckflight/Chart.yaml index a8f55e1..a65815b 100644 --- a/charts/duckflight/Chart.yaml +++ b/charts/duckflight/Chart.yaml @@ -2,6 +2,6 @@ apiVersion: v2 name: duckflight description: Stateless DuckFlight query workers for external lake catalogs type: application -version: 0.1.0 +version: 0.2.0 kubeVersion: ">=1.25.0-0" home: https://github.com/sidequery/duckflight-extension diff --git a/charts/duckflight/README.md b/charts/duckflight/README.md index d180a78..8fb2eb3 100644 --- a/charts/duckflight/README.md +++ b/charts/duckflight/README.md @@ -160,6 +160,72 @@ PostgreSQL's database qualifier follows the client's selected database; an unrelated startup database name such as `duckflight` is not an alias for qualified `lake.*` SQL. +## Optional idle sleep and wake-up + +Set `idleScale.enabled=true` to run a small, always-on TCP proxy and zero or one +query worker. PostgreSQL and Flight keep the same public Service, port numbers, +TLS identity, and authentication. The first protocol request starts the worker +and waits for bootstrap/readiness, then the proxy forwards the original bytes. +Allow for cold startup in client connection/RPC deadlines; the proxy's default +wait limit is 600 seconds. Connections without protocol data do not wake a worker. + +Use an image built from this chart's matching source revision. It must include +the wake-up proxy and idle-control protocol v1, available in core v0.1.13 onward. +Older images fail the opt-in worker readiness check; missing or invalid activity +evidence never authorizes scale-down. + +Create an existing Secret with a private `token` (at least 32 alphanumeric +characters) and `ca.crt` containing the worker's issuing CA or trusted self-signed +certificate. Generate/store it through your secret manager. Both worker and +proxy mount this Secret; the token must never be put in Helm values. The worker's +existing TLS certificate must include the public Service's short DNS name +(`lake-duckflight` in this example), which the proxy verifies when connecting to +the pod's HTTPS control endpoint. Query TLS continues to pass through unchanged, +including client certificates and PostgreSQL CancelRequest connections. + +```yaml +idleScale: + enabled: true + existingSecret: duckflight-idle + idleTimeoutSeconds: 3000 # 50 minutes after the worker becomes idle + coldStartTimeoutSeconds: 600 +``` + +The idle interval begins only after all protected activity is gone. Open TCP +sessions, even idle sessions, keep the worker awake. Detached Flight password +sessions, negotiated sessions, prepared statements, transactions, outstanding +result tickets, active response streams, and unfinished engine/cleanup workers +also prevent sleep. Close Flight sessions explicitly with `CloseSession`; closing +only a gRPC channel does not release server-owned sessions. Existing session and +transaction expiration settings are unchanged. Abandoned state with expiration +disabled can therefore keep a worker awake indefinitely. This is intentional: +the idle timeout never force-closes a session or interrupts a query. + +Readiness/startup probes use authenticated HTTPS control checks in this mode; +they do not create query sessions or reset the idle clock. Immediately before +scaling to zero, the core atomically fences both listeners and checks the +remaining activity. The proxy serializes sleep/wake and fences Kubernetes writes +using the Deployment's resource version, including recovery after timeouts or +proxy restarts. A new connection arriving during sleep waits for a fresh worker. +The proxy is a single replica with `Recreate` updates; do not attach an HPA or +another replica controller to either Deployment in this mode. + +`replicaCount` applies to normal mode; idle mode always owns zero or one worker. +The chart preserves that desired count on Helm upgrades. Enabling it on an +existing release with live normal-mode workers is rejected: install a fresh +release or deliberately drain and scale the old deployment to zero first. +Explicit rollouts, pod deletion, +node failures, and Kubernetes termination-grace limits remain separate lifecycle +events; the idle controller's safety guarantee applies to its automatic sleep. + +The chart installs a dedicated proxy ServiceAccount and a namespace Role limited +to reading/patching its worker Deployment and listing pods. The worker does not +receive Kubernetes API credentials. A NetworkPolicy restricts both worker +protocols and control traffic to the proxy pods; use a CNI that enforces it, and +route clients through the public Service. The private worker Service is not a +client endpoint. Worker memory/CPU requests disappear at sleep, but underlying VM +costs decrease only if your cluster also removes unused nodes. + ## Validate ```sh diff --git a/charts/duckflight/templates/_helpers.tpl b/charts/duckflight/templates/_helpers.tpl index a130ab6..5bc396b 100644 --- a/charts/duckflight/templates/_helpers.tpl +++ b/charts/duckflight/templates/_helpers.tpl @@ -2,6 +2,19 @@ {{- printf "%s-duckflight" .Release.Name | trunc 63 | trimSuffix "-" -}} {{- end -}} +{{- define "duckflight.proxyName" -}} +{{- printf "%s-idle-proxy" (include "duckflight.name" . | trunc 52 | trimSuffix "-") -}} +{{- end -}} + +{{- define "duckflight.workerService" -}} +{{- printf "%s-workers" (include "duckflight.name" . | trunc 55 | trimSuffix "-") -}} +{{- end -}} + +{{- define "duckflight.proxySelectorLabels" -}} +app.kubernetes.io/name: duckflight-idle-proxy +app.kubernetes.io/instance: {{ .Release.Name }} +{{- end -}} + {{- define "duckflight.selectorLabels" -}} app.kubernetes.io/name: duckflight app.kubernetes.io/instance: {{ .Release.Name }} diff --git a/charts/duckflight/templates/deployment.yaml b/charts/duckflight/templates/deployment.yaml index 20a1d08..4184305 100644 --- a/charts/duckflight/templates/deployment.yaml +++ b/charts/duckflight/templates/deployment.yaml @@ -7,8 +7,17 @@ {{- $auth := required "auth.existingSecret is required (TOML and TLS identity)" .Values.auth.existingSecret }} {{- $bootstrap := required "bootstrap.existingSecret is required (lake startup SQL)" .Values.bootstrap.existingSecret }} {{- $probePort := ternary "flight" "postgres" .Values.flight.enabled }} +{{- if .Values.idleScale.enabled }} +{{- $_ := required "idleScale.existingSecret is required (token and worker CA)" .Values.idleScale.existingSecret }} +{{- if or (eq (int .Values.idleScale.controlPort) 4090) (and .Values.postgres.enabled (eq (int .Values.idleScale.controlPort) (int .Values.postgres.port))) (and .Values.flight.enabled (eq (int .Values.idleScale.controlPort) (int .Values.flight.port))) }} +{{- fail "idleScale.controlPort must differ from enabled listeners and proxy health port 4090" }} +{{- end }} +{{- if or (and .Values.postgres.enabled (eq (int .Values.postgres.port) 4090)) (and .Values.flight.enabled (eq (int .Values.flight.port) 4090)) }} +{{- fail "idleScale listeners must differ from proxy health port 4090" }} +{{- end }} +{{- end }} {{- range .Values.extraEnv }} -{{- if has .name (list "DUCKFLIGHT_DATABASE" "DUCKFLIGHT_CONFIG" "DUCKFLIGHT_INIT_SQL" "DUCKFLIGHT_PG_ADDRESS" "DUCKFLIGHT_FLIGHT_ADDRESS" "DUCKFLIGHT_MEMORY_LIMIT" "DUCKFLIGHT_THREADS" "DUCKFLIGHT_TEMP_DIRECTORY" "HOME") }} +{{- if has .name (list "DUCKFLIGHT_DATABASE" "DUCKFLIGHT_CONFIG" "DUCKFLIGHT_INIT_SQL" "DUCKFLIGHT_PG_ADDRESS" "DUCKFLIGHT_FLIGHT_ADDRESS" "DUCKFLIGHT_MEMORY_LIMIT" "DUCKFLIGHT_THREADS" "DUCKFLIGHT_TEMP_DIRECTORY" "DUCKFLIGHT_IDLE_CONTROL_ADDRESS" "DUCKFLIGHT_IDLE_TOKEN_FILE" "DUCKFLIGHT_IDLE_EXPECTED_LISTENERS" "DUCKFLIGHT_IDLE_SERVER_NAME" "HOME") }} {{- fail (printf "extraEnv must not override chart-managed variable %s" .name) }} {{- end }} {{- end }} @@ -16,10 +25,27 @@ apiVersion: apps/v1 kind: Deployment metadata: name: {{ include "duckflight.name" . }} + {{- if .Values.idleScale.enabled }} + annotations: + duckflight.sidequery.dev/idle-scale: "true" + {{- end }} labels: {{- include "duckflight.labels" . | nindent 4 }} spec: + {{- if .Values.idleScale.enabled }} + {{- $existing := lookup "apps/v1" "Deployment" .Release.Namespace (include "duckflight.name" .) }} + {{- if and $existing (gt (int $existing.spec.replicas) 0) (ne (dig "annotations" "duckflight.sidequery.dev/idle-scale" "false" $existing.metadata) "true") }} + {{- fail "enabling idleScale on live normal-mode workers requires draining them first; use a fresh release" }} + {{- end }} + {{- if and $existing (gt (int $existing.spec.replicas) 1) }} + {{- fail "enabling idleScale on multiple live workers requires draining them first; use a fresh release" }} + {{- end }} + # Preserve the controller's desired count across Helm upgrades; cold installs + # have zero workers. This mode deliberately owns exactly one awake worker. + replicas: {{ if $existing }}{{ $existing.spec.replicas }}{{ else }}0{{ end }} + {{- else }} replicas: {{ .Values.replicaCount }} + {{- end }} strategy: type: RollingUpdate rollingUpdate: @@ -77,10 +103,24 @@ spec: value: {{ .Values.duckdb.threads | quote }} - name: DUCKFLIGHT_TEMP_DIRECTORY value: /data/spill + {{- if .Values.idleScale.enabled }} + - name: DUCKFLIGHT_IDLE_CONTROL_ADDRESS + value: {{ printf "0.0.0.0:%v" .Values.idleScale.controlPort | quote }} + - name: DUCKFLIGHT_IDLE_TOKEN_FILE + value: /run/idle-secret/token + - name: DUCKFLIGHT_IDLE_EXPECTED_LISTENERS + value: {{ add (ternary 1 0 .Values.postgres.enabled) (ternary 1 0 .Values.flight.enabled) | quote }} + - name: DUCKFLIGHT_IDLE_SERVER_NAME + value: {{ include "duckflight.name" . | quote }} + {{- end }} {{- with .Values.extraEnv }} {{- toYaml . | nindent 12 }} {{- end }} ports: + {{- if .Values.idleScale.enabled }} + - name: idle-control + containerPort: {{ .Values.idleScale.controlPort }} + {{- end }} {{- if .Values.postgres.enabled }} - name: postgres containerPort: {{ .Values.postgres.port }} @@ -92,16 +132,31 @@ spec: # Flight starts last when both protocols are enabled. Kubernetes' native # gRPC probe cannot negotiate this listener's required TLS transport. startupProbe: + {{- if .Values.idleScale.enabled }} + exec: + command: [/usr/local/bin/duckflight-idle-proxy, worker-ready] + {{- else }} tcpSocket: port: {{ $probePort }} + {{- end }} {{- toYaml .Values.startupProbe | nindent 12 }} readinessProbe: + {{- if .Values.idleScale.enabled }} + exec: + command: [/usr/local/bin/duckflight-idle-proxy, worker-ready] + {{- else }} tcpSocket: port: {{ $probePort }} + {{- end }} {{- toYaml .Values.readinessProbe | nindent 12 }} resources: {{- toYaml .Values.resources | nindent 12 }} volumeMounts: + {{- if .Values.idleScale.enabled }} + - name: idle-secret + mountPath: /run/idle-secret + readOnly: true + {{- end }} - name: data mountPath: /data - name: tmp @@ -113,6 +168,17 @@ spec: mountPath: /run/bootstrap readOnly: true volumes: + {{- if .Values.idleScale.enabled }} + - name: idle-secret + secret: + secretName: {{ .Values.idleScale.existingSecret }} + defaultMode: 0440 + items: + - key: token + path: token + - key: ca.crt + path: ca.crt + {{- end }} - name: data emptyDir: sizeLimit: {{ .Values.spill.sizeLimit }} diff --git a/charts/duckflight/templates/idle-scale.yaml b/charts/duckflight/templates/idle-scale.yaml new file mode 100644 index 0000000..cd9a2ea --- /dev/null +++ b/charts/duckflight/templates/idle-scale.yaml @@ -0,0 +1,190 @@ +{{- if .Values.idleScale.enabled }} +apiVersion: v1 +kind: ServiceAccount +metadata: + name: {{ include "duckflight.proxyName" . }} + labels: + {{- include "duckflight.proxySelectorLabels" . | nindent 4 }} +automountServiceAccountToken: true +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: {{ include "duckflight.proxyName" . }} +rules: + - apiGroups: [apps] + resources: [deployments] + resourceNames: [{{ include "duckflight.name" . | quote }}] + verbs: [get, patch] + - apiGroups: [""] + resources: [pods] + verbs: [list] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: {{ include "duckflight.proxyName" . }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: {{ include "duckflight.proxyName" . }} +subjects: + - kind: ServiceAccount + name: {{ include "duckflight.proxyName" . }} + namespace: {{ .Release.Namespace }} +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "duckflight.proxyName" . }} +spec: + replicas: 1 + # A single controller owns sleep/wake decisions. Resource-version fences also + # prevent a delayed old-controller scale request from racing a new wake-up. + strategy: + type: Recreate + selector: + matchLabels: + {{- include "duckflight.proxySelectorLabels" . | nindent 6 }} + template: + metadata: + labels: + {{- include "duckflight.proxySelectorLabels" . | nindent 8 }} + spec: + serviceAccountName: {{ include "duckflight.proxyName" . }} + automountServiceAccountToken: true + terminationGracePeriodSeconds: {{ .Values.idleScale.proxy.terminationGracePeriodSeconds }} + securityContext: + runAsNonRoot: true + runAsUser: 10001 + runAsGroup: 10001 + fsGroup: 10001 + seccompProfile: + type: RuntimeDefault + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: proxy + image: {{ include "duckflight.image" . | quote }} + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: [/usr/local/bin/duckflight-idle-proxy] + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: [ALL] + env: + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: DUCKFLIGHT_PROXY_DEPLOYMENT + value: {{ include "duckflight.name" . | quote }} + - name: DUCKFLIGHT_PROXY_SELECTOR + value: {{ printf "app.kubernetes.io/name=duckflight,app.kubernetes.io/instance=%s" .Release.Name | quote }} + - name: DUCKFLIGHT_PROXY_HOSTNAME + value: {{ include "duckflight.name" . | quote }} + - name: DUCKFLIGHT_PROXY_CONTROL_PORT + value: {{ .Values.idleScale.controlPort | quote }} + - name: DUCKFLIGHT_PROXY_IDLE_TIMEOUT_SECONDS + value: {{ .Values.idleScale.idleTimeoutSeconds | quote }} + - name: DUCKFLIGHT_PROXY_COLD_START_TIMEOUT_SECONDS + value: {{ .Values.idleScale.coldStartTimeoutSeconds | quote }} + - name: DUCKFLIGHT_PROXY_TOKEN_FILE + value: /run/idle-secret/token + - name: DUCKFLIGHT_PROXY_CA_FILE + value: /run/idle-secret/ca.crt + {{- if .Values.postgres.enabled }} + - name: DUCKFLIGHT_PROXY_POSTGRES_PORT + value: {{ .Values.postgres.port | quote }} + {{- end }} + {{- if .Values.flight.enabled }} + - name: DUCKFLIGHT_PROXY_FLIGHT_PORT + value: {{ .Values.flight.port | quote }} + {{- end }} + ports: + - name: health + containerPort: 4090 + {{- if .Values.postgres.enabled }} + - name: postgres + containerPort: {{ .Values.postgres.port }} + {{- end }} + {{- if .Values.flight.enabled }} + - name: flight + containerPort: {{ .Values.flight.port }} + {{- end }} + startupProbe: + httpGet: {path: /healthz, port: health} + periodSeconds: 2 + failureThreshold: 60 + readinessProbe: + httpGet: {path: /healthz, port: health} + periodSeconds: 5 + resources: + {{- toYaml .Values.idleScale.proxy.resources | nindent 12 }} + volumeMounts: + - name: idle-secret + mountPath: /run/idle-secret + readOnly: true + volumes: + - name: idle-secret + secret: + secretName: {{ .Values.idleScale.existingSecret }} + defaultMode: 0440 + items: + - key: token + path: token + - key: ca.crt + path: ca.crt +--- +apiVersion: v1 +kind: Service +metadata: + name: {{ include "duckflight.workerService" . }} +spec: + type: ClusterIP + selector: + {{- include "duckflight.selectorLabels" . | nindent 4 }} + ports: + - name: idle-control + port: {{ .Values.idleScale.controlPort }} + targetPort: idle-control + {{- if .Values.postgres.enabled }} + - name: postgres + port: {{ .Values.postgres.port }} + targetPort: postgres + {{- end }} + {{- if .Values.flight.enabled }} + - name: flight + port: {{ .Values.flight.port }} + targetPort: flight + {{- end }} +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: {{ include "duckflight.workerService" . }} +spec: + podSelector: + matchLabels: + {{- include "duckflight.selectorLabels" . | nindent 6 }} + policyTypes: [Ingress] + ingress: + - from: + - podSelector: + matchLabels: + {{- include "duckflight.proxySelectorLabels" . | nindent 14 }} + ports: + - protocol: TCP + port: {{ .Values.idleScale.controlPort }} + {{- if .Values.postgres.enabled }} + - protocol: TCP + port: {{ .Values.postgres.port }} + {{- end }} + {{- if .Values.flight.enabled }} + - protocol: TCP + port: {{ .Values.flight.port }} + {{- end }} +{{- end }} diff --git a/charts/duckflight/templates/service.yaml b/charts/duckflight/templates/service.yaml index dfc251c..4a9c5b6 100644 --- a/charts/duckflight/templates/service.yaml +++ b/charts/duckflight/templates/service.yaml @@ -11,7 +11,11 @@ metadata: spec: type: {{ .Values.service.type }} selector: + {{- if .Values.idleScale.enabled }} + {{- include "duckflight.proxySelectorLabels" . | nindent 4 }} + {{- else }} {{- include "duckflight.selectorLabels" . | nindent 4 }} + {{- end }} ports: {{- if .Values.postgres.enabled }} - name: postgres diff --git a/charts/duckflight/values.schema.json b/charts/duckflight/values.schema.json index d699eea..793f441 100644 --- a/charts/duckflight/values.schema.json +++ b/charts/duckflight/values.schema.json @@ -2,6 +2,22 @@ "$schema": "https://json-schema.org/draft-07/schema#", "type": "object", "properties": { + "idleScale": { + "type": "object", + "properties": { + "enabled": {"type": "boolean"}, + "idleTimeoutSeconds": {"type": "integer", "minimum": 1}, + "coldStartTimeoutSeconds": {"type": "integer", "minimum": 1}, + "controlPort": {"type": "integer", "minimum": 1024, "maximum": 65535}, + "existingSecret": {"type": "string"}, + "proxy": { + "type": "object", + "properties": { + "terminationGracePeriodSeconds": {"type": "integer", "minimum": 1} + } + } + } + }, "replicaCount": {"type": "integer", "minimum": 1}, "postgres": {"$ref": "#/definitions/protocol"}, "flight": {"$ref": "#/definitions/protocol"}, diff --git a/charts/duckflight/values.yaml b/charts/duckflight/values.yaml index 44d3c98..c3ff32d 100644 --- a/charts/duckflight/values.yaml +++ b/charts/duckflight/values.yaml @@ -1,5 +1,22 @@ # Each pod is an independent query engine; no shared DuckDB files or PVCs. replicaCount: 2 +# Optional single-worker sleep/wake mode. The public Service then targets a +# small, always-on TCP proxy. Open connections and core-owned state block sleep. +idleScale: + enabled: false + idleTimeoutSeconds: 3000 + coldStartTimeoutSeconds: 600 + controlPort: 4091 + # Existing Secret: token (>=32 alphanumeric characters), ca.crt (worker CA). + existingSecret: "" + proxy: + resources: + requests: + cpu: 25m + memory: 64Mi + limits: + memory: 128Mi + terminationGracePeriodSeconds: 3600 image: repository: ghcr.io/sidequery/duckflight-extension # Set a tested commit tag or digest that includes Iceberg's librt dependency. diff --git a/core-assets.lock b/core-assets.lock index 54e2c2a..be9c7f9 100644 --- a/core-assets.lock +++ b/core-assets.lock @@ -1,7 +1,7 @@ # platform sha256 immutable_url # # Each immutable release asset is fetched, checksum-verified, and embedded by `make release`. -linux_amd64 6e8793e655dc243f362fdcbe1eece2730645bf5dfad4a4fceffaaaa3ee9e548f https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.12/libduckflight_core_ffi-linux_amd64.so -linux_arm64 dd770cec2b2f726fdd1263b65aa77c6ed6276ea590f894be0336e5df7d310eab https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.12/libduckflight_core_ffi-linux_arm64.so -osx_amd64 b0299e2a52b4b81c7f7d499d1f8da26d53cbb5e8d920c8199a3efc17b82c2726 https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.12/libduckflight_core_ffi-osx_amd64.dylib -osx_arm64 9545539bfd93d123536bd9b1c13d6186fc00feae53c1ba62c746e14bbb6deade https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.12/libduckflight_core_ffi-osx_arm64.dylib +linux_amd64 e07a2832f502a7ef69e12f5b2cb2ab935e2cdbee4f2479e4b243785eab9ce5c0 https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.13/libduckflight_core_ffi-linux_amd64.so +linux_arm64 1bf70395a07ad385edacd9a44ff7e34b0f454141d56b71f124bb9af45a338265 https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.13/libduckflight_core_ffi-linux_arm64.so +osx_amd64 c74488c8bac1fb5a55d310d1238b3a26ac99467715a1721ce5e16e73576f74ae https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.13/libduckflight_core_ffi-osx_amd64.dylib +osx_arm64 24f8e5e9deeed925d59d084f1ab4bea5172504b2c8a93dcf0efa2b4b3a70ea56 https://github.com/sidequery/duckflight-extension/releases/download/core-v0.1.13/libduckflight_core_ffi-osx_arm64.dylib diff --git a/crates/duckflight-idle-proxy/Cargo.toml b/crates/duckflight-idle-proxy/Cargo.toml new file mode 100644 index 0000000..99a0310 --- /dev/null +++ b/crates/duckflight-idle-proxy/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "duckflight-idle-proxy" +version = "0.1.0" +edition = "2024" +rust-version = "1.90" +publish = false + +[dependencies] +anyhow = "1" +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls-native-roots"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["full"] } +uuid = { version = "1", features = ["v4"] } + +[dev-dependencies] +rcgen = "0.13" +tempfile = "3" +tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"] } diff --git a/crates/duckflight-idle-proxy/src/main.rs b/crates/duckflight-idle-proxy/src/main.rs new file mode 100644 index 0000000..e4fc6a9 --- /dev/null +++ b/crates/duckflight-idle-proxy/src/main.rs @@ -0,0 +1,603 @@ +//! A single-owner, TLS-pass-through wake proxy for PostgreSQL and Flight SQL. +//! Sleep requires a core-issued admission fence, not merely zero TCP sockets. +use anyhow::{Context, Result, bail}; +use reqwest::{Certificate, Client, Method}; +use serde::Deserialize; +use serde_json::{Value, json}; +use std::{env, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + sync::Mutex, + task::JoinSet, + time::{Instant, sleep, timeout}, +}; + +const OWNER_ANNOTATION: &str = "duckflight.sidequery.dev/idle-owner"; +const OPERATION_ANNOTATION: &str = "duckflight.sidequery.dev/idle-operation"; + +struct Config { + namespace: String, + deployment: String, + selector: String, + hostname: String, + control_port: u16, + idle_timeout: Duration, + cold_start_timeout: Duration, + token_file: PathBuf, + ca: Certificate, + listeners: Vec, +} + +impl Config { + fn from_env() -> Result { + let mut listeners = Vec::new(); + for name in [ + "DUCKFLIGHT_PROXY_POSTGRES_PORT", + "DUCKFLIGHT_PROXY_FLIGHT_PORT", + ] { + if let Ok(value) = env::var(name) { + listeners.push(value.parse()?); + } + } + if listeners.is_empty() { + bail!("enable at least one proxy listener") + } + Ok(Self { + namespace: env::var("POD_NAMESPACE")?, + deployment: env::var("DUCKFLIGHT_PROXY_DEPLOYMENT")?, + selector: env::var("DUCKFLIGHT_PROXY_SELECTOR")?, + hostname: env::var("DUCKFLIGHT_PROXY_HOSTNAME")?, + control_port: env::var("DUCKFLIGHT_PROXY_CONTROL_PORT")?.parse()?, + idle_timeout: Duration::from_secs( + env::var("DUCKFLIGHT_PROXY_IDLE_TIMEOUT_SECONDS")?.parse()?, + ), + cold_start_timeout: Duration::from_secs( + env::var("DUCKFLIGHT_PROXY_COLD_START_TIMEOUT_SECONDS")?.parse()?, + ), + token_file: env::var("DUCKFLIGHT_PROXY_TOKEN_FILE")?.into(), + ca: Certificate::from_pem(&std::fs::read(env::var("DUCKFLIGHT_PROXY_CA_FILE")?)?)?, + listeners, + }) + } +} + +struct Kubernetes { + client: Client, + base: String, + token_file: PathBuf, + deployment_path: String, + pods_path: String, + selector: String, +} + +impl Kubernetes { + fn new(config: &Config) -> Result { + let directory = PathBuf::from("/var/run/secrets/kubernetes.io/serviceaccount"); + let ca = Certificate::from_pem(&std::fs::read(directory.join("ca.crt"))?)?; + let host = env::var("KUBERNETES_SERVICE_HOST")?; + let port = env::var("KUBERNETES_SERVICE_PORT_HTTPS").unwrap_or_else(|_| "443".into()); + let host = if host.contains(':') { + format!("[{host}]") + } else { + host + }; + Ok(Self { + client: Client::builder() + .add_root_certificate(ca) + .timeout(Duration::from_secs(15)) + .build()?, + base: format!("https://{host}:{port}"), + token_file: directory.join("token"), + deployment_path: format!( + "/apis/apps/v1/namespaces/{}/deployments/{}", + config.namespace, config.deployment + ), + pods_path: format!("/api/v1/namespaces/{}/pods", config.namespace), + selector: config.selector.clone(), + }) + } + + async fn deployment(&self) -> Result { + Ok(self + .client + .get(format!("{}{}", self.base, self.deployment_path)) + .bearer_auth(std::fs::read_to_string(&self.token_file)?.trim()) + .send() + .await? + .error_for_status()? + .json() + .await?) + } + + async fn patch( + &self, + current: &Value, + owner: &str, + operation: &str, + replicas: Option, + ) -> Result { + let mut patch = json!({"metadata": { + "resourceVersion": current["metadata"]["resourceVersion"], + "uid": current["metadata"]["uid"], + "annotations": {OWNER_ANNOTATION: owner, OPERATION_ANNOTATION: operation} + }}); + if let Some(replicas) = replicas { + patch["spec"] = json!({"replicas": replicas}); + } + Ok(self + .client + .patch(format!("{}{}", self.base, self.deployment_path)) + .header("Content-Type", "application/merge-patch+json") + .bearer_auth(std::fs::read_to_string(&self.token_file)?.trim()) + .json(&patch) + .send() + .await? + .error_for_status()? + .json() + .await?) + } + + async fn pods(&self) -> Result> { + let result: Value = self + .client + .get(format!("{}{}", self.base, self.pods_path)) + .query(&[("labelSelector", &self.selector)]) + .bearer_auth(std::fs::read_to_string(&self.token_file)?.trim()) + .send() + .await? + .error_for_status()? + .json() + .await?; + Ok(result["items"] + .as_array() + .context("missing pod list")? + .clone()) + } +} + +#[derive(Clone)] +struct Backend { + uid: String, + ip: std::net::IpAddr, + control: Client, + url: String, +} + +#[derive(Debug, Deserialize)] +struct IdleStatus { + protocol_version: u32, + boot_id: String, + ready: bool, + busy: bool, + permit: Option, +} + +impl Backend { + fn from_pod(pod: &Value, config: &Config) -> Result { + let ip = pod["status"]["podIP"] + .as_str() + .context("missing pod IP")? + .parse()?; + Ok(Self { + uid: pod["metadata"]["uid"] + .as_str() + .context("missing pod UID")? + .to_string(), + ip, + control: Client::builder() + .add_root_certificate(config.ca.clone()) + .https_only(true) + .no_proxy() + .resolve(&config.hostname, SocketAddr::new(ip, config.control_port)) + .timeout(Duration::from_secs(10)) + .build()?, + url: format!("https://{}:{}", config.hostname, config.control_port), + }) + } + + async fn request( + &self, + method: Method, + path: &str, + config: &Config, + ) -> Result { + Ok(self + .control + .request(method, format!("{}{path}", self.url)) + .bearer_auth(std::fs::read_to_string(&config.token_file)?.trim()) + .send() + .await? + .error_for_status()?) + } + + async fn status(&self, suspend: bool, config: &Config) -> Result { + let status: IdleStatus = self + .request( + if suspend { Method::POST } else { Method::GET }, + if suspend { "/suspend" } else { "/status" }, + config, + ) + .await? + .json() + .await?; + if status.protocol_version != 1 || status.boot_id.is_empty() { + bail!("unsupported idle-control capability") + } + Ok(status) + } + + async fn resume(&self, permit: &str, config: &Config) -> Result<()> { + self.request(Method::POST, &format!("/resume/{permit}"), config) + .await?; + Ok(()) + } +} + +fn ready(pod: &Value) -> bool { + pod["metadata"]["deletionTimestamp"].is_null() + && pod["status"]["conditions"] + .as_array() + .is_some_and(|conditions| { + conditions + .iter() + .any(|condition| condition["type"] == "Ready" && condition["status"] == "True") + }) +} + +struct Controller { + config: Config, + kube: Kubernetes, + owner: String, + uid: String, + // A timed-out write may still be in flight. A resourceVersion-changing + // fence must succeed before any suspended worker is resumed or reused. + needs_fence: bool, + connections: usize, + idle_since: Option<(String, String, Instant)>, +} + +impl Controller { + async fn new(config: Config) -> Result { + let kube = Kubernetes::new(&config)?; + let current = kube.deployment().await?; + let uid = current["metadata"]["uid"] + .as_str() + .context("missing deployment UID")? + .to_string(); + let mut controller = Self { + config, + kube, + owner: uuid::Uuid::new_v4().to_string(), + uid, + needs_fence: true, + connections: 0, + idle_since: None, + }; + controller.fence().await?; + Ok(controller) + } + + async fn current(&self) -> Result { + let current = self.kube.deployment().await?; + if current["metadata"]["uid"] != self.uid { + bail!("worker Deployment was replaced; restart proxy") + } + if !self.needs_fence && current["metadata"]["annotations"][OWNER_ANNOTATION] != self.owner { + bail!("another controller owns this worker Deployment; restart proxy") + } + Ok(current) + } + + async fn fence(&mut self) -> Result<()> { + let current = self.current().await?; + self.needs_fence = true; + self.kube + .patch( + ¤t, + &self.owner, + &uuid::Uuid::new_v4().to_string(), + None, + ) + .await?; + self.needs_fence = false; + Ok(()) + } + + async fn scale(&mut self, current: &Value, replicas: u32) -> Result<()> { + self.needs_fence = true; + self.kube + .patch( + current, + &self.owner, + &uuid::Uuid::new_v4().to_string(), + Some(replicas), + ) + .await?; + self.needs_fence = false; + Ok(()) + } + + async fn ready_backend(&self) -> Result> { + let pods = self.kube.pods().await?; + // A rolling upgrade can overlap workers. Never let one idle replacement + // justify scaling down an old pod that may still own sessions. + if pods.len() != 1 { + return Ok(None); + } + let ready_pods: Vec<_> = pods.iter().filter(|pod| ready(pod)).collect(); + if ready_pods.len() > 1 { + bail!("idle mode requires exactly one worker") + } + ready_pods + .first() + .map(|pod| Backend::from_pod(pod, &self.config)) + .transpose() + } + + async fn wake(&mut self) -> Result { + self.idle_since = None; + let deadline = Instant::now() + self.config.cold_start_timeout; + loop { + if Instant::now() >= deadline { + bail!("DuckFlight cold start timed out") + } + if self.needs_fence { + self.fence().await?; + } + let current = self.current().await?; + match current["spec"]["replicas"].as_u64() { + Some(0) => { + // A deleting pod may still accept sockets. Never reuse it + // or wake a replacement until all old pods have gone. + if self.kube.pods().await?.is_empty() { + self.scale(¤t, 1).await?; + } + } + Some(1) => { + if let Some(backend) = self.ready_backend().await? { + let status = backend.status(false, &self.config).await?; + if let Some(permit) = status.permit { + backend.resume(&permit, &self.config).await?; + } + return Ok(backend); + } + } + _ => bail!("idle mode requires worker replicas to be zero or one"), + } + sleep(Duration::from_millis(500)).await; + } + } + + async fn idle_tick(&mut self) -> Result<()> { + if self.needs_fence { + self.fence().await?; + } + if self.connections > 0 { + self.idle_since = None; + return Ok(()); + } + let current = self.current().await?; + if current["spec"]["replicas"] == 0 { + self.idle_since = None; + return Ok(()); + } + if current["spec"]["replicas"] != 1 { + bail!("idle mode requires exactly one worker") + } + let Some(backend) = self.ready_backend().await? else { + self.idle_since = None; + return Ok(()); + }; + let status = backend.status(false, &self.config).await?; + if let Some(permit) = status.permit { + // Recovery follows the API fence above: a delayed old sleep write + // cannot now kill this worker after it is resumed. + backend.resume(&permit, &self.config).await?; + self.idle_since = None; + return Ok(()); + } + if status.busy { + self.idle_since = None; + return Ok(()); + } + let since = match &self.idle_since { + Some((uid, boot, since)) if *uid == backend.uid && *boot == status.boot_id => *since, + _ => { + let since = Instant::now(); + self.idle_since = Some((backend.uid.clone(), status.boot_id.clone(), since)); + since + } + }; + if since.elapsed() < self.config.idle_timeout { + return Ok(()); + } + let suspended = backend.status(true, &self.config).await?; + let Some(permit) = suspended.permit else { + self.idle_since = None; + return Ok(()); + }; + if suspended.busy || suspended.boot_id != status.boot_id { + bail!("worker changed during idle suspension") + } + // The core gate remains closed through Kubernetes deletion. On any + // ambiguous write outcome, do not resume until a new API fence wins. + match self.scale(¤t, 0).await { + Ok(()) => eprintln!("DuckFlight worker suspended after idle timeout"), + Err(error) => { + self.fence().await?; + let observed = self.current().await?; + if observed["spec"]["replicas"] == 1 { + backend.resume(&permit, &self.config).await?; + } + return Err(error); + } + } + self.idle_since = None; + Ok(()) + } +} + +async fn proxy_connection( + mut client: TcpStream, + port: u16, + controller: Arc>, +) -> Result<()> { + // TCP probes that send no protocol data never wake a worker. Established + // protocol sockets have no idle/read timeout and remain pinned until close. + let mut first = [0_u8; 1]; + if timeout(Duration::from_secs(10), client.read(&mut first)).await?? == 0 { + return Ok(()); + } + let backend = { + let mut controller = controller.lock().await; + controller.connections += 1; + match controller.wake().await { + Ok(backend) => backend, + Err(error) => { + controller.connections -= 1; + return Err(error); + } + } + }; + let result = async { + let mut upstream = TcpStream::connect(SocketAddr::new(backend.ip, port)).await?; + upstream.set_nodelay(true)?; + client.set_nodelay(true)?; + upstream.write_all(&first).await?; + tokio::io::copy_bidirectional(&mut client, &mut upstream).await?; + Ok(()) + } + .await; + let mut controller = controller.lock().await; + controller.connections -= 1; + controller.idle_since = None; + result +} + +async fn health(listener: TcpListener) -> Result<()> { + loop { + let (mut stream, _) = listener.accept().await?; + tokio::spawn(async move { + let _ = timeout(Duration::from_secs(2), async { + let mut request = [0_u8; 1024]; + if stream.read(&mut request).await? == 0 { + return Ok(()); + } + stream + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok", + ) + .await + }) + .await; + }); + } +} + +#[tokio::main] +async fn main() -> Result<()> { + if env::args().nth(1).as_deref() == Some("worker-ready") { + return worker_ready().await; + } + let config = Config::from_env()?; + let mut listeners = Vec::new(); + for port in &config.listeners { + listeners.push((*port, TcpListener::bind(("0.0.0.0", *port)).await?)); + } + let controller = Arc::new(Mutex::new(Controller::new(config).await?)); + let health_listener = TcpListener::bind("0.0.0.0:4090").await?; + let health_task = tokio::spawn(health(health_listener)); + let idle_controller = controller.clone(); + let idle_task = tokio::spawn(async move { + loop { + sleep(Duration::from_secs(5)).await; + let mut controller = idle_controller.lock().await; + if let Err(error) = controller.idle_tick().await { + controller.idle_since = None; + eprintln!("idle check refused: {error:#}"); + } + } + }); + let mut acceptors = JoinSet::new(); + let (accepted_tx, mut accepted_rx) = tokio::sync::mpsc::channel(128); + for (port, listener) in listeners { + let accepted_tx = accepted_tx.clone(); + acceptors.spawn(async move { + loop { + let (stream, _) = listener.accept().await?; + if accepted_tx.send((port, stream)).await.is_err() { + return Ok::<_, anyhow::Error>(()); + } + } + }); + } + drop(accepted_tx); + let mut connections = JoinSet::new(); + let terminate = async { + #[cfg(unix)] + { + let mut signal = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("register SIGTERM"); + tokio::select! { _ = signal.recv() => {}, _ = tokio::signal::ctrl_c() => {} } + } + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + } + }; + tokio::pin!(terminate); + loop { + tokio::select! { + _ = &mut terminate => break, + Some(_) = connections.join_next(), if !connections.is_empty() => {}, + connection = accepted_rx.recv() => { + let Some((port, stream)) = connection else { bail!("proxy listeners stopped") }; + let controller = controller.clone(); + connections.spawn(async move { + if let Err(error) = proxy_connection(stream, port, controller).await { eprintln!("connection failed: {error:#}"); } + }); + } + } + } + acceptors.abort_all(); + idle_task.abort(); + health_task.abort(); + // No proxy-imposed session timeout. Kubernetes' termination grace still + // bounds an explicit proxy rollout/deletion, independently of idle sleep. + while connections.join_next().await.is_some() {} + Ok(()) +} + +async fn worker_ready() -> Result<()> { + let address: SocketAddr = env::var("DUCKFLIGHT_IDLE_CONTROL_ADDRESS")?.parse()?; + let hostname = env::var("DUCKFLIGHT_IDLE_SERVER_NAME")?; + let client = Client::builder() + .https_only(true) + .no_proxy() + .add_root_certificate(Certificate::from_pem(&std::fs::read( + "/run/idle-secret/ca.crt", + )?)?) + .resolve( + &hostname, + SocketAddr::new("127.0.0.1".parse()?, address.port()), + ) + .timeout(Duration::from_secs(2)) + .build()?; + let status: IdleStatus = client + .get(format!("https://{hostname}:{}/status", address.port())) + .bearer_auth(std::fs::read_to_string(env::var("DUCKFLIGHT_IDLE_TOKEN_FILE")?)?.trim()) + .send() + .await? + .error_for_status()? + .json() + .await?; + if status.protocol_version != 1 || !status.ready { + bail!("idle-capable listeners are not ready") + } + Ok(()) +} + +#[cfg(test)] +mod tests; diff --git a/crates/duckflight-idle-proxy/src/tests.rs b/crates/duckflight-idle-proxy/src/tests.rs new file mode 100644 index 0000000..c1363f5 --- /dev/null +++ b/crates/duckflight-idle-proxy/src/tests.rs @@ -0,0 +1,340 @@ +use super::*; +use std::sync::atomic::{AtomicBool, Ordering}; +use tokio::io::{AsyncRead, AsyncWrite}; + +struct TestState { + deployment: Value, + busy: bool, + permit: Option, + scale_downs: usize, + extra_pod: bool, + unsupported: bool, +} + +struct Fixture { + state: Arc>, + slow_sleep: Arc, + sleep_started: Arc, + release_sleep: Arc, + api: SocketAddr, + control: SocketAddr, + certificate: Certificate, + directory: tempfile::TempDir, + tasks: JoinSet<()>, +} + +async fn read_request(stream: &mut S) -> Result<(String, Value)> { + let mut header = Vec::new(); + while !header.ends_with(b"\r\n\r\n") { + header.push(stream.read_u8().await?); + if header.len() > 8192 { + bail!("oversized test request") + } + } + let header = String::from_utf8(header)?; + let length = header + .lines() + .find_map(|line| { + let (key, value) = line.split_once(':')?; + key.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().unwrap()) + }) + .unwrap_or(0); + let mut body = vec![0; length]; + stream.read_exact(&mut body).await?; + Ok(( + header.lines().next().unwrap().to_string(), + if body.is_empty() { + Value::Null + } else { + serde_json::from_slice(&body)? + }, + )) +} + +async fn respond(stream: &mut S, code: u32, body: Value) -> Result<()> { + let body = body.to_string(); + stream.write_all(format!("HTTP/1.1 {code} Result\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await?; + stream.shutdown().await?; + Ok(()) +} + +impl Fixture { + async fn new() -> Self { + let state = Arc::new(Mutex::new(TestState { + deployment: json!({"metadata":{"uid":"deployment-one","resourceVersion":"1","annotations":{}},"spec":{"replicas":0}}), + busy: false, + permit: None, + scale_downs: 0, + extra_pod: false, + unsupported: false, + })); + let slow_sleep = Arc::new(AtomicBool::new(false)); + let sleep_started = Arc::new(tokio::sync::Notify::new()); + let release_sleep = Arc::new(tokio::sync::Notify::new()); + let api_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let api = api_listener.local_addr().unwrap(); + let control_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let control = control_listener.local_addr().unwrap(); + let rcgen::CertifiedKey { cert, key_pair } = + rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let certificate = Certificate::from_pem(cert.pem().as_bytes()).unwrap(); + let tls = tokio_rustls::rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![cert.der().clone()], + tokio_rustls::rustls::pki_types::PrivateKeyDer::Pkcs8( + key_pair.serialize_der().into(), + ), + ) + .unwrap(); + let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(tls)); + let mut tasks = JoinSet::new(); + let (api_state, slow, started) = (state.clone(), slow_sleep.clone(), sleep_started.clone()); + let release = release_sleep.clone(); + tasks.spawn(async move { + while let Ok((mut stream, _)) = api_listener.accept().await { + let (state, slow, started) = (api_state.clone(), slow.clone(), started.clone()); + let release = release.clone(); + tokio::spawn(async move { + let (request, patch) = read_request(&mut stream).await.unwrap(); + if request.starts_with("PATCH ") && patch["spec"]["replicas"] == 0 && slow.load(Ordering::SeqCst) { + started.notify_one(); + release.notified().await; + } + let mut state = state.lock().await; + let (code, body) = if request.starts_with("PATCH ") { + if patch["metadata"]["resourceVersion"] != state.deployment["metadata"]["resourceVersion"] { + (409, json!({"message":"resourceVersion conflict"})) + } else { + let version: u32 = state.deployment["metadata"]["resourceVersion"].as_str().unwrap().parse().unwrap(); + state.deployment["metadata"]["resourceVersion"] = json!((version + 1).to_string()); + state.deployment["metadata"]["annotations"] = patch["metadata"]["annotations"].clone(); + if let Some(count) = patch["spec"]["replicas"].as_u64() { + state.deployment["spec"]["replicas"] = json!(count); + if count == 0 { state.scale_downs += 1; state.permit = None; } + } + (200, state.deployment.clone()) + } + } else if request.contains("/pods?") { + let mut pods = Vec::new(); + if state.deployment["spec"]["replicas"] == 1 { + pods.push(json!({"metadata":{"uid":"worker-one"},"status":{"podIP":"127.0.0.1","conditions":[{"type":"Ready","status":"True"}]}})); + if state.extra_pod { pods.push(json!({"metadata":{"uid":"worker-two","deletionTimestamp":"now"},"status":{}})); } + } + (200, json!({"items":pods})) + } else { (200, state.deployment.clone()) }; + drop(state); + let _ = respond(&mut stream, code, body).await; + }); + } + }); + let control_state = state.clone(); + tasks.spawn(async move { + while let Ok((stream, _)) = control_listener.accept().await { + let state = control_state.clone(); + let acceptor = acceptor.clone(); + tokio::spawn(async move { + let mut stream = acceptor.accept(stream).await.unwrap(); + let (request, _) = read_request(&mut stream).await.unwrap(); + let mut state = state.lock().await; + let (code, body) = if state.unsupported { (404, json!({})) } + else if let Some(permit) = request.strip_prefix("POST /resume/").and_then(|s| s.strip_suffix(" HTTP/1.1")) { + if state.permit.as_deref() == Some(permit) { state.permit = None; (200, json!({})) } + else { (409, json!({})) } + } else { + if request.starts_with("POST /suspend ") && !state.busy { state.permit = Some("permit-one".into()); } + (200, json!({"protocol_version":1,"boot_id":"boot-one","ready":true,"busy":state.busy,"permit":state.permit})) + }; + drop(state); + let _ = respond(&mut stream, code, body).await; + }); + } + }); + let directory = tempfile::tempdir().unwrap(); + std::fs::write(directory.path().join("token"), "t".repeat(32)).unwrap(); + Self { + state, + slow_sleep, + sleep_started, + release_sleep, + api, + control, + certificate, + directory, + tasks, + } + } + + async fn controller(&self) -> Controller { + let config = Config { + namespace: "test".into(), + deployment: "worker".into(), + selector: "worker=test".into(), + hostname: "localhost".into(), + control_port: self.control.port(), + idle_timeout: Duration::from_millis(1), + cold_start_timeout: Duration::from_secs(5), + token_file: self.directory.path().join("token"), + ca: self.certificate.clone(), + listeners: vec![], + }; + let kube = Kubernetes { + client: Client::builder() + .timeout(Duration::from_secs(2)) + .build() + .unwrap(), + base: format!("http://{}", self.api), + token_file: config.token_file.clone(), + deployment_path: "/deployments/worker".into(), + pods_path: "/pods".into(), + selector: config.selector.clone(), + }; + let mut controller = Controller { + config, + kube, + owner: uuid::Uuid::new_v4().to_string(), + uid: "deployment-one".into(), + needs_fence: true, + connections: 0, + idle_since: None, + }; + controller.fence().await.unwrap(); + controller + } + + fn expire(controller: &mut Controller) { + controller.idle_since = Some(( + "worker-one".into(), + "boot-one".into(), + Instant::now() - Duration::from_secs(1), + )); + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.tasks.abort_all(); + } +} + +#[tokio::test] +async fn cold_wake_then_idle_sleep_and_busy_state_refusal() { + let fixture = Fixture::new().await; + let mut controller = fixture.controller().await; + controller.wake().await.unwrap(); + assert_eq!(fixture.state.lock().await.deployment["spec"]["replicas"], 1); + fixture.state.lock().await.busy = true; // Detached Flight session/query. + Fixture::expire(&mut controller); + controller.idle_tick().await.unwrap(); + assert!(controller.idle_since.is_none()); + assert_eq!(fixture.state.lock().await.scale_downs, 0); + fixture.state.lock().await.busy = false; + controller.idle_tick().await.unwrap(); + assert_eq!(fixture.state.lock().await.scale_downs, 0); // Fresh idle interval. + Fixture::expire(&mut controller); + controller.idle_tick().await.unwrap(); + assert_eq!(fixture.state.lock().await.scale_downs, 1); + controller.wake().await.unwrap(); + assert_eq!(fixture.state.lock().await.deployment["spec"]["replicas"], 1); +} + +#[tokio::test] +async fn missing_capability_or_overlapping_worker_never_scales_down() { + let fixture = Fixture::new().await; + let mut controller = fixture.controller().await; + controller.wake().await.unwrap(); + fixture.state.lock().await.unsupported = true; + Fixture::expire(&mut controller); + assert!(controller.idle_tick().await.is_err()); + assert_eq!(fixture.state.lock().await.scale_downs, 0); + fixture.state.lock().await.unsupported = false; + fixture.state.lock().await.extra_pod = true; + Fixture::expire(&mut controller); + controller.idle_tick().await.unwrap(); + assert_eq!(fixture.state.lock().await.scale_downs, 0); +} + +#[tokio::test] +async fn timed_out_sleep_is_fenced_before_worker_resume() { + let fixture = Fixture::new().await; + let mut controller = fixture.controller().await; + controller.wake().await.unwrap(); + controller.kube.client = Client::builder() + .timeout(Duration::from_secs(1)) + .build() + .unwrap(); + fixture.slow_sleep.store(true, Ordering::SeqCst); + Fixture::expire(&mut controller); + assert!(controller.idle_tick().await.is_err()); + assert!(!controller.needs_fence); + assert!(fixture.state.lock().await.permit.is_none()); + fixture.release_sleep.notify_one(); + sleep(Duration::from_millis(20)).await; + let state = fixture.state.lock().await; + assert_eq!(state.deployment["spec"]["replicas"], 1); + assert_eq!(state.scale_downs, 0); +} + +#[tokio::test] +async fn proxy_restart_fences_old_sleep_before_worker_resume() { + let fixture = Fixture::new().await; + let mut old = fixture.controller().await; + old.wake().await.unwrap(); + fixture.slow_sleep.store(true, Ordering::SeqCst); + Fixture::expire(&mut old); + let sleeping = tokio::spawn(async move { old.idle_tick().await }); + fixture.sleep_started.notified().await; + sleeping.abort(); // Crash after submitting sleep, before receiving its result. + let mut replacement = fixture.controller().await; + replacement.wake().await.unwrap(); + assert!(fixture.state.lock().await.permit.is_none()); + fixture.release_sleep.notify_one(); + sleep(Duration::from_millis(20)).await; + let state = fixture.state.lock().await; + assert_eq!(state.deployment["spec"]["replicas"], 1); + assert_eq!(state.scale_downs, 0); +} + +#[tokio::test] +async fn both_protocol_ports_pass_bytes_and_keep_open_sessions_awake() { + let fixture = Fixture::new().await; + let controller = Arc::new(Mutex::new(fixture.controller().await)); + for payload in [ + b"\0\0\0\x08\x04\xd2\x16\x2f".as_slice(), + b"\x16\x03\x01flight-tls".as_slice(), + ] { + let upstream = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = upstream.local_addr().unwrap().port(); + let echo = tokio::spawn(async move { + let (mut stream, _) = upstream.accept().await.unwrap(); + let (mut read, mut write) = stream.split(); + tokio::io::copy(&mut read, &mut write).await.unwrap(); + }); + let proxy = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let mut client = TcpStream::connect(proxy.local_addr().unwrap()) + .await + .unwrap(); + let (stream, _) = proxy.accept().await.unwrap(); + let task_controller = controller.clone(); + let copying = tokio::spawn(proxy_connection(stream, port, task_controller)); + client.write_all(payload).await.unwrap(); + let mut reply = vec![0; payload.len()]; + client.read_exact(&mut reply).await.unwrap(); + assert_eq!(reply, payload); + let mut state = controller.lock().await; + Fixture::expire(&mut state); + state.idle_tick().await.unwrap(); + assert_eq!(state.connections, 1); + drop(state); + client.write_all(b"still-open").await.unwrap(); + let mut reply = [0; 10]; + client.read_exact(&mut reply).await.unwrap(); + assert_eq!(&reply, b"still-open"); + drop(client); + copying.await.unwrap().unwrap(); + echo.await.unwrap(); + assert_eq!(controller.lock().await.connections, 0); + } + assert_eq!(fixture.state.lock().await.scale_downs, 0); +} diff --git a/test/kubernetes/Dockerfile b/test/kubernetes/Dockerfile index ae300af..ede5df5 100644 --- a/test/kubernetes/Dockerfile +++ b/test/kubernetes/Dockerfile @@ -6,6 +6,7 @@ COPY test/kubernetes/pyproject.toml test/kubernetes/uv.lock ./ RUN uv sync --frozen --no-dev COPY test/docker_smoke.py ./ COPY test/kubernetes/client.py ./ +COPY test/kubernetes/idle_client.py ./ ENV PATH="/e2e/.venv/bin:$PATH" ENV PYTHONUNBUFFERED=1 ENTRYPOINT ["uv", "run", "--frozen", "--offline", "--no-dev", "/e2e/client.py"] diff --git a/test/kubernetes/README.md b/test/kubernetes/README.md index 17a59de..98c29f9 100644 --- a/test/kubernetes/README.md +++ b/test/kubernetes/README.md @@ -53,3 +53,31 @@ Failures save redacted pod status, events, and logs to `--artifacts DIRECTORY` before cleanup. CI uploads those diagnostics. The suite does not claim to test cloud IAM refresh, production catalog availability, in-flight query migration, or provider-specific catalog authorization. + +## Idle mode + +CI runs the idle suite alongside the lake catalog suites. To run it locally with +the image built from this checkout: + +```sh +docker build --tag duckflight:idle-e2e . +docker build --file test/kubernetes/Dockerfile --tag duckflight-idle-e2e-client:local . +uv run --no-project test/kubernetes/run_idle.py --image duckflight:idle-e2e +``` + +The separate idle suite creates its own disposable cluster and tests both TLS +protocols against the real worker. It shortens the idle interval to 10 seconds, +then verifies cold wake from zero, idle PostgreSQL session preservation, an active +backpressured PostgreSQL result, a Flight Basic-handshake session and result +ticket retained across channel closure, an active Flight stream, explicit +`CloseSession`, sleep after cleanup, and a repeated cold wake. Readiness probes +keep running throughout and must not prevent sleep. The chart default remains +3000 seconds. API timeout/restart fencing and unsupported-core refusal are also +covered by the proxy's Rust tests; detached state and orphan engine work have +core tests. The idle suite does not need an external catalog fixture. + +For local testing of an unreleased core, put the built Linux payload at +`/libduckflight_core_ffi.so` in a scratch test image and pass +`--build-arg DUCKFLIGHT_CORE_TEST_IMAGE=` and +`--build-arg DUCKFLIGHT_CORE_TEST_SHA256=` to the server build. The build +verifies the digest. Production builds use `core-assets.lock`. diff --git a/test/kubernetes/client.py b/test/kubernetes/client.py index a815543..c60f7d6 100644 --- a/test/kubernetes/client.py +++ b/test/kubernetes/client.py @@ -226,9 +226,15 @@ def verify(updated: bool) -> None: if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("mode", choices=["seed", "append", "verify", "verify-updated"]) + parser.add_argument( + "mode", choices=["seed", "append", "verify", "verify-updated", "idle"] + ) mode = parser.parse_args().mode - if mode in ("seed", "append"): + if mode == "idle": + from idle_client import main as idle_cycle + + idle_cycle() + elif mode in ("seed", "append"): seed(mode == "append") else: verify(mode == "verify-updated") diff --git a/test/kubernetes/idle_client.py b/test/kubernetes/idle_client.py new file mode 100644 index 0000000..97c5062 --- /dev/null +++ b/test/kubernetes/idle_client.py @@ -0,0 +1,144 @@ +"""Real-client sleep/wake and activity protection, inside the disposable cluster.""" + +from __future__ import annotations + +import json +import os +import ssl +import time +import urllib.request +from pathlib import Path + +import psycopg +from docker_smoke import flight_execute, flight_query, protobuf_strings +from pyarrow import flight + +HOST = "lake-duckflight" +CA = Path("/run/test/server.crt") +SERVICE_ACCOUNT = Path("/var/run/secrets/kubernetes.io/serviceaccount") + + +def replicas() -> int: + namespace = (SERVICE_ACCOUNT / "namespace").read_text().strip() + host = os.environ["KUBERNETES_SERVICE_HOST"] + if ":" in host: + host = f"[{host}]" + url = f"https://{host}:{os.environ['KUBERNETES_SERVICE_PORT_HTTPS']}/apis/apps/v1/namespaces/{namespace}/deployments/{HOST}" + request = urllib.request.Request( + url, + headers={ + "Authorization": "Bearer " + (SERVICE_ACCOUNT / "token").read_text().strip() + }, + ) + context = ssl.create_default_context(cafile=str(SERVICE_ACCOUNT / "ca.crt")) + with urllib.request.urlopen(request, context=context, timeout=5) as response: + return json.load(response)["spec"]["replicas"] + + +def remains_awake(seconds: int = 25) -> None: + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + assert replicas() == 1, "idle controller killed protected activity" + time.sleep(1) + + +def sleeps() -> None: + deadline = time.monotonic() + 90 + while time.monotonic() < deadline: + if replicas() == 0: + return + time.sleep(1) + raise AssertionError("worker did not sleep after sessions/results were closed") + + +def postgres(): + return psycopg.connect( + host=HOST, + port=5433, + user="e2e", + password=os.environ["CLIENT_PASSWORD"], + dbname="duckflight", + sslmode="verify-full", + sslrootcert=str(CA), + connect_timeout=120, + autocommit=True, + ) + + +def flight_client(): + return flight.FlightClient( + f"grpc+tls://{HOST}:31337", tls_root_certs=CA.read_bytes() + ) + + +def flight_info(client, options, query): + command = protobuf_strings( + { + 1: b"type.googleapis.com/arrow.flight.protocol.sql.CommandStatementQuery", + 2: protobuf_strings({1: query.encode()}), + } + ) + return client.get_flight_info(flight.FlightDescriptor.for_command(command), options) + + +def main() -> None: + assert replicas() == 0, "idle installation must start asleep" + with postgres() as pg: + assert pg.execute("select 42").fetchone() == (42,) + pg.execute("create temp table retained as select 17 as value") + remains_awake() + assert pg.execute("select value from retained").fetchone() == (17,) + with pg.cursor() as cursor: + # Exceed socket buffers without exhausting a 1GiB test worker. + stream = cursor.stream("select range from range(1000000)") + assert next(stream) == (0,) + remains_awake() + assert next(stream) == (1,) + pg.close() # Explicit client disconnect ends the pending result. + stream.close() + sleeps() + print( + "PASS PostgreSQL: cold wake, idle session and active result survive, then sleep", + flush=True, + ) + + with flight_client() as client: + header = client.authenticate_basic_token( + "e2e", os.environ["CLIENT_PASSWORD"], flight.FlightCallOptions(timeout=120) + ) + options = flight.FlightCallOptions(headers=[header], timeout=120) + flight_execute( + client, options, "create temp table retained as select 23 as value" + ) + pending = flight_info(client, options, "select value from retained") + # Closing a channel does not close the Basic-handshake session or its ticket. + remains_awake() + with flight_client() as client: + assert client.do_get( + pending.endpoints[0].ticket, options + ).read_all().to_pylist() == [{"value": 23}] + assert flight_query(client, options, "select value from retained") == [ + {"value": 23} + ] + info = flight_info(client, options, "select range from range(1000000)") + reader = client.do_get(info.endpoints[0].ticket, options) + assert reader.read_chunk().data.num_rows > 0 + remains_awake() + assert reader.read_chunk().data.num_rows > 0 + reader.cancel() + # A reader retains its transport even after cancellation. + del reader + list(client.do_action(flight.Action("CloseSession", b""), options)) + sleeps() + print( + "PASS Flight: cold wake, detached session/ticket and active stream survive, CloseSession allows sleep", + flush=True, + ) + + with postgres() as pg: + assert pg.execute("select 77").fetchone() == (77,) + sleeps() + print( + "PASS repeated wake after zero workers; idle probes did not prevent sleep", + flush=True, + ) diff --git a/test/kubernetes/run.py b/test/kubernetes/run.py index 9158d47..90b50c1 100755 --- a/test/kubernetes/run.py +++ b/test/kubernetes/run.py @@ -22,6 +22,7 @@ class Suite: def __init__(self, arguments, directory: Path): self.arguments = arguments + self.client_image = getattr(arguments, "client_image", CLIENT_IMAGE) self.directory = directory self.cluster = arguments.cluster or f"duckflight-e2e-{secrets.token_hex(4)}" self.namespace = f"lake-e2e-{secrets.token_hex(4)}" @@ -121,7 +122,7 @@ def setup(self) -> None: "image", "import", self.arguments.image, - CLIENT_IMAGE, + self.client_image, "--cluster", self.cluster, ) @@ -255,7 +256,13 @@ def configure_workers(self) -> None: "5m", ) - def job(self, mode: str, pod_ip: str = "") -> None: + def job( + self, + mode: str, + pod_ip: str = "", + service_account: str = "default", + timeout_seconds: int = 240, + ) -> None: name = f"{mode}-{secrets.token_hex(3)}" self.apply( { @@ -264,14 +271,15 @@ def job(self, mode: str, pod_ip: str = "") -> None: "metadata": {"name": name}, "spec": { "backoffLimit": 0, - "activeDeadlineSeconds": 240, + "activeDeadlineSeconds": timeout_seconds, "template": { "spec": { + "serviceAccountName": service_account, "restartPolicy": "Never", "containers": [ { "name": "client", - "image": CLIENT_IMAGE, + "image": self.client_image, "imagePullPolicy": "Never", "args": [mode], "envFrom": [ @@ -311,7 +319,7 @@ def job(self, mode: str, pod_ip: str = "") -> None: }, } ) - deadline = time.monotonic() + 270 + deadline = time.monotonic() + timeout_seconds + 30 while time.monotonic() < deadline: job = json.loads(self.kubectl("get", "job", name, "-o", "json")) status = job.get("status", {}) diff --git a/test/kubernetes/run_idle.py b/test/kubernetes/run_idle.py new file mode 100755 index 0000000..9a220f5 --- /dev/null +++ b/test/kubernetes/run_idle.py @@ -0,0 +1,206 @@ +#!/usr/bin/env -S uv run --script +"""Run the opt-in chart's real PostgreSQL/Flight idle cycle on disposable k3s.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import secrets +import tempfile +from pathlib import Path + +from run import K3S_IMAGE, ROOT, Suite + + +class IdleSuite(Suite): + def setup(self) -> None: + self.command( + self.arguments.k3d, + "cluster", + "create", + self.cluster, + "--image", + K3S_IMAGE, + "--agents", + "1", + "--kubeconfig-update-default=false", + "--kubeconfig-switch-context=false", + "--k3s-arg", + "--disable=traefik@server:0", + "--k3s-arg", + "--disable=servicelb@server:0", + "--wait", + "--timeout", + "180s", + ) + config = self.command(self.arguments.k3d, "kubeconfig", "get", self.cluster) + (self.directory / "kubeconfig").write_text(config) + (self.directory / "kubeconfig").chmod(0o600) + self.command("kubectl", "create", "namespace", self.namespace) + self.namespace_created = True + self.command( + self.arguments.k3d, + "image", + "import", + self.arguments.image, + self.client_image, + "--cluster", + self.cluster, + ) + self.secret( + "test-credentials", + {key: value.encode() for key, value in self.credentials.items()}, + ) + certificate = self.directory / "server.crt" + key = self.directory / "server.key" + self.command( + "openssl", + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + str(key), + "-out", + str(certificate), + "-days", + "1", + "-subj", + "/CN=lake-duckflight", + "-addext", + f"subjectAltName=DNS:lake-duckflight,DNS:lake-duckflight.{self.namespace}.svc", + "-addext", + "basicConstraints=critical,CA:FALSE", + ) + salt = secrets.token_bytes(16) + digest = hashlib.pbkdf2_hmac( + "sha256", self.credentials["CLIENT_PASSWORD"].encode(), salt, 10000 + ).hex() + auth = ( + f'[users.e2e]\npassword_hash="{digest}"\nsalt={list(salt)}\niterations=10000\n' + '[tls]\ncert="server.crt"\nkey="server.key"\n' + ) + self.secret( + "worker-auth", + { + "duckflight.toml": auth.encode(), + "server.crt": certificate.read_bytes(), + "server.key": key.read_bytes(), + }, + ) + self.secret("client-tls", {"server.crt": certificate.read_bytes()}) + self.secret("worker-bootstrap", {"init.sql": b"select 1;\n"}) + self.secret( + "idle-control", + { + "token": secrets.token_hex(32).encode(), + "ca.crt": certificate.read_bytes(), + }, + ) + self.apply( + { + "apiVersion": "v1", + "kind": "ServiceAccount", + "metadata": {"name": "idle-client"}, + }, + { + "apiVersion": "rbac.authorization.k8s.io/v1", + "kind": "Role", + "metadata": {"name": "idle-client"}, + "rules": [ + { + "apiGroups": ["apps"], + "resources": ["deployments"], + "resourceNames": ["lake-duckflight"], + "verbs": ["get"], + } + ], + }, + { + "apiVersion": "rbac.authorization.k8s.io/v1", + "kind": "RoleBinding", + "metadata": {"name": "idle-client"}, + "roleRef": { + "apiGroup": "rbac.authorization.k8s.io", + "kind": "Role", + "name": "idle-client", + }, + "subjects": [ + { + "kind": "ServiceAccount", + "name": "idle-client", + "namespace": self.namespace, + } + ], + }, + ) + repository, tag = self.arguments.image.rsplit(":", 1) + values = { + "image": {"repository": repository, "tag": tag, "pullPolicy": "Never"}, + "auth": {"existingSecret": "worker-auth"}, + "bootstrap": {"existingSecret": "worker-bootstrap"}, + "idleScale": { + "enabled": True, + "existingSecret": "idle-control", + "idleTimeoutSeconds": 10, + }, + "duckdb": {"memoryLimit": "512MB", "threads": 1}, + "resources": { + "requests": { + "cpu": "100m", + "memory": "512Mi", + "ephemeral-storage": "256Mi", + }, + "limits": {"cpu": "1", "memory": "1Gi", "ephemeral-storage": "2Gi"}, + }, + } + values_file = self.directory / "values.json" + values_file.write_text(json.dumps(values)) + self.command( + "helm", + "upgrade", + "--install", + "lake", + str(ROOT / "charts/duckflight"), + "--namespace", + self.namespace, + "--values", + str(values_file), + "--wait", + "--timeout", + "5m", + ) + + def run(self) -> None: + self.setup() + self.job("idle", service_account="idle-client", timeout_seconds=600) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--image", required=True, help="Locally built idle-capable server image" + ) + parser.add_argument("--k3d", default="k3d") + parser.add_argument("--client-image", default="duckflight-idle-e2e-client:local") + parser.add_argument( + "--artifacts", type=Path, default=Path("/tmp/duckflight-idle-e2e-artifacts") + ) + arguments = parser.parse_args() + arguments.catalog = "idle" + arguments.cluster = None + with tempfile.TemporaryDirectory(prefix="duckflight-idle-k3s-") as directory: + suite = IdleSuite(arguments, Path(directory)) + try: + suite.run() + except Exception: + suite.diagnose() + raise + finally: + suite.command(arguments.k3d, "cluster", "delete", suite.cluster) + + +if __name__ == "__main__": + main() diff --git a/test/release/client_regressions.py b/test/release/client_regressions.py index d6bdd65..30321bf 100755 --- a/test/release/client_regressions.py +++ b/test/release/client_regressions.py @@ -18,6 +18,7 @@ import hashlib import os import secrets +import signal import sys import tempfile import time @@ -310,7 +311,8 @@ def execute(sql: str): # Each mode must exceed the bundled core's 16 connection slots. for explicit in (True, False): for _ in range(24): - info = execute("select i from range(1000000000) t(i)") + # Exceed transport buffers while staying within CI memory limits. + info = execute("select i from range(4000000) t(i)") reader = client.do_get(info.endpoints[0].ticket, options) if reader.read_chunk().data.num_rows == 0: raise AssertionError("expected a running, nonempty query") @@ -475,12 +477,24 @@ def main() -> None: description="Run public-bundle DuckFlight client regressions" ) parser.add_argument("extension", type=Path, help="bundled .duckdb_extension path") + parser.add_argument( + "--native-sigpipe", + action="store_true", + help="test disconnected Flight writers with a native host's default SIGPIPE policy", + ) parser.add_argument( "--require-noexec-tmp", action="store_true", help="require Linux temporary storage mounted noexec before exercising the bundle", ) args = parser.parse_args() + if args.native_sigpipe: + if not hasattr(signal, "SIGPIPE"): + parser.error("--native-sigpipe requires a Unix host") + # Python normally ignores SIGPIPE, which masks crashes in native hosts. + # This standalone test process owns its signal policy; the core must not + # change it when clients abandon their Flight response streams. + signal.signal(signal.SIGPIPE, signal.SIG_DFL) if args.require_noexec_tmp: if sys.platform != "linux": parser.error("--require-noexec-tmp requires Linux") diff --git a/test/test_helm_chart.py b/test/test_helm_chart.py index 6f9ad10..d73c4c9 100755 --- a/test/test_helm_chart.py +++ b/test/test_helm_chart.py @@ -110,8 +110,104 @@ def test_existing_service_account_and_digest(self): "ghcr.io/sidequery/duckflight-extension@sha256:" + "a" * 64, ) + def test_idle_mode_routes_both_protocols_through_one_controller(self): + result = render("idleScale.enabled=true", "idleScale.existingSecret=idle") + self.assertEqual(result.returncode, 0, result.stderr) + objects = { + (obj["kind"], obj["metadata"]["name"]): obj + for obj in yaml.safe_load_all(result.stdout) + } + worker = objects["Deployment", "lake-duckflight"] + proxy = objects["Deployment", "lake-duckflight-idle-proxy"] + self.assertEqual(worker["spec"]["replicas"], 0) + self.assertEqual(proxy["spec"]["replicas"], 1) + self.assertEqual(proxy["spec"]["strategy"], {"type": "Recreate"}) + pod = proxy["spec"]["template"]["spec"] + container = pod["containers"][0] + environment = {item["name"]: item.get("value") for item in container["env"]} + self.assertEqual(environment["DUCKFLIGHT_PROXY_IDLE_TIMEOUT_SECONDS"], "3000") + self.assertEqual(environment["DUCKFLIGHT_PROXY_POSTGRES_PORT"], "5433") + self.assertEqual(environment["DUCKFLIGHT_PROXY_FLIGHT_PORT"], "31337") + service = objects["Service", "lake-duckflight"] + backend = objects["Service", "lake-duckflight-workers"] + self.assertEqual( + service["spec"]["selector"], + proxy["spec"]["template"]["metadata"]["labels"], + ) + self.assertEqual( + backend["spec"]["selector"], worker["spec"]["selector"]["matchLabels"] + ) + role = objects["Role", "lake-duckflight-idle-proxy"] + self.assertEqual(role["rules"][0]["resourceNames"], ["lake-duckflight"]) + self.assertEqual(role["rules"][0]["verbs"], ["get", "patch"]) + self.assertEqual(role["rules"][1]["verbs"], ["list"]) + self.assertTrue(pod["automountServiceAccountToken"]) + worker_pod = worker["spec"]["template"]["spec"] + self.assertFalse(worker_pod["automountServiceAccountToken"]) + worker_env = { + item["name"]: item["value"] for item in worker_pod["containers"][0]["env"] + } + self.assertEqual(worker_env["DUCKFLIGHT_IDLE_EXPECTED_LISTENERS"], "2") + policy = objects["NetworkPolicy", "lake-duckflight-workers"]["spec"] + self.assertEqual( + policy["podSelector"]["matchLabels"], backend["spec"]["selector"] + ) + self.assertEqual( + policy["ingress"][0]["from"][0]["podSelector"]["matchLabels"], + service["spec"]["selector"], + ) + self.assertEqual( + {port["port"] for port in policy["ingress"][0]["ports"]}, + {5433, 31337, 4091}, + ) + + def test_idle_mode_supports_either_protocol_independently(self): + for disabled, remaining in (("postgres", "flight"), ("flight", "postgres")): + with self.subTest(disabled=disabled): + result = render( + "idleScale.enabled=true", + "idleScale.existingSecret=idle", + f"{disabled}.enabled=false", + ) + self.assertEqual(result.returncode, 0, result.stderr) + objects = list(yaml.safe_load_all(result.stdout)) + proxy = next( + obj + for obj in objects + if obj["kind"] == "Deployment" + and obj["metadata"]["name"].endswith("idle-proxy") + ) + environment = { + item["name"]: item.get("value") + for item in proxy["spec"]["template"]["spec"]["containers"][0][ + "env" + ] + } + self.assertNotIn( + f"DUCKFLIGHT_PROXY_{disabled.upper()}_PORT", environment + ) + self.assertIn(f"DUCKFLIGHT_PROXY_{remaining.upper()}_PORT", environment) + def test_invalid_runtime_configuration_fails_before_install(self): cases = ( + (("idleScale.enabled=true",), "idleScale.existingSecret"), + (("idleScale.idleTimeoutSeconds=0",), "idleTimeoutSeconds"), + ( + ( + "idleScale.enabled=true", + "idleScale.existingSecret=idle", + "idleScale.controlPort=5433", + ), + "must differ", + ), + ( + ( + "idleScale.enabled=true", + "idleScale.existingSecret=idle", + "flight.port=4090", + ), + "health port", + ), (("postgres.enabled=false", "flight.enabled=false"), "at least one"), (("image.tag=",), "image.tag or image.digest"), (("replicaCount=-1",), "replicaCount"),