diff --git a/CHANGELOG.md b/CHANGELOG.md index e46c95abb..3ce4d4181 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ * The server scheduler now contains a safety limit for computation, configurable via `--scheduler-time-limit` (default: 5s) * Better scheduling policy (prefill) for heterogenous clusters +* Added a new automatic allocation backend that submits Slurm allocations through a + [FirecREST](https://eth-cscs.github.io/firecrest-v2/) API (`hq alloc add firecrest`). + Unlike the PBS/Slurm backends, it does not require the server to run on the target + cluster: allocations are submitted via HTTPS using OAuth2 client credentials, and the + spawned workers connect back to the server over TCP. This enables long-running + HyperQueue servers placed outside of clusters with login-node wall-time limits + (e.g. CSCS). Enabled with the `firecrest` cargo feature (on by default). ### Fixes @@ -13,6 +20,10 @@ * Fixed server crash in a specific situation when an unschedulable high-priority task occurs * Fixed server crash caused by invalid handling of prefill * Fixed canceling of prefilled tasks +* When a server is restored from a journal and one of its allocation queues cannot be + recreated (for example because its backend is unavailable in the current build or its + configuration is no longer valid), the queue is now skipped with an error in the log + instead of silently aborting the rest of the state restoration. ## v0.26.2 diff --git a/Cargo.lock b/Cargo.lock index c2346babe..84690c881 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -80,7 +80,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -91,7 +91,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -124,12 +124,30 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[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" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "bincode" version = "1.3.3" @@ -156,7 +174,7 @@ dependencies = [ "regex", "rustc-hash", "shlex 1.3.0", - "syn", + "syn 2.0.118", ] [[package]] @@ -377,7 +395,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -436,7 +454,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -461,7 +479,7 @@ checksum = "4fe5f465a4f6fee88fad41b85d990f84c835335e85b5d9e6e63e0d06d28cba7c" dependencies = [ "encode_unicode", "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -649,7 +667,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn", + "syn 2.0.118", ] [[package]] @@ -662,7 +680,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn", + "syn 2.0.118", ] [[package]] @@ -673,7 +691,7 @@ checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ "darling_core 0.20.11", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -684,7 +702,7 @@ checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" dependencies = [ "darling_core 0.23.0", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -706,7 +724,7 @@ dependencies = [ "defmt-parser", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -742,7 +760,7 @@ dependencies = [ "darling 0.20.11", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -752,7 +770,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab63b0e2bf4d5928aff72e83a7dace85d7bba5fe12dcc3c5a572d78caffd3f3c" dependencies = [ "derive_builder_core", - "syn", + "syn 2.0.118", ] [[package]] @@ -774,7 +792,7 @@ dependencies = [ "proc-macro2", "quote", "rustc_version", - "syn", + "syn 2.0.118", ] [[package]] @@ -795,7 +813,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -808,6 +826,17 @@ dependencies = [ "objc2", ] +[[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.6", +] + [[package]] name = "document-features" version = "0.2.12" @@ -865,7 +894,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -904,6 +933,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" +[[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" version = "0.3.32" @@ -960,7 +998,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1018,8 +1056,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1029,9 +1069,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi", "rand_core", + "wasm-bindgen", ] [[package]] @@ -1123,12 +1165,111 @@ dependencies = [ "cmake", ] +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "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 = "humantime" version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15" +[[package]] +name = "hyper" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "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", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots", +] + +[[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 = "hyperqueue" version = "0.26.2" @@ -1166,6 +1307,7 @@ dependencies = [ "orion", "rand", "ratatui", + "reqwest", "serde", "serde_bytes", "serde_json", @@ -1208,12 +1350,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 = "indent_write" version = "2.2.0" @@ -1261,9 +1507,15 @@ dependencies = [ "indoc", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] +[[package]] +name = "ipnet" +version = "2.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" + [[package]] name = "is-docker" version = "0.2.0" @@ -1373,7 +1625,7 @@ checksum = "d0879bd39df99c4c5e2c6615ccc026391a423dde10532c573e6086eb94a802cc" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1471,6 +1723,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 = "litrs" version = "1.0.0" @@ -1501,6 +1759,12 @@ dependencies = [ "hashbrown 0.17.1", ] +[[package]] +name = "lru-slab" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" + [[package]] name = "matrixmultiply" version = "0.3.10" @@ -1543,7 +1807,7 @@ dependencies = [ "libc", "log", "wasi", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1791,7 +2055,7 @@ dependencies = [ "by_address", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1817,6 +2081,12 @@ dependencies = [ "windows-link", ] +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -1872,6 +2142,15 @@ dependencies = [ "portable-atomic", ] +[[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.0" @@ -1885,7 +2164,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn", + "syn 2.0.118", ] [[package]] @@ -1982,7 +2261,7 @@ dependencies = [ "proc-macro2", "pyo3-macros-backend", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1994,7 +2273,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2007,6 +2286,62 @@ dependencies = [ "serde", ] +[[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.46" @@ -2039,6 +2374,15 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core", +] + [[package]] name = "ratatui" version = "0.30.2" @@ -2180,6 +2524,58 @@ 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-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", + "webpki-roots", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -2205,7 +2601,42 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls" +version = "0.23.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[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", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", ] [[package]] @@ -2278,7 +2709,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2303,6 +2734,18 @@ dependencies = [ "serde_core", ] +[[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 = "1.3.0" @@ -2390,7 +2833,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2405,6 +2848,12 @@ dependencies = [ "smallvec", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "stacker" version = "0.1.24" @@ -2415,7 +2864,7 @@ dependencies = [ "cfg-if", "libc", "psm", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2448,7 +2897,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2468,6 +2917,37 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +dependencies = [ + "proc-macro2", + "quote", + "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.6", +] + [[package]] name = "sysinfo" version = "0.39.6" @@ -2536,7 +3016,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2585,7 +3065,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2609,6 +3089,16 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" +[[package]] +name = "tinystr" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" +dependencies = [ + "displaydoc", + "zerovec", +] + [[package]] name = "tinytemplate" version = "1.2.1" @@ -2619,6 +3109,12 @@ dependencies = [ "serde_json", ] +[[package]] +name = "tinyvec" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" + [[package]] name = "tokio" version = "1.52.3" @@ -2633,7 +3129,7 @@ dependencies = [ "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2644,7 +3140,17 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db" +dependencies = [ + "rustls", + "tokio", ] [[package]] @@ -2699,6 +3205,51 @@ version = "1.1.1+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "756daf9b1013ebe47a8776667b466417e2d4c5679d441c26230efd9ef78692db" +[[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 2.13.0", + "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" @@ -2718,7 +3269,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2730,6 +3281,12 @@ 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" @@ -2771,6 +3328,30 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[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_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + [[package]] name = "utf8parse" version = "0.2.2" @@ -2793,6 +3374,15 @@ dependencies = [ "winapi-util", ] +[[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" @@ -2812,6 +3402,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.76" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.126" @@ -2831,7 +3431,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.118", "wasm-bindgen-shared", ] @@ -2864,6 +3464,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "which" version = "8.0.4" @@ -2895,7 +3504,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2957,7 +3566,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2968,7 +3577,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3005,6 +3614,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -3014,6 +3632,22 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + [[package]] name = "windows-threading" version = "0.2.1" @@ -3023,12 +3657,89 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + [[package]] name = "winnow" version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0592e1c9d151f854e6fd382574c3a0855250e1d9b2f99d9281c6e6391af352f1" +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[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.6", + "synstructure", +] + [[package]] name = "zerocopy" version = "0.8.54" @@ -3046,7 +3757,28 @@ checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", +] + +[[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.6", + "synstructure", ] [[package]] @@ -3055,6 +3787,39 @@ 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.6", +] + [[package]] name = "zmij" version = "1.0.21" diff --git a/crates/hyperqueue/Cargo.toml b/crates/hyperqueue/Cargo.toml index 89cf41c88..6e23aa73e 100644 --- a/crates/hyperqueue/Cargo.toml +++ b/crates/hyperqueue/Cargo.toml @@ -61,6 +61,7 @@ tako = { path = "../tako", default-features = false } # Optional dependencies jemallocator = { version = "0.5", optional = true } +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"], optional = true } which = "8" [dev-dependencies] @@ -69,11 +70,13 @@ insta = "1.15.0" criterion = { workspace = true } [features] -default = ["jemalloc", "dashboard", "highs"] +default = ["jemalloc", "dashboard", "highs", "firecrest"] # Use the jemalloc allocator jemalloc = ["jemallocator"] # Enable the dashboard dashboard = ["dep:ratatui", "dep:crossterm", "dep:unicode-width"] +# Enable the FirecREST autoalloc backend +firecrest = ["dep:reqwest"] highs = ["tako/highs"] microlp = ["tako/microlp"] diff --git a/crates/hyperqueue/src/client/commands/autoalloc.rs b/crates/hyperqueue/src/client/commands/autoalloc.rs index 38dcecb94..1d19148e7 100644 --- a/crates/hyperqueue/src/client/commands/autoalloc.rs +++ b/crates/hyperqueue/src/client/commands/autoalloc.rs @@ -9,7 +9,7 @@ use crate::common::manager::info::ManagerType; use crate::common::utils::time::parse_hms_or_human_time; use crate::rpc_call; use crate::server::autoalloc::{ - Allocation, AllocationId, AllocationState, QueueId, QueueParameters, + Allocation, AllocationId, AllocationState, FirecrestQueueParams, QueueId, QueueParameters, }; use crate::server::bootstrap::get_client_session; use crate::transfer::connection::ClientSession; @@ -82,6 +82,56 @@ enum AddQueueCommand { Pbs(SharedQueueOpts), /// Create a SLURM allocation queue Slurm(SharedQueueOpts), + /// Create an allocation queue submitting Slurm jobs through a FirecREST API + Firecrest(FirecrestQueueOpts), +} + +#[derive(Parser)] +struct FirecrestQueueOpts { + /// Base URL of the FirecREST API + /// + /// Example: `https://api.cscs.ch/hpc/firecrest/v2` + #[arg(long)] + api_url: String, + + /// Name of the target system (cluster) as known to FirecREST, e.g. `daint` + #[arg(long)] + system: String, + + /// URL of the OAuth2 token endpoint used for the client credentials grant + #[arg(long)] + token_url: String, + + /// OAuth2 client ID + #[arg(long)] + client_id: String, + + /// Name of an environment variable that holds the OAuth2 client secret + /// + /// The variable is read in the environment of the **server**, at the time the queue + /// is created or restored. The secret itself is never stored by HyperQueue. + #[arg(long, default_value = "HQ_FIRECREST_CLIENT_SECRET")] + client_secret_env: String, + + /// Path to the `hq` binary on the target cluster + /// + /// It must be the same version as the server. + #[arg(long)] + remote_hq_path: String, + + /// Directory on the target cluster containing the worker access file (`access.json`) + /// + /// Generate the access file with `hq server generate-access` and copy it to the + /// target cluster before creating the queue. + #[arg(long)] + remote_server_dir: String, + + /// Existing directory on the target cluster where allocation stdout/stderr will be written + #[arg(long)] + remote_workdir: String, + + #[clap(flatten)] + shared: SharedQueueOpts, } fn parse_backlog(value: &str) -> Result { @@ -192,6 +242,8 @@ enum DryRunCommand { Pbs(SharedQueueOpts), /// Try to create a SLURM allocation Slurm(SharedQueueOpts), + /// Try to create a Slurm allocation through a FirecREST API + Firecrest(FirecrestQueueOpts), } #[derive(Parser)] @@ -413,9 +465,37 @@ wasted allocation duration." worker_args, idle_timeout, cli_resource_descriptor, + firecrest: None, }) } +fn firecrest_args_to_params(opts: FirecrestQueueOpts) -> anyhow::Result { + let FirecrestQueueOpts { + api_url, + system, + token_url, + client_id, + client_secret_env, + remote_hq_path, + remote_server_dir, + remote_workdir, + shared, + } = opts; + + let mut params = args_to_params(ManagerType::Firecrest, shared)?; + params.firecrest = Some(FirecrestQueueParams { + api_url, + system, + token_url, + client_id, + client_secret_env, + remote_hq_path, + remote_server_dir, + remote_workdir, + }); + Ok(params) +} + fn construct_resources_from_cli(args: &SharedWorkerStartOpts) -> Option { let SharedWorkerStartOpts { resource, cpus, .. } = args; @@ -444,6 +524,7 @@ async fn dry_run_command(mut session: ClientSession, opts: DryRunOpts) -> anyhow let parameters = match opts.subcmd { DryRunCommand::Pbs(params) => args_to_params(ManagerType::Pbs, params), DryRunCommand::Slurm(params) => args_to_params(ManagerType::Slurm, params), + DryRunCommand::Firecrest(params) => firecrest_args_to_params(params), }; let message = FromClientMessage::AutoAlloc(AutoAllocRequest::DryRun { parameters: parameters?, @@ -471,6 +552,10 @@ async fn add_queue(mut session: ClientSession, opts: AddQueueOpts) -> anyhow::Re let no_dry_run = params.no_dry_run; (args_to_params(ManagerType::Slurm, params), !no_dry_run) } + AddQueueCommand::Firecrest(params) => { + let no_dry_run = params.shared.no_dry_run; + (firecrest_args_to_params(params), !no_dry_run) + } }; let message = FromClientMessage::AutoAlloc(AutoAllocRequest::AddQueue { diff --git a/crates/hyperqueue/src/client/output/json.rs b/crates/hyperqueue/src/client/output/json.rs index 29532c715..0745d42ca 100644 --- a/crates/hyperqueue/src/client/output/json.rs +++ b/crates/hyperqueue/src/client/output/json.rs @@ -472,6 +472,7 @@ impl Serialize for FormattedManagerType { match self.0 { ManagerType::Pbs => serializer.serialize_str("PBS"), ManagerType::Slurm => serializer.serialize_str("Slurm"), + ManagerType::Firecrest => serializer.serialize_str("FirecREST"), } } } diff --git a/crates/hyperqueue/src/common/manager/info.rs b/crates/hyperqueue/src/common/manager/info.rs index 5232c980a..a90802bcd 100644 --- a/crates/hyperqueue/src/common/manager/info.rs +++ b/crates/hyperqueue/src/common/manager/info.rs @@ -9,6 +9,9 @@ pub const WORKER_EXTRA_MANAGER_KEY: &str = "JobManager"; pub enum ManagerType { Pbs, Slurm, + /// Slurm allocations submitted through a remote FirecREST API. + /// Workers spawned by this manager run inside ordinary Slurm jobs. + Firecrest, } impl Display for ManagerType { @@ -16,6 +19,7 @@ impl Display for ManagerType { match self { ManagerType::Pbs => f.write_str("PBS"), ManagerType::Slurm => f.write_str("SLURM"), + ManagerType::Firecrest => f.write_str("FirecREST"), } } } diff --git a/crates/hyperqueue/src/server/autoalloc/mod.rs b/crates/hyperqueue/src/server/autoalloc/mod.rs index 4893e5c84..72e7ca779 100644 --- a/crates/hyperqueue/src/server/autoalloc/mod.rs +++ b/crates/hyperqueue/src/server/autoalloc/mod.rs @@ -13,7 +13,7 @@ mod state; pub type AutoAllocResult = anyhow::Result; pub use process::try_submit_allocation; -pub use queue::{QueueInfo, QueueParameters}; +pub use queue::{FirecrestQueueParams, QueueInfo, QueueParameters}; pub use service::{AutoAllocService, LostWorkerDetails, create_autoalloc_service}; pub use state::{Allocation, AllocationId, AllocationState, QueueId}; diff --git a/crates/hyperqueue/src/server/autoalloc/process.rs b/crates/hyperqueue/src/server/autoalloc/process.rs index dbd60981c..974c5ea20 100644 --- a/crates/hyperqueue/src/server/autoalloc/process.rs +++ b/crates/hyperqueue/src/server/autoalloc/process.rs @@ -125,11 +125,8 @@ pub async fn autoalloc_process( /// given allocation parameters. pub async fn try_submit_allocation(params: QueueParameters) -> anyhow::Result<()> { let tmpdir = TempDir::with_prefix("hq")?; - let mut handler = create_allocation_handler( - ¶ms.manager, - params.name.clone(), - tmpdir.as_ref().to_path_buf(), - )?; + let mut handler = + create_allocation_handler(¶ms, params.name.clone(), tmpdir.as_ref().to_path_buf())?; // Try the highest possible worker count, to ensure that it works let worker_count = params.max_workers_per_alloc; @@ -164,11 +161,11 @@ pub async fn try_submit_allocation(params: QueueParameters) -> anyhow::Result<() // The code doesn't compile if the Box closures are removed #[allow(clippy::redundant_closure)] pub fn create_allocation_handler( - manager: &ManagerType, + params: &QueueParameters, name: Option, directory: PathBuf, ) -> anyhow::Result> { - match manager { + match ¶ms.manager { ManagerType::Pbs => { let handler = PbsHandler::new(directory, name); handler.map::, _>(|handler| Box::new(handler)) @@ -177,6 +174,26 @@ pub fn create_allocation_handler( let handler = SlurmHandler::new(directory, name); handler.map::, _>(|handler| Box::new(handler)) } + ManagerType::Firecrest => { + let config = params.firecrest.clone().ok_or_else(|| { + anyhow::anyhow!("FirecREST queue parameters are missing in the queue definition") + })?; + #[cfg(feature = "firecrest")] + { + let handler = crate::server::autoalloc::queue::firecrest::FirecrestHandler::new( + directory, name, config, + ); + handler.map::, _>(|handler| Box::new(handler)) + } + #[cfg(not(feature = "firecrest"))] + { + let _ = (config, directory, name); + Err(anyhow::anyhow!( + "This hq build does not include FirecREST support. \ +Rebuild HyperQueue with the `firecrest` feature enabled." + )) + } + } } } @@ -683,7 +700,7 @@ fn create_queue( worker_resources: Option, ) -> anyhow::Result { let name = params.name.clone(); - let handler = create_allocation_handler(¶ms.manager, name.clone(), server_directory); + let handler = create_allocation_handler(¶ms, name.clone(), server_directory); let queue_info = QueueInfo::new(params.clone()); match handler { @@ -2681,6 +2698,7 @@ mod tests { cli_resource_descriptor: cli_resources, worker_args: vec![], idle_timeout: None, + firecrest: None, }; ( diff --git a/crates/hyperqueue/src/server/autoalloc/queue/common.rs b/crates/hyperqueue/src/server/autoalloc/queue/common.rs index 4f1a8bf10..786001282 100644 --- a/crates/hyperqueue/src/server/autoalloc/queue/common.rs +++ b/crates/hyperqueue/src/server/autoalloc/queue/common.rs @@ -138,7 +138,9 @@ pub fn build_worker_args( ) -> String { let manager = match manager { ManagerType::Pbs => "pbs", - ManagerType::Slurm => "slurm", + // FirecREST allocations are ordinary Slurm jobs on the target cluster, + // so the workers inside them use the Slurm manager. + ManagerType::Slurm | ManagerType::Firecrest => "slurm", }; let idle_timeout = params diff --git a/crates/hyperqueue/src/server/autoalloc/queue/firecrest.rs b/crates/hyperqueue/src/server/autoalloc/queue/firecrest.rs new file mode 100644 index 000000000..3a9f7435b --- /dev/null +++ b/crates/hyperqueue/src/server/autoalloc/queue/firecrest.rs @@ -0,0 +1,595 @@ +//! Allocation queue backend that submits Slurm allocations through a (remote) FirecREST API +//! (). +//! +//! Unlike the PBS/Slurm backends, this backend does not assume that the server runs on the +//! target cluster: submission, status queries and cancellation are performed via HTTPS +//! requests, authenticated with an OAuth2 client-credentials token. The spawned workers then +//! connect back to the server over TCP, exactly like manually started workers. +//! +//! Because the server is not on the cluster, the queue carries cluster-side paths +//! (`hq` binary, worker access file directory, working directory) in +//! [`FirecrestQueueParams`]; see its documentation for details. + +use std::future::Future; +use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::rc::Rc; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use tokio::sync::Mutex; + +use anyhow::Context; +use serde::Deserialize; + +use crate::common::manager::info::ManagerType; +use crate::common::utils::time::AbsoluteTime; +use crate::server::autoalloc::queue::common::{ + add_start_stop_worker_commands, build_worker_args, create_allocation_dir, + format_allocation_name, +}; +use crate::server::autoalloc::queue::slurm::build_slurm_submit_script; +use crate::server::autoalloc::queue::{ + AllocationExternalStatus, AllocationStatusMap, AllocationSubmissionResult, QueueHandler, + SubmitMode, +}; +use crate::server::autoalloc::{ + Allocation, AutoAllocResult, FirecrestQueueParams, QueueId, QueueInfo, +}; +use tako::Map; + +/// How long before its expiration do we consider a cached token unusable. +/// CSCS tokens are valid for ~5 minutes; a request started just before the expiration +/// could otherwise be rejected mid-flight. +const TOKEN_EXPIRATION_SLACK: Duration = Duration::from_secs(30); + +const HTTP_TIMEOUT: Duration = Duration::from_secs(30); + +#[derive(Clone)] +struct CachedToken { + token: String, + expires_at: Instant, +} + +/// Shared context for performing authenticated FirecREST requests. +/// It is cheaply cloneable so that it can be moved into the `'static` futures returned +/// by the [`QueueHandler`] methods. +/// +/// The token cache is guarded by an async mutex that is held across the token fetch, +/// so that concurrent requests do not each fetch their own token; they wait for the +/// first fetch and then reuse the cached result. +#[derive(Clone)] +struct FirecrestContext { + client: reqwest::Client, + config: Rc, + client_secret: Rc, + token: Rc>>, +} + +#[derive(Deserialize)] +struct TokenResponse { + access_token: String, + expires_in: u64, +} + +#[derive(Deserialize)] +struct SubmitResponse { + #[serde(rename = "jobId")] + job_id: Option, +} + +#[derive(Deserialize)] +struct JobStatus { + state: String, +} + +#[derive(Deserialize)] +struct JobTime { + start: Option, + end: Option, +} + +#[derive(Deserialize)] +struct JobModel { + #[serde(rename = "jobId", default)] + job_id: Option, + status: JobStatus, + time: Option, +} + +impl JobModel { + fn job_id_string(&self) -> Option { + self.job_id.as_ref().map(json_value_to_string) + } +} + +/// The job id is documented as a string, but be lenient if it arrives as a number. +fn json_value_to_string(value: &serde_json::Value) -> String { + match value { + serde_json::Value::String(id) => id.clone(), + other => other.to_string(), + } +} + +#[derive(Deserialize)] +struct GetJobResponse { + jobs: Option>, +} + +impl FirecrestContext { + /// Returns a valid bearer token, fetching a fresh one through the OAuth2 + /// client-credentials grant if the cached one is missing or about to expire. + async fn bearer_token(&self) -> AutoAllocResult { + let mut cached = self.token.lock().await; + if let Some(token) = cached.as_ref() + && token.expires_at > Instant::now() + TOKEN_EXPIRATION_SLACK + { + return Ok(token.token.clone()); + } + + log::debug!( + "Fetching FirecREST auth token from {}", + self.config.token_url + ); + let response = self + .client + .post(&self.config.token_url) + .form(&[ + ("grant_type", "client_credentials"), + ("client_id", &self.config.client_id), + ("client_secret", &self.client_secret), + ]) + .send() + .await + .context("Cannot reach the OAuth2 token endpoint")?; + let response = check_response(response) + .await + .context("OAuth2 token request failed")?; + let token: TokenResponse = response + .json() + .await + .context("Cannot parse OAuth2 token response")?; + + let fresh = CachedToken { + token: token.access_token, + expires_at: Instant::now() + Duration::from_secs(token.expires_in), + }; + *cached = Some(fresh.clone()); + Ok(fresh.token) + } + + /// Drops the cached token, but only if it is still the one that was just rejected; + /// a concurrent request may have already fetched a fresh one. + async fn invalidate_token(&self, rejected: &str) { + let mut cached = self.token.lock().await; + if cached.as_ref().is_some_and(|token| token.token == rejected) { + *cached = None; + } + } + + /// Sends an authenticated request. If the API rejects the token (HTTP 401) even + /// though it has not expired yet (e.g. it was revoked, or the API and the token + /// endpoint disagree about clocks), the request is retried once with a fresh token. + async fn request_with_auth_retry( + &self, + build: impl Fn(String) -> reqwest::RequestBuilder, + ) -> AutoAllocResult { + let token = self.bearer_token().await?; + let response = build(token.clone()) + .send() + .await + .context("Cannot reach the FirecREST API")?; + if response.status() != reqwest::StatusCode::UNAUTHORIZED { + return Ok(response); + } + + log::debug!("FirecREST rejected the auth token, retrying with a fresh one"); + self.invalidate_token(&token).await; + let token = self.bearer_token().await?; + build(token) + .send() + .await + .context("Cannot reach the FirecREST API") + } + + fn jobs_url(&self) -> String { + format!( + "{}/compute/{}/jobs", + self.config.api_url.trim_end_matches('/'), + self.config.system + ) + } + + /// Submits the given submit script and returns the created Slurm job id. + async fn submit_job(&self, name: &str, script: &str) -> AutoAllocResult { + let url = self.jobs_url(); + let body = serde_json::json!({ + "job": { + "name": name, + "workingDirectory": self.config.remote_workdir, + "script": script, + } + }); + let response = self + .request_with_auth_retry(|token| self.client.post(&url).bearer_auth(token).json(&body)) + .await?; + let response = check_response(response) + .await + .context("FirecREST job submission failed")?; + let response: SubmitResponse = response + .json() + .await + .context("Cannot parse FirecREST submit response")?; + + let job_id = response + .job_id + .ok_or_else(|| anyhow::anyhow!("FirecREST submit response is missing the job id"))?; + Ok(json_value_to_string(&job_id)) + } + + /// Fetches all jobs currently visible to the user in one request and returns them + /// keyed by job id. Used to refresh the status of all allocations at once, which + /// keeps the number of API requests (and thus rate-limit pressure) independent of + /// the number of active allocations. + async fn list_jobs(&self) -> AutoAllocResult> { + let url = self.jobs_url(); + let response = self + .request_with_auth_retry(|token| self.client.get(&url).bearer_auth(token)) + .await?; + let response = check_response(response) + .await + .context("FirecREST job list query failed")?; + let response: GetJobResponse = response + .json() + .await + .context("Cannot parse FirecREST job list response")?; + + Ok(response + .jobs + .unwrap_or_default() + .into_iter() + .filter_map(|job| job.job_id_string().map(|id| (id, job))) + .collect()) + } + + async fn job_status(&self, allocation_id: &str) -> AutoAllocResult { + let url = format!("{}/{allocation_id}", self.jobs_url()); + let response = self + .request_with_auth_retry(|token| self.client.get(&url).bearer_auth(token)) + .await?; + let response = check_response(response) + .await + .with_context(|| format!("FirecREST status query for job {allocation_id} failed"))?; + let response: GetJobResponse = response + .json() + .await + .context("Cannot parse FirecREST job status response")?; + + let job = response + .jobs + .unwrap_or_default() + .into_iter() + .next() + .ok_or_else(|| anyhow::anyhow!("FirecREST returned no record for the job"))?; + parse_job_status(&job) + } + + async fn cancel_job(&self, allocation_id: &str) -> AutoAllocResult<()> { + let url = format!("{}/{allocation_id}", self.jobs_url()); + let response = self + .request_with_auth_retry(|token| self.client.delete(&url).bearer_auth(token)) + .await?; + // A job that has already finished or was removed from the scheduler's memory + // cannot be canceled, which is fine for our purposes. + if response.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(()); + } + check_response(response) + .await + .with_context(|| format!("FirecREST cancellation of job {allocation_id} failed"))?; + Ok(()) + } +} + +async fn check_response(response: reqwest::Response) -> AutoAllocResult { + let status = response.status(); + if status.is_success() { + Ok(response) + } else { + let body = response.text().await.unwrap_or_default(); + Err(anyhow::anyhow!("HTTP status {status}: {}", body.trim())) + } +} + +fn parse_job_status(job: &JobModel) -> AutoAllocResult { + let epoch_to_time = |secs: i64| -> AbsoluteTime { + AbsoluteTime::from(UNIX_EPOCH + Duration::from_secs(secs.max(0) as u64)) + }; + let started_at = job + .time + .as_ref() + .and_then(|time| time.start) + .map(epoch_to_time); + let finished_at = job + .time + .as_ref() + .and_then(|time| time.end) + .map(epoch_to_time) + .unwrap_or_else(|| AbsoluteTime::from(SystemTime::now())); + + // FirecREST reports Slurm job states; the mapping mirrors `slurm.rs::parse_slurm_status`. + // Cancelled jobs can be reported e.g. as `CANCELLED by `. + let state = job.status.state.as_str(); + let status = match state { + "PENDING" | "CONFIGURING" | "REQUEUED" | "SUSPENDED" => AllocationExternalStatus::Queued, + "RUNNING" | "COMPLETING" => AllocationExternalStatus::Running, + "COMPLETED" | "TIMEOUT" => AllocationExternalStatus::Finished { + started_at, + finished_at, + }, + _ if state == "FAILED" + || state.starts_with("CANCELLED") + || state == "NODE_FAIL" + || state == "OUT_OF_MEMORY" + || state == "PREEMPTED" + || state == "BOOT_FAIL" + || state == "DEADLINE" => + { + AllocationExternalStatus::Failed { + started_at, + finished_at, + } + } + other => anyhow::bail!("Unknown Slurm job status {other}"), + }; + Ok(status) +} + +pub struct FirecrestHandler { + ctx: FirecrestContext, + server_directory: PathBuf, + name: Option, + allocation_counter: u64, +} + +impl FirecrestHandler { + pub fn new( + server_directory: PathBuf, + name: Option, + config: FirecrestQueueParams, + ) -> anyhow::Result { + let client_secret = std::env::var(&config.client_secret_env).map_err(|_| { + anyhow::anyhow!( + "The environment variable `{}` with the FirecREST client secret is not set \ +in the environment of the server", + config.client_secret_env + ) + })?; + let client = reqwest::Client::builder() + .timeout(HTTP_TIMEOUT) + .build() + .context("Cannot create HTTP client")?; + + Ok(Self { + ctx: FirecrestContext { + client, + config: Rc::new(config), + client_secret: Rc::new(client_secret), + token: Rc::new(Mutex::new(None)), + }, + server_directory, + name, + allocation_counter: 0, + }) + } + + fn create_allocation_id(&mut self) -> u64 { + self.allocation_counter += 1; + self.allocation_counter + } +} + +impl QueueHandler for FirecrestHandler { + fn submit_allocation( + &mut self, + queue_id: QueueId, + queue_info: &QueueInfo, + worker_count: u64, + _mode: SubmitMode, + ) -> Pin>>> { + let params = queue_info.params().clone(); + let ctx = self.ctx.clone(); + let name = self.name.clone(); + let server_directory = self.server_directory.clone(); + let allocation_num = self.create_allocation_id(); + + Box::pin(async move { + // Local directory holding a debug copy of the submitted script + let working_dir = create_allocation_dir( + server_directory.clone(), + queue_id, + name.as_ref(), + allocation_num, + )?; + + let allocation_name = format_allocation_name(name, queue_id, allocation_num); + + let worker_args = build_worker_args( + Path::new(&ctx.config.remote_hq_path), + // The allocation is an ordinary Slurm job on the target cluster + ManagerType::Slurm, + Path::new(&ctx.config.remote_server_dir), + ¶ms, + ); + let worker_args = add_start_stop_worker_commands( + worker_args, + params.worker_start_cmd.as_deref(), + params.worker_stop_cmd.as_deref(), + ); + + // stdout/stderr live directly in the remote working directory, which is + // required to exist; Slurm does not create missing directories for them. + // These are paths on the target cluster: compose them as strings. + let remote_workdir = ctx.config.remote_workdir.trim_end_matches('/'); + let stdout = format!("{remote_workdir}/{allocation_name}.stdout"); + let stderr = format!("{remote_workdir}/{allocation_name}.stderr"); + + let script = build_slurm_submit_script( + worker_count, + params.timelimit, + &allocation_name, + &stdout, + &stderr, + ¶ms.additional_args.join(" "), + &worker_args, + ); + + // Keep a debug copy of the script, consistent with the PBS/Slurm backends + std::fs::write(working_dir.submit_script(), &script) + .context("Cannot write a debug copy of the submit script")?; + + let id = ctx.submit_job(&allocation_name, &script).await; + if let Ok(id) = &id { + std::fs::write(working_dir.jobid_file(), id)?; + } + + Ok(AllocationSubmissionResult::new(id, working_dir)) + }) + } + + fn get_status_of_allocations( + &self, + allocations: &[&Allocation], + ) -> Pin>>> { + let allocation_ids: Vec = + allocations.iter().map(|alloc| alloc.id.clone()).collect(); + let ctx = self.ctx.clone(); + + Box::pin(async move { + let mut result = Map::with_capacity(allocation_ids.len()); + if allocation_ids.is_empty() { + return Ok(result); + } + + // A single list request covers all allocations; only allocations that have + // already dropped out of the list (the API merges squeue and recent sacct + // history) are queried individually. + let jobs = ctx.list_jobs().await?; + for allocation_id in allocation_ids { + let status = match jobs.get(&allocation_id) { + Some(job) => parse_job_status(job), + None => ctx.job_status(&allocation_id).await, + }; + result.insert(allocation_id, status); + } + Ok(result) + }) + } + + fn remove_allocation( + &self, + allocation: &Allocation, + ) -> Pin>>> { + let allocation_id = allocation.id.clone(); + let ctx = self.ctx.clone(); + + Box::pin(async move { ctx.cancel_job(&allocation_id).await }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn job(state: &str, start: Option, end: Option) -> JobModel { + JobModel { + job_id: None, + status: JobStatus { + state: state.to_string(), + }, + time: Some(JobTime { start, end }), + } + } + + #[test] + fn job_id_accepts_string_and_number() { + let parse = |json: &str| -> Option { + serde_json::from_str::(json) + .unwrap() + .job_id_string() + }; + assert_eq!( + parse(r#"{"jobId": "123", "status": {"state": "PENDING"}}"#), + Some("123".to_string()) + ); + assert_eq!( + parse(r#"{"jobId": 123, "status": {"state": "PENDING"}}"#), + Some("123".to_string()) + ); + assert_eq!(parse(r#"{"status": {"state": "PENDING"}}"#), None); + } + + #[test] + fn map_queued_and_running_states() { + for state in ["PENDING", "CONFIGURING", "REQUEUED", "SUSPENDED"] { + assert!(matches!( + parse_job_status(&job(state, None, None)).unwrap(), + AllocationExternalStatus::Queued + )); + } + for state in ["RUNNING", "COMPLETING"] { + assert!(matches!( + parse_job_status(&job(state, Some(1), None)).unwrap(), + AllocationExternalStatus::Running + )); + } + } + + #[test] + fn map_finished_states() { + assert!(matches!( + parse_job_status(&job("COMPLETED", Some(1), Some(2))).unwrap(), + AllocationExternalStatus::Finished { .. } + )); + assert!(matches!( + parse_job_status(&job("TIMEOUT", Some(1), Some(2))).unwrap(), + AllocationExternalStatus::Finished { .. } + )); + } + + #[test] + fn map_failed_states() { + for state in [ + "FAILED", + "CANCELLED", + "CANCELLED by 12345", + "NODE_FAIL", + "OUT_OF_MEMORY", + "PREEMPTED", + "BOOT_FAIL", + "DEADLINE", + ] { + assert!(matches!( + parse_job_status(&job(state, Some(1), Some(2))).unwrap(), + AllocationExternalStatus::Failed { .. } + )); + } + } + + #[test] + fn unknown_state_is_an_error() { + assert!(parse_job_status(&job("FROBNICATING", None, None)).is_err()); + } + + #[test] + fn finished_without_end_time_falls_back_to_now() { + let before = SystemTime::now(); + let status = parse_job_status(&job("COMPLETED", None, None)).unwrap(); + let after = SystemTime::now(); + match status { + AllocationExternalStatus::Finished { finished_at, .. } => { + assert!(before <= finished_at.inner() && finished_at.inner() <= after); + } + _ => panic!("expected a finished allocation"), + } + } +} diff --git a/crates/hyperqueue/src/server/autoalloc/queue/mod.rs b/crates/hyperqueue/src/server/autoalloc/queue/mod.rs index 59d56e3e9..41c207d13 100644 --- a/crates/hyperqueue/src/server/autoalloc/queue/mod.rs +++ b/crates/hyperqueue/src/server/autoalloc/queue/mod.rs @@ -1,4 +1,6 @@ mod common; +#[cfg(feature = "firecrest")] +pub mod firecrest; pub mod pbs; pub mod slurm; @@ -34,6 +36,39 @@ pub struct QueueParameters { // Black-box worker args that will be passed to `worker start` pub worker_args: Vec, pub idle_timeout: Option, + + /// Configuration of the FirecREST backend. + /// Has to be filled when `manager` is [`ManagerType::Firecrest`]. + #[serde(default)] + pub firecrest: Option, +} + +/// Parameters for submitting allocations through a (remote) FirecREST API. +/// +/// Note that the server submitting through FirecREST usually does NOT run on the target +/// cluster, therefore all paths in this struct are paths *on the target cluster*, and they +/// are intentionally stored as strings (they must not be interpreted in terms of the +/// filesystem of the machine where the server runs). +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct FirecrestQueueParams { + /// Base URL of the FirecREST API, e.g. `https://api.cscs.ch/hpc/firecrest/v2` + pub api_url: String, + /// Name of the target system (cluster), e.g. `daint` + pub system: String, + /// OAuth2 token endpoint used for the client credentials grant + pub token_url: String, + /// OAuth2 client ID + pub client_id: String, + /// Name of an environment variable (in the environment of the **server**) that holds + /// the OAuth2 client secret. The secret itself is deliberately never stored here, + /// because queue parameters are persisted into the journal. + pub client_secret_env: String, + /// Path to the `hq` binary on the target cluster + pub remote_hq_path: String, + /// Directory on the target cluster that contains the worker access file (`access.json`) + pub remote_server_dir: String, + /// Existing directory on the target cluster where allocation stdout/stderr are written + pub remote_workdir: String, } // Wrapper that exposes some methods and doesn't allow pub access to the queue parameter diff --git a/crates/hyperqueue/src/server/autoalloc/queue/slurm.rs b/crates/hyperqueue/src/server/autoalloc/queue/slurm.rs index ba12fbe54..afd6f3346 100644 --- a/crates/hyperqueue/src/server/autoalloc/queue/slurm.rs +++ b/crates/hyperqueue/src/server/autoalloc/queue/slurm.rs @@ -189,7 +189,7 @@ fn parse_slurm_status(items: Map<&str, &str>) -> AutoAllocResult \ + --remote-hq-path /users//hq \ + --remote-server-dir /users//hq-worker-dir \ + --remote-workdir /scratch//hq-alloc \ + --time-limit 1h -- --account= --partition= +``` + +The queue behaves like a Slurm allocation queue (the allocations are ordinary Slurm jobs, trailing arguments are +passed to `sbatch` via `#SBATCH` directives, workers detect the Slurm environment), and all the generic +[parameters](#parameters) apply. The additional parameters are: + +- **`--api-url`**: base URL of the FirecREST API. +- **`--system`**: name of the target cluster as known to FirecREST. +- **`--token-url`**: OAuth2 token endpoint. HyperQueue authenticates with the client-credentials grant and manages + (caches, refreshes) the short-lived access tokens automatically. +- **`--client-id`**: OAuth2 client ID. +- **`--client-secret-env`**: name of an environment variable that holds the OAuth2 client secret + (`HQ_FIRECREST_CLIENT_SECRET` by default). The variable has to be set **in the environment of the server**. Only the + variable name is stored in the queue configuration and in the journal; the secret itself is never persisted. This + also means that when you restore a server from a [journal](server.md#resuming-stoppedcrashed-server), the variable + has to be set again, otherwise the queue will not be restored. + +Because the server cannot assume anything about the cluster's filesystem, you also have to provide cluster-side paths +(they are used verbatim on the cluster and are not interpreted by the server): + +- **`--remote-hq-path`**: path to the `hq` binary **on the cluster**, used to start the workers. It should be the same + version as the server binary. +- **`--remote-server-dir`**: directory **on the cluster** containing the worker access file (`access.json`), which + tells the workers where the server runs and provides authentication keys. Generate it with + [`hq server generate-access`](cloud.md#splitting-access-for-client-and-workers) and copy it to the cluster. +- **`--remote-workdir`**: directory **on the cluster** used as the working directory of the allocations (stdout/stderr + of the allocations are placed there). It has to exist before the first allocation starts. + +!!! important "Network requirements" + + The server itself only needs outbound HTTPS access to the FirecREST API. The spawned workers, however, initiate a + TCP connection **from the compute nodes to the server** (to the worker port advertised in the access file), so the + server has to run on an address reachable from the compute nodes. If the cluster firewall does not allow that + connection, workers will fail to connect: you can either run the server somewhere reachable (e.g. inside the + cluster's network) or tunnel the worker port (e.g. with a reverse SSH tunnel). + + If the server runs outside the cluster's network, this typically means it needs a public IP address with the worker + port open for inbound connections. The access file has to advertise that address, otherwise the workers will try + to connect to the server machine's local hostname. Set it with `--worker-host` when generating the access file + (`hq server generate-access ... --worker-host=`), see + [different hostnames for workers and clients](cloud.md#setting-different-server-hostname-for-workers-and-clients). + +!!! tip "Submitting jobs from outside the cluster" + + When the client submitting HyperQueue jobs also runs outside the cluster, the default working directory and + stdout/stderr paths of tasks are derived from the client's submission directory, which most likely does not exist + on the cluster. Pass explicit cluster-side paths to `hq submit` using `--cwd`, `--stdout` and `--stderr`. + ## Behavior The (automatic) allocator submits allocations into PBS/Slurm based on current computational demand. When an allocation diff --git a/tests/autoalloc/mock/firecrest.py b/tests/autoalloc/mock/firecrest.py new file mode 100644 index 000000000..f47d165fb --- /dev/null +++ b/tests/autoalloc/mock/firecrest.py @@ -0,0 +1,213 @@ +""" +In-process mock of a Keycloak token endpoint + a FirecREST v2 API server, used to test +the `firecrest` automatic allocation backend. + +Unlike the PBS/Slurm mocks, which replace scheduler binaries (`qsub`, `sbatch`, ...), +the firecrest backend talks to an HTTP API, so this mock runs a real HTTP server on an +ephemeral localhost port inside the pytest process. + +Mocked endpoints (response shapes follow the firecrest-v2 sources): + POST /token -> {"access_token", "expires_in"} + POST /compute/{system}/jobs -> {"jobId": ""} + GET /compute/{system}/jobs -> {"jobs": [...]} (includes "jobId") + GET /compute/{system}/jobs/{id} -> {"jobs": []} or 404 + DELETE /compute/{system}/jobs/{id} -> 204 (job state -> CANCELLED) or 404 +""" + +import json +import re +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Dict, List, Optional + + +class MockFirecrest: + def __init__(self): + self.jobs: Dict[str, dict] = {} + # Submissions as dicts with "name", "working_directory" and "script" keys + self.submitted: List[dict] = [] + self.token_counter = 0 + # "METHOD endpoint" -> number of requests + self.counters: Dict[str, int] = {} + self.deleted_jobs: List[str] = [] + self.fail_api: List[int] = [] + self.fail_token: List[int] = [] + self.token_expires_in = 300 + self.lock = threading.Lock() + + self.server: Optional[ThreadingHTTPServer] = None + self.thread: Optional[threading.Thread] = None + self.port: Optional[int] = None + + @property + def api_url(self) -> str: + return f"http://127.0.0.1:{self.port}" + + @property + def token_url(self) -> str: + return f"{self.api_url}/token" + + def set_job_state(self, job_id: str, state: str): + with self.lock: + job = self.jobs[job_id] + job["state"] = state + now = int(time.time()) + if state == "RUNNING": + job["start"] = now + elif state not in ("PENDING", "CONFIGURING"): + job["start"] = job["start"] or now + job["end"] = now + + def fail_next_api_request(self, code: int, count: int = 1): + with self.lock: + self.fail_api.extend([code] * count) + + def fail_next_token_request(self, code: int, count: int = 1): + with self.lock: + self.fail_token.extend([code] * count) + + def counter(self, key: str) -> int: + with self.lock: + return self.counters.get(key, 0) + + def __enter__(self) -> "MockFirecrest": + mock = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def reply(self, code: int, payload=None): + body = json.dumps(payload).encode() if payload is not None else b"" + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def count(self, key: str): + with mock.lock: + mock.counters[key] = mock.counters.get(key, 0) + 1 + + def check_auth(self) -> bool: + auth = self.headers.get("Authorization", "") + if not auth.startswith("Bearer mock-token-"): + self.reply(401, {"message": "missing or invalid bearer token"}) + return False + return True + + def injected_fault(self) -> bool: + with mock.lock: + if mock.fail_api: + code = mock.fail_api.pop(0) + self.reply(code, {"message": f"injected failure {code}"}) + return True + return False + + def read_body(self) -> bytes: + length = int(self.headers.get("Content-Length") or 0) + return self.rfile.read(length) if length else b"" + + def do_POST(self): + body = self.read_body() + + if self.path == "/token": + self.count("POST /token") + with mock.lock: + if mock.fail_token: + code = mock.fail_token.pop(0) + self.reply(code, {"error": f"injected token failure {code}"}) + return + mock.token_counter += 1 + n = mock.token_counter + expires = mock.token_expires_in + if b"client_id=" not in body or b"client_secret=" not in body: + self.reply(401, {"error": "invalid_client"}) + return + self.reply(200, {"access_token": f"mock-token-{n}", "expires_in": expires}) + return + + if re.fullmatch(r"/compute/[^/]+/jobs", self.path): + self.count("POST /compute/jobs") + if not self.check_auth() or self.injected_fault(): + return + payload = json.loads(body)["job"] + with mock.lock: + job_id = str(len(mock.jobs) + 1) + mock.jobs[job_id] = {"state": "PENDING", "start": None, "end": None} + mock.submitted.append( + { + "job_id": job_id, + "name": payload["name"], + "working_directory": payload["workingDirectory"], + "script": payload["script"], + } + ) + self.reply(200, {"jobId": job_id}) + return + + self.reply(404, {"message": "unknown endpoint"}) + + def do_GET(self): + if re.fullmatch(r"/compute/[^/]+/jobs", self.path): + self.count("GET /compute/jobs") + if not self.check_auth() or self.injected_fault(): + return + with mock.lock: + jobs = [job_response(job_id, job) for (job_id, job) in mock.jobs.items()] + self.reply(200, {"jobs": jobs}) + return + + m = re.fullmatch(r"/compute/[^/]+/jobs/([^/]+)", self.path) + if m: + self.count("GET /compute/jobs/{id}") + if not self.check_auth() or self.injected_fault(): + return + with mock.lock: + job = mock.jobs.get(m.group(1)) + payload = {"jobs": [job_response(m.group(1), job)]} if job else None + if payload is None: + self.reply(404, {"message": "job not found"}) + else: + self.reply(200, payload) + return + + self.reply(404, {"message": "unknown endpoint"}) + + def do_DELETE(self): + m = re.fullmatch(r"/compute/[^/]+/jobs/([^/]+)", self.path) + if m: + self.count("DELETE /compute/jobs/{id}") + if not self.check_auth() or self.injected_fault(): + return + with mock.lock: + job = mock.jobs.get(m.group(1)) + if job is None: + self.reply(404, {"message": "job not found"}) + return + job["state"] = "CANCELLED" + job["end"] = int(time.time()) + mock.deleted_jobs.append(m.group(1)) + self.reply(204) + return + self.reply(404, {"message": "unknown endpoint"}) + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.port = self.server.server_address[1] + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + +def job_response(job_id: str, job: dict) -> dict: + return { + "jobId": job_id, + "status": {"state": job["state"]}, + "time": {"start": job["start"], "end": job["end"]}, + } diff --git a/tests/autoalloc/test_firecrest.py b/tests/autoalloc/test_firecrest.py new file mode 100644 index 000000000..639a2ba7c --- /dev/null +++ b/tests/autoalloc/test_firecrest.py @@ -0,0 +1,256 @@ +import os +from typing import List, Optional + +from ..conftest import HqEnv +from ..utils.wait import wait_until +from .mock.firecrest import MockFirecrest +from .utils import ( + extract_script_args, + prepare_tasks, + remove_queue, + wait_for_alloc, +) + +CLIENT_SECRET = "mock-client-secret-12345" +REMOTE_HQ_PATH = "/remote/bin/hq" +REMOTE_SERVER_DIR = "/remote/server-dir" +REMOTE_WORKDIR = "/remote/workdir" +DEFAULT_TIME_LIMIT = "3m" + + +def start_server(hq_env: HqEnv, secret: Optional[str] = CLIENT_SECRET, **kwargs): + """ + Starts the HQ server with fast autoalloc refresh and (by default) the FirecREST + client secret in its environment. + """ + env = { + "HQ_AUTOALLOC_REFRESH_INTERVAL_MS": "100", + "HQ_AUTOALLOC_MAX_SCHEDULE_DELAY_MS": "100", + "HQ_AUTOALLOC_SCHEDULE_TICK_INTERVAL_MS": "100", + } + if secret is not None: + env["HQ_FIRECREST_CLIENT_SECRET"] = secret + return hq_env.start_server(env=env, **kwargs) + + +def firecrest_queue_args( + mock: MockFirecrest, + time_limit: str = DEFAULT_TIME_LIMIT, + additional_args: Optional[str] = None, +) -> List[str]: + args = [ + "--api-url", + mock.api_url, + "--system", + "daint", + "--token-url", + mock.token_url, + "--client-id", + "test-client", + "--remote-hq-path", + REMOTE_HQ_PATH, + "--remote-server-dir", + REMOTE_SERVER_DIR, + "--remote-workdir", + REMOTE_WORKDIR, + "--time-limit", + time_limit, + ] + if additional_args is not None: + args.append("--") + args.extend(additional_args.split(" ")) + return args + + +def add_queue( + hq_env: HqEnv, + mock: MockFirecrest, + dry_run=False, + time_limit: str = DEFAULT_TIME_LIMIT, + additional_args: Optional[str] = None, + expect_fail: Optional[str] = None, +) -> str: + args = ["alloc", "add", "firecrest"] + if not dry_run: + args.append("--no-dry-run") + args += firecrest_queue_args(mock, time_limit=time_limit, additional_args=additional_args) + return hq_env.command(args, expect_fail=expect_fail) + + +def test_firecrest_submit_script(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + prepare_tasks(hq_env) + + add_queue(hq_env, mock, time_limit="3m", additional_args="--account=foo --partition=bar") + wait_until(lambda: len(mock.submitted) > 0) + + submission = mock.submitted[0] + assert submission["name"] == "hq-1-1" + assert submission["working_directory"] == REMOTE_WORKDIR + + script = submission["script"] + assert extract_script_args(script, "#SBATCH") == [ + "--nodes=1", + "--job-name=hq-1-1", + f"--output={REMOTE_WORKDIR}/hq-1-1.stdout", + f"--error={REMOTE_WORKDIR}/hq-1-1.stderr", + "--time=00:03:00", + "--account=foo --partition=bar", + ] + worker_command = script.splitlines()[-1] + assert f"{REMOTE_HQ_PATH} worker start" in worker_command + assert '--manager "slurm"' in worker_command + assert f'--server-dir "{REMOTE_SERVER_DIR}"' in worker_command + + +def test_firecrest_dry_run_success(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + hq_env.command(["alloc", "dry-run", "firecrest"] + firecrest_queue_args(mock)) + + # The trial allocation should be submitted and immediately canceled + assert mock.deleted_jobs == ["1"] + assert mock.jobs["1"]["state"] == "CANCELLED" + + +def test_firecrest_dry_run_token_error(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + mock.fail_next_token_request(401) + hq_env.command( + ["alloc", "dry-run", "firecrest"] + firecrest_queue_args(mock), + expect_fail="OAuth2 token request failed", + ) + + +def test_firecrest_dry_run_submit_error(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + mock.fail_next_api_request(500) + hq_env.command( + ["alloc", "dry-run", "firecrest"] + firecrest_queue_args(mock), + expect_fail="FirecREST job submission failed", + ) + + +def test_firecrest_add_queue_missing_secret(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env, secret=None) + add_queue( + hq_env, + mock, + expect_fail="The environment variable `HQ_FIRECREST_CLIENT_SECRET` with the FirecREST client secret is not set", + ) + + +def test_firecrest_allocation_lifecycle(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + prepare_tasks(hq_env) + + add_queue(hq_env, mock) + wait_for_alloc(hq_env, "QUEUED", "1") + + mock.set_job_state("1", "RUNNING") + wait_for_alloc(hq_env, "RUNNING", "1") + + mock.set_job_state("1", "COMPLETED") + wait_for_alloc(hq_env, "FINISHED", "1") + + +def test_firecrest_allocation_failure(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + prepare_tasks(hq_env) + + add_queue(hq_env, mock) + wait_for_alloc(hq_env, "QUEUED", "1") + + mock.set_job_state("1", "NODE_FAIL") + wait_for_alloc(hq_env, "FAILED", "1") + + +def test_firecrest_cancel_on_queue_remove(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + prepare_tasks(hq_env) + + add_queue(hq_env, mock) + wait_for_alloc(hq_env, "QUEUED", "1") + + remove_queue(hq_env, 1, force=True) + wait_until(lambda: mock.jobs["1"]["state"] == "CANCELLED") + + +def test_firecrest_status_queries_are_batched(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + prepare_tasks(hq_env) + + add_queue(hq_env, mock) + wait_for_alloc(hq_env, "QUEUED", "1") + + # Let several refresh cycles pass; the status of allocations should be + # queried through the job-list endpoint, not per job id + wait_until(lambda: mock.counter("GET /compute/jobs") >= 3) + assert mock.counter("GET /compute/jobs/{id}") == 0 + + +def test_firecrest_retry_with_fresh_token_on_401(hq_env: HqEnv): + with MockFirecrest() as mock: + start_server(hq_env) + prepare_tasks(hq_env) + + add_queue(hq_env, mock) + wait_for_alloc(hq_env, "QUEUED", "1") + + tokens_before = mock.token_counter + mock.fail_next_api_request(401) + + # The 401 should be recovered from transparently: a fresh token is fetched + # and the allocation status keeps being refreshed without errors + wait_until(lambda: mock.token_counter == tokens_before + 1) + mock.set_job_state("1", "RUNNING") + wait_for_alloc(hq_env, "RUNNING", "1") + + +def test_firecrest_journal_restore(hq_env: HqEnv, tmp_path): + journal_path = os.path.join(tmp_path, "journal") + with MockFirecrest() as mock: + start_server(hq_env, args=["--journal", journal_path]) + add_queue(hq_env, mock) + hq_env.stop_server() + + start_server(hq_env, args=["--journal", journal_path]) + table = hq_env.command(["alloc", "list"], as_table=True) + table.check_columns_value(["ID", "Manager"], 0, ["1", "FirecREST"]) + + +def test_firecrest_journal_restore_missing_secret(hq_env: HqEnv, tmp_path): + journal_path = os.path.join(tmp_path, "journal") + with MockFirecrest() as mock: + start_server(hq_env, args=["--journal", journal_path]) + add_queue(hq_env, mock) + hq_env.stop_server() + + # Without the secret the queue cannot be restored; the server should + # nevertheless start and work + start_server(hq_env, secret=None, args=["--journal", journal_path]) + table = hq_env.command(["alloc", "list"], as_table=True) + assert len(table) == 0 + + +def test_firecrest_secret_not_persisted(hq_env: HqEnv, tmp_path): + journal_path = os.path.join(tmp_path, "journal") + with MockFirecrest() as mock: + start_server(hq_env, args=["--journal", journal_path]) + add_queue(hq_env, mock) + hq_env.stop_server() + + with open(journal_path, "rb") as f: + journal = f.read() + # The journal stores the name of the environment variable holding the secret, + # never the secret itself + assert CLIENT_SECRET.encode() not in journal + assert b"HQ_FIRECREST_CLIENT_SECRET" in journal