use std::{collections::HashMap, time::Duration};
use talos_certifier::{
model::{metrics::TxProcessingTimeline, CandidateMessage},
ports::MessageReciever,
ChannelMessage,
};
use talos_certifier_adapters::KafkaConsumer;
use talos_metrics::model::MinMax;
use talos_rdkafka_utils::kafka_config::KafkaConfig;
use time::OffsetDateTime;
use tokio::time::timeout;
#[tokio::main]
async fn main() -> Result<(), String> {
env_logger::builder().format_timestamp_millis().init();
let mut kafka_config = KafkaConfig::from_env(None);
kafka_config.extend(
None,
Some(HashMap::from([
("auto.commit.enable".into(), "false".into()),
("auto.offset.reset".into(), "earliest".into()),
])),
);
kafka_config.group_id = format!("talos-metric-histogram-timings-{}", uuid::Uuid::new_v4());
let mut kafka_consumer = KafkaConsumer::new(&kafka_config);
kafka_consumer.subscribe().await.unwrap();
log::warn!("Aggregating timelines...");
let aggregates = aggregate_timelines(&mut kafka_consumer).await?;
let total_decisions = aggregates.1;
let progress_frequency = if total_decisions < 100 {
10
} else {
(total_decisions as f32 / 100_f32 * 5_f32) as u64
};
kafka_consumer.unsubscribe().await;
kafka_config.group_id = format!("talos-metric-histogram-{}", uuid::Uuid::new_v4());
let mut kafka_consumer2: KafkaConsumer<CandidateMessage> = KafkaConsumer::new(&kafka_config);
kafka_consumer2.subscribe().await.unwrap();
log::warn!("Computing histograms of {} decisions", total_decisions);
let mut bukets_start_at = MinMax::default();
bukets_start_at.add(aggregates.0.candidate_published.min);
bukets_start_at.add(aggregates.0.candidate_received.min);
bukets_start_at.add(aggregates.0.candidate_processing_started.min);
bukets_start_at.add(aggregates.0.decision_created_at.min);
bukets_start_at.add(aggregates.0.db_save_started.min);
bukets_start_at.add(aggregates.0.db_save_ended.min);
let earliest_bucket_start = bukets_start_at.min;
log::warn!(
"Earliest bucket starts at {:?}",
OffsetDateTime::from_unix_timestamp_nanos(earliest_bucket_start)
);
let mut hist_candidate_published = Histogram::new(earliest_bucket_start);
let mut hist_candidate_received = Histogram::new(earliest_bucket_start);
let mut hist_candidate_processing_started = Histogram::new(earliest_bucket_start);
let mut hist_decision_created_at = Histogram::new(earliest_bucket_start);
let mut hist_db_save_started = Histogram::new(earliest_bucket_start);
let mut hist_db_save_ended = Histogram::new(earliest_bucket_start);
let mut item_number = 0_u64;
loop {
let rslt_consumed = timeout(Duration::from_secs(10), kafka_consumer2.consume_message()).await;
let message = match rslt_consumed {
Ok(Ok(Some(message))) => message,
Ok(Err(_)) => continue,
_ => break,
};
if let ChannelMessage::Decision(decision) = message {
let decision_message = decision.message;
item_number += 1;
if item_number.is_multiple_of(progress_frequency) {
log::warn!("Progress: {} of {}", item_number, total_decisions);
}
hist_candidate_published.include(&decision_message.metrics, decision_message.metrics.candidate_published);
hist_candidate_received.include(&decision_message.metrics, decision_message.metrics.candidate_received);
hist_candidate_processing_started.include(&decision_message.metrics, decision_message.metrics.candidate_processing_started);
hist_decision_created_at.include(&decision_message.metrics, decision_message.metrics.decision_created_at);
hist_db_save_started.include(&decision_message.metrics, decision_message.metrics.db_save_started);
hist_db_save_ended.include(&decision_message.metrics, decision_message.metrics.db_save_ended);
if item_number == total_decisions {
break;
}
}
}
log::warn!("Reading of messages ended");
kafka_consumer2.unsubscribe().await;
let mut bucket_length = MinMax::default();
bucket_length.add(hist_candidate_published.buckets.len() as i128);
bucket_length.add(hist_candidate_received.buckets.len() as i128);
bucket_length.add(hist_candidate_processing_started.buckets.len() as i128);
bucket_length.add(hist_decision_created_at.buckets.len() as i128);
bucket_length.add(hist_db_save_started.buckets.len() as i128);
bucket_length.add(hist_db_save_ended.buckets.len() as i128);
log::warn!("Histograms are ready, there are: {} 1 second buckets.", bucket_length.max);
log::warn!("Headers: 'By Publish Time',,,,,,'-','By Receive Time',,,,,,'-','Processing Started',,,,,,'-','By Decision Created',,,,,,'-','By DB Save Started',,,,,,'-','By DB Save Ended',,,,,");
for bucket in 1..bucket_length.max {
log::warn!(
"METRIC (histograms MAX): {},'-',{},'-',{},'-',{},'-',{},'-',{}",
hist_candidate_published.to_csv_max(bucket).await,
hist_candidate_received.to_csv_max(bucket).await,
hist_candidate_processing_started.to_csv_max(bucket).await,
hist_decision_created_at.to_csv_max(bucket).await,
hist_db_save_started.to_csv_max(bucket).await,
hist_db_save_ended.to_csv_max(bucket).await,
);
}
Ok(())
}
async fn aggregate_timelines(consumer: &mut KafkaConsumer<CandidateMessage>) -> Result<(TimelineAggregates, u64), String> {
let mut total = 0_u64;
let mut aggregates = TimelineAggregates::default();
loop {
let rslt_consumed = timeout(Duration::from_secs(3), consumer.consume_message()).await;
let message = match rslt_consumed {
Ok(Ok(Some(message))) => message,
Ok(Err(_)) => continue,
_ => break,
};
if let ChannelMessage::Decision(decision) = message {
total += 1;
aggregates.merge(decision.message.metrics);
}
}
Ok((aggregates, total))
}
#[derive(Default)]
struct TimelineAggregates {
pub candidate_published: MinMax,
pub candidate_received: MinMax,
pub candidate_processing_started: MinMax,
pub decision_created_at: MinMax,
pub db_save_started: MinMax,
pub db_save_ended: MinMax,
}
impl TimelineAggregates {
fn merge(&mut self, data: TxProcessingTimeline) {
self.candidate_published.add(data.candidate_published);
self.candidate_received.add(data.candidate_received);
self.candidate_processing_started.add(data.candidate_processing_started);
self.decision_created_at.add(data.decision_created_at);
self.db_save_started.add(data.db_save_started);
self.db_save_ended.add(data.db_save_ended);
}
}
#[derive(Clone)]
struct Spans {
pub span1_candidate_kafka_trip: MinMax,
pub span2_talos_internal_bus: MinMax,
pub span3_deciding: MinMax,
pub span4_prepare_for_save: MinMax,
pub span5_save_to_db: MinMax,
}
impl Spans {
fn new() -> Self {
Self {
span1_candidate_kafka_trip: MinMax::default(),
span2_talos_internal_bus: MinMax::default(),
span3_deciding: MinMax::default(),
span4_prepare_for_save: MinMax::default(),
span5_save_to_db: MinMax::default(),
}
}
fn include(&mut self, times: &TxProcessingTimeline) {
self.span1_candidate_kafka_trip.add(times.candidate_received - times.candidate_published);
self.span2_talos_internal_bus.add(times.candidate_processing_started - times.candidate_received);
self.span3_deciding.add(times.decision_created_at - times.candidate_processing_started);
self.span4_prepare_for_save.add(times.db_save_started - times.decision_created_at);
self.span5_save_to_db.add(times.db_save_ended - times.db_save_started);
}
}
struct Histogram {
pub start_time: i128,
pub buckets: Vec<(u64, Spans)>,
}
impl Histogram {
fn new(start_time: i128) -> Self {
Histogram {
start_time,
buckets: Vec::new(),
}
}
fn include(&mut self, metrics: &TxProcessingTimeline, value: i128) {
let elapsed = Duration::from_nanos((value - self.start_time) as u64);
let bucket = elapsed.as_secs_f32().ceil() as usize;
while self.buckets.len() < bucket + 1 {
self.buckets.push((0, Spans::new()));
}
if let Some((count, spans)) = self.buckets.get_mut(bucket) {
spans.include(metrics);
self.buckets[bucket].0 = *count + 1;
}
}
async fn get(&self, bucket: usize) -> (u64, Spans) {
if let Some(data) = self.buckets.get(bucket) {
(data.0, data.1.clone())
} else {
(0, Spans::new())
}
}
async fn to_csv_max(&self, bucket: i128) -> String {
let (count, spans) = self.get(bucket as usize).await;
format!(
"{},{},{},{},{},{}",
count,
spans.span1_candidate_kafka_trip.max,
spans.span2_talos_internal_bus.max,
spans.span3_deciding.max,
spans.span4_prepare_for_save.max,
spans.span5_save_to_db.max,
)
}
}