the-distro/ofborg
2
0
Fork 1
Code Issues 1 Pull requests 1 Packages Projects Releases Wiki Activity

convert stats to easyamqp

Browse source
This commit is contained in:
Daiderd Jordan 2020-05-20 23:28:56 +02:00
parent 874bfcd028
commit 41c2899e3c
No known key found for this signature in database
GPG key ID: D02435D05B810C96
1 changed files with 55 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

123
ofborg/src/bin/stats.rs
View file

@ -1,96 +1,83 @@
use ofborg::easyamqp::{ChannelExt, ConsumerExt};
use ofborg::{config, easyamqp, stats, tasks, worker};
use std::env; use std::env;
use std::error::Error;
use std::thread; use std::thread;
use amqp::Basic; use async_std::task;
use hyper::server::{Request, Response, Server}; use hyper::server::{Request, Response, Server};
use tracing::info; use tracing::info;
fn main() { use ofborg::easyamqp::{ChannelExt, ConsumerExt};
let cfg = config::load(env::args().nth(1).unwrap().as_ref()); use ofborg::{config, easyamqp, easylapin, stats, tasks};
fn main() -> Result<(), Box<dyn Error>> {
ofborg::setup_log(); ofborg::setup_log();
info!("Hello, world!"); let arg = env::args().nth(1).expect("usage: stats <config>");
let cfg = config::load(arg.as_ref());
let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap(); let conn = easylapin::from_config(&cfg.rabbitmq)?;
info!("Connected to rabbitmq"); let mut chan = task::block_on(conn.create_channel())?;
let events = stats::RabbitMQ::from_amqp( let events = stats::RabbitMQ::from_lapin(
&format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()), &format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()),
session.open_channel(3).unwrap(), task::block_on(conn.create_channel())?,
); );
let metrics = stats::MetricCollector::new(); let metrics = stats::MetricCollector::new();
let collector = tasks::statscollector::StatCollectorWorker::new(events, metrics.clone()); let collector = tasks::statscollector::StatCollectorWorker::new(events, metrics.clone());
let mut channel = session.open_channel(1).unwrap(); chan.declare_exchange(easyamqp::ExchangeConfig {
channel exchange: "stats".to_owned(),
.declare_exchange(easyamqp::ExchangeConfig { exchange_type: easyamqp::ExchangeType::Fanout,
exchange: "stats".to_owned(), passive: false,
exchange_type: easyamqp::ExchangeType::Fanout, durable: true,
passive: false, auto_delete: false,
durable: true, no_wait: false,
auto_delete: false, internal: false,
no_wait: false, })?;
internal: false,
})
.unwrap();
channel let queue_name = String::from("stats-events");
.declare_queue(easyamqp::QueueConfig { chan.declare_queue(easyamqp::QueueConfig {
queue: queue_name.clone(),
passive: false,
durable: true,
exclusive: false,
auto_delete: false,
no_wait: false,
})?;
chan.bind_queue(easyamqp::BindQueueConfig {
queue: queue_name.clone(),
exchange: "stats".to_owned(),
routing_key: None,
no_wait: false,
})?;
let handle = chan.consume(
collector,
easyamqp::ConsumeConfig {
queue: "stats-events".to_owned(), queue: "stats-events".to_owned(),
passive: false, consumer_tag: format!("{}-prometheus-stats-collector", cfg.whoami()),
durable: true, no_local: false,
no_ack: false,
no_wait: false,
exclusive: false, exclusive: false,
auto_delete: false, },
no_wait: false, )?;
})
.unwrap();
channel
.bind_queue(easyamqp::BindQueueConfig {
queue: "stats-events".to_owned(),
exchange: "stats".to_owned(),
routing_key: None,
no_wait: false,
})
.unwrap();
channel.basic_prefetch(1).unwrap();
let mut channel = channel
.consume(
worker::new(collector),
easyamqp::ConsumeConfig {
queue: "stats-events".to_owned(),
consumer_tag: format!("{}-prometheus-stats-collector", cfg.whoami()),
no_local: false,
no_ack: false,
no_wait: false,
exclusive: false,
},
)
.unwrap();
thread::spawn(|| { thread::spawn(|| {
let addr = "0.0.0.0:9898"; let addr = "0.0.0.0:9898";
info!("listening addr {:?}", addr); info!("listening addr {:?}", addr);
Server::http(addr) Server::http(addr)?.handle(move |_: Request, res: Response| {
.unwrap() res.send(metrics.prometheus_output().as_bytes()).unwrap();
.handle(move |_: Request, res: Response| { })?;
res.send(metrics.prometheus_output().as_bytes()).unwrap(); Ok::<_, Box<dyn Error + Sync + Send + '_>>(())
})
.unwrap();
}); });
channel.start_consuming(); info!("Fetching jobs from {}", &queue_name);
task::block_on(handle);
info!("Finished consuming?"); drop(conn); // Close connection.
channel.close(200, "Bye").unwrap();
info!("Closed the channel");
session.close(200, "Good Bye");
info!("Closed the session... EOF"); info!("Closed the session... EOF");
Ok(())
} }