manta_server/server/common/
kafka.rs

1//! Lazily-initialised Kafka producer used by the audit subsystem.
2//!
3//! The producer is a fire-and-forget `FutureProducer` cached behind
4//! a `OnceLock`, so the first audit message pays the connection cost
5//! and subsequent messages reuse the same client. Delivery uses a
6//! zero-duration wait — the audit path never blocks the request
7//! that triggered it.
8
9use std::{fmt, sync::OnceLock, time::Duration};
10
11use manta_shared::common::error::MantaError;
12use rdkafka::{
13  ClientConfig,
14  producer::{FutureProducer, FutureRecord},
15};
16use serde::{Deserialize, Serialize};
17
18use crate::server::common::audit::Audit;
19
20/// Default Kafka message delivery timeout (milliseconds), used when
21/// `server.toml`'s `[auditor.kafka].message_timeout_ms` is absent.
22const DEFAULT_KAFKA_MESSAGE_TIMEOUT_MS: u32 = 5000;
23
24/// Default Kafka delivery-confirmation wait (seconds), used when
25/// `server.toml`'s `[auditor.kafka].delivery_wait_secs` is absent.
26/// Zero means fire-and-forget.
27const DEFAULT_KAFKA_DELIVERY_WAIT_SECS: u64 = 0;
28
29fn default_kafka_message_timeout_ms() -> u32 {
30  DEFAULT_KAFKA_MESSAGE_TIMEOUT_MS
31}
32fn default_kafka_delivery_wait_secs() -> u64 {
33  DEFAULT_KAFKA_DELIVERY_WAIT_SECS
34}
35
36/// Kafka client configuration for audit message production.
37///
38/// The [`FutureProducer`] is lazily created on the first
39/// call to [`Audit::produce_message`] and reused for all
40/// subsequent calls via an internal [`OnceLock`].
41#[derive(Serialize, Deserialize)]
42pub struct Kafka {
43  /// Bootstrap broker list, e.g. `vec!["kafka.example.com:9092"]`.
44  pub brokers: Vec<String>,
45  /// Kafka topic that audit messages are published to.
46  pub topic: String,
47  /// librdkafka `message.timeout.ms`: how long a queued audit
48  /// message tries to deliver before being dropped.
49  #[serde(default = "default_kafka_message_timeout_ms")]
50  pub message_timeout_ms: u32,
51  /// How long `produce_message` blocks waiting for delivery
52  /// confirmation. Zero (default) is fire-and-forget.
53  #[serde(default = "default_kafka_delivery_wait_secs")]
54  pub delivery_wait_secs: u64,
55  #[serde(skip)]
56  producer: OnceLock<FutureProducer>,
57}
58
59impl fmt::Debug for Kafka {
60  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
61    f.debug_struct("Kafka")
62      .field("brokers", &self.brokers)
63      .field("topic", &self.topic)
64      .field(
65        "producer",
66        &if self.producer.get().is_some() {
67          "Some(<FutureProducer>)"
68        } else {
69          "None"
70        },
71      )
72      .finish()
73  }
74}
75
76impl Clone for Kafka {
77  /// Clone the configuration only; the cached producer is
78  /// not cloned and will be lazily recreated.
79  fn clone(&self) -> Self {
80    Self {
81      brokers: self.brokers.clone(),
82      topic: self.topic.clone(),
83      message_timeout_ms: self.message_timeout_ms,
84      delivery_wait_secs: self.delivery_wait_secs,
85      producer: OnceLock::new(),
86    }
87  }
88}
89
90impl Kafka {
91  /// Create a new `Kafka` instance with the given broker list and
92  /// topic name. Uses the crate-internal `message.timeout.ms` and
93  /// delivery-wait defaults (5000ms and fire-and-forget,
94  /// respectively); for non-default values, deserialize a `Kafka`
95  /// from `server.toml`'s `[auditor.kafka]` block instead.
96  ///
97  /// The actual `FutureProducer` is built lazily on the first
98  /// [`Audit::produce_message`] call, so this constructor is cheap
99  /// and infallible.
100  pub fn new(brokers: Vec<String>, topic: String) -> Self {
101    Self {
102      brokers,
103      topic,
104      message_timeout_ms: DEFAULT_KAFKA_MESSAGE_TIMEOUT_MS,
105      delivery_wait_secs: DEFAULT_KAFKA_DELIVERY_WAIT_SECS,
106      producer: OnceLock::new(),
107    }
108  }
109
110  /// Return the cached [`FutureProducer`], creating it on
111  /// first call.
112  fn get_or_init_producer(&self) -> Result<&FutureProducer, MantaError> {
113    if let Some(p) = self.producer.get() {
114      return Ok(p);
115    }
116    let brokers = self.brokers.join(",");
117    let p: FutureProducer = ClientConfig::new()
118      .set("bootstrap.servers", &brokers)
119      .set("message.timeout.ms", self.message_timeout_ms.to_string())
120      .create()
121      .map_err(|e| {
122        MantaError::KafkaError(format!("Failed to create Kafka producer: {e}"))
123      })?;
124    // Another thread may have raced us; either value is
125    // fine since they are configured identically.
126    Ok(self.producer.get_or_init(|| p))
127  }
128}
129
130impl Audit for Kafka {
131  async fn produce_message(&self, data: &[u8]) -> Result<(), MantaError> {
132    let producer = self.get_or_init_producer()?;
133
134    let delivery_status = producer
135      .send::<Vec<u8>, _, _>(
136        FutureRecord::to(&self.topic).payload(data),
137        Duration::from_secs(self.delivery_wait_secs),
138      )
139      .await;
140
141    match delivery_status {
142      Ok(_) => {
143        tracing::info!("Delivery status for message received");
144      }
145      Err(e) => {
146        return Err(MantaError::KafkaError(format!(
147          "Delivery status for message failed: {:?}",
148          e.0
149        )));
150      }
151    }
152
153    Ok(())
154  }
155}
156
157#[cfg(test)]
158mod tests {
159  //! Tests for the non-IO parts of [`Kafka`]: configuration plumbing,
160  //! clone semantics, and the redacted Debug representation.
161  //! `produce_message` and `get_or_init_producer` require a broker
162  //! (or a librdkafka mock) and are exercised via integration tests.
163
164  use super::*;
165
166  #[test]
167  fn new_round_trips_brokers_and_topic() {
168    let k = Kafka::new(
169      vec!["broker1:9092".into(), "broker2:9092".into()],
170      "audit-events".into(),
171    );
172    assert_eq!(k.brokers, vec!["broker1:9092", "broker2:9092"]);
173    assert_eq!(k.topic, "audit-events");
174    assert!(
175      k.producer.get().is_none(),
176      "producer must be uninitialised on construction (lazy init)"
177    );
178  }
179
180  #[test]
181  fn clone_resets_the_producer_cache() {
182    // The `Clone` impl deliberately drops the cached producer —
183    // otherwise two `Kafka` values would share rdkafka state in a
184    // way the rdkafka APIs don't sanction. A future "fix" that
185    // shares the OnceLock would break the lazy-init contract; this
186    // test makes that change deliberate.
187    let original = Kafka::new(vec!["b:9092".into()], "t".into());
188    let cloned = original.clone();
189    assert_eq!(cloned.brokers, original.brokers);
190    assert_eq!(cloned.topic, original.topic);
191    assert!(
192      cloned.producer.get().is_none(),
193      "cloned producer cache must be empty regardless of source state"
194    );
195  }
196
197  #[test]
198  fn debug_masks_the_producer_and_shows_init_state() {
199    // The Debug impl deliberately substitutes a placeholder string
200    // for the FutureProducer — librdkafka internals would otherwise
201    // appear in log lines if a Kafka value is debug-printed. Pin
202    // both the placeholder string and that brokers/topic remain
203    // visible (they're not secret).
204    let uninit = Kafka::new(vec!["b:9092".into()], "audit".into());
205    let s = format!("{uninit:?}");
206    assert!(s.contains("brokers"), "brokers field must be visible");
207    assert!(s.contains("\"b:9092\""), "broker value must be visible");
208    assert!(s.contains("audit"), "topic must be visible");
209    assert!(
210      s.contains("None"),
211      "uninitialised producer must show as `None`, got: {s}"
212    );
213    // The literal placeholder string used for an initialised producer
214    // is pinned indirectly: if `Some(<FutureProducer>)` ever leaks
215    // through to Debug output of an uninit Kafka, this assertion
216    // catches it.
217    assert!(
218      !s.contains("FutureProducer"),
219      "uninitialised Kafka must not mention FutureProducer in Debug output"
220    );
221  }
222}
manta_server/server/common/kafka.rs

manta_server/server/common/
kafka.rs
