feat(flux): add coold heartbeats and logs

Send Flux ping frames to connected coold streams, track pong timeouts,
publish coold reachability updates, and route coold.logs requests.

Author: andrasbacsai

coold/src/grpc/client.rs

@@ -14,7 +14,7 @@ use tracing::{info, warn};
 use crate::config::{Config, VERSION};
 use crate::grpc::handlers::handle;
 use crate::grpc::proto::{
-    agent_client::AgentClient, client_msg, ClientMsg, Hello, ResourceStatusUpdate,
+    agent_client::AgentClient, client_msg, server_msg, ClientMsg, Hello, Pong, ResourceStatusUpdate,
 };
 use crate::podman::PodmanClient;
 
@@ -212,6 +212,7 @@ fn primitive_capabilities() -> Vec<String> {
         "firewall.revoke",
         "firewall.list",
         "firewall.reconcile",
+        "coold.logs",
     ]
     .into_iter()
     .map(str::to_string)
@@ -336,6 +337,11 @@ async fn connect_and_serve(
             continue;
         };
 
+        if matches!(command, server_msg::Command::Ping(_)) {
+            tx.send(pong_for(&request_id)).await.context("send Pong")?;
+            continue;
+        }
+
         let tx = tx.clone();
         let podman = podman.clone();
         tokio::spawn(async move {
@@ -348,6 +354,14 @@ async fn connect_and_serve(
     Ok(())
 }
 
+fn pong_for(request_id: &str) -> ClientMsg {
+    ClientMsg {
+        payload: Some(client_msg::Payload::Pong(Pong {
+            request_id: request_id.to_string(),
+        })),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::load_host_jwt;
@@ -375,6 +389,18 @@ mod tests {
         assert!(msg.contains("insecure perms"), "got: {msg}");
     }
 
+    #[test]
+    fn builds_pong_for_ping_request_id() {
+        let msg = super::pong_for("ping-1");
+
+        match msg.payload {
+            Some(super::client_msg::Payload::Pong(pong)) => {
+                assert_eq!(pong.request_id, "ping-1");
+            }
+            other => panic!("expected pong payload, got {other:?}"),
+        }
+    }
+
     #[tokio::test]
     async fn rejects_group_readable() {
         let dir = tempfile::tempdir().unwrap();

coold/src/grpc/handlers.rs

@@ -11,9 +11,10 @@ use crate::grpc::proto::{
     client_msg, response, server_msg, ApplyIngressResp, ClientMsg, ContainerSummary,
     ContainersCreateResp, ContainersDeleteResp, ContainersExecResp, ContainersHealthcheckRunResp,
     ContainersInspectResp, ContainersListResp, ContainersLogsResp, ContainersRestartResp,
-    ContainersStartResp, ContainersStopResp, Error, FirewallAllowResp, FirewallListResp,
-    FirewallReconcileResp, FirewallRevokeResp, FirewallRule as ProtoFirewallRule, ImageSummary,
-    ImagesDeleteResp, ImagesListResp, ImagesPullResp, IngressAppConfig, Response, StopIngressResp,
+    ContainersStartResp, ContainersStopResp, CooldLogsResp, Error, FirewallAllowResp,
+    FirewallListResp, FirewallReconcileResp, FirewallRevokeResp, FirewallRule as ProtoFirewallRule,
+    ImageSummary, ImagesDeleteResp, ImagesListResp, ImagesPullResp, IngressAppConfig, Response,
+    StopIngressResp,
 };
 use crate::podman::client::{CreateContainerInput, CreatePortMapping};
 use crate::podman::PodmanClient;
@@ -25,6 +26,9 @@ pub async fn handle(
     tx: mpsc::Sender<ClientMsg>,
 ) {
     match command {
+        server_msg::Command::Ping(_) => {
+            debug!(%request_id, "ping command reached handler after fast-path; ignoring");
+        }
         server_msg::Command::ImagesPull(req) => {
             let body = match podman.pull_image(&req.reference).await {
                 Ok((digest, output)) => {
@@ -223,6 +227,20 @@ pub async fn handle(
             )
             .await;
         }
+        server_msg::Command::CooldLogs(req) => {
+            let body = match coold_logs(req.tail).await {
+                Ok(output) => response::Body::CooldLogs(CooldLogsResp { output }),
+                Err(e) => error_body(e),
+            };
+            send_response(
+                &tx,
+                Response {
+                    request_id,
+                    body: Some(body),
+                },
+            )
+            .await;
+        }
         server_msg::Command::ContainersExec(req) => {
             let body = match podman.exec_container(&req.id, req.command).await {
                 Ok((exit_code, output)) => {
@@ -1192,6 +1210,43 @@ async fn run_command(command: &mut Command) -> Result<String> {
     Ok(combined)
 }
 
+async fn coold_logs(tail: u32) -> Result<String> {
+    let tail = tail.clamp(1, 1000).to_string();
+    let output = Command::new("journalctl")
+        .args([
+            "--unit",
+            "coold",
+            "--no-pager",
+            "--output",
+            "short-iso",
+            "--lines",
+            &tail,
+        ])
+        .output()
+        .await
+        .context("read coold journal logs")?;
+
+    let combined = String::from_utf8_lossy(&output.stdout)
+        .lines()
+        .chain(String::from_utf8_lossy(&output.stderr).lines())
+        .filter(|line| !line.is_empty())
+        .collect::<Vec<_>>()
+        .join("\n");
+
+    if !output.status.success() {
+        return Err(anyhow!(
+            "{}",
+            if combined.is_empty() {
+                format!("journalctl exited with {}", output.status)
+            } else {
+                combined
+            }
+        ));
+    }
+
+    Ok(combined)
+}
+
 async fn containers_list(podman: &PodmanClient) -> Result<ContainersListResp> {
     let containers = podman.containers_list().await?;
 

flux/src/config.rs

@@ -99,6 +99,22 @@ pub struct Config {
         default_value = "10"
     )]
     pub laravel_heartbeat_interval_secs: u64,
+
+    /// Seconds between Flux ping frames sent to each connected coold stream.
+    #[arg(
+        long,
+        env = "COOLIFY_FLUX_COOLD_PING_INTERVAL_SECS",
+        default_value = "10"
+    )]
+    pub coold_ping_interval_secs: u64,
+
+    /// Seconds without a coold pong before Flux marks that server unreachable.
+    #[arg(
+        long,
+        env = "COOLIFY_FLUX_COOLD_PONG_TIMEOUT_SECS",
+        default_value = "45"
+    )]
+    pub coold_pong_timeout_secs: u64,
 }
 
 impl Config {
@@ -127,6 +143,14 @@ mod tests {
         );
     }
 
+    #[test]
+    fn defaults_coold_heartbeat_timing() {
+        let config = Config::parse_from(["flux", "--grpc-bind", "127.0.0.1:6443"]);
+
+        assert_eq!(config.coold_ping_interval_secs, 10);
+        assert_eq!(config.coold_pong_timeout_secs, 45);
+    }
+
     #[test]
     fn rejects_flux_file_log_path_override() {
         let result = Config::try_parse_from([

flux/src/envelope.rs

@@ -139,6 +139,11 @@ pub enum CommandPayload {
     },
     #[serde(rename = "firewall.reconcile")]
     FirewallReconcile,
+    #[serde(rename = "coold.logs")]
+    CooldLogs {
+        #[serde(default = "default_coold_logs_tail")]
+        tail: u32,
+    },
 }
 
 #[derive(Debug, Deserialize)]
@@ -153,6 +158,10 @@ fn default_logs_stdout() -> bool {
     true
 }
 
+fn default_coold_logs_tail() -> u32 {
+    200
+}
+
 #[derive(Debug, Deserialize)]
 pub struct IngressAppConfig {
     pub name: String,
@@ -293,6 +302,9 @@ impl ResponseBody {
             Some(Body::FirewallReconcile(r)) => Some(ResponseBody::Ok {
                 data: serde_json::json!({ "output": r.output }),
             }),
+            Some(Body::CooldLogs(r)) => Some(ResponseBody::Ok {
+                data: serde_json::json!({ "output": r.output }),
+            }),
             Some(Body::Error(e)) => Some(ResponseBody::Error {
                 code: e.code,
                 message: e.message,
@@ -312,7 +324,7 @@ mod tests {
         response, ContainerSummary, ContainersCreateResp, ContainersDeleteResp, ContainersExecResp,
         ContainersHealthcheckRunResp, ContainersInspectResp, ContainersListResp,
         ContainersLogsResp, ContainersRestartResp, ContainersStartResp, ContainersStopResp,
-        ImageSummary, ImagesDeleteResp, ImagesListResp, ImagesPullResp,
+        CooldLogsResp, ImageSummary, ImagesDeleteResp, ImagesListResp, ImagesPullResp,
     };
 
     fn data_for(body: response::Body) -> serde_json::Value {

flux/src/main.rs

@@ -37,6 +37,7 @@ async fn main() -> Result<()> {
         unix_bridge::run(config.clone(), streams.clone(), pending.clone()),
         pending_sweeper::run(config.clone(), pending.clone()),
         registry::heartbeat_loop(config.clone(), streams.clone()),
+        coold_heartbeat::run(config.clone(), streams.clone()),
     )?;
 
     Ok(())
@@ -262,6 +263,19 @@ mod grpc_server {
             let registry = self.registry.clone();
             let resource_status = self.resource_status.clone();
             let host_id_clone = host_id.clone();
+            {
+                let publisher = resource_status.clone();
+                let host_id = host_id.clone();
+                tokio::spawn(async move {
+                    publisher
+                        .publish(coold_status_update(
+                            &host_id,
+                            "installed",
+                            "coold stream connected.",
+                        ))
+                        .await;
+                });
+            }
             if let Some(registry) = registry.clone() {
                 let host_id = host_id.clone();
                 let capabilities = hello.capabilities.clone();
@@ -297,6 +311,23 @@ mod grpc_server {
                                 publisher.publish(update).await;
                             });
                         }
+                        Ok(ClientMsg {
+                            payload: Some(client_msg::Payload::Pong(_)),
+                        }) => {
+                            if streams.touch(&host_id_clone) {
+                                let publisher = resource_status.clone();
+                                let host_id = host_id_clone.clone();
+                                tokio::spawn(async move {
+                                    publisher
+                                        .publish(coold_status_update(
+                                            &host_id,
+                                            "installed",
+                                            "coold heartbeat restored.",
+                                        ))
+                                        .await;
+                                });
+                            }
+                        }
                         Ok(ClientMsg {
                             payload: Some(client_msg::Payload::Hello(_)),
                         }) => warn!(host_id = %host_id_clone, "duplicate Hello ignored"),
@@ -310,6 +341,13 @@ mod grpc_server {
                 }
                 info!(host_id = %host_id_clone, "coold stream disconnected");
                 streams.remove(&host_id_clone);
+                resource_status
+                    .publish(coold_status_update(
+                        &host_id_clone,
+                        "unreachable",
+                        "coold stream disconnected.",
+                    ))
+                    .await;
                 if let Some(registry) = registry {
                     if let Err(e) = registry.disconnect(&host_id_clone, disconnect_reason).await {
                         warn!(
@@ -351,9 +389,24 @@ mod grpc_server {
         pending.deliver(&request_id, data);
     }
 
+    pub(super) fn coold_status_update(
+        host_id: &str,
+        status: &str,
+        status_message: &str,
+    ) -> coolify_proto::agent::v1::ResourceStatusUpdate {
+        coolify_proto::agent::v1::ResourceStatusUpdate {
+            resource_type: "server".into(),
+            host_id: host_id.into(),
+            container_id: String::new(),
+            container_name: String::new(),
+            status: status.into(),
+            status_message: status_message.into(),
+        }
+    }
+
     #[cfg(test)]
     mod tests {
-        use super::{advertised_capability_not_granted, validate_bind};
+        use super::{advertised_capability_not_granted, coold_status_update, validate_bind};
         use std::net::SocketAddr;
 
         fn parse(s: &str) -> SocketAddr {
@@ -424,6 +477,68 @@ mod grpc_server {
                 Some("ingress.apply")
             );
         }
+
+        #[test]
+        fn builds_server_status_update_for_coold_liveness() {
+            let update =
+                coold_status_update("100.64.0.5", "unreachable", "coold heartbeat timed out.");
+
+            assert_eq!(update.resource_type, "server");
+            assert_eq!(update.host_id, "100.64.0.5");
+            assert_eq!(update.status, "unreachable");
+            assert_eq!(update.status_message, "coold heartbeat timed out.");
+        }
+    }
+}
+
+mod coold_heartbeat {
+    use anyhow::Result;
+    use coolify_proto::agent::v1::{server_msg, Ping, ServerMsg};
+    use tracing::warn;
+    use uuid::Uuid;
+
+    use crate::{
+        config::Config, grpc_server::coold_status_update, resource_status::ResourceStatusPublisher,
+        state::Streams,
+    };
+
+    pub async fn run(config: Config, streams: Streams) -> Result<()> {
+        let publisher = ResourceStatusPublisher::new(
+            config.laravel_api_url.clone(),
+            config.laravel_api_token.clone(),
+        );
+        let mut ticker = tokio::time::interval(std::time::Duration::from_secs(
+            config.coold_ping_interval_secs.max(1),
+        ));
+        let timeout = std::time::Duration::from_secs(config.coold_pong_timeout_secs.max(1));
+
+        loop {
+            ticker.tick().await;
+
+            for host_id in streams.mark_stale(timeout) {
+                publisher
+                    .publish(coold_status_update(
+                        &host_id,
+                        "unreachable",
+                        "coold heartbeat timed out.",
+                    ))
+                    .await;
+            }
+
+            for (host_id, tx) in streams.ping_targets() {
+                let request_id = Uuid::new_v4().to_string();
+                if tx
+                    .send(ServerMsg {
+                        request_id,
+                        command: Some(server_msg::Command::Ping(Ping {})),
+                    })
+                    .await
+                    .is_err()
+                {
+                    warn!(%host_id, "coold heartbeat ping send failed");
+                }
+            }
+        }
     }
 }