use std::cell::Cell;
use std::fs;
use std::path::PathBuf;
use std::time::Duration;
use shep_client::RequestError;
use shep_client::shep_core::protocol::{ProcessInfo, RpcError, RpcErrorCode};
use shep_client::shep_core::status::ProcStatus;
use tempfile::TempDir;
use crate::daemon::Daemon;
use crate::error::Error;
use crate::fixtures;
use crate::paths::Tree;
use crate::state::{State, Watch};
use crate::swap;
pub(super) fn test_config_keeping(retention: usize) -> crate::config::DogConfig {
crate::config::DogConfig {
retention,
..fixtures::dog_config()
}
}
pub(super) const FLOCKFILE: &str = "[[app]]\nname = 'web'\nscript = './run.sh'\n\n\
[app.readiness_probe]\nkind = 'http'\ntarget = 'http://127.0.0.1:1/health'\n";
pub(super) struct Fixture {
pub(super) home: TempDir,
pub(super) origin: TempDir,
pub(super) tree: Tree,
pub(super) state: State,
}
pub(super) const RECLAIMED: &str = "0000000000000000000000000000000000000000";
pub(super) fn fixture_before_any_release() -> Fixture {
let home = tempfile::tempdir().expect("tempdir");
let origin = fixtures::checkout(&[("Flockfile.toml", FLOCKFILE)]);
let tree = Tree::for_sheep(home.path(), "web");
fs::create_dir_all(tree.git()).expect("create git dir");
fixtures::run_git(&tree.git(), &["init", "-q", "--bare"]);
let remote = origin.path().to_str().expect("utf-8 path").to_owned();
crate::git::fetch(&tree.git(), &remote, fixtures::TEST_BUDGET).expect("fetch");
let state = State {
remote,
verify: crate::state::Verify::Probed,
watch: crate::state::Watch::Manual,
checkout: origin.path().to_owned(),
..fixtures::state()
};
state.write(&tree.state_file()).expect("write deploy.toml");
Fixture {
home,
origin,
tree,
state,
}
}
pub(super) fn fixture_with_previous_release() -> Fixture {
let mut fixture = fixture_before_any_release();
let first = crate::git::remote_head(&fixture.tree.git(), "main").expect("head");
install_release(&fixture, &first);
fixture.state.deployed = Some(first);
fixture
.state
.write(&fixture.tree.state_file())
.expect("write deploy.toml");
fixture
}
pub(super) fn install_release(fixture: &Fixture, sha: &str) {
let tree = &fixture.tree;
if !tree.release(sha).exists() {
crate::git::worktree_add(&tree.git(), &tree.release(sha), sha).expect("worktree");
}
swap::point_at(&tree.current(), &tree.release(sha)).expect("swap");
}
pub(super) fn commit_on_origin(fixture: &Fixture, name: &str) -> String {
fs::write(fixture.origin.path().join(name), "x").expect("write");
fixtures::run_git(fixture.origin.path(), &["add", "."]);
fixtures::run_git(fixture.origin.path(), &["commit", "-q", "-m", name]);
fixtures::head_of(fixture.origin.path())
}
pub(super) struct Shepherd {
settles_to: ProcStatus,
replaces: bool,
describe_fails_from: Option<u32>,
plant: Option<PathBuf>,
flip_watch: Option<PathBuf>,
refusals: Cell<u32>,
refusal_code: RpcErrorCode,
refuse_from_the_first: Cell<bool>,
replies_lost: Cell<u32>,
attempts: Cell<u32>,
flapping: bool,
instances: u32,
turnover_after: u32,
describes: Cell<u32>,
reloads: Cell<u32>,
}
const FIRST_PID: u32 = 12835;
impl Shepherd {
pub(super) fn ready() -> Self {
Self {
settles_to: ProcStatus::Online,
replaces: true,
describe_fails_from: None,
plant: None,
flip_watch: None,
refusals: Cell::new(0),
refusal_code: RpcErrorCode::Internal,
refuse_from_the_first: Cell::new(false),
replies_lost: Cell::new(0),
attempts: Cell::new(0),
flapping: false,
instances: 1,
turnover_after: 0,
describes: Cell::new(0),
reloads: Cell::new(0),
}
}
pub(super) fn never_ready() -> Self {
Self {
settles_to: ProcStatus::Starting,
..Self::ready()
}
}
pub(super) fn keeps_the_old_instance() -> Self {
Self {
replaces: false,
..Self::ready()
}
}
pub(super) fn describe_fails() -> Self {
Self {
describe_fails_from: Some(1),
..Self::ready()
}
}
pub(super) fn unreachable() -> Self {
Self {
describe_fails_from: Some(0),
..Self::ready()
}
}
pub(super) fn too_busy_to_start() -> Self {
let shepherd = Self::ready();
shepherd.refusals.set(u32::MAX);
shepherd.refuse_from_the_first.set(true);
shepherd
}
pub(super) fn flipping_watch(record: PathBuf) -> Self {
Self {
flip_watch: Some(record),
..Self::ready()
}
}
pub(super) fn planting(plant: PathBuf) -> Self {
Self {
plant: Some(plant),
..Self::never_ready()
}
}
pub(super) fn busy_for(times: u32) -> Self {
let shepherd = Self::never_ready();
shepherd.refusals.set(times);
shepherd
}
pub(super) fn flapping() -> Self {
Self {
flapping: true,
..Self::ready()
}
}
pub(super) fn ready_after(describes: u32) -> Self {
Self {
turnover_after: describes,
..Self::ready()
}
}
pub(super) fn running(self, instances: u32) -> Self {
Self { instances, ..self }
}
pub(super) fn losing_replies(count: u32) -> Self {
let shepherd = Self::ready();
shepherd.replies_lost.set(count);
shepherd
}
pub(super) fn unregistered() -> Self {
Self {
refusal_code: RpcErrorCode::NotFound,
..Self::busy_for(u32::MAX)
}
}
pub(super) fn attempt_count(&self) -> u32 {
self.attempts.get()
}
pub(super) fn reload_count(&self) -> u32 {
self.reloads.get()
}
fn landed(&self) -> u32 {
if self.describes.get() > self.turnover_after {
self.reloads.get()
} else {
0
}
}
fn pid(&self, index: u32) -> u32 {
if !self.replaces {
return FIRST_PID + index;
}
let churn = if self.flapping {
self.describes.get()
} else {
0
};
FIRST_PID + (self.landed() + churn) * 100 + index
}
fn status(&self) -> ProcStatus {
if self.landed() == 0 {
ProcStatus::Online
} else {
self.settles_to
}
}
}
impl Daemon for Shepherd {
async fn describe(&self, sheep: &str) -> Result<Vec<ProcessInfo>, Error> {
if let Some(plant) = &self.plant
&& plant.symlink_metadata().is_err()
{
std::fs::write(plant, b"in the way").expect("plant something at the tmp path");
}
self.describes.set(self.describes.get() + 1);
if self
.describe_fails_from
.is_some_and(|from| self.reloads.get() >= from)
{
return Err(Error::Protocol("the shepherd stopped answering".to_owned()));
}
Ok((0..self.instances)
.map(|index| {
ProcessInfo::builder(index, sheep, self.status())
.pid(Some(self.pid(index)))
.build()
})
.collect())
}
async fn reload(&self, sheep: &str) -> Result<(), Error> {
if let Some(record) = &self.flip_watch
&& self.attempts.get() == 0
{
let mut state = State::read(record).expect("the record to flip");
state.watch = Watch::Manual;
state
.write(record)
.expect("flip watch to manual mid-deploy");
}
self.attempts.set(self.attempts.get() + 1);
let refusals = self.refusals.get();
if (self.reloads.get() > 0 || self.refuse_from_the_first.get()) && refusals > 0 {
if refusals != u32::MAX {
self.refusals.set(refusals - 1);
}
return Err(Error::Request(RequestError::Rpc(RpcError {
code: self.refusal_code,
message: format!("{sheep} is already being reloaded"),
daemon_version: None,
})));
}
self.reloads.set(self.reloads.get() + 1);
let lost = self.replies_lost.get();
if lost > 0 {
if lost != u32::MAX {
self.replies_lost.set(lost - 1);
}
return Err(Error::Request(RequestError::Timeout {
after: Duration::from_secs(7),
}));
}
Ok(())
}
crate::fixtures::daemon_methods!(unimplemented;
dog_config, list_flock, start, delete, restart, save_roll, set_smit,
);
}