manta_server/server/audit/
kafka.rs

1//! Kafka destination for audit events.
2//!
3//! [`Kafka`] is the `[auditor.kafka]` block of `server.toml`;
4//! [`Kafka::build_sink`] turns it into a [`KafkaSink`] once, at
5//! startup, so a malformed librdkafka setting fails the boot instead
6//! of the first audited request.
7//!
8//! Emission never blocks a request: [`KafkaSink::emit`] only enqueues
9//! the message into librdkafka's in-memory queue and hands the
10//! delivery report to a spawned task, which logs a warning if the
11//! broker never acknowledges it within `message_timeout_ms`. A down
12//! broker therefore costs log lines, not latency. Messages still
13//! queued at shutdown get [`KafkaSink::flush`]ed within a bounded
14//! window.
15
16use std::{collections::BTreeMap, fmt, time::Duration};
17
18use manta_shared::common::error::MantaError;
19use rdkafka::{
20  ClientConfig,
21  producer::{FutureProducer, FutureRecord, Producer},
22};
23use serde::{Deserialize, Serialize};
24
25use super::{AuditSink, event::AuditEvent};
26
27/// Default librdkafka `message.timeout.ms`, used when
28/// `[auditor.kafka].message_timeout_ms` is absent.
29const DEFAULT_KAFKA_MESSAGE_TIMEOUT_MS: u32 = 5000;
30
31fn default_kafka_message_timeout_ms() -> u32 {
32  DEFAULT_KAFKA_MESSAGE_TIMEOUT_MS
33}
34
35/// librdkafka keys owned by the typed fields; setting them again in
36/// [`Kafka::properties`] is rejected rather than silently resolved.
37const RESERVED_PROPERTIES: [&str; 2] =
38  ["bootstrap.servers", "message.timeout.ms"];
39
40/// `[auditor.kafka]`: where audit events are produced to.
41#[derive(Serialize, Deserialize, Clone)]
42pub struct Kafka {
43  /// Bootstrap broker list, e.g. `["kafka.example.com:9092"]`.
44  pub brokers: Vec<String>,
45  /// Topic every audit event is produced to.
46  pub topic: String,
47  /// librdkafka `message.timeout.ms`: how long a queued event keeps
48  /// retrying delivery before it is dropped (and a warning logged).
49  #[serde(default = "default_kafka_message_timeout_ms")]
50  pub message_timeout_ms: u32,
51  /// Extra librdkafka client properties, passed through verbatim —
52  /// typically transport security: `security.protocol`,
53  /// `ssl.ca.location`, `sasl.mechanism`, `sasl.username`,
54  /// `sasl.password`, … Values are never printed (they may hold
55  /// credentials); only the keys are.
56  #[serde(default)]
57  pub properties: BTreeMap<String, String>,
58}
59
60impl fmt::Debug for Kafka {
61  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
62    f.debug_struct("Kafka")
63      .field("brokers", &self.brokers)
64      .field("topic", &self.topic)
65      .field("message_timeout_ms", &self.message_timeout_ms)
66      .field("properties", &self.properties.keys().collect::<Vec<_>>())
67      .finish()
68  }
69}
70
71impl Kafka {
72  /// Assemble the librdkafka client configuration.
73  ///
74  /// # Errors
75  ///
76  /// [`MantaError::KafkaError`] when `brokers` is empty or
77  /// `properties` sets a key owned by a typed field.
78  fn client_config(&self) -> Result<ClientConfig, MantaError> {
79    if self.brokers.is_empty() {
80      return Err(MantaError::KafkaError(
81        "[auditor.kafka].brokers must not be empty".to_string(),
82      ));
83    }
84    if let Some(key) = RESERVED_PROPERTIES
85      .iter()
86      .find(|k| self.properties.contains_key(**k))
87    {
88      return Err(MantaError::KafkaError(format!(
89        "[auditor.kafka].properties must not set `{key}`; use the \
90         `brokers` / `message_timeout_ms` fields instead"
91      )));
92    }
93    let mut config = ClientConfig::new();
94    for (key, value) in &self.properties {
95      config.set(key, value);
96    }
97    config
98      .set("bootstrap.servers", self.brokers.join(","))
99      .set("message.timeout.ms", self.message_timeout_ms.to_string());
100    Ok(config)
101  }
102
103  /// Build the producer. librdkafka connects lazily, so this does not
104  /// need a reachable broker — but it does validate every property.
105  ///
106  /// # Errors
107  ///
108  /// [`MantaError::KafkaError`] on an invalid configuration (see
109  /// `client_config`) or a property librdkafka rejects.
110  pub fn build_sink(&self) -> Result<KafkaSink, MantaError> {
111    let producer: FutureProducer =
112      self.client_config()?.create().map_err(|e| {
113        MantaError::KafkaError(format!("Failed to create Kafka producer: {e}"))
114      })?;
115    Ok(KafkaSink {
116      producer,
117      topic: self.topic.clone(),
118    })
119  }
120}
121
122/// A live Kafka producer for one topic.
123pub struct KafkaSink {
124  producer: FutureProducer,
125  topic: String,
126}
127
128impl AuditSink for KafkaSink {
129  fn emit(&self, event: &AuditEvent) {
130    let payload = match serde_json::to_vec(event) {
131      Ok(p) => p,
132      Err(e) => {
133        tracing::warn!(event_id = %event.event_id, "audit event not serialisable: {e}");
134        return;
135      }
136    };
137    let record = FutureRecord::to(&self.topic)
138      .key(event.event_id.as_str())
139      .payload(&payload);
140    match self.producer.send_result(record) {
141      Ok(delivery) => {
142        let event_id = event.event_id.clone();
143        tokio::spawn(async move {
144          match delivery.await {
145            Ok(Ok(_)) => {}
146            Ok(Err((e, _))) => tracing::warn!(
147              %event_id, "audit event not delivered to Kafka: {e}"
148            ),
149            Err(_) => tracing::warn!(
150              %event_id, "audit event delivery report lost (producer dropped)"
151            ),
152          }
153        });
154      }
155      Err((e, _)) => tracing::warn!(
156        event_id = %event.event_id, "audit event not queued for Kafka: {e}"
157      ),
158    }
159  }
160
161  fn flush(&self, timeout: Duration) {
162    if let Err(e) = self.producer.flush(timeout) {
163      tracing::warn!("audit events still queued for Kafka at shutdown: {e}");
164    }
165  }
166}
167
168#[cfg(test)]
169mod tests {
170  use super::*;
171
172  fn kafka(properties: &[(&str, &str)]) -> Kafka {
173    Kafka {
174      brokers: vec!["b1:9092".into(), "b2:9092".into()],
175      topic: "events.manta".into(),
176      message_timeout_ms: DEFAULT_KAFKA_MESSAGE_TIMEOUT_MS,
177      properties: properties
178        .iter()
179        .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
180        .collect(),
181    }
182  }
183
184  #[test]
185  fn deserialises_minimal_block_with_defaults() {
186    let k: Kafka =
187      toml::from_str("brokers = [\"b:9092\"]\ntopic = \"t\"").unwrap();
188    assert_eq!(k.message_timeout_ms, DEFAULT_KAFKA_MESSAGE_TIMEOUT_MS);
189    assert!(k.properties.is_empty());
190  }
191
192  #[test]
193  fn deserialises_properties_table() {
194    let k: Kafka = toml::from_str(
195      "brokers = [\"b:9092\"]\ntopic = \"t\"\n[properties]\n\
196       \"security.protocol\" = \"SASL_SSL\"",
197    )
198    .unwrap();
199    assert_eq!(k.properties["security.protocol"], "SASL_SSL");
200  }
201
202  #[test]
203  fn client_config_joins_brokers_and_passes_properties_through() {
204    let c = kafka(&[("security.protocol", "SSL")])
205      .client_config()
206      .unwrap();
207    assert_eq!(c.get("bootstrap.servers"), Some("b1:9092,b2:9092"));
208    assert_eq!(c.get("message.timeout.ms"), Some("5000"));
209    assert_eq!(c.get("security.protocol"), Some("SSL"));
210  }
211
212  #[test]
213  fn client_config_rejects_reserved_property() {
214    let err = kafka(&[("bootstrap.servers", "x:1")])
215      .client_config()
216      .unwrap_err();
217    assert!(err.to_string().contains("bootstrap.servers"), "{err}");
218  }
219
220  #[test]
221  fn client_config_rejects_empty_brokers() {
222    let mut k = kafka(&[]);
223    k.brokers.clear();
224    assert!(k.client_config().is_err());
225  }
226
227  #[test]
228  fn build_sink_rejects_unknown_librdkafka_property() {
229    // librdkafka validates property names at create(); no broker needed.
230    assert!(kafka(&[("no.such.property", "1")]).build_sink().is_err());
231  }
232
233  #[test]
234  fn build_sink_succeeds_without_a_reachable_broker() {
235    assert!(kafka(&[]).build_sink().is_ok());
236  }
237
238  #[test]
239  fn debug_hides_property_values() {
240    let s = format!("{:?}", kafka(&[("sasl.password", "hunter2")]));
241    assert!(s.contains("sasl.password"), "{s}");
242    assert!(!s.contains("hunter2"), "{s}");
243  }
244}
manta_server/server/audit/kafka.rs

manta_server/server/audit/
kafka.rs
