chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! What the runner logs of what it does (`knotless::telemetry`): its runs
//! from queued to ended, their steps and containers, its queues'
//! batches, and what went wrong.
use crate::{Subject, Watch};
use knotless::telemetry::{self, Fields, Level, millis};
use std::collections::HashMap;
use std::sync::{LazyLock, Mutex};
use std::time::{Duration, Instant};
/// Logs `message` about the run of `workflow` at `subject`: its ref,
/// commit and pull, and how it ended and how long it took, if it has.
pub(crate) fn run(
message: &str,
watch: &Watch,
subject: &Subject,
workflow: &str,
status: Option<&str>,
took: Option<Duration>,
) {
let fields = Fields {
repo: Some(&watch.label),
workflow: Some(workflow),
ref_name: Some(&subject.ref_name),
commit: Some(&subject.commit),
pull: subject.pull.as_ref().map(|p| p.uri.as_str()),
count: subject.batch.as_ref().map(|b| b.pulls.len() as u64),
status,
duration_ms: took.map(millis),
..Fields::default()
};
telemetry::note(Level::INFO, message, &fields);
}
/// Logs that `step` of `workflow` started, or, with its `status`, ended.
pub(crate) fn step(
commit: &str,
workflow: &str,
step: &str,
status: Option<&str>,
exit_code: Option<i32>,
took: Option<Duration>,
) {
let fields = Fields {
commit: Some(commit),
workflow: Some(workflow),
step: Some(step),
status,
exit_code: exit_code.map(i64::from),
duration_ms: took.map(millis),
..Fields::default()
};
let said = ["step started", "step ended"][usize::from(status.is_some())];
telemetry::note(Level::INFO, said, &fields);
}
/// Logs `message` about the container `id`, of `image` if known.
pub(crate) fn container(message: &str, id: &str, image: Option<&str>) {
let fields = Fields {
container: Some(&id[..id.len().min(12)]),
source: image,
..Fields::default()
};
telemetry::note(Level::INFO, message, &fields);
}
/// How often a failure going on is logged again, as `failed` with a count.
const REPEAT_EVERY: Duration = Duration::from_secs(60);
/// How long a failure not seen again is remembered: past it, it is logged
/// at once, as a new one.
const FORGET: Duration = Duration::from_secs(60 * 60);
/// The failures logged, by label and error: when last, and how many
/// times seen since without being logged.
type Failing = HashMap<(String, String), (Instant, u64)>;
static FAILING: LazyLock<Mutex<Failing>> = LazyLock::new(Mutex::default);
/// Logs, at warn, what went wrong with `label`'s work: the first time it
/// goes wrong so, then once each [`REPEAT_EVERY`] while it goes on, with
/// how many times it went wrong so meanwhile. Another error is logged at
/// once, and the same again once a success of `label`'s cleared it.
pub(crate) fn failed(label: &str, err: &anyhow::Error) {
let error = format!("{err:#}");
let seen = repeat(&mut FAILING.lock().unwrap(), label, &error, Instant::now());
let Some(repeats) = seen else {
return;
};
let fields = Fields {
repo: Some(label),
error: Some(error),
count: (repeats > 0).then_some(repeats),
..Fields::default()
};
telemetry::note(Level::WARN, "failed", &fields);
}
/// Forgets what went wrong with `label`'s work, which went right since.
pub(crate) fn succeeded(label: &str) {
FAILING.lock().unwrap().retain(|(l, _), _| l != label);
}
/// Whether `error` of `label`, at `now`, is to be logged, with how many
/// times it was seen since last it was; or `None`, counting it.
fn repeat(seen: &mut Failing, label: &str, error: &str, now: Instant) -> Option<u64> {
seen.retain(|_, (at, _)| now.duration_since(*at) < FORGET);
let key = (label.to_string(), error.to_string());
match seen.get_mut(&key) {
Some((at, skipped)) if now.duration_since(*at) < REPEAT_EVERY => {
*skipped += 1;
None
}
Some((at, skipped)) => {
let repeats = *skipped;
(*at, *skipped) = (now, 0);
Some(repeats)
}
None => {
seen.insert(key, (now, 0));
Some(0)
}
}
}
/// Logs that a pass over `label`, reconciling it, ended after `took`: at
/// debug, or info if it was slow, past `KNOTLESS_SLOW_MS`.
pub(crate) fn pass(label: &str, took: Duration, ok: bool) {
let fields = Fields {
repo: Some(label),
status: Some(["failed", "ok"][usize::from(ok)]),
duration_ms: Some(millis(took)),
..Fields::default()
};
let level = [Level::DEBUG, Level::INFO][usize::from(telemetry::slow(took))];
telemetry::note(level, "pass ended", &fields);
}
/// Logs, at warn, that a pass over `label` has gone on for `took`, and is
/// at `stage`.
pub(crate) fn long_pass(label: &str, stage: &str, took: Duration) {
let fields = Fields {
repo: Some(label),
stage: Some(stage),
duration_ms: Some(millis(took)),
..Fields::default()
};
telemetry::note(Level::WARN, "pass running long", &fields);
}
/// Logs that the disk is `short` of the runner's floor, as a run sheds
/// caches, or as it waits for room once it has.
pub(crate) fn short_of_disk(short: &str, shedding: bool) {
let fields = Fields {
reason: Some(short),
..Fields::default()
};
let message = if shedding {
"disk short: shedding caches"
} else {
"disk short: waiting for room"
};
telemetry::note(Level::WARN, message, &fields);
}
/// Logs `message` about a batch of `pulls` pulls into `branch` of
/// `label`, at `tip`, and why, if said.
pub(crate) fn batch(
message: &str,
label: &str,
branch: &str,
tip: &str,
pulls: usize,
why: Option<&str>,
) {
let fields = Fields {
repo: Some(label),
ref_name: Some(branch),
commit: Some(tip),
count: Some(pulls as u64),
reason: why,
..Fields::default()
};
telemetry::note(Level::INFO, message, &fields);
}
/// Logs that the pull `uri` was ejected from `label`'s queue, and why.
pub(crate) fn ejected(label: &str, uri: &str, why: &str) {
let fields = Fields {
repo: Some(label),
pull: Some(uri),
reason: Some(why),
..Fields::default()
};
telemetry::note(Level::INFO, "pull ejected", &fields);
}
/// Logs a look at `label`'s queue records, which took `took`: at debug,
/// or info if it found them changed.
pub(crate) fn noticed(label: &str, took: Duration, changed: bool) {
let fields = Fields {
repo: Some(label),
duration_ms: Some(millis(took)),
..Fields::default()
};
let level = [Level::DEBUG, Level::INFO][usize::from(changed)];
telemetry::note(level, "queue noticed", &fields);
}
/// Logs, at warn, a write to the PDS tried again after `wait`.
pub(crate) fn retried(err: &anyhow::Error, wait: Duration) {
let fields = Fields {
error: Some(format!("{err:#}")),
duration_ms: Some(millis(wait)),
..Fields::default()
};
telemetry::note(Level::WARN, "pds write retried", &fields);
}
/// Logs that Jetstream, at `url`, connected, or could not be or dropped.
pub(crate) fn feed(url: &str, connected: bool) {
let fields = Fields {
feed: Some("jetstream"),
source: Some(url),
connected: Some(connected),
..Fields::default()
};
let (level, said) =
[(Level::WARN, "feed down"), (Level::INFO, "feed connected")][usize::from(connected)];
telemetry::note(level, said, &fields);
}
/// Logs what the runner watches, as whom.
pub(crate) fn watching(repos: &[String], did: &str) {
let repos = repos.join(" ");
let fields = Fields {
repo: Some(&repos),
did: Some(did),
..Fields::default()
};
telemetry::note(Level::INFO, "watching", &fields);
}
/// Logs that `label` was found naming this runner, and is watched.
pub(crate) fn found(label: &str) {
let fields = Fields {
repo: Some(label),
..Fields::default()
};
telemetry::note(Level::INFO, "repository found", &fields);
}
/// Logs that `label`, found before, no longer names this runner.
pub(crate) fn left(label: &str) {
let fields = Fields {
repo: Some(label),
..Fields::default()
};
telemetry::note(Level::INFO, "repository left", &fields);
}
/// Logs that `gone` checks of `label` were deleted.
pub(crate) fn pruned(label: &str, gone: usize) {
let fields = Fields {
repo: Some(label),
count: Some(gone as u64),
..Fields::default()
};
telemetry::note(Level::INFO, "checks pruned", &fields);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn notes() {
// Nothing is set up to log here: each only has to run.
step("a/b", "test", "build", None, None, None);
step(
"a/b",
"test",
"build",
Some("success"),
Some(0),
Some(Duration::ZERO),
);
container("container started", "0123456789abcdef", Some("rust"));
container("container stopped", "short", None);
failed("a/b", &anyhow::anyhow!("no PDS"));
failed("a/b", &anyhow::anyhow!("no PDS"));
succeeded("a/b");
pass("a/b", Duration::ZERO, true);
pass("a/b", Duration::from_secs(5), false);
long_pass("a/b", "landing", Duration::from_secs(300));
batch("batch started", "a/b", "main", "abc", 2, None);
ejected("a/b", "at://did:plc:x/dev.knotless.pull/1", "conflicts");
noticed("a/b", Duration::ZERO, true);
retried(&anyhow::anyhow!("503"), Duration::from_secs(1));
feed("ws://127.0.0.1:9", false);
feed("ws://127.0.0.1:9", true);
watching(&["a/b".into()], "did:plc:x");
found("did:plc:x/b");
left("did:plc:x/b");
short_of_disk("1.0 GiB free", true);
short_of_disk("1.0 GiB free", false);
}
#[test]
fn failures_going_on_are_logged_now_and_then() {
let mut seen = Failing::new();
let start = Instant::now();
let at = |secs| start + Duration::from_secs(secs);
// The first time, at once; the same again only each minute, with
// how often it was seen meanwhile.
assert_eq!(repeat(&mut seen, "a/b", "gone", at(0)), Some(0));
assert_eq!(repeat(&mut seen, "a/b", "gone", at(5)), None);
assert_eq!(repeat(&mut seen, "a/b", "gone", at(10)), None);
assert_eq!(repeat(&mut seen, "a/b", "gone", at(60)), Some(2));
assert_eq!(repeat(&mut seen, "a/b", "gone", at(61)), None);
// Another error, or another repository's, is logged at once.
assert_eq!(repeat(&mut seen, "a/b", "refused", at(62)), Some(0));
assert_eq!(repeat(&mut seen, "c/d", "gone", at(62)), Some(0));
// One not seen for an hour is new again.
assert_eq!(repeat(&mut seen, "a/b", "gone", at(2 * 60 * 60)), Some(0));
assert_eq!(seen.len(), 1);
}
}