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

Dockerfile.chronos

@@ -1,6 +1,8 @@
 ARG RUST_VERSION=1.96.0
 FROM rust:${RUST_VERSION}-bookworm AS build
 # Install software
+ENV CARGO_HTTP_CAINFO=/etc/ssl/certs/zscaler_cert.pem
+COPY ./zscaler_cert.pem /etc/ssl/certs/zscaler_cert.pem
 RUN update-ca-certificates && apt-get update && apt-get install -y libsasl2-dev
 # Create appuser
 ENV USER=chronos
@@ -25,6 +27,8 @@ RUN cargo build -p chronos_bin --release
 FROM debian:bookworm-slim AS run
 # SASL supports
 ENV USER=chronos
+ENV CARGO_HTTP_CAINFO=/etc/ssl/certs/zscaler_cert.pem
+COPY ./zscaler_cert.pem /etc/ssl/certs/zscaler_cert.pem
 RUN apt-get update && apt-get install -y ca-certificates && update-ca-certificates && apt-get install -y libsasl2-dev
 WORKDIR /opt/build
 # Import users from build

chronos_bin/src/message_receiver.rs

@@ -1,12 +1,16 @@
 use chrono::{DateTime, Utc};
+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::time::UNIX_EPOCH;
 use std::{collections::HashMap, str::FromStr, sync::Arc};
 
 pub struct MessageReceiver {
@@ -28,6 +32,15 @@ impl MessageReceiver {
         //retry loop
         loop {
             if let Some(payload) = get_payload_utf8(new_message) {
+                // This is a bug.
+                // The README says:
+                //
+                // The `message_value` field will almost always be JSON in practice
+                // but Chronos doesn't attempt to parse its contents — it simply forwards it on.
+                // The value may be quite large — beyond the `varchar` limit — hence the use of a `blob`.
+                //
+                // This attempts to serialize the message body to JSON.
+                // This makes Chronos incompatible with any non-json schemas
                 if let Ok(message_value) = &serde_json::from_slice(payload) {
                     if let Some(message_key) = get_message_key(new_message) {
                         let params = TableInsertRow {
@@ -64,6 +77,10 @@ impl MessageReceiver {
     async fn prepare_and_publish(&self, message: &BorrowedMessage<'_>, reqd_headers: HashMap<String, String>) -> Option<String> {
         match get_payload_utf8(message) {
             Some(string_payload) => {
+                // This check smells fishy
+                // Nothing in the README requires that
+                // the message has a key.
+                // It only says we need the chronosMessageId and chronosDeadline.
                 if let Some(message_key) = get_message_key(message) {
                     let string_payload = String::from_utf8_lossy(string_payload).to_string();
                     tracing::Span::current().record("correlationId", &message_key);
@@ -81,22 +98,97 @@ impl MessageReceiver {
 
     #[tracing::instrument(name = "receiver_handle_message", skip_all, fields(correlationId, error))]
     pub async fn handle_message(&self, message: &BorrowedMessage<'_>) {
+        // Must be system time
+        let start = std::time::SystemTime::now();
+        let dest: metrics::ConsumedMessageDestinations;
+        let status: metrics::Status;
         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);
+        // Check for headers
+        match required_headers(new_message) {
+            Some(reqd_headers) => {
+                tracing::Span::current().record("correlationId", &reqd_headers[CHRONOS_ID]);
+                // Get the deadline
+                let message_deadline = DateTime::<Utc>::from_str(&reqd_headers[DEADLINE]);
+                match message_deadline {
+                    Ok(message_deadline) => {
+                        // I think this should also include the timing advance
+                        // dl<=Utc::now()+timing
+                        // In the worst case, a message will be delayed poll-1ns
+                        // Lets use the numbers in the docs to
+                        // In the README, the poll interval is set to 100ms, and the
+                        // timing advance is set to 50ms.
+                        // msg_dl_check: 23:59:59.99
+                        // message_deadline: 00:00:00.00
+                        // db_check: 00:00:00.00
+                        // msg stored: 00:00:00.01
+                        // msg_published: 00:00:00.10
+                        // We should have sent the message before storing
+                        if message_deadline <= Utc::now() {
+                            dest = metrics::ConsumedMessageDestinations::KAFKA;
+                            match self.prepare_and_publish(new_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(new_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;
+                                }
+                            };
+                        }
                     }
-                } 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);
+                    Err(e) => {
+                        log::warn!("message receiver: time parser error {e}");
+                        dest = metrics::ConsumedMessageDestinations::DROPPED;
+                        status = metrics::Status::ERROR;
+                    }
+                }
+            }
+            None => {
+                log::warn!("message receiver: required headers not found");
+                dest = metrics::ConsumedMessageDestinations::DROPPED;
+                status = metrics::Status::ERROR;
+            }
+        }
+        let end = std::time::SystemTime::now();
+        // Get the duration between
+        let delta = end.duration_since(start);
+        match delta {
+            Ok(o) => {
+                metrics::record_msg_consume(o.as_secs_f64(), dest, status);
+            }
+            Err(e) => {
+                log::error!("system time error: {e}");
+            }
+        }
+        let msg_ts = new_message.timestamp();
+        match msg_ts.to_millis() {
+            Some(msg_ts) => match end.duration_since(UNIX_EPOCH) {
+                Ok(end) => {
+                    let delta_sec = end.as_secs_f64() - (msg_ts / 1000) as f64;
+                    metrics::record_msg_consume_latency(delta_sec as f64, new_message.partition());
                 }
+                Err(e) => log::error!("system time error: {}", e),
+            },
+            None => {
+                log::error!(
+                    "no message timestamp for message {} on partition {}",
+                    new_message.offset(),
+                    new_message.partition()
+                );
             }
-        } else {
-            log::warn!("message receiver: required headers not found");
         }
     }
 
@@ -112,9 +204,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/telemetry/metrics/metrics.rs

@@ -0,0 +1,79 @@
+use super::super::register_telemetry::DEFAULT_OTEL_SERVICE_NAME;
+use opentelemetry::{
+    global,
+    metrics::{Counter, Histogram},
+    KeyValue,
+};
+use std::sync::LazyLock;
+
+struct Metrics {
+    msg_consume_seconds: Histogram<f64>,
+    msg_consume_latency_seconds: Histogram<f64>,
+    msg_publish_seconds: Histogram<f64>,
+    msg_jitter_seconds: Histogram<f64>,
+    msg_resets: Counter<u64>,
+}
+
+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")
+                .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")
+                .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<Metrics> = LazyLock::new(Metrics::new);
+
+pub enum ConsumedMessageDestinations {
+    KAFKA,
+    DATABASE,
+    DROPPED,
+}
+
+pub enum Status {
+    SUCCESS,
+    ERROR,
+}
+
+pub fn record_msg_consume(process_time: f64, 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",
+    };
+    METRICS
+        .msg_consume_seconds
+        .record(process_time, &[KeyValue::new("destination", d), KeyValue::new("status", s)]);
+}
+
+pub fn record_msg_consume_latency(latency: f64, partition: i32) {
+    METRICS
+        .msg_consume_latency_seconds
+        .record(latency, &[KeyValue::new("partition", partition.to_string())]);
+}

chronos_bin/src/telemetry/metrics/mod.rs

@@ -1 +1,2 @@
+pub mod metrics;
 pub mod prometheus_exporter;

chronos_bin/src/telemetry/mod.rs

@@ -1,3 +1,3 @@
-mod metrics;
+pub mod metrics;
 pub mod register_telemetry;
 mod traces;

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,

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:

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 10;
+done;