Compare commits

...

3 commits

Author SHA1 Message Date
raito 3265bef2eb feat: statcheck worker
Signed-off-by: Raito Bezarius <masterancpp@gmail.com>
2024-11-14 22:19:22 +01:00
raito 46b6519fc8 feat: add a debug Gerrit event listener
This should be piped into `ofborgctl listen gerrit-events` or something
later on.

Signed-off-by: Raito Bezarius <masterancpp@gmail.com>
2024-11-14 22:19:22 +01:00
raito ba4b005719 feat: add Gerrit SSH API
Make also the event-streamer use it.

Signed-off-by: Raito Bezarius <masterancpp@gmail.com>
2024-11-14 20:38:33 +01:00
10 changed files with 507 additions and 86 deletions

236
Cargo.lock generated
View file

@ -102,9 +102,9 @@ dependencies = [
[[package]]
name = "anstream"
version = "0.6.17"
version = "0.6.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23a1e53f0f5d86382dafe1cf314783b2044280f406e7e1506368220ad11b1338"
checksum = "8acc5369981196006228e28809f761875c0327210a891e941f4c683b3a99529b"
dependencies = [
"anstyle",
"anstyle-parse",
@ -117,9 +117,9 @@ dependencies = [
[[package]]
name = "anstyle"
version = "1.0.9"
version = "1.0.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8365de52b16c035ff4fcafe0092ba9390540e3e352870ac09933bebcaa2c8c56"
checksum = "55cc3b69f167a1ef2e161439aa98aed94e6028e5f9a59be9a6ffb47aef1651f9"
[[package]]
name = "anstyle-parse"
@ -161,7 +161,7 @@ dependencies = [
"nom 7.1.3",
"num-traits",
"rusticata-macros",
"thiserror",
"thiserror 1.0.69",
"time 0.3.36",
]
@ -219,8 +219,8 @@ checksum = "30ca9a001c1e8ba5149f91a74362376cc6bc5b919d92d988668657bd570bdcec"
dependencies = [
"async-task",
"concurrent-queue",
"fastrand 2.1.1",
"futures-lite 2.4.0",
"fastrand 2.2.0",
"futures-lite 2.5.0",
"slab",
]
@ -232,10 +232,10 @@ checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c"
dependencies = [
"async-channel 2.3.1",
"async-executor",
"async-io 2.3.4",
"async-io 2.4.0",
"async-lock 3.4.0",
"blocking",
"futures-lite 2.4.0",
"futures-lite 2.5.0",
"once_cell",
"tokio",
]
@ -273,18 +273,18 @@ dependencies = [
[[package]]
name = "async-io"
version = "2.3.4"
version = "2.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "444b0228950ee6501b3568d3c93bf1176a1fdbc3b758dcd9475046d30f4dc7e8"
checksum = "43a2b323ccce0a1d90b449fd71f2a06ca7faa7c54c2751f06c9bd851fc061059"
dependencies = [
"async-lock 3.4.0",
"cfg-if",
"concurrent-queue",
"futures-io",
"futures-lite 2.4.0",
"futures-lite 2.5.0",
"parking",
"polling 3.7.3",
"rustix 0.38.38",
"polling 3.7.4",
"rustix 0.38.40",
"slab",
"tracing",
"windows-sys 0.59.0",
@ -323,7 +323,7 @@ dependencies = [
"cfg-if",
"event-listener 3.1.0",
"futures-lite 1.13.0",
"rustix 0.38.38",
"rustix 0.38.40",
"windows-sys 0.48.0",
]
@ -345,13 +345,13 @@ version = "0.2.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "637e00349800c0bdf8bfc21ebbc0b6524abea702b0da4168ac00d070d0c0b9f3"
dependencies = [
"async-io 2.3.4",
"async-io 2.4.0",
"async-lock 3.4.0",
"atomic-waker",
"cfg-if",
"futures-core",
"futures-io",
"rustix 0.38.38",
"rustix 0.38.40",
"signal-hook-registry",
"slab",
"windows-sys 0.59.0",
@ -384,6 +384,28 @@ dependencies = [
"wasm-bindgen-futures",
]
[[package]]
name = "async-stream"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
dependencies = [
"async-stream-impl",
"futures-core",
"pin-project-lite",
]
[[package]]
name = "async-stream-impl"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "async-task"
version = "4.7.1"
@ -507,7 +529,7 @@ dependencies = [
"async-channel 2.3.1",
"async-task",
"futures-io",
"futures-lite 2.4.0",
"futures-lite 2.5.0",
"piper",
]
@ -546,9 +568,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.1.31"
version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2e7962b54006dcfcc61cb72735f4d89bb97061dd6a7ed882ec6b8ee53714c6f"
checksum = "fd9de9f2205d5ef3fd67e685b0df337994ddd4495e2a28d185500d0e1edfea47"
dependencies = [
"jobserver",
"libc",
@ -587,9 +609,9 @@ dependencies = [
[[package]]
name = "clap"
version = "4.5.20"
version = "4.5.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b97f376d85a664d5837dbae44bf546e6477a679ff6610010f17276f686d867e8"
checksum = "fb3b4b9e5a7c7514dfa52869339ee98b3156b0bfb4e8a77c4ff4babb64b1604f"
dependencies = [
"clap_builder",
"clap_derive",
@ -597,9 +619,9 @@ dependencies = [
[[package]]
name = "clap_builder"
version = "4.5.20"
version = "4.5.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "19bc80abd44e4bed93ca373a0704ccbd1b710dc5749406201bb018272808dc54"
checksum = "b17a95aa67cc7b5ebd32aa5370189aa0d79069ef1c64ce893bd30fb24bff20ec"
dependencies = [
"anstream",
"anstyle",
@ -621,9 +643,9 @@ dependencies = [
[[package]]
name = "clap_lex"
version = "0.7.2"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"
checksum = "afb84c814227b90d6895e01398aee0d8033c00e7466aca416fb6a8e0eb19d8a7"
[[package]]
name = "cms"
@ -682,9 +704,9 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
[[package]]
name = "cpufeatures"
version = "0.2.14"
version = "0.2.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "608697df725056feaccfa42cffdaeeec3fccc4ffc38358ecd19b243e716a78e0"
checksum = "0ca741a962e1b0bff6d724a1a0958b686406e853bb14061f218562e1896f95e6"
dependencies = [
"libc",
]
@ -905,9 +927,9 @@ dependencies = [
[[package]]
name = "fastrand"
version = "2.1.1"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e8c02a5121d4ea3eb16a80748c74f5549a5665e4c21333c6098f283870fbdea6"
checksum = "486f806e73c5707928240ddc295403b1b93c96a02038563881c4a2fd84b81ac4"
[[package]]
name = "flagset"
@ -960,6 +982,7 @@ checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
dependencies = [
"futures-channel",
"futures-core",
"futures-executor",
"futures-io",
"futures-sink",
"futures-task",
@ -982,6 +1005,17 @@ version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
[[package]]
name = "futures-executor"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
dependencies = [
"futures-core",
"futures-task",
"futures-util",
]
[[package]]
name = "futures-io"
version = "0.3.31"
@ -1005,11 +1039,11 @@ dependencies = [
[[package]]
name = "futures-lite"
version = "2.4.0"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f1fa2f9765705486b33fd2acf1577f8ec449c2ba1f318ae5447697b7c08d210"
checksum = "cef40d21ae2c515b51041df9ed313ed21e572df340ea58a922a0aefe7e8891a1"
dependencies = [
"fastrand 2.1.1",
"fastrand 2.2.0",
"futures-core",
"futures-io",
"parking",
@ -1045,10 +1079,13 @@ version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
dependencies = [
"futures-channel",
"futures-core",
"futures-io",
"futures-macro",
"futures-sink",
"futures-task",
"memchr",
"pin-project-lite",
"pin-utils",
"slab",
@ -1479,9 +1516,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
[[package]]
name = "libc"
version = "0.2.161"
version = "0.2.162"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8e9489c2807c139ffd9c1794f4af0ebe86a828db53ecdc7fea2111d0fed085d1"
checksum = "18d287de67fe55fd7e1581fe933d965a5a9477b38e949cfa9f8574ef01506398"
[[package]]
name = "libredox"
@ -1716,12 +1753,14 @@ name = "ofborg"
version = "0.90.0"
dependencies = [
"async-std",
"async-stream",
"base64 0.22.1",
"brace-expand",
"chrono",
"clap",
"either",
"fs2",
"futures",
"futures-util",
"http",
"hubcaps",
@ -1731,6 +1770,7 @@ dependencies = [
"lru-cache",
"md5",
"nom 4.2.3",
"openssh",
"regex",
"rustls-pemfile 1.0.4",
"separator",
@ -1740,7 +1780,9 @@ dependencies = [
"shellexpand",
"sys-info",
"tempfile",
"thiserror",
"thiserror 1.0.69",
"tokio",
"tokio-stream",
"tracing",
"tracing-subscriber",
"uuid",
@ -1770,6 +1812,20 @@ version = "1.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1261fe7e33c73b354eab43b1273a57c8f967d0391e80353e51f764ac02cf6775"
[[package]]
name = "openssh"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b52987a10526b8daef7f1946b0aadfc214479f897ba624776327fd3beec2722c"
dependencies = [
"libc",
"once_cell",
"shell-escape",
"tempfile",
"thiserror 2.0.3",
"tokio",
]
[[package]]
name = "openssl-probe"
version = "0.1.5"
@ -1806,7 +1862,7 @@ dependencies = [
"rc2",
"sha1",
"sha2",
"thiserror",
"thiserror 1.0.69",
"x509-parser",
]
@ -1912,7 +1968,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96c8c490f422ef9a4efd2cb5b42b76c8613d7e7dfc1caf667b8a3350a5acc066"
dependencies = [
"atomic-waker",
"fastrand 2.1.1",
"fastrand 2.2.0",
"futures-io",
]
@ -1970,15 +2026,15 @@ dependencies = [
[[package]]
name = "polling"
version = "3.7.3"
version = "3.7.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc2790cd301dec6cd3b7a025e4815cf825724a51c98dccfe6a3e55f05ffb6511"
checksum = "a604568c3202727d1507653cb121dbd627a58684eb09a820fd746bee38b4442f"
dependencies = [
"cfg-if",
"concurrent-queue",
"hermit-abi 0.4.0",
"pin-project-lite",
"rustix 0.38.38",
"rustix 0.38.40",
"tracing",
"windows-sys 0.59.0",
]
@ -2083,7 +2139,7 @@ checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43"
dependencies = [
"getrandom",
"libredox",
"thiserror",
"thiserror 1.0.69",
]
[[package]]
@ -2094,7 +2150,7 @@ checksum = "b544ef1b4eac5dc2db33ea63606ae9ffcfac26c1416a2806ae0bf5f56b201191"
dependencies = [
"aho-corasick",
"memchr",
"regex-automata 0.4.8",
"regex-automata 0.4.9",
"regex-syntax 0.8.5",
]
@ -2109,9 +2165,9 @@ dependencies = [
[[package]]
name = "regex-automata"
version = "0.4.8"
version = "0.4.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "368758f23274712b504848e9d5a6f010445cc8b87a7cdb4d7cbee666c1288da3"
checksum = "809e8dc61f6de73b46c85f4c96486310fe304c434cfa43669d7b40f711150908"
dependencies = [
"aho-corasick",
"memchr",
@ -2232,9 +2288,9 @@ dependencies = [
[[package]]
name = "rustix"
version = "0.38.38"
version = "0.38.40"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aa260229e6538e52293eeb577aabd09945a09d6d9cc0fc550ed7529056c2e32a"
checksum = "99e4ea3e1cdc4b559b8e5650f9c8e5998e3e5c1343b4eaf034565f32318d63c0"
dependencies = [
"bitflags 2.6.0",
"errno",
@ -2412,9 +2468,9 @@ dependencies = [
[[package]]
name = "security-framework-sys"
version = "2.12.0"
version = "2.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ea4a292869320c0272d7bc55a5a6aafaff59b4f63404a003887b679a2e05b4b6"
checksum = "fa39c7303dc58b5543c94d22c1766b0d31f2ee58306363ea622b10bbc075eaa2"
dependencies = [
"core-foundation-sys",
"libc",
@ -2428,18 +2484,18 @@ checksum = "f97841a747eef040fcd2e7b3b9a220a7205926e60488e673d9e4926d27772ce5"
[[package]]
name = "serde"
version = "1.0.214"
version = "1.0.215"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f55c3193aca71c12ad7890f1785d2b73e1b9f63a0bbc353c08ef26fe03fc56b5"
checksum = "6513c1ad0b11a9376da888e3e0baa0077f1aed55c17f50e7b2397136129fb88f"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
version = "1.0.214"
version = "1.0.215"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "de523f781f095e28fa605cdce0f8307e451cc0fd14e2eb4cd2e98a355b147766"
checksum = "ad1e866f866923f252f05c889987993144fb74e722403468a4ebd70c3cd756c0"
dependencies = [
"proc-macro2",
"quote",
@ -2501,6 +2557,12 @@ dependencies = [
"lazy_static",
]
[[package]]
name = "shell-escape"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "45bb67a18fa91266cc7807181f62f9178a6873bfad7dc788c42e6430db40184f"
[[package]]
name = "shellexpand"
version = "3.1.0"
@ -2681,31 +2743,51 @@ dependencies = [
[[package]]
name = "tempfile"
version = "3.13.0"
version = "3.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0f2c9fc62d0beef6951ccffd757e241266a2c833136efbe35af6cd2567dca5b"
checksum = "28cce251fcbc87fac86a866eeb0d6c2d536fc16d06f184bb61aeae11aa4cee0c"
dependencies = [
"cfg-if",
"fastrand 2.1.1",
"fastrand 2.2.0",
"once_cell",
"rustix 0.38.38",
"rustix 0.38.40",
"windows-sys 0.59.0",
]
[[package]]
name = "thiserror"
version = "1.0.67"
version = "1.0.69"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3b3c6efbfc763e64eb85c11c25320f0737cb7364c4b6336db90aa9ebe27a0bbd"
checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
dependencies = [
"thiserror-impl",
"thiserror-impl 1.0.69",
]
[[package]]
name = "thiserror"
version = "2.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c006c85c7651b3cf2ada4584faa36773bd07bac24acfb39f3c431b36d7e667aa"
dependencies = [
"thiserror-impl 2.0.3",
]
[[package]]
name = "thiserror-impl"
version = "1.0.67"
version = "1.0.69"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b607164372e89797d78b8e23a6d67d5d1038c1c65efd52e1389ef8b77caba2a6"
checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "thiserror-impl"
version = "2.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f077553d607adc1caf65430528a576c757a71ed73944b66ebb58ef2bbd243568"
dependencies = [
"proc-macro2",
"quote",
@ -2781,19 +2863,32 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "tokio"
version = "1.41.0"
version = "1.41.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "145f3413504347a2be84393cc8a7d2fb4d863b375909ea59f2158261aa258bbb"
checksum = "22cfb5bee7a6a52939ca9224d6ac897bb669134078daa8735560897f69de4d33"
dependencies = [
"backtrace",
"bytes",
"libc",
"mio",
"pin-project-lite",
"signal-hook-registry",
"socket2 0.5.7",
"tokio-macros",
"windows-sys 0.52.0",
]
[[package]]
name = "tokio-macros"
version = "2.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "693d596312e88961bc67d7f1f97af8a70227d9f90c31bba5806eec004978d752"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "tokio-rustls"
version = "0.24.1"
@ -2804,6 +2899,17 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-stream"
version = "0.1.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4f4e6ce100d0eb49a2734f8c0812bcd324cf357d21810932c5df6b96ef2b86f1"
dependencies = [
"futures-core",
"pin-project-lite",
"tokio",
]
[[package]]
name = "tokio-util"
version = "0.7.12"
@ -3356,7 +3462,7 @@ dependencies = [
"nom 7.1.3",
"oid-registry",
"rusticata-macros",
"thiserror",
"thiserror 1.0.69",
"time 0.3.36",
]

View file

@ -5,6 +5,10 @@
"checkout": {
"root": "$STATE_DIRECTORY/.checkouts"
},
"vcs": "Gerrit",
"gerrit": {
"instance_uri": "cl.forkos.org"
},
"nix": {
"system": "x86_64-linux",
"remote": "daemon",

View file

@ -43,4 +43,9 @@ zstd = "0.13.2"
jfs = "0.9.0"
base64 = "0.22.1"
thiserror = "1.0.67"
openssh = { version = "0.11.3", features = ["process-mux"], default-features = false }
futures = "0.3.31"
tokio = "1.41.1"
tokio-stream = { version = "0.1.16", features = ["io-util"] }
async-stream = "0.3.6"
# reqwest = "0.12.9"

View file

@ -4,7 +4,10 @@
use std::env;
use std::error::Error;
use async_std::stream::StreamExt;
use async_std::task;
use futures_util::TryStreamExt;
use ofborg::vcs::gerrit::ssh::GerritSSHApi;
use tracing::info;
use ofborg::config;
@ -35,6 +38,23 @@ fn main() -> Result<(), Box<dyn Error>> {
})?;
info!("Publishing events from Gerrit into {}", &exchange_name);
let gerrit_cfg = cfg
.gerrit
.expect("Gerrit event streamer requires Gerrit configuration");
let gerrit_api = GerritSSHApi::new(
gerrit_cfg.ssh_private_key_file,
&format!("ssh://{}:{}", gerrit_cfg.instance_uri, gerrit_cfg.ssh_port),
);
task::block_on(async {
while let Some(event) = gerrit_api.stream_events().await?.try_next().await? {
//chan.basic_publish();
// publish the event in the exchange!
todo!();
}
});
//task::block_on(handle);
drop(conn); // Close connection.

View file

@ -0,0 +1,42 @@
/// This is a Gerrit listener for events which puts them on stdout for debugging purposes.
/// The list of event type listened to is static.
use std::env;
use std::error::Error;
use async_std::stream::StreamExt;
use async_std::task;
use futures::pin_mut;
use ofborg::vcs::gerrit::ssh::GerritSSHApi;
use tracing::info;
use ofborg::config;
fn main() -> Result<(), Box<dyn Error>> {
ofborg::setup_log();
let arg = env::args()
.nth(1)
.expect("usage: listen-gerrit-events <config>");
let cfg = config::load(arg.as_ref());
let gerrit_cfg = cfg
.gerrit
.expect("Gerrit event streaming requires a Gerrit configuration");
let gerrit_ssh_uri = format!("ssh://{}:{}", gerrit_cfg.instance_uri, gerrit_cfg.ssh_port);
info!("Listening events from Gerrit on {}", gerrit_ssh_uri);
let mut gerrit_api = GerritSSHApi::new(gerrit_cfg.ssh_private_key_file, &gerrit_ssh_uri);
task::block_on(async {
let event_stream = gerrit_api.stream_events().await.unwrap();
pin_mut!(event_stream);
loop {
let thing = event_stream.next().await;
println!("{:?}", thing);
if let Some(Ok(event)) = thing {
println!("{:#?}", event);
}
}
});
Ok(())
}

View file

@ -0,0 +1,79 @@
/// Statuses and checks worker
/// - will keep a database of changes
/// - their statuses
/// - their checks
/// - is VCS/CI agnostic
use std::env;
use std::error::Error;
use async_std::task;
use ofborg::config;
use ofborg::easyamqp;
use ofborg::easyamqp::ChannelExt;
use ofborg::easyamqp::ConsumerExt;
use ofborg::easylapin;
use ofborg::tasks;
use tracing::info;
fn main() -> Result<(), Box<dyn Error>> {
ofborg::setup_log();
let arg = env::args()
.nth(1)
.expect("usage: statcheck-worker <config>");
let cfg = config::load(arg.as_ref());
let conn = easylapin::from_config(&cfg.rabbitmq)?;
let mut chan = task::block_on(conn.create_channel())?;
// an RPC queue for verbs
let api_queue_name = "statcheck-api".to_owned();
// an event queue to be notified about statuses & checks changes.
let event_queue_name = "statcheck-events".to_owned();
chan.declare_exchange(easyamqp::ExchangeConfig {
exchange: api_queue_name.clone(),
exchange_type: easyamqp::ExchangeType::Topic,
passive: false,
durable: true,
auto_delete: false,
no_wait: false,
internal: false,
})?;
chan.declare_queue(easyamqp::QueueConfig {
queue: api_queue_name.clone(),
passive: false,
durable: true,
exclusive: false,
auto_delete: false,
no_wait: false,
})?;
chan.bind_queue(easyamqp::BindQueueConfig {
queue: api_queue_name.clone(),
exchange: api_queue_name.clone(),
routing_key: None,
no_wait: false,
})?;
let handle = easylapin::WorkerChannel(chan).consume(
tasks::statcheck_collector::StatusCheckCollector::new(cfg.statcheck.clone().db),
easyamqp::ConsumeConfig {
queue: api_queue_name.clone(),
consumer_tag: format!("{}-{}", cfg.whoami(), api_queue_name),
no_local: false,
no_ack: false,
no_wait: false,
exclusive: false,
},
)?;
info!("Waiting for API calls on {}", api_queue_name);
info!("Notifying of new changes on {}", event_queue_name);
task::block_on(handle);
drop(conn); // Close connection.
info!("Closed the session... EOF");
Ok(())
}

View file

@ -0,0 +1,46 @@
use std::path::PathBuf;
use tracing::{debug_span, error};
use crate::worker;
#[derive(Serialize, Deserialize, Debug)]
enum StatusCheckRPCMessage {
ListStatuses,
ListChecks,
}
#[allow(dead_code)]
struct StatusCheckCollector {
db_path: PathBuf,
}
// RPC API worker
impl worker::SimpleWorker for StatusCheckCollector {
type J = StatusCheckRPCMessage;
fn msg_to_job(
&mut self,
_method: &str,
_headers: &Option<String>,
body: &[u8],
) -> Result<Self::J, String> {
match serde_json::from_slice(body) {
Ok(e) => Ok(e),
Err(e) => {
error!(
"Failed to deserialize StatusCheckRPCMessage: {:?}",
String::from_utf8(body.to_vec())
);
panic!("{:?}", e);
}
}
}
fn consumer(&mut self, job: &Self::J) -> worker::Actions {
let span = debug_span!("command");
let _enter = span.enter();
vec![worker::Action::Ack]
}
}

View file

@ -28,12 +28,12 @@ impl From<Vec<Account>> for crate::vcs::generic::ChangeReviewers {
pub struct Approval {
pub r#type: String, // Internal name of the approval
pub description: String, // Human-readable category of the approval
pub value: i32, // Value assigned by the approval (usually a numerical score)
pub value: String, // Value assigned by the approval (usually a numerical score)
#[serde(rename = "oldValue")]
pub old_value: Option<i32>, // Previous approval score, if present
pub old_value: Option<String>, // Previous approval score, if present
#[serde(rename = "grantedOn")]
pub granted_on: u64, // Time in seconds since the UNIX epoch
pub by: Account, // Reviewer of the patch set
pub granted_on: Option<u64>, // Time in seconds since the UNIX epoch
pub by: Option<Account>, // Reviewer of the patch set
}
#[derive(Serialize, Deserialize, Debug)]
@ -87,8 +87,11 @@ pub struct PatchSet {
#[serde(rename = "createdOn")]
pub created_on: u64, // Time in seconds since the UNIX epoch
pub kind: Kind, // Kind of change ("REWORK", "TRIVIAL_REBASE", etc.)
#[serde(default = "Default::default")]
pub approvals: Vec<Approval>, // Approvals granted
#[serde(default = "Default::default")]
pub comments: Vec<PatchSetComment>, // All comments for this patchset
#[serde(default = "Default::default")]
pub files: Vec<String>, // All changed files in this patchset
#[serde(rename = "sizeInsertions")]
pub size_insertions: i64, // Size of insertions
@ -179,30 +182,32 @@ pub struct Change {
pub subject: String,
pub owner: Account,
pub url: String,
pub commit_message: String,
pub commit_message: Option<String>,
#[serde(default = "Default::default")]
pub hashtags: Vec<String>,
#[serde(rename = "createdOn")]
pub created_on: u64, // Time in seconds since UNIX epoch
#[serde(rename = "lastUpdated")]
pub last_updated: u64, // Time in seconds since UNIX epoch
pub open: bool,
pub last_updated: Option<u64>, // Time in seconds since UNIX epoch
pub open: Option<bool>,
pub status: ChangeStatus, // "NEW", "MERGED", or "ABANDONED"
pub private: bool,
pub wip: bool, // Work in progress
pub private: Option<bool>,
pub wip: Option<bool>, // Work in progress
#[serde(default = "Default::default")]
pub comments: Vec<ReviewerMessage>, // Inline/file comments
#[serde(rename = "trackingIds")]
#[serde(rename = "trackingIds", default = "Default::default")]
pub tracking_ids: Vec<TrackingId>, // Links to issue tracking systems
#[serde(rename = "currentPatchSet")]
pub current_patch_set: PatchSet,
#[serde(rename = "patchSets")]
pub current_patch_set: Option<PatchSet>,
#[serde(rename = "patchSets", default = "Default::default")]
pub patch_sets: Vec<PatchSet>, // All patch sets
#[serde(rename = "dependsOn")]
#[serde(rename = "dependsOn", default = "Default::default")]
pub depends_on: Vec<ChangeDependency>, // Dependencies
#[serde(rename = "neededBy")]
#[serde(rename = "neededBy", default = "Default::default")]
pub needed_by: Vec<ChangeDependency>, // Reverse dependencies
#[serde(rename = "submitRecords")]
#[serde(rename = "submitRecords", default = "Default::default")]
pub submit_records: Vec<SubmitRecord>, // Submission information
#[serde(rename = "allReviewers")]
#[serde(rename = "allReviewers", default = "Default::default")]
pub all_reviewers: Vec<Account>, // List of all reviewers
}
@ -213,7 +218,8 @@ impl From<Change> for crate::message::Change {
// While the change number is deprecated, we actually need it.
// FIXME: enforce type level checking of this.
number: value.change_number.unwrap(),
head_sha: value.current_patch_set.revision,
// FIXME: that's not good…
head_sha: value.current_patch_set.unwrap().revision,
}
}
}
@ -224,18 +230,21 @@ pub struct RefUpdate {
pub old_rev: String, // The old value of the ref, prior to the update
#[serde(rename = "newRev")]
pub new_rev: String, // The new value the ref was updated to
#[serde(rename = "refName")]
pub ref_name: String, // Full ref name within the project
pub project: String, // Project path in Gerrit
}
#[derive(Serialize, Deserialize, Debug)]
#[serde(rename_all = "camelCase", tag = "type")]
#[serde(rename_all = "kebab-case", tag = "type")]
pub enum GerritStreamEvent {
ChangeAbandoned {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
abandoner: Account,
reason: String,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
ChangeDeleted {
@ -244,24 +253,32 @@ pub enum GerritStreamEvent {
},
ChangeMerged {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
submitter: Account,
#[serde(rename = "newRev")]
new_rev: String,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
ChangeRestored {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
restorer: Account,
reason: String,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
CommentAdded {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
author: Account,
#[serde(default = "Default::default")]
approvals: Vec<Approval>,
comment: String,
comment: Option<String>,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
DroppedOutput,
@ -271,43 +288,57 @@ pub enum GerritStreamEvent {
added: Vec<String>,
removed: Vec<String>,
hashtags: Vec<String>,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
ProjectCreated {
#[serde(rename = "projectName")]
project_name: String,
#[serde(rename = "projectHead")]
project_head: String,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
PatchSetCreated {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
uploader: Account,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
RefUpdated {
submitter: Account,
#[serde(rename = "refUpdate")]
ref_update: RefUpdate,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
BatchRefUpdated {
submitter: Account,
#[serde(rename = "refUpdates")]
ref_updates: Vec<RefUpdate>,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
ReviewerAdded {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
reviewer: Account,
adder: Account,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
ReviewerDeleted {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
reviewer: Account,
remover: Account,
approvals: Vec<Approval>,
comment: String,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
TopicChanged {
@ -315,22 +346,28 @@ pub enum GerritStreamEvent {
old_topic: Option<String>,
new_topic: Option<String>,
changer: Account,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
WorkInProgressStateChanged {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
changer: Account,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
PrivateStateChanged {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
changer: Account,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
VoteDeleted {
change: Change,
#[serde(rename = "patchSet")]
patch_set: PatchSet,
reviewer: Account,
remover: Account,
@ -338,8 +375,11 @@ pub enum GerritStreamEvent {
comment: String,
},
ProjectHeadUpdate {
#[serde(rename = "oldHead")]
old_head: String,
#[serde(rename = "newHead")]
new_head: String,
#[serde(rename = "eventCreatedOn")]
event_created_on: u64,
},
}

View file

@ -2,4 +2,5 @@ pub mod checks;
pub mod data_structures;
pub mod http;
pub mod r#impl;
pub mod ssh;
// pub mod events;

View file

@ -0,0 +1,78 @@
use std::{error::Error, path::PathBuf};
use async_std::stream::StreamExt;
use async_stream::stream;
use futures_util::Stream;
use openssh::{Session, SessionBuilder, Stdio};
use tokio::io::AsyncBufReadExt;
use super::data_structures::GerritStreamEvent;
pub struct GerritSSHApi {
session: Session,
}
impl GerritSSHApi {
pub fn new(private_key_file: PathBuf, uri: &str) -> Self {
let mut builder = SessionBuilder::default();
let (builder, destination) = builder.keyfile(&private_key_file).resolve(uri);
let tempdir = async_std::task::block_on(builder.launch_master(destination)).expect(
&format!("Failed to launch SSH master to destination '{}'", uri),
);
Self {
session: Session::new_process_mux(tempdir),
}
}
pub async fn raw_command(
&mut self,
args: Vec<&str>,
) -> Result<impl Stream<Item = Result<String, std::io::Error>> + '_, Box<dyn Error>> {
self.session
.check()
.await
.expect("Session is not in a good state.");
let mut child = self
.session
.raw_command("gerrit")
.raw_args(args)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.await
.expect("Failed to spawn a command");
Ok(stream! {
let child_stdout = child.stdout().take().expect("Failed to obtain stdout");
let stdout = tokio::io::BufReader::new(child_stdout);
let mut line_stream = stdout.lines();
loop {
match line_stream.next_line().await {
Ok(Some(line)) => yield Ok(line),
Ok(None) => {
break;
}
Err(err) => yield Err(err)
}
}
})
}
pub async fn stream_events(
&mut self,
) -> Result<impl Stream<Item = Result<GerritStreamEvent, Box<dyn Error>>> + '_, Box<dyn Error>>
{
let lines = self.raw_command(vec!["stream-events"]).await?;
let events = lines.filter_map(|line| line.ok()).map(|line| {
let event: Result<GerritStreamEvent, _> = serde_json::from_str(&line);
match event {
Ok(event) => Ok(event),
Err(err) => Err(Box::new(err) as Box<dyn Error>),
}
});
Ok(events)
}
}