Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Send per queue metrics #3

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Send per queue metrics #3

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7c63589
Print welcome message, sleep after sending metrics
Mongey Nov 20, 2019
9174b0d
Push per-queue metrics with queue_name tag
Mongey Nov 20, 2019
0240ffa
Format
Mongey Nov 20, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
171 changes: 124 additions & 47 deletions src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,24 @@
extern crate redis;
extern crate curl;
extern crate redis;
extern crate serde;
#[macro_use] extern crate serde_derive;
#[macro_use]
extern crate serde_derive;
extern crate serde_json;
extern crate time;
#[macro_use] extern crate log;
#[macro_use]
extern crate log;
extern crate env_logger;

use curl::easy::{Easy, List as HeaderList};
use redis::{cmd, Client as RedisClient, Commands, Connection, RedisError};
use std::collections::HashMap;
use std::collections::HashSet;
use std::env;
use std::io::Read;
use std::error::Error;
use std::fmt::Display;
use std::io::Read;
use std::thread::sleep;
use std::time::Duration;
use std::collections::HashSet;
use std::error::Error;
use redis::{Client as RedisClient, Commands, cmd, Connection, RedisError};
use curl::easy::{Easy, List as HeaderList};

const DD_SERIES_URL: &str = "https://app.datadoghq.com/api/v1/series";
const DEFAULT_INTERVAL: u64 = 60;
Expand All @@ -39,18 +42,28 @@ impl Series {
struct Metric {
metric: String,
points: Vec<(i64, u32)>,
#[serde(rename="type")]
#[serde(rename = "type")]
metric_type: String,
tags: Vec<String>,
}

impl Metric {
fn new<M: Into<String>, MT: Into<String>>(metric: M, value: u32, metric_type: MT, tags: Vec<String>) -> Self {
fn new<M: Into<String>, MT: Into<String>>(
metric: M,
value: u32,
metric_type: MT,
tags: Vec<String>,
) -> Self {
let metric = metric.into();
let points = vec![(time::get_time().sec, value)];
let metric_type = metric_type.into();

Metric { metric, points, metric_type, tags }
Metric {
metric,
points,
metric_type,
tags,
}
}

fn gauge<M: Into<String>>(metric: M, value: u32, tags: Vec<String>) -> Self {
Expand Down Expand Up @@ -79,13 +92,19 @@ fn main() {
env_logger::init().unwrap();

let interval = env::var("INTERVAL")
.map(|value| value.parse::<u64>().expect("INTERVAL is not a valid number."))
.map(|value| {
value
.parse::<u64>()
.expect("INTERVAL is not a valid number.")
})
.unwrap_or(DEFAULT_INTERVAL);

let redis_url = env::var("REDIS_URL").expect("REDIS_URL is missing.");
let redis_ns = Namespace::new(env::var("REDIS_NAMESPACE").ok());
let redis = RedisClient::open(&*redis_url).expect("Could not connect to the provided REDIS_URL.");
let mut conn = establish_redis_connection(&redis).expect("Could not establish a connection to Redis.");
let redis =
RedisClient::open(&*redis_url).expect("Could not connect to the provided REDIS_URL.");
let mut conn =
establish_redis_connection(&redis).expect("Could not establish a connection to Redis.");
let mut reconnect = false;

let dd_api_key = env::var("DD_API_KEY").expect("DD_API_KEY is missing.");
Expand All @@ -98,9 +117,8 @@ fn main() {

let mut previous_total_processed: Option<u32> = None;

debug!("Welcome to datadog-sidekiq");
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

given the usage of info! around the existing code base I could see this as info! as well (rather than debug!).

loop {
sleep(Duration::new(interval, 0));

// Redis was down. Try to re-establish a connection.
if reconnect {
info!("Trying to connect to Redis again.");
Expand All @@ -109,21 +127,39 @@ fn main() {
Ok(new_connection) => {
conn = new_connection;
reconnect = false;
},
}
Err(redis_err) => {
error!("An error occured while trying to re-establish the Redis connection: {}", redis_err);
error!(
"An error occured while trying to re-establish the Redis connection: {}",
redis_err
);
continue;
}
}
}

let mut series = Series::new();

match get_total_enqueued(&conn, &redis_ns) {
Ok(Some(total_enqueued)) => {
series.push(Metric::gauge("sidekiq.enqueued", total_enqueued, tags.clone()));
},
Ok(None) => {},
match get_enqueued(&conn, &redis_ns) {
Ok(ref enqueued_count_by_queue) => {
for (queue_name, count) in enqueued_count_by_queue {
let mut extra = vec!["queue_name:".to_string() + queue_name];
let mut new_tags = tags.clone();
new_tags.append(&mut extra);
series.push(Metric::gauge("sidekiq.enqueued", *count, new_tags));
}

let mut total_enqueued = 0;
for (_queue_name, count) in enqueued_count_by_queue {
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please use something like https://doc.rust-lang.org/std/iter/trait.Sum.html

total_enqueued += count;
}
series.push(Metric::gauge(
"sidekiq.total_enqueued",
total_enqueued,
tags.clone(),
));
}

Err(redis_err) => {
error!("A Redis error occured: {}", redis_err);
reconnect = true;
Expand All @@ -134,42 +170,64 @@ fn main() {
match get_total_processed(&conn, &redis_ns) {
Ok(total_processed) => {
if previous_total_processed.is_some() && total_processed.is_some() {
let current_processed = total_processed.unwrap() - previous_total_processed.unwrap();
series.push(Metric::gauge("sidekiq.processed", current_processed, tags.clone()));
let current_processed =
total_processed.unwrap() - previous_total_processed.unwrap();
series.push(Metric::gauge(
"sidekiq.processed",
current_processed,
tags.clone(),
));
}

previous_total_processed = total_processed;
},
}
Err(redis_err) => {
error!("A Redis error occured: {}", redis_err);
reconnect = true;
continue;
},
}
}

match deliver_series(&url, series) {
Ok(_) => {},
Err(err) => { error!("An error occured while sending the series to DataDog: {}", err); }
Ok(_) => {}
Err(err) => {
error!(
"An error occured while sending the series to DataDog: {}",
err
);
}
}

sleep(Duration::new(interval, 0));
}
}

fn establish_redis_connection(client: &RedisClient) -> Result<Connection, RedisError> {
client.get_connection()
}

fn get_total_enqueued(conn: &Connection, redis_ns: &Namespace) -> Result<Option<u32>, RedisError> {
fn get_enqueued(
conn: &Connection,
redis_ns: &Namespace,
) -> Result<HashMap<String, u32>, RedisError> {
let mut actual_result: HashMap<String, u32> = HashMap::new();
let queues: HashSet<String> = conn.smembers(redis_ns.wrap("queues"))?;

let mut pipe = redis::pipe();
for queue in queues {

for queue in &queues {
let queue_name = ["queue", &queue].join(":");
pipe.add_command(cmd("LLEN").arg(redis_ns.wrap(queue_name)));
}

let total_per_queue: Vec<u32> = pipe.query(conn)?;
let mut i: usize = 0;
for queue in &queues {
actual_result.insert(queue.to_string(), total_per_queue[i]);
i += 1;
}

Ok(Some(total_per_queue.iter().sum()))
Ok(actual_result)
}

fn get_total_processed(conn: &Connection, redis_ns: &Namespace) -> Result<Option<u32>, RedisError> {
Expand All @@ -178,38 +236,57 @@ fn get_total_processed(conn: &Connection, redis_ns: &Namespace) -> Result<Option

fn deliver_series(url: &str, series: Series) -> Result<(), String> {
let mut request = Easy::new();
request.post(true).map_err(|err| err.description().to_string())?;
request.url(url).map_err(|err| err.description().to_string())?;
request
.post(true)
.map_err(|err| err.description().to_string())?;
request
.url(url)
.map_err(|err| err.description().to_string())?;

let mut headers = HeaderList::new();
headers.append("content-type: application/json").map_err(|err| err.description().to_string())?;
request.http_headers(headers).map_err(|err| err.description().to_string())?;
headers
.append("content-type: application/json")
.map_err(|err| err.description().to_string())?;
request
.http_headers(headers)
.map_err(|err| err.description().to_string())?;

let body = serde_json::to_string(&series).map_err(|err| err.to_string())?;
info!("\nPOST {}\n{}", url, body);

request.post_field_size(body.len() as u64).map_err(|err| err.description().to_string())?;
request
.post_field_size(body.len() as u64)
.map_err(|err| err.description().to_string())?;

let mut response = vec![];
{
let mut transfer = request.transfer();

transfer.read_function(|into| {
Ok(body.as_bytes().read(into).unwrap_or(0))
}).map_err(|err| err.description().to_string())?;
transfer
.read_function(|into| Ok(body.as_bytes().read(into).unwrap_or(0)))
.map_err(|err| err.description().to_string())?;

transfer.write_function(|data| {
response.extend_from_slice(data);
Ok(data.len())
}).map_err(|err| err.description().to_string())?;
transfer
.write_function(|data| {
response.extend_from_slice(data);
Ok(data.len())
})
.map_err(|err| err.description().to_string())?;

transfer.perform().map_err(|err| err.description().to_string())?;
transfer
.perform()
.map_err(|err| err.description().to_string())?;
}

let status_code = request.response_code().map_err(|err| err.description().to_string())?;
let status_code = request
.response_code()
.map_err(|err| err.description().to_string())?;
if status_code == 202 {
Ok(())
} else {
Err(format!("The DataDog server replied with an error: {}", String::from_utf8_lossy(&response)))
Err(format!(
"The DataDog server replied with an error: {}",
String::from_utf8_lossy(&response)
))
}
}