fix(media-proxy): probe health with the binary not /dev/tcp (#2324)

fluxer_media_proxy/Dockerfile

@@ -80,7 +80,7 @@ WORKDIR /var/lib/fluxer
 EXPOSE 8080
 
 HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
-    CMD ["sh", "-c", "exec 3<>/dev/tcp/127.0.0.1/${FLUXER_MEDIA_PROXY_PORT:-8080} && printf 'GET /_health HTTP/1.0\\r\\n\\r\\n' >&3 && grep -q '200' <&3"]
+    CMD ["/usr/local/bin/fluxer-media-proxy", "healthcheck"]
 
 ENTRYPOINT ["/usr/bin/tini", "--"]
 CMD ["/usr/local/bin/fluxer-media-proxy"]

fluxer_media_proxy/src/cli.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use crate::config::{Config, DeploymentMode, StorageBackend};
-use clap::{ArgAction, Parser, ValueEnum};
+use clap::{ArgAction, Parser, Subcommand, ValueEnum};
 
 #[derive(Debug, Parser)]
 #[command(name = "fluxer-media-proxy", disable_help_subcommand = true)]
@@ -23,6 +23,14 @@ pub struct Args {
 
     #[arg(long = "read-only", action = ArgAction::SetTrue)]
     pub read_only: bool,
+
+    #[command(subcommand)]
+    pub command: Option<Command>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Subcommand)]
+pub enum Command {
+    Healthcheck,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
@@ -137,6 +145,22 @@ mod tests {
         assert!(cfg.read_only);
     }
 
+    #[test]
+    fn cli_parses_healthcheck_subcommand() {
+        assert_eq!(
+            Some(Command::Healthcheck),
+            Args::try_parse_from(["fluxer-media-proxy", "healthcheck"])
+                .unwrap()
+                .command
+        );
+        assert!(
+            Args::try_parse_from(["fluxer-media-proxy"])
+                .unwrap()
+                .command
+                .is_none()
+        );
+    }
+
     #[test]
     fn cli_rejects_empty_bind_host() {
         let args = Args::try_parse_from(["fluxer-media-proxy", "--bind-host", ""]).unwrap();

fluxer_media_proxy/src/healthcheck.rs

@@ -0,0 +1,100 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use anyhow::Context as _;
+use std::{
+    env,
+    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
+    time::Duration,
+};
+
+pub async fn run() -> anyhow::Result<()> {
+    let addr = target(
+        env::var("FLUXER_MEDIA_PROXY_HOST").ok().as_deref(),
+        env::var("FLUXER_MEDIA_PROXY_PORT").ok().as_deref(),
+    )?;
+    probe(addr).await
+}
+
+fn target(host: Option<&str>, port: Option<&str>) -> anyhow::Result<SocketAddr> {
+    let host = host
+        .map(str::trim)
+        .filter(|host| !host.is_empty())
+        .unwrap_or("127.0.0.1");
+    let ip = host
+        .parse::<IpAddr>()
+        .with_context(|| format!("FLUXER_MEDIA_PROXY_HOST is not an IP address: {host}"))?;
+    let ip = match ip {
+        IpAddr::V4(ip) if ip.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
+        IpAddr::V6(ip) if ip.is_unspecified() => IpAddr::V6(Ipv6Addr::LOCALHOST),
+        ip => ip,
+    };
+    let port = match port.map(str::trim).filter(|port| !port.is_empty()) {
+        Some(port) => port
+            .parse::<u16>()
+            .with_context(|| format!("FLUXER_MEDIA_PROXY_PORT is not a port number: {port}"))?,
+        None => 8080,
+    };
+    Ok(SocketAddr::new(ip, port))
+}
+
+async fn probe(addr: SocketAddr) -> anyhow::Result<()> {
+    let client = reqwest::Client::builder()
+        .connect_timeout(Duration::from_millis(500))
+        .timeout(Duration::from_millis(2_000))
+        .no_proxy()
+        .build()?;
+    let status = client
+        .get(format!("http://{addr}/_health"))
+        .send()
+        .await
+        .with_context(|| format!("health request to {addr} failed"))?
+        .status();
+    anyhow::ensure!(
+        status == reqwest::StatusCode::OK,
+        "health returned {status}"
+    );
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn target_defaults_to_loopback() {
+        assert_eq!(
+            SocketAddr::from(([127, 0, 0, 1], 8080)),
+            target(Some("0.0.0.0"), None).unwrap()
+        );
+        assert_eq!(
+            SocketAddr::from(([127, 0, 0, 1], 9000)),
+            target(None, Some("9000")).unwrap()
+        );
+        assert_eq!(
+            "[::1]:8080".parse::<SocketAddr>().unwrap(),
+            target(Some("::"), Some("")).unwrap()
+        );
+        assert!(target(Some("0.0.0.0"), Some("nope")).is_err());
+    }
+
+    #[tokio::test]
+    async fn probe_accepts_ok() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = axum::Router::new().route("/_health", axum::routing::get(async || "OK"));
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        probe(addr).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn probe_rejects_non_ok() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = axum::Router::new().route(
+            "/_health",
+            axum::routing::get(async || axum::http::StatusCode::SERVICE_UNAVAILABLE),
+        );
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        assert!(probe(addr).await.is_err());
+    }
+}

fluxer_media_proxy/src/lib.rs

@@ -11,6 +11,7 @@ pub mod config;
 pub mod constants;
 pub mod disposition;
 pub mod external_path;
+pub mod healthcheck;
 pub mod http_client;
 pub mod http_headers;
 pub mod media_process;

fluxer_media_proxy/src/main.rs

@@ -1,17 +1,21 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use clap::Parser;
-use fluxer_media_proxy::{cli, run};
+use fluxer_media_proxy::{cli, healthcheck, run};
 use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt};
 
 #[tokio::main(flavor = "multi_thread")]
 async fn main() -> anyhow::Result<()> {
+    let args = cli::Args::parse();
+    if matches!(args.command, Some(cli::Command::Healthcheck)) {
+        return healthcheck::run().await;
+    }
+
     tracing_subscriber::registry()
         .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
         .with(tracing_subscriber::fmt::layer().json())
         .init();
 
-    let args = cli::Args::parse();
     let cfg = cli::load_config(&args)?;
     run(cfg).await
 }