tracing_betterstack/
client.rs1use serde::Serialize;
2use std::future::Future;
3use std::pin::Pin;
4
5use crate::{dispatch::LogEvent, export::LogDestination};
6
7#[derive(Debug)]
8pub enum BetterstackError {
9 HttpError(reqwest::Error),
10 InvalidConfig(String),
11}
12
13impl std::fmt::Display for BetterstackError {
14 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
15 match self {
16 BetterstackError::HttpError(e) => write!(f, "HTTP error: {}", e),
17 BetterstackError::InvalidConfig(msg) => write!(f, "Invalid configuration: {}", msg),
18 }
19 }
20}
21
22impl std::error::Error for BetterstackError {
23 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
24 match self {
25 BetterstackError::HttpError(e) => Some(e),
26 BetterstackError::InvalidConfig(_) => None,
27 }
28 }
29}
30
31impl From<reqwest::Error> for BetterstackError {
32 fn from(err: reqwest::Error) -> Self {
33 BetterstackError::HttpError(err)
34 }
35}
36
37pub trait BetterstackClientTrait: Send + Sync {
38 fn put_logs<'a>(
39 &'a self,
40 dest: LogDestination,
41 logs: Vec<LogEvent>,
42 ) -> Pin<Box<dyn Future<Output = Result<(), BetterstackError>> + Send + 'a>>;
43}
44
45#[derive(Debug, Clone)]
46pub struct BetterstackClient {
47 http_client: reqwest::Client,
48 source_token: String,
49 ingestion_url: String,
50}
51
52impl BetterstackClient {
53 pub fn new(source_token: impl Into<String>, ingestion_url: impl Into<String>) -> Self {
54 Self {
55 http_client: reqwest::Client::new(),
56 source_token: source_token.into(),
57 ingestion_url: ingestion_url.into(),
58 }
59 }
60
61 pub fn with_client(
62 http_client: reqwest::Client,
63 source_token: impl Into<String>,
64 ingestion_url: impl Into<String>,
65 ) -> Self {
66 Self {
67 http_client,
68 source_token: source_token.into(),
69 ingestion_url: ingestion_url.into(),
70 }
71 }
72}
73
74#[derive(Serialize)]
75struct BetterstackEvent {
76 message: String,
77 dt: i64,
78 #[serde(skip_serializing_if = "Option::is_none")]
79 level: Option<String>,
80 #[serde(skip_serializing_if = "Option::is_none")]
81 target: Option<String>,
82 #[serde(skip_serializing_if = "Option::is_none")]
83 thread_id: Option<String>,
84 #[serde(skip_serializing_if = "Option::is_none")]
85 file: Option<String>,
86 #[serde(skip_serializing_if = "Option::is_none")]
87 line: Option<u32>,
88}
89
90impl From<LogEvent> for BetterstackEvent {
91 fn from(event: LogEvent) -> Self {
92 Self {
93 message: event.message,
94 dt: event.timestamp.to_utc().timestamp_millis(),
95 level: event.level,
96 target: event.target,
97 thread_id: event.thread_id,
98 file: event.file,
99 line: event.line,
100 }
101 }
102}
103
104impl BetterstackClientTrait for BetterstackClient {
105 fn put_logs<'a>(
106 &'a self,
107 _: LogDestination,
108 logs: Vec<LogEvent>,
109 ) -> Pin<Box<dyn Future<Output = Result<(), BetterstackError>> + Send + 'a>> {
110 Box::pin(async move {
111 if self.source_token.is_empty() {
112 return Err(BetterstackError::InvalidConfig(
113 "Source token cannot be empty".into(),
114 ));
115 }
116
117 let events: Vec<BetterstackEvent> = logs.into_iter().map(Into::into).collect();
118 let body = serde_json::to_string(&events).map_err(|e| {
119 BetterstackError::InvalidConfig(format!("Failed to serialize events: {}", e))
120 })?;
121
122 self.http_client
123 .post(&self.ingestion_url)
124 .header("Authorization", format!("Bearer {}", self.source_token))
125 .header("Content-Type", "application/json")
126 .body(body)
127 .send()
128 .await?
129 .error_for_status()?;
130
131 Ok(())
132 })
133 }
134}
135
136pub struct NoopBetterstackClient;
137
138impl BetterstackClientTrait for NoopBetterstackClient {
139 fn put_logs<'a>(
140 &'a self,
141 _: LogDestination,
142 _: Vec<LogEvent>,
143 ) -> Pin<Box<dyn Future<Output = Result<(), BetterstackError>> + Send + 'a>> {
144 Box::pin(async { Ok(()) })
145 }
146}
147
148impl NoopBetterstackClient {
149 pub fn new() -> Self {
150 Self
151 }
152}
153
154#[cfg(test)]
155mod tests {
156 use super::*;
157 use chrono::Utc;
158
159 #[tokio::test]
160 async fn test_client_empty_token() {
161 let client = BetterstackClient::new("", "");
162 let logs = vec![LogEvent::new("test".into())];
163
164 let result = client.put_logs(LogDestination, logs).await;
165 assert!(matches!(result, Err(BetterstackError::InvalidConfig(_))));
166 }
167
168 #[tokio::test]
169 async fn test_noop_client() {
170 let client = NoopBetterstackClient::new();
171 let logs = vec![LogEvent::new("test".into())];
172
173 let result = client.put_logs(LogDestination, logs).await;
174 assert!(result.is_ok());
175 }
176
177 #[test]
178 fn test_event_serialization() {
179 let now = Utc::now();
180 let event = LogEvent {
181 message: "test message".into(),
182 timestamp: now,
183 level: Some("INFO".into()),
184 target: Some("test_target".into()),
185 thread_id: Some("ThreadId(1)".into()),
186 file: Some("test.rs".into()),
187 line: Some(42),
188 };
189
190 let betterstack_event: BetterstackEvent = event.into();
191 assert_eq!(betterstack_event.message, "test message");
192 assert_eq!(betterstack_event.dt, now.timestamp_millis());
193 assert_eq!(betterstack_event.level, Some("INFO".into()));
194 assert_eq!(betterstack_event.target, Some("test_target".into()));
195 assert_eq!(betterstack_event.thread_id, Some("ThreadId(1)".into()));
196 assert_eq!(betterstack_event.file, Some("test.rs".into()));
197 assert_eq!(betterstack_event.line, Some(42));
198 }
199
200 #[test]
201 fn test_client_new() {
202 let client = BetterstackClient::new("token", "url");
203 assert_eq!(client.source_token, "token");
204 assert_eq!(client.ingestion_url, "url");
205 }
206
207 #[test]
208 fn test_client_with_custom_http_client() {
209 let http_client = reqwest::Client::new();
210 let client = BetterstackClient::with_client(http_client, "token", "url");
211 assert_eq!(client.source_token, "token");
212 assert_eq!(client.ingestion_url, "url");
213 }
214}