diff --git a/Dockerfile.chronos b/Dockerfile.chronos index b4b6515..b10b08b 100644 --- a/Dockerfile.chronos +++ b/Dockerfile.chronos @@ -15,9 +15,26 @@ RUN adduser \ "${USER}" WORKDIR /tmp -COPY ./ . # Build binary in release mode -RUN cargo build -p chronos_bin --release +COPY ./Cargo.lock \ + ./rust-toolchain.toml \ + ./Cargo.toml \ + ./ + +# Copy in pg_mig and examples so we don't break +# cargo builds (cargo will fail if it can't find pg_mig and examples) +# One per line so its cache optimized +COPY chronos_bin /tmp/chronos_bin +COPY pg_mig /tmp/pg_mig +COPY examples /tmp/examples + +# Local caches to optimize the local build exp +RUN --mount=type=cache,target=/usr/local/cargo/registry/ \ + --mount=type=cache,target=/tmp/target/ \ + cargo build \ + -p chronos_bin \ + --release \ + --target-dir /build # # Run image based on bookworm-slim to reduce image size while still using glibc @@ -31,7 +48,7 @@ WORKDIR /opt/build COPY --from=build /etc/passwd /etc/passwd COPY --from=build /etc/group /etc/group # Copy binary from build -COPY --from=build /tmp/target/release/chronos ./ +COPY --from=build /build/release/chronos /opt/build/chronos # Use an unprivileged user USER ${USER}:${USER} # Entry point diff --git a/chronos_bin/src/message_processor.rs b/chronos_bin/src/message_processor.rs index 4dac45a..b20e411 100644 --- a/chronos_bin/src/message_processor.rs +++ b/chronos_bin/src/message_processor.rs @@ -2,7 +2,7 @@ use crate::kafka::producer::KafkaProducer; use crate::postgres::pg::{GetReady, Pg, TableRow}; use crate::utils::config::ChronosConfig; use crate::utils::delay_controller::DelayController; -use chrono::Utc; +use chrono::Local; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; @@ -115,9 +115,7 @@ impl MessageProcessor { async fn processor_message_ready(&self, node_id: Uuid) { loop { let method_name = "processor_message_ready"; - - let deadline = Utc::now() - Duration::from_secs(ChronosConfig::from_env().time_advance); - + let deadline = Local::now() - Duration::from_secs(ChronosConfig::from_env().time_advance); let param = GetReady { readied_at: deadline, readied_by: node_id, @@ -157,7 +155,6 @@ impl MessageProcessor { if e.contains("could not serialize access due to concurrent update") { log::warn!("{}: could not serialize access due to concurrent update", method_name); } - log::error!("{}: occurred while processing message ready {}", method_name, e); } } @@ -175,7 +172,6 @@ impl MessageProcessor { log::debug!("MessageProcessor loop"); tokio::time::sleep(Duration::from_millis(10)).await; self.processor_message_ready(node_id).await; - delay_controller.sleep().await; } } diff --git a/chronos_bin/src/message_receiver.rs b/chronos_bin/src/message_receiver.rs index 3a98cfe..3b03b1c 100644 --- a/chronos_bin/src/message_receiver.rs +++ b/chronos_bin/src/message_receiver.rs @@ -1,10 +1,12 @@ -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Local}; +use rdkafka::Message; use serde_json::json; use tracing::instrument; use crate::kafka::consumer::KafkaConsumer; use crate::kafka::producer::KafkaProducer; use crate::postgres::pg::{Pg, TableInsertRow}; +use crate::telemetry::metrics::metrics; use crate::utils::util::{get_message_key, get_payload_utf8, required_headers, CHRONOS_ID, DEADLINE}; use rdkafka::message::BorrowedMessage; use std::{collections::HashMap, str::FromStr, sync::Arc}; @@ -21,7 +23,7 @@ impl MessageReceiver { &self, new_message: &BorrowedMessage<'_>, reqd_headers: HashMap, - message_deadline: DateTime, + message_deadline: DateTime, ) -> Option { let max_retry_count = 3; let mut retry_count = 0; @@ -48,7 +50,6 @@ impl MessageReceiver { } tracing::Span::current().record("correlationId", &message_key); } - log::debug!("Message publish success {:?}", new_message); return None; } else { @@ -81,23 +82,76 @@ impl MessageReceiver { #[tracing::instrument(name = "receiver_handle_message", skip_all, fields(correlationId, error))] pub async fn handle_message(&self, message: &BorrowedMessage<'_>) { - let new_message = &message; - if let Some(reqd_headers) = required_headers(new_message) { - tracing::Span::current().record("correlationId", &reqd_headers[CHRONOS_ID]); - if let Ok(message_deadline) = DateTime::::from_str(&reqd_headers[DEADLINE]) { - if message_deadline <= Utc::now() { - if let Some(err) = self.prepare_and_publish(new_message, reqd_headers).await { - log::error!("{}", err); - tracing::Span::current().record("error", &err); + // Metrics + // start instant for safe time recordings w no error handling + let start_i = std::time::Instant::now(); + // We need the system TS to compare to the kafka timestamp + let start_ts = chrono::Local::now(); + // Declare but don't set, this helps enumerate all + // code paths for our recordings + let dest: metrics::ConsumedMessageDestinations; + let status: metrics::Status; + // Check for headers + match required_headers(message) { + Some(reqd_headers) => { + tracing::Span::current().record("correlationId", &reqd_headers[CHRONOS_ID]); + // Get the deadline header + let message_deadline = DateTime::::from_str(&reqd_headers[DEADLINE]); + match message_deadline { + Ok(message_deadline) => { + if message_deadline <= start_ts { + dest = metrics::ConsumedMessageDestinations::KAFKA; + match self.prepare_and_publish(message, reqd_headers).await { + Some(err) => { + log::error!("{}", err); + tracing::Span::current().record("error", &err); + status = metrics::Status::ERROR; + } + None => { + status = metrics::Status::SUCCESS; + } + } + } else { + dest = metrics::ConsumedMessageDestinations::DATABASE; + match self.insert_into_db(message, reqd_headers, message_deadline).await { + Some(err) => { + log::error!("{}", err); + tracing::Span::current().record("error", &err); + status = metrics::Status::ERROR; + } + None => { + status = metrics::Status::SUCCESS; + } + }; + } + } + Err(e) => { + // The user provided a bad time stamp + // If we see a TON of em, it could also indicate a bug in our + // time parsing or lots of messages with bad timestamps + log::warn!( + "message receiver: offset {} on partition {} caused time parser error {} ", + message.offset(), + message.partition(), + e + ); + (dest, status) = (metrics::ConsumedMessageDestinations::DROPPED, metrics::Status::SUCCESS); } - } else if let Some(err_string) = self.insert_into_db(new_message, reqd_headers, message_deadline).await { - log::error!("{}", err_string); - tracing::Span::current().record("error", &err_string); } } - } else { - log::warn!("message receiver: required headers not found"); + None => { + log::warn!( + "message receiver: required headers not found for offset {} on partition {}", + message.offset(), + message.partition(), + ); + (dest, status) = (metrics::ConsumedMessageDestinations::DROPPED, metrics::Status::SUCCESS); + // This is a success as the producer messed up, not us + } } + // We use an instant because no error handling + let dur = std::time::Instant::now().duration_since(start_i); + metrics::record_consumer_metrics(&start_ts, &dur, message, &dest, &status); } pub async fn run(&self) { @@ -112,9 +166,6 @@ impl MessageReceiver { log::error!("error while consuming message {:?}", e); } } - // if let Ok(message) = &self.consumer.kafka_consume_message().await { - // self.handle_message(message).await; - // } } } } diff --git a/chronos_bin/src/monitor.rs b/chronos_bin/src/monitor.rs index eccf329..6354bac 100644 --- a/chronos_bin/src/monitor.rs +++ b/chronos_bin/src/monitor.rs @@ -1,6 +1,6 @@ use crate::postgres::pg::Pg; use crate::utils::config::ChronosConfig; -use chrono::Utc; +use chrono::Local; use std::sync::Arc; use std::time::Duration; use tokio_postgres::Row; @@ -56,7 +56,7 @@ impl FailureDetector { async fn monitor_failed_fire_records(&self) { match &self .data_store - .failed_to_fire_db(&(Utc::now() - Duration::from_secs(ChronosConfig::from_env().fail_detect_interval))) + .failed_to_fire_db(&(Local::now() - Duration::from_secs(ChronosConfig::from_env().fail_detect_interval))) .await { Ok(fetched_rows) => { diff --git a/chronos_bin/src/persistence_store.rs b/chronos_bin/src/persistence_store.rs index 5a7588b..2ddd620 100644 --- a/chronos_bin/src/persistence_store.rs +++ b/chronos_bin/src/persistence_store.rs @@ -9,6 +9,6 @@ pub trait PersistenceStore { // async fn queuing_fetch(pg_client: &Client, deadline: String, limit: u16) -> Vec; async fn delete_fired(&self, ids: &String) -> u64; async fn ready_to_fire(&self, params: &Vec) -> Vec; - async fn failed_to_fire(&self, delay_time: DateTime) -> Vec; + async fn failed_to_fire(&self, delay_time: DateTime) -> Vec; async fn reset_to_init(&self, to_init_list: &Vec) -> Vec; } \ No newline at end of file diff --git a/chronos_bin/src/postgres/pg.rs b/chronos_bin/src/postgres/pg.rs index c444d1a..880ef17 100644 --- a/chronos_bin/src/postgres/pg.rs +++ b/chronos_bin/src/postgres/pg.rs @@ -1,4 +1,4 @@ -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Local}; use deadpool_postgres::{Config, GenericClient, ManagerConfig, Object, Pool, PoolConfig, Runtime, Transaction}; use log::error; use std::time::{Duration, Instant}; @@ -20,7 +20,7 @@ pub struct Pg { #[derive(Debug)] pub struct TableInsertColumns<'a> { pub id: &'a str, - pub deadline: DateTime, + pub deadline: DateTime, pub message_headers: serde_json::Value, pub message_key: &'a str, pub message_value: serde_json::Value, @@ -29,8 +29,8 @@ pub struct TableInsertColumns<'a> { #[derive(Debug)] pub struct TableRow<'a> { pub id: &'a str, - pub deadline: DateTime, - pub readied_at: DateTime, + pub deadline: DateTime, + pub readied_at: DateTime, pub readied_by: Uuid, pub message_headers: serde_json::Value, pub message_key: &'a str, @@ -40,16 +40,16 @@ pub struct TableRow<'a> { #[derive(Debug)] pub struct TableInsertRow<'a> { pub id: &'a str, - pub deadline: DateTime, + pub deadline: DateTime, pub message_headers: &'a serde_json::Value, pub message_key: &'a str, pub message_value: &'a serde_json::Value, } #[derive(Debug)] pub struct GetReady { - pub readied_at: DateTime, + pub readied_at: DateTime, pub readied_by: Uuid, - pub deadline: DateTime, + pub deadline: DateTime, // pub limit: i64, // pub order: &'a str, } @@ -307,7 +307,7 @@ impl Pg { } #[tracing::instrument(name = "failed_to_fire_db", skip_all)] - pub(crate) async fn failed_to_fire_db(&self, delay_time: &DateTime) -> Result, PgError> { + pub(crate) async fn failed_to_fire_db(&self, delay_time: &DateTime) -> Result, PgError> { let method_name = "failed_to_fire_db"; let query_execute_instant = Instant::now(); let pg_client = self.get_client().await?; diff --git a/chronos_bin/src/telemetry/metrics/metrics.rs b/chronos_bin/src/telemetry/metrics/metrics.rs new file mode 100644 index 0000000..520ab99 --- /dev/null +++ b/chronos_bin/src/telemetry/metrics/metrics.rs @@ -0,0 +1,128 @@ +use super::super::register_telemetry::DEFAULT_OTEL_SERVICE_NAME; +use chrono::Local; +use opentelemetry::{global, metrics::Histogram, KeyValue}; +use rdkafka::Message; +use std::sync::LazyLock; + +pub enum ConsumedMessageDestinations { + KAFKA, + DATABASE, + DROPPED, +} +pub enum Status { + SUCCESS, + ERROR, +} + +struct Metrics { + msg_consume_seconds: Histogram, + msg_consume_latency_seconds: Histogram, + // Will add fr in the next PR + // msg_publish_seconds: Histogram, + // msg_jitter_seconds: Histogram, + // msg_resets: Counter, +} + +impl Metrics { + fn new() -> Self { + let meter = global::meter(DEFAULT_OTEL_SERVICE_NAME); + Self { + // input consumption metrics + msg_consume_seconds: meter + .f64_histogram("msg.consume.service.time") + .with_description("Service time after receiving a message from the input queue") + // 5ms, 10ms, 25ms, 50ms, 100ms, 200ms, 500ms, 1s, 2s, 2.5s, 5s + .with_boundaries(vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 2.5, 5.0]) + .with_unit("s") + .build(), + msg_consume_latency_seconds: meter // Aka "lag time" + .f64_histogram("msg.consume.latency") + .with_description( + "Message latency on the input queue. Recorded from the start of the message handler function (does not include processing time)", + ) + .with_boundaries(vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 2.5, 5.0]) + .with_unit("s") + .build(), + // msg_publish_seconds: meter + // .f64_histogram("msg.process.service.time") + // .with_description("The service time of \"message ready\" processing loop. Only increments when messages are found") + // .with_unit("s") + // .build(), + // msg_jitter_seconds: meter + // .f64_histogram("msg.jitter") + // .with_unit("s") + // .with_description("The delta between the desired published time and confirmed published time of messages to the output topic") + // .build(), + // msg_resets: meter.u64_counter("msg.reset").with_description("The count of messages reset").build(), + } + } +} + +static METRICS: LazyLock = LazyLock::new(Metrics::new); + +/// Records the message consumption service time (time spent processing), in seconds +fn record_msg_consume(process_time: f64, destination: &str, status: &str) { + METRICS.msg_consume_seconds.record( + process_time, + &[ + KeyValue::new("destination", destination.to_string()), + KeyValue::new("status", status.to_string()), + ], + ); +} + +/// Records the message conume latency (time in queue), in seconds +fn record_msg_consume_latency(latency: f64, partition: i32) { + METRICS + .msg_consume_latency_seconds + .record(latency, &[KeyValue::new("partition", partition.to_string())]); +} + +/// Records consumer metrics +/// Should be run as the last step of message processing +pub fn record_consumer_metrics( + start: &chrono::DateTime, + duration: &std::time::Duration, + message: &rdkafka::message::BorrowedMessage<'_>, + destination: &ConsumedMessageDestinations, + status: &Status, +) { + let d = match destination { + ConsumedMessageDestinations::DATABASE => "database", + ConsumedMessageDestinations::KAFKA => "kafka", + ConsumedMessageDestinations::DROPPED => "dropped", + }; + let s = match status { + Status::ERROR => "error", + Status::SUCCESS => "success", + }; + record_msg_consume(duration.as_secs_f64(), d, s); + match message.timestamp().to_millis() { + Some(msg_ts) => { + // Time stamp millis returns unix epoch which is the same + // unit as the Kafka timestamp implementation. + // This is "safe" (won't panic), + // but we log here if we see a negative integer as it means that + // the kafka and system clock are out of sync + // We will discard the metric as it doesn't reflect the actual performance + // of the system. + let delta_sec = ((start.timestamp_millis() - (msg_ts) as i64) / 1000) as f64; + if delta_sec < 0 as f64 { + record_msg_consume_latency(delta_sec as f64, message.partition()); + } else { + log::error!( + "negative time delta found when comparing system UTC and kafka TS UTC for message {} on partition {}", + message.offset(), + message.partition() + ) + } + } + None => { + log::error!( + "metrics: no message timestamp for message {} on partition {}", + message.offset(), + message.partition() + ); + } + } +} diff --git a/chronos_bin/src/telemetry/metrics/mod.rs b/chronos_bin/src/telemetry/metrics/mod.rs index 0253970..c3aadb0 100644 --- a/chronos_bin/src/telemetry/metrics/mod.rs +++ b/chronos_bin/src/telemetry/metrics/mod.rs @@ -1 +1,2 @@ +pub mod metrics; pub mod prometheus_exporter; diff --git a/chronos_bin/src/telemetry/mod.rs b/chronos_bin/src/telemetry/mod.rs index 1f67268..f9445cf 100644 --- a/chronos_bin/src/telemetry/mod.rs +++ b/chronos_bin/src/telemetry/mod.rs @@ -1,3 +1,3 @@ -mod metrics; +pub mod metrics; pub mod register_telemetry; mod traces; diff --git a/chronos_bin/src/telemetry/register_telemetry.rs b/chronos_bin/src/telemetry/register_telemetry.rs index 5e3bfd9..9b7f43f 100644 --- a/chronos_bin/src/telemetry/register_telemetry.rs +++ b/chronos_bin/src/telemetry/register_telemetry.rs @@ -15,6 +15,8 @@ pub enum MetricsExporterType { NoOp, } +pub const DEFAULT_OTEL_SERVICE_NAME: &str = "chronos"; + pub struct TelemetryCollector { pub traces_collector_type: TracesExporterType, pub metrics_collector_type: MetricsExporterType, @@ -43,7 +45,7 @@ impl TelemetryCollector { MetricsExporterType::NoOp } }; - let service_name = std::env::var("OTEL_SERVICE_NAME").unwrap_or("chronos".to_string()); + let service_name = std::env::var("OTEL_SERVICE_NAME").unwrap_or(DEFAULT_OTEL_SERVICE_NAME.to_string()); TelemetryCollector { traces_collector_type, metrics_collector_type, diff --git a/docker-compose.yml b/docker-compose.yml index e502fa2..edcdde9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,7 +14,6 @@ services: timeout: 5s retries: 10 networks: [chronos] - chronos-pg-mig: build: @@ -66,6 +65,45 @@ services: KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER" KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:9093" KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1" + healthcheck: + test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 || exit 1"] + interval: 1s + timeout: 5s + retries: 60 + + kafka-bootstrap: + image: apache/kafka:4.1.1 + container_name: chronos-kafka-bootstrap + networks: [chronos] + restart: on-failure + depends_on: + kafka: + condition: service_healthy + command: |- + /opt/kafka/bin/kafka-topics.sh + --bootstrap-server kafka:9092 + --create + --if-not-exists + --config "message.timestamp.type=LogAppendTime" + --topic chronos\.in && + /opt/kafka/bin/kafka-topics.sh + --bootstrap-server kafka:9092 \ + --create + --if-not-exists + --config "message.timestamp.type=LogAppendTime" + --topic chronos\.out + + kafka-producer: + image: apache/kafka:4.1.1 + container_name: chronos-kafka-producer + networks: [chronos] + restart: always + depends_on: + kafka-bootstrap: + condition: service_completed_successfully + entrypoint: bash /tmp/console-producer.sh + volumes: + - ./scripts/console-producer.sh:/tmp/console-producer.sh kowl: image: quay.io/cloudhut/kowl:v1.5.0 @@ -114,8 +152,8 @@ services: depends_on: postgres: condition: service_healthy - kafka: - condition: service_started + kafka-bootstrap: + condition: service_completed_successfully otel-collector: condition: service_started ports: diff --git a/scripts/console-producer.sh b/scripts/console-producer.sh new file mode 100644 index 0000000..fa7d046 --- /dev/null +++ b/scripts/console-producer.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash + +# Produces a message every 5 seconds +# tested in image: +# apache/kafka:4.1.1 +while true +do + for i in "-5" "+10"; + do + echo "chronosMessageId:$(cat /proc/sys/kernel/random/uuid),chronosDeadline:$(date --date="$i seconds" --iso-8601=seconds) $(cat /proc/sys/kernel/random/uuid)::{\"msg\": \"I'm a msg!\"}" > /tmp/msg.$i ; + /opt/kafka/bin/kafka-console-producer.sh \ + --topic "chronos.in" \ + --property "parse.key=true" \ + --property "parse.headers=true" \ + --property "key.separator=::" \ + --property "headers.delimiter= " \ + --bootstrap-server "kafka:9092" < /tmp/msg.$i ; + done; + sleep 5; +done;