feat: add metric implementation and address some suspected logic bugs.

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

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;
         }
     }

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<String, String>,
-        message_deadline: DateTime<Utc>,
+        message_deadline: DateTime<Local>,
     ) -> Option<String> {
         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::<Utc>::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::<Local>::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;
-            // }
         }
     }
 }

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) => {

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<TableRow>;
     async fn delete_fired(&self, ids: &String) -> u64;
     async fn ready_to_fire(&self, params: &Vec<GetReady>) -> Vec<Row>;
-    async fn failed_to_fire(&self, delay_time: DateTime<Utc>) -> Vec<Row>;
+    async fn failed_to_fire(&self, delay_time: DateTime<Local>) -> Vec<Row>;
     async fn reset_to_init(&self,  to_init_list: &Vec<Row>) -> Vec<String>;
 }

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<Utc>,
+    pub deadline: DateTime<Local>,
     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<Utc>,
-    pub readied_at: DateTime<Utc>,
+    pub deadline: DateTime<Local>,
+    pub readied_at: DateTime<Local>,
     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<Utc>,
+    pub deadline: DateTime<Local>,
     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<Utc>,
+    pub readied_at: DateTime<Local>,
     pub readied_by: Uuid,
-    pub deadline: DateTime<Utc>,
+    pub deadline: DateTime<Local>,
     // 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<Utc>) -> Result<Vec<Row>, PgError> {
+    pub(crate) async fn failed_to_fire_db(&self, delay_time: &DateTime<Local>) -> Result<Vec<Row>, PgError> {
         let method_name = "failed_to_fire_db";
         let query_execute_instant = Instant::now();
         let pg_client = self.get_client().await?;