This repository has been archived by the owner on Feb 21, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Matthew Fisher <[email protected]>
- Loading branch information
Matthew Fisher
committed
May 21, 2020
1 parent
86305c2
commit 48f3cc1
Showing
3 changed files
with
122 additions
and
102 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,104 +1,115 @@ | ||
| use std::error; | ||
| use std::fmt; | ||
| use std::path::Path; | ||
| use std::sync::Arc; | ||
|
|
||
| use wasm3::Environment; | ||
| use tempfile::NamedTempFile; | ||
| use tokio::sync::watch::{self, Sender}; | ||
| use tokio::task::JoinHandle; | ||
| use wasm3::{Environment, Module}; | ||
| use kubelet::handle::{RuntimeHandle, Stop}; | ||
| use kubelet::status::ContainerStatus; | ||
|
|
||
| type Result<T> = std::result::Result<T, RuntimeError>; | ||
|
|
||
| #[derive(Debug, Clone, PartialEq, Eq)] | ||
| pub struct RuntimeError { | ||
| kind: RuntimeErrorKind, | ||
| } | ||
|
|
||
| #[derive(Debug, Clone, PartialEq, Eq)] | ||
| enum RuntimeErrorKind { | ||
| AlreadyStarted, | ||
| CannotCreateRuntime, | ||
| CannotLinkWASI, | ||
| CannotLoadModule, | ||
| NoEntrypoint, | ||
| RunFailure, | ||
| } | ||
|
|
||
| #[derive(Debug, Clone, PartialEq, Eq)] | ||
| enum RuntimeStatus { | ||
| Running, | ||
| Stopped, | ||
| } | ||
|
|
||
| impl RuntimeError { | ||
| fn new(kind: RuntimeErrorKind) -> Self { | ||
| Self { kind: kind } | ||
| } | ||
|
|
||
| fn __description(&self) -> &str { | ||
| match self.kind { | ||
| RuntimeErrorKind::AlreadyStarted => "runtime already started", | ||
| RuntimeErrorKind::CannotCreateRuntime => "cannot create runtime", | ||
| RuntimeErrorKind::CannotLinkWASI => "cannot link module to the WASI runtime", | ||
| RuntimeErrorKind::CannotLoadModule => "cannot load module", | ||
| RuntimeErrorKind::NoEntrypoint => "no entrypoint function called '_start' found", | ||
| RuntimeErrorKind::RunFailure => "failure during function call", | ||
| } | ||
| } | ||
| pub struct HandleStopper { | ||
| handle: JoinHandle<anyhow::Result<()>>, | ||
| } | ||
|
|
||
| impl fmt::Display for RuntimeError { | ||
| fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { | ||
| write!(f, "{}", self.__description()) | ||
| #[async_trait::async_trait] | ||
| impl Stop for HandleStopper { | ||
| async fn stop(&mut self) -> anyhow::Result<()> { | ||
| // no nothing | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| impl error::Error for RuntimeError { | ||
| fn source(&self) -> Option<&(dyn error::Error + 'static)> { | ||
| // source is not tracked | ||
| None | ||
| async fn wait(&mut self) -> anyhow::Result<()> { | ||
| (&mut self.handle).await??; | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| /// A runtime context for running a wasm module with wasm3 | ||
| pub struct Runtime { | ||
| module_bytes: Vec<u8>, | ||
| stack_size: u32, | ||
| current_status: RuntimeStatus, | ||
| output: Arc<NamedTempFile>, | ||
| } | ||
|
|
||
| impl Runtime { | ||
| pub fn new(module_bytes: Vec<u8>, stack_size: u32) -> Self { | ||
| Self { | ||
| pub async fn new<L: AsRef<Path> + Send + Sync + 'static>(module_bytes: Vec<u8>, stack_size: u32, log_dir: L) -> anyhow::Result<Self> { | ||
| let temp = tokio::task::spawn_blocking(move || -> anyhow::Result<NamedTempFile> { | ||
| Ok(NamedTempFile::new_in(log_dir)?) | ||
| }) | ||
| .await??; | ||
|
|
||
| Ok(Self { | ||
| module_bytes: module_bytes, | ||
| stack_size: stack_size, | ||
| current_status: RuntimeStatus::Stopped, | ||
| } | ||
| output: Arc::new(temp), | ||
| }) | ||
| } | ||
|
|
||
| pub fn start(&mut self) -> Result<()> { | ||
| if self.current_status == RuntimeStatus::Running { | ||
| return Err(RuntimeError::new(RuntimeErrorKind::AlreadyStarted)); | ||
| } | ||
| let env = Environment::new() | ||
| .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotCreateRuntime))?; | ||
| let rt = env | ||
| .create_runtime(self.stack_size) | ||
| .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotCreateRuntime))?; | ||
| let mut module = rt | ||
| .parse_and_load_module(&self.module_bytes) | ||
| .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotLoadModule))?; | ||
| module.link_wasi().map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotLinkWASI))?; | ||
| let func = module | ||
| .find_function::<(), ()>("_start") | ||
| .map_err(|_| RuntimeError::new(RuntimeErrorKind::NoEntrypoint))?; | ||
| self.current_status = RuntimeStatus::Running; | ||
| // FIXME: run this in the background | ||
| // for now, we block until the function is complete, then call .stop() | ||
| func.call() | ||
| .map_err(|_| RuntimeError::new(RuntimeErrorKind::RunFailure))?; | ||
| self.stop() | ||
| pub async fn start(&mut self) -> anyhow::Result<RuntimeHandle<HandleStopper, LogHandleFactory>> { | ||
| let temp = self.output.clone(); | ||
| let output_write = tokio::task::spawn_blocking(move || -> anyhow::Result<std::fs::File> { | ||
| Ok(temp.reopen()?) | ||
| }) | ||
| .await??; | ||
|
|
||
| let (status_sender, status_recv) = watch::channel(ContainerStatus::Waiting { | ||
| timestamp: chrono::Utc::now(), | ||
| message: "No status has been received from the process".into(), | ||
| }); | ||
| let handle = spawn_wasm3(self.module_bytes.clone(), self.stack_size, status_sender, output_write).await?; | ||
|
|
||
|
|
||
| let log_handle_factory = LogHandleFactory { | ||
| temp: self.output.clone(), | ||
| }; | ||
|
|
||
| Ok(RuntimeHandle::new( | ||
| HandleStopper{handle}, | ||
| log_handle_factory, | ||
| status_recv, | ||
| )) | ||
| } | ||
| } | ||
|
|
||
| pub fn stop(&mut self) -> Result<()> { | ||
| // it is OK for the runtime to stop an already stopped module. Effectively a no-op | ||
| self.current_status = RuntimeStatus::Stopped; | ||
| Ok(()) | ||
| /// Holds our tempfile handle. | ||
| pub struct LogHandleFactory { | ||
| temp: Arc<NamedTempFile>, | ||
| } | ||
|
|
||
| impl kubelet::handle::LogHandleFactory<tokio::fs::File> for LogHandleFactory { | ||
| /// Creates `tokio::fs::File` on demand for log reading. | ||
| fn new_handle(&self) -> tokio::fs::File { | ||
| tokio::fs::File::from_std(self.temp.reopen().unwrap()) | ||
| } | ||
| } | ||
|
|
||
| // Spawns a running wasmtime instance with the given context and status | ||
| // channel. Due to the Instance type not being Send safe, all of the logic | ||
| // needs to be done within the spawned task | ||
| async fn spawn_wasm3( | ||
| module_bytes: Vec<u8>, | ||
| stack_size: u32, | ||
| status_sender: Sender<ContainerStatus>, | ||
| _output_write: std::fs::File, //TODO: hook this up such that log output will be written to the file | ||
| ) -> anyhow::Result<JoinHandle<anyhow::Result<()>>> { | ||
| let handle = tokio::task::spawn_blocking(move || -> anyhow::Result<_> { | ||
| let env = Environment::new().expect("cannot create environment"); | ||
| let rt = env.create_runtime(stack_size).expect("cannot create runtime"); | ||
| let module = Module::parse(&env, &module_bytes).expect("cannot parse module"); | ||
| let mut module = rt.load_module(module).expect("cannot load module"); | ||
| module.link_wasi().expect("cannot link WASI"); | ||
| let func = module.find_function::<(), ()>("_start").expect("cannot find function '_start' in module"); | ||
| func.call().expect("cannot call '_start' in module"); | ||
| status_sender | ||
| .broadcast(ContainerStatus::Terminated { | ||
| failed: false, | ||
| message: "Module run completed".into(), | ||
| timestamp: chrono::Utc::now(), | ||
| }) | ||
| .expect("status should be able to send"); | ||
| Ok(()) | ||
| }); | ||
|
|
||
| Ok(handle) | ||
| } |