Surface X-Logfire-Warning response headers from the Logfire API

logfire-client/src/client.rs

@@ -3,6 +3,8 @@
 use reqwest::StatusCode;
 use serde::de::DeserializeOwned;
 
+use logfire_core::server_response::{WARNING_HEADER_NAME, emit_warning};
+
 use crate::types::{ReadTokenInfo, RowQueryResults, SchemasResponse};
 
 /// Errors that can occur when executing queries.
@@ -30,6 +32,13 @@ pub enum ClientError {
     RowDeserialize(#[source] serde_json::Error),
 }
 
+/// The warning which the Logfire backend sent with a response, if any.
+fn warning_header(headers: &reqwest::header::HeaderMap) -> Option<&str> {
+    headers
+        .get(WARNING_HEADER_NAME)
+        .and_then(|value| value.to_str().ok())
+}
+
 /// HTTP client for querying Logfire.
 #[derive(Clone, Debug)]
 pub struct LogfireClient {
@@ -58,6 +67,10 @@ impl LogfireClient {
 
         let response = request.send().await.map_err(ClientError::Request)?;
 
+        if let Some(warning) = warning_header(response.headers()) {
+            emit_warning(warning);
+        }
+
         let status = response.status();
         if !status.is_success() {
             let body = response.text().await.unwrap_or_default();
@@ -102,9 +115,21 @@ impl LogfireClient {
 
 #[cfg(test)]
 mod tests {
-    use super::{ClientError, LogfireClient, StatusCode};
+    use super::{ClientError, LogfireClient, StatusCode, warning_header};
     use crate::builder::LogfireClientBuilder;
 
+    #[test]
+    fn warning_header_is_read() {
+        let mut headers = reqwest::header::HeaderMap::new();
+        assert_eq!(warning_header(&headers), None);
+
+        headers.insert(
+            "X-Logfire-Warning",
+            reqwest::header::HeaderValue::from_static("endpoint is deprecated"),
+        );
+        assert_eq!(warning_header(&headers), Some("endpoint is deprecated"));
+    }
+
     #[test]
     fn query_failed_error_format() {
         let err = ClientError::QueryFailed {

logfire-core/src/lib.rs

@@ -3,6 +3,7 @@
 //! This crate provides common functionality used by both the instrumentation SDK (`logfire`)
 //! and the query client (`logfire-client`).
 
+pub mod server_response;
 mod token;
 
 pub use token::{Region, parse_region};

logfire-core/src/server_response.rs

@@ -0,0 +1,45 @@
+//! Out-of-band signals which the Logfire backend attaches to its API responses.
+//!
+//! The backend sets the [`WARNING_HEADER_NAME`] header on responses to signal something it wants
+//! the user to see (for example the deprecation of an endpoint), without relying on the response
+//! body. SDKs surface the header value as a warning on stderr.
+
+use std::{
+    collections::HashSet,
+    sync::{Mutex, OnceLock, PoisonError},
+};
+
+/// The response header used by the Logfire backend to send a warning to the SDK.
+pub const WARNING_HEADER_NAME: &str = "x-logfire-warning";
+
+/// Print a warning received from the Logfire backend to stderr.
+///
+/// Repeats of the same message are suppressed, so that a chatty server only warns once per
+/// process (matching the deduplication which Python's `warnings` module does for the Python SDK).
+pub fn emit_warning(message: &str) {
+    if is_first_occurrence(message) {
+        eprintln!("Logfire server warning: {message}");
+    }
+}
+
+/// Record `message` as seen, returning whether it had not been seen before.
+fn is_first_occurrence(message: &str) -> bool {
+    static SEEN: OnceLock<Mutex<HashSet<String>>> = OnceLock::new();
+    SEEN.get_or_init(Mutex::default)
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner)
+        .insert(message.to_owned())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::is_first_occurrence;
+
+    #[test]
+    fn repeated_warnings_are_deduplicated() {
+        let message = "repeated_warnings_are_deduplicated"; // unique to this test
+        assert!(is_first_occurrence(message));
+        assert!(!is_first_occurrence(message));
+        assert!(is_first_occurrence("a different message"));
+    }
+}

logfire/Cargo.toml

@@ -15,10 +15,15 @@ logfire-core.workspace = true
 log = "0.4"
 env_filter = "2"
 
+http = "1.2"
+
 # deps for grpc export
-http = { version = "1.2", optional = true }
 tonic = { version = "0.14", optional = true }
 
+# deps for http export
+async-trait = { version = "0.1", optional = true }
+opentelemetry-http = { version = "0.32", optional = true, default-features = false }
+
 # TLS for OTLP exports. `reqwest` must stay on the same major version as `opentelemetry-http`'s,
 # so that the TLS features enabled here apply to the client it builds.
 reqwest = { version = "0.13", optional = true, default-features = false }
@@ -108,20 +113,23 @@ data-dir = ["dep:serde", "dep:serde_json"]
 export-grpc = [
     "opentelemetry-otlp/grpc-tonic",
     "tonic/tls-native-roots",
-    "dep:http",
     "dep:rustls",
     "dep:tonic",
 ]
 export-http-protobuf = [
     "opentelemetry-otlp/http-proto",
     "opentelemetry-otlp/reqwest-client",
     "reqwest/rustls-no-provider",
+    "dep:async-trait",
+    "dep:opentelemetry-http",
     "dep:rustls",
 ]
 export-http-json = [
     "opentelemetry-otlp/http-json",
     "opentelemetry-otlp/reqwest-client",
     "reqwest/rustls-no-provider",
+    "dep:async-trait",
+    "dep:opentelemetry-http",
     "dep:rustls",
 ]
 # Use aws-lc-rs for TLS when the application has not installed a rustls `CryptoProvider`.

logfire/src/config.rs

@@ -19,7 +19,12 @@ use opentelemetry_sdk::{
 };
 use tracing::{Level, level_filters::LevelFilter};
 
-use crate::{ConfigureError, internal::env::get_optional_env, logfire::Logfire};
+use crate::{
+    ConfigureError,
+    internal::env::get_optional_env,
+    logfire::Logfire,
+    server_response::{ServerResponse, ServerResponseHook},
+};
 
 /// Builder for logfire configuration, returned from [`logfire::configure()`][crate::configure].
 #[must_use = "call `.finish()` to complete logfire configuration."]
@@ -393,6 +398,7 @@ pub struct AdvancedOptions {
     pub(crate) resources: Vec<opentelemetry_sdk::Resource>,
     pub(crate) log_record_processors: Vec<BoxedLogProcessor>,
     pub(crate) enable_tracing_metrics: bool,
+    pub(crate) server_response_hook: Option<ServerResponseHook>,
     //
     //
     // TODO: arguments below supported by Python
@@ -444,6 +450,38 @@ impl AdvancedOptions {
         self.enable_tracing_metrics = enable;
         self
     }
+
+    /// Set a hook which is called for every response received from the Logfire API.
+    ///
+    /// This is experimental and may change in a future release.
+    ///
+    /// The hook replaces the default behaviour, which prints any warning sent by the Logfire
+    /// backend (the `x-logfire-warning` response header) to stderr. Call
+    /// [`ServerResponse::default_hook`] from the hook to keep that behaviour, or set a hook which
+    /// does nothing to opt out of it entirely.
+    ///
+    /// The hook is called on the SDK's export thread, so it should not block.
+    ///
+    /// Responses are only inspected for exports over HTTP (the default protocol); the
+    /// `export-grpc` feature has no equivalent support.
+    ///
+    /// ```rust
+    /// use logfire::config::AdvancedOptions;
+    ///
+    /// let advanced = AdvancedOptions::default().with_server_response_hook(|response| {
+    ///     if let Some(warning) = response.warning() {
+    ///         tracing::warn!("logfire: {warning}");
+    ///     }
+    /// });
+    /// ```
+    #[must_use]
+    pub fn with_server_response_hook<F>(mut self, hook: F) -> Self
+    where
+        F: Fn(&ServerResponse<'_>) + Send + Sync + 'static,
+    {
+        self.server_response_hook = Some(Arc::new(hook));
+        self
+    }
 }
 
 /// Configuration of metrics.

logfire/src/exporters.rs

@@ -4,7 +4,15 @@
         feature = "export-http-protobuf",
         feature = "export-http-json"
     )),
-    expect(unreachable_code, unused_variables)
+    expect(unreachable_code)
+)]
+#![cfg_attr(
+    not(any(feature = "export-http-protobuf", feature = "export-http-json")),
+    allow(
+        unused_variables,
+        clippy::needless_pass_by_value,
+        reason = "the endpoint configuration and server response hook are only used for exports over HTTP"
+    )
 )]
 //! Helper functions to configure exporters via OTLP.
 use std::collections::HashMap;
@@ -14,7 +22,7 @@ use opentelemetry_sdk::{
     logs::LogExporter, metrics::exporter::PushMetricExporter, trace::SpanExporter,
 };
 
-use crate::{ConfigureError, internal::env::get_optional_env};
+use crate::{ConfigureError, internal::env::get_optional_env, server_response::ServerResponseHook};
 
 /// The `User-Agent` sent with OTLP exports, e.g. `logfire-rust/0.13.0`.
 ///
@@ -85,6 +93,15 @@ macro_rules! feature_required {
 pub fn span_exporter(
     endpoint: &str,
     headers: Option<HashMap<String, String>>,
+) -> Result<impl SpanExporter + use<>, ConfigureError> {
+    span_exporter_with_hook(endpoint, headers, None)
+}
+
+/// As [`span_exporter`], with a hook called for each response received from the Logfire API.
+pub(crate) fn span_exporter_with_hook(
+    endpoint: &str,
+    headers: Option<HashMap<String, String>>,
+    server_response_hook: Option<ServerResponseHook>,
 ) -> Result<impl SpanExporter + use<>, ConfigureError> {
     let protocol = protocol_from_env("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL")?;
 
@@ -113,6 +130,10 @@ pub fn span_exporter(
                 .with_http()
                 .with_protocol(Protocol::HttpBinary)
                 .with_headers(headers_with_user_agent(headers))
+                .with_http_client(http_client(
+                    "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT",
+                    server_response_hook,
+                )?)
                 .with_endpoint(format!("{endpoint}/v1/traces"))
                 .build()?
         }
@@ -125,6 +146,10 @@ pub fn span_exporter(
                 .with_http()
                 .with_protocol(Protocol::HttpJson)
                 .with_headers(headers_with_user_agent(headers))
+                .with_http_client(http_client(
+                    "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT",
+                    server_response_hook,
+                )?)
                 .with_endpoint(format!("{endpoint}/v1/traces"))
                 .build()?
         }
@@ -170,6 +195,15 @@ pub fn span_exporter(
 pub fn metric_exporter(
     endpoint: &str,
     headers: Option<HashMap<String, String>>,
+) -> Result<impl PushMetricExporter + use<>, ConfigureError> {
+    metric_exporter_with_hook(endpoint, headers, None)
+}
+
+/// As [`metric_exporter`], with a hook called for each response received from the Logfire API.
+pub(crate) fn metric_exporter_with_hook(
+    endpoint: &str,
+    headers: Option<HashMap<String, String>>,
+    server_response_hook: Option<ServerResponseHook>,
 ) -> Result<impl PushMetricExporter + use<>, ConfigureError> {
     let protocol = protocol_from_env("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL")?;
 
@@ -200,6 +234,10 @@ pub fn metric_exporter(
                 .with_http()
                 .with_protocol(Protocol::HttpBinary)
                 .with_headers(headers_with_user_agent(headers))
+                .with_http_client(http_client(
+                    "OTEL_EXPORTER_OTLP_METRICS_TIMEOUT",
+                    server_response_hook,
+                )?)
                 .with_endpoint(format!("{endpoint}/v1/metrics"))
                 .build()?)
         }
@@ -213,6 +251,10 @@ pub fn metric_exporter(
                 .with_http()
                 .with_protocol(Protocol::HttpJson)
                 .with_headers(headers_with_user_agent(headers))
+                .with_http_client(http_client(
+                    "OTEL_EXPORTER_OTLP_METRICS_TIMEOUT",
+                    server_response_hook,
+                )?)
                 .with_endpoint(format!("{endpoint}/v1/metrics"))
                 .build()?)
         }
@@ -249,6 +291,15 @@ pub fn metric_exporter(
 pub fn log_exporter(
     endpoint: &str,
     headers: Option<HashMap<String, String>>,
+) -> Result<impl LogExporter + use<>, ConfigureError> {
+    log_exporter_with_hook(endpoint, headers, None)
+}
+
+/// As [`log_exporter`], with a hook called for each response received from the Logfire API.
+pub(crate) fn log_exporter_with_hook(
+    endpoint: &str,
+    headers: Option<HashMap<String, String>>,
+    server_response_hook: Option<ServerResponseHook>,
 ) -> Result<impl LogExporter + use<>, ConfigureError> {
     let protocol = protocol_from_env("OTEL_EXPORTER_OTLP_LOGS_PROTOCOL")?;
 
@@ -277,6 +328,10 @@ pub fn log_exporter(
                 .with_http()
                 .with_protocol(Protocol::HttpBinary)
                 .with_headers(headers_with_user_agent(headers))
+                .with_http_client(http_client(
+                    "OTEL_EXPORTER_OTLP_LOGS_TIMEOUT",
+                    server_response_hook,
+                )?)
                 .with_endpoint(format!("{endpoint}/v1/logs"))
                 .build()?)
         }
@@ -289,6 +344,10 @@ pub fn log_exporter(
                 .with_http()
                 .with_protocol(Protocol::HttpJson)
                 .with_headers(headers_with_user_agent(headers))
+                .with_http_client(http_client(
+                    "OTEL_EXPORTER_OTLP_LOGS_TIMEOUT",
+                    server_response_hook,
+                )?)
                 .with_endpoint(format!("{endpoint}/v1/logs"))
                 .build()?)
         }
@@ -309,6 +368,41 @@ pub fn log_exporter(
     }
 }
 
+/// The default OTLP export timeout, matching the default used by `opentelemetry-otlp`.
+#[cfg(any(feature = "export-http-protobuf", feature = "export-http-json"))]
+const DEFAULT_EXPORT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+/// Build the `reqwest` client used for HTTP exports.
+///
+/// `opentelemetry-otlp` would build one itself, but logfire needs its own so that responses can
+/// be passed to the [server response hook][crate::config::AdvancedOptions::with_server_response_hook].
+/// That means resolving the export timeout here too, the same way `opentelemetry-otlp` does.
+#[cfg(any(feature = "export-http-protobuf", feature = "export-http-json"))]
+fn http_client(
+    signal_timeout_var: &str,
+    server_response_hook: Option<ServerResponseHook>,
+) -> Result<crate::server_response::LogfireHttpClient, ConfigureError> {
+    let mut timeout = DEFAULT_EXPORT_TIMEOUT;
+    for var_name in [signal_timeout_var, "OTEL_EXPORTER_OTLP_TIMEOUT"] {
+        if let Some(value) = get_optional_env(var_name, None)?
+            && let Ok(millis) = value.parse()
+        {
+            timeout = std::time::Duration::from_millis(millis);
+            break;
+        }
+    }
+
+    let client = reqwest::Client::builder()
+        .timeout(timeout)
+        .build()
+        .map_err(|e| ConfigureError::Other(e.into()))?;
+
+    Ok(crate::server_response::LogfireHttpClient::new(
+        client,
+        server_response_hook,
+    ))
+}
+
 /// Build the `tonic` endpoint used for gRPC exports.
 ///
 /// TLS is configured here because `tonic` only configures it itself in `Endpoint::new`, which