From 3ce9f3edd1046b7a15ab558773ed70d0ac5a166e Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Wed, 20 May 2020 23:35:37 +0200 Subject: [PATCH 1/3] generalize amqp references in SysEvents --- ofborg/src/bin/mass-rebuilder.rs | 2 +- ofborg/src/bin/stats.rs | 2 +- ofborg/src/stats.rs | 54 +++++++++++++++++++++++++++----- 3 files changed, 48 insertions(+), 10 deletions(-) diff --git a/ofborg/src/bin/mass-rebuilder.rs b/ofborg/src/bin/mass-rebuilder.rs index 8db5be9..b449468 100644 --- a/ofborg/src/bin/mass-rebuilder.rs +++ b/ofborg/src/bin/mass-rebuilder.rs @@ -40,7 +40,7 @@ fn main() { let cloner = checkout::cached_cloner(Path::new(&cfg.checkout.root)); let nix = cfg.nix(); - let events = stats::RabbitMQ::new( + let events = stats::RabbitMQ::from_amqp( &format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()), session.open_channel(3).unwrap(), ); diff --git a/ofborg/src/bin/stats.rs b/ofborg/src/bin/stats.rs index 06a4ffc..bcf3dd1 100644 --- a/ofborg/src/bin/stats.rs +++ b/ofborg/src/bin/stats.rs @@ -17,7 +17,7 @@ fn main() { let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap(); info!("Connected to rabbitmq"); - let events = stats::RabbitMQ::new( + let events = stats::RabbitMQ::from_amqp( &format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()), session.open_channel(3).unwrap(), ); diff --git a/ofborg/src/stats.rs b/ofborg/src/stats.rs index c9bde24..97bccef 100644 --- a/ofborg/src/stats.rs +++ b/ofborg/src/stats.rs @@ -1,5 +1,8 @@ -use amqp::protocol::basic::BasicProperties; -use amqp::{Basic, Channel}; +use amqp::protocol::basic; +use amqp::Basic; +use async_std::task; +use lapin::options::BasicPublishOptions; +use lapin::CloseOnDrop; include!(concat!(env!("OUT_DIR"), "/events.rs")); @@ -19,13 +22,13 @@ pub struct EventMessage { pub events: Vec, } -pub struct RabbitMQ { +pub struct RabbitMQ { identity: String, - channel: Channel, + channel: C, } -impl RabbitMQ { - pub fn new(identity: &str, channel: Channel) -> RabbitMQ { +impl RabbitMQ { + pub fn from_amqp(identity: &str, channel: amqp::Channel) -> Self { RabbitMQ { identity: identity.to_owned(), channel, @@ -33,9 +36,9 @@ impl RabbitMQ { } } -impl SysEvents for RabbitMQ { +impl SysEvents for RabbitMQ { fn notify(&mut self, event: Event) { - let props = BasicProperties { + let props = basic::BasicProperties { ..Default::default() }; self.channel @@ -55,3 +58,38 @@ impl SysEvents for RabbitMQ { .unwrap(); } } + +impl RabbitMQ> { + pub fn from_lapin(identity: &str, channel: CloseOnDrop) -> Self { + RabbitMQ { + identity: identity.to_owned(), + channel, + } + } +} + +impl SysEvents for RabbitMQ> { + fn notify(&mut self, event: Event) { + let props = lapin::BasicProperties::default().with_content_type("application/json".into()); + task::block_on(async { + let _confirmaton = self + .channel + .basic_publish( + &String::from("stats"), + &"".to_owned(), + BasicPublishOptions::default(), + serde_json::to_string(&EventMessage { + sender: self.identity.clone(), + events: vec![event], + }) + .unwrap() + .into_bytes(), + props, + ) + .await + .unwrap() + .await + .unwrap(); + }); + } +} From 9120d16bd75b23d9c525d3946712c60557e08a7c Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Thu, 21 May 2020 21:09:33 +0200 Subject: [PATCH 2/3] convert mass rebuilder to lapin --- ofborg/src/bin/mass-rebuilder.rs | 112 ++++++++++++++----------------- 1 file changed, 51 insertions(+), 61 deletions(-) diff --git a/ofborg/src/bin/mass-rebuilder.rs b/ofborg/src/bin/mass-rebuilder.rs index b449468..d72f171 100644 --- a/ofborg/src/bin/mass-rebuilder.rs +++ b/ofborg/src/bin/mass-rebuilder.rs @@ -1,20 +1,26 @@ -use ofborg::checkout; -use ofborg::config; -use ofborg::easyamqp::{self, ChannelExt, ConsumerExt}; -use ofborg::stats; -use ofborg::tasks; -use ofborg::worker; - use std::env; +use std::error::Error; use std::path::Path; use std::process; -use amqp::Basic; +use async_std::task; use tracing::{error, info}; +use ofborg::checkout; +use ofborg::config; +use ofborg::easyamqp::{self, ChannelExt, ConsumerExt}; +use ofborg::easylapin; +use ofborg::stats; +use ofborg::tasks; + // FIXME: remove with rust/cargo update #[allow(clippy::cognitive_complexity)] -fn main() { +fn main() -> Result<(), Box> { + ofborg::setup_log(); + + let arg = env::args().nth(1).expect("usage: mass-rebuilder "); + let cfg = config::load(arg.as_ref()); + let memory_info = sys_info::mem_info().expect("Unable to get memory information from OS"); if memory_info.avail < 8 * 1024 * 1024 { @@ -26,68 +32,52 @@ fn main() { process::exit(1); }; - let cfg = config::load(env::args().nth(1).unwrap().as_ref()); - - ofborg::setup_log(); - - info!("Hello, world!"); - - let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap(); - info!("Connected to rabbitmq"); - - let mut channel = session.open_channel(1).unwrap(); + let conn = easylapin::from_config(&cfg.rabbitmq)?; + let mut chan = task::block_on(conn.create_channel())?; let cloner = checkout::cached_cloner(Path::new(&cfg.checkout.root)); let nix = cfg.nix(); - let events = stats::RabbitMQ::from_amqp( + let events = stats::RabbitMQ::from_lapin( &format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()), - session.open_channel(3).unwrap(), + task::block_on(conn.create_channel())?, ); - let mrw = tasks::evaluate::EvaluationWorker::new( - cloner, - &nix, - cfg.github(), - cfg.github_app_vendingmachine(), - cfg.acl(), - cfg.runner.identity.clone(), - events, - cfg.tag_paths.clone().unwrap(), - ); + let queue_name = String::from("mass-rebuild-check-jobs"); + chan.declare_queue(easyamqp::QueueConfig { + queue: queue_name.clone(), + passive: false, + durable: true, + exclusive: false, + auto_delete: false, + no_wait: false, + })?; - channel - .declare_queue(easyamqp::QueueConfig { - queue: "mass-rebuild-check-jobs".to_owned(), - passive: false, - durable: true, - exclusive: false, - auto_delete: false, + let handle = chan.consume( + tasks::evaluate::EvaluationWorker::new( + cloner, + &nix, + cfg.github(), + cfg.github_app_vendingmachine(), + cfg.acl(), + cfg.runner.identity.clone(), + events, + cfg.tag_paths.clone().unwrap(), + ), + easyamqp::ConsumeConfig { + queue: queue_name.clone(), + consumer_tag: format!("{}-mass-rebuild-checker", cfg.whoami()), + no_local: false, + no_ack: false, no_wait: false, - }) - .unwrap(); + exclusive: false, + }, + )?; - channel.basic_prefetch(1).unwrap(); - let mut channel = channel - .consume( - worker::new(mrw), - easyamqp::ConsumeConfig { - queue: "mass-rebuild-check-jobs".to_owned(), - consumer_tag: format!("{}-mass-rebuild-checker", cfg.whoami()), - no_local: false, - no_ack: false, - no_wait: false, - exclusive: false, - }, - ) - .unwrap(); + info!("Fetching jobs from {}", queue_name); + task::block_on(handle); - channel.start_consuming(); - - info!("Finished consuming?"); - - channel.close(200, "Bye").unwrap(); - info!("Closed the channel"); - session.close(200, "Good Bye"); + drop(conn); // Close connection. info!("Closed the session... EOF"); + Ok(()) } From 910330c28dd6007d3a3285d2429e4bc19f6f9bd6 Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Thu, 21 May 2020 21:14:25 +0200 Subject: [PATCH 3/3] start span for evaluations in consumer --- ofborg/src/tasks/evaluate.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/ofborg/src/tasks/evaluate.rs b/ofborg/src/tasks/evaluate.rs index 31315cd..82c962c 100644 --- a/ofborg/src/tasks/evaluate.rs +++ b/ofborg/src/tasks/evaluate.rs @@ -80,6 +80,9 @@ impl worker::SimpleWorker for EvaluationWorker } fn consumer(&mut self, job: &evaluationjob::EvaluationJob) -> worker::Actions { + let span = debug_span!("job", pr = ?job.pr.number); + let _enter = span.enter(); + let mut vending_machine = self .github_vend .write() @@ -249,9 +252,6 @@ impl<'a, E: stats::SysEvents + 'static> OneEval<'a, E> { // FIXME: remove with rust/cargo update #[allow(clippy::cognitive_complexity)] fn evaluate_job(&mut self) -> Result { - let span = debug_span!("job", pr = ?self.job.pr.number); - let _enter = span.enter(); - let job = self.job; let repo = self .client_app