1pub mod probe;
69
70use std::io::Write;
71use std::net::SocketAddr;
72use std::path::PathBuf;
73use std::sync::mpsc::{Receiver, RecvTimeoutError};
74use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
75
76use gunnar_common::logtap::{self, LogLine};
77use tracing::Level;
78
79pub const TICK: Duration = Duration::from_secs(2);
81pub const EVENT_ROWS: usize = 20;
84pub const DEFAULT_COLS: usize = 80;
87pub const DEFAULT_ROWS: usize = 25;
88pub const DEVICE_ENV: &str = "GUNNAR_DASHBOARD";
90const DEVICE: &str = "/dev/tty0";
91
92#[derive(Debug, Clone)]
94pub struct Facts {
95 pub ssh_bind: SocketAddr,
96 pub grpc_bind: Option<SocketAddr>,
98 pub data_root: PathBuf,
99 pub build: String,
101 pub engine: Option<crate::storage::Engine>,
104 pub replication: crate::replication_view::ReplicationView,
113 pub tenants: TenantView,
116 pub flow: FlowView,
121}
122
123#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
131pub struct TenantCounts {
132 pub paid: u64,
133 pub grace: u64,
134 pub suspended: u64,
137}
138
139#[derive(Clone, Default)]
144pub struct TenantView {
145 source: std::sync::Arc<std::sync::OnceLock<Box<dyn Fn() -> TenantCounts + Send + Sync>>>,
146}
147
148impl TenantView {
149 pub fn new() -> Self {
150 Self::default()
151 }
152
153 pub fn fill(&self, source: impl Fn() -> TenantCounts + Send + Sync + 'static) {
155 let _ = self.source.set(Box::new(source));
156 }
157
158 pub fn counts(&self) -> Option<TenantCounts> {
161 self.source.get().map(|source| source())
162 }
163}
164
165impl std::fmt::Debug for TenantView {
166 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
167 f.debug_struct("TenantView")
168 .field("filled", &self.source.get().is_some())
169 .finish()
170 }
171}
172
173pub const SILENT_AFTER_DAYS: i64 = 14;
181
182type FleetSource = Box<dyn Fn() -> Option<gunnar_core::Fleet> + Send + Sync>;
184
185#[derive(Clone, Default)]
194pub struct FlowView {
195 source: std::sync::Arc<std::sync::OnceLock<FleetSource>>,
196}
197
198impl FlowView {
199 pub fn new() -> Self {
200 Self::default()
201 }
202
203 pub fn fill(&self, source: impl Fn() -> Option<gunnar_core::Fleet> + Send + Sync + 'static) {
205 let _ = self.source.set(Box::new(source));
206 }
207
208 pub fn fleet(&self) -> Option<gunnar_core::Fleet> {
213 self.source.get().and_then(|source| source())
214 }
215}
216
217impl std::fmt::Debug for FlowView {
218 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
219 f.debug_struct("FlowView")
220 .field("filled", &self.source.get().is_some())
221 .finish()
222 }
223}
224#[derive(Debug, Clone, PartialEq, Eq)]
226pub struct Row {
227 pub at: u64,
229 pub kind: &'static str,
230 pub text: String,
231 pub warn: bool,
232}
233
234#[derive(Debug, Clone)]
236pub struct Screen {
237 pub facts: Facts,
238 pub hostname: Option<String>,
239 pub addresses: Vec<probe::IfAddr>,
240 pub dns: Option<Vec<std::net::Ipv4Addr>>,
246 pub cpu_model: Option<String>,
247 pub cpus: usize,
248 pub cpu_percent: Option<u8>,
249 pub load1: Option<String>,
250 pub memory: Option<(u64, u64)>,
252 pub disk: Option<(u64, u64)>,
253 pub explode: Option<(u64, u64)>,
260 pub disk_kind: Option<probe::DiskKind>,
262 pub writer: Option<&'static str>,
265 pub uptime_secs: Option<u64>,
266 pub now_unix: u64,
267 pub rows: std::collections::VecDeque<Row>,
268 pub dropped: u64,
270 ticks: Option<probe::CpuTicks>,
271 pub index: Vec<Row>,
276 index_last: Vec<(PathBuf, u64, Instant)>,
278 pub standby: Vec<Row>,
291 pub tenants: Vec<Row>,
295 pub flow: Vec<Row>,
305}
306
307impl Screen {
308 pub fn new(facts: Facts) -> Self {
309 Screen {
310 facts,
311 hostname: None,
312 addresses: Vec::new(),
313 dns: None,
314 cpu_model: None,
315 cpus: 0,
316 cpu_percent: None,
317 load1: None,
318 memory: None,
319 disk: None,
320 explode: None,
321 disk_kind: None,
322 writer: None,
323 uptime_secs: None,
324 now_unix: 0,
325 rows: std::collections::VecDeque::with_capacity(EVENT_ROWS + 1),
326 dropped: 0,
327 ticks: None,
328 index: Vec::new(),
329 index_last: Vec::new(),
330 standby: Vec::new(),
331 tenants: Vec::new(),
332 flow: Vec::new(),
333 }
334 }
335
336 pub fn sample_tenants(&mut self, counts: Option<TenantCounts>) {
341 self.tenants = counts
342 .map(|c| Row {
343 at: self.now_unix,
344 kind: "tenants",
345 text: format!(
346 "paid {} grace {} suspended {}",
347 c.paid, c.grace, c.suspended
348 ),
349 warn: c.suspended > 0,
350 })
351 .into_iter()
352 .collect();
353 }
354
355 pub fn sample_flow(&mut self, fleet: Option<gunnar_core::Fleet>) {
364 self.flow = fleet
365 .map(|f| Row {
366 at: self.now_unix,
367 kind: "flow",
368 text: format!(
369 "in {} out {} today silent {} of {}",
370 short_bytes(f.bytes_in_today),
371 short_bytes(f.bytes_out_today),
372 f.silent,
373 f.namespaces
374 ),
375 warn: f.silent > 0,
376 })
377 .into_iter()
378 .collect();
379 }
380
381 pub fn sample_replication(&mut self, report: &crate::replication_view::ReplicationReport) {
388 self.standby = report
389 .standbys
390 .iter()
391 .map(|row| Row {
392 at: self.now_unix,
393 kind: "standby",
394 text: row.describe(report.synced, report.failed, self.now_unix as i64 * 1000),
395 warn: row.stale,
400 })
401 .collect();
402 if report.configured && report.standbys.is_empty() {
406 self.standby.push(Row {
407 at: self.now_unix,
408 kind: "standby",
409 text: "replication is configured and reports NO standby rows".to_owned(),
410 warn: true,
411 });
412 }
413 }
414
415 pub fn sample_index(&mut self, facts: &[probe::IndexFacts], now: Instant) {
418 let mut next_last = Vec::with_capacity(facts.len());
419 let mut rows = Vec::new();
420 for f in facts {
421 if !f.snap.active {
422 continue;
423 }
424 let prev = self.index_last.iter().find(|(d, _, _)| *d == f.git_dir);
425 let rate = prev.and_then(|(_, bytes, at)| {
426 let secs = now.saturating_duration_since(*at).as_secs_f64();
427 (secs > 0.0 && f.snap.bytes_done >= *bytes)
428 .then(|| ((f.snap.bytes_done - bytes) as f64 / secs) as u64)
429 });
430 next_last.push((f.git_dir.clone(), f.snap.bytes_done, now));
431 rows.push(Row {
432 at: self.now_unix,
433 kind: "index",
434 text: index_row_text(
435 &repo_label(&f.git_dir, &self.facts.data_root),
436 &f.snap,
437 rate,
438 ),
439 warn: f.snap.phase == "paused",
440 });
441 }
442 self.index_last = next_last;
443 self.index = rows;
444 }
445
446 pub fn sample(&mut self) {
448 self.hostname = probe::hostname();
449 self.addresses = probe::addresses();
450 self.dns = appliance_pid1::resolver::published();
451 if self.cpu_model.is_none() {
452 self.cpu_model = probe::cpu_model();
453 }
454 let (ticks, cpus) = probe::cpu_ticks();
455 if cpus > 0 {
456 self.cpus = cpus;
457 }
458 if let (Some(now), Some(then)) = (ticks, self.ticks) {
459 if let Some(p) = now.busy_percent_since(then) {
460 self.cpu_percent = Some(p);
461 }
462 }
463 if ticks.is_some() {
464 self.ticks = ticks;
465 }
466 self.load1 = probe::load1();
467 self.memory = probe::memory();
468 self.disk = probe::disk(&self.facts.data_root);
469 if self.disk_kind.is_none() {
470 self.disk_kind = probe::disk_kind(&self.facts.data_root);
471 }
472 if self.writer.is_none() {
473 self.writer = crate::storage::writer_arm();
474 }
475 self.uptime_secs = probe::uptime_secs();
476 self.now_unix = now_unix();
477 self.dropped = logtap::dropped();
478 if let Some(engine) = self.facts.engine {
479 let facts = probe::index_progress(engine);
480 self.sample_index(&facts, Instant::now());
481 let bands = engine.explode_use();
482 self.explode = (!bands.is_empty()).then(|| {
483 bands.iter().fold((0u64, 0u64), |(u, b), (_, band)| {
484 (u + band.used(), b + band.budget)
485 })
486 });
487 }
488 self.sample_replication(&self.facts.replication.report());
491 self.sample_tenants(self.facts.tenants.counts());
492 self.sample_flow(self.facts.flow.fleet());
493 }
494
495 pub fn push(&mut self, row: Row) {
497 self.rows.push_back(row);
498 while self.rows.len() > EVENT_ROWS {
499 self.rows.pop_front();
500 }
501 }
502
503 pub fn take(&mut self, line: &LogLine) {
505 if let Some((kind, text, warn)) = describe(line) {
506 self.push(Row {
507 at: now_unix(),
508 kind,
509 text,
510 warn,
511 });
512 }
513 }
514
515 pub fn render_frame(&self, cols: usize, rows: usize) -> String {
524 let width = cols.saturating_sub(1).max(20);
525 let mut lines = self.header_lines(width);
526 lines.push("\u{2500}".repeat(width));
527 for row in self
530 .index
531 .iter()
532 .chain(self.standby.iter())
533 .chain(self.tenants.iter())
534 .chain(self.flow.iter())
535 {
536 let mark = if row.warn { "!" } else { " " };
537 lines.push(format!(
538 "{mark}{} {:<7} {}",
539 clock(row.at),
540 row.kind,
541 row.text
542 ));
543 }
544 let body_rows = rows.saturating_sub(lines.len());
545 let shown = self.rows.iter().rev().take(body_rows).collect::<Vec<_>>();
546 for row in shown.into_iter().rev() {
547 let mark = if row.warn { "!" } else { " " };
548 lines.push(format!(
549 "{mark}{} {:<7} {}",
550 clock(row.at),
551 row.kind,
552 row.text
553 ));
554 }
555 while lines.len() < rows {
556 lines.push(String::new());
557 }
558 lines.truncate(rows);
559 let mut out = String::with_capacity(rows * (cols + 8) + 8);
560 out.push_str("\x1b[H");
561 for (i, line) in lines.iter().enumerate() {
562 if i > 0 {
563 out.push_str("\r\n");
564 }
565 out.push_str(&pad(line, width));
566 out.push_str("\x1b[K");
567 }
568 out
569 }
570
571 fn header_lines(&self, width: usize) -> Vec<String> {
572 let up = self
573 .uptime_secs
574 .map(fmt_uptime)
575 .unwrap_or_else(|| "-".into());
576 let dropped = if self.dropped > 0 {
577 format!(" (log tap dropped {})", self.dropped)
578 } else {
579 String::new()
580 };
581 let line1 = format!(
582 " gunnar {} appliance up {} {} UTC{}",
583 self.facts.build,
584 up,
585 clock(self.now_unix),
586 dropped
587 );
588 let mut line2 = format!(" host {}", self.hostname.as_deref().unwrap_or("(none)"));
589 if self.addresses.is_empty() {
590 line2.push_str(" no address");
591 }
592 for a in &self.addresses {
593 let piece = format!(" {a}");
594 if line2.len() + piece.len() > width {
595 break;
596 }
597 line2.push_str(&piece);
598 }
599 let dns = match &self.dns {
600 None => " dns none".to_string(),
601 Some(v) if v.is_empty() => " dns (file names none)".to_string(),
602 Some(v) => format!(
603 " dns {}",
604 v.iter()
605 .map(|s| s.to_string())
606 .collect::<Vec<_>>()
607 .join(",")
608 ),
609 };
610 if line2.len() + dns.len() <= width {
611 line2.push_str(&dns);
612 }
613 let cpu = match (&self.cpu_model, self.cpus) {
614 (Some(m), n) if n > 0 => format!("{n}x {m}"),
615 (Some(m), _) => m.clone(),
616 (None, n) if n > 0 => format!("{n} cpu"),
617 (None, _) => "-".into(),
618 };
619 let cpu = clip(&cpu, CPU_MODEL_COLS);
623 let pct = self
624 .cpu_percent
625 .map(|p| format!("{p:>3}%"))
626 .unwrap_or_else(|| " -%".into());
627 let ram = match self.memory {
628 Some((total, avail)) => {
629 let used = total.saturating_sub(avail);
630 format!(
631 "{}/{} {:>2}%",
632 fmt_bytes(used),
633 fmt_bytes(total),
634 pct_of(used, total)
635 )
636 }
637 None => "-".into(),
638 };
639 let line3 = format!(
640 " cpu {cpu} {pct} load {} ram {ram}",
641 self.load1.as_deref().unwrap_or("-")
642 );
643 let disk = match self.disk {
644 Some((total, avail)) => {
645 let used = total.saturating_sub(avail);
646 format!(
647 "{}/{} {:>2}%",
648 fmt_bytes(used),
649 fmt_bytes(total),
650 pct_of(used, total)
651 )
652 }
653 None => "-".into(),
654 };
655 let kind = self
656 .disk_kind
657 .as_ref()
658 .map(|k| format!(" {k}"))
659 .unwrap_or_default();
660 let kind = match self.writer {
661 Some(w) if kind.is_empty() => format!(" {w}"),
662 Some(w) => format!("{kind} {w}"),
663 None => kind,
664 };
665 let explode = match self.explode {
670 Some((used, budget)) if budget > 0 => {
671 format!(" explode {}/{}", fmt_bytes(used), fmt_bytes(budget))
672 }
673 Some((used, _)) => format!(" explode {} unlimited", fmt_bytes(used)),
674 None => String::new(),
675 };
676 let grpc = self
677 .facts
678 .grpc_bind
679 .map(|g| format!(" grpc {}", fmt_bind(g)))
680 .unwrap_or_default();
681 let line4 = format!(
682 " disk {} {disk}{kind}{explode} ssh {}{grpc}",
683 clip(&self.facts.data_root.display().to_string(), CPU_MODEL_COLS),
684 fmt_bind(self.facts.ssh_bind)
685 );
686 vec![line1, line2, line3, line4]
687 }
688}
689
690pub fn describe(line: &LogLine) -> Option<(&'static str, String, bool)> {
696 let warn = line.level <= Level::WARN;
697 let get = |k: &str| line.get(k).unwrap_or("-").to_owned();
698 let short_fp = |fp: &str| {
699 let body = fp.strip_prefix("SHA256:").unwrap_or(fp);
701 format!("{}\u{2026}", &body[..body.len().min(12)])
702 };
703 if let Some(event) = line.event() {
704 return match event {
705 "ssh.auth.accepted" => Some((
706 "login",
707 format!(
708 "{} {} {}",
709 get("principal"),
710 short_fp(&get("fingerprint")),
711 get("peer")
712 ),
713 false,
714 )),
715 "ssh.auth.anonymous" => {
716 Some(("anon", format!("{} public read only", get("peer")), false))
717 }
718 "ssh.auth.refused" | "ssh.auth.none.refused" => Some((
719 "REFUSED",
720 format!("login {} {}", get("peer"), get("reason")),
721 true,
722 )),
723 "git.transfer.finished" => {
724 let verb = match line.get("service") {
725 Some("git-receive-pack") => "push",
726 Some("git-upload-pack") => "fetch",
727 _ => "git",
728 };
729 Some((
730 verb,
731 format!(
732 "{} {} {} {} ms",
733 get("repo"),
734 get("principal"),
735 get("status"),
736 get("elapsed_ms")
737 ),
738 warn,
739 ))
740 }
741 "git.transfer.waited" => Some((
745 "waiting",
746 format!(
747 "{} waited mean {} ms worst {} ms limit {}",
748 get("waited_total"),
749 get("mean_wait_ms"),
750 get("peak_wait_ms"),
751 get("limit")
752 ),
753 false,
754 )),
755 "git.transfer.refused" | "http.transfer.refused" => Some((
756 "REFUSED",
757 format!("{} {} {}", get("repo"), get("principal"), line.message),
758 true,
759 )),
760 "policy.gate.refused" => Some((
761 "REFUSED",
762 format!("push {} {}", get("repo"), get("principal")),
763 true,
764 )),
765 "repo.created_on_first_push" => Some((
766 "repo+",
767 format!("{} {}", get("repo"), get("principal")),
768 false,
769 )),
770 "grpc.call" => {
771 let decision = line.get("decision").unwrap_or("-");
772 Some((
773 if decision == "allow" {
774 "admin"
775 } else {
776 "DENIED"
777 },
778 format!(
779 "{} {} {}",
780 get("method"),
781 get("principal"),
782 get("reason").replace('-', "")
783 ),
784 decision != "allow",
785 ))
786 }
787 "store.maintained" => Some((
788 "gc",
789 format!(
790 "{} {} -> {} B",
791 get("repo"),
792 get("bytes_before"),
793 get("bytes_after")
794 ),
795 false,
796 )),
797 "store.writer_selected" => Some((
801 "writer",
802 match line.get("fallback").filter(|f| *f != "-" && !f.is_empty()) {
803 Some(why) => format!("{} ({why})", get("writer")),
804 None => get("writer"),
805 },
806 false,
807 )),
808 "replication.synced" | "replication.failed" => {
809 Some(("mirror", format!("{} {}", get("repo"), line.message), warn))
810 }
811 "http.listening" | "metrics.listening" => Some((
812 "listen",
813 format!("{} {}", event.split('.').next().unwrap_or(""), get("addr")),
814 false,
815 )),
816 "install.mode" => Some((
824 "install",
825 match (line.get("mode"), line.get("kept")) {
826 (Some("upgrade"), Some("data-set")) => {
827 match line.get("kept_set").filter(|s| *s != "-" && !s.is_empty()) {
828 Some(set) => format!("upgrade /data kept (data set {set})"),
829 None => "upgrade /data kept (data set)".to_owned(),
830 }
831 }
832 (Some("upgrade"), _) => match line
833 .get("kept_partno")
834 .filter(|p| *p != "-" && !p.is_empty())
835 {
836 Some(partno) => {
837 format!("upgrade /data kept (p{partno} {}G)", get("kept_gib"))
838 }
839 None => "upgrade /data kept".to_owned(),
840 },
841 (Some(m), _) => format!("{m} /data new"),
842 (None, _) => "- /data new".to_string(),
843 },
844 false,
845 )),
846 "limits.budget" => Some((
852 "limits",
853 format!(
854 "transfers {} repos {} git mem {} ({}c {}fd {})",
855 get("transfers"),
856 get("open_repos"),
857 crate::budget::human(get("git_memory_bytes").parse().unwrap_or(0)),
858 get("cores"),
859 get("fds"),
860 crate::budget::human(get("mem_available").parse().unwrap_or(0)),
861 ),
862 false,
863 )),
864 "serve.starting" => Some((
869 "starting",
870 format!("git refused (retry) until {} finishes", get("waiting_for")),
871 true,
872 )),
873 "serve.ready" => Some((
874 "serving",
875 format!("git: {} took {} ms", get("waited_for"), get("ms")),
876 false,
877 )),
878 "limits.handles.reclaimed" => Some((
881 "limits",
882 format!(
883 "repos {} -> {} ({} dropped, {} available)",
884 get("was"),
885 get("now"),
886 get("dropped"),
887 crate::budget::human(get("mem_available").parse().unwrap_or(0)),
888 ),
889 true,
890 )),
891 "disktest" => Some((
894 "disktest",
895 format!("{} {}", clip(&get("path"), CPU_MODEL_COLS), line.message),
896 warn,
897 )),
898 "layout" => Some((
904 "layout",
905 format!("{} {}", clip(&get("path"), CPU_MODEL_COLS), line.message),
906 warn,
907 )),
908 "index.paused" => Some((
913 "index",
914 format!(
915 "paused: {} available < {} budget {}",
916 fmt_bytes(
917 line.get("available")
918 .and_then(|v| v.parse().ok())
919 .unwrap_or(0)
920 ),
921 fmt_bytes(line.get("budget").and_then(|v| v.parse().ok()).unwrap_or(0)),
922 short_repo(&get("repo"))
923 ),
924 true,
925 )),
926 "index.resumed" => Some((
927 "index",
928 format!(
929 "resumed after {} ms: {} available {}",
930 get("paused_ms"),
931 fmt_bytes(
932 line.get("available")
933 .and_then(|v| v.parse().ok())
934 .unwrap_or(0)
935 ),
936 short_repo(&get("repo"))
937 ),
938 false,
939 )),
940 "index.gave_up" => Some((
946 "index",
947 format!(
948 "GAVE UP ({}) after {} ms: {} available, {} needed {}",
949 get("reason"),
950 get("paused_ms"),
951 fmt_bytes(
952 line.get("available")
953 .and_then(|v| v.parse().ok())
954 .unwrap_or(0)
955 ),
956 fmt_bytes(line.get("budget").and_then(|v| v.parse().ok()).unwrap_or(0)),
957 short_repo(&get("repo"))
958 ),
959 true,
960 )),
961 "index.progress" => Some((
967 "index",
968 format!(
969 "{} {}% {}/{} {} {}/s {} re-inflations {} pauses rss {} avail {}",
970 short_repo(&get("repo")),
971 get("percent"),
972 fmt_count(
973 line.get("objects_done")
974 .and_then(|v| v.parse().ok())
975 .unwrap_or(0)
976 ),
977 fmt_count(
978 line.get("objects_total")
979 .and_then(|v| v.parse().ok())
980 .unwrap_or(0)
981 ),
982 get("phase"),
983 fmt_count(
984 line.get("objects_per_s")
985 .and_then(|v| v.parse().ok())
986 .unwrap_or(0)
987 ),
988 get("reinflations"),
989 get("pauses"),
990 fmt_bytes(
991 line.get("peak_rss")
992 .and_then(|v| v.parse().ok())
993 .unwrap_or(0)
994 ),
995 fmt_bytes(
996 line.get("mem_available")
997 .and_then(|v| v.parse().ok())
998 .unwrap_or(0)
999 ),
1000 ),
1001 false,
1002 )),
1003 "index.finished" => Some((
1004 "index",
1005 format!(
1006 "{} done: {} objects in {} {} re-inflations {} pauses peak rss {}",
1007 short_repo(&get("repo")),
1008 fmt_count(
1009 line.get("objects")
1010 .and_then(|v| v.parse().ok())
1011 .unwrap_or(0)
1012 ),
1013 fmt_ms(
1014 line.get("total_ms")
1015 .and_then(|v| v.parse().ok())
1016 .unwrap_or(0)
1017 ),
1018 get("reinflations"),
1019 get("pauses"),
1020 fmt_bytes(
1021 line.get("peak_rss")
1022 .and_then(|v| v.parse().ok())
1023 .unwrap_or(0)
1024 )
1025 ),
1026 false,
1027 )),
1028 "repo.purged" => Some((
1029 "delete",
1030 format!(
1031 "{} {} reclaimed",
1032 get("repo"),
1033 fmt_bytes(
1034 line.get("bytes_before")
1035 .and_then(|v| v.parse::<u64>().ok())
1036 .unwrap_or(0)
1037 .saturating_sub(
1038 line.get("bytes_after")
1039 .and_then(|v| v.parse::<u64>().ok())
1040 .unwrap_or(0)
1041 )
1042 )
1043 ),
1044 false,
1045 )),
1046 _ => {
1047 if warn {
1048 Some(("WARN", format!("{event} {}", line.message), true))
1049 } else {
1050 None
1051 }
1052 }
1053 };
1054 }
1055 let msg = line.message.as_str();
1056 if msg.starts_with("DHCP")
1057 || msg.contains("lease") && line.target.contains("dhcp")
1058 || line.target.contains("pid1") && msg.contains("DHCP")
1059 {
1060 let some = |v: String| {
1061 v.trim_start_matches("Some(")
1062 .trim_end_matches(')')
1063 .to_owned()
1064 };
1065 let text = if msg.starts_with("DHCP ACK") {
1066 format!(
1067 "{} {}/{} via {} lease {}s",
1068 get("interface"),
1069 get("address"),
1070 get("prefix"),
1071 some(get("gateway")),
1072 get("lease_secs")
1073 )
1074 } else if msg.contains("renewed") {
1075 format!(
1076 "{} renewed via {} next in {}s",
1077 get("interface"),
1078 get("via"),
1079 get("next_renew_in_secs")
1080 )
1081 } else if msg.contains("CHANGED") {
1082 format!(
1083 "{} NEW ADDRESS {}/{} via {}",
1084 get("interface"),
1085 get("address"),
1086 get("prefix"),
1087 some(get("gateway"))
1088 )
1089 } else {
1090 format!("{} {}", get("interface"), first_clause(msg))
1091 };
1092 return Some(("dhcp", text, warn));
1093 }
1094 if msg.starts_with("gunnar ssh listener") && line.get("addr").is_some() {
1095 return Some(("listen", format!("ssh {}", get("addr")), false));
1096 }
1097 if warn && (line.target.starts_with("gunnar") || line.target.starts_with("appliance_pid1")) {
1098 return Some((
1099 if line.level == Level::ERROR {
1100 "ERROR"
1101 } else {
1102 "WARN"
1103 },
1104 first_clause(msg),
1105 true,
1106 ));
1107 }
1108 None
1109}
1110
1111pub fn index_row_text(
1115 repo: &str,
1116 s: &crate::storage::IndexSnapshot,
1117 rate_bytes_per_s: Option<u64>,
1118) -> String {
1119 let mut t = format!(
1120 "{repo} {}{}% {}/{} {}",
1121 if s.resumed { "resumed " } else { "" },
1122 s.percent(),
1123 fmt_count(s.objects_done),
1124 fmt_count(s.objects_total),
1125 s.phase
1126 );
1127 if let Some(r) = rate_bytes_per_s {
1128 t.push_str(&format!(" {}/s", fmt_bytes(r)));
1129 }
1130 if s.pauses > 0 {
1131 t.push_str(&format!(" {}p", s.pauses));
1132 }
1133 t
1134}
1135
1136pub fn fmt_count(n: u64) -> String {
1138 if n >= 1_000_000 {
1139 format!("{:.1}M", n as f64 / 1e6)
1140 } else if n >= 10_000 {
1141 format!("{}k", n / 1000)
1142 } else {
1143 n.to_string()
1144 }
1145}
1146
1147pub fn fmt_ms(ms: u64) -> String {
1149 if ms >= 3_600_000 {
1150 format!("{:.1}h", ms as f64 / 3_600_000.0)
1151 } else if ms >= 60_000 {
1152 format!("{}m", ms / 60_000)
1153 } else if ms >= 1000 {
1154 format!("{:.1}s", ms as f64 / 1000.0)
1155 } else {
1156 format!("{ms} ms")
1157 }
1158}
1159
1160pub fn repo_label(git_dir: &std::path::Path, data_root: &std::path::Path) -> String {
1163 let rel = git_dir.strip_prefix(data_root).unwrap_or(git_dir);
1164 let parts: Vec<String> = rel
1165 .components()
1166 .filter_map(|c| match c {
1167 std::path::Component::Normal(s) => Some(s.to_string_lossy().into_owned()),
1168 _ => None,
1169 })
1170 .collect();
1171 let n = parts.len();
1172 if n >= 2 {
1173 let name = parts[n - 1].strip_suffix(".git").unwrap_or(&parts[n - 1]);
1174 format!("{}/{name}", parts[n - 2])
1175 } else {
1176 rel.display().to_string()
1177 }
1178}
1179
1180fn short_repo(root: &str) -> String {
1183 let p = std::path::Path::new(root);
1184 let dir = if p.file_name().is_some_and(|f| f == "gunnar-git.znippy") {
1185 p.parent().unwrap_or(p)
1186 } else {
1187 p
1188 };
1189 repo_label(dir, std::path::Path::new("/"))
1190}
1191
1192fn first_clause(msg: &str) -> String {
1194 let end = msg.find([':', '.']).unwrap_or(msg.len());
1195 msg[..end].trim().to_owned()
1196}
1197
1198const CPU_MODEL_COLS: usize = 26;
1200
1201fn fmt_bind(addr: SocketAddr) -> String {
1206 if addr.ip().is_unspecified() {
1207 format!("*:{}", addr.port())
1208 } else {
1209 addr.to_string()
1210 }
1211}
1212
1213fn clip(s: &str, n: usize) -> String {
1215 if s.chars().count() <= n {
1216 return s.to_owned();
1217 }
1218 let mut out: String = s.chars().take(n.saturating_sub(1)).collect();
1219 out.push('\u{2026}');
1220 out
1221}
1222
1223fn pad(line: &str, width: usize) -> String {
1224 let mut s: String = line.chars().take(width).collect();
1225 let n = s.chars().count();
1226 s.extend(std::iter::repeat_n(' ', width.saturating_sub(n)));
1227 s
1228}
1229
1230fn pct_of(part: u64, whole: u64) -> u64 {
1231 if whole == 0 {
1232 0
1233 } else {
1234 (part as u128 * 100 / whole as u128) as u64
1235 }
1236}
1237
1238pub fn fmt_bytes(n: u64) -> String {
1240 const UNITS: [&str; 5] = ["B", "K", "M", "G", "T"];
1241 let mut v = n as f64;
1242 let mut u = 0;
1243 while v >= 1000.0 && u < UNITS.len() - 1 {
1244 v /= 1024.0;
1245 u += 1;
1246 }
1247 if u == 0 {
1248 format!("{n}B")
1249 } else if v < 10.0 {
1250 format!("{v:.1}{}", UNITS[u])
1251 } else {
1252 format!("{v:.0}{}", UNITS[u])
1253 }
1254}
1255
1256pub fn fmt_uptime(secs: u64) -> String {
1258 let d = secs / 86_400;
1259 let h = secs % 86_400 / 3600;
1260 let m = secs % 3600 / 60;
1261 let s = secs % 60;
1262 if d > 0 {
1263 format!("{d}d {h:02}:{m:02}:{s:02}")
1264 } else {
1265 format!("{h:02}:{m:02}:{s:02}")
1266 }
1267}
1268
1269pub fn clock(unix: u64) -> String {
1271 let t = unix % 86_400;
1272 format!("{:02}:{:02}:{:02}", t / 3600, t % 3600 / 60, t % 60)
1273}
1274
1275fn short_bytes(n: u64) -> String {
1285 const K: u64 = 1 << 10;
1286 for (unit, per) in [("T", 1u64 << 40), ("G", 1 << 30), ("M", 1 << 20), ("K", K)] {
1287 if n >= per {
1288 return format!("{}.{}{unit}", n / per, ((n % per) * 10) / per);
1289 }
1290 }
1291 n.to_string()
1292}
1293
1294fn now_unix() -> u64 {
1295 SystemTime::now()
1296 .duration_since(UNIX_EPOCH)
1297 .map(|d| d.as_secs())
1298 .unwrap_or(0)
1299}
1300
1301pub fn spawn(facts: Facts) -> bool {
1308 let explicit = std::env::var(DEVICE_ENV).ok().filter(|p| !p.is_empty());
1309 let gated = appliance_pid1::net::is_init() && appliance_pid1::net::booted_by_korp_start();
1310 if explicit.is_none() && !gated {
1311 tracing::debug!(
1312 "console dashboard: not the appliance's PID 1 and {DEVICE_ENV} unset; not drawing"
1313 );
1314 return false;
1315 }
1316 let device = match &explicit {
1317 Some(path) => std::fs::OpenOptions::new().write(true).open(path).ok(),
1318 None => appliance_pid1::console::open_gfx_console(),
1319 };
1320 let Some(device) = device else {
1321 tracing::info!(
1322 device = explicit.as_deref().unwrap_or(DEVICE),
1323 "console dashboard: the device cannot be opened (no VT in this kernel?); not drawing"
1324 );
1325 return false;
1326 };
1327 let Some(rx) = logtap::take_receiver() else {
1328 tracing::info!(
1329 "console dashboard: the log tap was already taken or never installed; not drawing"
1330 );
1331 return false;
1332 };
1333 if explicit.is_none() {
1334 quiet_printk();
1335 }
1336 tracing::info!(
1337 device = explicit.as_deref().unwrap_or(DEVICE),
1338 "console dashboard drawing"
1339 );
1340 std::thread::Builder::new()
1342 .name("gunnar-dashboard".into())
1343 .spawn(move || {
1344 let result =
1347 std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run(device, rx, facts)));
1348 if result.is_err() {
1349 tracing::error!(
1350 "console dashboard thread panicked; the screen is frozen, serving continues"
1351 );
1352 }
1353 })
1354 .is_ok()
1355}
1356
1357fn quiet_printk() {
1361 let _ = std::fs::write("/proc/sys/kernel/printk", "1 4 1 7\n");
1362}
1363
1364fn run(mut device: std::fs::File, rx: Receiver<LogLine>, facts: Facts) {
1365 let (cols, rows) = winsize(&device).unwrap_or((DEFAULT_COLS, DEFAULT_ROWS));
1366 let mut screen = Screen::new(facts);
1367 let _ = device.write_all(b"\x1b[?25l\x1b[2J");
1369 screen.sample();
1370 let mut next_tick = Instant::now();
1371 loop {
1372 let now = Instant::now();
1375 if now >= next_tick {
1376 screen.sample();
1377 next_tick = now + TICK;
1378 }
1379 let wait = next_tick.saturating_duration_since(now);
1380 match rx.recv_timeout(wait) {
1381 Ok(line) => {
1382 screen.take(&line);
1383 while let Ok(more) = rx.try_recv() {
1384 screen.take(&more);
1385 }
1386 }
1387 Err(RecvTimeoutError::Timeout) => {}
1388 Err(RecvTimeoutError::Disconnected) => break,
1389 }
1390 if device
1391 .write_all(screen.render_frame(cols, rows).as_bytes())
1392 .is_err()
1393 {
1394 break;
1397 }
1398 let _ = device.flush();
1399 }
1400}
1401
1402fn winsize(device: &std::fs::File) -> Option<(usize, usize)> {
1403 let ws = rustix::termios::tcgetwinsize(device).ok()?;
1404 if ws.ws_col == 0 || ws.ws_row == 0 {
1405 return None;
1406 }
1407 Some((ws.ws_col as usize, ws.ws_row as usize))
1408}
1409
1410#[cfg(test)]
1411mod tests {
1412 use super::*;
1413
1414 fn line(
1415 level: Level,
1416 target: &'static str,
1417 event: Option<&str>,
1418 msg: &str,
1419 fields: &[(&'static str, &str)],
1420 spans: &[(&'static str, &str)],
1421 ) -> LogLine {
1422 let mut f: Vec<(&'static str, String)> =
1423 fields.iter().map(|(k, v)| (*k, v.to_string())).collect();
1424 if let Some(e) = event {
1425 f.insert(0, ("event", e.to_owned()));
1426 }
1427 LogLine {
1428 level,
1429 target,
1430 message: msg.to_owned(),
1431 fields: f,
1432 span_fields: spans.iter().map(|(k, v)| (*k, v.to_string())).collect(),
1433 }
1434 }
1435
1436 #[test]
1437 fn every_shown_event_reads_as_one_short_row() {
1438 let conn = [
1439 ("peer", "10.0.2.2:5123"),
1440 ("principal", "alice"),
1441 ("repo", "ns/repo"),
1442 ("service", "git-receive-pack"),
1443 ];
1444 let cases: Vec<(LogLine, (&str, &str, bool))> = vec![
1445 (
1446 line(
1447 Level::INFO,
1448 "gunnar::ssh",
1449 Some("ssh.auth.accepted"),
1450 "publickey accepted",
1451 &[(
1452 "fingerprint",
1453 "SHA256:abcdefghijklmnopqrstuvwxyz0123456789ABCDEFG",
1454 )],
1455 &conn,
1456 ),
1457 ("login", "alice abcdefghijkl\u{2026} 10.0.2.2:5123", false),
1458 ),
1459 (
1460 line(
1461 Level::INFO,
1462 "gunnar::git",
1463 Some("git.transfer.finished"),
1464 "transfer finished",
1465 &[("status", "ok"), ("elapsed_ms", "812")],
1466 &conn,
1467 ),
1468 ("push", "ns/repo alice ok 812 ms", false),
1469 ),
1470 (
1471 line(
1472 Level::WARN,
1473 "gunnar::audit",
1474 Some("grpc.call"),
1475 "refused",
1476 &[
1477 ("method", "Admin.X"),
1478 ("principal", "eve"),
1479 ("decision", "deny"),
1480 ("reason", "not admin"),
1481 ],
1482 &[],
1483 ),
1484 ("DENIED", "Admin.X eve not admin", true),
1485 ),
1486 (
1487 line(
1488 Level::INFO,
1489 "gunnar::audit",
1490 Some("grpc.call"),
1491 "call",
1492 &[
1493 ("method", "Identity.AddKey"),
1494 ("principal", "alice"),
1495 ("decision", "allow"),
1496 ],
1497 &[],
1498 ),
1499 ("admin", "Identity.AddKey alice ", false),
1500 ),
1501 (
1502 line(
1503 Level::INFO,
1504 "appliance_pid1::dhcp",
1505 None,
1506 "DHCP ACK: lease taken. The renewal loop keeps it alive",
1507 &[
1508 ("interface", "eth0"),
1509 ("address", "10.0.2.15"),
1510 ("prefix", "24"),
1511 ("gateway", "Some(10.0.2.2)"),
1512 ("lease_secs", "3600"),
1513 ],
1514 &[],
1515 ),
1516 ("dhcp", "eth0 10.0.2.15/24 via 10.0.2.2 lease 3600s", false),
1517 ),
1518 (
1519 line(
1520 Level::WARN,
1521 "appliance_pid1::runtime",
1522 None,
1523 "DHCP renewal CHANGED THE ADDRESS mid-life — a big event",
1524 &[
1525 ("interface", "eth0"),
1526 ("address", "10.0.2.99"),
1527 ("prefix", "24"),
1528 ("gateway", "Some(10.0.2.2)"),
1529 ],
1530 &[],
1531 ),
1532 ("dhcp", "eth0 NEW ADDRESS 10.0.2.99/24 via 10.0.2.2", true),
1533 ),
1534 (
1535 line(
1536 Level::INFO,
1537 "gunnar::git",
1538 Some("repo.created_on_first_push"),
1539 "created",
1540 &[],
1541 &conn,
1542 ),
1543 ("repo+", "ns/repo alice", false),
1544 ),
1545 (
1546 line(
1547 Level::WARN,
1548 "gunnar_server::wg",
1549 None,
1550 "the tunnel refused: no address. details follow",
1551 &[],
1552 &[],
1553 ),
1554 ("WARN", "the tunnel refused", true),
1555 ),
1556 (
1557 line(
1558 Level::INFO,
1559 "gunnar_server::boot",
1560 Some("install.mode"),
1561 "install record read off /boot/korp/install",
1562 &[
1563 ("mode", "upgrade"),
1564 ("kept", "korp-data"),
1565 ("kept_partno", "3"),
1566 ("kept_gib", "225.0"),
1567 ],
1568 &[],
1569 ),
1570 ("install", "upgrade /data kept (p3 225.0G)", false),
1571 ),
1572 (
1573 line(
1574 Level::INFO,
1575 "gunnar_server::boot",
1576 Some("install.mode"),
1577 "install record read off /boot/korp/install",
1578 &[
1579 ("mode", "fresh"),
1580 ("kept", "-"),
1581 ("kept_partno", "-"),
1582 ("kept_gib", "0.0"),
1583 ],
1584 &[],
1585 ),
1586 ("install", "fresh /data new", false),
1587 ),
1588 (
1589 line(
1590 Level::INFO,
1591 "gunnar_server::boot",
1592 Some("install.mode"),
1593 "install record read off /boot/korp/install",
1594 &[
1595 ("mode", "upgrade"),
1596 ("kept", "data-set"),
1597 ("kept_set", "4\u{d7}75G"),
1598 ],
1599 &[],
1600 ),
1601 (
1602 "install",
1603 "upgrade /data kept (data set 4\u{d7}75G)",
1604 false,
1605 ),
1606 ),
1607 (
1608 line(
1609 Level::INFO,
1610 "gunnar_server::boot",
1611 Some("install.mode"),
1612 "install record read off /boot/korp/install",
1613 &[("mode", "upgrade"), ("kept", "data-set"), ("kept_set", "-")],
1614 &[],
1615 ),
1616 ("install", "upgrade /data kept (data set)", false),
1617 ),
1618 (
1619 line(
1621 Level::INFO,
1622 "gunnar_server::boot",
1623 Some("install.mode"),
1624 "install record read off /boot/korp/install",
1625 &[("mode", "upgrade"), ("kept", "-")],
1626 &[],
1627 ),
1628 ("install", "upgrade /data kept", false),
1629 ),
1630 (
1631 line(
1632 Level::INFO,
1633 "gunnar_server::disktest",
1634 Some("disktest"),
1635 "seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s (1.6 s)",
1636 &[("path", "/data")],
1637 &[],
1638 ),
1639 (
1640 "disktest",
1641 "/data seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s (1.6 s)",
1642 false,
1643 ),
1644 ),
1645 (
1650 line(
1651 Level::INFO,
1652 "gunnar::slow",
1653 Some("layout"),
1654 "INTEGRITY ONLY \u{b7} off-stripe: p3 /dev/vda3",
1655 &[("path", "/boot/korp/layout.conf"), ("claim", "off-stripe")],
1656 &[],
1657 ),
1658 (
1659 "layout",
1660 "/boot/korp/layout.conf INTEGRITY ONLY \u{b7} off-stripe: p3 /dev/vda3",
1661 false,
1662 ),
1663 ),
1664 (
1669 line(
1670 Level::WARN,
1671 "gunnar::slow",
1672 Some("layout"),
1673 "INTEGRITY ONLY \u{b7} SAME DEVICE as /data \u{2014} not off-stripe \
1674 (fallback /data)",
1675 &[("path", "/boot/korp/layout.conf"), ("claim", "same-device")],
1676 &[],
1677 ),
1678 (
1679 "layout",
1680 "/boot/korp/layout.conf INTEGRITY ONLY \u{b7} SAME DEVICE as /data \
1681 \u{2014} not off-stripe (fallback /data)",
1682 true,
1683 ),
1684 ),
1685 (
1690 line(
1691 Level::INFO,
1692 "gunnar::limits",
1693 Some("limits.budget"),
1694 "transfers 64 \u{b7} open repos 204",
1695 &[
1696 ("transfers", "64"),
1697 ("open_repos", "204"),
1698 ("git_memory_bytes", "4289783800"),
1699 ("cores", "4"),
1700 ("fds", "1048576"),
1701 ("mem_available", "8579567616"),
1702 ],
1703 &[],
1704 ),
1705 (
1706 "limits",
1707 "transfers 64 repos 204 git mem 4.0G (4c 1048576fd 8.0G)",
1708 false,
1709 ),
1710 ),
1711 (
1712 line(
1713 Level::INFO,
1714 "gunnar::git",
1715 Some("git.transfer.waited"),
1716 "transfers are WAITING",
1717 &[
1718 ("waited_total", "412"),
1719 ("mean_wait_ms", "180"),
1720 ("peak_wait_ms", "1974"),
1721 ("limit", "64"),
1722 ],
1723 &[],
1724 ),
1725 (
1726 "waiting",
1727 "412 waited mean 180 ms worst 1974 ms limit 64",
1728 false,
1729 ),
1730 ),
1731 (
1733 line(
1734 Level::INFO,
1735 "gunnar::limits",
1736 Some("serve.starting"),
1737 "git transfers are refused",
1738 &[("waiting_for", "the start-up disk test")],
1739 &[],
1740 ),
1741 (
1742 "starting",
1743 "git refused (retry) until the start-up disk test finishes",
1744 true,
1745 ),
1746 ),
1747 (
1748 line(
1749 Level::INFO,
1750 "gunnar::limits",
1751 Some("serve.ready"),
1752 "serving git",
1753 &[("waited_for", "the start-up disk test"), ("ms", "710")],
1754 &[],
1755 ),
1756 ("serving", "git: the start-up disk test took 710 ms", false),
1757 ),
1758 (
1760 line(
1761 Level::WARN,
1762 "gunnar::limits",
1763 Some("limits.handles.reclaimed"),
1764 "memory is short",
1765 &[
1766 ("was", "50"),
1767 ("now", "25"),
1768 ("dropped", "25"),
1769 ("mem_available", "432013312"),
1770 ("mem_total", "2147483648"),
1771 ],
1772 &[],
1773 ),
1774 (
1775 "limits",
1776 "repos 50 -> 25 (25 dropped, 412M available)",
1777 true,
1778 ),
1779 ),
1780 (
1781 line(
1782 Level::WARN,
1783 "gunnar_server::disktest",
1784 Some("disktest"),
1785 "failed: write: No space left on device",
1786 &[("path", "/data")],
1787 &[],
1788 ),
1789 (
1790 "disktest",
1791 "/data failed: write: No space left on device",
1792 true,
1793 ),
1794 ),
1795 (
1796 line(
1797 Level::INFO,
1798 "gunnar::store",
1799 Some("store.writer_selected"),
1800 "the writer arm",
1801 &[
1802 ("writer", "uring"),
1803 ("source", "cmdline"),
1804 ("fallback", "-"),
1805 ],
1806 &[],
1807 ),
1808 ("writer", "uring", false),
1809 ),
1810 (
1811 line(
1812 Level::INFO,
1813 "gunnar::store",
1814 Some("store.writer_selected"),
1815 "the writer arm",
1816 &[
1817 ("writer", "safe"),
1818 ("source", "cmdline"),
1819 ("fallback", "io_uring unavailable: ENOSYS"),
1820 ],
1821 &[],
1822 ),
1823 ("writer", "safe (io_uring unavailable: ENOSYS)", false),
1824 ),
1825 ];
1826 for (l, (kind, text, warn)) in cases {
1827 let got = describe(&l).unwrap_or_else(|| panic!("{:?} must be shown", l.event()));
1828 assert_eq!((got.0, got.1.as_str(), got.2), (kind, text, warn));
1829 }
1830 }
1831
1832 #[test]
1833 fn the_noise_stays_in_the_log() {
1834 let quiet = [
1835 line(
1836 Level::INFO,
1837 "gunnar::git",
1838 Some("git.upload_pack.advertised"),
1839 "served the advertisement",
1840 &[],
1841 &[],
1842 ),
1843 line(
1844 Level::INFO,
1845 "gunnar_server::serve",
1846 None,
1847 "issuer ready (persisted)",
1848 &[],
1849 &[],
1850 ),
1851 line(
1852 Level::INFO,
1853 "gunnar::audit",
1854 Some("grpc.request.done"),
1855 "done",
1856 &[],
1857 &[],
1858 ),
1859 line(
1860 Level::WARN,
1861 "hyper::proto",
1862 None,
1863 "some third-party warning",
1864 &[],
1865 &[],
1866 ),
1867 ];
1868 for l in &quiet {
1869 assert_eq!(describe(l), None, "{:?} / {:?}", l.event(), l.message);
1870 }
1871 }
1872
1873 #[test]
1874 fn numbers_format_for_a_narrow_cell() {
1875 assert_eq!(fmt_bytes(900), "900B");
1876 assert_eq!(fmt_bytes(12 * 1024), "12K");
1877 assert_eq!(fmt_bytes(3_300_000_000), "3.1G");
1878 assert_eq!(fmt_bytes(300 * 1024 * 1024 * 1024), "300G");
1879 assert_eq!(
1880 fmt_uptime(3 * 86_400 + 4 * 3600 + 12 * 60 + 7),
1881 "3d 04:12:07"
1882 );
1883 assert_eq!(fmt_uptime(59), "00:00:59");
1884 assert_eq!(clock(1_700_000_000), "22:13:20");
1885 assert_eq!(pad("ab", 4), "ab ");
1886 assert_eq!(pad("abcdef", 4), "abcd");
1887 assert_eq!(
1888 pad("åäö", 4),
1889 "åäö ",
1890 "padding counts characters, not bytes"
1891 );
1892 assert_eq!(
1893 clip(
1894 "32x AMD Ryzen Threadripper PRO 3975WX 32-Cores",
1895 CPU_MODEL_COLS
1896 ),
1897 "32x AMD Ryzen Threadrippe\u{2026}"
1898 );
1899 assert_eq!(
1900 clip("2x AMD EPYC 7542 32-Core", CPU_MODEL_COLS),
1901 "2x AMD EPYC 7542 32-Core"
1902 );
1903 assert_eq!(fmt_bind("0.0.0.0:2222".parse().unwrap()), "*:2222");
1904 assert_eq!(fmt_bind("[::]:50051".parse().unwrap()), "*:50051");
1905 assert_eq!(
1906 fmt_bind("127.0.0.1:50051".parse().unwrap()),
1907 "127.0.0.1:50051"
1908 );
1909 }
1910
1911 #[test]
1917 fn the_index_row_says_what_the_indexer_is_doing_and_the_events_read_as_rows() {
1918 use crate::storage::IndexSnapshot;
1919 let snap = |done: u64, phase: &'static str, resumed: bool, pauses: u64| IndexSnapshot {
1920 active: true,
1921 resumed,
1922 pack_id: 3,
1923 objects_done: done,
1924 objects_total: 13_948_244,
1925 bytes_done: done * 600,
1926 bytes_total: 8_700_000_000,
1927 phase,
1928 started_at_ms: 1,
1929 reinflations: 0,
1930 pauses,
1931 };
1932 assert_eq!(
1933 index_row_text(
1934 "alice/linux",
1935 &snap(4_700_000, "resolve", false, 0),
1936 Some(2_100_000_000)
1937 ),
1938 "alice/linux 33% 4.7M/13.9M resolve 2.0G/s"
1939 );
1940 assert_eq!(
1941 index_row_text("alice/linux", &snap(0, "walk", true, 0), None),
1942 "alice/linux resumed 0% 0/13.9M walk"
1943 );
1944 assert_eq!(
1945 index_row_text("a/b", &snap(13_948_244, "fold", false, 2), None),
1946 "a/b 100% 13.9M/13.9M fold 2p"
1947 );
1948
1949 let mut s = Screen::new(facts());
1952 s.now_unix = 1_700_000_000;
1953 let dir = PathBuf::from("/data/repos/alice/linux.git");
1954 let t0 = Instant::now();
1955 let f1 = probe::IndexFacts {
1956 git_dir: dir.clone(),
1957 snap: snap(1_000_000, "resolve", false, 0),
1958 };
1959 s.sample_index(&[f1], t0);
1960 assert_eq!(s.index.len(), 1);
1961 assert_eq!(s.index[0].kind, "index");
1962 assert_eq!(s.index[0].text, "alice/linux 7% 1.0M/13.9M resolve");
1963 assert!(!s.index[0].warn);
1964 let f2 = probe::IndexFacts {
1965 git_dir: dir.clone(),
1966 snap: snap(3_000_000, "paused", false, 1),
1967 };
1968 s.sample_index(&[f2], t0 + Duration::from_secs(2));
1969 assert_eq!(
1970 s.index[0].text,
1971 "alice/linux 21% 3.0M/13.9M paused 572M/s 1p"
1972 );
1973 assert!(s.index[0].warn, "a paused pack is a warning row");
1974 let mut idle = snap(3_000_000, "done", false, 1);
1975 idle.active = false;
1976 s.sample_index(
1977 &[probe::IndexFacts {
1978 git_dir: dir,
1979 snap: idle,
1980 }],
1981 t0 + Duration::from_secs(4),
1982 );
1983 assert!(s.index.is_empty(), "an idle counter draws no row");
1984 s.sample_index(
1986 &[probe::IndexFacts {
1987 git_dir: PathBuf::from("/data/repos/a/b.git"),
1988 snap: snap(5, "walk", false, 0),
1989 }],
1990 t0,
1991 );
1992 s.push(Row {
1993 at: 1_700_000_000,
1994 kind: "login",
1995 text: "x".into(),
1996 warn: false,
1997 });
1998 let frame = s.render_frame(80, 25);
1999 let lines: Vec<&str> = frame.split("\r\n").collect();
2000 assert!(
2001 lines[5].contains("index a/b 0% 5/13.9M walk"),
2002 "row 5: {:?}",
2003 lines[5]
2004 );
2005 assert!(lines[6].contains("login x"), "row 6: {:?}", lines[6]);
2006
2007 let paused = line(
2009 Level::WARN,
2010 "gunnar::index",
2011 Some("index.paused"),
2012 "index paused",
2013 &[
2014 ("available", "3113851289"),
2015 ("budget", "3435973836"),
2016 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2017 ],
2018 &[],
2019 );
2020 assert_eq!(
2021 describe(&paused),
2022 Some((
2023 "index",
2024 "paused: 2.9G available < 3.2G budget alice/linux".to_string(),
2025 true
2026 ))
2027 );
2028 let resumed = line(
2029 Level::INFO,
2030 "gunnar::index",
2031 Some("index.resumed"),
2032 "index resumed",
2033 &[
2034 ("available", "5368709120"),
2035 ("budget", "3435973836"),
2036 ("paused_ms", "8120"),
2037 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2038 ],
2039 &[],
2040 );
2041 assert_eq!(
2042 describe(&resumed),
2043 Some((
2044 "index",
2045 "resumed after 8120 ms: 5.0G available alice/linux".to_string(),
2046 false
2047 ))
2048 );
2049 let finished = line(
2050 Level::INFO,
2051 "gunnar::index",
2052 Some("index.finished"),
2053 "indexed",
2054 &[
2055 ("objects", "13948244"),
2056 ("total_ms", "2520000"),
2057 ("reinflations", "1234"),
2058 ("pauses", "2"),
2059 ("peak_rss", "9126805504"),
2060 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2061 ],
2062 &[],
2063 );
2064 assert_eq!(describe(&finished), Some(("index", "alice/linux done: 13.9M objects in 42m 1234 re-inflations 2 pauses peak rss 8.5G".to_string(), false)));
2065 let progress = line(
2068 Level::INFO,
2069 "gunnar::index",
2070 Some("index.progress"),
2071 "index",
2072 &[
2073 ("percent", "78"),
2074 ("objects_done", "11000000"),
2075 ("objects_total", "13948244"),
2076 ("phase", "resolve"),
2077 ("objects_per_s", "744000"),
2078 ("reinflations", "12"),
2079 ("pauses", "1"),
2080 ("peak_rss", "13958643712"),
2081 ("mem_available", "3286511616"),
2082 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2083 ],
2084 &[],
2085 );
2086 assert_eq!(
2087 describe(&progress),
2088 Some((
2089 "index",
2090 "alice/linux 78% 11.0M/13.9M resolve 744k/s 12 re-inflations 1 pauses rss 13G avail 3.1G".to_string(),
2091 false
2092 ))
2093 );
2094 let gave_up = line(
2095 Level::WARN,
2096 "gunnar::index",
2097 Some("index.gave_up"),
2098 "gave up",
2099 &[
2100 ("reason", "nothing-to-give"),
2101 ("paused_ms", "31200"),
2102 ("available", "3113851289"),
2103 ("budget", "3435973836"),
2104 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2105 ],
2106 &[],
2107 );
2108 assert_eq!(
2109 describe(&gave_up),
2110 Some((
2111 "index",
2112 "GAVE UP (nothing-to-give) after 31200 ms: 2.9G available, 3.2G needed alice/linux".to_string(),
2113 true
2114 ))
2115 );
2116 let purged = line(
2117 Level::INFO,
2118 "gunnar::store",
2119 Some("repo.purged"),
2120 "reclaimed",
2121 &[
2122 ("repo", "alice/linux"),
2123 ("bytes_before", "26843545600"),
2124 ("bytes_after", "0"),
2125 ],
2126 &[],
2127 );
2128 assert_eq!(
2129 describe(&purged),
2130 Some(("delete", "alice/linux 25G reclaimed".to_string(), false))
2131 );
2132 assert_eq!(fmt_count(812), "812");
2133 assert_eq!(fmt_count(45_000), "45k");
2134 assert_eq!(fmt_ms(3_700_000), "1.0h");
2135 assert_eq!(
2136 repo_label(
2137 std::path::Path::new("/data/repos/ns/x.git"),
2138 std::path::Path::new("/data")
2139 ),
2140 "ns/x"
2141 );
2142 }
2143
2144 fn facts() -> Facts {
2145 Facts {
2146 ssh_bind: "0.0.0.0:2222".parse().unwrap(),
2147 grpc_bind: Some("0.0.0.0:50051".parse().unwrap()),
2148 data_root: PathBuf::from("/data"),
2149 build: "v1.1.3-145-g1eb7083".into(),
2150 engine: None,
2151 replication: crate::replication_view::ReplicationView::new(),
2152 tenants: TenantView::new(),
2153 flow: FlowView::new(),
2154 }
2155 }
2156
2157 #[test]
2164 fn the_tenants_row_is_drawn_when_filled_and_absent_when_not() {
2165 let mut s = Screen::new(facts());
2166 s.now_unix = 1_000;
2167 s.sample_tenants(None);
2168 let frame = s.render_frame(80, 24);
2169 assert!(
2170 !frame.contains("tenants"),
2171 "an unfilled view must draw no tenants row:\n{frame}"
2172 );
2173 s.sample_tenants(Some(TenantCounts {
2174 paid: 118,
2175 grace: 4,
2176 suspended: 3,
2177 }));
2178 let frame = s.render_frame(80, 24);
2179 assert!(
2180 frame.contains("tenants paid 118 grace 4 suspended 3"),
2181 "the row names all three counts:\n{frame}"
2182 );
2183 assert!(
2185 frame.contains("!00:16:40 tenants"),
2186 "suspended > 0 is a `!` row:\n{frame}"
2187 );
2188 s.sample_tenants(Some(TenantCounts {
2189 paid: 1,
2190 grace: 0,
2191 suspended: 0,
2192 }));
2193 let frame = s.render_frame(80, 24);
2194 assert!(
2195 frame.contains(" 00:16:40 tenants paid 1 grace 0 suspended 0"),
2196 "nobody suspended is an ordinary row:\n{frame}"
2197 );
2198 }
2199
2200 #[test]
2208 fn a_standby_that_has_never_synced_is_a_warn_row_on_the_band() {
2209 use crate::replication_view::{ReplicationReport, StandbyRow};
2210 let mut s = Screen::new(facts());
2211 s.now_unix = 1_000;
2212 s.sample_replication(&ReplicationReport {
2213 configured: true,
2214 failed: 8,
2215 standbys: vec![StandbyRow {
2216 name: "standby".into(),
2217 remote: "ssh://git@standby:2222".into(),
2218 last_success_unix_ms: 0,
2219 stale: true,
2220 ..StandbyRow::default()
2221 }],
2222 ..ReplicationReport::default()
2223 });
2224 let frame = s.render_frame(100, 25);
2225 assert!(
2226 frame.contains("last successful replication: NEVER"),
2227 "the band must say it in those words:\n{frame}"
2228 );
2229 assert!(
2230 frame.contains("!00:16:40 standby"),
2231 "a stale standby must be a WARN row (`!`), not an ordinary one:\n{frame}"
2232 );
2233
2234 let mut s = Screen::new(facts());
2237 s.now_unix = 1_000;
2238 s.sample_replication(&ReplicationReport {
2239 configured: true,
2240 synced: 105,
2241 failed: 40,
2242 standbys: vec![StandbyRow {
2243 name: "standby".into(),
2244 last_success_unix_ms: 997_000,
2245 stale: false,
2246 ..StandbyRow::default()
2247 }],
2248 ..ReplicationReport::default()
2249 });
2250 let frame = s.render_frame(100, 25);
2251 assert!(!frame.contains("NEVER"), "{frame}");
2252 assert!(frame.contains(" 00:16:40 standby"), "{frame}");
2253 assert!(
2254 frame.contains("synced 105 failed 40 last 00:00:03 ago"),
2255 "{frame}"
2256 );
2257
2258 let mut s = Screen::new(facts());
2262 s.now_unix = 1_000;
2263 s.sample_replication(&ReplicationReport {
2264 configured: true,
2265 ..ReplicationReport::default()
2266 });
2267 assert!(s.render_frame(100, 25).contains("reports NO standby rows"));
2268
2269 let mut s = Screen::new(facts());
2272 s.now_unix = 1_000;
2273 s.sample_replication(&ReplicationReport::default());
2274 assert!(s.standby.is_empty());
2275 }
2276
2277 #[test]
2278 fn a_frame_is_exactly_rows_lines_of_width_minus_one_and_the_newest_events_win() {
2279 let mut s = Screen::new(facts());
2280 s.hostname = Some("gunnar-appliance".into());
2281 s.addresses = vec![probe::IfAddr {
2282 interface: "eth0".into(),
2283 address: "198.51.100.155".parse().unwrap(),
2284 prefix: 24,
2285 }];
2286 s.memory = Some((16 * 1024 * 1024 * 1024, 15 * 1024 * 1024 * 1024));
2287 s.disk = Some((300 * 1024 * 1024 * 1024, 297 * 1024 * 1024 * 1024));
2288 s.uptime_secs = Some(3 * 86_400 + 4 * 3600 + 12 * 60 + 7);
2289 s.cpus = 4;
2290 s.cpu_model = Some("AMD EPYC 7542".into());
2291 s.disk_kind = Some(probe::DiskKind {
2292 fs: "xfs".into(),
2293 set: Some(probe::SetFacts {
2294 level: "raid0".into(),
2295 members: 4,
2296 member_bytes: 75 * 1024 * 1024 * 1024,
2297 chunk_bytes: 524288,
2298 }),
2299 });
2300 s.cpu_percent = Some(12);
2301 s.load1 = Some("0.10".into());
2302 s.now_unix = 1_700_000_000;
2303 for i in 0..(EVENT_ROWS + 5) {
2304 s.push(Row {
2305 at: i as u64,
2306 kind: "login",
2307 text: format!("user{i}"),
2308 warn: i % 7 == 0,
2309 });
2310 }
2311 let frame = s.render_frame(80, 25);
2312 assert!(
2313 frame.starts_with("\x1b[H"),
2314 "homes the cursor, never clears"
2315 );
2316 let raw: Vec<&str> = frame["\x1b[H".len()..].split("\r\n").collect();
2317 assert_eq!(raw.len(), 25);
2318 assert!(
2319 raw.iter().all(|l| l.ends_with("\x1b[K")),
2320 "every row ends with erase-to-EOL so column 80 is cleared: {:?}",
2321 raw.iter()
2322 .map(|l| &l[l.len().saturating_sub(4)..])
2323 .collect::<Vec<_>>()
2324 );
2325 let lines: Vec<&str> = raw
2326 .iter()
2327 .map(|l| l.strip_suffix("\x1b[K").unwrap())
2328 .collect();
2329 assert!(
2330 lines.iter().all(|l| l.chars().count() == 79),
2331 "every row is padded to cols-1: {:?}",
2332 lines.iter().map(|l| l.chars().count()).collect::<Vec<_>>()
2333 );
2334 assert!(
2335 lines[0].contains("gunnar v1.1.3-145-g1eb7083")
2336 && lines[0].contains("up 3d 04:12:07")
2337 && lines[0].contains("22:13:20 UTC"),
2338 "{}",
2339 lines[0]
2340 );
2341 assert!(
2342 lines[1].contains("host gunnar-appliance eth0 198.51.100.155/24"),
2343 "{}",
2344 lines[1]
2345 );
2346 assert!(
2347 lines[2].contains("cpu 4x AMD EPYC 7542 12% load 0.10 ram 1.0G/16G 6%"),
2348 "{}",
2349 lines[2]
2350 );
2351 assert!(
2352 lines[3].contains(
2353 "disk /data 3.0G/300G 1% 4\u{d7}75G raid0 xfs ssh *:2222 grpc *:50051"
2354 ),
2355 "{}",
2356 lines[3]
2357 );
2358 s.writer = Some("uring");
2360 let frame = s.render_frame(80, 25);
2361 let l3 = frame["\x1b[H".len()..].split("\r\n").nth(3).unwrap();
2362 assert!(
2363 l3.contains("4\u{d7}75G raid0 xfs uring ssh *:2222 grpc *:50051"),
2364 "{l3}"
2365 );
2366 assert_eq!(
2367 l3.strip_suffix("\x1b[K").unwrap().chars().count(),
2368 79,
2369 "still fits: {l3}"
2370 );
2371 s.writer = None;
2372 assert!(lines[4].starts_with("\u{2500}\u{2500}\u{2500}"));
2373 assert_eq!(s.rows.len(), EVENT_ROWS, "the ring holds EVENT_ROWS");
2374 assert!(
2375 lines[5].contains("user5"),
2376 "the oldest kept row is the first body row: {}",
2377 lines[5]
2378 );
2379 assert!(
2380 lines[24].contains("user24"),
2381 "the newest row is the last: {}",
2382 lines[24]
2383 );
2384 assert!(
2385 lines[7].starts_with('!'),
2386 "a warning row is marked: {}",
2387 lines[7]
2388 );
2389 assert!(
2390 lines[5].starts_with(' '),
2391 "a plain row is not: {}",
2392 lines[5]
2393 );
2394 }
2395
2396 #[test]
2397 fn a_dark_screen_says_dash_not_zero() {
2398 let s = Screen::new(facts());
2401 let frame = s.render_frame(80, 25);
2402 assert!(
2403 frame.contains("up -")
2404 && frame.contains("host (none) no address")
2405 && frame.contains("cpu - -% load - ram -")
2406 && frame.contains("disk /data - ssh *:2222"),
2407 "{frame}"
2408 );
2409 }
2410
2411 #[test]
2412 fn the_disk_cell_says_what_data_is_from_fixture_mounts_and_sysfs() {
2413 const MOUNTS: &str = "rootfs / rootfs rw 0 0\nproc /proc proc rw 0 0\n/dev/md/korp-data /data xfs rw,nosuid,nodev,noexec,relatime,attr2,inode64 0 0\n/dev/vda3 /mnt/fat\\040box vfat rw,dirsync 0 0\n";
2416 let md = probe::parse_mounts(MOUNTS, "/data").unwrap();
2417 assert_eq!(
2418 md,
2419 probe::MountFacts {
2420 source: "/dev/md/korp-data".into(),
2421 fstype: "xfs".into()
2422 }
2423 );
2424 let fat = probe::parse_mounts(MOUNTS, "/mnt/fat box").unwrap();
2425 assert_eq!(fat.fstype, "vfat");
2426 assert_eq!(
2427 probe::parse_mounts(MOUNTS, "/tmp/x"),
2428 None,
2429 "a plain directory is no mount"
2430 );
2431
2432 let dir = tempfile::tempdir().unwrap();
2435 let mdir = dir.path().join("md");
2436 std::fs::create_dir_all(&mdir).unwrap();
2437 std::fs::write(mdir.join("level"), "raid0\n").unwrap();
2438 std::fs::write(mdir.join("raid_disks"), "4\n").unwrap();
2439 std::fs::write(mdir.join("chunk_size"), "524288\n").unwrap();
2440 for m in ["vdb1", "vdc1", "vdd1", "vde1"] {
2441 let d = mdir.join(format!("dev-{m}"));
2442 std::fs::create_dir_all(&d).unwrap();
2443 std::fs::write(d.join("size"), "78643200\n").unwrap();
2444 }
2445 let set = probe::read_set_facts(dir.path()).unwrap();
2446 assert_eq!(
2447 set,
2448 probe::SetFacts {
2449 level: "raid0".into(),
2450 members: 4,
2451 member_bytes: 75 * 1024 * 1024 * 1024,
2452 chunk_bytes: 524288
2453 }
2454 );
2455 assert_eq!(
2456 probe::DiskKind::from_mount(&md, Some(set)).to_string(),
2457 "4\u{d7}75G raid0 xfs"
2458 );
2459 assert_eq!(probe::DiskKind::from_mount(&fat, None).to_string(), "fat");
2460 assert_eq!(
2461 probe::DiskKind::from_mount(&md, None).to_string(),
2462 "xfs",
2463 "a plain xfs disk names only the filesystem"
2464 );
2465
2466 let half = tempfile::tempdir().unwrap();
2468 std::fs::create_dir_all(half.path().join("md")).unwrap();
2469 std::fs::write(half.path().join("md/level"), "raid0\n").unwrap();
2470 std::fs::write(half.path().join("md/raid_disks"), "4\n").unwrap();
2471 assert_eq!(probe::read_set_facts(half.path()), None);
2472 assert_eq!(probe::read_set_facts(&dir.path().join("nowhere")), None);
2473 }
2474
2475 #[test]
2476 fn the_address_line_does_not_overflow_the_width() {
2477 let mut s = Screen::new(facts());
2478 for i in 0..12u8 {
2479 s.addresses.push(probe::IfAddr {
2480 interface: format!("eth{i}"),
2481 address: format!("10.0.{i}.15").parse().unwrap(),
2482 prefix: 24,
2483 });
2484 }
2485 let frame = s.render_frame(80, 25);
2486 let line2 = frame.split("\r\n").nth(1).unwrap();
2487 assert!(
2488 line2.contains("eth0 10.0.0.15/24") && !line2.contains("eth11"),
2489 "{line2}"
2490 );
2491 }
2492}