containerd
/
rust-extensions
mirror of https://github.com/containerd/rust-extensions.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make one function that works for both platforms 

Signed-off-by: James Sturtevant <jstur@microsoft.com>
This commit is contained in:
James Sturtevant 2023-04-27 15:58:12 -07:00
parent bceaf4aca3
commit fa4a0045e6
No known key found for this signature in database
7 changed files with 161 additions and 203 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/ci-windows.yml vendored
View File

@ -97,3 +97,5 @@ jobs:
cargo run --example skeleton -- -namespace default -id 1234 -address "\\.\pipe\containerd-containerd" -publish-binary ./bin/containerd start
ps skeleton
cargo run --example shim-proto-connect \\.\pipe\containerd-shim-17630016127144989388-pipe
$skeleton = get-process skeleton -ErrorAction SilentlyContinue
if ($skeleton) { exit 1 }

2
crates/shim/src/asynchronous/mod.rs
View File

@ -166,7 +166,7 @@ where
}
_ => {
if !config.no_setup_logger {
logger::init(flags.debug)?;
logger::init(flags.debug, &flags.namespace, &flags.id)?;
}
let publisher = RemotePublisher::new(&ttrpc_address).await?;

18
crates/shim/src/lib.rs
View File

@ -181,17 +181,15 @@ pub fn socket_address(socket_path: &str, namespace: &str, id: &str) -> String {
hasher.finish()
};
format_address(hash)
}
#[cfg(unix)]
{
format!("unix://{}/{:x}.sock", SOCKET_ROOT, hash)
}
#[cfg(windows)]
fn format_address(hash: u64) -> String {
format!(r"\\.\pipe\containerd-shim-{}-pipe", hash)
}
#[cfg(unix)]
fn format_address(hash: u64) -> String {
format!("unix://{}/{:x}.sock", SOCKET_ROOT, hash)
#[cfg(windows)]
{
format!(r"\\.\pipe\containerd-shim-{}-pipe", hash)
}
}
#[cfg(unix)]

32
crates/shim/src/logger.rs
View File

@ -78,37 +78,27 @@ impl log::Log for FifoLogger {
}
}
#[cfg(unix)]
pub fn init(debug: bool) -> Result<(), Error> {
pub fn init(debug: bool, _namespace: &str, _id: &str) -> Result<(), Error> {
#[cfg(unix)]
let logger = FifoLogger::new().map_err(io_error!(e, "failed to init logger"))?;
let boxed_logger = Box::new(logger);
configure_logger(boxed_logger, debug)
}
#[cfg(windows)]
// Containerd on windows expects the log to be a named pipe in the format of \\.\pipe\containerd-<namespace>-<id>-log
// There is an assumption that there is always only one client connected which is containerd.
// If there is a restart of containerd then logs during that time period will be lost.
//
// https://github.com/containerd/containerd/blob/v1.7.0/runtime/v2/shim_windows.go#L77
// https://github.com/microsoft/hcsshim/blob/5871d0c4436f131c377655a3eb09fc9b5065f11d/cmd/containerd-shim-runhcs-v1/serve.go#L132-L137
pub fn init(debug: bool, namespace: &String, id: &String) -> Result<(), Error> {
// Containerd on windows expects the log to be a named pipe in the format of \\.\pipe\containerd-<namespace>-<id>-log
// There is an assumption that there is always only one client connected which is containerd.
// If there is a restart of containerd then logs during that time period will be lost.
//
// https://github.com/containerd/containerd/blob/v1.7.0/runtime/v2/shim_windows.go#L77
// https://github.com/microsoft/hcsshim/blob/5871d0c4436f131c377655a3eb09fc9b5065f11d/cmd/containerd-shim-runhcs-v1/serve.go#L132-L137
#[cfg(windows)]
let logger =
NamedPipeLogger::new(namespace, id).map_err(io_error!(e, "failed to init logger"))?;
NamedPipeLogger::new(_namespace, _id).map_err(io_error!(e, "failed to init logger"))?;
let boxed_logger = Box::new(logger);
configure_logger(boxed_logger, debug)
}
fn configure_logger(logger: Box<dyn log::Log>, debug: bool) -> Result<(), Error> {
let level = if debug {
log::LevelFilter::Debug
} else {
log::LevelFilter::Info
};
log::set_boxed_logger(logger)?;
log::set_boxed_logger(Box::new(logger))?;
log::set_max_level(level);
Ok(())
}

242
crates/shim/src/synchronous/mod.rs
View File

@ -74,7 +74,6 @@ use crate::{
};
cfg_unix! {
use std::os::unix::net::UnixListener;
use crate::{SOCKET_FD, parse_sockaddr};
use command_fds::{CommandFdExt, FdMapping};
use libc::{SIGCHLD, SIGINT, SIGPIPE, SIGTERM};
@ -93,20 +92,23 @@ cfg_unix! {
}
cfg_windows! {
use std::{io, ptr};
use std::{
io, ptr,
fs::OpenOptions,
os::windows::prelude::{AsRawHandle, OpenOptionsExt},
};
use windows_sys::Win32::{
Foundation::{CloseHandle, HANDLE},
System::{
Console::SetConsoleCtrlHandler,
Threading::{CreateSemaphoreA, ReleaseSemaphore, },
Threading::{CreateSemaphoreA, ReleaseSemaphore, WaitForSingleObject, INFINITE},
},
Storage::FileSystem::FILE_FLAG_OVERLAPPED
};
static mut SEMAPHORE: HANDLE = 0 as HANDLE;
const MAX_SEM_COUNT: i32 = 255;
use windows_sys::Win32::System::Threading::WaitForSingleObject;
use windows_sys::Win32::System::Threading::INFINITE;
}
pub mod monitor;
@ -248,9 +250,6 @@ where
util::setup_debugger_event();
if !config.no_setup_logger {
#[cfg(unix)]
logger::init(flags.debug)?;
#[cfg(windows)]
logger::init(flags.debug, &flags.namespace, &flags.id)?;
}
@ -281,18 +280,20 @@ where
}
}
#[cfg(windows)]
fn create_server(flags: args::Flags) -> Result<Server> {
let mut server = Server::new();
let address = socket_address(&flags.address, &flags.namespace, &flags.id);
server = server.bind(address.as_str())?;
Ok(server)
}
#[cfg(unix)]
fn create_server(_flags: args::Flags) -> Result<Server> {
let mut server = Server::new();
server = server.add_listener(SOCKET_FD)?;
#[cfg(unix)]
{
server = server.add_listener(SOCKET_FD)?;
}
#[cfg(windows)]
{
let address = socket_address(&_flags.address, &_flags.namespace, &_flags.id);
server = server.bind(address.as_str())?;
}
Ok(server)
}
@ -399,34 +400,29 @@ fn remove_socket_silently(address: &str) {
remove_socket(address).unwrap_or_else(|e| warn!("failed to remove file {} {:?}", address, e))
}
#[cfg(unix)]
fn remove_socket(address: &str) -> Result<()> {
let path = parse_sockaddr(address);
if let Ok(md) = Path::new(path).metadata() {
if md.file_type().is_socket() {
fs::remove_file(path).map_err(io_error!(e, "remove socket"))?;
#[cfg(unix)]
{
let path = parse_sockaddr(address);
if let Ok(md) = Path::new(path).metadata() {
if md.file_type().is_socket() {
fs::remove_file(path).map_err(io_error!(e, "remove socket"))?;
}
}
}
Ok(())
}
#[cfg(windows)]
fn remove_socket(address: &str) -> Result<()> {
use std::{
fs::OpenOptions,
os::windows::prelude::{AsRawHandle, OpenOptionsExt},
};
use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OVERLAPPED;
let mut opts = OpenOptions::new();
opts.read(true)
.write(true)
.custom_flags(FILE_FLAG_OVERLAPPED);
if let Ok(f) = opts.open(address) {
info!("attempting to remove existing named pipe: {}", address);
unsafe { CloseHandle(f.as_raw_handle() as isize) };
#[cfg(windows)]
{
let mut opts = OpenOptions::new();
opts.read(true)
.write(true)
.custom_flags(FILE_FLAG_OVERLAPPED);
if let Ok(f) = opts.open(address) {
info!("attempting to remove existing named pipe: {}", address);
unsafe { CloseHandle(f.as_raw_handle() as isize) };
}
}
Ok(())
}
@ -438,9 +434,10 @@ pub fn spawn(opts: StartOpts, grouping: &str, vars: Vec<(&str, &str)>) -> Result
let address = socket_address(&opts.address, &opts.namespace, grouping);
// Create socket and prepare listener.
// We'll use `add_listener` when creating TTRPC server.
// On Linux, We'll use `add_listener` when creating TTRPC server, on Windows the value isn't used hence the clippy allow
// (see note below about activation process for windows)
#[allow(clippy::let_unit_value)]
let listener = match start_listener(&address) {
let _listener = match start_listener(&address) {
Ok(l) => l,
Err(e) => {
if e.kind() != std::io::ErrorKind::AddrInUse {
@ -449,7 +446,7 @@ pub fn spawn(opts: StartOpts, grouping: &str, vars: Vec<(&str, &str)>) -> Result
err: e,
});
};
// If the Address is already in use then make sure it is up and running and return the address
// If the address is already in use then make sure it is up and running and return the address
// This allows for running a single shim per container scenarios
if let Ok(()) = wait_socket_working(&address, 5, 200) {
write_address(&address)?;
@ -460,108 +457,75 @@ pub fn spawn(opts: StartOpts, grouping: &str, vars: Vec<(&str, &str)>) -> Result
}
};
run_shim(cmd, cwd, opts, vars, listener, address)
}
#[cfg(unix)]
fn run_shim(
cmd: std::path::PathBuf,
cwd: std::path::PathBuf,
opts: StartOpts,
vars: Vec<(&str, &str)>,
listener: UnixListener,
address: String,
) -> Result<(u32, String)> {
let mut command = Command::new(cmd);
command
.current_dir(cwd)
.stdout(Stdio::null())
.stdin(Stdio::null())
.stderr(Stdio::null())
.fd_mappings(vec![FdMapping {
parent_fd: listener.as_raw_fd(),
child_fd: SOCKET_FD,
}])?
.args([
"-namespace",
&opts.namespace,
"-id",
&opts.id,
"-address",
&opts.address,
]);
if opts.debug {
command.arg("-debug");
}
command.envs(vars);
command
.spawn()
.map_err(io_error!(e, "spawn shim"))
.map(|child| {
// Ownership of `listener` has been passed to child.
std::mem::forget(listener);
(child.id(), address)
})
}
// Activation pattern for Windows comes from the hcsshim: https://github.com/microsoft/hcsshim/blob/v0.10.0-rc.7/cmd/containerd-shim-runhcs-v1/serve.go#L57-L70
// another way to do it would to create named pipe and pass it to the child process through handle inheritence but that would require duplicating
// the logic in Rust's 'command' for process creation. There is an issue in Rust to make it simplier to specify handle inheritence and this could
// be revisited once https://github.com/rust-lang/rust/issues/54760 is implemented.
#[cfg(windows)]
fn run_shim(
cmd: std::path::PathBuf,
cwd: std::path::PathBuf,
opts: StartOpts,
vars: Vec<(&str, &str)>,
_listener: (),
address: String,
) -> Result<(u32, String)> {
let mut command = Command::new(cmd);
let (mut reader, writer) = os_pipe::pipe().map_err(io_error!(e, "create pipe"))?;
let stdio_writer = writer.try_clone().unwrap();
command
.current_dir(cwd)
.stdout(stdio_writer)
.stdin(Stdio::null())
.stderr(Stdio::null())
.args([
"-namespace",
&opts.namespace,
"-id",
&opts.id,
"-address",
&opts.address,
]);
command.current_dir(cwd).envs(vars).args([
"-namespace",
&opts.namespace,
"-id",
&opts.id,
"-address",
&opts.address,
]);
if opts.debug {
command.arg("-debug");
}
command.envs(vars);
disable_handle_inheritance();
command
.spawn()
.map_err(io_error!(e, "spawn shim"))
.map(|child| {
// IMPORTANT: we must drop the writer and command to close up handles before we copy the reader to stderr
// AND the shim Start method must NOT write to stdout/stderr
drop(writer);
drop(command);
io::copy(&mut reader, &mut io::stderr()).unwrap();
(child.id(), address)
})
#[cfg(unix)]
{
command
.stdout(Stdio::null())
.stdin(Stdio::null())
.stderr(Stdio::null())
.fd_mappings(vec![FdMapping {
parent_fd: _listener.as_raw_fd(),
child_fd: SOCKET_FD,
}])?;
command
.spawn()
.map_err(io_error!(e, "spawn shim"))
.map(|child| {
// Ownership of `listener` has been passed to child.
std::mem::forget(_listener);
(child.id(), address)
})
}
#[cfg(windows)]
{
// Activation pattern for Windows comes from the hcsshim: https://github.com/microsoft/hcsshim/blob/v0.10.0-rc.7/cmd/containerd-shim-runhcs-v1/serve.go#L57-L70
// another way to do it would to create named pipe and pass it to the child process through handle inheritence but that would require duplicating
// the logic in Rust's 'command' for process creation. There is an issue in Rust to make it simplier to specify handle inheritence and this could
// be revisited once https://github.com/rust-lang/rust/issues/54760 is implemented.
let (mut reader, writer) = os_pipe::pipe().map_err(io_error!(e, "create pipe"))?;
let stdio_writer = writer.try_clone().unwrap();
command
.stdout(stdio_writer)
.stdin(Stdio::null())
.stderr(Stdio::null());
// On Windows Rust currently sets the `HANDLE_FLAG_INHERIT` flag to true when using Command::spawn.
// When a child process is spawned by another process (containerd) the child process inherits the parent's stdin, stdout, and stderr handles.
// Due to the HANDLE_FLAG_INHERIT flag being set to true this will cause containerd to hand until the child process closes the handles.
// As a workaround we can Disables inheritance on the io pipe handles.
// This workaround comes from https://github.com/rust-lang/rust/issues/54760#issuecomment-1045940560
disable_handle_inheritance();
command
.spawn()
.map_err(io_error!(e, "spawn shim"))
.map(|child| {
// IMPORTANT: we must drop the writer and command to close up handles before we copy the reader to stderr
// AND the shim Start method must NOT write to stdout/stderr
drop(writer);
drop(command);
io::copy(&mut reader, &mut io::stderr()).unwrap();
(child.id(), address)
})
}
}
// On Windows Rust currently sets the `HANDLE_FLAG_INHERIT` flag to true when using Command::spawn.
// When a child process is spawned by another process (containerd) the child process inherits the parent's stdin, stdout, and stderr handles.
// Due to the HANDLE_FLAG_INHERIT flag being set to true this will cause containerd to hand until the child process closes the handles.
// As a workaround we can Disables inheritance on the io pipe handles.
// This workaround comes from https://github.com/rust-lang/rust/issues/54760#issuecomment-1045940560
#[cfg(windows)]
fn disable_handle_inheritance() {
use windows_sys::Win32::{

66
crates/shim/src/synchronous/publisher.rs
View File

@ -49,18 +49,20 @@ impl RemotePublisher {
})
}
#[cfg(unix)]
fn connect(address: impl AsRef<str>) -> Result<Client> {
let fd = connect(address)?;
// Client::new() takes ownership of the RawFd.
Ok(Client::new_from_fd(fd)?)
}
#[cfg(unix)]
{
let fd = connect(address)?;
// Client::new() takes ownership of the RawFd.
Ok(Client::new_from_fd(fd)?)
}
#[cfg(windows)]
fn connect(address: impl AsRef<str>) -> Result<Client> {
match Client::connect(address.as_ref()) {
Ok(client) => Ok(client),
Err(e) => Err(e.into()),
#[cfg(windows)]
{
match Client::connect(address.as_ref()) {
Ok(client) => Ok(client),
Err(e) => Err(e.into()),
}
}
}
@ -165,29 +167,31 @@ mod tests {
thread.join().unwrap();
}
#[cfg(unix)]
fn create_server(server_address: &String) -> Server {
use std::os::unix::{io::AsRawFd, net::UnixListener};
let listener = UnixListener::bind(server_address).unwrap();
listener.set_nonblocking(true).unwrap();
let t = Arc::new(Box::new(FakeServer {}) as Box<dyn Events + Send + Sync>);
let service = client::create_events(t);
let server = Server::new()
.add_listener(listener.as_raw_fd())
.unwrap()
.register_service(service);
std::mem::forget(listener);
server
}
#[cfg(unix)]
{
use std::os::unix::{io::AsRawFd, net::UnixListener};
let listener = UnixListener::bind(server_address).unwrap();
listener.set_nonblocking(true).unwrap();
let t = Arc::new(Box::new(FakeServer {}) as Box<dyn Events + Send + Sync>);
let service = client::create_events(t);
let server = Server::new()
.add_listener(listener.as_raw_fd())
.unwrap()
.register_service(service);
std::mem::forget(listener);
server
}
#[cfg(windows)]
fn create_server(server_address: &String) -> Server {
let t = Arc::new(Box::new(FakeServer {}) as Box<dyn Events + Send + Sync>);
let service = client::create_events(t);
#[cfg(windows)]
{
let t = Arc::new(Box::new(FakeServer {}) as Box<dyn Events + Send + Sync>);
let service = client::create_events(t);
Server::new()
.bind(server_address)
.unwrap()
.register_service(service)
Server::new()
.bind(server_address)
.unwrap()
.register_service(service)
}
}
}

2
crates/shim/src/sys/windows/named_pipe_logger.rs
View File

@ -28,7 +28,7 @@ pub struct NamedPipeLogger {
}
impl NamedPipeLogger {
pub fn new(namespace: &String, id: &String) -> Result<NamedPipeLogger, io::Error> {
pub fn new(namespace: &str, id: &str) -> Result<NamedPipeLogger, io::Error> {
let pipe_name = format!("\\\\.\\pipe\\containerd-shim-{}-{}-log", namespace, id);
let mut pipe_server = NamedPipe::new(pipe_name).unwrap();
let mut poll = Poll::new().unwrap();