Compare commits
1 Commits
e0bb98e855
...
798b113ebb
Author | SHA1 | Date | |
---|---|---|---|
798b113ebb |
19
Cargo.lock
generated
19
Cargo.lock
generated
@ -135,9 +135,9 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "async-trait"
|
name = "async-trait"
|
||||||
version = "0.1.84"
|
version = "0.1.83"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "1b1244b10dcd56c92219da4e14caa97e312079e185f04ba3eea25061561dc0a0"
|
checksum = "721cae7de5c34fbb2acd27e21e6d2cf7b886dce0c27388d46c4e6c47ea4318dd"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"proc-macro2",
|
"proc-macro2",
|
||||||
"quote",
|
"quote",
|
||||||
@ -428,7 +428,6 @@ dependencies = [
|
|||||||
"futures",
|
"futures",
|
||||||
"nodrift",
|
"nodrift",
|
||||||
"notmad",
|
"notmad",
|
||||||
"petname",
|
|
||||||
"prost",
|
"prost",
|
||||||
"prost-types",
|
"prost-types",
|
||||||
"reqwest",
|
"reqwest",
|
||||||
@ -1834,20 +1833,6 @@ version = "2.3.1"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
|
checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "petname"
|
|
||||||
version = "2.0.2"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "9cd31dcfdbbd7431a807ef4df6edd6473228e94d5c805e8cf671227a21bad068"
|
|
||||||
dependencies = [
|
|
||||||
"anyhow",
|
|
||||||
"clap",
|
|
||||||
"itertools",
|
|
||||||
"proc-macro2",
|
|
||||||
"quote",
|
|
||||||
"rand",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "pin-project"
|
name = "pin-project"
|
||||||
version = "1.1.7"
|
version = "1.1.7"
|
||||||
|
@ -37,4 +37,3 @@ reqwest = { version = "0.12.9", default-features = false, features = [
|
|||||||
serde_json = "1.0.133"
|
serde_json = "1.0.133"
|
||||||
wasmtime = "28.0.0"
|
wasmtime = "28.0.0"
|
||||||
wasmtime-wasi = "28.0.0"
|
wasmtime-wasi = "28.0.0"
|
||||||
petname = "2.0.2"
|
|
||||||
|
@ -44,7 +44,7 @@ impl State {
|
|||||||
let config = AgentConfig::new().await?;
|
let config = AgentConfig::new().await?;
|
||||||
let discovery = DiscoveryClient::new(&config.discovery).discover().await?;
|
let discovery = DiscoveryClient::new(&config.discovery).discover().await?;
|
||||||
let grpc = GrpcClient::new(&discovery.process_host);
|
let grpc = GrpcClient::new(&discovery.process_host);
|
||||||
let plugin_store = PluginStore::new(config.clone())?;
|
let plugin_store = PluginStore::new()?;
|
||||||
let scheduled_tasks = ScheduledTasks::new(plugin_store.clone());
|
let scheduled_tasks = ScheduledTasks::new(plugin_store.clone());
|
||||||
let scheduler = Scheduler::new(scheduled_tasks);
|
let scheduler = Scheduler::new(scheduled_tasks);
|
||||||
let queue = AgentQueue::new(scheduler);
|
let queue = AgentQueue::new(scheduler);
|
||||||
|
@ -8,8 +8,6 @@ use uuid::Uuid;
|
|||||||
pub struct AgentConfig {
|
pub struct AgentConfig {
|
||||||
pub agent_id: String,
|
pub agent_id: String,
|
||||||
pub discovery: String,
|
pub discovery: String,
|
||||||
|
|
||||||
pub labels: BTreeMap<String, String>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AgentConfig {
|
impl AgentConfig {
|
||||||
@ -19,7 +17,6 @@ impl AgentConfig {
|
|||||||
Ok(Self {
|
Ok(Self {
|
||||||
agent_id: config.agent_id,
|
agent_id: config.agent_id,
|
||||||
discovery: config.discovery,
|
discovery: config.discovery,
|
||||||
labels: config.labels.unwrap_or_default(),
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -8,9 +8,6 @@ use wasmtime::component::*;
|
|||||||
use wasmtime::{Config, Engine, Store};
|
use wasmtime::{Config, Engine, Store};
|
||||||
use wasmtime_wasi::{DirPerms, FilePerms, WasiCtx, WasiCtxBuilder, WasiView};
|
use wasmtime_wasi::{DirPerms, FilePerms, WasiCtx, WasiCtxBuilder, WasiView};
|
||||||
|
|
||||||
use super::agent_state::State;
|
|
||||||
use super::config::AgentConfig;
|
|
||||||
|
|
||||||
wasmtime::component::bindgen!({
|
wasmtime::component::bindgen!({
|
||||||
path: "wit/world.wit",
|
path: "wit/world.wit",
|
||||||
//world: "churn",
|
//world: "churn",
|
||||||
@ -20,14 +17,9 @@ wasmtime::component::bindgen!({
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
pub struct CustomProcess {
|
#[derive(Default)]
|
||||||
agent_config: AgentConfig,
|
pub struct CustomProcess {}
|
||||||
}
|
|
||||||
impl CustomProcess {
|
impl CustomProcess {
|
||||||
pub fn new(agent_config: AgentConfig) -> Self {
|
|
||||||
Self { agent_config }
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn run(&self, args: Vec<String>) -> String {
|
pub fn run(&self, args: Vec<String>) -> String {
|
||||||
tracing::info!("calling function");
|
tracing::info!("calling function");
|
||||||
|
|
||||||
@ -50,10 +42,6 @@ impl CustomProcess {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn get_label(&self, label_key: &str) -> Option<String> {
|
|
||||||
self.agent_config.labels.get(label_key).cloned()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
@ -62,9 +50,9 @@ pub struct PluginStore {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl PluginStore {
|
impl PluginStore {
|
||||||
pub fn new(config: AgentConfig) -> anyhow::Result<Self> {
|
pub fn new() -> anyhow::Result<Self> {
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
inner: Arc::new(Mutex::new(InnerPluginStore::new(config)?)),
|
inner: Arc::new(Mutex::new(InnerPluginStore::new()?)),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -86,7 +74,7 @@ pub struct InnerPluginStore {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl InnerPluginStore {
|
impl InnerPluginStore {
|
||||||
pub fn new(agent_config: AgentConfig) -> anyhow::Result<Self> {
|
pub fn new() -> anyhow::Result<Self> {
|
||||||
let mut config = Config::default();
|
let mut config = Config::default();
|
||||||
config.wasm_component_model(true);
|
config.wasm_component_model(true);
|
||||||
config.async_support(true);
|
config.async_support(true);
|
||||||
@ -101,7 +89,7 @@ impl InnerPluginStore {
|
|||||||
|state: &mut ServerWasiView| state,
|
|state: &mut ServerWasiView| state,
|
||||||
)?;
|
)?;
|
||||||
|
|
||||||
let wasi_view = ServerWasiView::new(agent_config);
|
let wasi_view = ServerWasiView::new();
|
||||||
let store = Store::new(&engine, wasi_view);
|
let store = Store::new(&engine, wasi_view);
|
||||||
|
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
@ -124,21 +112,11 @@ impl InnerPluginStore {
|
|||||||
pub async fn execute(&mut self, plugin: &str) -> anyhow::Result<()> {
|
pub async fn execute(&mut self, plugin: &str) -> anyhow::Result<()> {
|
||||||
let plugin = self.ensure_plugin(plugin).await?;
|
let plugin = self.ensure_plugin(plugin).await?;
|
||||||
|
|
||||||
if plugin
|
plugin
|
||||||
.interface0
|
.interface0
|
||||||
.call_should_run(&mut self.store)
|
.call_execute(&mut self.store)
|
||||||
.await
|
.await
|
||||||
.context("Failed to call should run")?
|
.context("Failed to call add function")
|
||||||
{
|
|
||||||
tracing::info!("job was marked as required to run");
|
|
||||||
return plugin
|
|
||||||
.interface0
|
|
||||||
.call_execute(&mut self.store)
|
|
||||||
.await
|
|
||||||
.context("Failed to call add function");
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok(())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn ensure_plugin(&mut self, plugin: &str) -> anyhow::Result<Churn> {
|
async fn ensure_plugin(&mut self, plugin: &str) -> anyhow::Result<Churn> {
|
||||||
@ -170,12 +148,6 @@ impl InnerPluginStore {
|
|||||||
|
|
||||||
let req = reqwest::get(format!("https://api-minio.front.kjuulh.io/churn-registry/{plugin_name}/{plugin_version}/{plugin_name}.wasm")).await.context("failed to get plugin from registry")?;
|
let req = reqwest::get(format!("https://api-minio.front.kjuulh.io/churn-registry/{plugin_name}/{plugin_version}/{plugin_name}.wasm")).await.context("failed to get plugin from registry")?;
|
||||||
let mut stream = req.bytes_stream();
|
let mut stream = req.bytes_stream();
|
||||||
|
|
||||||
tracing::info!(
|
|
||||||
plugin_name = plugin_name,
|
|
||||||
plugin_path = plugin_path.display().to_string(),
|
|
||||||
"writing plugin to file"
|
|
||||||
);
|
|
||||||
let mut file = tokio::fs::File::create(&plugin_path).await?;
|
let mut file = tokio::fs::File::create(&plugin_path).await?;
|
||||||
while let Some(chunk) = stream.next().await {
|
while let Some(chunk) = stream.next().await {
|
||||||
let chunk = chunk?;
|
let chunk = chunk?;
|
||||||
@ -205,17 +177,14 @@ struct ServerWasiView {
|
|||||||
table: ResourceTable,
|
table: ResourceTable,
|
||||||
ctx: WasiCtx,
|
ctx: WasiCtx,
|
||||||
processes: ResourceTable,
|
processes: ResourceTable,
|
||||||
agent_config: AgentConfig,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ServerWasiView {
|
impl ServerWasiView {
|
||||||
fn new(agent_config: AgentConfig) -> Self {
|
fn new() -> Self {
|
||||||
let table = ResourceTable::new();
|
let table = ResourceTable::new();
|
||||||
|
|
||||||
let ctx = WasiCtxBuilder::new()
|
let ctx = WasiCtxBuilder::new()
|
||||||
.inherit_stdio()
|
.inherit_stdio()
|
||||||
.inherit_stdout()
|
|
||||||
.inherit_stderr()
|
|
||||||
.inherit_env()
|
.inherit_env()
|
||||||
.inherit_network()
|
.inherit_network()
|
||||||
.preopened_dir("/", "/", DirPerms::all(), FilePerms::all())
|
.preopened_dir("/", "/", DirPerms::all(), FilePerms::all())
|
||||||
@ -226,7 +195,6 @@ impl ServerWasiView {
|
|||||||
table,
|
table,
|
||||||
ctx,
|
ctx,
|
||||||
processes: ResourceTable::default(),
|
processes: ResourceTable::default(),
|
||||||
agent_config,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -248,9 +216,7 @@ impl HostProcess for ServerWasiView {
|
|||||||
async fn new(
|
async fn new(
|
||||||
&mut self,
|
&mut self,
|
||||||
) -> wasmtime::component::Resource<component::churn_tasks::process::Process> {
|
) -> wasmtime::component::Resource<component::churn_tasks::process::Process> {
|
||||||
self.processes
|
self.processes.push(CustomProcess::default()).unwrap()
|
||||||
.push(CustomProcess::new(self.agent_config.clone()))
|
|
||||||
.unwrap()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn run_process(
|
async fn run_process(
|
||||||
@ -262,15 +228,6 @@ impl HostProcess for ServerWasiView {
|
|||||||
process.run(inputs)
|
process.run(inputs)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn get_variable(
|
|
||||||
&mut self,
|
|
||||||
self_: wasmtime::component::Resource<component::churn_tasks::process::Process>,
|
|
||||||
key: wasmtime::component::__internal::String,
|
|
||||||
) -> String {
|
|
||||||
let process = self.processes.get(&self_).unwrap();
|
|
||||||
process.get_label(&key).unwrap()
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn drop(
|
async fn drop(
|
||||||
&mut self,
|
&mut self,
|
||||||
rep: wasmtime::component::Resource<component::churn_tasks::process::Process>,
|
rep: wasmtime::component::Resource<component::churn_tasks::process::Process>,
|
||||||
|
@ -31,13 +31,6 @@ pub async fn execute() -> anyhow::Result<()> {
|
|||||||
setup_labels.insert(k, v);
|
setup_labels.insert(k, v);
|
||||||
}
|
}
|
||||||
|
|
||||||
if !setup_labels.contains_key("node_name") {
|
|
||||||
setup_labels.insert(
|
|
||||||
"node_name".into(),
|
|
||||||
petname::petname(2, "-").expect("to be able to generate a valid petname"),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
agent::setup_config(discovery, force, setup_labels).await?;
|
agent::setup_config(discovery, force, setup_labels).await?;
|
||||||
tracing::info!("wrote default agent config");
|
tracing::info!("wrote default agent config");
|
||||||
}
|
}
|
||||||
|
@ -4,7 +4,6 @@ interface process {
|
|||||||
resource process {
|
resource process {
|
||||||
constructor();
|
constructor();
|
||||||
run-process: func(inputs: list<string>) -> string;
|
run-process: func(inputs: list<string>) -> string;
|
||||||
get-variable: func(key: string) -> string;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user