From cc8aaae5c353a311955c313c828b66f3b6a7cd31 Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Mon, 11 May 2020 23:25:17 +0200 Subject: [PATCH 1/2] convert comment filter to easylapin --- ofborg/src/bin/github-comment-filter.rs | 137 +++++++++++------------- ofborg/src/tasks/githubcommentfilter.rs | 2 + 2 files changed, 64 insertions(+), 75 deletions(-) diff --git a/ofborg/src/bin/github-comment-filter.rs b/ofborg/src/bin/github-comment-filter.rs index af18796..31a90b4 100644 --- a/ofborg/src/bin/github-comment-filter.rs +++ b/ofborg/src/bin/github-comment-filter.rs @@ -1,91 +1,78 @@ -use ofborg::config; -use ofborg::easyamqp::{self, ChannelExt, ConsumerExt}; -use ofborg::tasks; -use ofborg::worker; - use std::env; +use std::error::Error; -use amqp::Basic; +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(); - info!("Hello, world!"); + let arg = env::args() + .nth(1) + .expect("usage: github-comment-filter "); + let cfg = config::load(arg.as_ref()); - let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap(); - info!("Connected to rabbitmq"); + let conn = easylapin::from_config(&cfg.rabbitmq)?; + let mut chan = task::block_on(conn.create_channel())?; - let mut channel = session.open_channel(1).unwrap(); - channel - .declare_exchange(easyamqp::ExchangeConfig { - exchange: "github-events".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: "github-events".to_owned(), + exchange_type: easyamqp::ExchangeType::Topic, + passive: false, + durable: true, + auto_delete: false, + no_wait: false, + internal: false, + })?; - channel - .declare_exchange(easyamqp::ExchangeConfig { - exchange: "build-jobs".to_owned(), - exchange_type: easyamqp::ExchangeType::Fanout, - passive: false, - durable: true, - auto_delete: false, - no_wait: false, - internal: false, - }) - .unwrap(); + chan.declare_exchange(easyamqp::ExchangeConfig { + exchange: "build-jobs".to_owned(), + exchange_type: easyamqp::ExchangeType::Fanout, + passive: false, + durable: true, + auto_delete: false, + no_wait: false, + internal: false, + })?; - channel - .declare_queue(easyamqp::QueueConfig { + let queue_name = "build-inputs"; + chan.declare_queue(easyamqp::QueueConfig { + queue: queue_name.to_owned(), + passive: false, + durable: true, + exclusive: false, + auto_delete: false, + no_wait: false, + })?; + + chan.bind_queue(easyamqp::BindQueueConfig { + queue: "build-inputs".to_owned(), + exchange: "github-events".to_owned(), + routing_key: Some("issue_comment.*".to_owned()), + no_wait: false, + })?; + + let handle = chan.consume( + tasks::githubcommentfilter::GitHubCommentWorker::new(cfg.acl(), cfg.github()), + easyamqp::ConsumeConfig { queue: "build-inputs".to_owned(), - passive: false, - durable: true, + consumer_tag: format!("{}-github-comment-filter", cfg.whoami()), + no_local: false, + no_ack: false, + no_wait: false, exclusive: false, - auto_delete: false, - no_wait: false, - }) - .unwrap(); + }, + )?; - channel - .bind_queue(easyamqp::BindQueueConfig { - queue: "build-inputs".to_owned(), - exchange: "github-events".to_owned(), - routing_key: Some("issue_comment.*".to_owned()), - no_wait: false, - }) - .unwrap(); + info!("Fetching jobs from {}", &queue_name); + task::block_on(handle); - channel.basic_prefetch(1).unwrap(); - let mut channel = channel - .consume( - worker::new(tasks::githubcommentfilter::GitHubCommentWorker::new( - cfg.acl(), - cfg.github(), - )), - easyamqp::ConsumeConfig { - queue: "build-inputs".to_owned(), - consumer_tag: format!("{}-github-comment-filter", 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(()) } diff --git a/ofborg/src/tasks/githubcommentfilter.rs b/ofborg/src/tasks/githubcommentfilter.rs index 03a1c0d..3e121b8 100644 --- a/ofborg/src/tasks/githubcommentfilter.rs +++ b/ofborg/src/tasks/githubcommentfilter.rs @@ -34,6 +34,8 @@ impl worker::SimpleWorker for GitHubCommentWorker { } } + // FIXME: remove with rust/cargo update + #[allow(clippy::cognitive_complexity)] fn consumer(&mut self, job: &ghevent::IssueComment) -> worker::Actions { if job.action == ghevent::IssueCommentAction::Deleted { return vec![worker::Action::Ack]; From 41883df47b62000b56db02b29ee6cd9961b5d354 Mon Sep 17 00:00:00 2001 From: Daiderd Jordan Date: Mon, 11 May 2020 23:28:57 +0200 Subject: [PATCH 2/2] add basic structured logging to comment filter --- ofborg/src/tasks/githubcommentfilter.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/ofborg/src/tasks/githubcommentfilter.rs b/ofborg/src/tasks/githubcommentfilter.rs index 3e121b8..9314f67 100644 --- a/ofborg/src/tasks/githubcommentfilter.rs +++ b/ofborg/src/tasks/githubcommentfilter.rs @@ -4,7 +4,7 @@ use crate::ghevent; use crate::message::{buildjob, evaluationjob, Pr, Repo}; use crate::worker; -use tracing::{error, info}; +use tracing::{debug_span, error, info}; use uuid::Uuid; pub struct GitHubCommentWorker { @@ -37,6 +37,9 @@ impl worker::SimpleWorker for GitHubCommentWorker { // FIXME: remove with rust/cargo update #[allow(clippy::cognitive_complexity)] fn consumer(&mut self, job: &ghevent::IssueComment) -> worker::Actions { + let span = debug_span!("job", pr = ?job.issue.number); + let _enter = span.enter(); + if job.action == ghevent::IssueCommentAction::Deleted { return vec![worker::Action::Ack]; }