feat(logs): opt-in log collection over iroh-services

Cargo.toml

@@ -60,3 +60,6 @@ default = []
 
 [[example]]
 name = "net_diagnostics"
+
+[[example]]
+name = "logs"

examples/logs.rs

@@ -0,0 +1,96 @@
+//! Log collection example.
+//!
+//! Demonstrates how to install the iroh-services log collector alongside the
+//! standard `tracing-subscriber` `fmt` layer, ship records to the cloud over
+//! the iroh-services RPC channel, and let the cloud override the local
+//! filter at runtime via `SetLogLevel`.
+//!
+//! The level filter starts at `off`. The cloud pushes a level after this
+//! endpoint is opted into log collection from the dashboard or REST API; both
+//! the buffered cloud-shipping layer and the stderr fmt layer respect that
+//! level. Anything emitted before the cloud responds is silently dropped.
+//!
+//! Run with: cargo run --example logs
+
+use std::time::Duration;
+use tracing_subscriber::prelude::*;
+
+use anyhow::Result;
+use iroh::{Endpoint, endpoint::presets, protocol::Router};
+use iroh_services::{
+    API_SECRET_ENV_VAR_NAME, ApiSecret, CLIENT_HOST_ALPN, Client, ClientHost, caps::LogsCap, logs,
+};
+use tracing::info;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // 1. Build the buffer layer and compose it with a stderr fmt layer behind
+    //    the same reloadable filter, so local console output mirrors what
+    //    gets shipped. The cloud raises the filter from `off` after this
+    //    endpoint is opted in via the dashboard.
+    let (collector, log_layer) = logs::layer();
+    tracing_subscriber::registry()
+        .with(log_layer)
+        .with(tracing_subscriber::fmt::layer().with_writer(std::io::stderr))
+        .try_init()
+        .map_err(|e| anyhow::anyhow!("failed to install tracing subscriber: {e}"))?;
+
+    // 2. Create the endpoint and parse the API secret so we know which
+    //    cloud endpoint to grant SetLevel to.
+    let endpoint = Endpoint::bind(presets::N0).await?;
+    let secret = ApiSecret::from_env_var(API_SECRET_ENV_VAR_NAME)?;
+    let cloud_id = secret.addr().id;
+
+    let name = format!("logs-example-{}", &endpoint.id().to_string()[..8]);
+
+    // 3. Build the client. `with_log_collection(collector.clone())` starts a
+    //    background task that drains the buffer every second and ships the
+    //    batch as a `PutLogs` RPC.
+    let client = Client::builder(&endpoint)
+        .api_secret(secret)?
+        .name(name)?
+        .with_log_collection(collector.clone())
+        .build()
+        .await?;
+
+    // 4. Grant `LogsCap::SetLevel` so the cloud can dial us back and apply a
+    //    filter override. Spawned so a momentarily-down cloud does not block
+    //    startup.
+    let client_for_grant = client.clone();
+    let grant_task = tokio::spawn(async move {
+        if let Err(err) = client_for_grant
+            .grant_capability(cloud_id, [LogsCap::SetLevel])
+            .await
+        {
+            eprintln!("failed to grant LogsCap::SetLevel: {err:?}");
+        }
+    });
+
+    // 5. Accept the cloud's callback connections on `CLIENT_HOST_ALPN`. The
+    //    `ClientHost` needs the same collector so the `SetLogLevel` request
+    //    can hot-reload the filter.
+    let host = ClientHost::new(&endpoint).with_log_collector(collector);
+    let router = Router::builder(endpoint)
+        .accept(CLIENT_HOST_ALPN, host)
+        .spawn();
+
+    // 6. Emit an info log every other second forever. These will surface on
+    //    the dashboard's Logs page (and on this process's stderr) once the
+    //    endpoint is opted into log collection at `info` level or above.
+    println!("emitting logs; ctrl+c to exit.");
+    let mut tick = tokio::time::interval(Duration::from_secs(2));
+    let mut counter: u64 = 0;
+    loop {
+        tokio::select! {
+            _ = tick.tick() => {
+                counter += 1;
+                info!(counter, "logs example heartbeat");
+            }
+            _ = tokio::signal::ctrl_c() => break,
+        }
+    }
+
+    grant_task.abort();
+    router.endpoint().close().await;
+    Ok(())
+}

src/caps.rs

@@ -81,6 +81,8 @@ pub enum Cap {
     Metrics(MetricsCap),
     #[strum(to_string = "net-diagnostics:{0}")]
     NetDiagnostics(NetDiagnosticsCap),
+    #[strum(to_string = "logs:{0}")]
+    Logs(LogsCap),
 }
 
 impl FromStr for Cap {
@@ -94,6 +96,7 @@ impl FromStr for Cap {
                 "metrics" => Self::Metrics(MetricsCap::from_str(inner)?),
                 "relay" => Self::Relay(RelayCap::from_str(inner)?),
                 "net-diagnostics" => Self::NetDiagnostics(NetDiagnosticsCap::from_str(inner)?),
+                "logs" => Self::Logs(LogsCap::from_str(inner)?),
                 _ => bail!("invalid cap domain"),
             })
         } else {
@@ -121,6 +124,16 @@ cap_enum!(
     }
 );
 
+cap_enum!(
+    /// Capabilities for the log collection feature.
+    pub enum LogsCap {
+        /// Permits the bearer to push log lines to the cloud.
+        Push,
+        /// Permits the bearer to set the log level filter on the issuer at runtime.
+        SetLevel,
+    }
+);
+
 impl Caps {
     pub fn new(caps: impl IntoIterator<Item = impl Into<Cap>>) -> Self {
         Self::V0(CapSet::new(caps))
@@ -179,6 +192,7 @@ impl Capability for Cap {
             (Cap::Relay(slf), Cap::Relay(other)) => slf.permits(other),
             (Cap::Metrics(slf), Cap::Metrics(other)) => slf.permits(other),
             (Cap::NetDiagnostics(slf), Cap::NetDiagnostics(other)) => slf.permits(other),
+            (Cap::Logs(slf), Cap::Logs(other)) => slf.permits(other),
             (_, _) => false,
         }
     }
@@ -192,6 +206,8 @@ fn client_capabilities(other: &Cap) -> bool {
         Cap::Metrics(MetricsCap::PutAny) => true,
         Cap::NetDiagnostics(NetDiagnosticsCap::PutAny) => true,
         Cap::NetDiagnostics(NetDiagnosticsCap::GetAny) => true,
+        Cap::Logs(LogsCap::Push) => true,
+        Cap::Logs(LogsCap::SetLevel) => true,
     }
 }
 
@@ -221,6 +237,16 @@ impl Capability for NetDiagnosticsCap {
     }
 }
 
+impl Capability for LogsCap {
+    fn permits(&self, other: &Self) -> bool {
+        match (self, other) {
+            (LogsCap::Push, LogsCap::Push) => true,
+            (LogsCap::SetLevel, LogsCap::SetLevel) => true,
+            (_, _) => false,
+        }
+    }
+}
+
 /// A set of capabilities
 #[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Clone, Serialize, Deserialize)]
 pub struct CapSet<C: Capability + Ord>(BTreeSet<C>);

src/client.rs

@@ -17,9 +17,10 @@ use uuid::Uuid;
 use crate::{
     api_secret::ApiSecret,
     caps::Caps,
+    logs::LogCollector,
     net_diagnostics::{DiagnosticsReport, checks::run_diagnostics},
     protocol::{
-        ALPN, Auth, IrohServicesClient, NameEndpoint, Ping, Pong, PutMetrics,
+        ALPN, Auth, IrohServicesClient, NameEndpoint, Ping, Pong, PutLogs, PutMetrics,
         PutNetworkDiagnostics, RemoteError,
     },
 };
@@ -54,6 +55,7 @@ pub struct Client {
     endpoint: Endpoint,
     message_channel: tokio::sync::mpsc::Sender<ClientActorMessage>,
     _actor_task: Arc<AbortOnDropHandle<()>>,
+    _log_flush_task: Option<Arc<AbortOnDropHandle<()>>>,
 }
 
 /// ClientBuilder provides configures and builds a iroh-services client, typically
@@ -67,11 +69,19 @@ pub struct ClientBuilder {
     metrics_interval: Option<Duration>,
     remote: Option<EndpointAddr>,
     registry: Registry,
+    log_collector: Option<LogCollector>,
+    log_flush_interval: Duration,
+    log_max_batch: usize,
 }
 
 const DEFAULT_CAP_EXPIRY: Duration = Duration::from_secs(60 * 60 * 24 * 30); // 1 month
 pub const API_SECRET_ENV_VAR_NAME: &str = "IROH_SERVICES_API_SECRET";
 
+/// Default interval between log batch flushes when log collection is enabled.
+pub const DEFAULT_LOG_FLUSH_INTERVAL: Duration = Duration::from_secs(1);
+/// Default maximum batch size pushed in a single PutLogs request.
+pub const DEFAULT_LOG_MAX_BATCH: usize = 200;
+
 impl ClientBuilder {
     pub fn new(endpoint: &Endpoint) -> Self {
         let mut registry = Registry::default();
@@ -85,9 +95,35 @@ impl ClientBuilder {
             metrics_interval: Some(Duration::from_secs(60)),
             remote: None,
             registry,
+            log_collector: None,
+            log_flush_interval: DEFAULT_LOG_FLUSH_INTERVAL,
+            log_max_batch: DEFAULT_LOG_MAX_BATCH,
         }
     }
 
+    /// Enables periodic shipment of buffered log lines to iroh-services.
+    ///
+    /// The collector is shared with [`crate::client_host::ClientHost`] when
+    /// runtime log-level overrides are needed; clone it before passing so both
+    /// sides hold a handle.
+    pub fn with_log_collection(mut self, collector: LogCollector) -> Self {
+        self.log_collector = Some(collector);
+        self
+    }
+
+    /// Override the log batch flush interval. Defaults to one second.
+    pub fn log_flush_interval(mut self, interval: Duration) -> Self {
+        self.log_flush_interval = interval;
+        self
+    }
+
+    /// Override the maximum number of lines included in a single PutLogs
+    /// request. Defaults to [`DEFAULT_LOG_MAX_BATCH`].
+    pub fn log_max_batch(mut self, max: usize) -> Self {
+        self.log_max_batch = max;
+        self
+    }
+
     /// Register a metrics group to forward to iroh-services
     ///
     /// The default registered metrics uses only the endpoint
@@ -213,22 +249,37 @@ impl ClientBuilder {
         let conn = IrohLazyRemoteConnection::new(self.endpoint.clone(), remote, ALPN.to_vec());
         let irpc_client = IrohServicesClient::boxed(conn);
 
-        let (tx, rx) = tokio::sync::mpsc::channel(1);
+        let session_id = Uuid::new_v4();
+        // The actor mailbox is only used for control-plane messages (auth,
+        // ping, name, grant_cap) plus the periodic metrics + log flush. A
+        // small buffer is enough but `1` head-of-line-blocks log flushes
+        // behind metrics ticks, so leave a little room.
+        let (tx, rx) = tokio::sync::mpsc::channel(8);
         let actor_task = AbortOnDropHandle::new(n0_future::task::spawn(
             ClientActor {
                 capabilities,
                 client: irpc_client,
                 name: self.name.clone(),
-                session_id: Uuid::new_v4(),
+                session_id,
                 authorized: false,
             }
             .run(self.name, self.registry, self.metrics_interval, rx),
         ));
 
+        let log_flush_task = self.log_collector.map(|collector| {
+            let message_channel = tx.clone();
+            let interval = self.log_flush_interval;
+            let max_batch = self.log_max_batch;
+            Arc::new(AbortOnDropHandle::new(n0_future::task::spawn(
+                run_log_flush(message_channel, collector, interval, max_batch, session_id),
+            )))
+        });
+
         Ok(Client {
             endpoint: self.endpoint,
             message_channel: tx,
             _actor_task: Arc::new(actor_task),
+            _log_flush_task: log_flush_task,
         })
     }
 }
@@ -425,6 +476,10 @@ enum ClientActorMessage {
         report: Box<DiagnosticsReport>,
         done: oneshot::Sender<Result<(), Error>>,
     },
+    PutLogs {
+        request: PutLogs,
+        done: oneshot::Sender<Result<(), Error>>,
+    },
     ReadName {
         done: oneshot::Sender<Option<String>>,
     },
@@ -505,6 +560,13 @@ impl ClientActor {
                                 warn!("failed to publish network diagnostics: {:#?}", err);
                             }
                         }
+                        ClientActorMessage::PutLogs{ request, done } => {
+                            let res = self.put_logs(request).await;
+                            if let Err(err) = done.send(res) {
+                                debug!("failed to publish logs: {:#?}", err);
+                                self.authorized = false;
+                            }
+                        }
                     }
                 }
                 _ = async {
@@ -613,6 +675,77 @@ impl ClientActor {
 
         Ok(())
     }
+
+    async fn put_logs(&mut self, request: PutLogs) -> Result<(), Error> {
+        trace!(
+            lines = request.lines.len(),
+            dropped = request.dropped,
+            "client actor put logs"
+        );
+        self.auth().await?;
+
+        self.client
+            .rpc(request)
+            .await
+            .map_err(|_| RemoteError::InternalServerError)??;
+
+        Ok(())
+    }
+}
+
+async fn run_log_flush(
+    message_channel: tokio::sync::mpsc::Sender<ClientActorMessage>,
+    collector: LogCollector,
+    interval: Duration,
+    max_batch: usize,
+    session_id: Uuid,
+) {
+    const INITIAL_BACKOFF: Duration = Duration::from_millis(500);
+    const MAX_BACKOFF: Duration = Duration::from_secs(30);
+
+    let mut ticker = n0_future::time::interval(interval);
+    // After a slow RPC the default `Burst` behavior would fire several
+    // ticks back-to-back; `Delay` waits a full interval from the previous
+    // completed tick.
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    let mut backoff = INITIAL_BACKOFF;
+    loop {
+        ticker.tick().await;
+        let (lines, dropped) = collector.drain(max_batch);
+        if lines.is_empty() && dropped == 0 {
+            backoff = INITIAL_BACKOFF;
+            continue;
+        }
+        let request = PutLogs {
+            session_id,
+            lines,
+            dropped,
+        };
+        let (tx, rx) = oneshot::channel();
+        if message_channel
+            .send(ClientActorMessage::PutLogs { request, done: tx })
+            .await
+            .is_err()
+        {
+            // Mailbox closed only when the actor task has terminated; that
+            // means the entire client is gone and there is nothing to do.
+            debug!("log flush stopped: client actor channel closed");
+            return;
+        }
+        match rx.await {
+            Ok(Ok(())) => {
+                backoff = INITIAL_BACKOFF;
+            }
+            // Either the RPC failed (Ok(Err)) or the actor dropped the
+            // response sender mid-handoff (Err(_)). Both are transient: keep
+            // ticking and back off so the next attempt happens later.
+            other => {
+                debug!(?other, ?backoff, "log flush attempt failed; backing off");
+                n0_future::time::sleep(backoff).await;
+                backoff = (backoff * 2).min(MAX_BACKOFF);
+            }
+        }
+    }
 }
 
 async fn set_name_inner(