From 829a3916fd8b375aed48f68cefa782edae87ec1c Mon Sep 17 00:00:00 2001 From: Jean Date: Mon, 20 Jul 2026 15:28:00 +0000 Subject: [PATCH] Add an echorpc control CLI over gRPC: version/status/rehash live, start/stop/restart via systemctl --- build.rs | 4 +- echorpc | 1 + proto/echo.proto | 23 +++++++++++ src/control.rs | 102 ++++++++++++++++++++++++++++++++++++++++++++++ src/engine/mod.rs | 13 ++++++ src/grpc.rs | 45 +++++++++++++++++--- src/main.rs | 19 ++++++++- 7 files changed, 198 insertions(+), 9 deletions(-) create mode 120000 echorpc create mode 100644 src/control.rs diff --git a/build.rs b/build.rs index 86a2b6d..bf4149b 100644 --- a/build.rs +++ b/build.rs @@ -9,7 +9,9 @@ fn main() -> Result<(), Box> { // Django (or any other consumer) generates its own stub from proto/echo.proto. tonic_build::configure() .build_server(true) - .build_client(false) + // The `echorpc` control CLI (in this same binary) dials the Admin service, + // so we need the client stubs too. Django generates its own from the proto. + .build_client(true) .compile_protos(&["proto/echo.proto"], &["proto"])?; println!("cargo:rerun-if-changed=proto/echo.proto"); emit_build_info(); diff --git a/echorpc b/echorpc new file mode 120000 index 0000000..dc7ee44 --- /dev/null +++ b/echorpc @@ -0,0 +1 @@ +target/release/echo \ No newline at end of file diff --git a/proto/echo.proto b/proto/echo.proto index 3625dcf..7be4f74 100644 --- a/proto/echo.proto +++ b/proto/echo.proto @@ -155,3 +155,26 @@ message StatsResponse { map counters = 1; uint64 collected_at = 2; // unix seconds when this snapshot was taken } + +// Local admin/control surface, for the `echorpc` CLI: read the running daemon's +// build + liveness, and reload its config live. Process lifecycle +// (start/stop/restart) is deliberately NOT here — you can't RPC a daemon that +// isn't running; that's systemd's job. +service Admin { + rpc Version(AdminRequest) returns (VersionResponse); + rpc Rehash(AdminRequest) returns (AdminResponse); +} + +message AdminRequest {} +message VersionResponse { + string version = 1; + string revision = 2; + string built = 3; + string rustc = 4; + uint64 uptime_secs = 5; // 0 = not linked to the uplink yet + bool linked = 6; +} +message AdminResponse { + bool ok = 1; + string message = 2; +} diff --git a/src/control.rs b/src/control.rs new file mode 100644 index 0000000..4765f5b --- /dev/null +++ b/src/control.rs @@ -0,0 +1,102 @@ +//! `echorpc` — a small control CLI that drives the running daemon over its gRPC +//! Admin service (`version`/`status`/`rehash`) and hands process lifecycle +//! (`start`/`stop`/`restart`) to systemd (you can't RPC a daemon that isn't +//! running). Invoked as `echorpc ` (a symlink to the echo binary) or +//! `echo ctl `. An optional trailing arg overrides the config path +//! (default `config.toml` in the working directory). + +use crate::grpc::pb; + +pub async fn run(args: Vec) -> anyhow::Result<()> { + let verb = args.first().map(String::as_str).unwrap_or(""); + let config_path = args.get(1).cloned().unwrap_or_else(|| "config.toml".to_string()); + match verb { + // Process lifecycle is systemd's job — a dead daemon has no gRPC to answer. + "start" | "stop" | "restart" => { + let status = std::process::Command::new("systemctl").arg(verb).arg("echo").status()?; + std::process::exit(status.code().unwrap_or(1)); + } + "version" | "status" => version(&config_path).await, + "rehash" => rehash(&config_path).await, + "" | "help" | "-h" | "--help" => { + print_usage(); + Ok(()) + } + other => { + eprintln!("echorpc: unknown command '{other}'\n"); + print_usage(); + std::process::exit(2); + } + } +} + +async fn connect(config_path: &str) -> anyhow::Result<(pb::admin_client::AdminClient, String)> { + let cfg = crate::config::Config::load(config_path)?; + let grpc = cfg + .grpc + .ok_or_else(|| anyhow::anyhow!("[grpc] is not configured in {config_path}; the control API needs it"))?; + let scheme = if grpc.tls.is_some() { "https" } else { "http" }; + let endpoint = format!("{scheme}://{}", grpc.bind); + let client = pb::admin_client::AdminClient::connect(endpoint) + .await + .map_err(|e| anyhow::anyhow!("cannot reach echo at {} ({e}); is the daemon running?", grpc.bind))?; + Ok((client, grpc.token)) +} + +// Every Admin RPC carries the same bearer token the daemon's other gRPC clients use. +fn authed(token: &str) -> tonic::Request { + let mut req = tonic::Request::new(pb::AdminRequest {}); + req.metadata_mut() + .insert("authorization", format!("Bearer {token}").parse().expect("valid header value")); + req +} + +async fn version(config_path: &str) -> anyhow::Result<()> { + let (mut client, token) = connect(config_path).await?; + let v = client.version(authed(&token)).await?.into_inner(); + let rustc = v.rustc.strip_prefix("rustc ").unwrap_or(&v.rustc); + let (status, uptime) = if v.linked { + ("linked", humanize(v.uptime_secs)) + } else { + ("starting (not linked to the uplink yet)", "—".to_string()) + }; + println!("echo {} ({})", v.version, v.revision); + println!(" built {}", v.built); + println!(" rustc {rustc}"); + println!(" status {status}"); + println!(" uptime {uptime}"); + Ok(()) +} + +async fn rehash(config_path: &str) -> anyhow::Result<()> { + let (mut client, token) = connect(config_path).await?; + let r = client.rehash(authed(&token)).await?.into_inner(); + println!("{} {}", if r.ok { "✓" } else { "✗" }, r.message); + Ok(()) +} + +fn humanize(secs: u64) -> String { + let (d, h, m) = (secs / 86400, (secs % 86400) / 3600, (secs % 3600) / 60); + let mut out = Vec::new(); + if d > 0 { + out.push(format!("{d}d")); + } + if h > 0 { + out.push(format!("{h}h")); + } + if m > 0 || out.is_empty() { + out.push(format!("{m}m")); + } + out.join(" ") +} + +fn print_usage() { + eprintln!( + "usage: echorpc [config.toml]\n\n\ + talks to the running daemon (gRPC):\n \ + version, status the running build, uptime, and uplink state\n \ + rehash reload config.toml live (no restart)\n\n\ + process lifecycle (systemctl):\n \ + start, stop, restart\n" + ); +} diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 001da68..9d21cfc 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -950,6 +950,19 @@ impl Engine { self.config_path = path.into(); } + // Whether our pseudo-clients have signed on to the uplink (a fresh burst sets + // `services_signon`), and for how long — for the Admin/`echorpc` status view. + pub fn linked(&self) -> bool { + self.services_signon > 0 + } + pub fn uptime_secs(&self) -> u64 { + if self.services_signon == 0 { + 0 + } else { + self.now_secs().saturating_sub(self.services_signon) + } + } + // Re-read config.toml and apply the settings that are safe to change while // linked, reporting the outcome to the oper who ran REHASH. A parse error // keeps the running configuration untouched (validate-or-keep, like an ircd diff --git a/src/grpc.rs b/src/grpc.rs index 95656e2..00001fa 100644 --- a/src/grpc.rs +++ b/src/grpc.rs @@ -21,16 +21,17 @@ pub mod pb { } use pb::accounts_server::{Accounts, AccountsServer}; +use pb::admin_server::{Admin, AdminServer}; use pb::directory_server::{Directory, DirectoryServer}; use pb::replication_event::Kind; use pb::stats_server::{Stats, StatsServer}; use pb::{ AccountDropped, AccountEmailSet, AccountRecord, AccountRegistered, AccountReply, AccountVerified, - AuthenticateReply, AuthenticateRequest, ChannelDescSet, ChannelDropped, ChannelFounderSet, ChannelRecord, - ChannelRegistered, ConfirmRequest, DropRequest, ForceLogoutReply, ForceLogoutRequest, GroupNickRequest, - NickGrouped, NickUngrouped, ProvisionRequest, RegisterRequest, ReplicationEvent, SetEmailRequest, SetPasswordRequest, - SnapshotRequest, SnapshotResponse, Status as PbStatus, StatsRequest, StatsResponse, SubscribeRequest, - UngroupNickRequest, + AdminRequest, AdminResponse, AuthenticateReply, AuthenticateRequest, ChannelDescSet, ChannelDropped, + ChannelFounderSet, ChannelRecord, ChannelRegistered, ConfirmRequest, DropRequest, ForceLogoutReply, + ForceLogoutRequest, GroupNickRequest, NickGrouped, NickUngrouped, ProvisionRequest, RegisterRequest, + ReplicationEvent, SetEmailRequest, SetPasswordRequest, SnapshotRequest, SnapshotResponse, + Status as PbStatus, StatsRequest, StatsResponse, SubscribeRequest, UngroupNickRequest, VersionResponse, }; type Shared = Arc>; @@ -409,6 +410,36 @@ impl Stats for StatsService { } } +struct AdminService { + engine: Shared, + token: String, +} + +#[tonic::async_trait] +impl Admin for AdminService { + async fn version(&self, req: Request) -> Result, Status> { + authorize(&req, &self.token)?; + let e = self.engine.lock().await; + Ok(Response::new(VersionResponse { + version: crate::version::VERSION.to_string(), + revision: crate::version::revision(), + built: crate::version::built(), + rustc: crate::version::RUSTC.to_string(), + uptime_secs: e.uptime_secs(), + linked: e.linked(), + })) + } + + async fn rehash(&self, req: Request) -> Result, Status> { + authorize(&req, &self.token)?; + // Same reload the OperServ REHASH path runs; the returned staff-feed notice + // has no oper to reach over gRPC, so it's dropped (the reload is logged). + let _ = self.engine.lock().await.rehash("gRPC control", ""); + tracing::info!("configuration reloaded via gRPC control"); + Ok(Response::new(AdminResponse { ok: true, message: "configuration reloaded".to_string() })) + } +} + // Start the listener, if configured. Absent [grpc] in config.toml = no-op, same // pattern as gossip being optional. pub async fn run(engine: Shared, cfg: GrpcCfg, outbound: broadcast::Sender) { @@ -425,6 +456,7 @@ pub async fn run(engine: Shared, cfg: GrpcCfg, outbound: broadcast::Sender return tracing::error!(%e, "grpc TLS setup failed"), }; } - tracing::info!(%addr, tls = has_tls, "grpc directory + accounts + stats API listening"); + tracing::info!(%addr, tls = has_tls, "grpc directory + accounts + stats + admin API listening"); if let Err(e) = server .add_service(DirectoryServer::new(directory_svc)) .add_service(AccountsServer::new(accounts_svc)) .add_service(StatsServer::new(stats_svc)) + .add_service(AdminServer::new(admin_svc)) .serve(addr) .await { diff --git a/src/main.rs b/src/main.rs index 4744090..5fa9ae9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ // crates (echo-nickserv, echo-chanserv, echo-inspircd), each depending // only on the echo-api SDK. mod config; +mod control; mod dict; mod engine; mod gossip; @@ -58,8 +59,22 @@ async fn main() -> Result<()> { // One-off migration: `echo import [node-name]` builds a // fresh event log from an Anope db_json and exits, touching nothing else. let argv: Vec = std::env::args().collect(); - // `echo --version` / `-V` prints the build banner (version, git revision, build - // date, toolchain, target) and exits. + // `echorpc ` (a symlink to this binary) or `echo ctl ` runs the + // control CLI against the daemon's gRPC Admin API instead of starting up. + // Checked before `version` so `echorpc version` hits the live daemon, not the + // static build banner. + let arg0 = std::path::Path::new(argv.first().map(String::as_str).unwrap_or("")) + .file_name() + .and_then(std::ffi::OsStr::to_str) + .unwrap_or(""); + if arg0 == "echorpc" || arg0 == "echoctl" { + return control::run(argv[1..].to_vec()).await; + } + if argv.get(1).map(String::as_str) == Some("ctl") { + return control::run(argv.get(2..).unwrap_or(&[]).to_vec()).await; + } + // `echo --version` / `-V` / `echo version` prints the build banner (version, + // git revision, build date, toolchain, target) and exits. if matches!(argv.get(1).map(String::as_str), Some("--version") | Some("-V") | Some("version")) { use std::io::IsTerminal; println!("{}", version::banner(std::io::stdout().is_terminal()));