|
1 | | -use std::error; |
2 | | -use std::fmt; |
| 1 | +use std::path::Path; |
| 2 | +use std::sync::Arc; |
3 | 3 |
|
4 | | -use wasm3::Environment; |
| 4 | +use tempfile::NamedTempFile; |
| 5 | +use tokio::sync::watch::{self, Sender}; |
| 6 | +use tokio::task::JoinHandle; |
| 7 | +use wasm3::{Environment, Module}; |
| 8 | +use kubelet::handle::{RuntimeHandle, Stop}; |
| 9 | +use kubelet::status::ContainerStatus; |
5 | 10 |
|
6 | | -type Result<T> = std::result::Result<T, RuntimeError>; |
7 | | - |
8 | | -#[derive(Debug, Clone, PartialEq, Eq)] |
9 | | -pub struct RuntimeError { |
10 | | - kind: RuntimeErrorKind, |
11 | | -} |
12 | | - |
13 | | -#[derive(Debug, Clone, PartialEq, Eq)] |
14 | | -enum RuntimeErrorKind { |
15 | | - AlreadyStarted, |
16 | | - CannotCreateRuntime, |
17 | | - CannotLinkWASI, |
18 | | - CannotLoadModule, |
19 | | - NoEntrypoint, |
20 | | - RunFailure, |
21 | | -} |
22 | | - |
23 | | -#[derive(Debug, Clone, PartialEq, Eq)] |
24 | | -enum RuntimeStatus { |
25 | | - Running, |
26 | | - Stopped, |
27 | | -} |
28 | | - |
29 | | -impl RuntimeError { |
30 | | - fn new(kind: RuntimeErrorKind) -> Self { |
31 | | - Self { kind: kind } |
32 | | - } |
33 | | - |
34 | | - fn __description(&self) -> &str { |
35 | | - match self.kind { |
36 | | - RuntimeErrorKind::AlreadyStarted => "runtime already started", |
37 | | - RuntimeErrorKind::CannotCreateRuntime => "cannot create runtime", |
38 | | - RuntimeErrorKind::CannotLinkWASI => "cannot link module to the WASI runtime", |
39 | | - RuntimeErrorKind::CannotLoadModule => "cannot load module", |
40 | | - RuntimeErrorKind::NoEntrypoint => "no entrypoint function called '_start' found", |
41 | | - RuntimeErrorKind::RunFailure => "failure during function call", |
42 | | - } |
43 | | - } |
| 11 | +pub struct HandleStopper { |
| 12 | + handle: JoinHandle<anyhow::Result<()>>, |
44 | 13 | } |
45 | 14 |
|
46 | | -impl fmt::Display for RuntimeError { |
47 | | - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { |
48 | | - write!(f, "{}", self.__description()) |
| 15 | +#[async_trait::async_trait] |
| 16 | +impl Stop for HandleStopper { |
| 17 | + async fn stop(&mut self) -> anyhow::Result<()> { |
| 18 | + // no nothing |
| 19 | + Ok(()) |
49 | 20 | } |
50 | | -} |
51 | 21 |
|
52 | | -impl error::Error for RuntimeError { |
53 | | - fn source(&self) -> Option<&(dyn error::Error + 'static)> { |
54 | | - // source is not tracked |
55 | | - None |
| 22 | + async fn wait(&mut self) -> anyhow::Result<()> { |
| 23 | + (&mut self.handle).await??; |
| 24 | + Ok(()) |
56 | 25 | } |
57 | 26 | } |
58 | 27 |
|
59 | 28 | /// A runtime context for running a wasm module with wasm3 |
60 | 29 | pub struct Runtime { |
61 | 30 | module_bytes: Vec<u8>, |
62 | 31 | stack_size: u32, |
63 | | - current_status: RuntimeStatus, |
| 32 | + output: Arc<NamedTempFile>, |
64 | 33 | } |
65 | 34 |
|
66 | 35 | impl Runtime { |
67 | | - pub fn new(module_bytes: Vec<u8>, stack_size: u32) -> Self { |
68 | | - Self { |
| 36 | + pub async fn new<L: AsRef<Path> + Send + Sync + 'static>(module_bytes: Vec<u8>, stack_size: u32, log_dir: L) -> anyhow::Result<Self> { |
| 37 | + let temp = tokio::task::spawn_blocking(move || -> anyhow::Result<NamedTempFile> { |
| 38 | + Ok(NamedTempFile::new_in(log_dir)?) |
| 39 | + }) |
| 40 | + .await??; |
| 41 | + |
| 42 | + Ok(Self { |
69 | 43 | module_bytes: module_bytes, |
70 | 44 | stack_size: stack_size, |
71 | | - current_status: RuntimeStatus::Stopped, |
72 | | - } |
| 45 | + output: Arc::new(temp), |
| 46 | + }) |
73 | 47 | } |
74 | 48 |
|
75 | | - pub fn start(&mut self) -> Result<()> { |
76 | | - if self.current_status == RuntimeStatus::Running { |
77 | | - return Err(RuntimeError::new(RuntimeErrorKind::AlreadyStarted)); |
78 | | - } |
79 | | - let env = Environment::new() |
80 | | - .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotCreateRuntime))?; |
81 | | - let rt = env |
82 | | - .create_runtime(self.stack_size) |
83 | | - .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotCreateRuntime))?; |
84 | | - let mut module = rt |
85 | | - .parse_and_load_module(&self.module_bytes) |
86 | | - .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotLoadModule))?; |
87 | | - module.link_wasi().map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotLinkWASI))?; |
88 | | - let func = module |
89 | | - .find_function::<(), ()>("_start") |
90 | | - .map_err(|_| RuntimeError::new(RuntimeErrorKind::NoEntrypoint))?; |
91 | | - self.current_status = RuntimeStatus::Running; |
92 | | - // FIXME: run this in the background |
93 | | - // for now, we block until the function is complete, then call .stop() |
94 | | - func.call() |
95 | | - .map_err(|_| RuntimeError::new(RuntimeErrorKind::RunFailure))?; |
96 | | - self.stop() |
| 49 | + pub async fn start(&mut self) -> anyhow::Result<RuntimeHandle<HandleStopper, LogHandleFactory>> { |
| 50 | + let temp = self.output.clone(); |
| 51 | + let output_write = tokio::task::spawn_blocking(move || -> anyhow::Result<std::fs::File> { |
| 52 | + Ok(temp.reopen()?) |
| 53 | + }) |
| 54 | + .await??; |
| 55 | + |
| 56 | + let (status_sender, status_recv) = watch::channel(ContainerStatus::Waiting { |
| 57 | + timestamp: chrono::Utc::now(), |
| 58 | + message: "No status has been received from the process".into(), |
| 59 | + }); |
| 60 | + let handle = spawn_wasm3(self.module_bytes.clone(), self.stack_size, status_sender, output_write).await?; |
| 61 | + |
| 62 | + |
| 63 | + let log_handle_factory = LogHandleFactory { |
| 64 | + temp: self.output.clone(), |
| 65 | + }; |
| 66 | + |
| 67 | + Ok(RuntimeHandle::new( |
| 68 | + HandleStopper{handle}, |
| 69 | + log_handle_factory, |
| 70 | + status_recv, |
| 71 | + )) |
97 | 72 | } |
| 73 | +} |
98 | 74 |
|
99 | | - pub fn stop(&mut self) -> Result<()> { |
100 | | - // it is OK for the runtime to stop an already stopped module. Effectively a no-op |
101 | | - self.current_status = RuntimeStatus::Stopped; |
102 | | - Ok(()) |
| 75 | +/// Holds our tempfile handle. |
| 76 | +pub struct LogHandleFactory { |
| 77 | + temp: Arc<NamedTempFile>, |
| 78 | +} |
| 79 | + |
| 80 | +impl kubelet::handle::LogHandleFactory<tokio::fs::File> for LogHandleFactory { |
| 81 | + /// Creates `tokio::fs::File` on demand for log reading. |
| 82 | + fn new_handle(&self) -> tokio::fs::File { |
| 83 | + tokio::fs::File::from_std(self.temp.reopen().unwrap()) |
103 | 84 | } |
104 | 85 | } |
| 86 | + |
| 87 | +// Spawns a running wasmtime instance with the given context and status |
| 88 | +// channel. Due to the Instance type not being Send safe, all of the logic |
| 89 | +// needs to be done within the spawned task |
| 90 | +async fn spawn_wasm3( |
| 91 | + module_bytes: Vec<u8>, |
| 92 | + stack_size: u32, |
| 93 | + status_sender: Sender<ContainerStatus>, |
| 94 | + _output_write: std::fs::File, //TODO: hook this up such that log output will be written to the file |
| 95 | +) -> anyhow::Result<JoinHandle<anyhow::Result<()>>> { |
| 96 | + let handle = tokio::task::spawn_blocking(move || -> anyhow::Result<_> { |
| 97 | + let env = Environment::new().expect("cannot create environment"); |
| 98 | + let rt = env.create_runtime(stack_size).expect("cannot create runtime"); |
| 99 | + let module = Module::parse(&env, &module_bytes).expect("cannot parse module"); |
| 100 | + let mut module = rt.load_module(module).expect("cannot load module"); |
| 101 | + module.link_wasi().expect("cannot link WASI"); |
| 102 | + let func = module.find_function::<(), ()>("_start").expect("cannot find function '_start' in module"); |
| 103 | + func.call().expect("cannot call '_start' in module"); |
| 104 | + status_sender |
| 105 | + .broadcast(ContainerStatus::Terminated { |
| 106 | + failed: false, |
| 107 | + message: "Module run completed".into(), |
| 108 | + timestamp: chrono::Utc::now(), |
| 109 | + }) |
| 110 | + .expect("status should be able to send"); |
| 111 | + Ok(()) |
| 112 | + }); |
| 113 | + |
| 114 | + Ok(handle) |
| 115 | +} |
0 commit comments