From 4cbe1e89a7434190a76c08fc8e3c072deed7fc3b Mon Sep 17 00:00:00 2001 From: Aidan Hall Date: Mon, 20 Jul 2026 20:02:50 +1000 Subject: [PATCH 1/7] chore: commit wip Need to flesh out the bugs/spec drift with people in the office. Implemented metrics are working, but could use a clean up. Added in some infra bits around kafka setup/data production. --- Dockerfile.chronos | 4 + chronos_bin/src/message_receiver.rs | 119 +++++++++++++++--- chronos_bin/src/telemetry/metrics/metrics.rs | 79 ++++++++++++ chronos_bin/src/telemetry/metrics/mod.rs | 1 + chronos_bin/src/telemetry/mod.rs | 2 +- .../src/telemetry/register_telemetry.rs | 4 +- docker-compose.yml | 44 ++++++- scripts/console-producer.sh | 20 +++ 8 files changed, 253 insertions(+), 20 deletions(-) create mode 100644 chronos_bin/src/telemetry/metrics/metrics.rs create mode 100644 scripts/console-producer.sh diff --git a/Dockerfile.chronos b/Dockerfile.chronos index b4b6515..35af5de 100644 --- a/Dockerfile.chronos +++ b/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 diff --git a/chronos_bin/src/message_receiver.rs b/chronos_bin/src/message_receiver.rs index 3a98cfe..ee78fc6 100644 --- a/chronos_bin/src/message_receiver.rs +++ b/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) -> Option { 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::::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::::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; - // } } } } diff --git a/chronos_bin/src/telemetry/metrics/metrics.rs b/chronos_bin/src/telemetry/metrics/metrics.rs new file mode 100644 index 0000000..54ec597 --- /dev/null +++ b/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, + msg_consume_latency_seconds: Histogram, + 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") + .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 = 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())]); +} 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..39f7de4 --- /dev/null +++ b/scripts/console-producer.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash + +# Produces a message every 5 minutes +# 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; From 59ae09619cdb7ade07f88224958bcd5afca92008 Mon Sep 17 00:00:00 2001 From: Aidan Hall Date: Mon, 20 Jul 2026 20:15:27 +1000 Subject: [PATCH 2/7] chore: typo --- scripts/console-producer.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/console-producer.sh b/scripts/console-producer.sh index 39f7de4..3e8af53 100644 --- a/scripts/console-producer.sh +++ b/scripts/console-producer.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash -# Produces a message every 5 minutes +# Produces a message every 5 seconds # tested in image: # apache/kafka:4.1.1 while true From b6b83ed3646553e58951a08cd183af4cfe2857dd Mon Sep 17 00:00:00 2001 From: Aidan Hall Date: Wed, 22 Jul 2026 10:33:24 +1000 Subject: [PATCH 3/7] chore: push up some refactoring and add some experimental histogram boundaries --- chronos_bin/src/message_receiver.rs | 71 +++++++++----------- chronos_bin/src/telemetry/metrics/metrics.rs | 62 +++++++++++++---- 2 files changed, 80 insertions(+), 53 deletions(-) diff --git a/chronos_bin/src/message_receiver.rs b/chronos_bin/src/message_receiver.rs index ee78fc6..e9ef3eb 100644 --- a/chronos_bin/src/message_receiver.rs +++ b/chronos_bin/src/message_receiver.rs @@ -8,9 +8,7 @@ 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 { @@ -61,7 +59,6 @@ impl MessageReceiver { } tracing::Span::current().record("correlationId", &message_key); } - log::debug!("Message publish success {:?}", new_message); return None; } else { @@ -98,13 +95,17 @@ 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(); + // 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 = std::time::SystemTime::now(); + // Declare but don't set, this helps enumerate all + // code paths for our recordings let dest: metrics::ConsumedMessageDestinations; let status: metrics::Status; - let new_message = &message; // Check for headers - match required_headers(new_message) { + match required_headers(message) { Some(reqd_headers) => { tracing::Span::current().record("correlationId", &reqd_headers[CHRONOS_ID]); // Get the deadline @@ -125,7 +126,7 @@ impl MessageReceiver { // 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 { + match self.prepare_and_publish(message, reqd_headers).await { Some(err) => { log::error!("{}", err); tracing::Span::current().record("error", &err); @@ -137,7 +138,7 @@ impl MessageReceiver { } } else { dest = metrics::ConsumedMessageDestinations::DATABASE; - match self.insert_into_db(new_message, reqd_headers, message_deadline).await { + match self.insert_into_db(message, reqd_headers, message_deadline).await { Some(err) => { log::error!("{}", err); tracing::Span::current().record("error", &err); @@ -150,46 +151,34 @@ impl MessageReceiver { } } Err(e) => { - log::warn!("message receiver: time parser error {e}"); + // The user provided a bad time stamp + // If we see a TON of em, it could also indicate a bug in our time parser + // Or lots of messages with bad TS's + log::warn!( + "message receiver: offset {} on partition {} caused time parser error {} ", + message.offset(), + message.partition(), + e + ); dest = metrics::ConsumedMessageDestinations::DROPPED; - status = metrics::Status::ERROR; + status = metrics::Status::SUCCESS; } } } 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() + log::warn!( + "message receiver: required headers not found for offset {} on partition {}", + message.offset(), + message.partition(), ); + dest = metrics::ConsumedMessageDestinations::DROPPED; + // This is a success as the producer messed up, not us + status = metrics::Status::SUCCESS; } } + // 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) { diff --git a/chronos_bin/src/telemetry/metrics/metrics.rs b/chronos_bin/src/telemetry/metrics/metrics.rs index 54ec597..d015f8b 100644 --- a/chronos_bin/src/telemetry/metrics/metrics.rs +++ b/chronos_bin/src/telemetry/metrics/metrics.rs @@ -4,8 +4,9 @@ use opentelemetry::{ metrics::{Counter, Histogram}, KeyValue, }; +use rdkafka::Message; use std::sync::LazyLock; - +use std::time::UNIX_EPOCH; struct Metrics { msg_consume_seconds: Histogram, msg_consume_latency_seconds: Histogram, @@ -26,7 +27,11 @@ impl Metrics { .build(), msg_consume_latency_seconds: meter // Aka "lag time" .f64_histogram("msg.consume.latency") - .with_description("Message latency on the input queue") + .with_description( + "Message latency on the input queue. Recorded from the start of the message handler function (does not include processing time)", + ) + // 10ms, 100ms, 200ms, 500ms, 750ms, 750ms + .with_boundaries(vec![0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 2.5, 5.0]) .with_unit("s") .build(), msg_publish_seconds: meter @@ -57,7 +62,29 @@ pub enum Status { ERROR, } -pub fn record_msg_consume(process_time: f64, destination: ConsumedMessageDestinations, status: Status) { +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()), + ], + ); +} + +fn record_msg_consume_latency(latency: f64, partition: i32) { + METRICS + .msg_consume_latency_seconds + .record(latency, &[KeyValue::new("partition", partition.to_string())]); +} + +pub fn record_consumer_metrics( + start: std::time::SystemTime, + duration: std::time::Duration, + message: &rdkafka::message::BorrowedMessage<'_>, + destination: ConsumedMessageDestinations, + status: Status, +) { let d = match destination { ConsumedMessageDestinations::DATABASE => "database", ConsumedMessageDestinations::KAFKA => "kafka", @@ -67,13 +94,24 @@ pub fn record_msg_consume(process_time: f64, destination: ConsumedMessageDestina 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())]); + // consumer function latency + record_msg_consume(duration.as_secs_f64(), d, s); + // Consumer lag time + // Requires error handling because we are using system time (time can go backwards!) + match message.timestamp().to_millis() { + Some(msg_ts) => match start.duration_since(UNIX_EPOCH) { + Ok(dur) => { + let delta_sec = dur.as_secs_f64() - (msg_ts / 1000) as f64; + record_msg_consume_latency(delta_sec as f64, message.partition()); + } + Err(e) => log::error!("metrics: system time error: {}", e), + }, + None => { + log::error!( + "metrics: no message timestamp for message {} on partition {}", + message.offset(), + message.partition() + ); + } + } } From 000fc91d15466cbfb831f3fc577010fadca3006c Mon Sep 17 00:00:00 2001 From: Aidan Hall Date: Mon, 17 Aug 2026 10:15:31 +1000 Subject: [PATCH 4/7] chore: prepare demo push --- chronos_bin/src/message_processor.rs | 2 - chronos_bin/src/message_receiver.rs | 37 ++--------- chronos_bin/src/telemetry/metrics/metrics.rs | 69 +++++++++----------- 3 files changed, 37 insertions(+), 71 deletions(-) diff --git a/chronos_bin/src/message_processor.rs b/chronos_bin/src/message_processor.rs index 4dac45a..32ff7c9 100644 --- a/chronos_bin/src/message_processor.rs +++ b/chronos_bin/src/message_processor.rs @@ -157,7 +157,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 +174,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 e9ef3eb..8eeef8e 100644 --- a/chronos_bin/src/message_receiver.rs +++ b/chronos_bin/src/message_receiver.rs @@ -30,15 +30,6 @@ 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 { @@ -74,10 +65,6 @@ impl MessageReceiver { async fn prepare_and_publish(&self, message: &BorrowedMessage<'_>, reqd_headers: HashMap) -> Option { 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); @@ -112,18 +99,6 @@ impl MessageReceiver { let message_deadline = DateTime::::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(message, reqd_headers).await { @@ -152,16 +127,15 @@ impl MessageReceiver { } 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 parser - // Or lots of messages with bad TS's + // 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 = metrics::ConsumedMessageDestinations::DROPPED; - status = metrics::Status::SUCCESS; + (dest, status) = (metrics::ConsumedMessageDestinations::DROPPED, metrics::Status::SUCCESS); } } } @@ -171,14 +145,13 @@ impl MessageReceiver { message.offset(), message.partition(), ); - dest = metrics::ConsumedMessageDestinations::DROPPED; + (dest, status) = (metrics::ConsumedMessageDestinations::DROPPED, metrics::Status::SUCCESS); // This is a success as the producer messed up, not us - status = metrics::Status::SUCCESS; } } // 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); + metrics::record_consumer_metrics(&start_ts, &dur, message, &dest, &status); } pub async fn run(&self) { diff --git a/chronos_bin/src/telemetry/metrics/metrics.rs b/chronos_bin/src/telemetry/metrics/metrics.rs index d015f8b..ecd0338 100644 --- a/chronos_bin/src/telemetry/metrics/metrics.rs +++ b/chronos_bin/src/telemetry/metrics/metrics.rs @@ -1,18 +1,26 @@ use super::super::register_telemetry::DEFAULT_OTEL_SERVICE_NAME; -use opentelemetry::{ - global, - metrics::{Counter, Histogram}, - KeyValue, -}; +use opentelemetry::{global, metrics::Histogram, KeyValue}; use rdkafka::Message; use std::sync::LazyLock; use std::time::UNIX_EPOCH; + +pub enum ConsumedMessageDestinations { + KAFKA, + DATABASE, + DROPPED, +} +pub enum Status { + SUCCESS, + ERROR, +} + struct Metrics { msg_consume_seconds: Histogram, msg_consume_latency_seconds: Histogram, - msg_publish_seconds: Histogram, - msg_jitter_seconds: Histogram, - msg_resets: Counter, + // Will add fr in the next PR + // msg_publish_seconds: Histogram, + // msg_jitter_seconds: Histogram, + // msg_resets: Counter, } impl Metrics { @@ -23,6 +31,7 @@ impl Metrics { msg_consume_seconds: meter .f64_histogram("msg.consume.service.time") .with_description("Service time after receiving a message from the input queue") + .with_boundaries(vec![0.01, 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" @@ -30,38 +39,26 @@ impl Metrics { .with_description( "Message latency on the input queue. Recorded from the start of the message handler function (does not include processing time)", ) - // 10ms, 100ms, 200ms, 500ms, 750ms, 750ms .with_boundaries(vec![0.01, 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(), + // 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); -pub enum ConsumedMessageDestinations { - KAFKA, - DATABASE, - DROPPED, -} - -pub enum Status { - SUCCESS, - ERROR, -} - fn record_msg_consume(process_time: f64, destination: &str, status: &str) { METRICS.msg_consume_seconds.record( process_time, @@ -79,11 +76,11 @@ fn record_msg_consume_latency(latency: f64, partition: i32) { } pub fn record_consumer_metrics( - start: std::time::SystemTime, - duration: std::time::Duration, + start: &std::time::SystemTime, + duration: &std::time::Duration, message: &rdkafka::message::BorrowedMessage<'_>, - destination: ConsumedMessageDestinations, - status: Status, + destination: &ConsumedMessageDestinations, + status: &Status, ) { let d = match destination { ConsumedMessageDestinations::DATABASE => "database", @@ -94,9 +91,7 @@ pub fn record_consumer_metrics( Status::ERROR => "error", Status::SUCCESS => "success", }; - // consumer function latency record_msg_consume(duration.as_secs_f64(), d, s); - // Consumer lag time // Requires error handling because we are using system time (time can go backwards!) match message.timestamp().to_millis() { Some(msg_ts) => match start.duration_since(UNIX_EPOCH) { From c203754ce37f1501c352ff6c4d715e74da08d3d8 Mon Sep 17 00:00:00 2001 From: Aidan Hall Date: Wed, 19 Aug 2026 13:08:31 +1000 Subject: [PATCH 5/7] fix: fix utc v local time spec bug + add consumer metrics The implementation was comparing all produced timestamps to UTC time. The spec says README.md#L136: A message requiring a delay is inserted into the database so that it can be published later. Before insertion, the queuing node compares the deadline to its local clock. The code was reading all timestamps as utc time. Also spent time optimizing the histogram bucket sizes to capture sub 100ms processing and queue latency times. --- Dockerfile.chronos | 23 +++++++++++++++--- chronos_bin/src/message_processor.rs | 6 ++--- chronos_bin/src/message_receiver.rs | 12 +++++----- chronos_bin/src/monitor.rs | 4 ++-- chronos_bin/src/persistence_store.rs | 2 +- chronos_bin/src/postgres/pg.rs | 16 ++++++------- chronos_bin/src/telemetry/metrics/metrics.rs | 25 ++++++++++---------- scripts/console-producer.sh | 2 +- 8 files changed, 53 insertions(+), 37 deletions(-) diff --git a/Dockerfile.chronos b/Dockerfile.chronos index 35af5de..3035329 100644 --- a/Dockerfile.chronos +++ b/Dockerfile.chronos @@ -17,9 +17,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 @@ -35,7 +52,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 32ff7c9..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, diff --git a/chronos_bin/src/message_receiver.rs b/chronos_bin/src/message_receiver.rs index 8eeef8e..3b03b1c 100644 --- a/chronos_bin/src/message_receiver.rs +++ b/chronos_bin/src/message_receiver.rs @@ -1,4 +1,4 @@ -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Local}; use rdkafka::Message; use serde_json::json; use tracing::instrument; @@ -23,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; @@ -86,7 +86,7 @@ impl MessageReceiver { // 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 = std::time::SystemTime::now(); + let start_ts = chrono::Local::now(); // Declare but don't set, this helps enumerate all // code paths for our recordings let dest: metrics::ConsumedMessageDestinations; @@ -95,11 +95,11 @@ impl MessageReceiver { match required_headers(message) { Some(reqd_headers) => { tracing::Span::current().record("correlationId", &reqd_headers[CHRONOS_ID]); - // Get the deadline - let message_deadline = DateTime::::from_str(&reqd_headers[DEADLINE]); + // Get the deadline header + let message_deadline = DateTime::::from_str(&reqd_headers[DEADLINE]); match message_deadline { Ok(message_deadline) => { - if message_deadline <= Utc::now() { + if message_deadline <= start_ts { dest = metrics::ConsumedMessageDestinations::KAFKA; match self.prepare_and_publish(message, reqd_headers).await { Some(err) => { 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 index ecd0338..c9e749e 100644 --- a/chronos_bin/src/telemetry/metrics/metrics.rs +++ b/chronos_bin/src/telemetry/metrics/metrics.rs @@ -1,8 +1,8 @@ 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; -use std::time::UNIX_EPOCH; pub enum ConsumedMessageDestinations { KAFKA, @@ -31,7 +31,8 @@ impl Metrics { msg_consume_seconds: meter .f64_histogram("msg.consume.service.time") .with_description("Service time after receiving a message from the input queue") - .with_boundaries(vec![0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 2.5, 5.0]) + // 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" @@ -39,7 +40,7 @@ impl Metrics { .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.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 2.5, 5.0]) + .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 @@ -59,6 +60,7 @@ impl Metrics { 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, @@ -69,14 +71,17 @@ fn record_msg_consume(process_time: f64, destination: &str, status: &str) { ); } +/// 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: &std::time::SystemTime, + start: &chrono::DateTime, duration: &std::time::Duration, message: &rdkafka::message::BorrowedMessage<'_>, destination: &ConsumedMessageDestinations, @@ -92,15 +97,11 @@ pub fn record_consumer_metrics( Status::SUCCESS => "success", }; record_msg_consume(duration.as_secs_f64(), d, s); - // Requires error handling because we are using system time (time can go backwards!) match message.timestamp().to_millis() { - Some(msg_ts) => match start.duration_since(UNIX_EPOCH) { - Ok(dur) => { - let delta_sec = dur.as_secs_f64() - (msg_ts / 1000) as f64; - record_msg_consume_latency(delta_sec as f64, message.partition()); - } - Err(e) => log::error!("metrics: system time error: {}", e), - }, + Some(msg_ts) => { + let delta_sec = ((start.timestamp_millis() - (msg_ts) as i64) / 1000) as f64; + record_msg_consume_latency(delta_sec as f64, message.partition()); + } None => { log::error!( "metrics: no message timestamp for message {} on partition {}", diff --git a/scripts/console-producer.sh b/scripts/console-producer.sh index 3e8af53..fa7d046 100644 --- a/scripts/console-producer.sh +++ b/scripts/console-producer.sh @@ -16,5 +16,5 @@ do --property "headers.delimiter= " \ --bootstrap-server "kafka:9092" < /tmp/msg.$i ; done; - sleep 10; + sleep 5; done; From 653e5cc28fb516f09a6617a60409a41dc491c376 Mon Sep 17 00:00:00 2001 From: Aidan Hall Date: Wed, 19 Aug 2026 13:27:03 +1000 Subject: [PATCH 6/7] chore: add log line if we see time go backwards when recording kafka ts and local time compare --- chronos_bin/src/telemetry/metrics/metrics.rs | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/chronos_bin/src/telemetry/metrics/metrics.rs b/chronos_bin/src/telemetry/metrics/metrics.rs index c9e749e..520ab99 100644 --- a/chronos_bin/src/telemetry/metrics/metrics.rs +++ b/chronos_bin/src/telemetry/metrics/metrics.rs @@ -99,8 +99,23 @@ pub fn record_consumer_metrics( 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; - record_msg_consume_latency(delta_sec as f64, message.partition()); + 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!( From ac33cc4065c9d537faaf5c13198f2839a8c461aa Mon Sep 17 00:00:00 2001 From: Aidan Hall Date: Wed, 19 Aug 2026 13:30:32 +1000 Subject: [PATCH 7/7] chore: clean pr for merge --- Dockerfile.chronos | 4 ---- 1 file changed, 4 deletions(-) diff --git a/Dockerfile.chronos b/Dockerfile.chronos index 3035329..b10b08b 100644 --- a/Dockerfile.chronos +++ b/Dockerfile.chronos @@ -1,8 +1,6 @@ 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 @@ -44,8 +42,6 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry/ \ 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