From 2003f0cf14da64ff7021743f43d3464e200035f0 Mon Sep 17 00:00:00 2001 From: Patrick Meade Date: Mon, 9 Dec 2024 17:04:17 -0600 Subject: [PATCH] RESTful server with status and shutdown --- .gitignore | 3 + Cargo.lock | 525 +++++++++++++++++++++++++++---- Cargo.toml | 16 +- bin/setup | 21 +- datamove.example.toml | 32 +- src/bin/disk_archiver.rs | 66 +++- src/config.rs | 39 ++- src/sps.rs | 1 + src/sps/context.rs | 9 +- src/sps/database.rs | 49 ++- src/sps/models.rs | 25 ++ src/sps/process/disk_archiver.rs | 294 ++++++++++++++++- src/sps/utils.rs | 201 ++++++++++++ src/status.rs | 1 + src/status/net.rs | 34 ++ src/status/sps.rs | 64 +++- tools/init-docker-mysql | 13 + tools/shutdown-disk-archiver | 6 + tools/status-disk-archiver | 6 + 19 files changed, 1274 insertions(+), 131 deletions(-) create mode 100644 src/sps/utils.rs create mode 100644 src/status/net.rs create mode 100755 tools/init-docker-mysql create mode 100755 tools/shutdown-disk-archiver create mode 100755 tools/status-disk-archiver diff --git a/.gitignore b/.gitignore index 6db1d01..ab3281a 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,6 @@ # don't version production configuration datamove.toml + +# don't version the sql directory +/sql diff --git a/Cargo.lock b/Cargo.lock index 9b5ddaa..eb3526f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -90,12 +90,115 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "async-trait" +version = "0.1.83" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "721cae7de5c34fbb2acd27e21e6d2cf7b886dce0c27388d46c4e6c47ea4318dd" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "autocfg" version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" +[[package]] +name = "axum" +version = "0.7.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" +dependencies = [ + "async-trait", + "axum-core", + "axum-macros", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "rustversion", + "serde", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper 1.0.2", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09f2bd6146b97ae3359fa0cc6d6b376d9539582c7b4220f041a33ec24c226199" +dependencies = [ + "async-trait", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "rustversion", + "sync_wrapper 1.0.2", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-extra" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c794b30c904f0a1c2fb7740f7df7f7972dfaa14ef6f57cb6178dc63e5dca2f04" +dependencies = [ + "axum", + "axum-core", + "bytes", + "fastrand", + "futures-util", + "http", + "http-body", + "http-body-util", + "mime", + "multer", + "pin-project-lite", + "serde", + "serde_json", + "tower", + "tower-layer", + "tower-service", + "typed-json", +] + +[[package]] +name = "axum-macros" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57d123550fa8d071b7255cb0cc04dc302baa6c8c4a79f55701552684d8399bce" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "backtrace" version = "0.3.74" @@ -131,15 +234,15 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" [[package]] name = "bytes" -version = "1.8.0" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ac0150caa2ae65ca5bd83f25c7de183dea78d4d366469f148435e2acfbad0da" +checksum = "325918d6fe32f23b19878fe4b34794ae41fc19ddbe53b10571a4874d44ffd39b" [[package]] name = "cc" -version = "1.2.1" +version = "1.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd9de9f2205d5ef3fd67e685b0df337994ddd4495e2a28d185500d0e1edfea47" +checksum = "27f657647bcff5394bf56c7317665bbf790a137a50eaaa5c6bfbb9e27a518f2d" dependencies = [ "shlex", ] @@ -152,9 +255,9 @@ checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" [[package]] name = "chrono" -version = "0.4.38" +version = "0.4.39" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a21f936df1771bf62b77f047b726c4625ff2e8aa607c01ec06e5a05bd8463401" +checksum = "7e36cc9d416881d2e24f9a963be5fb1cd90966419ac844274161d10488b3e825" dependencies = [ "android-tzdata", "iana-time-zone", @@ -213,9 +316,9 @@ dependencies = [ [[package]] name = "diesel" -version = "2.2.4" +version = "2.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "158fe8e2e68695bd615d7e4f3227c0727b151330d3e253b525086c348d055d5e" +checksum = "ccf1bedf64cdb9643204a36dd15b19a6ce8e7aa7f7b105868e9f1fad5ffa7d12" dependencies = [ "bitflags", "byteorder", @@ -279,6 +382,15 @@ version = "1.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "60b1af1c220855b6ceac025d3f6ecdd2b7c4894bfe9cd9bda4fbb4bc7c0d4cf0" +[[package]] +name = "encoding_rs" +version = "0.8.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +dependencies = [ + "cfg-if", +] + [[package]] name = "env_filter" version = "0.1.2" @@ -308,6 +420,24 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5" +[[package]] +name = "fastrand" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" + +[[package]] +name = "filetime" +version = "0.2.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35c0522e981e68cbfa8c3f978441a5f34b30b96e146b33cd3359176b50fe8586" +dependencies = [ + "cfg-if", + "libc", + "libredox", + "windows-sys 0.59.0", +] + [[package]] name = "fnv" version = "1.0.7" @@ -323,6 +453,39 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures-channel" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" + +[[package]] +name = "futures-task" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" + +[[package]] +name = "futures-util" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "pin-utils", +] + [[package]] name = "gimli" version = "0.31.1" @@ -331,9 +494,9 @@ checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f" [[package]] name = "hashbrown" -version = "0.15.1" +version = "0.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a9bfc1af68b1726ea47d3d5109de126281def866b33970e10fbab11b5dafab3" +checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289" [[package]] name = "heck" @@ -341,12 +504,6 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" -[[package]] -name = "hermit-abi" -version = "0.3.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024" - [[package]] name = "hostname" version = "0.4.0" @@ -358,12 +515,93 @@ dependencies = [ "windows", ] +[[package]] +name = "http" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f16ca2af56261c99fba8bac40a10251ce8188205a4c448fbb745a2e4daa76fea" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "793429d76616a256bcb62c2a2ec2bed781c8307e797e2598c50010f2bee2544f" +dependencies = [ + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d71d3574edd2771538b901e6549113b4006ece66150fb69c0fb6d9a2adae946" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "humantime" version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4" +[[package]] +name = "hyper" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97818827ef4f364230e16705d4706e2897df2bb60617d6ca15d598025a3c481f" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df2dcfbe0677734ab2f3ffa7fa7bfd4706bfdc1ef393f2ee30184aed67e631b4" +dependencies = [ + "bytes", + "futures-util", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", + "tower-service", +] + [[package]] name = "iana-time-zone" version = "0.1.61" @@ -534,9 +772,9 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.6.0" +version = "2.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "707907fe3c25f5424cce2cb7e1cbcafee6bdbe735ca90ef77c29e84591e5b9da" +checksum = "62f822373a4fe84d4bb149bf54e584a7f4abec90e072ed49cda0edea5b95471f" dependencies = [ "equivalent", "hashbrown", @@ -550,30 +788,42 @@ checksum = "7943c866cc5cd64cbc25b2e01621d07fa8eb2a1a23160ee81ce38704e97b8ecf" [[package]] name = "itoa" -version = "1.0.11" +version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49f1f14873335454500d59611f1cf4a4b0f786f9ac11f4312a78e4cf2566695b" +checksum = "d75a2a4b1b190afb6f5425f10f6a8f959d2ea0b9c2b1d79553551850539e4674" [[package]] name = "js-sys" -version = "0.3.72" +version = "0.3.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a88f1bda2bd75b0452a14784937d796722fdebfe50df998aeb3f0b7603019a9" +checksum = "6717b6b5b077764fb5966237269cb3c64edddde4b14ce42647430a78ced9e7b7" dependencies = [ + "once_cell", "wasm-bindgen", ] [[package]] name = "libc" -version = "0.2.162" +version = "0.2.168" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "18d287de67fe55fd7e1581fe933d965a5a9477b38e949cfa9f8574ef01506398" +checksum = "5aaeb2981e0606ca11d79718f8bb01164f1d6ed75080182d3abf017e6d244b6d" + +[[package]] +name = "libredox" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d" +dependencies = [ + "bitflags", + "libc", + "redox_syscall", +] [[package]] name = "litemap" -version = "0.7.3" +version = "0.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "643cb0b8d4fcc284004d5fd0d67ccf61dfffadb7f75e1e71bc420f4688a3a704" +checksum = "4ee93343901ab17bd981295f2cf0026d4ad018c7c31ba84549a4ddbb47a45104" [[package]] name = "lock_api" @@ -591,12 +841,24 @@ version = "0.4.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24" +[[package]] +name = "matchit" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" + [[package]] name = "memchr" version = "2.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78ca9ab1a0babb1e7d5695e3530886289c18cf2f87ec19a575a0abdce112e3a3" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "miniz_oxide" version = "0.8.0" @@ -608,21 +870,37 @@ dependencies = [ [[package]] name = "mio" -version = "1.0.2" +version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "80e04d1dcff3aae0704555fe5fee3bcfaf3d1fdf8a7e521d5b9d2b42acb52cec" +checksum = "2886843bf800fba2e3377cff24abf6379b4c4d5c6681eaf9ea5b0d15090450bd" dependencies = [ - "hermit-abi", "libc", "wasi", "windows-sys 0.52.0", ] +[[package]] +name = "multer" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83e87776546dc87511aa5ee218730c92b666d7264ab6ed41f9d215af9cd5224b" +dependencies = [ + "bytes", + "encoding_rs", + "futures-util", + "http", + "httparse", + "memchr", + "mime", + "spin", + "version_check", +] + [[package]] name = "mysqlclient-sys" -version = "0.4.1" +version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "478e2040dbc35c73927b77a2be91a496de19deab376a6982ed61e89592434619" +checksum = "6bbb9b017b98c4cde5802998113e182eecc1ebce8d47e9ea1697b9a623d53870" dependencies = [ "pkg-config", "vcpkg", @@ -687,6 +965,12 @@ version = "0.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "915a1e146535de9163f3987b8944ed8cf49a18bb0056bcebcdcece385cece4ff" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "pkg-config" version = "0.3.31" @@ -695,9 +979,9 @@ checksum = "953ec861398dccce10c670dfeaf3ec4911ca479e9c02154b3a215178c5f566f2" [[package]] name = "proc-macro2" -version = "1.0.89" +version = "1.0.92" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f139b0662de085916d1fb67d2b4169d1addddda1919e696f3252b740b629986e" +checksum = "37d3544b3f2748c54e147655edb5025752e2303145b5aefb3c3ea2c78b973bb0" dependencies = [ "unicode-ident", ] @@ -755,6 +1039,12 @@ version = "0.1.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f" +[[package]] +name = "rustversion" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e819f2bc632f285be6d7cd36e25940d45b2391dd6d9b939e79de557f7014248" + [[package]] name = "ryu" version = "1.0.18" @@ -789,9 +1079,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.132" +version = "1.0.133" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d726bfaff4b320266d395898905d0eba0345aae23b54aee3a737e260fd46db03" +checksum = "c7fceb2473b9166b2294ef05efcb65a3db80803f0b03ef86a5fc88a2b85ee377" dependencies = [ "itoa", "memchr", @@ -799,6 +1089,16 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_path_to_error" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af99884400da37c88f5e9146b7f1fd0fbcae8f6eec4e9da38b67d05486f814a6" +dependencies = [ + "itoa", + "serde", +] + [[package]] name = "serde_spanned" version = "0.6.8" @@ -808,6 +1108,18 @@ dependencies = [ "serde", ] +[[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" @@ -831,14 +1143,20 @@ checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" [[package]] name = "socket2" -version = "0.5.7" +version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce305eb0b4296696835b71df73eb912e0f1ffd2556a501fcede6e0c50349191c" +checksum = "c970269d99b64e60ec3bd6ad27270092a5394c4e309314b18ae3fe575695fbe8" dependencies = [ "libc", "windows-sys 0.52.0", ] +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" + [[package]] name = "stable_deref_trait" version = "1.2.0" @@ -853,15 +1171,27 @@ checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" [[package]] name = "syn" -version = "2.0.87" +version = "2.0.90" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "25aa4ce346d03a6dcd68dd8b4010bcb74e54e62c90c573f394c46eae99aba32d" +checksum = "919d3b74a5dd0ccd15aeb8f93e7006bd9e14c295087c9896a110f490752bcf31" dependencies = [ "proc-macro2", "quote", "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160" + +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "synstructure" version = "0.13.1" @@ -885,9 +1215,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.41.1" +version = "1.42.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22cfb5bee7a6a52939ca9224d6ac897bb669134078daa8735560897f69de4d33" +checksum = "5cec9b21b0450273377fc97bd4c33a8acffc8c996c987a7c5b319a0083707551" dependencies = [ "backtrace", "bytes", @@ -946,17 +1276,75 @@ dependencies = [ "winnow", ] +[[package]] +name = "tower" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2873938d487c3cfb9aed7546dc9f2711d867c9f90c46b889989a2cb84eba6b4f" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper 0.1.2", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + +[[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.41" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0" +dependencies = [ + "log", + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c" +dependencies = [ + "once_cell", +] + +[[package]] +name = "typed-json" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6024a8d0025400b3f6b189366e9aa92012cf9c4fe1cd2620848dd61425c49eed" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "unicode-ident" -version = "1.0.13" +version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e91b56cd4cadaeb79bbf1a5645f6b4f8dc5bde8834ad5894a8db35fda9efa1fe" +checksum = "adb9e6ca4f869e1180728b7950e35922a7fc6397f7b641499e8f3ef06e50dc83" [[package]] name = "url" -version = "2.5.3" +version = "2.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8d157f1b96d14500ffdc1f10ba712e780825526c03d9a49b4d0324b0d9113ada" +checksum = "32f8b686cadd1473f4bd0117a5d28d36b1ade384ea9b5069a1c40aefed7fda60" dependencies = [ "form_urlencoded", "idna", @@ -987,6 +1375,12 @@ version = "0.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "wasi" version = "0.11.0+wasi-snapshot-preview1" @@ -995,9 +1389,9 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" [[package]] name = "wasm-bindgen" -version = "0.2.95" +version = "0.2.99" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "128d1e363af62632b8eb57219c8fd7877144af57558fb2ef0368d0087bddeb2e" +checksum = "a474f6281d1d70c17ae7aa6a613c87fce69a127e2624002df63dcb39d6cf6396" dependencies = [ "cfg-if", "once_cell", @@ -1006,13 +1400,12 @@ dependencies = [ [[package]] name = "wasm-bindgen-backend" -version = "0.2.95" +version = "0.2.99" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb6dd4d3ca0ddffd1dd1c9c04f94b868c37ff5fac97c30b97cff2d74fce3a358" +checksum = "5f89bb38646b4f81674e8f5c3fb81b562be1fd936d84320f3264486418519c79" dependencies = [ "bumpalo", "log", - "once_cell", "proc-macro2", "quote", "syn", @@ -1021,9 +1414,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.95" +version = "0.2.99" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e79384be7f8f5a9dd5d7167216f022090cf1f9ec128e6e6a482a2cb5c5422c56" +checksum = "2cc6181fd9a7492eef6fef1f33961e3695e4579b9872a6f7c83aee556666d4fe" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -1031,9 +1424,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.95" +version = "0.2.99" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26c6ab57572f7a24a4985830b120de1594465e5d500f24afe89e16b4e833ef68" +checksum = "30d7a95b763d3c45903ed6c81f156801839e5ee968bb07e534c44df0fcd330c2" dependencies = [ "proc-macro2", "quote", @@ -1044,9 +1437,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.95" +version = "0.2.99" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65fc09f10666a9f147042251e0dda9c18f166ff7de300607007e96bdebc1068d" +checksum = "943aab3fdaaa029a6e0271b35ea10b72b943135afe9bffca82384098ad0e06a6" [[package]] name = "windows" @@ -1162,12 +1555,16 @@ dependencies = [ name = "wipac_datamove" version = "0.0.2" dependencies = [ + "axum", + "axum-extra", "chrono", "diesel", "env_logger", + "filetime", "hostname", "log", "mysqlclient-sys", + "regex", "serde", "serde_json", "tokio", @@ -1188,9 +1585,9 @@ checksum = "1e9df38ee2d2c3c5948ea468a8406ff0db0b29ae1ffde1bcf20ef305bcc95c51" [[package]] name = "yoke" -version = "0.7.4" +version = "0.7.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6c5b1314b079b0930c31e3af543d8ee1757b1951ae1e1565ec704403a7240ca5" +checksum = "120e6aef9aa629e3d4f52dc8cc43a015c7724194c97dfaf45180d2daf2b77f40" dependencies = [ "serde", "stable_deref_trait", @@ -1200,9 +1597,9 @@ dependencies = [ [[package]] name = "yoke-derive" -version = "0.7.4" +version = "0.7.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28cc31741b18cb6f1d5ff12f5b7523e3d6eb0852bbbad19d73905511d9849b95" +checksum = "2380878cad4ac9aac1e2435f3eb4020e8374b5f13c296cb75b4620ff8e229154" dependencies = [ "proc-macro2", "quote", @@ -1212,18 +1609,18 @@ dependencies = [ [[package]] name = "zerofrom" -version = "0.1.4" +version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91ec111ce797d0e0784a1116d0ddcdbea84322cd79e5d5ad173daeba4f93ab55" +checksum = "cff3ee08c995dee1859d998dea82f7374f2826091dd9cd47def953cae446cd2e" dependencies = [ "zerofrom-derive", ] [[package]] name = "zerofrom-derive" -version = "0.1.4" +version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ea7b4a3637ea8669cedf0f1fd5c286a17f3de97b8dd5a70a6c167a1730e63a5" +checksum = "595eed982f7d355beb85837f651fa22e90b3c044842dc7f2c2842c086f295808" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index d522e3d..82a85ef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,13 +8,19 @@ edition = "2021" publish = false [dependencies] -chrono = "0.4.38" -diesel = { version = "2.2.4", features = ["chrono", "mysql"] } +axum = { version = "0.7.9", features = ["macros"] } +axum-extra = { version = "0.9.6", features = [ "erased-json" ] } +chrono = "0.4.39" +diesel = { version = "2.2.6", features = ["chrono", "mysql"] } env_logger = "0.11.5" hostname = "0.4.0" log = "0.4.22" -mysqlclient-sys = { version = "0.4.1" } # , features = ["bundled"] } +mysqlclient-sys = { version = "0.4.2" } # , features = ["bundled"] } +regex = "1.11.1" serde = { version = "1.0.215", features = ["derive"] } -serde_json = "1.0.132" -tokio = { version = "1.41.1", features = ["full"] } +serde_json = "1.0.133" +tokio = { version = "1.42.0", features = ["full"] } toml = "0.8.19" + +[dev-dependencies] +filetime = "0.2.25" diff --git a/bin/setup b/bin/setup index fce7377..0e4d806 100755 --- a/bin/setup +++ b/bin/setup @@ -7,16 +7,17 @@ if ! dpkg -l | grep -q libmysqlclient-dev; then sudo apt install libmysqlclient-dev --yes fi -if ! dpkg -l | grep -q libncurses5-dev; then - echo "libncurses5-dev is not installed. Installing..." - sudo apt update - sudo apt install libncurses5-dev --yes -fi +# if ! dpkg -l | grep -q libncurses5-dev; then +# echo "libncurses5-dev is not installed. Installing..." +# sudo apt update +# sudo apt install libncurses5-dev --yes +# fi -if ! dpkg -l | grep -q libstdc++-12-dev; then - echo "libstdc++-12-dev is not installed. Installing..." - sudo apt update - sudo apt install libstdc++-12-dev -fi +# if ! dpkg -l | grep -q libstdc++-12-dev; then +# echo "libstdc++-12-dev is not installed. Installing..." +# sudo apt update +# sudo apt install libstdc++-12-dev +# fi cargo install diesel_cli --no-default-features --features mysql +cargo install cargo-outdated diff --git a/datamove.example.toml b/datamove.example.toml index 17f71a8..3081b08 100644 --- a/datamove.example.toml +++ b/datamove.example.toml @@ -10,7 +10,33 @@ password = "hunter2" database_name = "jade" [sps_disk_archiver] -archive_path = "/mnt/slot1" -max_threads = 5 -retry_attempts = 3 +# The number of bytes to leave free before considering the disk to be +# "full". Note that this value is subtracted from the free space on an +# archival disk, so it should be set high enough to allow useful +# activity, but low enough not to waste archival disk space. +# Unit: bytes +archive_headroom = 10000000000 + +# The cache path where the DiskArchiver will store data files until it +# can confirm they are written to a sufficient number of closed copies. +cache_dir = "/mnt/data/jade/disk_cache" + +# Configuration of the disk archives +disk_archives_json_path = "/mnt/data/jade/etc/diskArchives.json" + +# The inbox path where the DiskArchiver will consume archive files +# that should be sent to archival disks. +inbox_dir = "/mnt/data/jade/disk_queue" + +# The problem files directory where the disk archiver should +# place files that it is having a difficult time processing. +problem_files_dir = "/mnt/data/jade/problem_files/disk-archiver" + +# The port used by the component to listen and respond to requests +# for status information. +status_port = 13002 + +# The delay after a work cycle completes, before a disk archiver +# begins to perform work again. +# Unit: seconds work_cycle_sleep_seconds = 5 diff --git a/src/bin/disk_archiver.rs b/src/bin/disk_archiver.rs index a32fe5c..ddb220f 100644 --- a/src/bin/disk_archiver.rs +++ b/src/bin/disk_archiver.rs @@ -3,11 +3,13 @@ // #![feature(trivial_bounds)] #![forbid(unsafe_code)] -use log::{error, info, warn}; -use std::sync::atomic::Ordering; -use tokio::time::{sleep, Duration}; +use axum::{routing::get, routing::post, Router}; +use log::{error, info}; +use std::sync::Arc; +use tokio::{net::TcpListener, sync::Notify, try_join}; use wipac_datamove::sps::{context::load_context, process::disk_archiver::DiskArchiver}; +use wipac_datamove::status::net::{get_disk_archiver_status, post_shutdown_disk_archiver}; #[tokio::main] async fn main() { @@ -17,19 +19,53 @@ async fn main() { // create a new DiskArchiver let context = load_context(); - let mut disk_archiver = DiskArchiver::new(context); - - // tell the DiskArchiver to shut down in ~30 seconds - let shutdown = disk_archiver.shutdown.clone(); - tokio::spawn(async move { - sleep(Duration::from_secs(30)).await; - warn!("Shutting down the DiskArchiver."); - shutdown.store(true, Ordering::Relaxed); + let status_port = context.config.sps_disk_archiver.status_port; + let disk_archiver = Arc::new(DiskArchiver::new(context)); + let shutdown_notify = Arc::new(Notify::new()); + + // start an Axum server to provide JSON status responses + let da = Arc::clone(&disk_archiver); + let sn_await = Arc::clone(&shutdown_notify); + let sn_notify = Arc::clone(&shutdown_notify); + let handle_axum_server = tokio::spawn(async move { + // establish our listening port + let listener = TcpListener::bind(format!("0.0.0.0:{}", status_port)) + .await + .unwrap_or_else(|_| panic!("Unable to listen on port {}", status_port)); + // build our status serving route(s) + let app: Router = Router::new() + .route( + "/shutdown", + post(move |state| post_shutdown_disk_archiver(state, sn_notify)), + ) + .route("/status", get(get_disk_archiver_status)) + .with_state(da); + // start the status service + info!( + "DiskArchiver status service: http://{}/status", + listener.local_addr().unwrap() + ); + axum::serve(listener, app) + .with_graceful_shutdown(async move { + sn_await.notified().await; + }) + .await + .unwrap(); + }); + + // start the DiskArchiver daemon process + let da = Arc::clone(&disk_archiver); + let handle_jade_process = tokio::spawn(async move { + da.run().await.unwrap(); }); - // log about the final result - match disk_archiver.run().await { - Ok(_) => info!("DiskArchiver ends successfully."), - Err(e) => error!("DiskArchiver ends with an error: {e}"), + // wait for both tasks to complete, if either one has an error + if let Err(e) = try_join!(handle_axum_server, handle_jade_process) { + // log and print the error + error!("An error occurred: {}", e); + eprintln!("An error occurred: {:?}", e); } + + // log about the fact that the DiskArchiver has finally shut down + info!("DiskArchiver has shut down."); } diff --git a/src/config.rs b/src/config.rs index 445a78f..e8c1eb4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -19,8 +19,41 @@ pub struct JadeDatabaseConfig { #[derive(Clone, Debug, Deserialize)] pub struct SpsDiskArchiverConfig { - pub archive_path: String, - pub max_threads: u8, - pub retry_attempts: u32, + pub archive_headroom: u64, + pub cache_dir: String, + pub disk_archives_json_path: String, + pub inbox_dir: String, + pub problem_files_dir: String, + pub status_port: u16, pub work_cycle_sleep_seconds: u64, } + +#[derive(Clone, Debug, Deserialize)] +pub struct DiskArchives { + #[serde(rename = "diskArchives")] + pub disk_archives: Vec, +} + +#[derive(Clone, Debug, Deserialize)] +pub struct DiskArchive { + #[serde(rename = "id")] + pub id: u64, + + #[serde(rename = "uuid", skip_serializing_if = "Option::is_none")] + pub uuid: Option, + + #[serde(rename = "description", skip_serializing_if = "Option::is_none")] + pub description: Option, + + #[serde(rename = "name", skip_serializing_if = "Option::is_none")] + pub name: Option, + + #[serde(rename = "numCopies")] + pub num_copies: u64, + + #[serde(rename = "paths")] + pub paths: Vec, + + #[serde(rename = "shortName", skip_serializing_if = "Option::is_none")] + pub short_name: Option, +} diff --git a/src/sps.rs b/src/sps.rs index bfefb91..c6f05a9 100644 --- a/src/sps.rs +++ b/src/sps.rs @@ -5,3 +5,4 @@ pub mod database; pub mod models; pub mod process; pub mod schema; +pub mod utils; diff --git a/src/sps/context.rs b/src/sps/context.rs index 7e321ce..2cb6e52 100644 --- a/src/sps/context.rs +++ b/src/sps/context.rs @@ -3,12 +3,16 @@ use diesel::prelude::*; use log::info; use std::fs; +use std::sync::{Arc, Mutex}; use crate::config::DatamoveConfiguration; +type SharedDatabase = Arc>; + +#[derive(Clone)] pub struct Context { pub config: DatamoveConfiguration, - pub db: MysqlConnection, + pub db: SharedDatabase, pub hostname: String, } @@ -33,6 +37,7 @@ pub fn load_context() -> Context { // db: set up the database connection let database_url = format!("mysql://{username}:{password}@{host}:{port}/{database_name}"); let connection = MysqlConnection::establish(&database_url).unwrap(); + let db = Arc::new(Mutex::new(connection)); // hostname: determine the hostname running the process let full_hostname = hostname::get().expect("Failed to get hostname"); @@ -46,7 +51,7 @@ pub fn load_context() -> Context { // return the application Context object to the caller Context { config: datamove_config.clone(), - db: connection, + db, hostname: hostname.to_string(), } } diff --git a/src/sps/database.rs b/src/sps/database.rs index b18e35a..ab92e2a 100644 --- a/src/sps/database.rs +++ b/src/sps/database.rs @@ -6,7 +6,7 @@ use diesel::prelude::*; use log::{error, info, trace, warn}; use crate::sps::context::Context; -use crate::sps::models::{JadeHost, NewJadeHost}; +use crate::sps::models::{JadeDisk, JadeHost, NewJadeHost}; use crate::sps::models::{MYSQL_FALSE, MYSQL_TRUE}; pub type Error = Box; @@ -16,22 +16,25 @@ pub type Result = core::result::Result; // -- Database API ---------------------------------------------------------- // -------------------------------------------------------------------------- -pub fn ensure_host(context: &mut Context) -> Result { +pub fn ensure_host(context: &Context) -> Result { trace!("ensure_host({})", context.hostname); - let conn = &mut context.db; + let mut conn = match context.db.lock() { + Ok(conn) => conn, + Err(x) => return Err(format!("Unable to lock MysqlConnection: {x}").into()), + }; let hostname = &context.hostname; // if we already have this host in the database - if let Some(host) = select_host_by_hostname(conn, hostname)? { + if let Some(host) = select_host_by_hostname(&mut conn, hostname)? { // return it to the caller return Ok(host); } // otherwise, we need to create the host in the database - insert_host(conn, hostname)?; + insert_host(&mut conn, hostname)?; // now let's return the host that we just created - if let Some(host) = select_host_by_hostname(conn, hostname)? { + if let Some(host) = select_host_by_hostname(&mut conn, hostname)? { return Ok(host); } @@ -40,6 +43,40 @@ pub fn ensure_host(context: &mut Context) -> Result { Err("Unable to ensure_host".into()) } +// -------------------------------------------------------------------------- +// -- jade_disk ------------------------------------------------------------- +// -------------------------------------------------------------------------- + +pub fn select_disk_by_uuid( + conn: &mut MysqlConnection, + find_uuid: &str, +) -> Result> { + trace!("select_disk_by_uuid({find_uuid})"); + use crate::sps::schema::jade_disk::dsl::*; + + // query the DB to find the row for the provided hostname + // SQL: select * from jade_disk where uuid = $uuid order jade_host_id limit 1; + let disks = jade_disk + .select(JadeDisk::as_select()) + .filter(uuid.eq(find_uuid)) + .order(jade_disk_id) + .limit(1) + .load(conn)?; + + // if we got more than one row back, log a warning about that + if disks.len() > 1 { + warn!("Multiple rows returned from jade_disk for uuid = '{find_uuid}'") + } + + // we found it (hopefully only one), so return it to the caller + if !disks.is_empty() { + return Ok(Some(disks[0].clone())); + } + + // oops, we did not find anything + Ok(None) +} + // -------------------------------------------------------------------------- // -- jade_host ------------------------------------------------------------- // -------------------------------------------------------------------------- diff --git a/src/sps/models.rs b/src/sps/models.rs index 76ce8d3..4d5b345 100644 --- a/src/sps/models.rs +++ b/src/sps/models.rs @@ -37,6 +37,31 @@ impl FromSql for BitBool { } } +// -------------------------------------------------------------------------- +// -- jade_disk ------------------------------------------------------------- +// -------------------------------------------------------------------------- + +#[derive(Clone, Queryable, Selectable)] +#[diesel(table_name = crate::sps::schema::jade_disk)] +#[diesel(check_for_backend(diesel::mysql::Mysql))] +pub struct JadeDisk { + pub jade_disk_id: i64, + pub bad: Option, + pub capacity: Option, + pub closed: Option, + pub copy_id: Option, + pub date_created: Option, + pub date_updated: Option, + pub device_path: Option, + pub label: Option, + pub on_hold: Option, + pub uuid: Option, + pub version: Option, + pub jade_host_id: Option, + pub disk_archive_uuid: Option, + pub hardware_metadata: Option, +} + // -------------------------------------------------------------------------- // -- jade_host ------------------------------------------------------------- // -------------------------------------------------------------------------- diff --git a/src/sps/process/disk_archiver.rs b/src/sps/process/disk_archiver.rs index 19703fd..a940bec 100644 --- a/src/sps/process/disk_archiver.rs +++ b/src/sps/process/disk_archiver.rs @@ -1,39 +1,61 @@ // disk_archiver.rs -use log::{info, warn}; -use std::sync::{ - atomic::{AtomicBool, Ordering}, - Arc, -}; +use log::{error, info, warn}; +use regex::Regex; +use std::collections::{BTreeSet, HashMap}; +use std::fs::{self, File, OpenOptions}; +use std::path::Path; +use std::sync::{Arc, Mutex}; use tokio::time::{sleep, Duration}; -use crate::sps::{context::Context, database::ensure_host, models::JadeHost}; +use crate::config::DiskArchives; +use crate::sps::database::select_disk_by_uuid; +use crate::sps::utils::{get_file_count, get_oldest_file_age_in_secs, is_mount_point}; +use crate::status::sps::{Disk, DiskArchiverStatus, DiskStatus}; +use crate::{ + sps::{context::Context, database::ensure_host, models::JadeHost}, + status::sps::DiskArchiverWorkerStatus, +}; + +pub type SharedFlag = Arc>; pub type Error = Box; pub type Result = core::result::Result; +#[derive(Clone)] pub struct DiskArchiver { pub context: Context, - pub host: Option, - pub shutdown: Arc, + pub disk_archives: DiskArchives, + pub host: JadeHost, + pub shutdown: SharedFlag, } impl DiskArchiver { pub fn new(context: Context) -> Self { + let host = + ensure_host(&context).expect("Unable to determine JadeHost running DiskArchiver"); + + let disk_archives_json_path = &context.config.sps_disk_archiver.disk_archives_json_path; + let disk_archives = load_disk_archives(disk_archives_json_path) + .expect("Unable to load disk_archives from JSON configuration"); + Self { context, - host: None, - shutdown: Arc::new(AtomicBool::new(false)), + disk_archives, + host, + shutdown: Arc::new(Mutex::new(false)), } } - pub async fn run(&mut self) -> Result<()> { - // ensure host - let jade_host = ensure_host(&mut self.context)?; - self.host = Some(jade_host); + pub async fn get_status(&self) -> DiskArchiverStatus { + build_disk_archiver_status(self).await + } + + pub async fn run(&self) -> Result<()> { + let mut graceful_shutdown = false; // until a shutdown is requested - while !self.shutdown.load(Ordering::Relaxed) { + while !graceful_shutdown { // perform the work of the disk archiver do_work_cycle(self).await?; // sleep until the next work cycle @@ -44,6 +66,14 @@ impl DiskArchiver { .work_cycle_sleep_seconds; info!("Will sleep for {} seconds.", work_cycle_sleep_seconds); sleep(Duration::from_secs(work_cycle_sleep_seconds)).await; + // check if we need to shut down before starting the next work cycle + graceful_shutdown = match self.shutdown.lock() { + Ok(flag) => *flag, + Err(x) => { + error!("Unable to lock SharedFlag shutdown: {x}"); + true + } + }; } // we received a shutdown command @@ -51,8 +81,203 @@ impl DiskArchiver { Ok(()) } - pub fn request_shutdown(&mut self) { - self.shutdown.store(true, Ordering::Relaxed); + pub fn request_shutdown(&self) { + // if we can get hold of the shutdown flag + let mut flag = match self.shutdown.lock() { + Ok(flag) => flag, + Err(x) => { + error!("Unable to lock SharedFlag shutdown: {x}"); + return; + } + }; + // raise the flag to indicate that we want to shut down + *flag = true; + } +} + +fn build_archival_disk_status(disk_archiver: &DiskArchiver, disk_path: &str) -> Disk { + // ostensibly, this is a path to an archival disk. now let's put it + // through the gauntlet and see what we've really got here... + let path = Path::new(disk_path); + // if the path doesn't exist at all, we can't use it + if !path.exists() { + error!("Archival disk path '{}' does not exist.", path.display()); + return Disk::for_status(DiskStatus::NotUsable); + } + // if we can't write to the path, we can't use it + let temp_path = path.join(".temp_check_writable"); + match OpenOptions::new() + .write(true) + .create_new(true) + .open(&temp_path) + { + Ok(_) => { + let _ = fs::remove_file(&temp_path); + } + Err(_) => { + error!("Archival disk path '{}' is not writable.", path.display()); + return Disk::for_status(DiskStatus::NotUsable); + } + } + // if the path isn't a mount point, it's not mounted + if !is_mount_point(disk_path) { + return Disk::for_status(DiskStatus::NotMounted); + } + // let's see if the mounted disk has any labels + let disk_labels = match read_disk_labels(path) { + Ok(labels) => labels, + Err(_) => { + error!( + "Unable to read disk labels from archival disk path '{}'", + path.display() + ); + return Disk::for_status(DiskStatus::NotUsable); + } + }; + // if there are too many labels, this disk is not usable + if disk_labels.len() > 1 { + error!( + "Archival disk path '{}' contains multiple UUID disk label files.", + path.display() + ); + return Disk::for_status(DiskStatus::NotUsable); + } + // if there are no labels, this disk is available + if disk_labels.is_empty() { + return Disk::for_status(DiskStatus::Available); + } + // there is exactly one label, so look up database information about that disk + let mut conn = match disk_archiver.context.db.lock() { + Ok(conn) => conn, + Err(x) => { + error!("Unable to lock MysqlConnection: {x}"); + return Disk::for_status(DiskStatus::NotUsable); + } + }; + let find_uuid = &disk_labels[0]; + let disk = match select_disk_by_uuid(&mut conn, find_uuid) { + Ok(disk) => { + if let Some(disk) = disk { + disk + } else { + error!( + "Database table jade_disk has no entry for uuid '{}'.", + find_uuid + ); + return Disk::for_status(DiskStatus::NotUsable); + } + } + Err(_) => { + error!( + "Unable to read database table jade_disk for uuid '{}'.", + find_uuid + ); + return Disk::for_status(DiskStatus::NotUsable); + } + }; + // return the fully realized disk status to the caller + match Disk::try_from(disk) { + Ok(disk_status) => disk_status, + Err(e) => { + error!( + "Error when reading database table jade_disk for uuid '{}'.", + find_uuid + ); + error!("Error was: {}", e); + Disk::for_status(DiskStatus::NotUsable) + } + } +} + +fn build_archival_disks_status(disk_archiver: &DiskArchiver) -> HashMap { + // create a hashmap to hold our archival disks + let mut archival_disks = HashMap::new(); + + // gather up all the paths we're configured to use + let mut disk_path_set: BTreeSet = BTreeSet::new(); + for disk_archive in &disk_archiver.disk_archives.disk_archives { + for path in &disk_archive.paths { + disk_path_set.insert(path.to_string()); + } + } + let disk_paths: Vec = disk_path_set.into_iter().collect(); + + // for each disk path + for disk_path in disk_paths { + let disk = build_archival_disk_status(disk_archiver, &disk_path); + archival_disks.insert(disk_path, disk); + } + + // return the hashmap to the caller + archival_disks +} + +/// TODO: documentation comment +pub async fn build_disk_archiver_status(disk_archiver: &DiskArchiver) -> DiskArchiverStatus { + let cache_dir = &disk_archiver.context.config.sps_disk_archiver.cache_dir; + let inbox_dir = &disk_archiver.context.config.sps_disk_archiver.inbox_dir; + let problem_files_dir = &disk_archiver + .context + .config + .sps_disk_archiver + .problem_files_dir; + + let cache_age = match get_oldest_file_age_in_secs(cache_dir) { + Ok(age) => age, + Err(e) => { + error!( + "Unable to determine age of oldest file in the cache directory: {}", + e + ); + 0 + } + }; + + let inbox_count = match get_file_count(inbox_dir) { + Ok(age) => age, + Err(e) => { + error!( + "Unable to determine count of files in the inbox directory: {}", + e + ); + 0 + } + }; + + let inbox_age = match get_oldest_file_age_in_secs(inbox_dir) { + Ok(age) => age, + Err(e) => { + error!( + "Unable to determine age of oldest file in the inbox directory: {}", + e + ); + 0 + } + }; + + let problem_file_count = match get_file_count(problem_files_dir) { + Ok(age) => age, + Err(e) => { + error!( + "Unable to determine count of files in the problem_files directory: {}", + e + ); + 0 + } + }; + + let archival_disks = build_archival_disks_status(disk_archiver); + + DiskArchiverStatus { + workers: vec![DiskArchiverWorkerStatus { + archival_disks, + inbox_count, + }], + cache_age, + inbox_age, + problem_file_count, + message: None, + status: Some("OK".to_string()), } } @@ -67,6 +292,41 @@ pub async fn do_work_cycle(_disk_archiver: &DiskArchiver) -> Result<()> { Ok(()) } +fn load_disk_archives(path: &str) -> Result { + // open the disk archives JSON configuration file + let file = File::open(path).map_err(|e| format!("Failed to open file {}: {}", path, e))?; + // deserialize the JSON into the DiskArchives structure + let disk_archives: DiskArchives = + serde_json::from_reader(&file).map_err(|e| format!("Failed to deserialize JSON: {}", e))?; + // return the DiskArchives structure to the caller + Ok(disk_archives) +} + +fn read_disk_labels(path: &Path) -> Result> { + // disk labels follow a UUID pattern + let uuid_pattern = Regex::new( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$", + ) + .expect("Invalid regex"); + // read the files from the provided path + let disk_labels = fs::read_dir(path)? + // filter out any directory entries that we couldn't read due to error + .filter_map(|entry| entry.ok()) + // keep the ones that match the UUID filename pattern + .filter(|entry| { + let file_name = entry.file_name(); + let name = file_name.to_string_lossy(); + uuid_pattern.is_match(&name) + }) + // map the file names of the ones we keep to a String + .map(|entry| entry.file_name().to_string_lossy().to_string()) + // collect them all up into a Vec<> for the caller + .collect(); + + // return the list of disk label filenames to the caller + Ok(disk_labels) +} + //--------------------------------------------------------------------------- //--------------------------------------------------------------------------- //--------------------------------------------------------------------------- diff --git a/src/sps/utils.rs b/src/sps/utils.rs new file mode 100644 index 0000000..7279579 --- /dev/null +++ b/src/sps/utils.rs @@ -0,0 +1,201 @@ +// utils.rs + +use std::cmp::max; +use std::fs; +use std::path::Path; +use std::process::Command; +use std::time::SystemTime; + +pub type Error = Box; +pub type Result = core::result::Result; + +/// Determine the count of files in the provided directory. +/// +/// This function scans the provided directory and determines how many files are +/// present in the directory. It ignores any subdirectories and only considers +/// regular files in the top level of the specified directory. +/// +/// If no files are found, it returns 0 as the count. +/// +/// # Arguments +/// +/// * `directory` - A string slice representing the path to the directory to scan. +/// +/// # Returns +/// +/// Returns a `Result` containing: +/// - `Ok(u64)` - The count of the number of files if successful. +/// - `Err(Box)` - An error if the directory does not exist, +/// is not accessible, or an error occurs while reading file metadata. +pub fn get_file_count(directory: &str) -> Result { + // convert the provided directory path to a Path + let dir_path = Path::new(directory); + // ensure the provided path is a directory + if !dir_path.is_dir() { + return Err("Provided path is not a directory".into()); + } + // initialize a counter for files + let mut file_count = 0; + // iterate over the entries in the directory + for entry in fs::read_dir(dir_path)? { + let entry = entry?; + let metadata = entry.metadata()?; + // increment the counter if this is a regular file + if metadata.is_file() { + file_count += 1; + } + } + // return the total count of files to the caller + Ok(file_count) +} + +/// Determine the age of the oldest file in the provided directory, in seconds. +/// +/// This function scans the provided directory and determines the file with the oldest +/// modification time. It ignores any subdirectories and only considers regular files +/// in the top level of the specified directory. +/// +/// If no files are found, it returns 0 as the age. +/// +/// # Arguments +/// +/// * `directory` - A string slice representing the path to the directory to scan. +/// +/// # Returns +/// +/// Returns a `Result` containing: +/// - `Ok(u64)` - The age of the oldest file in seconds if successful. +/// - `Err(Box)` - An error if the directory does not exist, +/// is not accessible, or an error occurs while reading file metadata. +pub fn get_oldest_file_age_in_secs(directory: &str) -> Result { + // convert the provided directory path to a Path + let dir_path = Path::new(directory); + // ensure the provided path is a directory + if !dir_path.is_dir() { + return Err("Provided path is not a directory".into()); + } + // get the current system time + let now = SystemTime::now(); + // iterate over the entries in the directory + let mut oldest_age = 0; + // for each entry in the provided directory + for entry in fs::read_dir(dir_path)? { + // get the data about the entry + let entry = entry?; + let metadata = entry.metadata()?; + // if this is a file and not a directory + if metadata.is_file() { + // if we can get the modification time of the file + if let Ok(mtime) = metadata.modified() { + // if we can calculate the duration between the modification time and now + if let Ok(mtime_duration) = now.duration_since(mtime) { + // calculate the age of the file in seconds + let age = mtime_duration.as_secs(); + // update if this file is the oldest one that we've seen so far + oldest_age = max(oldest_age, age); + } + } + } + } + // return the age of the oldest file to the caller + Ok(oldest_age) +} + +/// Determine if the provided path is a mount point or not. +/// +/// This function runs the system command `mountpoint`, which reads the +/// file `/proc/self/mountinfo` to determine if the mount is listed there. +/// If it finds the path listed there, it will end with status code 0. +/// A non-zero status code indicates an error (1) or not a mountpoint (32). +/// +/// If no files are found, it returns 0 as the age. +/// +/// # Arguments +/// +/// * `path` - A string slice representing the path to the directory to check. +/// +/// # Returns +/// +/// Returns `true` if the provided path is a mount point, otherwise `false`. +pub fn is_mount_point(path: &str) -> bool { + // if the path doesn't exist, it can't be a mount point + if !Path::new(path).exists() { + return false; + } + // run the mountpoint command to see if it's a mountpoint or not + let output = Command::new("mountpoint").arg(path).output(); + // return the result to the caller + match output { + // if the exit code was 0, return true, otherwise false + Ok(output) => output.status.success(), + // if there was some kind of error, just call it false + Err(_) => false, + } +} + +//--------------------------------------------------------------------------- +//--------------------------------------------------------------------------- +//--------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use filetime::FileTime; + use std::fs::{self, File}; + use std::path::Path; + use std::time::{Duration, SystemTime}; + + #[test] + fn test_always_succeed() { + assert!(true); + } + + #[test] + fn test_get_file_count() -> Result<()> { + // create a temporary directory for testing + let temp_dir = "/tmp/test_get_file_count"; + std::fs::create_dir_all(temp_dir)?; + // create some files in the directory + std::fs::File::create(format!("{}/file1.txt", temp_dir))?; + std::fs::File::create(format!("{}/file2.txt", temp_dir))?; + std::fs::File::create(format!("{}/file3.txt", temp_dir))?; + // create a subdirectory (should not be counted) + std::fs::create_dir(format!("{}/subdir", temp_dir))?; + // test our file counting function + let count = get_file_count(temp_dir)?; + // clean up + std::fs::remove_dir_all(temp_dir)?; + // check that we got the correct result from our test + assert_eq!(count, 3); + Ok(()) + } + + #[test] + fn test_get_oldest_file_age_in_secs() -> Result<()> { + // create a temporary directory + let temp_dir = "/tmp/test_get_oldest_file_age"; + fs::create_dir_all(temp_dir)?; + // create temporary files with specific creation times + let now = SystemTime::now(); + // helper function to create files with specific ages + let create_file_with_age = |file_name: &str, age_in_secs: u64| -> Result<()> { + let file_path = Path::new(temp_dir).join(file_name); + File::create(&file_path)?; + let creation_time = now - Duration::from_secs(age_in_secs); + let ft = FileTime::from_system_time(creation_time); + filetime::set_file_times(&file_path, ft, ft)?; + Ok(()) + }; + // create files with varying ages + create_file_with_age("file1.txt", 30)?; + create_file_with_age("file2.txt", 60)?; + create_file_with_age("file3.txt", 90)?; + // test our oldest file age function + let oldest_age = get_oldest_file_age_in_secs(temp_dir)?; + // clean up + fs::remove_dir_all(temp_dir)?; + // check that we got the correct result from our test + assert_eq!(oldest_age, 90); + Ok(()) + } +} diff --git a/src/status.rs b/src/status.rs index af31edb..5b21c25 100644 --- a/src/status.rs +++ b/src/status.rs @@ -1,3 +1,4 @@ // status.rs +pub mod net; pub mod sps; diff --git a/src/status/net.rs b/src/status/net.rs new file mode 100644 index 0000000..c23f41f --- /dev/null +++ b/src/status/net.rs @@ -0,0 +1,34 @@ +// net.rs + +use axum::{extract::State, http::StatusCode, response::IntoResponse}; +use axum_extra::response::ErasedJson; +use log::warn; +use serde_json::json; +use std::sync::Arc; +use tokio::sync::Notify; + +use crate::sps::process::disk_archiver::DiskArchiver; + +pub async fn get_disk_archiver_status(State(state): State>) -> impl IntoResponse { + let disk_archiver_status = state.get_status().await; + ErasedJson::pretty(disk_archiver_status) +} + +pub async fn post_shutdown_disk_archiver( + State(state): State>, + shutdown_notify: Arc, +) -> impl IntoResponse { + // signal the DiskArchiver to shutdown + state.request_shutdown(); + // signal the Axum status server to shutdown + shutdown_notify.notify_one(); + // log about the fact that we're shutting down + warn!("Shutdown requested for DiskArchiver."); + // tell the caller that shutdown has been initiated + ( + StatusCode::ACCEPTED, + ErasedJson::pretty(json!({ + "status": "shutdown initiated" + })), + ) +} diff --git a/src/status/sps.rs b/src/status/sps.rs index 4d3403a..e992ef4 100644 --- a/src/status/sps.rs +++ b/src/status/sps.rs @@ -3,6 +3,8 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use crate::sps::models::JadeDisk; + // -------------------------------------------------------------------------- // -- disk_archiver --------------------------------------------------------- // -------------------------------------------------------------------------- @@ -19,7 +21,7 @@ pub struct DiskArchiverStatus { pub inbox_age: u64, #[serde(rename = "problemFileCount")] - pub problem_file_count: u32, + pub problem_file_count: u64, #[serde(rename = "message", skip_serializing_if = "Option::is_none")] pub message: Option, @@ -34,7 +36,7 @@ pub struct DiskArchiverWorkerStatus { pub archival_disks: HashMap, #[serde(rename = "inboxCount")] - pub inbox_count: u32, + pub inbox_count: u64, } #[derive(Clone, Deserialize, Serialize)] @@ -42,13 +44,13 @@ pub struct Disk { pub status: DiskStatus, #[serde(rename = "id")] - pub id: u64, + pub id: i64, #[serde(rename = "closed", skip_serializing_if = "Option::is_none")] pub closed: Option, #[serde(rename = "copyId", skip_serializing_if = "Option::is_none")] - pub copy_id: Option, + pub copy_id: Option, #[serde(rename = "onHold", skip_serializing_if = "Option::is_none")] pub on_hold: Option, @@ -69,6 +71,56 @@ pub struct Disk { pub serial: Option, } +impl Disk { + pub fn for_status(status: DiskStatus) -> Self { + Disk { + status, + id: 0, + closed: None, + copy_id: None, + on_hold: None, + uuid: None, + archive: None, + available: None, + label: None, + serial: None, + } + } +} + +impl TryFrom for Disk { + type Error = Box; + + fn try_from(value: JadeDisk) -> Result { + let closed: bool = value + .closed + .expect("Database returned NULL for jade_disk.closed") + .into(); + let on_hold: bool = value + .on_hold + .expect("Database returned NULL for jade_disk.on_hold") + .into(); + let status = if closed { + DiskStatus::Finished + } else { + DiskStatus::InUse + }; + // return the Disk + Ok(Disk { + status, + id: value.jade_disk_id, + closed: Some(closed), + copy_id: value.copy_id, + on_hold: Some(on_hold), + uuid: value.uuid, + archive: value.disk_archive_uuid, + available: Some(!closed), + label: value.label, + serial: None, // TODO: need to add this!! + }) + } +} + #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] pub enum DiskStatus { #[serde(rename = "Not Mounted")] @@ -109,13 +161,13 @@ pub struct LiveDiskArchiverStatus { pub message: Option, #[serde(rename = "problemFileCount")] - pub problem_file_count: u32, + pub problem_file_count: u64, #[serde(rename = "status", skip_serializing_if = "Option::is_none")] pub status: Option, #[serde(rename = "inboxCount")] - pub inbox_count: u32, + pub inbox_count: u64, #[serde(rename = "archivalDisks")] pub archival_disks: HashMap, diff --git a/tools/init-docker-mysql b/tools/init-docker-mysql new file mode 100755 index 0000000..55ae575 --- /dev/null +++ b/tools/init-docker-mysql @@ -0,0 +1,13 @@ +#!/usr/bin/env bash +# init-docker-mysql + +# create the container and import the JADE database backup +docker run \ + --env=MYSQL_DATABASE="jade" \ + --env=MYSQL_PASSWORD="hunter2" \ + --env=MYSQL_ROOT_PASSWORD="hunter2" \ + --env=MYSQL_USER="jade" \ + --name=sps-jade-20231226 \ + --publish 3306:3306 \ + --volume ${PWD}/sql:/docker-entrypoint-initdb.d:ro \ + mariadb:latest diff --git a/tools/shutdown-disk-archiver b/tools/shutdown-disk-archiver new file mode 100755 index 0000000..92edac6 --- /dev/null +++ b/tools/shutdown-disk-archiver @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# shutdown-disk-archiver + +# submit a command to a local DiskArchiver's RESTful status server +# to tell the DiskArchiver to initiate a graceful shutdown +curl -X POST http://127.0.0.1:13002/shutdown diff --git a/tools/status-disk-archiver b/tools/status-disk-archiver new file mode 100755 index 0000000..a50454c --- /dev/null +++ b/tools/status-disk-archiver @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# status-disk-archiver + +# query a local DiskArchiver's RESTful status server +# to determine the current status of the DiskArchiver +curl -X GET http://127.0.0.1:13002/status