ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Compose cold-start resources under Runtime Supervisor ownership

Browse source
This commit is contained in:
ms2k 2026-07-16 17:29:23 +09:00
commit 33ac0ba4ee
1 changed files with 1827 additions and 139 deletions
Download patch file Download diff file Expand all files Collapse all files

454
crates/lazaret-runtime-supervisor/src/supervision.rs
View file

@ -21,6 +21,9 @@ use lazaret_linux::process::{
ChildStandardIo, ParentEdge, ProcessOwnershipContext, RetainedProcess, TargetCommand,
child_broker, current_process_identity,
};
use lazaret_state_anchor::{
RuntimeStateAnchorAuthority, RuntimeStateAnchorMount, StateAnchorLaunchAuthority,
};
use rustix::event::{PollFd, PollFlags, Timespec, poll};
use rustix::fs::{MemfdFlags, memfd_create};
use rustix::io::{FdFlags, fcntl_setfd, read, write};
@ -58,6 +61,12 @@ impl AttachedRunscSignal {
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum SupervisorEvent {
StateAnchorStaged(StagedProcessOwnership),
StateAnchorReady {
staged: StagedProcessOwnership,
mount: lazaret_core::identity::MountIdentity,
},
SidecarStarted,
HelperStaged(StagedProcessOwnership),
HelperReady(StagedProcessOwnership),
NetworkAttestation(NetworkReadinessAttestation),
@ -69,6 +78,10 @@ pub enum SupervisorEvent {
sentry_pid_inventory: u32,
},
SignalApplied(AttachedRunscSignal),
SandboxStopped,
SidecarStopped,
StateAnchorUnmounted,
Exited,
AttachedRunscExited,
Stopped,
Refused(String),
@ -76,12 +89,19 @@ pub enum SupervisorEvent {
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
enum CoordinatorMessage {
MountStateAnchor,
StateAnchorStagedAck(ProcessOwnershipCommitAck),
StartSidecar,
HelperStagedAck(ProcessOwnershipCommitAck),
NetworkAttestationAccepted(NetworkProbeRequest),
NetworkReadinessCommitted,
StagedAck(ProcessOwnershipCommitAck),
SandboxCommitted,
SignalAttachedRunsc(AttachedRunscSignal),
StopSandbox,
StopSidecar,
UnmountStateAnchor,
Exit,
Stop,
}
@ -101,6 +121,30 @@ impl CoordinatorChannel {
receive_message(&self.descriptor)
}
/// # Errors
/// Returns an error when exact State Anchor acquisition cannot be requested.
pub fn mount_state_anchor(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::MountStateAnchor)
}
/// # Errors
/// Returns an error when the exact daemon acknowledgment cannot be delivered.
pub fn acknowledge_state_anchor(
&self,
acknowledgment: ProcessOwnershipCommitAck,
) -> io::Result<()> {
send_message(
&self.descriptor,
&CoordinatorMessage::StateAnchorStagedAck(acknowledgment),
)
}
/// # Errors
/// Returns an error when the RS-owned sidecar install path cannot be released.
pub fn start_sidecar(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::StartSidecar)
}
/// # Errors
/// Returns an error when the exact typed acknowledgment cannot be delivered.
pub fn acknowledge_staged(&self, acknowledgment: ProcessOwnershipCommitAck) -> io::Result<()> {
@ -157,14 +201,41 @@ impl CoordinatorChannel {
pub fn stop(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::Stop)
}
/// # Errors
/// Returns an error when ordered sandbox stop cannot be requested.
pub fn stop_sandbox(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::StopSandbox)
}
/// # Errors
/// Returns an error when ordered sidecar stop cannot be requested.
pub fn stop_sidecar(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::StopSidecar)
}
/// # Errors
/// Returns an error when exact State Anchor unmount cannot be requested.
pub fn unmount_state_anchor(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::UnmountStateAnchor)
}
/// # Errors
/// Returns an error when graceful Runtime Supervisor exit cannot be requested.
pub fn exit(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::Exit)
}
}
pub struct RuntimeSupervisorLaunch {
target: TargetCommand,
ownership: ProcessOwnershipContext,
policy_file: Option<File>,
policy_digest: Option<DerivedPolicyDigest>,
configuration_fingerprint: Option<ConfigurationFingerprint>,
network_sidecar_id: Option<NetworkSidecarId>,
runsc_command: RunscCommand,
state_anchor: Option<StateAnchorLaunchAuthority>,
}
#[derive(Clone, Copy)]
@ -213,8 +284,11 @@ impl RuntimeSupervisorLaunch {
target: target.with_lifetime_pipe_parent_edge(),
ownership,
policy_file: None,
policy_digest: None,
configuration_fingerprint: None,
network_sidecar_id: None,
runsc_command: command.clone(),
state_anchor: None,
})
}
@ -260,10 +334,68 @@ impl RuntimeSupervisorLaunch {
launch.target = launch.target.with_lifetime_pipe_parent_edge();
drop(target);
launch.policy_file = Some(policy_file);
launch.policy_digest = Some(network.policy.digest());
launch.configuration_fingerprint = Some(network.configuration_fingerprint);
launch.network_sidecar_id = Some(network.network_sidecar_id);
Ok(launch)
}
/// Adds exact inherited State Anchor authorities without beginning physical acquisition.
#[must_use]
pub fn with_state_anchor(mut self, authority: StateAnchorLaunchAuthority) -> Self {
self.state_anchor = Some(authority);
self.rebuild_target();
self
}
fn rebuild_target(&mut self) {
let mut target = TargetCommand::new("/proc/self/exe")
.argument(PRIVATE_ROLE_ARGUMENT)
.argument("--domain")
.argument(hex(self.ownership.domain_id.as_bytes()))
.argument("--generation")
.argument(self.ownership.generation_id.get().to_string())
.argument("--coordinator")
.argument(hex(self.ownership.coordinator_id.as_bytes()));
if let Some(authority) = &self.state_anchor {
let (device, inode) = authority.expected_lower();
target = target
.argument("--state-lower-fd")
.argument(authority.lower_fd().to_string())
.argument("--state-persistent-fd")
.argument(authority.persistent_anchor_fd().to_string())
.argument("--state-runtime-fd")
.argument(authority.runtime_anchor_fd().to_string())
.argument("--state-lower-device")
.argument(device.to_string())
.argument("--state-lower-inode")
.argument(inode.to_string());
}
if let (Some(policy_file), Some(digest), Some(fingerprint), Some(sidecar)) = (
&self.policy_file,
self.policy_digest,
self.configuration_fingerprint,
self.network_sidecar_id,
) {
target = target
.argument("--network-policy-fd")
.argument(policy_file.as_raw_fd().to_string())
.argument("--network-policy-digest")
.argument(hex(digest.as_bytes()))
.argument("--configuration-fingerprint")
.argument(hex(fingerprint.as_bytes()))
.argument("--network-sidecar")
.argument(hex(sidecar.as_bytes()));
}
target = target
.argument("--runsc")
.argument(self.runsc_command.program());
for argument in self.runsc_command.arguments() {
target = target.argument(argument);
}
self.target = target.with_lifetime_pipe_parent_edge();
}
#[must_use]
pub const fn target(&self) -> &TargetCommand {
&self.target
@ -328,11 +460,19 @@ pub fn dispatch_private_runtime_role(arguments: &[OsString]) -> Option<io::Resul
struct RuntimeArguments {
ownership: ProcessOwnershipContext,
state_anchor: Option<RuntimeStateAnchorArguments>,
network: Option<RuntimeNetworkArguments>,
runsc_program: OsString,
runsc_arguments: Vec<OsString>,
}
struct RuntimeStateAnchorArguments {
lower_fd: i32,
persistent_anchor_fd: i32,
runtime_anchor_fd: i32,
expected_lower: (u64, u64),
}
struct RuntimeNetworkArguments {
policy_fd: i32,
policy_digest: DerivedPolicyDigest,
@ -348,6 +488,22 @@ fn parse_runtime_arguments(arguments: &[OsString]) -> io::Result<RuntimeArgument
.and_then(|value| value.parse().ok())
.ok_or_else(|| io::Error::other("Runtime Supervisor generation is invalid"))?;
let coordinator = parse_fixed_hex(named(&mut arguments, "--coordinator")?)?;
let state_anchor = if arguments
.peek()
.is_some_and(|value| *value == "--state-lower-fd")
{
Some(RuntimeStateAnchorArguments {
lower_fd: parse_number(named(&mut arguments, "--state-lower-fd")?)?,
persistent_anchor_fd: parse_number(named(&mut arguments, "--state-persistent-fd")?)?,
runtime_anchor_fd: parse_number(named(&mut arguments, "--state-runtime-fd")?)?,
expected_lower: (
parse_number(named(&mut arguments, "--state-lower-device")?)?,
parse_number(named(&mut arguments, "--state-lower-inode")?)?,
),
})
} else {
None
};
let network = if arguments
.peek()
.is_some_and(|value| *value == "--network-policy-fd")
@ -374,12 +530,23 @@ fn parse_runtime_arguments(arguments: &[OsString]) -> io::Result<RuntimeArgument
coordinator_id: CoordinatorId::new(coordinator),
slot: HostProcessOwnershipSlot::SandboxAttachedRunsc,
},
state_anchor,
network,
runsc_program: (*program).clone(),
runsc_arguments: arguments.cloned().collect(),
})
}
fn parse_number<T>(value: &OsStr) -> io::Result<T>
where
T: std::str::FromStr,
{
value
.to_str()
.and_then(|value| value.parse().ok())
.ok_or_else(|| io::Error::other("Runtime Supervisor numeric argument is invalid"))
}
fn parse_network_arguments<'a>(
arguments: &mut impl Iterator<Item = &'a OsString>,
) -> io::Result<RuntimeNetworkArguments> {
@ -439,9 +606,92 @@ fn run_supervisor(arguments: &RuntimeArguments) -> io::Result<()> {
require_lifetime_held()?;
let control = std::io::stdout();
let parent = current_process_identity()?;
let mut state_anchor = prepare_state_anchor(arguments, parent, &control)?;
let result = run_supervisor_with_state(arguments, parent, &control, &mut state_anchor);
let cleanup = state_anchor.take().map(RuntimeStateAnchorMount::unmount);
match (result, cleanup) {
(Err(error), Some(Err(cleanup))) => Err(io::Error::new(
error.kind(),
format!("{error}; exact State Anchor cleanup failed: {cleanup}"),
)),
(Err(error), _) => Err(error),
(Ok(()), Some(Err(error))) => Err(io::Error::other(error)),
(Ok(()), _) => Ok(()),
}
}
fn prepare_state_anchor(
arguments: &RuntimeArguments,
parent: lazaret_core::identity::PidfsProcessIdentityV1,
control: &std::io::Stdout,
) -> io::Result<Option<RuntimeStateAnchorMount>> {
let Some(state) = &arguments.state_anchor else {
return Ok(None);
};
let command: CoordinatorMessage = receive_message(control)?;
if command != CoordinatorMessage::MountStateAnchor {
return Err(io::Error::other(
"Runtime Supervisor received no State Anchor mount command",
));
}
let authority = RuntimeStateAnchorAuthority::from_inherited(
state.lower_fd,
state.persistent_anchor_fd,
state.runtime_anchor_fd,
state.expected_lower,
)
.map_err(io::Error::other)?;
let ownership = ProcessOwnershipContext {
slot: HostProcessOwnershipSlot::StateAnchorDaemon,
..arguments.ownership
};
authority
.acquire(
parent,
ownership,
|staged| send_message(control, &SupervisorEvent::StateAnchorStaged(staged)),
|staged| match receive_message(control)? {
CoordinatorMessage::StateAnchorStagedAck(acknowledgment)
if acknowledgment.acknowledges(&staged) =>
{
Ok(acknowledgment)
}
_ => Err(io::Error::other(
"Runtime Supervisor received a mismatched State Anchor acknowledgment",
)),
},
|readiness| {
send_message(
control,
&SupervisorEvent::StateAnchorReady {
staged: readiness.staged(),
mount: readiness.mount(),
},
)
},
)
.map(Some)
.map_err(io::Error::other)
}
fn run_supervisor_with_state(
arguments: &RuntimeArguments,
parent: lazaret_core::identity::PidfsProcessIdentityV1,
control: &std::io::Stdout,
state_anchor: &mut Option<RuntimeStateAnchorMount>,
) -> io::Result<()> {
if arguments.state_anchor.is_some() {
let message: CoordinatorMessage = receive_message(control)?;
if message != CoordinatorMessage::StartSidecar {
return Err(io::Error::other(
"Runtime Supervisor received no sidecar start command",
));
}
send_message(control, &SupervisorEvent::SidecarStarted)?;
}
let network = match &arguments.network {
Some(network) => Some(prepare_runtime_network(
arguments, network, parent, &control,
arguments, network, parent, control,
)?),
None => None,
};
@ -456,9 +706,9 @@ fn run_supervisor(arguments: &RuntimeArguments) -> io::Result<()> {
attached_runsc_standard_io(&runsc)?,
|ownership| {
staged.set(Some(ownership));
send_message(&control, &SupervisorEvent::Staged(ownership))
send_message(control, &SupervisorEvent::Staged(ownership))
},
|ownership| match receive_message(&control)? {
|ownership| match receive_message(control)? {
CoordinatorMessage::StagedAck(acknowledgment)
if acknowledgment.acknowledges(&ownership) =>
{
@ -493,7 +743,7 @@ fn run_supervisor(arguments: &RuntimeArguments) -> io::Result<()> {
return fail_with_network_cleanup(process, network, error);
}
if let Err(error) = send_message(
&control,
control,
&SupervisorEvent::Ready {
staged,
sentry_pid_inventory,
@ -501,7 +751,7 @@ fn run_supervisor(arguments: &RuntimeArguments) -> io::Result<()> {
) {
return fail_with_network_cleanup(process, network, error);
}
let committed: CoordinatorMessage = match receive_message(&control) {
let committed: CoordinatorMessage = match receive_message(control) {
Ok(committed) => committed,
Err(error) => return fail_with_network_cleanup(process, network, error),
};
@ -512,7 +762,7 @@ fn run_supervisor(arguments: &RuntimeArguments) -> io::Result<()> {
io::Error::other("sandbox commit acknowledgment was not received"),
);
}
supervise(&control, process, network)
supervise(control, process, network, state_anchor)
}
fn prepare_runtime_network(
@ -645,14 +895,23 @@ fn supervise(
control: &std::io::Stdout,
process: RetainedProcess,
mut network: Option<PreparedNetwork>,
state_anchor: &mut Option<RuntimeStateAnchorMount>,
) -> io::Result<()> {
let mut process = Some(process);
loop {
let exited = match process.has_exited() {
let exited = match process
.as_ref()
.map_or(Ok(false), RetainedProcess::has_exited)
{
Ok(exited) => exited,
Err(error) => return fail_with_network_cleanup(process, network, error),
Err(error) => return fail_active_runtime(process, network, error),
};
if exited {
child_broker().reap(process)?;
child_broker().reap(
process
.take()
.ok_or_else(|| io::Error::other("attached runsc handle is absent"))?,
)?;
stop_network(network)?;
send_message(control, &SupervisorEvent::AttachedRunscExited)?;
return Ok(());
@ -669,10 +928,12 @@ fn supervise(
}
let lease_lost = match lifetime_lost() {
Ok(lost) => lost,
Err(error) => return fail_with_network_cleanup(process, network, error),
Err(error) => return fail_active_runtime(process, network, error),
};
if lease_lost {
terminate_and_reap(process)?;
if let Some(process) = process.take() {
terminate_and_reap(process)?;
}
stop_network(network)?;
return Ok(());
}
@ -684,11 +945,11 @@ fn supervise(
tv_nsec: 10_000_000,
}),
) {
return fail_with_network_cleanup(process, network, error.into());
return fail_active_runtime(process, network, error.into());
}
let events = descriptors[0].revents();
if events.intersects(PollFlags::HUP | PollFlags::ERR) {
return fail_with_network_cleanup(
return fail_active_runtime(
process,
network,
io::Error::new(
@ -700,41 +961,110 @@ fn supervise(
if events.intersects(PollFlags::IN) {
let message = match receive_message(control) {
Ok(message) => message,
Err(error) => return fail_with_network_cleanup(process, network, error),
Err(error) => return fail_active_runtime(process, network, error),
};
match message {
CoordinatorMessage::SignalAttachedRunsc(signal) => {
if let Err(error) = process.send_signal(signal.signal()) {
return fail_with_network_cleanup(process, network, error);
}
if let Err(error) =
send_message(control, &SupervisorEvent::SignalApplied(signal))
{
return fail_with_network_cleanup(process, network, error);
}
}
CoordinatorMessage::Stop => {
terminate_and_reap(process)?;
stop_network(network)?;
send_message(control, &SupervisorEvent::Stopped)?;
return Ok(());
}
CoordinatorMessage::HelperStagedAck(_)
| CoordinatorMessage::NetworkAttestationAccepted(_)
| CoordinatorMessage::NetworkReadinessCommitted
| CoordinatorMessage::StagedAck(_)
| CoordinatorMessage::SandboxCommitted => {
return fail_with_network_cleanup(
process,
network,
io::Error::other("unexpected duplicate coordinator message"),
);
}
match handle_supervisor_message(
control,
&message,
&mut process,
&mut network,
state_anchor,
) {
Ok(true) => return Ok(()),
Ok(false) => {}
Err(error) => return fail_active_runtime(process, network, error),
}
}
}
}
fn handle_supervisor_message(
control: &std::io::Stdout,
message: &CoordinatorMessage,
process: &mut Option<RetainedProcess>,
network: &mut Option<PreparedNetwork>,
state_anchor: &mut Option<RuntimeStateAnchorMount>,
) -> io::Result<bool> {
match message {
CoordinatorMessage::SignalAttachedRunsc(signal) => {
process
.as_ref()
.ok_or_else(|| io::Error::other("attached runsc is already stopped"))?
.send_signal(signal.signal())?;
send_message(control, &SupervisorEvent::SignalApplied(*signal))?;
}
CoordinatorMessage::StopSandbox => {
terminate_and_reap(
process
.take()
.ok_or_else(|| io::Error::other("attached runsc is already stopped"))?,
)?;
send_message(control, &SupervisorEvent::SandboxStopped)?;
}
CoordinatorMessage::StopSidecar => {
if process.is_some() {
return Err(io::Error::other("sidecar stop preceded sandbox stop"));
}
stop_network(network.take())?;
send_message(control, &SupervisorEvent::SidecarStopped)?;
}
CoordinatorMessage::UnmountStateAnchor => {
if process.is_some() || network.is_some() {
return Err(io::Error::other(
"State Anchor unmount preceded runtime resource stop",
));
}
state_anchor
.take()
.ok_or_else(|| io::Error::other("State Anchor is absent"))?
.unmount()
.map_err(io::Error::other)?;
send_message(control, &SupervisorEvent::StateAnchorUnmounted)?;
}
CoordinatorMessage::Exit => {
if process.is_some() || network.is_some() || state_anchor.is_some() {
return Err(io::Error::other(
"Runtime Supervisor exit preceded resource cleanup",
));
}
send_message(control, &SupervisorEvent::Exited)?;
return Ok(true);
}
CoordinatorMessage::Stop => {
if let Some(process) = process.take() {
terminate_and_reap(process)?;
}
stop_network(network.take())?;
send_message(control, &SupervisorEvent::Stopped)?;
return Ok(true);
}
CoordinatorMessage::HelperStagedAck(_)
| CoordinatorMessage::MountStateAnchor
| CoordinatorMessage::StateAnchorStagedAck(_)
| CoordinatorMessage::StartSidecar
| CoordinatorMessage::NetworkAttestationAccepted(_)
| CoordinatorMessage::NetworkReadinessCommitted
| CoordinatorMessage::StagedAck(_)
| CoordinatorMessage::SandboxCommitted => {
return Err(io::Error::other("unexpected duplicate coordinator message"));
}
}
Ok(false)
}
fn fail_active_runtime(
process: Option<RetainedProcess>,
network: Option<PreparedNetwork>,
error: io::Error,
) -> io::Result<()> {
if let Some(process) = process {
fail_with_network_cleanup(process, network, error)
} else {
let _ = stop_network(network);
Err(error)
}
}
fn terminate_and_reap(process: RetainedProcess) -> io::Result<()> {
let term_error = process.send_signal(Signal::TERM).err();
let deadline = Instant::now() + TERMINATION_GRACE;
@ -890,6 +1220,42 @@ mod tests {
);
}
#[test]
fn coordinator_channel_preserves_phased_resource_commands() {
let (coordinator, supervisor) = private_seqpacket_pair().unwrap();
let channel = CoordinatorChannel::new(coordinator);
channel.mount_state_anchor().unwrap();
assert_eq!(
receive_message::<CoordinatorMessage>(&supervisor).unwrap(),
CoordinatorMessage::MountStateAnchor
);
channel.start_sidecar().unwrap();
assert_eq!(
receive_message::<CoordinatorMessage>(&supervisor).unwrap(),
CoordinatorMessage::StartSidecar
);
channel.stop_sandbox().unwrap();
assert_eq!(
receive_message::<CoordinatorMessage>(&supervisor).unwrap(),
CoordinatorMessage::StopSandbox
);
channel.stop_sidecar().unwrap();
assert_eq!(
receive_message::<CoordinatorMessage>(&supervisor).unwrap(),
CoordinatorMessage::StopSidecar
);
channel.unmount_state_anchor().unwrap();
assert_eq!(
receive_message::<CoordinatorMessage>(&supervisor).unwrap(),
CoordinatorMessage::UnmountStateAnchor
);
channel.exit().unwrap();
assert_eq!(
receive_message::<CoordinatorMessage>(&supervisor).unwrap(),
CoordinatorMessage::Exit
);
}
#[test]
fn runtime_supervisor_and_runsc_keep_distinct_edge_and_stdio_policies() {
let ownership = ProcessOwnershipContext {
@ -902,12 +1268,15 @@ mod tests {
target: TargetCommand::new("/proc/self/exe").with_lifetime_pipe_parent_edge(),
ownership,
policy_file: None,
policy_digest: None,
configuration_fingerprint: None,
runsc_command: RunscCommand::new(
"runsc",
[OsString::from("run"), OsString::from("container")],
[0; 32],
),
network_sidecar_id: None,
state_anchor: None,
};
let (_coordinator_control, supervisor_control) = private_seqpacket_pair().unwrap();
let (supervisor_lifetime, _coordinator_lifetime) = pipe_with(PipeFlags::CLOEXEC).unwrap();
@ -925,6 +1294,7 @@ mod tests {
slot: HostProcessOwnershipSlot::SandboxAttachedRunsc,
..ownership
},
state_anchor: None,
network: None,
runsc_program: "runsc".into(),
runsc_arguments: ["-root=/runtime", "run", "container"]