hulthe/compost
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Implement stuff

Browse Source
This commit is contained in:
Joakim Hulthe
2022-04-19 01:04:59 +02:00
parent b9e6eeeb80
commit 7ff5f62413
6 changed files with 208 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
src/collector.rs
View File

@ -2,13 +2,14 @@ use crate::docker;
use crate::state::{StackStats, StateEvent};
use std::collections::HashMap;
use tokio::sync::mpsc;
use tokio::task;
use tokio::time::{sleep, Duration};
pub(crate) async fn start_collector(events: mpsc::Sender<StateEvent>) -> anyhow::Result<()> {
let mut old_stacks: HashMap<String, StackStats> = HashMap::new();
loop {
let new_stacks = collect_data()?;
let new_stacks = collect_data().await?;
for (name, stats) in &new_stacks {
let mut send_put = false;
@ -44,42 +45,51 @@ pub(crate) async fn start_collector(events: mpsc::Sender<StateEvent>) -> anyhow:
}
}
pub(crate) fn collect_data() -> anyhow::Result<HashMap<String, StackStats>> {
let docker_stacks = docker::list_stacks()?;
let mut out = HashMap::new();
pub(crate) async fn collect_data() -> anyhow::Result<HashMap<String, StackStats>> {
let docker_stacks = docker::list_stacks().await?;
let mut stack_jobs = Vec::new();
for docker_stack in docker_stacks {
let containers = docker::list_containers(&docker_stack)?;
let processes = containers
.iter()
.map(|container| docker::list_processes(&docker_stack, &container))
.flatten(/* ignore errors */)
.flatten(/* flatten per-container process list */)
.collect::<Vec<_>>();
stack_jobs.push(task::spawn(collect_stack_data(docker_stack)));
}
let memory = (1usize << 20) * 16;
let memory_usage = processes.iter().map(|proc| proc.memory_usage).sum();
let mut out = HashMap::with_capacity(stack_jobs.len());
let container_count = containers.len() as u32;
let running_containers = containers
.iter()
.filter(|c| c.state.contains("running"))
.count() as u32;
let stopped_containers = container_count - running_containers;
let stats = StackStats {
containers: container_count,
running_containers,
stopped_containers,
process_count: processes.len() as u32,
cpu_percent: 0.5, // TODO
memory_usage,
memory_percent: (memory_usage as f64 / memory as f64),
};
out.insert(docker_stack.name, stats);
for stack_job in stack_jobs {
let (name, stats) = stack_job.await??;
out.insert(name, stats);
}
Ok(out)
}
async fn collect_stack_data(docker_stack: docker::Stack) -> anyhow::Result<(String, StackStats)> {
let containers = docker::list_containers(&docker_stack).await?;
let mut processes = Vec::new();
for container in &containers {
processes.append(&mut docker::list_processes(&docker_stack, container).await?);
}
let memory = (1usize << 20) * 16;
let memory_usage = processes.iter().map(|proc| proc.memory_usage).sum();
let container_count = containers.len() as u32;
let running_containers = containers
.iter()
.filter(|c| c.state.contains("running"))
.count() as u32;
let stopped_containers = container_count - running_containers;
let stats = StackStats {
containers: container_count,
running_containers,
stopped_containers,
process_count: processes.len() as u32,
cpu_percent: 0.5, // TODO
memory_usage,
memory_percent: (memory_usage as f64 / memory as f64),
};
Ok((docker_stack.name, stats))
}

55
src/docker.rs
View File

@ -1,6 +1,7 @@
use serde::Deserialize;
use std::fs;
use std::process::Command;
use tokio::fs;
use tokio::process::Command;
use tokio::task;
#[derive(Deserialize)]
pub struct Stack {
@ -33,21 +34,23 @@ pub struct Container {
}
/// Run `docker compose ls` and parse the output
pub fn list_stacks() -> anyhow::Result<Vec<Stack>> {
pub async fn list_stacks() -> anyhow::Result<Vec<Stack>> {
let output = Command::new("docker")
.args(&["compose", "ls", "--format", "json"])
.output()?;
.output()
.await?;
let stdout = std::str::from_utf8(&output.stdout)?;
Ok(serde_json::from_str(&stdout)?)
}
/// Run `docker compose ps` and parse the output
pub fn list_containers(stack: &Stack) -> anyhow::Result<Vec<Container>> {
pub async fn list_containers(stack: &Stack) -> anyhow::Result<Vec<Container>> {
let output = Command::new("docker")
.arg("compose")
.args(&["--file", &stack.config_file])
.args(&["ps", "--format", "json"])
.output()?;
.output()
.await?;
let stdout = std::str::from_utf8(&output.stdout)?;
Ok(serde_json::from_str(&stdout)?)
@ -64,16 +67,19 @@ pub struct Process {
}
/// Run `docker compose top` and parse the output
pub fn list_processes(stack: &Stack, container: &Container) -> anyhow::Result<Vec<Process>> {
pub async fn list_processes(stack: &Stack, container: &Container) -> anyhow::Result<Vec<Process>> {
let output = Command::new("docker")
.arg("compose")
.args(&["--file", &stack.config_file])
.args(&["top", &container.service])
.output()?;
.output()
.await?;
let stdout = std::str::from_utf8(&output.stdout)?;
let mut processes = Vec::new();
let mut proc_info_set = Vec::new();
for line in stdout.lines().skip(2) {
if line.trim().is_empty() {
continue;
@ -91,29 +97,32 @@ pub fn list_processes(stack: &Stack, container: &Container) -> anyhow::Result<Ve
let _time = words.next().ok_or_else(err)?;
let cmd: String = words.collect();
let mut memory_usage = 0;
let proc_info = fs::read_to_string(format!("/proc/{pid}/status"))?;
for (key, value) in proc_info.lines().flat_map(|line| line.split_once(':')) {
let value = value.trim();
match key {
"VmRSS" => {
memory_usage = value.trim_end_matches(" kB").parse()?;
}
_ => {}
}
}
proc_info_set.push(task::spawn(fs::read_to_string(format!(
"/proc/{pid}/status"
))));
processes.push(Process {
uid,
pid,
ppid,
cmd,
memory_usage,
memory_usage: 0,
})
}
for (process, proc_info) in processes.iter_mut().zip(proc_info_set.into_iter()) {
let proc_info = proc_info.await??;
for (key, value) in proc_info.lines().flat_map(|line| line.split_once(':')) {
let value = value.trim();
match key {
"VmRSS" => {
process.memory_usage = value.trim_end_matches(" kB").parse()?;
}
_ => {}
}
}
}
Ok(processes)
}

5
src/main.rs
View File

@ -17,10 +17,13 @@ async fn main() -> anyhow::Result<()> {
let mut ui = Ui::new(event_rx);
if let Err(e) = task::spawn_blocking(move || ui.start()).await? {
println!("{e}");
println!("Error: {e}");
}
collector.abort();
if let Ok(Err(e)) = collector.await {
println!("Error: {e}");
}
Ok(())
}

156
src/ui.rs
View File

@ -7,7 +7,7 @@ use crossterm::{
};
use std::collections::BTreeMap;
use std::time::Duration;
use std::{io, iter};
use std::{cmp, io, iter};
use tokio::sync::mpsc::{self, error::TryRecvError};
use tui::{
backend::{Backend, CrosstermBackend},
@ -19,15 +19,24 @@ use tui::{
Frame, Terminal,
};
const KEYS_DOWN: &[KeyCode] = &[KeyCode::Down, KeyCode::PageDown, KeyCode::Char('j')];
const KEYS_UP: &[KeyCode] = &[KeyCode::Up, KeyCode::PageUp, KeyCode::Char('k')];
pub struct Ui {
stacks: BTreeMap<String, StackStats>,
events: mpsc::Receiver<StateEvent>,
event_log: [String; 7],
event_count: usize,
scroll: usize,
}
impl Ui {
pub fn new(events: mpsc::Receiver<StateEvent>) -> Self {
Self {
stacks: Default::default(),
scroll: 0,
event_log: Default::default(),
event_count: 0,
events,
}
}
@ -63,14 +72,20 @@ impl Ui {
let mut draw = false;
let timeout = Duration::from_millis(16);
if crossterm::event::poll(timeout)? {
draw = true;
if let Event::Key(key) = event::read()? {
draw = true;
if let KeyCode::Char('q') = key.code {
return Ok(());
} else if KEYS_UP.contains(&key.code) {
self.scroll = self.scroll.saturating_sub(1);
} else if KEYS_DOWN.contains(&key.code) {
self.scroll += 1;
}
}
}
self.scroll = cmp::min(self.stacks.len(), self.scroll);
match self.events.try_recv() {
Err(TryRecvError::Empty) => {}
Err(e) => return Err(e.into()),
@ -88,53 +103,116 @@ impl Ui {
}
fn handle_event(&mut self, event: StateEvent) {
let log_msg;
match event {
StateEvent::Delete { name } => {
log_msg = format!("{:.4}: DELETE {name}", self.event_count);
self.stacks.remove(&name);
}
StateEvent::Put { name, stats } => {
log_msg = format!("{:.4}: UPDATE {name}", self.event_count);
self.stacks.insert(name, stats);
}
}
self.event_log.rotate_left(1);
self.event_log[self.event_log.len() - 1] = log_msg;
self.event_count += 1;
}
fn draw<B: Backend>(&self, f: &mut Frame<'_, B>) {
let size = f.size();
const BOX_HEIGHT: u16 = 9;
let fitted_boxes = size.height / BOX_HEIGHT;
const BOX_WIDTH: u16 = 52;
let fitted_boxes_x = cmp::max(1, size.width / BOX_WIDTH);
let fitted_boxes_y = size.height / BOX_HEIGHT;
let partial_box_size = size.height % BOX_HEIGHT;
let partial_box_exists = partial_box_size != 0;
let constraints: Vec<_> = iter::repeat(Constraint::Length(BOX_HEIGHT))
.take(fitted_boxes as usize)
.chain(partial_box_exists.then(|| Constraint::Length(partial_box_size)))
.collect();
let x_constraints = vec![Constraint::Length(BOX_WIDTH); fitted_boxes_x as usize];
let chunks = Layout::default()
let x_layout = Layout::default()
.direction(Direction::Horizontal)
.constraints(x_constraints);
let y_constraints: Vec<_> = iter::repeat(Constraint::Length(BOX_HEIGHT))
.take(fitted_boxes_y as usize)
.chain(iter::once(Constraint::Length(0))) // padding
//.chain(partial_box_exists.then(|| Constraint::Length(partial_box_size)))
.collect();
let y_chunks = Layout::default()
.direction(Direction::Vertical)
.constraints(constraints)
.constraints(y_constraints)
.split(f.size());
for (i, (name, info)) in (0..fitted_boxes).zip(self.stacks.iter()) {
let area = chunks[i as usize];
//self.draw_info(f, chunks[0]);
self.draw_stack(f, area, name, info);
}
let mut stacks = self
.stacks
.iter()
.skip((self.scroll * fitted_boxes_x as usize).saturating_sub(1));
if partial_box_exists {
let block = Block::default()
.title("Partial")
.borders(Borders::ALL.difference(Borders::BOTTOM));
f.render_widget(block, chunks[chunks.len() - 1]);
let mut first = self.scroll == 0;
'outer: for &y_chunk in &y_chunks[..fitted_boxes_y as usize] {
for x_chunk in x_layout.split(y_chunk) {
if first {
first = false;
self.draw_info(f, x_chunk);
} else if let Some((name, info)) = stacks.next() {
self.draw_stack(f, x_chunk, name, info);
} else {
break 'outer;
}
}
}
}
fn draw_info<B: Backend>(&self, f: &mut Frame<B>, area: Rect) {
let block = Block::default().borders(Borders::ALL);
let inner = block.inner(area);
f.render_widget(block, area);
let chunks = Layout::default()
.direction(Direction::Horizontal)
.constraints([Constraint::Length(20), Constraint::Min(10)])
.split(inner);
let unhealthy_stacks = self
.stacks
.values()
.filter(|stack| stack.stopped_containers > 0)
.count();
let notices = Paragraph::new(vec![
Spans::from("Status"),
Spans::from(format!("stacks: {}", self.stacks.len())),
Spans::from(""),
if unhealthy_stacks > 0 {
let style = Style::default().fg(Color::Red);
Span::styled(format!("unhealthy: {}", unhealthy_stacks), style).into()
} else {
Spans::from("")
},
]);
f.render_widget(notices, chunks[0]);
let log_style = Style::default()
.fg(Color::LightBlue)
.add_modifier(Modifier::ITALIC);
let event_log = Paragraph::new(
self.event_log
.clone()
.map(|msg| Span::styled(msg, log_style))
.map(Spans::from)
.to_vec(),
);
f.render_widget(event_log, chunks[1]);
}
fn draw_stack<B: Backend>(&self, f: &mut Frame<B>, area: Rect, name: &str, info: &StackStats) {
let title_style = Style::default()
.fg(Color::Black)
.bg(Color::White)
.add_modifier(Modifier::BOLD);
let title_style = Style::default().fg(Color::LightMagenta).bg(Color::Black);
let block = Block::default()
.title(Span::styled(name, title_style))
@ -162,11 +240,11 @@ impl Ui {
)
.into()
} else {
Spans::from("")
Spans::default()
},
Spans::from(""),
Spans::default(),
Spans::from(format!("processes: {}", info.process_count)),
Spans::from(format!("memory: {} KBs", info.memory_usage)),
Spans::from(format!("memory: {}", fmt_kilobytes(info.memory_usage))),
]);
f.render_widget(services, chunks[0]);
@ -175,19 +253,28 @@ impl Ui {
.x_bounds([-5.0, 5.0])
.y_bounds([-5.0, 5.0])
.paint(move |ctx| {
let (bg, fg, scaled_percent) = if percent < 0.33 {
(Color::Blue, Color::Green, percent * 3.0)
} else if percent < 0.67 {
(Color::Green, Color::Yellow, (percent - 0.33) * 3.0)
} else {
(Color::Yellow, Color::Red, (percent - 0.67) * 3.0)
};
ctx.draw(&Circle {
r: 4.0,
color: Color::Blue,
color: bg,
..Default::default()
});
ctx.draw(&Circle {
r: 4.0,
color: Color::Green,
start: 360 - (percent * 360.0) as u16,
color: fg,
start: 360 - (scaled_percent * 360.0) as u16,
stop: 360,
..Default::default()
});
ctx.print(-0.5, 0.0, name);
ctx.print(-0.5, -2.0, format!("{:02.0}%", percent * 100.0));
})
};
@ -195,3 +282,16 @@ impl Ui {
f.render_widget(gauge_canvas(info.memory_percent, "MEM"), chunks[2]);
}
}
fn fmt_kilobytes<'a>(kbs: usize) -> String {
let gibi = 1 << 20;
let mebi = 1 << 10;
if kbs > gibi * 2 {
format!("{} GBs", kbs / gibi)
} else if kbs > mebi * 10 {
format!("{} MBs", kbs / mebi)
} else {
format!("{} KBs", kbs)
}
}