github/kata-containers
1
0
Fork 1
mirror of https://github.com/kata-containers/kata-containers.git synced 2025-04-29 20:24:31 +00:00
Code Issues Projects Releases Wiki Activity

agent: Remove some unwrap and expect calls

Browse Source 
Replace some `unwrap()` and `expect()` calls with code to return the
error to the caller.

Fixes: #3011.

Signed-off-by: James O. D. Hunt <james.o.hunt@intel.com>
This commit is contained in:
James O. D. Hunt 2021-11-22 14:32:37 +00:00
parent 351cef7b6a
commit adab64349c
6 changed files with 184 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
src/agent/src/device.rs
View File

@ -11,7 +11,7 @@ use std::fmt;
use std::fs; use std::fs;
use std::os::unix::ffi::OsStrExt; use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::MetadataExt; use std::os::unix::fs::MetadataExt;
use std::path::Path; use std::path::{Path, PathBuf};
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::Mutex; use tokio::sync::Mutex;
@ -157,21 +157,23 @@ pub fn pcipath_to_sysfs(root_bus_sysfs: &str, pcipath: &pci::Path) -> Result<Str
let bridgebuspath = format!("{}{}/pci_bus", root_bus_sysfs, relpath); let bridgebuspath = format!("{}{}/pci_bus", root_bus_sysfs, relpath);
let mut files: Vec<_> = fs::read_dir(&bridgebuspath)?.collect(); let mut files: Vec<_> = fs::read_dir(&bridgebuspath)?.collect();
if files.len() != 1 { match files.pop() {
return Err(anyhow!( Some(busfile) if files.is_empty() => {
"Expected exactly one PCI bus in {}, got {} instead", bus = busfile?
bridgebuspath,
files.len()
));
}
// unwrap is safe, because of the length test above
let busfile = files.pop().unwrap()?;
bus = busfile
.file_name() .file_name()
.into_string() .into_string()
.map_err(|e| anyhow!("Bad filename under {}: {:?}", &bridgebuspath, e))?; .map_err(|e| anyhow!("Bad filename under {}: {:?}", &bridgebuspath, e))?;
} }
_ => {
return Err(anyhow!(
"Expected exactly one PCI bus in {}, got {} instead",
bridgebuspath,
// Adjust to original value as we've already popped
files.len() + 1
));
}
};
}
Ok(relpath) Ok(relpath)
} }
@ -218,8 +220,9 @@ impl VirtioBlkPciMatcher {
fn new(relpath: &str) -> VirtioBlkPciMatcher { fn new(relpath: &str) -> VirtioBlkPciMatcher {
let root_bus = create_pci_root_bus_path(); let root_bus = create_pci_root_bus_path();
let re = format!(r"^{}{}/virtio[0-9]+/block/", root_bus, relpath); let re = format!(r"^{}{}/virtio[0-9]+/block/", root_bus, relpath);
VirtioBlkPciMatcher { VirtioBlkPciMatcher {
rex: Regex::new(&re).unwrap(), rex: Regex::new(&re).expect("BUG: failed to compile VirtioBlkPciMatcher regex"),
} }
} }
} }
@ -257,7 +260,7 @@ impl VirtioBlkCCWMatcher {
root_bus_path, device root_bus_path, device
); );
VirtioBlkCCWMatcher { VirtioBlkCCWMatcher {
rex: Regex::new(&re).unwrap(), rex: Regex::new(&re).expect("BUG: failed to compile VirtioBlkCCWMatcher regex"),
} }
} }
} }
@ -413,12 +416,15 @@ fn scan_scsi_bus(scsi_addr: &str) -> Result<()> {
for entry in fs::read_dir(SYSFS_SCSI_HOST_PATH)? { for entry in fs::read_dir(SYSFS_SCSI_HOST_PATH)? {
let host = entry?.file_name(); let host = entry?.file_name();
let scan_path = format!(
"{}/{}/{}", let host_str = host.to_str().ok_or_else(|| {
SYSFS_SCSI_HOST_PATH, anyhow!(
host.to_str().unwrap(), "failed to convert directory entry to unicode for file {:?}",
"scan" host
); )
})?;
let scan_path = PathBuf::from(&format!("{}/{}/{}", SYSFS_SCSI_HOST_PATH, host_str, "scan"));
fs::write(scan_path, &scan_data)?; fs::write(scan_path, &scan_data)?;
} }
@ -722,25 +728,17 @@ async fn vfio_device_handler(device: &Device, sandbox: &Arc<Mutex<Sandbox>>) ->
if vfio_in_guest { if vfio_in_guest {
pci_driver_override(SYSFS_BUS_PCI_PATH, guestdev, "vfio-pci")?; pci_driver_override(SYSFS_BUS_PCI_PATH, guestdev, "vfio-pci")?;
let devgroup = pci_iommu_group(SYSFS_BUS_PCI_PATH, guestdev)?;
if devgroup.is_none() {
// Devices must have an IOMMU group to be usable via VFIO // Devices must have an IOMMU group to be usable via VFIO
return Err(anyhow!("{} has no IOMMU group", guestdev)); let devgroup = pci_iommu_group(SYSFS_BUS_PCI_PATH, guestdev)?
.ok_or_else(|| anyhow!("{} has no IOMMU group", guestdev))?;
if let Some(g) = group {
if g != devgroup {
return Err(anyhow!("{} is not in guest IOMMU group {}", guestdev, g));
}
} }
if group.is_some() && group != devgroup { group = Some(devgroup);
// If PCI devices associated with the same VFIO device
// (and therefore group) in the host don't end up in
// the same group in the guest, something has gone
// horribly wrong
return Err(anyhow!(
"{} is not in guest IOMMU group {}",
guestdev,
group.unwrap()
));
}
group = devgroup;
pci_fixups.push((host, guestdev)); pci_fixups.push((host, guestdev));
} }
@ -748,7 +746,8 @@ async fn vfio_device_handler(device: &Device, sandbox: &Arc<Mutex<Sandbox>>) ->
let dev_update = if vfio_in_guest { let dev_update = if vfio_in_guest {
// If there are any devices at all, logic above ensures that group is not None // If there are any devices at all, logic above ensures that group is not None
let group = group.unwrap(); let group = group.ok_or_else(|| anyhow!("failed to get VFIO group: {:?}"))?;
let vm_path = get_vfio_device_name(sandbox, group).await?; let vm_path = get_vfio_device_name(sandbox, group).await?;
Some(DevUpdate::from_vm_path(&vm_path, vm_path.clone())?) Some(DevUpdate::from_vm_path(&vm_path, vm_path.clone())?)
@ -844,11 +843,8 @@ pub fn update_device_cgroup(spec: &mut Spec) -> Result<()> {
.as_mut() .as_mut()
.ok_or_else(|| anyhow!("Spec didn't container linux field"))?; .ok_or_else(|| anyhow!("Spec didn't container linux field"))?;
if linux.resources.is_none() { let resources = linux.resources.get_or_insert(LinuxResources::default());
linux.resources = Some(LinuxResources::default());
}
let resources = linux.resources.as_mut().unwrap();
resources.devices.push(LinuxDeviceCgroup { resources.devices.push(LinuxDeviceCgroup {
allow: false, allow: false,
major: Some(major), major: Some(major),

8
src/agent/src/main.rs
View File

@ -113,10 +113,10 @@ async fn create_logger_task(rfd: RawFd, vsock_port: u32, shutdown: Receiver<bool
)?; )?;
let addr = SockAddr::new_vsock(libc::VMADDR_CID_ANY, vsock_port); let addr = SockAddr::new_vsock(libc::VMADDR_CID_ANY, vsock_port);
socket::bind(listenfd, &addr).unwrap(); socket::bind(listenfd, &addr)?;
socket::listen(listenfd, 1).unwrap(); socket::listen(listenfd, 1)?;
writer = Box::new(util::get_vsock_stream(listenfd).await.unwrap()); writer = Box::new(util::get_vsock_stream(listenfd).await?);
} else { } else {
writer = Box::new(tokio::io::stdout()); writer = Box::new(tokio::io::stdout());
} }
@ -326,7 +326,7 @@ async fn start_sandbox(
sandbox.lock().await.sender = Some(tx); sandbox.lock().await.sender = Some(tx);
// vsock:///dev/vsock, port // vsock:///dev/vsock, port
let mut server = rpc::start(sandbox.clone(), config.server_addr.as_str()); let mut server = rpc::start(sandbox.clone(), config.server_addr.as_str())?;
server.start().await?; server.start().await?;
rx.await?; rx.await?;

32
src/agent/src/metrics.rs
View File

@ -8,6 +8,7 @@ extern crate procfs;
use prometheus::{Encoder, Gauge, GaugeVec, IntCounter, TextEncoder}; use prometheus::{Encoder, Gauge, GaugeVec, IntCounter, TextEncoder};
use anyhow::Result; use anyhow::Result;
use slog::warn;
use tracing::instrument; use tracing::instrument;
const NAMESPACE_KATA_AGENT: &str = "kata_agent"; const NAMESPACE_KATA_AGENT: &str = "kata_agent";
@ -74,7 +75,7 @@ pub fn get_metrics(_: &protocols::agent::GetMetricsRequest) -> Result<String> {
AGENT_SCRAPE_COUNT.inc(); AGENT_SCRAPE_COUNT.inc();
// update agent process metrics // update agent process metrics
update_agent_metrics(); update_agent_metrics()?;
// update guest os metrics // update guest os metrics
update_guest_metrics(); update_guest_metrics();
@ -84,23 +85,26 @@ pub fn get_metrics(_: &protocols::agent::GetMetricsRequest) -> Result<String> {
let mut buffer = Vec::new(); let mut buffer = Vec::new();
let encoder = TextEncoder::new(); let encoder = TextEncoder::new();
encoder.encode(&metric_families, &mut buffer).unwrap(); encoder.encode(&metric_families, &mut buffer)?;
Ok(String::from_utf8(buffer).unwrap()) Ok(String::from_utf8(buffer)?)
} }
#[instrument] #[instrument]
fn update_agent_metrics() { fn update_agent_metrics() -> Result<()> {
let me = procfs::process::Process::myself(); let me = procfs::process::Process::myself();
if let Err(err) = me { let me = match me {
error!(sl!(), "failed to create process instance: {:?}", err); Ok(p) => p,
return; Err(e) => {
// FIXME: return Ok for all errors?
warn!(sl!(), "failed to create process instance: {:?}", e);
return Ok(());
} }
};
let me = me.unwrap(); let tps = procfs::ticks_per_second()?;
let tps = procfs::ticks_per_second().unwrap();
// process total time // process total time
AGENT_TOTAL_TIME.set((me.stat.utime + me.stat.stime) as f64 / (tps as f64)); AGENT_TOTAL_TIME.set((me.stat.utime + me.stat.stime) as f64 / (tps as f64));
@ -109,7 +113,7 @@ fn update_agent_metrics() {
AGENT_TOTAL_VM.set(me.stat.vsize as f64); AGENT_TOTAL_VM.set(me.stat.vsize as f64);
// Total resident set // Total resident set
let page_size = procfs::page_size().unwrap() as f64; let page_size = procfs::page_size()? as f64;
AGENT_TOTAL_RSS.set(me.stat.rss as f64 * page_size); AGENT_TOTAL_RSS.set(me.stat.rss as f64 * page_size);
// io // io
@ -132,11 +136,11 @@ fn update_agent_metrics() {
} }
match me.status() { match me.status() {
Err(err) => { Err(err) => error!(sl!(), "failed to get process status: {:?}", err),
info!(sl!(), "failed to get process status: {:?}", err);
}
Ok(status) => set_gauge_vec_proc_status(&AGENT_PROC_STATUS, &status), Ok(status) => set_gauge_vec_proc_status(&AGENT_PROC_STATUS, &status),
} }
return Ok(());
} }
#[instrument] #[instrument]

21
src/agent/src/mount.rs
View File

@ -628,8 +628,7 @@ pub fn get_mount_fs_type_from_file(mount_file: &str, mount_point: &str) -> Resul
let file = File::open(mount_file)?; let file = File::open(mount_file)?;
let reader = BufReader::new(file); let reader = BufReader::new(file);
let re = Regex::new(format!("device .+ mounted on {} with fstype (.+)", mount_point).as_str()) let re = Regex::new(format!("device .+ mounted on {} with fstype (.+)", mount_point).as_str())?;
.unwrap();
// Read the file line by line using the lines() iterator from std::io::BufRead. // Read the file line by line using the lines() iterator from std::io::BufRead.
for (_index, line) in reader.lines().enumerate() { for (_index, line) in reader.lines().enumerate() {
@ -707,20 +706,21 @@ pub fn get_cgroup_mounts(
} }
} }
if fields[0].is_empty() { let subsystem_name = fields[0];
if subsystem_name.is_empty() {
continue; continue;
} }
if fields[0] == "devices" { if subsystem_name == "devices" {
has_device_cgroup = true; has_device_cgroup = true;
} }
if let Some(value) = CGROUPS.get(&fields[0]) { if let Some((key, value)) = CGROUPS.get_key_value(subsystem_name) {
let key = CGROUPS.keys().find(|&&f| f == fields[0]).unwrap();
cg_mounts.push(InitMount { cg_mounts.push(InitMount {
fstype: "cgroup", fstype: "cgroup",
src: "cgroup", src: "cgroup",
dest: *value, dest: value,
options: vec!["nosuid", "nodev", "noexec", "relatime", key], options: vec!["nosuid", "nodev", "noexec", "relatime", key],
}); });
} }
@ -773,10 +773,9 @@ fn ensure_destination_file_exists(path: &Path) -> Result<()> {
return Err(anyhow!("{:?} exists but is not a regular file", path)); return Err(anyhow!("{:?} exists but is not a regular file", path));
} }
// The only way parent() can return None is if the path is /, let dir = path
// which always exists, so the test above will already have caught .parent()
// it, thus the unwrap() is safe .ok_or_else(|| anyhow!("failed to find parent path for {:?}", path))?;
let dir = path.parent().unwrap();
fs::create_dir_all(dir).context(format!("create_dir_all {:?}", dir))?; fs::create_dir_all(dir).context(format!("create_dir_all {:?}", dir))?;

155
src/agent/src/rpc.rs
View File

@ -183,7 +183,7 @@ impl AgentService {
update_device_cgroup(&mut oci)?; update_device_cgroup(&mut oci)?;
// Append guest hooks // Append guest hooks
append_guest_hooks(&s, &mut oci); append_guest_hooks(&s, &mut oci)?;
// write spec to bundle path, hooks might // write spec to bundle path, hooks might
// read ocispec // read ocispec
@ -205,21 +205,14 @@ impl AgentService {
LinuxContainer::new(cid.as_str(), CONTAINER_BASE, opts, &sl!())?; LinuxContainer::new(cid.as_str(), CONTAINER_BASE, opts, &sl!())?;
let pipe_size = AGENT_CONFIG.read().await.container_pipe_size; let pipe_size = AGENT_CONFIG.read().await.container_pipe_size;
let p = if oci.process.is_some() {
Process::new( let p = if let Some(p) = oci.process {
&sl!(), Process::new(&sl!(), &p, cid.as_str(), true, pipe_size)?
oci.process.as_ref().unwrap(),
cid.as_str(),
true,
pipe_size,
)?
} else { } else {
info!(sl!(), "no process configurations!"); info!(sl!(), "no process configurations!");
return Err(anyhow!(nix::Error::from_errno(nix::errno::Errno::EINVAL))); return Err(anyhow!(nix::Error::from_errno(nix::errno::Errno::EINVAL)));
}; };
ctr.start(p).await?; ctr.start(p).await?;
s.update_shared_pidns(&ctr)?; s.update_shared_pidns(&ctr)?;
s.add_container(ctr); s.add_container(ctr);
info!(sl!(), "created container!"); info!(sl!(), "created container!");
@ -241,11 +234,17 @@ impl AgentService {
ctr.exec()?; ctr.exec()?;
if sid == cid {
return Ok(());
}
// start oom event loop // start oom event loop
if sid != cid && ctr.cgroup_manager.is_some() { if let Some(ref ctr) = ctr.cgroup_manager {
let cg_path = ctr.cgroup_manager.as_ref().unwrap().get_cg_path("memory"); let cg_path = ctr.get_cg_path("memory");
if cg_path.is_some() {
let rx = notifier::notify_oom(cid.as_str(), cg_path.unwrap()).await?; if let Some(cg_path) = cg_path {
let rx = notifier::notify_oom(cid.as_str(), cg_path.to_string()).await?;
s.run_oom_event_monitor(rx, cid.clone()).await; s.run_oom_event_monitor(rx, cid.clone()).await;
} }
} }
@ -345,14 +344,13 @@ impl AgentService {
let s = self.sandbox.clone(); let s = self.sandbox.clone();
let mut sandbox = s.lock().await; let mut sandbox = s.lock().await;
let process = if req.process.is_some() { let process = req
req.process.as_ref().unwrap() .process
} else { .into_option()
return Err(anyhow!(nix::Error::from_errno(nix::errno::Errno::EINVAL))); .ok_or_else(|| anyhow!(nix::Error::from_errno(nix::errno::Errno::EINVAL)))?;
};
let pipe_size = AGENT_CONFIG.read().await.container_pipe_size; let pipe_size = AGENT_CONFIG.read().await.container_pipe_size;
let ocip = rustjail::process_grpc_to_oci(process); let ocip = rustjail::process_grpc_to_oci(&process);
let p = Process::new(&sl!(), &ocip, exec_id.as_str(), false, pipe_size)?; let p = Process::new(&sl!(), &ocip, exec_id.as_str(), false, pipe_size)?;
let ctr = sandbox let ctr = sandbox
@ -385,7 +383,12 @@ impl AgentService {
let p = find_process(&mut sandbox, cid.as_str(), eid.as_str(), init)?; let p = find_process(&mut sandbox, cid.as_str(), eid.as_str(), init)?;
let mut signal = Signal::try_from(req.signal as i32).unwrap(); let mut signal = Signal::try_from(req.signal as i32).map_err(|e| {
anyhow!(e).context(format!(
"failed to convert {:?} to signal (container-id: {}, exec-id: {})",
req.signal, cid, eid
))
})?;
// For container initProcess, if it hasn't installed handler for "SIGTERM" signal, // For container initProcess, if it hasn't installed handler for "SIGTERM" signal,
// it will ignore the "SIGTERM" signal sent to it, thus send it "SIGKILL" signal // it will ignore the "SIGTERM" signal sent to it, thus send it "SIGKILL" signal
@ -444,7 +447,11 @@ impl AgentService {
Some(p) => p, Some(p) => p,
None => { None => {
// Lost race, pick up exit code from channel // Lost race, pick up exit code from channel
resp.status = exit_recv.recv().await.unwrap(); resp.status = exit_recv
.recv()
.await
.ok_or_else(|| anyhow!("Failed to receive exit code"))?;
return Ok(resp); return Ok(resp);
} }
}; };
@ -486,7 +493,7 @@ impl AgentService {
} }
}; };
let writer = writer.unwrap(); let writer = writer.ok_or_else(|| anyhow!("cannot get writer"))?;
writer.lock().await.write_all(req.data.as_slice()).await?; writer.lock().await.write_all(req.data.as_slice()).await?;
let mut resp = WriteStreamResponse::new(); let mut resp = WriteStreamResponse::new();
@ -528,7 +535,7 @@ impl AgentService {
return Err(anyhow!(nix::Error::from_errno(nix::errno::Errno::EINVAL))); return Err(anyhow!(nix::Error::from_errno(nix::errno::Errno::EINVAL)));
} }
let reader = reader.unwrap(); let reader = reader.ok_or_else(|| anyhow!("cannot get stream reader"))?;
tokio::select! { tokio::select! {
_ = term_exit_notifier.notified() => { _ = term_exit_notifier.notified() => {
@ -646,8 +653,8 @@ impl protocols::agent_ttrpc::AgentService for AgentService {
let resp = Empty::new(); let resp = Empty::new();
if res.is_some() { if let Some(res) = res.as_ref() {
let oci_res = rustjail::resources_grpc_to_oci(&res.unwrap()); let oci_res = rustjail::resources_grpc_to_oci(res);
match ctr.set(oci_res) { match ctr.set(oci_res) {
Err(e) => { Err(e) => {
return Err(ttrpc_error(ttrpc::Code::INTERNAL, e.to_string())); return Err(ttrpc_error(ttrpc::Code::INTERNAL, e.to_string()));
@ -807,11 +814,7 @@ impl protocols::agent_ttrpc::AgentService for AgentService {
) )
})?; })?;
if p.term_master.is_none() { if let Some(fd) = p.term_master {
return Err(ttrpc_error(ttrpc::Code::UNAVAILABLE, "no tty".to_string()));
}
let fd = p.term_master.unwrap();
unsafe { unsafe {
let win = winsize { let win = winsize {
ws_row: req.row as c_ushort, ws_row: req.row as c_ushort,
@ -821,9 +824,12 @@ impl protocols::agent_ttrpc::AgentService for AgentService {
}; };
let err = libc::ioctl(fd, TIOCSWINSZ, &win); let err = libc::ioctl(fd, TIOCSWINSZ, &win);
Errno::result(err) Errno::result(err).map(drop).map_err(|e| {
.map(drop) ttrpc_error(ttrpc::Code::INTERNAL, format!("ioctl error: {:?}", e))
.map_err(|e| ttrpc_error(ttrpc::Code::INTERNAL, format!("ioctl error: {:?}", e)))?; })?;
}
} else {
return Err(ttrpc_error(ttrpc::Code::UNAVAILABLE, "no tty".to_string()));
} }
Ok(Empty::new()) Ok(Empty::new())
@ -1027,12 +1033,25 @@ impl protocols::agent_ttrpc::AgentService for AgentService {
let mut sandbox = s.lock().await; let mut sandbox = s.lock().await;
// destroy all containers, clean up, notify agent to exit // destroy all containers, clean up, notify agent to exit
// etc. // etc.
sandbox.destroy().await.unwrap(); sandbox
.destroy()
.await
.map_err(|e| ttrpc_error(ttrpc::Code::INTERNAL, e.to_string()))?;
// Close get_oom_event connection, // Close get_oom_event connection,
// otherwise it will block the shutdown of ttrpc. // otherwise it will block the shutdown of ttrpc.
sandbox.event_tx.take(); sandbox.event_tx.take();
sandbox.sender.take().unwrap().send(1).unwrap(); sandbox
.sender
.take()
.ok_or_else(|| {
ttrpc_error(
ttrpc::Code::INTERNAL,
"failed to get sandbox sender channel".to_string(),
)
})?
.send(1)
.map_err(|e| ttrpc_error(ttrpc::Code::INTERNAL, e.to_string()))?;
Ok(Empty::new()) Ok(Empty::new())
} }
@ -1291,11 +1310,7 @@ fn get_memory_info(block_size: bool, hotplug: bool) -> Result<(u64, bool)> {
match stat::stat(SYSFS_MEMORY_HOTPLUG_PROBE_PATH) { match stat::stat(SYSFS_MEMORY_HOTPLUG_PROBE_PATH) {
Ok(_) => plug = true, Ok(_) => plug = true,
Err(e) => { Err(e) => {
info!( info!(sl!(), "hotplug memory error: {:?}", e);
sl!(),
"hotplug memory error: {}",
e.as_errno().unwrap().desc()
);
match e { match e {
nix::Error::Sys(errno) => match errno { nix::Error::Sys(errno) => match errno {
Errno::ENOENT => plug = false, Errno::ENOENT => plug = false,
@ -1371,7 +1386,7 @@ fn find_process<'a>(
ctr.get_process(eid).map_err(|_| anyhow!("Invalid exec id")) ctr.get_process(eid).map_err(|_| anyhow!("Invalid exec id"))
} }
pub fn start(s: Arc<Mutex<Sandbox>>, server_address: &str) -> TtrpcServer { pub fn start(s: Arc<Mutex<Sandbox>>, server_address: &str) -> Result<TtrpcServer> {
let agent_service = Box::new(AgentService { sandbox: s }) let agent_service = Box::new(AgentService { sandbox: s })
as Box<dyn protocols::agent_ttrpc::AgentService + Send + Sync>; as Box<dyn protocols::agent_ttrpc::AgentService + Send + Sync>;
@ -1386,14 +1401,13 @@ pub fn start(s: Arc<Mutex<Sandbox>>, server_address: &str) -> TtrpcServer {
let hservice = protocols::health_ttrpc::create_health(health_worker); let hservice = protocols::health_ttrpc::create_health(health_worker);
let server = TtrpcServer::new() let server = TtrpcServer::new()
.bind(server_address) .bind(server_address)?
.unwrap()
.register_service(aservice) .register_service(aservice)
.register_service(hservice); .register_service(hservice);
info!(sl!(), "ttRPC server started"; "address" => server_address); info!(sl!(), "ttRPC server started"; "address" => server_address);
server Ok(server)
} }
// This function updates the container namespaces configuration based on the // This function updates the container namespaces configuration based on the
@ -1438,24 +1452,28 @@ fn update_container_namespaces(
// the create_sandbox request or create_container request. // the create_sandbox request or create_container request.
// Else set this to empty string so that a new pid namespace is // Else set this to empty string so that a new pid namespace is
// created for the container. // created for the container.
if sandbox_pidns && sandbox.sandbox_pidns.is_some() { if sandbox_pidns {
pid_ns.path = String::from(sandbox.sandbox_pidns.as_ref().unwrap().path.as_str()); if let Some(ref pidns) = &sandbox.sandbox_pidns {
pid_ns.path = String::from(pidns.path.as_str());
} else {
return Err(anyhow!("failed to get sandbox pidns"));
}
} }
linux.namespaces.push(pid_ns); linux.namespaces.push(pid_ns);
Ok(()) Ok(())
} }
fn append_guest_hooks(s: &Sandbox, oci: &mut Spec) { fn append_guest_hooks(s: &Sandbox, oci: &mut Spec) -> Result<()> {
if s.hooks.is_none() { if let Some(ref guest_hooks) = s.hooks {
return;
}
let guest_hooks = s.hooks.as_ref().unwrap();
let mut hooks = oci.hooks.take().unwrap_or_default(); let mut hooks = oci.hooks.take().unwrap_or_default();
hooks.prestart.append(&mut guest_hooks.prestart.clone()); hooks.prestart.append(&mut guest_hooks.prestart.clone());
hooks.poststart.append(&mut guest_hooks.poststart.clone()); hooks.poststart.append(&mut guest_hooks.poststart.clone());
hooks.poststop.append(&mut guest_hooks.poststop.clone()); hooks.poststop.append(&mut guest_hooks.poststop.clone());
oci.hooks = Some(hooks); oci.hooks = Some(hooks);
}
Ok(())
} }
// Check is the container process installed the // Check is the container process installed the
@ -1545,7 +1563,7 @@ fn do_copy_file(req: &CopyFileRequest) -> Result<()> {
PathBuf::from("/") PathBuf::from("/")
}; };
fs::create_dir_all(dir.to_str().unwrap()).or_else(|e| { fs::create_dir_all(&dir).or_else(|e| {
if e.kind() != std::io::ErrorKind::AlreadyExists { if e.kind() != std::io::ErrorKind::AlreadyExists {
return Err(e); return Err(e);
} }
@ -1553,10 +1571,7 @@ fn do_copy_file(req: &CopyFileRequest) -> Result<()> {
Ok(()) Ok(())
})?; })?;
std::fs::set_permissions( std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(req.dir_mode))?;
dir.to_str().unwrap(),
std::fs::Permissions::from_mode(req.dir_mode),
)?;
let mut tmpfile = path.clone(); let mut tmpfile = path.clone();
tmpfile.set_extension("tmp"); tmpfile.set_extension("tmp");
@ -1565,10 +1580,10 @@ fn do_copy_file(req: &CopyFileRequest) -> Result<()> {
.write(true) .write(true)
.create(true) .create(true)
.truncate(false) .truncate(false)
.open(tmpfile.to_str().unwrap())?; .open(&tmpfile)?;
file.write_all_at(req.data.as_slice(), req.offset as u64)?; file.write_all_at(req.data.as_slice(), req.offset as u64)?;
let st = stat::stat(tmpfile.to_str().unwrap())?; let st = stat::stat(&tmpfile)?;
if st.st_size != req.file_size { if st.st_size != req.file_size {
return Ok(()); return Ok(());
@ -1577,7 +1592,7 @@ fn do_copy_file(req: &CopyFileRequest) -> Result<()> {
file.set_permissions(std::fs::Permissions::from_mode(req.file_mode))?; file.set_permissions(std::fs::Permissions::from_mode(req.file_mode))?;
unistd::chown( unistd::chown(
tmpfile.to_str().unwrap(), &tmpfile,
Some(Uid::from_raw(req.uid as u32)), Some(Uid::from_raw(req.uid as u32)),
Some(Gid::from_raw(req.gid as u32)), Some(Gid::from_raw(req.gid as u32)),
)?; )?;
@ -1646,10 +1661,18 @@ fn setup_bundle(cid: &str, spec: &mut Spec) -> Result<PathBuf> {
readonly: spec_root.readonly, readonly: spec_root.readonly,
}); });
let _ = spec.save(config_path.to_str().unwrap()); let _ = spec.save(
config_path
.to_str()
.ok_or_else(|| anyhow!("cannot convert path to unicode"))?,
);
let olddir = unistd::getcwd().context("cannot getcwd")?; let olddir = unistd::getcwd().context("cannot getcwd")?;
unistd::chdir(bundle_path.to_str().unwrap())?; unistd::chdir(
bundle_path
.to_str()
.ok_or_else(|| anyhow!("cannot convert bundle path to unicode"))?,
)?;
Ok(olddir) Ok(olddir)
} }
@ -1682,8 +1705,8 @@ fn load_kernel_module(module: &protocols::agent::KernelModule) -> Result<()> {
match status.code() { match status.code() {
Some(code) => { Some(code) => {
let std_out: String = String::from_utf8(output.stdout).unwrap(); let std_out = String::from_utf8_lossy(&output.stdout);
let std_err: String = String::from_utf8(output.stderr).unwrap(); let std_err = String::from_utf8_lossy(&output.stderr);
let msg = format!( let msg = format!(
"load_kernel_module return code: {} stdout:{} stderr:{}", "load_kernel_module return code: {} stdout:{} stderr:{}",
code, std_out, std_err code, std_out, std_err
@ -1746,7 +1769,7 @@ mod tests {
let mut oci = Spec { let mut oci = Spec {
..Default::default() ..Default::default()
}; };
append_guest_hooks(&s, &mut oci); append_guest_hooks(&s, &mut oci).unwrap();
assert_eq!(s.hooks, oci.hooks); assert_eq!(s.hooks, oci.hooks);
} }

14
src/agent/src/util.rs
View File

@ -3,7 +3,7 @@
// SPDX-License-Identifier: Apache-2.0 // SPDX-License-Identifier: Apache-2.0
// //
use anyhow::Result; use anyhow::{anyhow, Result};
use futures::StreamExt; use futures::StreamExt;
use std::io; use std::io;
use std::io::ErrorKind; use std::io::ErrorKind;
@ -64,8 +64,12 @@ pub fn get_vsock_incoming(fd: RawFd) -> Incoming {
#[instrument] #[instrument]
pub async fn get_vsock_stream(fd: RawFd) -> Result<VsockStream> { pub async fn get_vsock_stream(fd: RawFd) -> Result<VsockStream> {
let stream = get_vsock_incoming(fd).next().await.unwrap()?; let stream = get_vsock_incoming(fd)
Ok(stream) .next()
.await
.ok_or_else(|| anyhow!("cannot handle incoming vsock connection"))?;
Ok(stream?)
} }
#[cfg(test)] #[cfg(test)]
@ -124,7 +128,9 @@ mod tests {
let mut vec_locked = vec_ref.lock(); let mut vec_locked = vec_ref.lock();
let v = vec_locked.as_deref_mut().unwrap(); let v = vec_locked
.as_deref_mut()
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
std::io::Write::flush(v) std::io::Write::flush(v)
} }