Adds bacalhau support

.gitignore

@@ -12,3 +12,5 @@ Cargo.lock
 
 # MSVC Windows builds of rustc generate these, which store debugging information
 *.pdb
+
+.vscode/

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "pipe/runtime",
     "pipe/arrow_msg",
     "pipe/section",
+    "pipe/section/section_impls/bacalhau",
     "pipe/section/section_impls/hello_world",
     "pipe/section/section_impls/sqlite_connector",
     "pipe/section/section_impls/excel_connector",

common/src/lib.rs

@@ -65,6 +65,7 @@ pub enum Destination {
     Sqlite_Connector(SqliteDestinationConfig),
     Snowflake(SnowflakeDestinationConfig),
     Sqlite_Physical_Replication(SqlitePhysicalReplicationDestinationConfig),
+    Bacalhau(BacalhauDestinationConfig),
     Hello_World(HelloWorldDestinationConfig),
     Kafka(KafkaDestinationConfig),
     Postgres_Connector(PostgresConnectorDestinationConfig),
@@ -160,6 +161,14 @@ pub struct SqlitePhysicalReplicationSourceConfig {
     // database path
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
+pub struct BacalhauDestinationConfig {
+    #[serde(flatten)]
+    pub common_attrs: CommonAttrs,
+    pub job: String,
+    pub jobstore: String,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 pub struct HelloWorldSourceConfig {
     #[serde(flatten)]

myceliald/Cargo.toml

@@ -11,16 +11,18 @@ clap = { version = "4", features = ["derive", "env"] }
 reqwest = { version = "0.11", features = ["json"] }
 anyhow = "1"
 base64 = { version = "0.21" }
-serde = { version = "1", features = ["derive"]}
+serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-sqlx = { version = "0.7", features = ["sqlite"]}
+sqlx = { version = "0.7", features = ["sqlite"] }
 log = "0.4"
 toml = "0.7"
 common = { path = "../common" }
-pipe = { version = "0.3", path = "../pipe/runtime/", package="runtime" }
+thiserror = "1"
+pipe = { version = "0.3", path = "../pipe/runtime/", package = "runtime" }
 section = { version = "0.3", path = "../pipe/section/" }
 
 ## sections
+bacalhau = { path = "../pipe/section/section_impls/bacalhau/" }
 hello_world = { path = "../pipe/section/section_impls/hello_world/" }
 sqlite_connector = { path = "../pipe/section/section_impls/sqlite_connector/" }
 excel_connector = { path = "../pipe/section/section_impls/excel_connector/" }

myceliald/config.example.toml

@@ -88,6 +88,13 @@ path = "Cargo.toml"
 # Define all data DESTINATIONS (data stores and directory paths)
 # accessible by the Node that should be exposed as DESTINATIONS to the Sqlite Physical Replication Server
 
+
+[[destinations]]
+display_name = "Bacalhau"
+jobstore = "c:/temp/jobstore"
+type = "bacalhau"
+job = "jobname"
+
 [[destinations]]
 type = "sqlite_physical_replication"
 display_name = "Sqlite Physical Replication Movie"

myceliald/src/constructors/bacalhau.rs

@@ -0,0 +1,28 @@
+use std::path::Path;
+
+use pipe::{config::Map, types::DynSection};
+use section::{command_channel::SectionChannel, SectionError};
+
+pub fn destination_ctor<S: SectionChannel>(
+    config: &Map,
+) -> Result<Box<dyn DynSection<S>>, SectionError> {
+    let job = config
+        .get("job")
+        .ok_or("bacalhau section requires 'job'")?
+        .as_str()
+        .ok_or("'job' should be a string")?;
+    let jobstore = config
+        .get("jobstore")
+        .ok_or("bacalhau section requires 'jobstore'")?
+        .as_str()
+        .ok_or("'jobstore' should be a string")?;
+
+    // Verify jobstore folder exists
+    if !Path::new(jobstore).exists() {
+        return Err(format!("'jobstore' path '{}' does not exist", jobstore).into());
+    }
+
+    Ok(Box::new(bacalhau::destination::Bacalhau::new(
+        job, jobstore,
+    )))
+}

myceliald/src/constructors/mod.rs

@@ -1,3 +1,4 @@
+pub mod bacalhau;
 pub mod excel_connector;
 pub mod file;
 pub mod hello_world;

myceliald/src/runtime.rs

@@ -15,6 +15,10 @@ use section::command_channel::SectionChannel;
 /// Setup & populate registry
 fn setup_registry<S: SectionChannel>() -> Registry<S> {
     let arr: &[(&str, Constructor<S>)] = &[
+        (
+            "bacalhau_destination",
+            constructors::bacalhau::destination_ctor,
+        ),
         (
             "hello_world_destination",
             constructors::hello_world::destination_ctor,

pipe/section/section_impls/bacalhau/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "bacalhau"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+arrow = { version = "50", features = ["prettyprint"] }
+arrow-json = "50"
+arrow_msg = { path = "../../../arrow_msg" }
+section = { version = "0.3", path = "../../../section/" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+reqwest = { version = "0.11.22", features = ["json"] }
+tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1.14"
+tokio-util = "0.7.10"
+handlebars = "4.4"
+serde_yaml = "0.9"
+
+
+[dev-dependencies]
+stub = { path = "../../section_impls/stub/" }

pipe/section/section_impls/bacalhau/src/api.rs

@@ -0,0 +1,49 @@
+use crate::StdError;
+use reqwest;
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+struct SubmitErrorResponse {
+    error: String,
+    #[allow(dead_code)]
+    message: String,
+}
+
+#[derive(Deserialize)]
+struct SubmitResponse {
+    #[serde(rename = "JobID")]
+    job_id: String,
+
+    #[serde(rename = "EvaluationID")]
+    _evaluation_id: String,
+
+    #[serde(rename = "Warnings")]
+    _warnings: Option<Vec<String>>,
+}
+
+pub async fn submit(body: &str) -> Result<String, StdError> {
+    // For now we'll assume a local instance and using the newer version
+    // of the jobs API.
+    let api_url = "http://127.0.0.1:20000/api/v1/orchestrator/jobs";
+    let client = reqwest::Client::new();
+    let response = client
+        .post(api_url)
+        .header("Content-Type", "application/json")
+        .body(body.to_owned())
+        .send()
+        .await?;
+
+    let success = &response.status().is_success();
+    let resp_body = &response.bytes().await?;
+
+    match success {
+        true => {
+            let sresp: SubmitResponse = serde_json::from_slice(resp_body.as_ref()).unwrap();
+            Ok(sresp.job_id)
+        }
+        _ => {
+            let err: SubmitErrorResponse = serde_json::from_slice(resp_body.as_ref()).unwrap();
+            Err(err.error.into())
+        }
+    }
+}