Add an echorpc control CLI over gRPC: version/status/rehash live, start/stop/restart via systemctl

build.rs

@@ -9,7 +9,9 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     // 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();

echorpc

@@ -0,0 +1 @@
+target/release/echo

proto/echo.proto

@@ -155,3 +155,26 @@ message StatsResponse {
   map<string, uint64> 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;
+}

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 <cmd>` (a symlink to the echo binary) or
+//! `echo ctl <cmd>`. 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<String>) -> 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<tonic::transport::Channel>, 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<pb::AdminRequest> {
+    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 <command> [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"
+    );
+}

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

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<Mutex<Engine>>;
@@ -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<AdminRequest>) -> Result<Response<VersionResponse>, 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<AdminRequest>) -> Result<Response<AdminResponse>, 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<LogEntry>) {
@@ -425,6 +456,7 @@ pub async fn run(engine: Shared, cfg: GrpcCfg, outbound: broadcast::Sender<LogEn
     let has_tls = cfg.tls.is_some();
     let accounts_svc = AccountsService { engine: engine.clone(), token: cfg.token.clone() };
     let stats_svc = StatsService { engine: engine.clone(), token: cfg.token.clone() };
+    let admin_svc = AdminService { engine: engine.clone(), token: cfg.token.clone() };
     let directory_svc = DirectoryService { engine, outbound, token: cfg.token };
     let mut server = Server::builder();
     if let Some(tls) = &cfg.tls {
@@ -438,11 +470,12 @@ pub async fn run(engine: Shared, cfg: GrpcCfg, outbound: broadcast::Sender<LogEn
             Err(e) => 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
     {

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 <anope.json> <out.log> [node-name]` builds a
     // fresh event log from an Anope db_json and exits, touching nothing else.
     let argv: Vec<String> = std::env::args().collect();
-    // `echo --version` / `-V` prints the build banner (version, git revision, build
-    // date, toolchain, target) and exits.
+    // `echorpc <cmd>` (a symlink to this binary) or `echo ctl <cmd>` 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()));