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