pub mod bark;
pub mod metrics;
use core::fmt;
use shep_client::{ConnectError, EventStream, ReconnectingClient, RequestError};
use shep_core::paths::ShepPaths;
use shep_core::protocol::{BusEvent, ProcessInfo, Request, Response, RpcError, RpcErrorCode};
use crate::exit::ExitCode;
const BUILT_IN_DOGS: [&str; 2] = ["metrics", "bark"];
pub struct DogRuntime {
pub client: ReconnectingClient,
pub section: String,
pub paths: ShepPaths,
name: String,
}
impl fmt::Debug for DogRuntime {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("DogRuntime")
.field("client", &self.client)
.field("section", &format!("<{} bytes>", self.section.len()))
.field("paths", &self.paths)
.field("name", &self.name)
.finish()
}
}
pub enum DogRunError {
Connect(ConnectError),
Request(RequestError),
UnexpectedReply,
Section {
name: String,
message: String,
},
}
impl fmt::Debug for DogRunError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Connect(err) => f.debug_tuple("Connect").field(err).finish(),
Self::Request(err) => f.debug_tuple("Request").field(err).finish(),
Self::UnexpectedReply => f.write_str("UnexpectedReply"),
Self::Section { name, .. } => f
.debug_struct("Section")
.field("name", name)
.field("message", &"<redacted: may quote the section>")
.finish(),
}
}
}
impl fmt::Display for DogRunError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Connect(err) => write!(f, "no shepherd answered at the socket: {err}"),
Self::Request(err) => write!(f, "the shepherd refused the config request: {err}"),
Self::UnexpectedReply => {
f.write_str("the shepherd answered with a response this client does not understand")
}
Self::Section { name, message } => {
write!(f, "dog {name}'s own configuration does not fit: {message}")
}
}
}
}
impl core::error::Error for DogRunError {
fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
match self {
Self::Connect(err) => Some(err),
Self::Request(err) => Some(err),
Self::UnexpectedReply | Self::Section { .. } => None,
}
}
}
impl From<ConnectError> for DogRunError {
fn from(source: ConnectError) -> Self {
Self::Connect(source)
}
}
impl From<RequestError> for DogRunError {
fn from(source: RequestError) -> Self {
Self::Request(source)
}
}
impl DogRuntime {
pub async fn start(name: &str, paths: ShepPaths) -> Result<Self, DogRunError> {
let client = ReconnectingClient::connect_as_dog(&paths.socket, name).await?;
let response = client
.request(Request::DogConfig {
name: name.to_string(),
})
.await?;
let Response::DogSection { toml } = response else {
return Err(DogRunError::UnexpectedReply);
};
Ok(Self {
section: toml.as_str().to_string(),
client,
paths,
name: name.to_string(),
})
}
pub fn config<T>(&self) -> Result<T, DogRunError>
where
T: serde::de::DeserializeOwned + Default,
{
if self.section.is_empty() {
return Ok(T::default());
}
toml::from_str(&self.section).map_err(|err| DogRunError::Section {
name: self.name.clone(),
message: err.to_string(),
})
}
}
fn exit_code_for(err: &DogRunError) -> ExitCode {
match err {
DogRunError::Connect(inner) => ExitCode::from(inner),
DogRunError::Request(inner) => ExitCode::from(inner),
DogRunError::Section { .. } => ExitCode::InvalidConfig,
DogRunError::UnexpectedReply => ExitCode::Internal,
}
}
pub async fn run_dog(name: &str, paths: ShepPaths) -> ExitCode {
if !BUILT_IN_DOGS.contains(&name) {
eprintln!("shep dog: unknown dog {name:?}; the built-in dogs are \"metrics\" and \"bark\"");
return ExitCode::Usage;
}
let runtime = match DogRuntime::start(name, paths).await {
Ok(runtime) => runtime,
Err(err) => {
eprintln!("shep dog {name}: {err}");
return exit_code_for(&err);
}
};
match name {
"metrics" => metrics::run(runtime).await,
"bark" => run_bark(runtime).await,
_ => unreachable!("checked against BUILT_IN_DOGS above"),
}
}
async fn run_bark(runtime: DogRuntime) -> ExitCode {
let config = match runtime.config::<bark::BarkConfig>() {
Ok(config) => config,
Err(_err) => {
eprintln!("shep dog bark: [dog.bark] does not parse; see `shep dogs`");
return ExitCode::InvalidConfig;
}
};
let rule_list = if config.rules.is_empty() {
bark::rules::Rules::default_rules(&config.sinks)
} else {
config.rules.clone()
};
let rules = match bark::rules::Rules::new(rule_list, &config.sinks) {
Ok(rules) => rules,
Err(err) => {
eprintln!("shep dog bark: {err}");
return ExitCode::InvalidConfig;
}
};
let events = match runtime.client.subscribe(vec!["process.*".to_owned()]).await {
Ok(events) => events,
Err(err) => {
eprintln!("shep dog bark: could not subscribe to the shepherd's bus: {err}");
return ExitCode::from(&err);
}
};
let barks_path = runtime.paths.barks.clone();
let flock = ClientFlockSource {
client: runtime.client,
};
bark::run_loop(events, flock, rules, &config, &barks_path).await
}
impl bark::EventSource for EventStream {
async fn next(&mut self) -> Option<Result<BusEvent, u64>> {
self.next()
.await
.map(|item| item.map_err(|lagged| lagged.count))
}
}
struct ClientFlockSource {
client: ReconnectingClient,
}
impl bark::FlockSource for ClientFlockSource {
async fn flock(&self) -> Result<Vec<ProcessInfo>, RequestError> {
match self.client.request(Request::ListFlock).await? {
Response::Flock(flock) => Ok(flock),
_ => Err(RequestError::Rpc(RpcError {
code: RpcErrorCode::Internal,
message: "the shepherd answered ListFlock with something other than \
Response::Flock"
.to_owned(),
daemon_version: None,
})),
}
}
}
#[cfg(test)]
mod tests {
use std::path::{Path, PathBuf};
use std::time::Duration;
use shep_client::testing::{fake_reconnecting_client_on, sample_ack, serve_one_request};
use super::*;
fn test_paths(dir: &Path, socket: PathBuf) -> ShepPaths {
let home = dir.to_path_buf();
ShepPaths {
daemon_config: home.join("shep.toml"),
snapshot: home.join("flock.json"),
logs: home.join("logs"),
pids: home.join("pids"),
run: home.join("run"),
socket,
barks: home.join("barks.jsonl"),
kv: home.join("kv.json"),
home,
}
}
fn runtime_with_section(section: &str) -> DogRuntime {
let dir = tempfile::tempdir().unwrap();
let socket = shep_client::testing::control_address(dir.path());
tokio::runtime::Runtime::new().unwrap().block_on(async {
let (client, _daemon) = fake_reconnecting_client_on(&socket).await;
DogRuntime {
client,
section: section.to_string(),
paths: test_paths(dir.path(), socket),
name: "testdog".to_string(),
}
})
}
#[test]
fn a_section_that_does_not_fit_is_refused_rather_than_defaulted() {
#[derive(Debug, Default, serde::Deserialize, PartialEq)]
#[serde(deny_unknown_fields, default)]
struct Cfg {
port: u16,
}
let runtime = runtime_with_section("port = \"nine thousand\"\n");
let err = runtime.config::<Cfg>().unwrap_err();
assert!(matches!(err, DogRunError::Section { .. }));
assert!(err.to_string().contains("port"));
let empty = runtime_with_section("");
assert_eq!(empty.config::<Cfg>().unwrap(), Cfg::default());
}
#[tokio::test]
async fn a_dog_asks_for_its_own_section_by_name() {
let dir = tempfile::tempdir().unwrap();
let socket = shep_client::testing::control_address(dir.path());
let response = Response::DogSection {
toml: "webhook = \"https://example.invalid/hook\"\n"
.to_string()
.into(),
};
let handle = serve_one_request(&socket, sample_ack(), response).await;
let paths = test_paths(dir.path(), socket);
let runtime = DogRuntime::start("bark", paths).await.unwrap();
let envelope = tokio::time::timeout(Duration::from_secs(5), handle)
.await
.expect("DogRuntime::start must reach the wire; it hung instead of connecting")
.unwrap();
assert_eq!(
envelope.body,
Request::DogConfig {
name: "bark".to_string()
}
);
assert_eq!(
runtime.section,
"webhook = \"https://example.invalid/hook\"\n"
);
}
#[tokio::test]
async fn a_dog_announces_its_own_name_at_the_handshake() {
let dir = tempfile::tempdir().unwrap();
let socket = shep_client::testing::control_address(dir.path());
let served = shep_client::testing::fake_daemon(&socket, Ok(sample_ack())).await;
let paths = test_paths(dir.path(), socket);
let _started = DogRuntime::start("bark", paths).await;
let hello = tokio::time::timeout(Duration::from_secs(5), served)
.await
.expect("DogRuntime::start must reach the wire; it hung instead of connecting")
.unwrap();
assert_eq!(
hello.dog_name.as_deref(),
Some("bark"),
"a dog must announce the name it was registered under"
);
}
#[tokio::test]
async fn an_unknown_dog_name_is_usage_without_touching_the_socket() {
let dir = tempfile::tempdir().unwrap();
let paths = test_paths(
dir.path(),
shep_client::testing::control_address(dir.path()),
);
let code = run_dog("otel", paths).await;
assert_eq!(code, ExitCode::Usage);
}
#[tokio::test]
async fn run_dog_reaches_bark() {
let dir = tempfile::tempdir().unwrap();
let socket = shep_client::testing::control_address(dir.path());
let response = Response::DogSection {
toml: String::new().into(),
};
let handle = serve_one_request(&socket, sample_ack(), response).await;
let paths = test_paths(dir.path(), socket);
let task = tokio::spawn(run_dog("bark", paths));
let envelope = tokio::time::timeout(Duration::from_secs(5), handle)
.await
.expect("run_dog must reach the wire")
.unwrap();
assert_eq!(
envelope.body,
Request::DogConfig {
name: "bark".to_string()
}
);
task.abort();
}
#[tokio::test]
async fn run_dog_reaches_metrics() {
let dir = tempfile::tempdir().unwrap();
let socket = shep_client::testing::control_address(dir.path());
let response = Response::DogSection {
toml: "bind = \"127.0.0.1:0\"\n".to_string().into(),
};
let handle = serve_one_request(&socket, sample_ack(), response).await;
let paths = test_paths(dir.path(), socket);
let task = tokio::spawn(run_dog("metrics", paths));
let envelope = tokio::time::timeout(Duration::from_secs(5), handle)
.await
.expect("run_dog must reach the wire")
.unwrap();
assert_eq!(
envelope.body,
Request::DogConfig {
name: "metrics".to_string()
}
);
task.abort();
}
#[tokio::test]
async fn run_dog_reports_daemon_unreachable_with_no_shepherd_running() {
let dir = tempfile::tempdir().unwrap();
let paths = test_paths(
dir.path(),
shep_client::testing::control_address(dir.path()),
);
let code = run_dog("metrics", paths).await;
assert_eq!(code, ExitCode::DaemonUnreachable);
}
#[test]
fn dog_run_error_section_debug_never_prints_the_message() {
let secret = "https://hooks.example.com/services/T00/B00/super-secret-token";
let err = DogRunError::Section {
name: "bark".to_string(),
message: format!("invalid type: string \"{secret}\", expected u16\nin `webhook`"),
};
let debug = format!("{err:?}");
assert!(!debug.contains(secret), "{debug}");
assert!(!debug.contains("webhook"), "{debug}");
assert_eq!(
debug,
"Section { name: \"bark\", message: \"<redacted: may quote the section>\" }"
);
}
#[test]
fn dog_runtime_debug_never_prints_the_section() {
let secret = "https://hooks.example.com/services/T00/B00/super-secret-token";
let section = format!("webhook = \"{secret}\"\n");
let byte_len = section.len();
let runtime = runtime_with_section(§ion);
let debug = format!("{runtime:?}");
assert!(!debug.contains(secret), "{debug}");
assert!(!debug.contains("webhook"), "{debug}");
assert!(
debug.contains(&format!("section: \"<{byte_len} bytes>\"")),
"{debug}"
);
}
}