fast_telemetry_export/
otlp.rs

1//! OTLP HTTP/protobuf metrics exporter.
2//!
3//! Exports metrics to an OTLP-compatible collector (OpenTelemetry Collector,
4//! Grafana Alloy, Datadog OTLP intake, etc.) via HTTP POST of protobuf-encoded
5//! `ExportMetricsServiceRequest` to `/v1/metrics`.
6//!
7//! Uses cumulative temporality — no state tracking needed between export cycles.
8//! Larger payloads are gzip-compressed automatically, and failed exports retry
9//! with exponential backoff.
10//!
11//! The actual metric collection is provided by the caller via a closure,
12//! making this exporter work with any metrics struct.
13
14use std::time::Duration;
15
16use fast_telemetry::otlp::{build_export_request, build_resource, pb};
17use prost::Message;
18use tokio::time::{MissedTickBehavior, interval};
19use tokio_util::sync::CancellationToken;
20
21/// Configuration for the OTLP HTTP metrics exporter.
22#[derive(Clone)]
23pub struct OtlpConfig {
24    /// OTLP collector endpoint (scheme + host + port), e.g. `"http://localhost:4318"`.
25    /// The path `/v1/metrics` is appended automatically.
26    pub endpoint: String,
27    /// Export interval (default: 60s).
28    pub interval: Duration,
29    /// `service.name` resource attribute.
30    pub service_name: String,
31    /// Instrumentation scope name (default: "fast-telemetry").
32    pub scope_name: String,
33    /// Additional resource attributes (e.g. `("service.version", "1.0")`).
34    pub resource_attributes: Vec<(String, String)>,
35    /// Request timeout (default: 10s).
36    pub timeout: Duration,
37    /// Extra HTTP headers sent with every export request.
38    ///
39    /// Use this for collector authentication, e.g.:
40    /// - `("Authorization", "Bearer <token>")`
41    /// - `("x-api-key", "<key>")`
42    pub headers: Vec<(String, String)>,
43}
44
45impl Default for OtlpConfig {
46    fn default() -> Self {
47        Self {
48            endpoint: "http://localhost:4318".to_string(),
49            interval: Duration::from_secs(60),
50            service_name: "unknown_service".to_string(),
51            scope_name: "fast-telemetry".to_string(),
52            resource_attributes: Vec::new(),
53            timeout: Duration::from_secs(10),
54            headers: Vec::new(),
55        }
56    }
57}
58
59impl OtlpConfig {
60    pub fn new(endpoint: impl Into<String>) -> Self {
61        Self {
62            endpoint: endpoint.into(),
63            ..Default::default()
64        }
65    }
66
67    pub fn with_interval(mut self, interval: Duration) -> Self {
68        self.interval = interval;
69        self
70    }
71
72    pub fn with_service_name(mut self, name: impl Into<String>) -> Self {
73        self.service_name = name.into();
74        self
75    }
76
77    pub fn with_scope_name(mut self, name: impl Into<String>) -> Self {
78        self.scope_name = name.into();
79        self
80    }
81
82    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
83        self.resource_attributes.push((key.into(), value.into()));
84        self
85    }
86
87    pub fn with_timeout(mut self, timeout: Duration) -> Self {
88        self.timeout = timeout;
89        self
90    }
91
92    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
93        self.headers.push((name.into(), value.into()));
94        self
95    }
96}
97
98/// Maximum backoff delay between retries after export failures.
99const MAX_BACKOFF: Duration = Duration::from_secs(300);
100
101/// Base backoff delay after the first failure.
102const BASE_BACKOFF: Duration = Duration::from_secs(5);
103
104/// Minimum payload size (bytes) before gzip compression is applied.
105/// Below this threshold, compression overhead exceeds savings.
106const GZIP_THRESHOLD: usize = 1024;
107
108/// Gzip-compress `data` into `out` using fast compression (level 1).
109///
110/// Returns `true` if compression was applied, `false` if the payload was below
111/// the threshold (in which case `out` is untouched).
112fn gzip_compress(data: &[u8], out: &mut Vec<u8>) -> bool {
113    if data.len() < GZIP_THRESHOLD {
114        return false;
115    }
116    use flate2::Compression;
117    use flate2::write::GzEncoder;
118    use std::io::Write;
119
120    out.clear();
121    let mut encoder = GzEncoder::new(out, Compression::fast());
122    let _ = encoder.write_all(data);
123    let _ = encoder.finish();
124    true
125}
126
127/// Send a protobuf-encoded body, applying gzip when beneficial.
128async fn send_otlp(
129    client: &reqwest::Client,
130    url: &str,
131    body: &[u8],
132    gzip_buf: &mut Vec<u8>,
133    extra_headers: &[(String, String)],
134) -> Result<reqwest::Response, reqwest::Error> {
135    let mut req = client
136        .post(url)
137        .header("Content-Type", "application/x-protobuf");
138
139    for (name, value) in extra_headers {
140        req = req.header(name, value);
141    }
142
143    if gzip_compress(body, gzip_buf) {
144        req.header("Content-Encoding", "gzip")
145            .body(gzip_buf.clone())
146            .send()
147            .await
148    } else {
149        req.body(body.to_vec()).send().await
150    }
151}
152
153/// Run the OTLP metrics export loop.
154///
155/// `collect_fn` is called each cycle with a `&mut Vec<pb::Metric>`. The closure
156/// should append OTLP metric messages (typically via `ExportMetrics::export_otlp`).
157/// The exporter handles protobuf encoding, gzip compression, HTTP transport, and
158/// exponential backoff on failures.
159///
160/// On cancellation, a final export is performed to flush pending metrics.
161///
162/// # Example
163///
164/// ```ignore
165/// use std::sync::Arc;
166/// use std::time::Duration;
167///
168/// use fast_telemetry_export::otlp::{OtlpConfig, run};
169/// use tokio_util::sync::CancellationToken;
170///
171/// let metrics = Arc::new(MyMetrics::new());
172/// let cancel = CancellationToken::new();
173/// let config = OtlpConfig::new("http://otel-collector:4318")
174///     .with_service_name("myapp")
175///     .with_scope_name("proxy")
176///     .with_attribute("service.version", "1.0")
177///     .with_header("Authorization", "Bearer <token>")
178///     .with_timeout(Duration::from_secs(5));
179///
180/// let m = metrics.clone();
181/// tokio::spawn(run(config, cancel, move |out| {
182///     m.export_otlp(out);
183/// }));
184/// ```
185pub async fn run<F>(config: OtlpConfig, cancel: CancellationToken, mut collect_fn: F)
186where
187    F: FnMut(&mut Vec<pb::Metric>),
188{
189    let url = format!("{}/v1/metrics", config.endpoint.trim_end_matches('/'));
190
191    log::info!(
192        "Starting OTLP metrics exporter, endpoint={url}, interval={}s",
193        config.interval.as_secs()
194    );
195
196    let attr_refs: Vec<(&str, &str)> = config
197        .resource_attributes
198        .iter()
199        .map(|(k, v)| (k.as_str(), v.as_str()))
200        .collect();
201    let resource = build_resource(&config.service_name, &attr_refs);
202
203    let client = match reqwest::Client::builder().timeout(config.timeout).build() {
204        Ok(c) => c,
205        Err(e) => {
206            log::error!("Failed to build HTTP client for OTLP exporter: {e}");
207            return;
208        }
209    };
210
211    let mut interval = interval(config.interval);
212    interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
213    interval.tick().await;
214
215    let mut consecutive_failures: u32 = 0;
216    let mut bufs = ExportBufs::default();
217
218    let ctx = ExportContext {
219        client: &client,
220        url: &url,
221        resource: &resource,
222        scope_name: &config.scope_name,
223        extra_headers: &config.headers,
224    };
225
226    loop {
227        tokio::select! {
228            _ = interval.tick() => {}
229            _ = cancel.cancelled() => {
230                log::info!("OTLP metrics exporter shutting down, performing final export");
231                export_once(&ctx, &mut collect_fn, &mut bufs).await;
232                return;
233            }
234        }
235
236        if consecutive_failures > 0 {
237            let backoff = backoff_with_jitter(consecutive_failures);
238            log::debug!(
239                "OTLP export backing off {}ms (failures={consecutive_failures})",
240                backoff.as_millis()
241            );
242            tokio::select! {
243                _ = tokio::time::sleep(backoff) => {}
244                _ = cancel.cancelled() => {
245                    export_once(&ctx, &mut collect_fn, &mut bufs).await;
246                    return;
247                }
248            }
249        }
250
251        let mut metric_messages = Vec::new();
252        collect_fn(&mut metric_messages);
253
254        if metric_messages.is_empty() {
255            continue;
256        }
257
258        let metric_count = metric_messages.len();
259        let request = build_export_request(&resource, &config.scope_name, metric_messages);
260
261        bufs.encode.clear();
262        if let Err(e) = request.encode(&mut bufs.encode) {
263            log::warn!("OTLP protobuf encode failed: {e}");
264            continue;
265        }
266        let body_len = bufs.encode.len();
267
268        match send_otlp(&client, &url, &bufs.encode, &mut bufs.gzip, &config.headers).await {
269            Ok(resp) if resp.status().is_success() => {
270                consecutive_failures = 0;
271                log::debug!("Exported {metric_count} OTLP metrics ({body_len} bytes)");
272            }
273            Ok(resp) => {
274                consecutive_failures = consecutive_failures.saturating_add(1);
275                let status = resp.status();
276                let body = resp.text().await.unwrap_or_default();
277                log::warn!("OTLP export failed: status={status}, body={body}");
278            }
279            Err(e) => {
280                consecutive_failures = consecutive_failures.saturating_add(1);
281                log::warn!("OTLP export request failed: {e}");
282            }
283        }
284    }
285}
286
287/// Run the OTLP metrics export loop on a monoio runtime.
288///
289/// This is a monoio-native alternative to [`run`] for applications that do not
290/// want their metrics exporter to run on Tokio. It sends OTLP HTTP/protobuf over
291/// a fresh monoio TCP connection per export attempt.
292///
293/// Currently this supports plaintext `http://` collector endpoints only. Keep
294/// using [`run`] for `https://` endpoints until a monoio TLS transport is added.
295/// The caller must run this inside a monoio runtime with timers enabled.
296#[cfg(feature = "monoio")]
297pub async fn run_monoio<F>(config: OtlpConfig, cancel: CancellationToken, mut collect_fn: F)
298where
299    F: FnMut(&mut Vec<pb::Metric>),
300{
301    use monoio::time::MissedTickBehavior;
302
303    let url = format!("{}/v1/metrics", config.endpoint.trim_end_matches('/'));
304    let target = match MonoioHttpTarget::parse(&url) {
305        Ok(target) => target,
306        Err(e) => {
307            log::error!("Invalid monoio OTLP endpoint {url}: {e}");
308            return;
309        }
310    };
311
312    log::info!(
313        "Starting monoio OTLP metrics exporter, endpoint={url}, interval={}s",
314        config.interval.as_secs()
315    );
316
317    let attr_refs: Vec<(&str, &str)> = config
318        .resource_attributes
319        .iter()
320        .map(|(k, v)| (k.as_str(), v.as_str()))
321        .collect();
322    let resource = build_resource(&config.service_name, &attr_refs);
323
324    let mut interval = monoio::time::interval(config.interval);
325    interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
326    interval.tick().await;
327
328    let mut consecutive_failures: u32 = 0;
329    let mut bufs = ExportBufs::default();
330
331    let ctx = MonoioExportContext {
332        target: &target,
333        resource: &resource,
334        scope_name: &config.scope_name,
335        extra_headers: &config.headers,
336        timeout: config.timeout,
337    };
338
339    loop {
340        monoio::select! {
341            _ = interval.tick() => {}
342            _ = cancel.cancelled() => {
343                log::info!("monoio OTLP metrics exporter shutting down, performing final export");
344                export_once_monoio(&ctx, &mut collect_fn, &mut bufs).await;
345                return;
346            }
347        }
348
349        if consecutive_failures > 0 {
350            let backoff = backoff_with_jitter(consecutive_failures);
351            log::debug!(
352                "monoio OTLP export backing off {}ms (failures={consecutive_failures})",
353                backoff.as_millis()
354            );
355            monoio::select! {
356                _ = monoio::time::sleep(backoff) => {}
357                _ = cancel.cancelled() => {
358                    export_once_monoio(&ctx, &mut collect_fn, &mut bufs).await;
359                    return;
360                }
361            }
362        }
363
364        let mut metric_messages = Vec::new();
365        collect_fn(&mut metric_messages);
366
367        if metric_messages.is_empty() {
368            continue;
369        }
370
371        let metric_count = metric_messages.len();
372        let request = build_export_request(&resource, &config.scope_name, metric_messages);
373
374        bufs.encode.clear();
375        if let Err(e) = request.encode(&mut bufs.encode) {
376            log::warn!("monoio OTLP protobuf encode failed: {e}");
377            continue;
378        }
379        let body_len = bufs.encode.len();
380
381        match send_otlp_monoio(
382            &target,
383            &bufs.encode,
384            &mut bufs.gzip,
385            &config.headers,
386            config.timeout,
387        )
388        .await
389        {
390            Ok(resp) if resp.is_success() => {
391                consecutive_failures = 0;
392                log::debug!("Exported {metric_count} monoio OTLP metrics ({body_len} bytes)");
393            }
394            Ok(resp) => {
395                consecutive_failures = consecutive_failures.saturating_add(1);
396                log::warn!(
397                    "monoio OTLP export failed: status={}, body={}",
398                    resp.status,
399                    resp.body_text_lossy()
400                );
401            }
402            Err(e) => {
403                consecutive_failures = consecutive_failures.saturating_add(1);
404                log::warn!("monoio OTLP export request failed: {e}");
405            }
406        }
407    }
408}
409
410struct ExportContext<'a> {
411    client: &'a reqwest::Client,
412    url: &'a str,
413    resource: &'a pb::Resource,
414    scope_name: &'a str,
415    extra_headers: &'a [(String, String)],
416}
417
418#[derive(Default)]
419struct ExportBufs {
420    encode: Vec<u8>,
421    gzip: Vec<u8>,
422}
423
424async fn export_once<F>(ctx: &ExportContext<'_>, collect_fn: &mut F, bufs: &mut ExportBufs)
425where
426    F: FnMut(&mut Vec<pb::Metric>),
427{
428    let mut metric_messages = Vec::new();
429    collect_fn(&mut metric_messages);
430
431    if metric_messages.is_empty() {
432        return;
433    }
434
435    let request = build_export_request(ctx.resource, ctx.scope_name, metric_messages);
436
437    bufs.encode.clear();
438    if let Err(e) = request.encode(&mut bufs.encode) {
439        log::warn!("Final OTLP protobuf encode failed: {e}");
440        return;
441    }
442
443    match send_otlp(
444        ctx.client,
445        ctx.url,
446        &bufs.encode,
447        &mut bufs.gzip,
448        ctx.extra_headers,
449    )
450    .await
451    {
452        Ok(resp) if !resp.status().is_success() => {
453            let status = resp.status();
454            let body = resp.text().await.unwrap_or_default();
455            log::warn!("Final OTLP export returned {status}: {body}");
456        }
457        Err(e) => log::warn!("Final OTLP export failed: {e}"),
458        _ => {}
459    }
460}
461
462#[cfg(feature = "monoio")]
463struct MonoioExportContext<'a> {
464    target: &'a MonoioHttpTarget,
465    resource: &'a pb::Resource,
466    scope_name: &'a str,
467    extra_headers: &'a [(String, String)],
468    timeout: Duration,
469}
470
471#[cfg(feature = "monoio")]
472async fn export_once_monoio<F>(
473    ctx: &MonoioExportContext<'_>,
474    collect_fn: &mut F,
475    bufs: &mut ExportBufs,
476) where
477    F: FnMut(&mut Vec<pb::Metric>),
478{
479    let mut metric_messages = Vec::new();
480    collect_fn(&mut metric_messages);
481
482    if metric_messages.is_empty() {
483        return;
484    }
485
486    let request = build_export_request(ctx.resource, ctx.scope_name, metric_messages);
487
488    bufs.encode.clear();
489    if let Err(e) = request.encode(&mut bufs.encode) {
490        log::warn!("Final monoio OTLP protobuf encode failed: {e}");
491        return;
492    }
493
494    match send_otlp_monoio(
495        ctx.target,
496        &bufs.encode,
497        &mut bufs.gzip,
498        ctx.extra_headers,
499        ctx.timeout,
500    )
501    .await
502    {
503        Ok(resp) if !resp.is_success() => {
504            log::warn!(
505                "Final monoio OTLP export returned {}: {}",
506                resp.status,
507                resp.body_text_lossy()
508            );
509        }
510        Err(e) => log::warn!("Final monoio OTLP export failed: {e}"),
511        _ => {}
512    }
513}
514
515#[cfg(feature = "monoio")]
516#[derive(Debug)]
517struct MonoioHttpTarget {
518    connect_addr: String,
519    host_header: String,
520    path: String,
521}
522
523#[cfg(feature = "monoio")]
524impl MonoioHttpTarget {
525    fn parse(url: &str) -> Result<Self, MonoioHttpError> {
526        let rest = url
527            .strip_prefix("http://")
528            .ok_or_else(|| MonoioHttpError::InvalidEndpoint("only http:// is supported".into()))?;
529        let (authority, path) = match rest.find('/') {
530            Some(idx) => (&rest[..idx], &rest[idx..]),
531            None => (rest, "/"),
532        };
533
534        if authority.is_empty() {
535            return Err(MonoioHttpError::InvalidEndpoint(
536                "missing host in endpoint".into(),
537            ));
538        }
539
540        let connect_addr = connect_addr_for_authority(authority)?;
541
542        Ok(Self {
543            connect_addr,
544            host_header: authority.to_string(),
545            path: path.to_string(),
546        })
547    }
548}
549
550#[cfg(feature = "monoio")]
551fn connect_addr_for_authority(authority: &str) -> Result<String, MonoioHttpError> {
552    if let Some(rest) = authority.strip_prefix('[') {
553        let end = rest.find(']').ok_or_else(|| {
554            MonoioHttpError::InvalidEndpoint("IPv6 hosts must use [addr] syntax".into())
555        })?;
556        let suffix = &rest[end + 1..];
557        return if suffix.is_empty() {
558            Ok(format!("{authority}:80"))
559        } else if let Some(port) = suffix.strip_prefix(':') {
560            validate_port(port)?;
561            Ok(authority.to_string())
562        } else {
563            Err(MonoioHttpError::InvalidEndpoint(
564                "unexpected characters after IPv6 host".into(),
565            ))
566        };
567    }
568
569    match authority.rsplit_once(':') {
570        Some((_host, port)) => {
571            validate_port(port)?;
572            Ok(authority.to_string())
573        }
574        None => Ok(format!("{authority}:80")),
575    }
576}
577
578#[cfg(feature = "monoio")]
579fn validate_port(port: &str) -> Result<(), MonoioHttpError> {
580    if port.parse::<u16>().is_ok() {
581        Ok(())
582    } else {
583        Err(MonoioHttpError::InvalidEndpoint(
584            "invalid port in endpoint".into(),
585        ))
586    }
587}
588
589#[cfg(feature = "monoio")]
590#[derive(Debug)]
591struct MonoioHttpResponse {
592    status: u16,
593    body: Vec<u8>,
594}
595
596#[cfg(feature = "monoio")]
597impl MonoioHttpResponse {
598    fn is_success(&self) -> bool {
599        (200..300).contains(&self.status)
600    }
601
602    fn body_text_lossy(&self) -> String {
603        String::from_utf8_lossy(&self.body).into_owned()
604    }
605}
606
607#[cfg(feature = "monoio")]
608#[derive(Debug)]
609enum MonoioHttpError {
610    InvalidEndpoint(String),
611    InvalidHeader(String),
612    Io(std::io::Error),
613    Timeout,
614    InvalidResponse,
615}
616
617#[cfg(feature = "monoio")]
618impl std::fmt::Display for MonoioHttpError {
619    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
620        match self {
621            Self::InvalidEndpoint(msg) => write!(f, "{msg}"),
622            Self::InvalidHeader(name) => write!(f, "invalid HTTP header {name:?}"),
623            Self::Io(e) => write!(f, "{e}"),
624            Self::Timeout => f.write_str("request timed out"),
625            Self::InvalidResponse => f.write_str("invalid HTTP response"),
626        }
627    }
628}
629
630#[cfg(feature = "monoio")]
631impl From<std::io::Error> for MonoioHttpError {
632    fn from(value: std::io::Error) -> Self {
633        Self::Io(value)
634    }
635}
636
637#[cfg(feature = "monoio")]
638async fn send_otlp_monoio(
639    target: &MonoioHttpTarget,
640    body: &[u8],
641    gzip_buf: &mut Vec<u8>,
642    extra_headers: &[(String, String)],
643    timeout: Duration,
644) -> Result<MonoioHttpResponse, MonoioHttpError> {
645    let compressed = gzip_compress(body, gzip_buf);
646    let request_body = if compressed {
647        gzip_buf.as_slice()
648    } else {
649        body
650    };
651    let request = build_monoio_http_request(target, request_body, compressed, extra_headers)?;
652
653    match monoio::time::timeout(timeout, send_monoio_http_request(target, request)).await {
654        Ok(result) => result,
655        Err(_) => Err(MonoioHttpError::Timeout),
656    }
657}
658
659#[cfg(feature = "monoio")]
660fn build_monoio_http_request(
661    target: &MonoioHttpTarget,
662    body: &[u8],
663    compressed: bool,
664    extra_headers: &[(String, String)],
665) -> Result<Vec<u8>, MonoioHttpError> {
666    use std::fmt::Write as _;
667
668    let mut head = String::with_capacity(512 + extra_headers.len() * 64);
669    write!(
670        &mut head,
671        "POST {} HTTP/1.1\r\nHost: {}\r\nContent-Type: application/x-protobuf\r\nContent-Length: {}\r\nConnection: close\r\n",
672        target.path,
673        target.host_header,
674        body.len()
675    )
676    .expect("writing to String cannot fail");
677
678    if compressed {
679        head.push_str("Content-Encoding: gzip\r\n");
680    }
681
682    for (name, value) in extra_headers {
683        validate_header(name, value)?;
684        write!(&mut head, "{name}: {value}\r\n").expect("writing to String cannot fail");
685    }
686
687    head.push_str("\r\n");
688    let mut request = head.into_bytes();
689    request.extend_from_slice(body);
690    Ok(request)
691}
692
693#[cfg(feature = "monoio")]
694fn validate_header(name: &str, value: &str) -> Result<(), MonoioHttpError> {
695    let invalid_name = name.is_empty()
696        || name
697            .bytes()
698            .any(|b| b <= b' ' || b == b':' || b == b'\r' || b == b'\n');
699    let invalid_value = value.bytes().any(|b| b == b'\r' || b == b'\n');
700
701    if invalid_name || invalid_value {
702        Err(MonoioHttpError::InvalidHeader(name.to_string()))
703    } else {
704        Ok(())
705    }
706}
707
708#[cfg(feature = "monoio")]
709async fn send_monoio_http_request(
710    target: &MonoioHttpTarget,
711    request: Vec<u8>,
712) -> Result<MonoioHttpResponse, MonoioHttpError> {
713    use monoio::io::{AsyncReadRent, AsyncWriteRentExt};
714    use monoio::net::TcpStream;
715
716    let mut stream = TcpStream::connect(target.connect_addr.as_str()).await?;
717    let (write_result, _request) = stream.write_all(request).await;
718    write_result?;
719
720    let mut response = Vec::with_capacity(8192);
721    loop {
722        let buf = Vec::with_capacity(8192);
723        let (read_result, buf) = stream.read(buf).await;
724        let n = read_result?;
725        if n == 0 {
726            break;
727        }
728        let available = buf.len().min(n);
729        response.extend_from_slice(&buf[..available]);
730        if response.len() >= 64 * 1024 {
731            break;
732        }
733    }
734
735    parse_monoio_http_response(response)
736}
737
738#[cfg(feature = "monoio")]
739fn parse_monoio_http_response(response: Vec<u8>) -> Result<MonoioHttpResponse, MonoioHttpError> {
740    let header_end = response
741        .windows(4)
742        .position(|w| w == b"\r\n\r\n")
743        .map(|idx| idx + 4)
744        .ok_or(MonoioHttpError::InvalidResponse)?;
745
746    let status_line_end = response[..header_end]
747        .windows(2)
748        .position(|w| w == b"\r\n")
749        .ok_or(MonoioHttpError::InvalidResponse)?;
750    let status_line = std::str::from_utf8(&response[..status_line_end])
751        .map_err(|_| MonoioHttpError::InvalidResponse)?;
752    let status = parse_status_code(status_line).ok_or(MonoioHttpError::InvalidResponse)?;
753    let body = response[header_end..].to_vec();
754
755    Ok(MonoioHttpResponse { status, body })
756}
757
758#[cfg(feature = "monoio")]
759fn parse_status_code(status_line: &str) -> Option<u16> {
760    let mut parts = status_line.split_whitespace();
761    let version = parts.next()?;
762    if !version.starts_with("HTTP/") {
763        return None;
764    }
765    parts.next()?.parse().ok()
766}
767
768/// Compute backoff with jitter: min(MAX_BACKOFF, BASE_BACKOFF * 2^failures) +/- 25% jitter.
769fn backoff_with_jitter(consecutive_failures: u32) -> Duration {
770    let exp = consecutive_failures.min(10);
771    let base_ms = BASE_BACKOFF.as_millis() as u64;
772    let backoff_ms = base_ms
773        .saturating_mul(1u64 << exp)
774        .min(MAX_BACKOFF.as_millis() as u64);
775
776    let nanos = std::time::SystemTime::now()
777        .duration_since(std::time::UNIX_EPOCH)
778        .unwrap_or_default()
779        .subsec_nanos();
780    let jitter_range = (backoff_ms / 4).max(1);
781    let jitter = (nanos as u64 % (jitter_range * 2 + 1)).saturating_sub(jitter_range);
782    let final_ms = backoff_ms.saturating_add(jitter);
783
784    Duration::from_millis(final_ms)
785}
786
787#[cfg(all(test, feature = "monoio"))]
788mod monoio_tests {
789    use super::*;
790
791    #[test]
792    fn parses_http_target_with_port() {
793        let target = MonoioHttpTarget::parse("http://localhost:4318/v1/metrics").unwrap();
794        assert_eq!(target.connect_addr, "localhost:4318");
795        assert_eq!(target.host_header, "localhost:4318");
796        assert_eq!(target.path, "/v1/metrics");
797    }
798
799    #[test]
800    fn parses_http_target_without_port() {
801        let target = MonoioHttpTarget::parse("http://collector/v1/metrics").unwrap();
802        assert_eq!(target.connect_addr, "collector:80");
803        assert_eq!(target.host_header, "collector");
804        assert_eq!(target.path, "/v1/metrics");
805    }
806
807    #[test]
808    fn rejects_https_target() {
809        let err = MonoioHttpTarget::parse("https://collector:4318/v1/metrics").unwrap_err();
810        assert!(matches!(err, MonoioHttpError::InvalidEndpoint(_)));
811    }
812
813    #[test]
814    fn rejects_invalid_port() {
815        let err = MonoioHttpTarget::parse("http://collector:nope/v1/metrics").unwrap_err();
816        assert!(matches!(err, MonoioHttpError::InvalidEndpoint(_)));
817    }
818
819    #[test]
820    fn builds_http_request() {
821        let target = MonoioHttpTarget::parse("http://localhost:4318/v1/metrics").unwrap();
822        let request = build_monoio_http_request(
823            &target,
824            b"abc",
825            true,
826            &[("Authorization".to_string(), "Bearer token".to_string())],
827        )
828        .unwrap();
829        let request = String::from_utf8(request).unwrap();
830        assert!(request.starts_with("POST /v1/metrics HTTP/1.1\r\n"));
831        assert!(request.contains("Host: localhost:4318\r\n"));
832        assert!(request.contains("Content-Length: 3\r\n"));
833        assert!(request.contains("Content-Encoding: gzip\r\n"));
834        assert!(request.ends_with("\r\n\r\nabc"));
835    }
836
837    #[test]
838    fn parses_http_status() {
839        let response = parse_monoio_http_response(
840            b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n".to_vec(),
841        )
842        .unwrap();
843        assert_eq!(response.status, 204);
844        assert!(response.body.is_empty());
845    }
846}
fast_telemetry_export/otlp.rs

fast_telemetry_export/
otlp.rs
