slog_cloud/
loggly.rs

1use std::time::Duration;
2
3use bytes::Bytes;
4use chrono::{SecondsFormat, Utc};
5use slog::{KV, Key, Level, OwnedKVList, Record, Serializer};
6
7use crate::{
8    batch::NDJSONBatchBuilder,
9    client::{Client, HttpClient, Method, SendError},
10    drain::{CloudDrain, CloudDrainBuilder, CloudDrainHandle, CloudDrainTask},
11    error::Error,
12    serializer::{AcceptAll, JsonMessageBuilder, KVFilter, LogMessageSerializer},
13};
14
15/// Builder for the Loggly log drain.
16pub struct LogglyDrainBuilder<F = AcceptAll> {
17    field_filter: F,
18    fallback_field: Key,
19    inner: CloudDrainBuilder,
20    request_timeout: Duration,
21    debug: bool,
22}
23
24impl LogglyDrainBuilder {
25    /// Create a new drain builder.
26    #[inline]
27    pub const fn new() -> Self {
28        Self {
29            field_filter: AcceptAll,
30            fallback_field: "",
31            inner: CloudDrainBuilder::new(),
32            request_timeout: Duration::from_secs(60),
33            debug: false,
34        }
35    }
36}
37
38impl Default for LogglyDrainBuilder {
39    #[inline]
40    fn default() -> Self {
41        Self::new()
42    }
43}
44
45impl<F> LogglyDrainBuilder<F> {
46    /// Set the log message queue capacity or `None` for unlimited (the default
47    /// is `None`).
48    #[inline]
49    pub fn queue_capacity(mut self, queue_capacity: Option<usize>) -> Self {
50        self.inner = self.inner.queue_capacity(queue_capacity);
51        self
52    }
53
54    /// Set the maximum number of concurrent message sends (the default is 16).
55    #[inline]
56    pub fn max_concurrency(mut self, max_concurrency: usize) -> Self {
57        self.inner = self.inner.max_concurrency(max_concurrency);
58        self
59    }
60
61    /// Set the request timeout (the default is 60 seconds).
62    #[inline]
63    pub fn request_timeout(mut self, timeout: Duration) -> Self {
64        self.request_timeout = timeout;
65        self
66    }
67
68    /// Set the initial retry delay (the default is 1 second).
69    #[inline]
70    pub fn initial_retry_delay(mut self, initial_retry_delay: Duration) -> Self {
71        self.inner = self.inner.initial_retry_delay(initial_retry_delay);
72        self
73    }
74
75    /// Set the maximum retry delay (the default is 60 seconds).
76    #[inline]
77    pub fn max_retry_delay(mut self, max_retry_delay: Duration) -> Self {
78        self.inner = self.inner.max_retry_delay(max_retry_delay);
79        self
80    }
81
82    /// Set the maximum number of retries or `None` for unlimited (the default
83    /// is unlimited).
84    #[inline]
85    pub fn max_retries(mut self, max_retries: Option<u32>) -> Self {
86        self.inner = self.inner.max_retries(max_retries);
87        self
88    }
89
90    /// Use a given key-value pair filter.
91    ///
92    /// All key-value pairs rejected by the filter will be serialized under a
93    /// given fallback field.
94    ///
95    /// This feature can be used if you want Loggly to index only a given
96    /// subset of fields.
97    pub fn kv_filter<K, T>(self, fallback_field: K, filter: T) -> LogglyDrainBuilder<T>
98    where
99        K: Into<Key>,
100    {
101        let mut fallback_field = fallback_field.into();
102
103        if fallback_field.is_empty() {
104            fallback_field = "misc";
105        }
106
107        LogglyDrainBuilder {
108            field_filter: filter,
109            fallback_field,
110            inner: self.inner,
111            request_timeout: self.request_timeout,
112            debug: self.debug,
113        }
114    }
115
116    /// Enable or disable debug mode (the default is disabled).
117    ///
118    /// In the debug mode you'll be able to see some runtime info on stderr
119    /// that will help you with setting up the drain (e.g. failed requests).
120    /// With debug mode disabled, all errors will be silently ignored.
121    #[inline]
122    pub fn debug_mode(mut self, enable: bool) -> Self {
123        self.inner = self.inner.debug(enable);
124        self.debug = enable;
125        self
126    }
127
128    /// Build the drain.
129    pub fn build(
130        self,
131        token: &str,
132        tag: &str,
133    ) -> Result<(LogglyDrain<F>, CloudDrainTask, LogglyDrainHandle), Error> {
134        let client = InternalClient::new(token, tag, self.request_timeout, self.debug)?;
135        let serializer = LogglySerializer::new(self.field_filter, self.fallback_field);
136        let batch_builder = NDJSONBatchBuilder::new(4_000_000);
137
138        let (drain, task, handle) = self.inner.build(client, serializer, batch_builder);
139
140        Ok((drain, task, handle))
141    }
142
143    /// Build the drain and spawn a tokio task responsible for sending log
144    /// messages.
145    pub fn spawn_task(
146        self,
147        token: &str,
148        tag: &str,
149    ) -> Result<(LogglyDrain<F>, LogglyDrainHandle), Error> {
150        let (drain, task, handle) = self.build(token, tag)?;
151
152        tokio::spawn(task);
153
154        Ok((drain, handle))
155    }
156
157    /// Build the drain and spawn a thread responsible for sending log
158    /// messages.
159    pub fn spawn_thread(
160        self,
161        token: &str,
162        tag: &str,
163    ) -> Result<(LogglyDrain<F>, LogglyDrainHandle), Error> {
164        let (drain, task, handle) = self.build(token, tag)?;
165
166        std::thread::spawn(move || {
167            let runtime = tokio::runtime::Builder::new_current_thread()
168                .enable_io()
169                .enable_time()
170                .build()
171                .expect("unable to create tokio runtime");
172
173            runtime.block_on(task)
174        });
175
176        Ok((drain, handle))
177    }
178}
179
180/// Loggly log drain.
181pub type LogglyDrain<F = AcceptAll> = CloudDrain<LogglySerializer<F>, Bytes>;
182
183/// Loggly drain handle.
184pub type LogglyDrainHandle = CloudDrainHandle<Bytes>;
185
186/// Loggly log message serializer.
187pub struct LogglySerializer<F = AcceptAll> {
188    field_filter: InternalKVFilter<F>,
189    fallback_field: Key,
190}
191
192impl<F> LogglySerializer<F> {
193    /// Create a new serializer with the given field filter and fallback field.
194    fn new(field_filter: F, fallback_field: Key) -> Self {
195        Self {
196            field_filter: InternalKVFilter::new(field_filter),
197            fallback_field,
198        }
199    }
200}
201
202impl<F> LogMessageSerializer for LogglySerializer<F>
203where
204    F: KVFilter,
205{
206    type Serialized = Bytes;
207
208    fn serialize(
209        &self,
210        record: &Record,
211        logger_values: &OwnedKVList,
212    ) -> slog::Result<Self::Serialized> {
213        let mut builder =
214            JsonMessageBuilder::new().with_field_filter(self.fallback_field, &self.field_filter);
215
216        let level = match record.level() {
217            Level::Critical => "critical",
218            Level::Error => "error",
219            Level::Warning => "warn",
220            Level::Info => "info",
221            Level::Debug => "debug",
222            Level::Trace => "trace",
223        };
224
225        let file = record.file();
226        let line = record.line();
227
228        builder.emit_str("level", level)?;
229        builder.emit_arguments("file", &format_args!("{}:{}", file, line))?;
230        builder.emit_arguments("message", record.msg())?;
231
232        let timestamp = Utc::now();
233
234        builder.emit_str(
235            "timestamp",
236            &timestamp.to_rfc3339_opts(SecondsFormat::Micros, true),
237        )?;
238
239        record.kv().serialize(record, &mut builder)?;
240
241        logger_values.serialize(record, &mut builder)?;
242
243        builder.finish()
244    }
245}
246
247/// Internal key-value filter that accepts required fields.
248struct InternalKVFilter<F> {
249    inner: F,
250}
251
252impl<F> InternalKVFilter<F> {
253    /// Create a new internal key-value filter.
254    fn new(inner: F) -> Self {
255        Self { inner }
256    }
257}
258
259impl<F> KVFilter for InternalKVFilter<F>
260where
261    F: KVFilter,
262{
263    fn is_accepted(&self, key: &Key) -> bool {
264        matches!(*key, "level" | "file" | "message" | "timestamp") || self.inner.is_accepted(key)
265    }
266}
267
268/// Internal client for sending log messages to Loggly.
269struct InternalClient {
270    inner: HttpClient,
271    debug: bool,
272}
273
274impl InternalClient {
275    /// Create a new client.
276    fn new(token: &str, tag: &str, request_timeout: Duration, debug: bool) -> Result<Self, Error> {
277        let url = format!("https://logs-01.loggly.com/bulk/{token}/tag/{tag}/")
278            .parse()
279            .map_err(|err| {
280                Error::from_static_msg_and_cause("unable to construct Loggly URL", err)
281            })?;
282
283        let inner = HttpClient::builder()
284            .method(Method::POST)
285            .header("Content-Type", "text/plain")?
286            .request_timeout(request_timeout)
287            .build(url)?;
288
289        let res = Self { inner, debug };
290
291        Ok(res)
292    }
293}
294
295impl Client for InternalClient {
296    type Message = Bytes;
297
298    async fn send(&self, msg: Self::Message) -> Result<(), SendError> {
299        self.inner.send(msg).await.map_err(|err| {
300            if self.debug {
301                eprintln!("Loggly request failed: {err:#}");
302            }
303
304            err.into()
305        })
306    }
307}
308
309#[cfg(test)]
310mod tests {
311    use chrono::{DateTime, Utc};
312    use slog::{Key, info, o, warn};
313
314    use crate::serializer::{AcceptAll, test_utils::serialize_logs};
315
316    use super::LogglySerializer;
317
318    #[test]
319    fn serializes_log_record() {
320        let serializer = LogglySerializer::new(AcceptAll, "misc");
321
322        let logs = serialize_logs(serializer, |log| {
323            warn!(log, "hello {}", "world"; "key" => "value");
324        });
325
326        let msg = &logs[0];
327
328        assert_eq!(msg["level"], "warn");
329        assert_eq!(msg["message"], "hello world");
330        assert_eq!(msg["key"], "value");
331
332        let file = msg["file"].as_str().unwrap();
333
334        assert!(file.starts_with(concat!(file!(), ":")), "{file}");
335
336        let timestamp: DateTime<Utc> = msg["timestamp"].as_str().unwrap().parse().unwrap();
337
338        assert!((Utc::now() - timestamp).num_seconds().abs() < 60);
339    }
340
341    #[test]
342    fn later_key_values_take_precedence() {
343        let serializer = LogglySerializer::new(AcceptAll, "misc");
344
345        let logs = serialize_logs(serializer, |log| {
346            let parent = log.new(o!("a" => "parent", "b" => "parent"));
347            let child = parent.new(o!("a" => "child"));
348
349            info!(child, "msg"; "b" => "record", "c" => "first", "c" => "second");
350        });
351
352        let msg = &logs[0];
353
354        assert_eq!(msg["a"], "child");
355        assert_eq!(msg["b"], "record");
356        assert_eq!(msg["c"], "second");
357    }
358
359    #[test]
360    fn reserved_fields_cannot_be_overridden() {
361        let serializer = LogglySerializer::new(AcceptAll, "misc");
362
363        let logs = serialize_logs(serializer, |log| {
364            let log = log.new(o!("level" => "fake"));
365
366            info!(log, "msg"; "message" => "fake", "file" => "fake", "timestamp" => "fake");
367        });
368
369        let msg = &logs[0];
370
371        assert_eq!(msg["level"], "info");
372        assert_eq!(msg["message"], "msg");
373        assert_ne!(msg["file"], "fake");
374        assert_ne!(msg["timestamp"], "fake");
375    }
376
377    #[test]
378    fn kv_filter_does_not_apply_to_reserved_fields() {
379        let serializer = LogglySerializer::new(|_: &Key| false, "extra");
380
381        let logs = serialize_logs(serializer, |log| {
382            info!(log, "msg"; "key" => "value");
383        });
384
385        let msg = &logs[0];
386
387        assert_eq!(msg["level"], "info");
388        assert_eq!(msg["message"], "msg");
389        assert!(msg["file"].is_string());
390        assert!(msg["timestamp"].is_string());
391        assert!(msg.get("key").is_none());
392        assert_eq!(msg["extra"], "key: value");
393    }
394}
slog_cloud/loggly.rs

slog_cloud/
loggly.rs
