From 51b4c3dbc94e18279251dc5d69e4677c8523a164 Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Fri, 22 May 2020 19:29:59 +0200 Subject: [PATCH 1/3] convert log-message-collector to easylapin --- ofborg/src/bin/log-message-collector.rs | 116 +++++++++++------------- 1 file changed, 55 insertions(+), 61 deletions(-) diff --git a/ofborg/src/bin/log-message-collector.rs b/ofborg/src/bin/log-message-collector.rs index a9e5cef..175eeb0 100644 --- a/ofborg/src/bin/log-message-collector.rs +++ b/ofborg/src/bin/log-message-collector.rs @@ -1,78 +1,72 @@ -use ofborg::config; -use ofborg::easyamqp::{self, ChannelExt, ConsumerExt}; -use ofborg::tasks; -use ofborg::worker; - use std::env; +use std::error::Error; use std::path::PathBuf; +use async_std::task; use tracing::info; -fn main() { - let cfg = config::load(env::args().nth(1).unwrap().as_ref()); +use ofborg::config; +use ofborg::easyamqp::{self, ChannelExt, ConsumerExt}; +use ofborg::easylapin; +use ofborg::tasks; + +fn main() -> Result<(), Box> { ofborg::setup_log(); - let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap(); - info!("Connected to rabbitmq"); + let arg = env::args() + .nth(1) + .expect("usage: log-message-collector "); + let cfg = config::load(arg.as_ref()); - let mut channel = session.open_channel(1).unwrap(); + let conn = easylapin::from_config(&cfg.rabbitmq)?; + let mut chan = task::block_on(conn.create_channel())?; - channel - .declare_exchange(easyamqp::ExchangeConfig { - exchange: "logs".to_owned(), - exchange_type: easyamqp::ExchangeType::Topic, - passive: false, - durable: true, - auto_delete: false, - no_wait: false, - internal: false, - }) - .unwrap(); + chan.declare_exchange(easyamqp::ExchangeConfig { + exchange: "logs".to_owned(), + exchange_type: easyamqp::ExchangeType::Topic, + passive: false, + durable: true, + auto_delete: false, + no_wait: false, + internal: false, + })?; let queue_name = "".to_owned(); - channel - .declare_queue(easyamqp::QueueConfig { + chan.declare_queue(easyamqp::QueueConfig { + queue: queue_name.clone(), + passive: false, + durable: false, + exclusive: true, + auto_delete: true, + no_wait: false, + })?; + + chan.bind_queue(easyamqp::BindQueueConfig { + queue: queue_name.clone(), + exchange: "logs".to_owned(), + routing_key: Some("*.*".to_owned()), + no_wait: false, + })?; + + let handle = chan.consume( + tasks::log_message_collector::LogMessageCollector::new( + PathBuf::from(cfg.log_storage.clone().unwrap().path), + 100, + ), + easyamqp::ConsumeConfig { queue: queue_name.clone(), - passive: false, - durable: false, - exclusive: true, - auto_delete: true, + consumer_tag: format!("{}-log-collector", cfg.whoami()), + no_local: false, + no_ack: false, no_wait: false, - }) - .unwrap(); + exclusive: false, + }, + )?; - channel - .bind_queue(easyamqp::BindQueueConfig { - queue: queue_name.clone(), - exchange: "logs".to_owned(), - routing_key: Some("*.*".to_owned()), - no_wait: false, - }) - .unwrap(); + info!("Fetching jobs from {}", &queue_name); + task::block_on(handle); - let mut channel = channel - .consume( - worker::new(tasks::log_message_collector::LogMessageCollector::new( - PathBuf::from(cfg.log_storage.clone().unwrap().path), - 100, - )), - easyamqp::ConsumeConfig { - queue: queue_name, - consumer_tag: format!("{}-log-collector", cfg.whoami()), - no_local: false, - no_ack: false, - no_wait: false, - exclusive: false, - }, - ) - .unwrap(); - - 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 f652901df227966574508afc59a88f0a74c41861 Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Fri, 22 May 2020 21:55:18 +0200 Subject: [PATCH 2/3] move prefetch qos to WorkerChannel This way the log collecter can use a channel with prefetching while the other workers still consume items one by one. --- ofborg/src/bin/evaluation-filter.rs | 2 +- ofborg/src/bin/github-comment-filter.rs | 2 +- ofborg/src/bin/github-comment-poster.rs | 2 +- ofborg/src/bin/log-message-collector.rs | 1 + ofborg/src/bin/mass-rebuilder.rs | 2 +- ofborg/src/easylapin.rs | 16 ++++++++++++++-- 6 files changed, 19 insertions(+), 6 deletions(-) diff --git a/ofborg/src/bin/evaluation-filter.rs b/ofborg/src/bin/evaluation-filter.rs index 76bf5b0..1c5c032 100644 --- a/ofborg/src/bin/evaluation-filter.rs +++ b/ofborg/src/bin/evaluation-filter.rs @@ -56,7 +56,7 @@ fn main() -> Result<(), Box> { no_wait: false, })?; - let handle = chan.consume( + let handle = easylapin::WorkerChannel(chan).consume( tasks::evaluationfilter::EvaluationFilterWorker::new(cfg.acl()), easyamqp::ConsumeConfig { queue: queue_name.clone(), diff --git a/ofborg/src/bin/github-comment-filter.rs b/ofborg/src/bin/github-comment-filter.rs index 31a90b4..acee59b 100644 --- a/ofborg/src/bin/github-comment-filter.rs +++ b/ofborg/src/bin/github-comment-filter.rs @@ -57,7 +57,7 @@ fn main() -> Result<(), Box> { no_wait: false, })?; - let handle = chan.consume( + let handle = easylapin::WorkerChannel(chan).consume( tasks::githubcommentfilter::GitHubCommentWorker::new(cfg.acl(), cfg.github()), easyamqp::ConsumeConfig { queue: "build-inputs".to_owned(), diff --git a/ofborg/src/bin/github-comment-poster.rs b/ofborg/src/bin/github-comment-poster.rs index 4e623b8..0521db2 100644 --- a/ofborg/src/bin/github-comment-poster.rs +++ b/ofborg/src/bin/github-comment-poster.rs @@ -46,7 +46,7 @@ fn main() -> Result<(), Box> { no_wait: false, })?; - let handle = chan.consume( + let handle = easylapin::WorkerChannel(chan).consume( tasks::githubcommentposter::GitHubCommentPoster::new(cfg.github_app_vendingmachine()), easyamqp::ConsumeConfig { queue: "build-results".to_owned(), diff --git a/ofborg/src/bin/log-message-collector.rs b/ofborg/src/bin/log-message-collector.rs index 175eeb0..c6d914a 100644 --- a/ofborg/src/bin/log-message-collector.rs +++ b/ofborg/src/bin/log-message-collector.rs @@ -48,6 +48,7 @@ fn main() -> Result<(), Box> { no_wait: false, })?; + // Regular channel, we want prefetching here. let handle = chan.consume( tasks::log_message_collector::LogMessageCollector::new( PathBuf::from(cfg.log_storage.clone().unwrap().path), diff --git a/ofborg/src/bin/mass-rebuilder.rs b/ofborg/src/bin/mass-rebuilder.rs index d72f171..d43cca6 100644 --- a/ofborg/src/bin/mass-rebuilder.rs +++ b/ofborg/src/bin/mass-rebuilder.rs @@ -53,7 +53,7 @@ fn main() -> Result<(), Box> { no_wait: false, })?; - let handle = chan.consume( + let handle = easylapin::WorkerChannel(chan).consume( tasks::evaluate::EvaluationWorker::new( cloner, &nix, diff --git a/ofborg/src/easylapin.rs b/ofborg/src/easylapin.rs index d6fa113..1820529 100644 --- a/ofborg/src/easylapin.rs +++ b/ofborg/src/easylapin.rs @@ -86,8 +86,6 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for CloseOnDrop { type Handle = Pin + 'a>>; fn consume(self, mut worker: W, config: ConsumeConfig) -> Result { - task::block_on(self.basic_qos(1, BasicQosOptions::default()))?; - let mut consumer = task::block_on(self.basic_consume( &config.queue, &config.consumer_tag, @@ -117,6 +115,20 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for CloseOnDrop { } } +// Same as a regular channel, but without prefetching, +// used for services with multiple instances. +pub struct WorkerChannel(pub CloseOnDrop); + +impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for WorkerChannel { + type Error = lapin::Error; + type Handle = Pin + 'a>>; + + fn consume(self, worker: W, config: ConsumeConfig) -> Result { + task::block_on(self.0.basic_qos(1, BasicQosOptions::default()))?; + self.0.consume(worker, config) + } +} + struct ChannelNotificationReceiver<'a> { channel: &'a mut CloseOnDrop, deliver: &'a Delivery, From b6e870f5cb3e0191c7622ca6e55cdd9fdeed777b Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Sat, 23 May 2020 21:41:11 +0200 Subject: [PATCH 3/3] Update ofborg/src/easylapin.rs Co-authored-by: Cole Helbling --- ofborg/src/easylapin.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ofborg/src/easylapin.rs b/ofborg/src/easylapin.rs index 1820529..5d97893 100644 --- a/ofborg/src/easylapin.rs +++ b/ofborg/src/easylapin.rs @@ -115,8 +115,8 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for CloseOnDrop { } } -// Same as a regular channel, but without prefetching, -// used for services with multiple instances. +/// Same as a regular channel, but without prefetching, +/// used for services with multiple instances. pub struct WorkerChannel(pub CloseOnDrop); impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for WorkerChannel {