refactor: split large modules into directories

crates/rginx-app/src/admin_cli/counters.rs

@@ -0,0 +1,89 @@
+use super::render::print_record;
+use super::socket::{query_admin_socket, unexpected_admin_response};
+use super::*;
+
+pub(super) fn print_admin_counters(config_path: &Path) -> anyhow::Result<()> {
+    match query_admin_socket(config_path, AdminRequest::GetCounters)? {
+        AdminResponse::Counters(counters) => {
+            print_record(
+                "counters",
+                [
+                    (
+                        "downstream_connections_accepted_total",
+                        counters.downstream_connections_accepted.to_string(),
+                    ),
+                    (
+                        "downstream_connections_rejected_total",
+                        counters.downstream_connections_rejected.to_string(),
+                    ),
+                    ("downstream_requests_total", counters.downstream_requests.to_string()),
+                    ("downstream_responses_total", counters.downstream_responses.to_string()),
+                    (
+                        "downstream_responses_1xx_total",
+                        counters.downstream_responses_1xx.to_string(),
+                    ),
+                    (
+                        "downstream_responses_2xx_total",
+                        counters.downstream_responses_2xx.to_string(),
+                    ),
+                    (
+                        "downstream_responses_3xx_total",
+                        counters.downstream_responses_3xx.to_string(),
+                    ),
+                    (
+                        "downstream_responses_4xx_total",
+                        counters.downstream_responses_4xx.to_string(),
+                    ),
+                    (
+                        "downstream_responses_5xx_total",
+                        counters.downstream_responses_5xx.to_string(),
+                    ),
+                    (
+                        "downstream_mtls_authenticated_connections_total",
+                        counters.downstream_mtls_authenticated_connections.to_string(),
+                    ),
+                    (
+                        "downstream_mtls_authenticated_requests_total",
+                        counters.downstream_mtls_authenticated_requests.to_string(),
+                    ),
+                    (
+                        "downstream_mtls_anonymous_requests_total",
+                        counters.downstream_mtls_anonymous_requests.to_string(),
+                    ),
+                    (
+                        "downstream_tls_handshake_failures_total",
+                        counters.downstream_tls_handshake_failures.to_string(),
+                    ),
+                    (
+                        "downstream_tls_handshake_failures_missing_client_cert_total",
+                        counters.downstream_tls_handshake_failures_missing_client_cert.to_string(),
+                    ),
+                    (
+                        "downstream_tls_handshake_failures_unknown_ca_total",
+                        counters.downstream_tls_handshake_failures_unknown_ca.to_string(),
+                    ),
+                    (
+                        "downstream_tls_handshake_failures_bad_certificate_total",
+                        counters.downstream_tls_handshake_failures_bad_certificate.to_string(),
+                    ),
+                    (
+                        "downstream_tls_handshake_failures_certificate_revoked_total",
+                        counters.downstream_tls_handshake_failures_certificate_revoked.to_string(),
+                    ),
+                    (
+                        "downstream_tls_handshake_failures_verify_depth_exceeded_total",
+                        counters
+                            .downstream_tls_handshake_failures_verify_depth_exceeded
+                            .to_string(),
+                    ),
+                    (
+                        "downstream_tls_handshake_failures_other_total",
+                        counters.downstream_tls_handshake_failures_other.to_string(),
+                    ),
+                ],
+            );
+            Ok(())
+        }
+        response => Err(unexpected_admin_response("counters", &response)),
+    }
+}

crates/rginx-app/src/admin_cli/mod.rs

@@ -0,0 +1,62 @@
+pub(super) use std::io::{BufReader, Read, Write};
+pub(super) use std::os::unix::net::UnixStream;
+pub(super) use std::path::Path;
+
+pub(super) use anyhow::{Context, anyhow};
+pub(super) use rginx_runtime::admin::{
+    AdminRequest, AdminResponse, RevisionSnapshot, admin_socket_path_for_config,
+};
+
+use crate::cli::Command;
+pub(super) use crate::cli::{DeltaArgs, SnapshotArgs, SnapshotModuleArg, WaitArgs, WindowArgs};
+
+mod counters;
+mod peers;
+mod render;
+mod snapshot;
+mod socket;
+mod status;
+mod traffic;
+mod upstreams;
+
+pub(crate) fn run_admin_command(config_path: &Path, command: &Command) -> anyhow::Result<bool> {
+    match command {
+        Command::Snapshot(args) => {
+            snapshot::print_admin_snapshot(config_path, args)?;
+            Ok(true)
+        }
+        Command::SnapshotVersion => {
+            snapshot::print_admin_snapshot_version(config_path)?;
+            Ok(true)
+        }
+        Command::Delta(args) => {
+            snapshot::print_admin_delta(config_path, args)?;
+            Ok(true)
+        }
+        Command::Wait(args) => {
+            snapshot::print_admin_wait(config_path, args)?;
+            Ok(true)
+        }
+        Command::Status => {
+            status::print_admin_status(config_path)?;
+            Ok(true)
+        }
+        Command::Counters => {
+            counters::print_admin_counters(config_path)?;
+            Ok(true)
+        }
+        Command::Traffic(args) => {
+            traffic::print_admin_traffic(config_path, args)?;
+            Ok(true)
+        }
+        Command::Peers => {
+            peers::print_admin_peers(config_path)?;
+            Ok(true)
+        }
+        Command::Upstreams(args) => {
+            upstreams::print_admin_upstreams(config_path, args)?;
+            Ok(true)
+        }
+        Command::Check | Command::MigrateNginx(_) => Ok(false),
+    }
+}

crates/rginx-app/src/admin_cli/peers.rs

@@ -0,0 +1,47 @@
+use super::render::print_record;
+use super::socket::{query_admin_socket, unexpected_admin_response};
+use super::*;
+
+pub(super) fn print_admin_peers(config_path: &Path) -> anyhow::Result<()> {
+    match query_admin_socket(config_path, AdminRequest::GetPeerHealth)? {
+        AdminResponse::PeerHealth(upstreams) => {
+            for upstream in upstreams {
+                let upstream_name = upstream.upstream_name.clone();
+                print_record(
+                    "peer_health_upstream",
+                    [
+                        ("upstream", upstream_name.clone()),
+                        ("unhealthy_after_failures", upstream.unhealthy_after_failures.to_string()),
+                        ("cooldown_ms", upstream.cooldown_ms.to_string()),
+                        ("active_health_enabled", upstream.active_health_enabled.to_string()),
+                    ],
+                );
+                for peer in upstream.peers {
+                    print_record(
+                        "peer_health_peer",
+                        [
+                            ("upstream", upstream_name.clone()),
+                            ("peer", peer.peer_url),
+                            ("backup", peer.backup.to_string()),
+                            ("weight", peer.weight.to_string()),
+                            ("available", peer.available.to_string()),
+                            ("passive_failures", peer.passive_consecutive_failures.to_string()),
+                            (
+                                "passive_cooldown_remaining_ms",
+                                peer.passive_cooldown_remaining_ms
+                                    .map(|value| value.to_string())
+                                    .unwrap_or_else(|| "-".to_string()),
+                            ),
+                            ("passive_pending_recovery", peer.passive_pending_recovery.to_string()),
+                            ("active_unhealthy", peer.active_unhealthy.to_string()),
+                            ("active_successes", peer.active_consecutive_successes.to_string()),
+                            ("active_requests", peer.active_requests.to_string()),
+                        ],
+                    );
+                }
+            }
+            Ok(())
+        }
+        response => Err(unexpected_admin_response("peers", &response)),
+    }
+}

crates/rginx-app/src/admin_cli/render.rs

@@ -0,0 +1,89 @@
+pub(super) fn render_last_reload(result: Option<&rginx_http::ReloadResultSnapshot>) -> String {
+    let Some(result) = result else {
+        return "-".to_string();
+    };
+
+    let finished_at = result
+        .finished_at_unix_ms
+        .checked_div(1000)
+        .map(|seconds| seconds.to_string())
+        .unwrap_or_else(|| result.finished_at_unix_ms.to_string());
+
+    match &result.outcome {
+        rginx_http::ReloadOutcomeSnapshot::Success { revision } => {
+            format!(
+                "success revision={revision} active_revision={} finished_at_unix_s={finished_at}",
+                result.active_revision
+            )
+        }
+        rginx_http::ReloadOutcomeSnapshot::Failure { error } => {
+            format!(
+                "failure active_revision={} rollback_preserved_revision={} error={error:?} finished_at_unix_s={finished_at}",
+                result.active_revision,
+                result
+                    .rollback_preserved_revision
+                    .map(|value| value.to_string())
+                    .unwrap_or_else(|| "-".to_string())
+            )
+        }
+    }
+}
+
+pub(super) fn render_reload_active_revision(
+    result: Option<&rginx_http::ReloadResultSnapshot>,
+) -> String {
+    result.map(|result| result.active_revision.to_string()).unwrap_or_else(|| "-".to_string())
+}
+
+pub(super) fn render_reload_rollback_revision(
+    result: Option<&rginx_http::ReloadResultSnapshot>,
+) -> String {
+    result
+        .and_then(|result| result.rollback_preserved_revision)
+        .map(|value| value.to_string())
+        .unwrap_or_else(|| "-".to_string())
+}
+
+pub(super) fn render_reload_tls_certificate_changes(
+    result: Option<&rginx_http::ReloadResultSnapshot>,
+) -> String {
+    let Some(result) = result else {
+        return "-".to_string();
+    };
+    if result.tls_certificate_changes.is_empty() {
+        return "-".to_string();
+    }
+    result.tls_certificate_changes.join(",")
+}
+
+pub(super) fn render_optional_string_list(values: Option<&[String]>) -> String {
+    match values {
+        Some(values) if !values.is_empty() => values.join(","),
+        _ => "-".to_string(),
+    }
+}
+
+pub(super) fn print_record<const N: usize>(kind: &str, fields: [(&str, String); N]) {
+    let mut rendered = String::from("kind=");
+    rendered.push_str(kind);
+    for (key, value) in fields {
+        rendered.push(' ');
+        rendered.push_str(key);
+        rendered.push('=');
+        rendered.push_str(&encode_record_value(&value));
+    }
+    println!("{rendered}");
+}
+
+fn encode_record_value(value: &str) -> String {
+    if !value.is_empty()
+        && value.chars().all(|ch| {
+            ch.is_ascii_alphanumeric()
+                || matches!(ch, '.' | ':' | '/' | '-' | '_' | ',' | '*' | '[' | ']' | '|')
+        })
+    {
+        value.to_string()
+    } else {
+        serde_json::to_string(value).expect("record value should encode as JSON string")
+    }
+}

crates/rginx-app/src/admin_cli/snapshot.rs

@@ -0,0 +1,85 @@
+use super::socket::{query_admin_socket, unexpected_admin_response};
+use super::*;
+
+pub(super) fn print_admin_snapshot(config_path: &Path, args: &SnapshotArgs) -> anyhow::Result<()> {
+    match query_admin_socket(
+        config_path,
+        AdminRequest::GetSnapshot {
+            include: requested_snapshot_modules(&args.include),
+            window_secs: args.window_secs,
+        },
+    )? {
+        AdminResponse::Snapshot(snapshot) => {
+            let rendered = serde_json::to_string_pretty(&snapshot)
+                .context("failed to encode snapshot JSON")?;
+            println!("{rendered}");
+            Ok(())
+        }
+        response => Err(unexpected_admin_response("snapshot", &response)),
+    }
+}
+
+pub(super) fn print_admin_snapshot_version(config_path: &Path) -> anyhow::Result<()> {
+    match query_admin_socket(config_path, AdminRequest::GetSnapshotVersion)? {
+        AdminResponse::SnapshotVersion(snapshot) => {
+            println!("snapshot_version={}", snapshot.snapshot_version);
+            Ok(())
+        }
+        response => Err(unexpected_admin_response("snapshot-version", &response)),
+    }
+}
+
+pub(super) fn print_admin_delta(config_path: &Path, args: &DeltaArgs) -> anyhow::Result<()> {
+    match query_admin_socket(
+        config_path,
+        AdminRequest::GetDelta {
+            since_version: args.since_version,
+            include: requested_snapshot_modules(&args.include),
+            window_secs: args.window_secs,
+        },
+    )? {
+        AdminResponse::Delta(delta) => {
+            let rendered =
+                serde_json::to_string_pretty(&delta).context("failed to encode delta JSON")?;
+            println!("{rendered}");
+            Ok(())
+        }
+        response => Err(unexpected_admin_response("delta", &response)),
+    }
+}
+
+pub(super) fn print_admin_wait(config_path: &Path, args: &WaitArgs) -> anyhow::Result<()> {
+    match query_admin_socket(
+        config_path,
+        AdminRequest::WaitForSnapshotChange {
+            since_version: args.since_version,
+            timeout_ms: args.timeout_ms,
+        },
+    )? {
+        AdminResponse::SnapshotVersion(snapshot) => {
+            println!("snapshot_version={}", snapshot.snapshot_version);
+            Ok(())
+        }
+        response => Err(unexpected_admin_response("wait", &response)),
+    }
+}
+
+fn requested_snapshot_modules(
+    include: &[SnapshotModuleArg],
+) -> Option<Vec<rginx_http::SnapshotModule>> {
+    if include.is_empty() {
+        return None;
+    }
+
+    Some(include.iter().copied().map(snapshot_module).collect())
+}
+
+fn snapshot_module(module: SnapshotModuleArg) -> rginx_http::SnapshotModule {
+    match module {
+        SnapshotModuleArg::Status => rginx_http::SnapshotModule::Status,
+        SnapshotModuleArg::Counters => rginx_http::SnapshotModule::Counters,
+        SnapshotModuleArg::Traffic => rginx_http::SnapshotModule::Traffic,
+        SnapshotModuleArg::PeerHealth => rginx_http::SnapshotModule::PeerHealth,
+        SnapshotModuleArg::Upstreams => rginx_http::SnapshotModule::Upstreams,
+    }
+}