use core::fmt;
use shep_client::{
EventStream, LinkLost, LinkState, ReconnectingClient,
shep_core::protocol::{HostUsage, ProcessInfo, Request, Response, SelectorSpec, SheepRefusal},
};
use crate::error::Error;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Verb {
Start,
Stop,
Restart,
Reload,
Delete,
Flush,
Save,
Reopen,
}
struct Redacted;
impl fmt::Debug for Redacted {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "<redacted>")
}
}
pub struct Live(ReconnectingClient);
impl fmt::Debug for Live {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Live").field("socket", &Redacted).finish()
}
}
impl Live {
#[must_use]
pub fn new(client: ReconnectingClient) -> Self {
Self(client)
}
#[must_use]
pub fn link(&self) -> LinkState {
self.0.link()
}
pub async fn section(&self, name: &str) -> Result<String, Error> {
let asked = Request::DogConfig {
name: name.to_owned(),
};
match self.0.request(asked).await? {
Response::DogSection { toml } => Ok(toml.as_str().to_owned()),
other => Err(unexpected("a DogSection", &other)),
}
}
pub async fn flock(&self) -> Result<Vec<ProcessInfo>, Error> {
match self.0.request(Request::ListFlock).await? {
Response::Flock(sheep) => Ok(sheep),
other => Err(unexpected("a Flock", &other)),
}
}
pub async fn host_usage(&self) -> Result<Option<HostUsage>, Error> {
match self.0.request(Request::HostUsage).await? {
Response::HostUsage(usage) => Ok(usage),
other => Err(unexpected("a HostUsage", &other)),
}
}
pub async fn act(&self, verb: Verb, selector: SelectorSpec) -> Result<String, Error> {
match verb {
Verb::Start => self.restart(selector, "started").await,
Verb::Restart => self.restart(selector, "restarted").await,
Verb::Reload => match self.0.request(Request::Reload { selector }).await? {
Response::Reloading { accepted, refused } => {
Ok(walked("reloading", &accepted, &refused))
}
other => Err(unexpected("a Reloading", &other)),
},
Verb::Stop => match self.0.request(Request::Stop { selector }).await? {
Response::Stopped(sheep) => Ok(format!("stopped {}", name_list(&sheep))),
other => Err(unexpected("a Stopped", &other)),
},
Verb::Delete => {
let named = describe_selector(&selector);
match self.0.request(Request::Delete { selector }).await? {
Response::Deleted(_) => Ok(format!("deleted {named}")),
other => Err(unexpected("a Deleted", &other)),
}
}
Verb::Flush => match self.0.request(Request::Flush { selector }).await? {
Response::Flushed(sheep) => Ok(format!("flushed {}", name_list(&sheep))),
other => Err(unexpected("a Flushed", &other)),
},
Verb::Reopen => match self.0.request(Request::Reopen { selector }).await? {
Response::Reopened(sheep) => Ok(format!("reopened {}", name_list(&sheep))),
other => Err(unexpected("a Reopened", &other)),
},
Verb::Save => match self.0.request(Request::SaveRoll).await? {
Response::RollSaved { path, apps } => Ok(format!("saved {apps} apps to {path}")),
other => Err(unexpected("a RollSaved", &other)),
},
}
}
async fn restart(&self, selector: SelectorSpec, verb_word: &str) -> Result<String, Error> {
match self.0.request(Request::Restart { selector }).await? {
Response::Restarted { accepted, refused } => Ok(walked(verb_word, &accepted, &refused)),
other => Err(unexpected("a Restarted", &other)),
}
}
pub async fn connected_within(&self, budget: core::time::Duration) -> Result<(), LinkLost> {
self.0.connected_within(budget).await
}
pub async fn subscribe(&self, topics: Vec<String>) -> Result<EventStream, Error> {
Ok(self.0.subscribe(topics).await?)
}
}
fn unexpected(asked: &str, got: &Response) -> Error {
Error::Unexpected {
asked: asked.to_owned(),
got: named(got),
}
}
fn named(response: &Response) -> String {
match response {
Response::Pong => "Pong".to_owned(),
Response::Flock(sheep) => format!("a Flock of {}", sheep.len()),
Response::HostUsage(_) => "a HostUsage".to_owned(),
Response::Described(sheep) => format!("a Described of {}", sheep.len()),
Response::DogSection { .. } => "a DogSection".to_owned(),
Response::Stopped(sheep) => format!("a Stopped of {}", sheep.len()),
Response::Restarted { accepted, .. } => format!("a Restarted of {}", accepted.len()),
Response::Reloading { accepted, .. } => format!("a Reloading of {}", accepted.len()),
Response::Deleted(ids) => format!("a Deleted of {}", ids.len()),
Response::Flushed(sheep) => format!("a Flushed of {}", sheep.len()),
Response::Reopened(sheep) => format!("a Reopened of {}", sheep.len()),
Response::RollSaved { apps, .. } => format!("a RollSaved of {apps}"),
other => format!("{other:?}").chars().take(60).collect(),
}
}
fn walked(verb_word: &str, accepted: &[ProcessInfo], refused: &[SheepRefusal]) -> String {
let names = name_list(accepted);
if refused.is_empty() {
format!("{verb_word} {names}")
} else {
let reasons: Vec<String> = refused
.iter()
.map(|refusal| format!("{}: {}", refusal.name, refusal.reason))
.collect();
format!("{verb_word} {names}, refused {}", reasons.join(", "))
}
}
fn name_list(sheep: &[ProcessInfo]) -> String {
if sheep.is_empty() {
return "no sheep".to_owned();
}
sheep
.iter()
.map(|info| info.name.as_str())
.collect::<Vec<_>>()
.join(", ")
}
fn describe_selector(selector: &SelectorSpec) -> String {
match selector {
SelectorSpec::All => "all sheep".to_owned(),
SelectorSpec::Id(id) => id.to_string(),
SelectorSpec::Name(name) | SelectorSpec::Fold(name) => name.clone(),
SelectorSpec::Regex(pattern) => pattern.clone(),
SelectorSpec::Instance { name, slot } => format!("{name}.{slot}"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::test_support::{assert_no_dashes, sample, test_live};
use shep_client::{ReconnectingClient, testing};
#[tokio::test]
async fn start_asks_a_restart_and_still_says_started() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Restart {
selector: SelectorSpec::Name("web".to_owned()),
})
.answer(Response::Restarted {
accepted: vec![sample("web")],
refused: Vec::new(),
});
let reply = live
.act(Verb::Start, SelectorSpec::Name("web".to_owned()))
.await
.expect("ok");
assert_eq!(reply, "started web");
}
#[tokio::test]
async fn restart_names_the_sheep_in_its_reply() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Restart {
selector: SelectorSpec::Name("web".to_owned()),
})
.answer(Response::Restarted {
accepted: vec![sample("web")],
refused: Vec::new(),
});
let reply = live
.act(Verb::Restart, SelectorSpec::Name("web".to_owned()))
.await
.expect("ok");
assert_eq!(reply, "restarted web");
}
#[tokio::test]
async fn a_refused_restart_names_the_refusal_too() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Restart {
selector: SelectorSpec::Fold("prod".to_owned()),
})
.answer(Response::Restarted {
accepted: vec![sample("web")],
refused: vec![SheepRefusal::new("worker", "exceeded restart budget")],
});
let reply = live
.act(Verb::Restart, SelectorSpec::Fold("prod".to_owned()))
.await
.expect("ok");
assert_eq!(
reply,
"restarted web, refused worker: exceeded restart budget"
);
}
#[tokio::test]
async fn reload_says_reloading_not_reloaded() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Reload {
selector: SelectorSpec::Name("web".to_owned()),
})
.answer(Response::Reloading {
accepted: vec![sample("web")],
refused: Vec::new(),
});
let reply = live
.act(Verb::Reload, SelectorSpec::Name("web".to_owned()))
.await
.expect("ok");
assert_eq!(reply, "reloading web");
}
#[tokio::test]
async fn stop_names_the_sheep_it_stopped() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Stop {
selector: SelectorSpec::Name("web".to_owned()),
})
.answer(Response::Stopped(vec![sample("web")]));
let reply = live
.act(Verb::Stop, SelectorSpec::Name("web".to_owned()))
.await
.expect("ok");
assert_eq!(reply, "stopped web");
}
#[tokio::test]
async fn delete_names_what_was_asked_since_ids_carry_no_name() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Delete {
selector: SelectorSpec::Name("web".to_owned()),
})
.answer(Response::Deleted(vec![7]));
let reply = live
.act(Verb::Delete, SelectorSpec::Name("web".to_owned()))
.await
.expect("ok");
assert_eq!(reply, "deleted web");
}
#[tokio::test]
async fn flush_names_the_sheep_it_flushed() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Flush {
selector: SelectorSpec::Name("web".to_owned()),
})
.answer(Response::Flushed(vec![sample("web")]));
let reply = live
.act(Verb::Flush, SelectorSpec::Name("web".to_owned()))
.await
.expect("ok");
assert_eq!(reply, "flushed web");
}
#[tokio::test]
async fn reopen_names_the_sheep_it_reopened() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Reopen {
selector: SelectorSpec::Name("web".to_owned()),
})
.answer(Response::Reopened(vec![sample("web")]));
let reply = live
.act(Verb::Reopen, SelectorSpec::Name("web".to_owned()))
.await
.expect("ok");
assert_eq!(reply, "reopened web");
}
#[tokio::test]
async fn save_names_the_roll_it_wrote_not_a_sheep() {
let (live, mut fake) = test_live().await;
fake.expect(Request::SaveRoll).answer(Response::RollSaved {
path: "roll.toml".to_owned(),
apps: 3,
});
let reply = live.act(Verb::Save, SelectorSpec::All).await.expect("ok");
assert_eq!(reply, "saved 3 apps to roll.toml");
}
#[tokio::test]
async fn host_usage_hands_back_what_the_shepherd_sampled() {
let (live, mut fake) = test_live().await;
fake.expect(Request::HostUsage)
.answer(Response::HostUsage(Some(HostUsage {
cpu_percent: Some(12.5),
memory_used_bytes: 1024,
memory_total_bytes: 4096,
disk_bytes_per_second: Some((1, 2)),
network_bytes_per_second: None,
})));
let usage = live.host_usage().await.expect("ok").expect("sampled");
assert_eq!(usage.cpu_percent, Some(12.5));
assert_eq!(usage.memory_used_bytes, 1024);
assert_eq!(usage.memory_total_bytes, 4096);
assert_eq!(usage.disk_bytes_per_second, Some((1, 2)));
assert_eq!(usage.network_bytes_per_second, None);
}
#[tokio::test]
async fn an_unsampled_host_is_none_rather_than_an_error() {
let (live, mut fake) = test_live().await;
fake.expect(Request::HostUsage)
.answer(Response::HostUsage(None));
assert_eq!(live.host_usage().await.expect("ok"), None);
}
#[tokio::test]
async fn a_wrong_shaped_answer_names_what_was_asked_and_what_came_back() {
let (live, mut fake) = test_live().await;
fake.expect(Request::ListFlock).answer(Response::Pong);
let err = live.flock().await.expect_err("refused").to_string();
assert!(err.contains("a Flock"), "{err}");
assert!(err.contains("Pong"), "{err}");
}
#[tokio::test]
async fn section_reads_the_dog_config_toml() {
let (live, mut fake) = test_live().await;
fake.expect(Request::DogConfig {
name: "discord".to_owned(),
})
.answer(Response::DogSection {
toml: "flush = \"1s\"\n".to_owned().into(),
});
let toml = live.section("discord").await.expect("ok");
assert_eq!(toml, "flush = \"1s\"\n");
}
#[tokio::test]
async fn subscribe_hands_back_an_event_stream() {
let (live, mut fake) = test_live().await;
fake.expect(Request::Subscribe {
topics: vec!["process.*".to_owned()],
})
.answer(Response::Subscribed);
live.subscribe(vec!["process.*".to_owned()])
.await
.expect("ok");
}
#[tokio::test]
async fn a_live_session_never_prints_its_socket_path() {
let dir = tempfile::tempdir().expect("tempdir");
let socket = testing::control_address(dir.path());
let _envelopes =
testing::fake_daemon_answering_with_ack(&socket, testing::sample_ack(), |_request| {
Response::Pong
})
.await;
let client = ReconnectingClient::connect(&socket)
.await
.expect("test dial");
let shown = format!("{:?}", Live::new(client));
assert_eq!(shown, "Live { socket: <redacted> }");
assert!(
!shown.contains(&socket.display().to_string()),
"the socket path leaked: {shown}"
);
}
#[tokio::test]
async fn no_sentence_act_hands_back_carries_a_dash() {
let (live, mut fake) = test_live().await;
let web = SelectorSpec::Name("web".to_owned());
let answers = [
(
Verb::Start,
Request::Restart {
selector: web.clone(),
},
Response::Restarted {
accepted: vec![sample("web")],
refused: Vec::new(),
},
),
(
Verb::Restart,
Request::Restart {
selector: web.clone(),
},
Response::Restarted {
accepted: vec![sample("web")],
refused: vec![SheepRefusal::new("worker", "exceeded restart budget")],
},
),
(
Verb::Reload,
Request::Reload {
selector: web.clone(),
},
Response::Reloading {
accepted: Vec::new(),
refused: vec![SheepRefusal::new("worker", "no reload signal configured")],
},
),
(
Verb::Stop,
Request::Stop {
selector: web.clone(),
},
Response::Stopped(vec![sample("web")]),
),
(
Verb::Delete,
Request::Delete {
selector: web.clone(),
},
Response::Deleted(vec![7]),
),
(
Verb::Flush,
Request::Flush {
selector: web.clone(),
},
Response::Flushed(Vec::new()),
),
(
Verb::Reopen,
Request::Reopen {
selector: web.clone(),
},
Response::Reopened(vec![sample("web")]),
),
(
Verb::Save,
Request::SaveRoll,
Response::RollSaved {
path: "/var/lib/shep/roll.json".to_owned(),
apps: 3,
},
),
];
for (verb, asked, answer) in answers {
fake.expect(asked).answer(answer);
let reply = live.act(verb, web.clone()).await.expect("ok");
assert_no_dashes(&reply);
}
}
#[test]
fn nothing_this_module_names_for_a_person_carries_a_dash() {
for selector in [
SelectorSpec::All,
SelectorSpec::Id(7),
SelectorSpec::Name("web".to_owned()),
SelectorSpec::Fold("prod".to_owned()),
SelectorSpec::Regex("^web".to_owned()),
SelectorSpec::Instance {
name: "web".to_owned(),
slot: 2,
},
] {
assert_no_dashes(&describe_selector(&selector));
}
for response in [
Response::Pong,
Response::Flock(vec![sample("web")]),
Response::HostUsage(None),
Response::Described(vec![sample("web")]),
Response::Stopped(vec![sample("web")]),
Response::Restarted {
accepted: vec![sample("web")],
refused: Vec::new(),
},
Response::Reloading {
accepted: vec![sample("web")],
refused: Vec::new(),
},
Response::Deleted(vec![7]),
Response::Flushed(vec![sample("web")]),
Response::Reopened(vec![sample("web")]),
Response::RollSaved {
path: "/var/lib/shep/roll.json".to_owned(),
apps: 3,
},
] {
assert_no_dashes(&named(&response));
}
assert_no_dashes(&name_list(&[]));
}
}