⎇ 324a0c2d5dcacc219bceac40c382c51adf364043
crates/apps/shipote-cli/src/main.rs
Historia
Quién tocó qué
crudo
feat(shipote): multi-core CPU% + quota report + restart-on-failure (fase K)
1//! `shipote` — CLI de administración del daemon.
2
3use anyhow::{anyhow, Context, Result};
4use clap::{Parser, Subcommand};
5use shipote_card::{load_pipeline_spec, load_workspace_spec, WorkspaceId};
6use shipote_protocol::{default_socket_path, read_frame, write_frame, Request, Response};
7use std::path::PathBuf;
8use tokio::net::UnixStream;
9use ulid::Ulid;
10
11#[derive(Parser, Debug)]
12#[command(name = "shipote", version, about = "Administración de shipote-daemon")]
13struct Cli {
14 /// Path al socket del daemon. Default: $XDG_RUNTIME_DIR/shipote.sock.
15 #[arg(long, global = true)]
16 socket: Option<PathBuf>,
17
18 #[command(subcommand)]
19 cmd: Cmd,
20}
21
22#[derive(Subcommand, Debug)]
23enum Cmd {
24 /// Health-check del daemon.
25 Ping,
26
27 /// Capacidades runtime detectadas por el daemon.
28 Caps,
29
30 /// Operaciones sobre Workspaces.
31 #[command(subcommand)]
32 Workspace(WsCmd),
33
34 /// Ejecutar un comando one-shot dentro de un workspace.
35 Run {
36 /// ULID del workspace destino.
37 #[arg(short = 'w', long)]
38 workspace: String,
39 /// Si exit != 0, relanzar con backoff exponencial.
40 #[arg(long)]
41 restart_on_failure: bool,
42 /// Path del ejecutable.
43 exec: String,
44 /// Argumentos del comando.
45 argv: Vec<String>,
46 },
47
48 /// Discernir el tipo de un archivo (ad-hoc, sin workspace).
49 Discern {
50 /// Path al archivo a discernir.
51 path: PathBuf,
52 },
53
54 /// Listar comandos de un workspace.
55 Commands {
56 /// ULID del workspace.
57 workspace: String,
58 },
59
60 /// Mostrar tail del log capturado de un comando.
61 Logs {
62 /// ULID del workspace.
63 workspace: String,
64 /// ULID del comando.
65 command: String,
66 /// Bytes desde el final (0 = todo).
67 #[arg(long, default_value_t = 0)]
68 tail: usize,
69 /// Stream a leer: stdout | stderr | both.
70 #[arg(long, default_value = "both")]
71 stream: String,
72 },
73
74 /// Pipeline DAG con flujo tipado.
75 #[command(subcommand)]
76 Pipeline(PipeCmd),
77
78 /// Flow data plane (subscribirse a streams enriquecidos).
79 #[command(subcommand)]
80 Flow(FlowCmd),
81}
82
83#[derive(Subcommand, Debug)]
84enum FlowCmd {
85 /// Listar pipelines activos con sus sockets de flow.
86 List,
87 /// Cerrar el data plane de un pipeline (drop de todos sus sockets).
88 Drop { pipeline: String },
89 /// Suscribirse a un flow socket y volcar bytes a stdout.
90 Tail {
91 /// Path al Unix socket del flow.
92 socket: PathBuf,
93 },
94}
95
96#[derive(Subcommand, Debug)]
97enum PipeCmd {
98 /// Lanzar un Pipeline desde un spec TOML/JSON.
99 Run {
100 /// Path al spec del pipeline.
101 spec: PathBuf,
102 /// Interponer un tap entre productor↔consumidor de cada edge para
103 /// discernir el TypeRef del flujo.
104 #[arg(long)]
105 tap: bool,
106 /// Variables `KEY=VALUE` para sustitución `${KEY}` en el spec.
107 #[arg(long = "var", value_parser = parse_kv)]
108 vars: Vec<(String, String)>,
109 /// Tras lanzar, suscribir al primer flow socket y volcar bytes
110 /// a stdout hasta EOF. Implica `--tap`.
111 #[arg(long)]
112 tail: bool,
113 },
114 /// Guardar un pipeline bajo un nombre (persiste con el snapshot).
115 Save {
116 /// Nombre simbólico.
117 name: String,
118 /// Path al spec.
119 spec: PathBuf,
120 },
121 /// Listar nombres de pipelines guardados.
122 SavedList,
123 /// Eliminar un pipeline guardado (no afecta runs en curso).
124 Drop { name: String },
125 /// Detener un pipeline en curso por ID (SIGTERM → grace → SIGKILL
126 /// sólo a sus comandos).
127 Stop {
128 /// ULID del pipeline (devuelto por `pipeline run`).
129 pipeline: String,
130 #[arg(long, default_value_t = 1000)]
131 grace_ms: u64,
132 },
133 /// Ejecutar un pipeline guardado por nombre.
134 RunSaved {
135 name: String,
136 #[arg(long)]
137 tap: bool,
138 #[arg(long = "var", value_parser = parse_kv)]
139 vars: Vec<(String, String)>,
140 #[arg(long)]
141 tail: bool,
142 },
143}
144
145fn parse_kv(s: &str) -> Result<(String, String), String> {
146 let (k, v) = s.split_once('=').ok_or_else(|| format!("expected KEY=VALUE, got `{s}`"))?;
147 Ok((k.to_string(), v.to_string()))
148}
149
150#[derive(Subcommand, Debug)]
151enum WsCmd {
152 /// Crear un workspace desde un spec TOML/JSON.
153 Create {
154 /// Path al spec del workspace.
155 spec: PathBuf,
156 },
157 /// Listar workspaces vivos.
158 List,
159 /// Detener un workspace por ID.
160 Stop {
161 id: String,
162 /// Milisegundos de gracia tras SIGTERM antes de SIGKILL.
163 #[arg(long, default_value_t = 1000)]
164 grace_ms: u64,
165 },
166 /// Resource accounting (RSS, CPU, comandos vivos).
167 Stats {
168 id: String,
169 },
170 /// Quota report: rlimits declarados vs uso actual.
171 Quota {
172 id: String,
173 },
174}
175
176#[tokio::main]
177async fn main() -> Result<()> {
178 let cli = Cli::parse();
179 let socket = cli.socket.unwrap_or_else(default_socket_path);
180 let mut stream = UnixStream::connect(&socket)
181 .await
182 .with_context(|| format!("connect {}", socket.display()))?;
183
184 match cli.cmd {
185 Cmd::Ping => {
186 let resp = round_trip(&mut stream, Request::Ping).await?;
187 match resp {
188 Response::Pong => println!("pong"),
189 other => print_unexpected(&other),
190 }
191 }
192
193 Cmd::Caps => {
194 let resp = round_trip(&mut stream, Request::Capabilities).await?;
195 match resp {
196 Response::Capabilities {
197 kernel_version,
198 user_ns,
199 cgroup_v2,
200 cgroup_delegated,
201 has_cap_sys_admin,
202 } => {
203 println!("kernel: {}.{}.{}", kernel_version.0, kernel_version.1, kernel_version.2);
204 println!("user_ns: {user_ns}");
205 println!("cgroup_v2: {cgroup_v2}");
206 println!("cgroup_delegated: {cgroup_delegated}");
207 println!("cap_sys_admin: {has_cap_sys_admin}");
208 }
209 other => print_unexpected(&other),
210 }
211 }
212
213 Cmd::Workspace(WsCmd::Create { spec }) => {
214 let ws = load_workspace_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
215 let resp = round_trip(&mut stream, Request::WorkspaceCreate { spec: ws }).await?;
216 match resp {
217 Response::WorkspaceCreated { id, warnings } => {
218 println!("{id}");
219 for w in warnings {
220 eprintln!("warning: {w}");
221 }
222 }
223 Response::Error { message } => return Err(anyhow!(message)),
224 other => print_unexpected(&other),
225 }
226 }
227
228 Cmd::Workspace(WsCmd::List) => {
229 let resp = round_trip(&mut stream, Request::WorkspaceList).await?;
230 match resp {
231 Response::WorkspaceList { items } => {
232 if items.is_empty() {
233 println!("(no workspaces)");
234 }
235 for it in items {
236 println!(
237 "{} {:<20} cmds={} uptime={}ms",
238 it.id, it.label, it.commands, it.uptime_ms
239 );
240 }
241 }
242 other => print_unexpected(&other),
243 }
244 }
245
246 Cmd::Workspace(WsCmd::Stats { id }) => {
247 let id = parse_ws_id(&id)?;
248 let resp = round_trip(&mut stream, Request::WorkspaceStats { workspace: id }).await?;
249 match resp {
250 Response::WorkspaceStats { info } => {
251 println!("commands: {} alive / {} total", info.commands_alive, info.commands_total);
252 let fmt_mib = |b: u64| format!("{:.2} MiB", b as f64 / 1024.0 / 1024.0);
253 let rss = info.rss_bytes.map(fmt_mib).unwrap_or_else(|| "—".into());
254 let peak = info.rss_peak_bytes.map(fmt_mib).unwrap_or_else(|| "—".into());
255 let cpu = info
256 .cpu_usec
257 .map(|u| format!("{:.3} s", u as f64 / 1_000_000.0))
258 .unwrap_or_else(|| "—".into());
259 let cpu_pct = info
260 .cpu_percent
261 .map(|p| format!("{p:.1} % ({:.1}% total / {} cores)",
262 if info.cpu_cores > 0 { p / info.cpu_cores as f32 } else { p },
263 info.cpu_cores))
264 .unwrap_or_else(|| "— (esperando 2do sample)".into());
265 println!("rss: {rss}");
266 println!("rss_peak: {peak}");
267 println!("cpu: {cpu}");
268 println!("cpu_pct: {cpu_pct}");
269 println!("source: {}", info.source);
270 println!("uptime: {} ms", info.uptime_ms);
271 }
272 Response::Error { message } => return Err(anyhow!(message)),
273 other => print_unexpected(&other),
274 }
275 }
276
277 Cmd::Workspace(WsCmd::Quota { id }) => {
278 let id = parse_ws_id(&id)?;
279 let resp = round_trip(&mut stream, Request::WorkspaceQuota { workspace: id }).await?;
280 match resp {
281 Response::WorkspaceQuota { info } => {
282 let mem = info
283 .mem_limit
284 .map(|b| format!("{:.2} MiB", b as f64 / 1024.0 / 1024.0))
285 .unwrap_or_else(|| "—".into());
286 let nproc = info
287 .nproc_limit
288 .map(|n| n.to_string())
289 .unwrap_or_else(|| "—".into());
290 println!("mem_limit: {mem}");
291 println!("nproc_limit: {nproc}");
292 if info.breaches.is_empty() {
293 println!("breaches: (none — dentro de quota)");
294 } else {
295 println!("breaches:");
296 for b in info.breaches {
297 println!(" - {b}");
298 }
299 }
300 }
301 Response::Error { message } => return Err(anyhow!(message)),
302 other => print_unexpected(&other),
303 }
304 }
305
306 Cmd::Workspace(WsCmd::Stop { id, grace_ms }) => {
307 let id = parse_ws_id(&id)?;
308 let resp = round_trip(&mut stream, Request::WorkspaceStop { id, grace_ms }).await?;
309 match resp {
310 Response::WorkspaceStopped { id, reaped } => {
311 println!("stopped {id} (reaped {reaped})");
312 }
313 Response::Error { message } => return Err(anyhow!(message)),
314 other => print_unexpected(&other),
315 }
316 }
317
318 Cmd::Run { workspace, exec, argv, restart_on_failure } => {
319 let id = parse_ws_id(&workspace)?;
320 let resp = round_trip(
321 &mut stream,
322 Request::Run {
323 workspace: id,
324 exec,
325 argv,
326 envp: vec![],
327 restart_on_failure,
328 },
329 )
330 .await?;
331 match resp {
332 Response::RunStarted { command_id, pid, .. } => {
333 println!("{command_id} pid={pid}");
334 }
335 Response::Error { message } => return Err(anyhow!(message)),
336 other => print_unexpected(&other),
337 }
338 }
339
340 Cmd::Pipeline(PipeCmd::Run { spec, tap, vars, tail }) => {
341 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
342 // --tail implica --tap (no hay flow socket sin tap).
343 let effective_tap = tap || tail;
344 let resp = round_trip(
345 &mut stream,
346 Request::PipelineRun {
347 spec: p,
348 tap: effective_tap,
349 vars: vars.into_iter().collect(),
350 },
351 )
352 .await?;
353 let socket = print_pipeline_started_returning_socket(resp)?;
354 if tail {
355 if let Some(sock) = socket {
356 eprintln!("--- tailing {} ---", sock.display());
357 tail_socket(&sock).await?;
358 } else {
359 eprintln!("--tail: no hay flow socket disponible");
360 }
361 }
362 }
363
364 Cmd::Pipeline(PipeCmd::Save { name, spec }) => {
365 let p = load_pipeline_spec(&spec).with_context(|| format!("load {}", spec.display()))?;
366 let resp = round_trip(&mut stream, Request::PipelineSave { name: name.clone(), spec: p }).await?;
367 match resp {
368 Response::PipelineSaved { name } => println!("saved {name}"),
369 Response::Error { message } => return Err(anyhow!(message)),
370 other => print_unexpected(&other),
371 }
372 }
373
374 Cmd::Pipeline(PipeCmd::SavedList) => {
375 let resp = round_trip(&mut stream, Request::PipelineSavedList).await?;
376 match resp {
377 Response::PipelineSavedList { names } => {
378 if names.is_empty() {
379 println!("(no saved pipelines)");
380 }
381 for n in names {
382 println!("{n}");
383 }
384 }
385 other => print_unexpected(&other),
386 }
387 }
388
389 Cmd::Pipeline(PipeCmd::Stop { pipeline, grace_ms }) => {
390 let pid = Ulid::from_string(&pipeline).map_err(|e| anyhow!("invalid pipeline id: {e}"))?;
391 let resp = round_trip(&mut stream, Request::PipelineStop { pipeline: pid, grace_ms }).await?;
392 match resp {
393 Response::PipelineStopped { pipeline, reaped } => {
394 println!("stopped pipeline {pipeline} (reaped {reaped})");
395 }
396 other => print_unexpected(&other),
397 }
398 }
399
400 Cmd::Pipeline(PipeCmd::Drop { name }) => {
401 let resp = round_trip(&mut stream, Request::PipelineDrop { name }).await?;
402 match resp {
403 Response::PipelineDropped { name, existed } => {
404 if existed {
405 println!("dropped {name}");
406 } else {
407 eprintln!("no existía: {name}");
408 }
409 }
410 other => print_unexpected(&other),
411 }
412 }
413
414 Cmd::Pipeline(PipeCmd::RunSaved { name, tap, vars, tail }) => {
415 let effective_tap = tap || tail;
416 let resp = round_trip(
417 &mut stream,
418 Request::PipelineRunSaved {
419 name,
420 tap: effective_tap,
421 vars: vars.into_iter().collect(),
422 },
423 )
424 .await?;
425 let socket = print_pipeline_started_returning_socket(resp)?;
426 if tail {
427 if let Some(sock) = socket {
428 eprintln!("--- tailing {} ---", sock.display());
429 tail_socket(&sock).await?;
430 } else {
431 eprintln!("--tail: no hay flow socket disponible");
432 }
433 }
434 }
435
436 Cmd::Commands { workspace } => {
437 let ws = parse_ws_id(&workspace)?;
438 let resp = round_trip(&mut stream, Request::CommandList { workspace: ws }).await?;
439 match resp {
440 Response::CommandList { items } => {
441 if items.is_empty() {
442 println!("(no commands)");
443 }
444 for c in items {
445 let alive = if c.alive { "alive" } else { "exited" };
446 let exit = c
447 .exit_status
448 .map(|s| format!("exit={s}"))
449 .unwrap_or_default();
450 println!(
451 "{} {:<24} pid={:<7} {:<8} logs={} {}",
452 c.id, c.label, c.pid, alive, c.log_bytes, exit
453 );
454 }
455 }
456 other => print_unexpected(&other),
457 }
458 }
459
460 Cmd::Logs { workspace, command, tail, stream: which_stream } => {
461 let ws = parse_ws_id(&workspace)?;
462 let cmd_id = Ulid::from_string(&command).map_err(|e| anyhow!("invalid command id: {e}"))?;
463 let resp = round_trip(
464 &mut stream,
465 Request::CommandLogs {
466 workspace: ws,
467 command: cmd_id,
468 tail_bytes: tail,
469 stream: which_stream,
470 },
471 )
472 .await?;
473 match resp {
474 Response::CommandLogs { bytes } => {
475 // stdout raw, sin decoding — el log puede tener bytes binarios.
476 use std::io::Write;
477 let _ = std::io::stdout().write_all(&bytes);
478 let _ = std::io::stdout().flush();
479 }
480 Response::Error { message } => return Err(anyhow!(message)),
481 other => print_unexpected(&other),
482 }
483 }
484
485 Cmd::Flow(FlowCmd::List) => {
486 let resp = round_trip(&mut stream, Request::FlowList).await?;
487 match resp {
488 Response::FlowList { items } => {
489 if items.is_empty() {
490 println!("(no active flows)");
491 }
492 for it in items {
493 println!("{}", it.pipeline);
494 for s in it.sockets {
495 println!(" {}", s.display());
496 }
497 }
498 }
499 other => print_unexpected(&other),
500 }
501 }
502
503 Cmd::Flow(FlowCmd::Drop { pipeline }) => {
504 let pid = Ulid::from_string(&pipeline).map_err(|e| anyhow!("invalid pipeline id: {e}"))?;
505 let resp = round_trip(&mut stream, Request::FlowDrop { pipeline: pid }).await?;
506 match resp {
507 Response::FlowDropped { pipeline, existed } => {
508 if existed {
509 println!("dropped {pipeline}");
510 } else {
511 eprintln!("no existía: {pipeline}");
512 }
513 }
514 other => print_unexpected(&other),
515 }
516 }
517
518 Cmd::Flow(FlowCmd::Tail { socket }) => {
519 // Subscribirse directo al socket — no pasamos por el daemon.
520 use tokio::io::AsyncReadExt;
521 let mut s = UnixStream::connect(&socket)
522 .await
523 .with_context(|| format!("connect {}", socket.display()))?;
524 let mut buf = [0u8; 4096];
525 loop {
526 let n = s.read(&mut buf).await?;
527 if n == 0 {
528 break;
529 }
530 use std::io::Write;
531 let _ = std::io::stdout().write_all(&buf[..n]);
532 let _ = std::io::stdout().flush();
533 }
534 }
535
536 Cmd::Discern { path } => {
537 let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?;
538 // Sample: hasta 4 KiB.
539 let sample = bytes.into_iter().take(4096).collect();
540 let resp = round_trip(
541 &mut stream,
542 Request::Discern {
543 sample,
544 hint_path: Some(path),
545 },
546 )
547 .await?;
548 match resp {
549 Response::Discernment { ty, confidence, mime, lens } => {
550 println!("type: {ty}");
551 println!("confidence: {confidence:.2}");
552 if let Some(m) = mime {
553 println!("mime: {m}");
554 }
555 if let Some(l) = lens {
556 println!("lens: {l}");
557 }
558 }
559 Response::Error { message } => return Err(anyhow!(message)),
560 other => print_unexpected(&other),
561 }
562 }
563 }
564
565 Ok(())
566}
567
568async fn round_trip(stream: &mut UnixStream, req: Request) -> Result<Response> {
569 write_frame(stream, &req).await?;
570 let resp: Response = read_frame(stream).await?;
571 Ok(resp)
572}
573
574fn parse_ws_id(s: &str) -> Result<WorkspaceId> {
575 let u = Ulid::from_string(s).map_err(|e| anyhow!("invalid workspace id: {e}"))?;
576 Ok(WorkspaceId(u))
577}
578
579fn print_unexpected(r: &Response) {
580 eprintln!("unexpected response: {r:?}");
581}
582
583/// Imprime el resultado del launch del pipeline y retorna el path del
584/// primer flow socket (si hay), útil para `--tail`.
585fn print_pipeline_started_returning_socket(resp: Response) -> Result<Option<PathBuf>> {
586 match resp {
587 Response::PipelineStarted { pipeline, command_pids, edges } => {
588 println!("pipeline {pipeline}");
589 for (label, pid) in command_pids {
590 println!(" {:<20} pid={pid}", label);
591 }
592 let mut first_socket: Option<PathBuf> = None;
593 if !edges.is_empty() {
594 println!("edges:");
595 for e in &edges {
596 println!(
597 " {}.{} → {}.{} ty={:?} mime={:?} conf={:.2}",
598 e.from_label, e.from_output, e.to_label, e.to_input,
599 e.ty, e.mime, e.confidence,
600 );
601 if first_socket.is_none() {
602 first_socket = e.flow_socket.clone();
603 }
604 }
605 }
606 Ok(first_socket)
607 }
608 Response::Error { message } => Err(anyhow!(message)),
609 other => {
610 print_unexpected(&other);
611 Ok(None)
612 }
613 }
614}
615
616async fn tail_socket(socket: &std::path::Path) -> Result<()> {
617 use tokio::io::AsyncReadExt;
618 // Pequeña ventana de retry — el daemon retiene el flow channel
619 // antes de retornar, así que en la práctica ya está bindeado.
620 let mut s = UnixStream::connect(socket)
621 .await
622 .with_context(|| format!("connect {}", socket.display()))?;
623 let mut buf = [0u8; 4096];
624 loop {
625 let n = s.read(&mut buf).await?;
626 if n == 0 {
627 break;
628 }
629 use std::io::Write;
630 let _ = std::io::stdout().write_all(&buf[..n]);
631 let _ = std::io::stdout().flush();
632 }
633 Ok(())
634}