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()
1337 .name("gunnar-dashboard".into())
1338 .spawn(move || {
1339 let result =
1342 std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run(device, rx, facts)));
1343 if result.is_err() {
1344 tracing::error!(
1345 "console dashboard thread panicked; the screen is frozen, serving continues"
1346 );
1347 }
1348 })
1349 .is_ok()
1350}
1351
1352fn quiet_printk() {
1356 let _ = std::fs::write("/proc/sys/kernel/printk", "1 4 1 7\n");
1357}
1358
1359fn run(mut device: std::fs::File, rx: Receiver<LogLine>, facts: Facts) {
1360 let (cols, rows) = winsize(&device).unwrap_or((DEFAULT_COLS, DEFAULT_ROWS));
1361 let mut screen = Screen::new(facts);
1362 let _ = device.write_all(b"\x1b[?25l\x1b[2J");
1364 screen.sample();
1365 let mut next_tick = Instant::now();
1366 loop {
1367 let now = Instant::now();
1370 if now >= next_tick {
1371 screen.sample();
1372 next_tick = now + TICK;
1373 }
1374 let wait = next_tick.saturating_duration_since(now);
1375 match rx.recv_timeout(wait) {
1376 Ok(line) => {
1377 screen.take(&line);
1378 while let Ok(more) = rx.try_recv() {
1379 screen.take(&more);
1380 }
1381 }
1382 Err(RecvTimeoutError::Timeout) => {}
1383 Err(RecvTimeoutError::Disconnected) => break,
1384 }
1385 if device
1386 .write_all(screen.render_frame(cols, rows).as_bytes())
1387 .is_err()
1388 {
1389 break;
1392 }
1393 let _ = device.flush();
1394 }
1395}
1396
1397fn winsize(device: &std::fs::File) -> Option<(usize, usize)> {
1398 let ws = rustix::termios::tcgetwinsize(device).ok()?;
1399 if ws.ws_col == 0 || ws.ws_row == 0 {
1400 return None;
1401 }
1402 Some((ws.ws_col as usize, ws.ws_row as usize))
1403}
1404
1405#[cfg(test)]
1406mod tests {
1407 use super::*;
1408
1409 fn line(
1410 level: Level,
1411 target: &'static str,
1412 event: Option<&str>,
1413 msg: &str,
1414 fields: &[(&'static str, &str)],
1415 spans: &[(&'static str, &str)],
1416 ) -> LogLine {
1417 let mut f: Vec<(&'static str, String)> =
1418 fields.iter().map(|(k, v)| (*k, v.to_string())).collect();
1419 if let Some(e) = event {
1420 f.insert(0, ("event", e.to_owned()));
1421 }
1422 LogLine {
1423 level,
1424 target,
1425 message: msg.to_owned(),
1426 fields: f,
1427 span_fields: spans.iter().map(|(k, v)| (*k, v.to_string())).collect(),
1428 }
1429 }
1430
1431 #[test]
1432 fn every_shown_event_reads_as_one_short_row() {
1433 let conn = [
1434 ("peer", "10.0.2.2:5123"),
1435 ("principal", "alice"),
1436 ("repo", "ns/repo"),
1437 ("service", "git-receive-pack"),
1438 ];
1439 let cases: Vec<(LogLine, (&str, &str, bool))> = vec![
1440 (
1441 line(
1442 Level::INFO,
1443 "gunnar::ssh",
1444 Some("ssh.auth.accepted"),
1445 "publickey accepted",
1446 &[(
1447 "fingerprint",
1448 "SHA256:abcdefghijklmnopqrstuvwxyz0123456789ABCDEFG",
1449 )],
1450 &conn,
1451 ),
1452 (
1453 "login",
1454 "alice abcdefghijkl\u{2026} 10.0.2.2:5123",
1455 false,
1456 ),
1457 ),
1458 (
1459 line(
1460 Level::INFO,
1461 "gunnar::git",
1462 Some("git.transfer.finished"),
1463 "transfer finished",
1464 &[("status", "ok"), ("elapsed_ms", "812")],
1465 &conn,
1466 ),
1467 ("push", "ns/repo alice ok 812 ms", false),
1468 ),
1469 (
1470 line(
1471 Level::WARN,
1472 "gunnar::audit",
1473 Some("grpc.call"),
1474 "refused",
1475 &[
1476 ("method", "Admin.X"),
1477 ("principal", "eve"),
1478 ("decision", "deny"),
1479 ("reason", "not admin"),
1480 ],
1481 &[],
1482 ),
1483 ("DENIED", "Admin.X eve not admin", true),
1484 ),
1485 (
1486 line(
1487 Level::INFO,
1488 "gunnar::audit",
1489 Some("grpc.call"),
1490 "call",
1491 &[
1492 ("method", "Identity.AddKey"),
1493 ("principal", "alice"),
1494 ("decision", "allow"),
1495 ],
1496 &[],
1497 ),
1498 ("admin", "Identity.AddKey alice ", false),
1499 ),
1500 (
1501 line(
1502 Level::INFO,
1503 "appliance_pid1::dhcp",
1504 None,
1505 "DHCP ACK: lease taken. The renewal loop keeps it alive",
1506 &[
1507 ("interface", "eth0"),
1508 ("address", "10.0.2.15"),
1509 ("prefix", "24"),
1510 ("gateway", "Some(10.0.2.2)"),
1511 ("lease_secs", "3600"),
1512 ],
1513 &[],
1514 ),
1515 ("dhcp", "eth0 10.0.2.15/24 via 10.0.2.2 lease 3600s", false),
1516 ),
1517 (
1518 line(
1519 Level::WARN,
1520 "appliance_pid1::runtime",
1521 None,
1522 "DHCP renewal CHANGED THE ADDRESS mid-life — a big event",
1523 &[
1524 ("interface", "eth0"),
1525 ("address", "10.0.2.99"),
1526 ("prefix", "24"),
1527 ("gateway", "Some(10.0.2.2)"),
1528 ],
1529 &[],
1530 ),
1531 ("dhcp", "eth0 NEW ADDRESS 10.0.2.99/24 via 10.0.2.2", true),
1532 ),
1533 (
1534 line(
1535 Level::INFO,
1536 "gunnar::git",
1537 Some("repo.created_on_first_push"),
1538 "created",
1539 &[],
1540 &conn,
1541 ),
1542 ("repo+", "ns/repo alice", false),
1543 ),
1544 (
1545 line(
1546 Level::WARN,
1547 "gunnar_server::wg",
1548 None,
1549 "the tunnel refused: no address. details follow",
1550 &[],
1551 &[],
1552 ),
1553 ("WARN", "the tunnel refused", true),
1554 ),
1555 (
1556 line(
1557 Level::INFO,
1558 "gunnar_server::boot",
1559 Some("install.mode"),
1560 "install record read off /boot/korp/install",
1561 &[
1562 ("mode", "upgrade"),
1563 ("kept", "korp-data"),
1564 ("kept_partno", "3"),
1565 ("kept_gib", "225.0"),
1566 ],
1567 &[],
1568 ),
1569 ("install", "upgrade /data kept (p3 225.0G)", false),
1570 ),
1571 (
1572 line(
1573 Level::INFO,
1574 "gunnar_server::boot",
1575 Some("install.mode"),
1576 "install record read off /boot/korp/install",
1577 &[
1578 ("mode", "fresh"),
1579 ("kept", "-"),
1580 ("kept_partno", "-"),
1581 ("kept_gib", "0.0"),
1582 ],
1583 &[],
1584 ),
1585 ("install", "fresh /data new", false),
1586 ),
1587 (
1588 line(
1589 Level::INFO,
1590 "gunnar_server::boot",
1591 Some("install.mode"),
1592 "install record read off /boot/korp/install",
1593 &[
1594 ("mode", "upgrade"),
1595 ("kept", "data-set"),
1596 ("kept_set", "4\u{d7}75G"),
1597 ],
1598 &[],
1599 ),
1600 (
1601 "install",
1602 "upgrade /data kept (data set 4\u{d7}75G)",
1603 false,
1604 ),
1605 ),
1606 (
1607 line(
1608 Level::INFO,
1609 "gunnar_server::boot",
1610 Some("install.mode"),
1611 "install record read off /boot/korp/install",
1612 &[("mode", "upgrade"), ("kept", "data-set"), ("kept_set", "-")],
1613 &[],
1614 ),
1615 ("install", "upgrade /data kept (data set)", false),
1616 ),
1617 (
1618 line(
1620 Level::INFO,
1621 "gunnar_server::boot",
1622 Some("install.mode"),
1623 "install record read off /boot/korp/install",
1624 &[("mode", "upgrade"), ("kept", "-")],
1625 &[],
1626 ),
1627 ("install", "upgrade /data kept", false),
1628 ),
1629 (
1630 line(
1631 Level::INFO,
1632 "gunnar_server::disktest",
1633 Some("disktest"),
1634 "seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s (1.6 s)",
1635 &[("path", "/data")],
1636 &[],
1637 ),
1638 (
1639 "disktest",
1640 "/data seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s (1.6 s)",
1641 false,
1642 ),
1643 ),
1644 (
1649 line(
1650 Level::INFO,
1651 "gunnar::slow",
1652 Some("layout"),
1653 "INTEGRITY ONLY \u{b7} off-stripe: p3 /dev/vda3",
1654 &[
1655 ("path", "/boot/korp/layout.conf"),
1656 ("claim", "off-stripe"),
1657 ],
1658 &[],
1659 ),
1660 (
1661 "layout",
1662 "/boot/korp/layout.conf INTEGRITY ONLY \u{b7} off-stripe: p3 /dev/vda3",
1663 false,
1664 ),
1665 ),
1666 (
1671 line(
1672 Level::WARN,
1673 "gunnar::slow",
1674 Some("layout"),
1675 "INTEGRITY ONLY \u{b7} SAME DEVICE as /data \u{2014} not off-stripe \
1676 (fallback /data)",
1677 &[
1678 ("path", "/boot/korp/layout.conf"),
1679 ("claim", "same-device"),
1680 ],
1681 &[],
1682 ),
1683 (
1684 "layout",
1685 "/boot/korp/layout.conf INTEGRITY ONLY \u{b7} SAME DEVICE as /data \
1686 \u{2014} not off-stripe (fallback /data)",
1687 true,
1688 ),
1689 ),
1690 (
1695 line(
1696 Level::INFO,
1697 "gunnar::limits",
1698 Some("limits.budget"),
1699 "transfers 64 \u{b7} open repos 204",
1700 &[
1701 ("transfers", "64"),
1702 ("open_repos", "204"),
1703 ("git_memory_bytes", "4289783800"),
1704 ("cores", "4"),
1705 ("fds", "1048576"),
1706 ("mem_available", "8579567616"),
1707 ],
1708 &[],
1709 ),
1710 (
1711 "limits",
1712 "transfers 64 repos 204 git mem 4.0G (4c 1048576fd 8.0G)",
1713 false,
1714 ),
1715 ),
1716 (
1717 line(
1718 Level::INFO,
1719 "gunnar::git",
1720 Some("git.transfer.waited"),
1721 "transfers are WAITING",
1722 &[
1723 ("waited_total", "412"),
1724 ("mean_wait_ms", "180"),
1725 ("peak_wait_ms", "1974"),
1726 ("limit", "64"),
1727 ],
1728 &[],
1729 ),
1730 (
1731 "waiting",
1732 "412 waited mean 180 ms worst 1974 ms limit 64",
1733 false,
1734 ),
1735 ),
1736 (
1738 line(
1739 Level::INFO,
1740 "gunnar::limits",
1741 Some("serve.starting"),
1742 "git transfers are refused",
1743 &[("waiting_for", "the start-up disk test")],
1744 &[],
1745 ),
1746 (
1747 "starting",
1748 "git refused (retry) until the start-up disk test finishes",
1749 true,
1750 ),
1751 ),
1752 (
1753 line(
1754 Level::INFO,
1755 "gunnar::limits",
1756 Some("serve.ready"),
1757 "serving git",
1758 &[("waited_for", "the start-up disk test"), ("ms", "710")],
1759 &[],
1760 ),
1761 ("serving", "git: the start-up disk test took 710 ms", false),
1762 ),
1763 (
1765 line(
1766 Level::WARN,
1767 "gunnar::limits",
1768 Some("limits.handles.reclaimed"),
1769 "memory is short",
1770 &[
1771 ("was", "50"),
1772 ("now", "25"),
1773 ("dropped", "25"),
1774 ("mem_available", "432013312"),
1775 ("mem_total", "2147483648"),
1776 ],
1777 &[],
1778 ),
1779 (
1780 "limits",
1781 "repos 50 -> 25 (25 dropped, 412M available)",
1782 true,
1783 ),
1784 ),
1785 (
1786 line(
1787 Level::WARN,
1788 "gunnar_server::disktest",
1789 Some("disktest"),
1790 "failed: write: No space left on device",
1791 &[("path", "/data")],
1792 &[],
1793 ),
1794 (
1795 "disktest",
1796 "/data failed: write: No space left on device",
1797 true,
1798 ),
1799 ),
1800 (
1801 line(
1802 Level::INFO,
1803 "gunnar::store",
1804 Some("store.writer_selected"),
1805 "the writer arm",
1806 &[
1807 ("writer", "uring"),
1808 ("source", "cmdline"),
1809 ("fallback", "-"),
1810 ],
1811 &[],
1812 ),
1813 ("writer", "uring", false),
1814 ),
1815 (
1816 line(
1817 Level::INFO,
1818 "gunnar::store",
1819 Some("store.writer_selected"),
1820 "the writer arm",
1821 &[
1822 ("writer", "safe"),
1823 ("source", "cmdline"),
1824 ("fallback", "io_uring unavailable: ENOSYS"),
1825 ],
1826 &[],
1827 ),
1828 ("writer", "safe (io_uring unavailable: ENOSYS)", false),
1829 ),
1830 ];
1831 for (l, (kind, text, warn)) in cases {
1832 let got = describe(&l).unwrap_or_else(|| panic!("{:?} must be shown", l.event()));
1833 assert_eq!((got.0, got.1.as_str(), got.2), (kind, text, warn));
1834 }
1835 }
1836
1837 #[test]
1838 fn the_noise_stays_in_the_log() {
1839 let quiet = [
1840 line(
1841 Level::INFO,
1842 "gunnar::git",
1843 Some("git.upload_pack.advertised"),
1844 "served the advertisement",
1845 &[],
1846 &[],
1847 ),
1848 line(
1849 Level::INFO,
1850 "gunnar_server::serve",
1851 None,
1852 "issuer ready (persisted)",
1853 &[],
1854 &[],
1855 ),
1856 line(
1857 Level::INFO,
1858 "gunnar::audit",
1859 Some("grpc.request.done"),
1860 "done",
1861 &[],
1862 &[],
1863 ),
1864 line(
1865 Level::WARN,
1866 "hyper::proto",
1867 None,
1868 "some third-party warning",
1869 &[],
1870 &[],
1871 ),
1872 ];
1873 for l in &quiet {
1874 assert_eq!(describe(l), None, "{:?} / {:?}", l.event(), l.message);
1875 }
1876 }
1877
1878 #[test]
1879 fn numbers_format_for_a_narrow_cell() {
1880 assert_eq!(fmt_bytes(900), "900B");
1881 assert_eq!(fmt_bytes(12 * 1024), "12K");
1882 assert_eq!(fmt_bytes(3_300_000_000), "3.1G");
1883 assert_eq!(fmt_bytes(300 * 1024 * 1024 * 1024), "300G");
1884 assert_eq!(
1885 fmt_uptime(3 * 86_400 + 4 * 3600 + 12 * 60 + 7),
1886 "3d 04:12:07"
1887 );
1888 assert_eq!(fmt_uptime(59), "00:00:59");
1889 assert_eq!(clock(1_700_000_000), "22:13:20");
1890 assert_eq!(pad("ab", 4), "ab ");
1891 assert_eq!(pad("abcdef", 4), "abcd");
1892 assert_eq!(
1893 pad("åäö", 4),
1894 "åäö ",
1895 "padding counts characters, not bytes"
1896 );
1897 assert_eq!(
1898 clip(
1899 "32x AMD Ryzen Threadripper PRO 3975WX 32-Cores",
1900 CPU_MODEL_COLS
1901 ),
1902 "32x AMD Ryzen Threadrippe\u{2026}"
1903 );
1904 assert_eq!(
1905 clip("2x AMD EPYC 7542 32-Core", CPU_MODEL_COLS),
1906 "2x AMD EPYC 7542 32-Core"
1907 );
1908 assert_eq!(fmt_bind("0.0.0.0:2222".parse().unwrap()), "*:2222");
1909 assert_eq!(fmt_bind("[::]:50051".parse().unwrap()), "*:50051");
1910 assert_eq!(
1911 fmt_bind("127.0.0.1:50051".parse().unwrap()),
1912 "127.0.0.1:50051"
1913 );
1914 }
1915
1916 #[test]
1922 fn the_index_row_says_what_the_indexer_is_doing_and_the_events_read_as_rows() {
1923 use crate::storage::IndexSnapshot;
1924 let snap = |done: u64, phase: &'static str, resumed: bool, pauses: u64| IndexSnapshot {
1925 active: true,
1926 resumed,
1927 pack_id: 3,
1928 objects_done: done,
1929 objects_total: 13_948_244,
1930 bytes_done: done * 600,
1931 bytes_total: 8_700_000_000,
1932 phase,
1933 started_at_ms: 1,
1934 reinflations: 0,
1935 pauses,
1936 };
1937 assert_eq!(
1938 index_row_text(
1939 "alice/linux",
1940 &snap(4_700_000, "resolve", false, 0),
1941 Some(2_100_000_000)
1942 ),
1943 "alice/linux 33% 4.7M/13.9M resolve 2.0G/s"
1944 );
1945 assert_eq!(
1946 index_row_text("alice/linux", &snap(0, "walk", true, 0), None),
1947 "alice/linux resumed 0% 0/13.9M walk"
1948 );
1949 assert_eq!(
1950 index_row_text("a/b", &snap(13_948_244, "fold", false, 2), None),
1951 "a/b 100% 13.9M/13.9M fold 2p"
1952 );
1953
1954 let mut s = Screen::new(facts());
1957 s.now_unix = 1_700_000_000;
1958 let dir = PathBuf::from("/data/repos/alice/linux.git");
1959 let t0 = Instant::now();
1960 let f1 = probe::IndexFacts {
1961 git_dir: dir.clone(),
1962 snap: snap(1_000_000, "resolve", false, 0),
1963 };
1964 s.sample_index(&[f1], t0);
1965 assert_eq!(s.index.len(), 1);
1966 assert_eq!(s.index[0].kind, "index");
1967 assert_eq!(s.index[0].text, "alice/linux 7% 1.0M/13.9M resolve");
1968 assert!(!s.index[0].warn);
1969 let f2 = probe::IndexFacts {
1970 git_dir: dir.clone(),
1971 snap: snap(3_000_000, "paused", false, 1),
1972 };
1973 s.sample_index(&[f2], t0 + Duration::from_secs(2));
1974 assert_eq!(
1975 s.index[0].text,
1976 "alice/linux 21% 3.0M/13.9M paused 572M/s 1p"
1977 );
1978 assert!(s.index[0].warn, "a paused pack is a warning row");
1979 let mut idle = snap(3_000_000, "done", false, 1);
1980 idle.active = false;
1981 s.sample_index(
1982 &[probe::IndexFacts {
1983 git_dir: dir,
1984 snap: idle,
1985 }],
1986 t0 + Duration::from_secs(4),
1987 );
1988 assert!(s.index.is_empty(), "an idle counter draws no row");
1989 s.sample_index(
1991 &[probe::IndexFacts {
1992 git_dir: PathBuf::from("/data/repos/a/b.git"),
1993 snap: snap(5, "walk", false, 0),
1994 }],
1995 t0,
1996 );
1997 s.push(Row {
1998 at: 1_700_000_000,
1999 kind: "login",
2000 text: "x".into(),
2001 warn: false,
2002 });
2003 let frame = s.render_frame(80, 25);
2004 let lines: Vec<&str> = frame.split("\r\n").collect();
2005 assert!(
2006 lines[5].contains("index a/b 0% 5/13.9M walk"),
2007 "row 5: {:?}",
2008 lines[5]
2009 );
2010 assert!(lines[6].contains("login x"), "row 6: {:?}", lines[6]);
2011
2012 let paused = line(
2014 Level::WARN,
2015 "gunnar::index",
2016 Some("index.paused"),
2017 "index paused",
2018 &[
2019 ("available", "3113851289"),
2020 ("budget", "3435973836"),
2021 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2022 ],
2023 &[],
2024 );
2025 assert_eq!(
2026 describe(&paused),
2027 Some((
2028 "index",
2029 "paused: 2.9G available < 3.2G budget alice/linux".to_string(),
2030 true
2031 ))
2032 );
2033 let resumed = line(
2034 Level::INFO,
2035 "gunnar::index",
2036 Some("index.resumed"),
2037 "index resumed",
2038 &[
2039 ("available", "5368709120"),
2040 ("budget", "3435973836"),
2041 ("paused_ms", "8120"),
2042 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2043 ],
2044 &[],
2045 );
2046 assert_eq!(
2047 describe(&resumed),
2048 Some((
2049 "index",
2050 "resumed after 8120 ms: 5.0G available alice/linux".to_string(),
2051 false
2052 ))
2053 );
2054 let finished = line(
2055 Level::INFO,
2056 "gunnar::index",
2057 Some("index.finished"),
2058 "indexed",
2059 &[
2060 ("objects", "13948244"),
2061 ("total_ms", "2520000"),
2062 ("reinflations", "1234"),
2063 ("pauses", "2"),
2064 ("peak_rss", "9126805504"),
2065 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2066 ],
2067 &[],
2068 );
2069 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)));
2070 let progress = line(
2073 Level::INFO,
2074 "gunnar::index",
2075 Some("index.progress"),
2076 "index",
2077 &[
2078 ("percent", "78"),
2079 ("objects_done", "11000000"),
2080 ("objects_total", "13948244"),
2081 ("phase", "resolve"),
2082 ("objects_per_s", "744000"),
2083 ("reinflations", "12"),
2084 ("pauses", "1"),
2085 ("peak_rss", "13958643712"),
2086 ("mem_available", "3286511616"),
2087 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2088 ],
2089 &[],
2090 );
2091 assert_eq!(
2092 describe(&progress),
2093 Some((
2094 "index",
2095 "alice/linux 78% 11.0M/13.9M resolve 744k/s 12 re-inflations 1 pauses rss 13G avail 3.1G".to_string(),
2096 false
2097 ))
2098 );
2099 let gave_up = line(
2100 Level::WARN,
2101 "gunnar::index",
2102 Some("index.gave_up"),
2103 "gave up",
2104 &[
2105 ("reason", "nothing-to-give"),
2106 ("paused_ms", "31200"),
2107 ("available", "3113851289"),
2108 ("budget", "3435973836"),
2109 ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
2110 ],
2111 &[],
2112 );
2113 assert_eq!(
2114 describe(&gave_up),
2115 Some((
2116 "index",
2117 "GAVE UP (nothing-to-give) after 31200 ms: 2.9G available, 3.2G needed alice/linux".to_string(),
2118 true
2119 ))
2120 );
2121 let purged = line(
2122 Level::INFO,
2123 "gunnar::store",
2124 Some("repo.purged"),
2125 "reclaimed",
2126 &[
2127 ("repo", "alice/linux"),
2128 ("bytes_before", "26843545600"),
2129 ("bytes_after", "0"),
2130 ],
2131 &[],
2132 );
2133 assert_eq!(
2134 describe(&purged),
2135 Some(("delete", "alice/linux 25G reclaimed".to_string(), false))
2136 );
2137 assert_eq!(fmt_count(812), "812");
2138 assert_eq!(fmt_count(45_000), "45k");
2139 assert_eq!(fmt_ms(3_700_000), "1.0h");
2140 assert_eq!(
2141 repo_label(
2142 std::path::Path::new("/data/repos/ns/x.git"),
2143 std::path::Path::new("/data")
2144 ),
2145 "ns/x"
2146 );
2147 }
2148
2149 fn facts() -> Facts {
2150 Facts {
2151 ssh_bind: "0.0.0.0:2222".parse().unwrap(),
2152 grpc_bind: Some("0.0.0.0:50051".parse().unwrap()),
2153 data_root: PathBuf::from("/data"),
2154 build: "v1.1.3-145-g1eb7083".into(),
2155 engine: None,
2156 replication: crate::replication_view::ReplicationView::new(),
2157 tenants: TenantView::new(),
2158 flow: FlowView::new(),
2159 }
2160 }
2161
2162 #[test]
2169 fn the_tenants_row_is_drawn_when_filled_and_absent_when_not() {
2170 let mut s = Screen::new(facts());
2171 s.now_unix = 1_000;
2172 s.sample_tenants(None);
2173 let frame = s.render_frame(80, 24);
2174 assert!(
2175 !frame.contains("tenants"),
2176 "an unfilled view must draw no tenants row:\n{frame}"
2177 );
2178 s.sample_tenants(Some(TenantCounts {
2179 paid: 118,
2180 grace: 4,
2181 suspended: 3,
2182 }));
2183 let frame = s.render_frame(80, 24);
2184 assert!(
2185 frame.contains("tenants paid 118 grace 4 suspended 3"),
2186 "the row names all three counts:\n{frame}"
2187 );
2188 assert!(
2190 frame.contains("!00:16:40 tenants"),
2191 "suspended > 0 is a `!` row:\n{frame}"
2192 );
2193 s.sample_tenants(Some(TenantCounts {
2194 paid: 1,
2195 grace: 0,
2196 suspended: 0,
2197 }));
2198 let frame = s.render_frame(80, 24);
2199 assert!(
2200 frame.contains(" 00:16:40 tenants paid 1 grace 0 suspended 0"),
2201 "nobody suspended is an ordinary row:\n{frame}"
2202 );
2203 }
2204
2205 #[test]
2213 fn a_standby_that_has_never_synced_is_a_warn_row_on_the_band() {
2214 use crate::replication_view::{ReplicationReport, StandbyRow};
2215 let mut s = Screen::new(facts());
2216 s.now_unix = 1_000;
2217 s.sample_replication(&ReplicationReport {
2218 configured: true,
2219 failed: 8,
2220 standbys: vec![StandbyRow {
2221 name: "standby".into(),
2222 remote: "ssh://git@standby:2222".into(),
2223 last_success_unix_ms: 0,
2224 stale: true,
2225 ..StandbyRow::default()
2226 }],
2227 ..ReplicationReport::default()
2228 });
2229 let frame = s.render_frame(100, 25);
2230 assert!(
2231 frame.contains("last successful replication: NEVER"),
2232 "the band must say it in those words:\n{frame}"
2233 );
2234 assert!(
2235 frame.contains("!00:16:40 standby"),
2236 "a stale standby must be a WARN row (`!`), not an ordinary one:\n{frame}"
2237 );
2238
2239 let mut s = Screen::new(facts());
2242 s.now_unix = 1_000;
2243 s.sample_replication(&ReplicationReport {
2244 configured: true,
2245 synced: 105,
2246 failed: 40,
2247 standbys: vec![StandbyRow {
2248 name: "standby".into(),
2249 last_success_unix_ms: 997_000,
2250 stale: false,
2251 ..StandbyRow::default()
2252 }],
2253 ..ReplicationReport::default()
2254 });
2255 let frame = s.render_frame(100, 25);
2256 assert!(!frame.contains("NEVER"), "{frame}");
2257 assert!(frame.contains(" 00:16:40 standby"), "{frame}");
2258 assert!(
2259 frame.contains("synced 105 failed 40 last 00:00:03 ago"),
2260 "{frame}"
2261 );
2262
2263 let mut s = Screen::new(facts());
2267 s.now_unix = 1_000;
2268 s.sample_replication(&ReplicationReport {
2269 configured: true,
2270 ..ReplicationReport::default()
2271 });
2272 assert!(s.render_frame(100, 25).contains("reports NO standby rows"));
2273
2274 let mut s = Screen::new(facts());
2277 s.now_unix = 1_000;
2278 s.sample_replication(&ReplicationReport::default());
2279 assert!(s.standby.is_empty());
2280 }
2281
2282 #[test]
2283 fn a_frame_is_exactly_rows_lines_of_width_minus_one_and_the_newest_events_win() {
2284 let mut s = Screen::new(facts());
2285 s.hostname = Some("gunnar-appliance".into());
2286 s.addresses = vec![probe::IfAddr {
2287 interface: "eth0".into(),
2288 address: "198.51.100.155".parse().unwrap(),
2289 prefix: 24,
2290 }];
2291 s.memory = Some((16 * 1024 * 1024 * 1024, 15 * 1024 * 1024 * 1024));
2292 s.disk = Some((300 * 1024 * 1024 * 1024, 297 * 1024 * 1024 * 1024));
2293 s.uptime_secs = Some(3 * 86_400 + 4 * 3600 + 12 * 60 + 7);
2294 s.cpus = 4;
2295 s.cpu_model = Some("AMD EPYC 7542".into());
2296 s.disk_kind = Some(probe::DiskKind {
2297 fs: "xfs".into(),
2298 set: Some(probe::SetFacts {
2299 level: "raid0".into(),
2300 members: 4,
2301 member_bytes: 75 * 1024 * 1024 * 1024,
2302 chunk_bytes: 524288,
2303 }),
2304 });
2305 s.cpu_percent = Some(12);
2306 s.load1 = Some("0.10".into());
2307 s.now_unix = 1_700_000_000;
2308 for i in 0..(EVENT_ROWS + 5) {
2309 s.push(Row {
2310 at: i as u64,
2311 kind: "login",
2312 text: format!("user{i}"),
2313 warn: i % 7 == 0,
2314 });
2315 }
2316 let frame = s.render_frame(80, 25);
2317 assert!(
2318 frame.starts_with("\x1b[H"),
2319 "homes the cursor, never clears"
2320 );
2321 let raw: Vec<&str> = frame["\x1b[H".len()..].split("\r\n").collect();
2322 assert_eq!(raw.len(), 25);
2323 assert!(
2324 raw.iter().all(|l| l.ends_with("\x1b[K")),
2325 "every row ends with erase-to-EOL so column 80 is cleared: {:?}",
2326 raw.iter()
2327 .map(|l| &l[l.len().saturating_sub(4)..])
2328 .collect::<Vec<_>>()
2329 );
2330 let lines: Vec<&str> = raw
2331 .iter()
2332 .map(|l| l.strip_suffix("\x1b[K").unwrap())
2333 .collect();
2334 assert!(
2335 lines.iter().all(|l| l.chars().count() == 79),
2336 "every row is padded to cols-1: {:?}",
2337 lines.iter().map(|l| l.chars().count()).collect::<Vec<_>>()
2338 );
2339 assert!(
2340 lines[0].contains("gunnar v1.1.3-145-g1eb7083")
2341 && lines[0].contains("up 3d 04:12:07")
2342 && lines[0].contains("22:13:20 UTC"),
2343 "{}",
2344 lines[0]
2345 );
2346 assert!(
2347 lines[1].contains("host gunnar-appliance eth0 198.51.100.155/24"),
2348 "{}",
2349 lines[1]
2350 );
2351 assert!(
2352 lines[2].contains("cpu 4x AMD EPYC 7542 12% load 0.10 ram 1.0G/16G 6%"),
2353 "{}",
2354 lines[2]
2355 );
2356 assert!(
2357 lines[3].contains(
2358 "disk /data 3.0G/300G 1% 4\u{d7}75G raid0 xfs ssh *:2222 grpc *:50051"
2359 ),
2360 "{}",
2361 lines[3]
2362 );
2363 s.writer = Some("uring");
2365 let frame = s.render_frame(80, 25);
2366 let l3 = frame["\x1b[H".len()..].split("\r\n").nth(3).unwrap();
2367 assert!(
2368 l3.contains("4\u{d7}75G raid0 xfs uring ssh *:2222 grpc *:50051"),
2369 "{l3}"
2370 );
2371 assert_eq!(
2372 l3.strip_suffix("\x1b[K").unwrap().chars().count(),
2373 79,
2374 "still fits: {l3}"
2375 );
2376 s.writer = None;
2377 assert!(lines[4].starts_with("\u{2500}\u{2500}\u{2500}"));
2378 assert_eq!(s.rows.len(), EVENT_ROWS, "the ring holds EVENT_ROWS");
2379 assert!(
2380 lines[5].contains("user5"),
2381 "the oldest kept row is the first body row: {}",
2382 lines[5]
2383 );
2384 assert!(
2385 lines[24].contains("user24"),
2386 "the newest row is the last: {}",
2387 lines[24]
2388 );
2389 assert!(
2390 lines[7].starts_with('!'),
2391 "a warning row is marked: {}",
2392 lines[7]
2393 );
2394 assert!(
2395 lines[5].starts_with(' '),
2396 "a plain row is not: {}",
2397 lines[5]
2398 );
2399 }
2400
2401 #[test]
2402 fn a_dark_screen_says_dash_not_zero() {
2403 let s = Screen::new(facts());
2406 let frame = s.render_frame(80, 25);
2407 assert!(
2408 frame.contains("up -")
2409 && frame.contains("host (none) no address")
2410 && frame.contains("cpu - -% load - ram -")
2411 && frame.contains("disk /data - ssh *:2222"),
2412 "{frame}"
2413 );
2414 }
2415
2416 #[test]
2417 fn the_disk_cell_says_what_data_is_from_fixture_mounts_and_sysfs() {
2418 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";
2421 let md = probe::parse_mounts(MOUNTS, "/data").unwrap();
2422 assert_eq!(
2423 md,
2424 probe::MountFacts {
2425 source: "/dev/md/korp-data".into(),
2426 fstype: "xfs".into()
2427 }
2428 );
2429 let fat = probe::parse_mounts(MOUNTS, "/mnt/fat box").unwrap();
2430 assert_eq!(fat.fstype, "vfat");
2431 assert_eq!(
2432 probe::parse_mounts(MOUNTS, "/tmp/x"),
2433 None,
2434 "a plain directory is no mount"
2435 );
2436
2437 let dir = tempfile::tempdir().unwrap();
2440 let mdir = dir.path().join("md");
2441 std::fs::create_dir_all(&mdir).unwrap();
2442 std::fs::write(mdir.join("level"), "raid0\n").unwrap();
2443 std::fs::write(mdir.join("raid_disks"), "4\n").unwrap();
2444 std::fs::write(mdir.join("chunk_size"), "524288\n").unwrap();
2445 for m in ["vdb1", "vdc1", "vdd1", "vde1"] {
2446 let d = mdir.join(format!("dev-{m}"));
2447 std::fs::create_dir_all(&d).unwrap();
2448 std::fs::write(d.join("size"), "78643200\n").unwrap();
2449 }
2450 let set = probe::read_set_facts(dir.path()).unwrap();
2451 assert_eq!(
2452 set,
2453 probe::SetFacts {
2454 level: "raid0".into(),
2455 members: 4,
2456 member_bytes: 75 * 1024 * 1024 * 1024,
2457 chunk_bytes: 524288
2458 }
2459 );
2460 assert_eq!(
2461 probe::DiskKind::from_mount(&md, Some(set)).to_string(),
2462 "4\u{d7}75G raid0 xfs"
2463 );
2464 assert_eq!(probe::DiskKind::from_mount(&fat, None).to_string(), "fat");
2465 assert_eq!(
2466 probe::DiskKind::from_mount(&md, None).to_string(),
2467 "xfs",
2468 "a plain xfs disk names only the filesystem"
2469 );
2470
2471 let half = tempfile::tempdir().unwrap();
2473 std::fs::create_dir_all(half.path().join("md")).unwrap();
2474 std::fs::write(half.path().join("md/level"), "raid0\n").unwrap();
2475 std::fs::write(half.path().join("md/raid_disks"), "4\n").unwrap();
2476 assert_eq!(probe::read_set_facts(half.path()), None);
2477 assert_eq!(probe::read_set_facts(&dir.path().join("nowhere")), None);
2478 }
2479
2480 #[test]
2481 fn the_address_line_does_not_overflow_the_width() {
2482 let mut s = Screen::new(facts());
2483 for i in 0..12u8 {
2484 s.addresses.push(probe::IfAddr {
2485 interface: format!("eth{i}"),
2486 address: format!("10.0.{i}.15").parse().unwrap(),
2487 prefix: 24,
2488 });
2489 }
2490 let frame = s.render_frame(80, 25);
2491 let line2 = frame.split("\r\n").nth(1).unwrap();
2492 assert!(
2493 line2.contains("eth0 10.0.0.15/24") && !line2.contains("eth11"),
2494 "{line2}"
2495 );
2496 }
2497}