sozu_command_lib/logging/access_logs.rs
1use std::{
2 collections::BTreeMap, fmt, fmt::Formatter, mem::ManuallyDrop, net::SocketAddr, time::Duration,
3};
4
5use rusty_ulid::Ulid;
6
7use crate::{
8 logging::{LogLevel, Rfc3339Time},
9 proto::command::{
10 HttpEndpoint, OpenTelemetry as ProtobufOpenTelemetry, ProtobufAccessLog, ProtobufEndpoint,
11 TcpEndpoint, protobuf_endpoint,
12 },
13};
14
15/// This uses unsafe to creates a "fake" owner of the underlying data.
16/// Beware that for the compiler it is as legitimate as the original owner.
17/// So you have to elide one of them (with std::mem::forget or ManuallyDrop)
18/// before it is drop to avoid a double free.
19///
20/// This trait works on &T and Option<&T> types
21///
22/// After performance review, it seems not any more efficient than calling `clone()`,
23/// probably because the cache of malloc is so well optimized these days.
24trait DuplicateOwnership {
25 type Target;
26 /// Don't forget to use std::mem::forget or ManuallyDrop over one of your owners
27 unsafe fn duplicate(self) -> Self::Target;
28}
29
30impl<T> DuplicateOwnership for &T {
31 type Target = T;
32 unsafe fn duplicate(self) -> T {
33 // SAFETY: `std::ptr::read` duplicates the bit-pattern of the
34 // pointee without copying. The result aliases the original value's
35 // ownership view, so the caller MUST ensure exactly one of the two
36 // owners is wrapped in `ManuallyDrop` or `mem::forget`'d before
37 // either is dropped (cf. the type-level guidance at line 26).
38 unsafe { std::ptr::read(self as *const T) }
39 }
40}
41impl<'a, T> DuplicateOwnership for Option<&'a T>
42where
43 T: ?Sized,
44 &'a T: DuplicateOwnership,
45{
46 type Target = Option<<&'a T as DuplicateOwnership>::Target>;
47 unsafe fn duplicate(self) -> Self::Target {
48 // SAFETY: delegates to the inner `&T::duplicate` impl above. The
49 // same double-free obligation propagates to the caller (cf. the
50 // type-level guidance at line 26).
51 unsafe { self.map(|t| t.duplicate()) }
52 }
53}
54impl DuplicateOwnership for &str {
55 type Target = String;
56 unsafe fn duplicate(self) -> Self::Target {
57 // SAFETY: `String::from_raw_parts` reuses the underlying allocation
58 // owned by `self` without copying, so the result aliases that
59 // ownership view. Caller MUST ensure exactly one owner is wrapped
60 // in `ManuallyDrop` or `mem::forget`'d to avoid double-free
61 // (cf. the type-level guidance at line 26). The bytes are valid
62 // UTF-8 because they came from a `&str`.
63 unsafe { String::from_raw_parts(self.as_ptr() as *mut _, self.len(), self.len()) }
64 }
65}
66impl<T> DuplicateOwnership for &[T] {
67 type Target = Vec<T>;
68 unsafe fn duplicate(self) -> Self::Target {
69 // SAFETY: `Vec::from_raw_parts` reuses the underlying allocation
70 // owned by `self` without copying, so the result aliases that
71 // ownership view. Caller MUST ensure exactly one owner is wrapped
72 // in `ManuallyDrop` or `mem::forget`'d to avoid double-free
73 // (cf. the type-level guidance at line 26).
74 unsafe { Vec::from_raw_parts(self.as_ptr() as *mut _, self.len(), self.len()) }
75 }
76}
77
78pub struct LogMessage<'a>(pub Option<&'a str>);
79pub struct LogDuration(pub Option<Duration>);
80/// Renders a socket address, or `-` when there is none, straight into the
81/// formatter: an access-log line must not build a `String` per address.
82pub struct LogAddress(pub Option<SocketAddr>);
83
84/// Prefix block attached to every log line. Rendered as
85/// `[<session_id> <request_id_or_-> <cluster_id_or_-> <backend_id_or_->]`
86/// so operators can grep a full TCP/TLS session (`session_id`) or drill
87/// into one HTTP exchange (`request_id`).
88///
89/// - `session_id` is generated once per mio-accepted socket and survives
90/// protocol upgrades (expect-proxy, TLS handshake, H1↔H2 multiplexing).
91/// - `request_id` is per-request: distinct for each H2 stream, and
92/// regenerated on each H1 keep-alive exchange.
93#[derive(Debug)]
94pub struct LogContext<'a> {
95 pub session_id: Ulid,
96 pub request_id: Option<Ulid>,
97 pub cluster_id: Option<&'a str>,
98 pub backend_id: Option<&'a str>,
99}
100
101pub enum EndpointRecord<'a> {
102 Http {
103 method: Option<&'a str>,
104 authority: Option<&'a str>,
105 path: Option<&'a str>,
106 status: Option<u16>,
107 reason: Option<&'a str>,
108 },
109 Tcp,
110}
111
112/// used to aggregate tags in a session
113#[derive(Debug, Clone, PartialEq, Eq)]
114pub struct CachedTags {
115 pub tags: BTreeMap<String, String>,
116 pub concatenated: String,
117}
118
119impl CachedTags {
120 pub fn new(tags: BTreeMap<String, String>) -> Self {
121 let concatenated = tags
122 .iter()
123 .map(|(k, v)| format!("{k}={v}"))
124 .collect::<Vec<_>>()
125 .join(", ");
126 Self { tags, concatenated }
127 }
128}
129
130#[derive(Debug)]
131pub struct FullTags<'a> {
132 pub concatenated: Option<&'a str>,
133 pub user_agent: Option<&'a str>,
134}
135
136#[derive(Default)]
137pub struct OpenTelemetry {
138 pub trace_id: [u8; 32],
139 pub span_id: [u8; 16],
140 pub parent_span_id: Option<[u8; 16]>,
141}
142
143impl fmt::Debug for OpenTelemetry {
144 fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
145 // SAFETY: `trace_id` is populated either by `parse_hex` (which
146 // verifies every byte is `is_ascii_hexdigit`) or by `random_id`
147 // (which fills from `b"0123456789abcdef"`). Both produce ASCII
148 // bytes, which are by construction valid single-byte UTF-8.
149 let trace_id = unsafe { std::str::from_utf8_unchecked(&self.trace_id) };
150 // SAFETY: same provenance as `trace_id` above — `parse_hex` or
151 // `random_id` from a hex charset, ASCII guaranteed.
152 let span_id = unsafe { std::str::from_utf8_unchecked(&self.span_id) };
153 let parent_span_id = self
154 .parent_span_id
155 .as_ref()
156 // SAFETY: same provenance as `trace_id` above — `parse_hex`
157 // from a hex charset, ASCII guaranteed.
158 .map(|id| unsafe { std::str::from_utf8_unchecked(id) })
159 .unwrap_or("-");
160 write!(f, "{trace_id} {span_id} {parent_span_id}")
161 }
162}
163
164/// Intermediate representation of an access log agnostic of the final format.
165/// Every field is a reference to avoid capturing ownership (as a logger should).
166pub struct RequestRecord<'a> {
167 pub message: Option<&'a str>,
168 pub context: LogContext<'a>,
169 pub session_address: Option<SocketAddr>,
170 pub backend_address: Option<SocketAddr>,
171 pub protocol: &'a str,
172 pub endpoint: EndpointRecord<'a>,
173 pub tags: Option<&'a CachedTags>,
174 pub client_rtt: Option<Duration>,
175 pub server_rtt: Option<Duration>,
176 pub user_agent: Option<&'a str>,
177 /// Value of the `x-request-id` header forwarded to the backend. Preserved
178 /// verbatim when the client supplied one; otherwise derived from the
179 /// request ULID (`context.request_id`). Used by downstream observability
180 /// pipelines as a universal correlation key.
181 pub x_request_id: Option<&'a str>,
182 /// Negotiated TLS protocol version short-form (e.g. `"TLSv1.3"`).
183 /// Static-string borrow plumbed straight from rustls. `None` for
184 /// plaintext listeners or when the version label is unknown.
185 pub tls_version: Option<&'static str>,
186 /// Negotiated TLS cipher suite short-form (e.g.
187 /// `"TLS_AES_128_GCM_SHA256"`). Static-string borrow plumbed straight
188 /// from rustls. `None` for plaintext listeners or when the cipher
189 /// label is unknown.
190 pub tls_cipher: Option<&'static str>,
191 /// TLS Server Name Indication sent by the client at handshake (already
192 /// pre-lowercased, no port). `None` for plaintext listeners or when the
193 /// client omitted the SNI extension.
194 pub tls_sni: Option<&'a str>,
195 /// Negotiated ALPN protocol (e.g. `"h2"`, `"http/1.1"`). `None` for
196 /// plaintext listeners or when no ALPN was negotiated.
197 pub tls_alpn: Option<&'static str>,
198 /// Verbatim value of the client-supplied `X-Forwarded-For` header as
199 /// observed before Sōzu appended its own hop. `None` if the request had
200 /// no `X-Forwarded-For` header.
201 pub xff_chain: Option<&'a str>,
202 pub service_time: Duration,
203 /// time from connecting to the backend until the end of the response
204 pub response_time: Option<Duration>,
205 /// time between first byte of the request and last byte of the response
206 pub request_time: Duration,
207 /// Wall-clock start time as nanoseconds since the Unix epoch. Captured
208 /// immediately alongside the monotonic `start` instant (via
209 /// `SessionMetrics::mark_request_start`), so consumers can use it
210 /// directly as the span start time instead of computing
211 /// `precise_time - request_time` (which mixes CLOCK_REALTIME and
212 /// CLOCK_MONOTONIC).
213 pub start_time_ns: Option<i128>,
214 pub bytes_in: usize,
215 pub bytes_out: usize,
216 pub otel: Option<&'a OpenTelemetry>,
217
218 // added by the logger itself
219 pub pid: i32,
220 pub tag: &'a str,
221 pub level: LogLevel,
222 pub now: Rfc3339Time,
223 pub precise_time: i128,
224}
225
226impl RequestRecord<'_> {
227 pub fn full_tags(&self) -> FullTags<'_> {
228 FullTags {
229 concatenated: self.tags.as_ref().map(|t| t.concatenated.as_str()),
230 user_agent: self.user_agent,
231 }
232 }
233
234 /// Converts the RequestRecord in its protobuf representation.
235 /// Prost needs ownership over all the fields but we don't want to take it from the user
236 /// or clone them, so we use the unsafe DuplicateOwnership.
237 pub fn into_binary_access_log(self) -> ManuallyDrop<ProtobufAccessLog> {
238 // SAFETY: Each `.duplicate()` call below borrows ownership without
239 // copying. The whole protobuf is wrapped in `ManuallyDrop` so its
240 // destructor never runs — the original `RequestRecord` references
241 // remain the sole owners. `std::str::from_utf8_unchecked` on `otel`
242 // fields is sound because `trace_id` / `span_id` are ASCII hex
243 // (see the `OpenTelemetry` Debug impl above for the same proof).
244 unsafe {
245 let endpoint = match self.endpoint {
246 EndpointRecord::Http {
247 method,
248 authority,
249 path,
250 status,
251 reason,
252 } => protobuf_endpoint::Inner::Http(HttpEndpoint {
253 method: method.duplicate(),
254 authority: authority.duplicate(),
255 path: path.duplicate(),
256 status: status.map(|s| s as u32),
257 reason: reason.duplicate(),
258 }),
259 EndpointRecord::Tcp => protobuf_endpoint::Inner::Tcp(TcpEndpoint {}),
260 };
261
262 ManuallyDrop::new(ProtobufAccessLog {
263 backend_address: self.backend_address.map(Into::into),
264 backend_id: self.context.backend_id.duplicate(),
265 bytes_in: self.bytes_in as u64,
266 bytes_out: self.bytes_out as u64,
267 client_rtt: self.client_rtt.map(|t| t.as_micros() as u64),
268 cluster_id: self.context.cluster_id.duplicate(),
269 endpoint: ProtobufEndpoint {
270 inner: Some(endpoint),
271 },
272 message: self.message.duplicate(),
273 protocol: self.protocol.duplicate(),
274 request_id: self
275 .context
276 .request_id
277 .unwrap_or(self.context.session_id)
278 .into(),
279 session_id: Some(self.context.session_id.into()),
280 response_time: self.response_time.map(|t| t.as_micros() as u64),
281 server_rtt: self.server_rtt.map(|t| t.as_micros() as u64),
282 service_time: self.service_time.as_micros() as u64,
283 session_address: self.session_address.map(Into::into),
284 tags: self
285 .tags
286 .map(|tags| tags.tags.duplicate())
287 .unwrap_or_default(),
288 user_agent: self.user_agent.duplicate(),
289 x_request_id: self.x_request_id.duplicate(),
290 tls_version: self.tls_version.duplicate(),
291 tls_cipher: self.tls_cipher.duplicate(),
292 tls_sni: self.tls_sni.duplicate(),
293 tls_alpn: self.tls_alpn.duplicate(),
294 xff_chain: self.xff_chain.duplicate(),
295 tag: self.tag.duplicate(),
296 time: self.precise_time.into(),
297 request_time: Some(self.request_time.as_micros() as u64),
298 otel: self.otel.map(|otel| ProtobufOpenTelemetry {
299 trace_id: std::str::from_utf8_unchecked(&otel.trace_id).duplicate(),
300 span_id: std::str::from_utf8_unchecked(&otel.span_id).duplicate(),
301 parent_span_id: otel
302 .parent_span_id
303 .as_ref()
304 .map(|id| std::str::from_utf8_unchecked(id).duplicate()),
305 }),
306 start_time: self.start_time_ns.map(Into::into),
307 })
308 }
309 }
310}