Merge pull request #5 from sparkzky/feat-provider

Feat provider

Cargo.lock

@@ -432,9 +432,9 @@ checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
 [[package]]
 name = "containerd-client"
-version = "0.6.0"
+version = "0.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce8bbfa492159a878a45bbd8fbd8d5a6fb2a4d6bd74836204a324523ba3233e0"
+checksum = "936cc6911381d95039f381080ec148024d2bb0044a03363542b0a6b0b2c46592"
 dependencies = [
  "hyper-util",
  "prost",
@@ -443,7 +443,7 @@ dependencies = [
  "tokio",
  "tonic",
  "tonic-build",
- "tower 0.4.13",
+ "tower 0.5.2",
 ]
 
 [[package]]
@@ -809,6 +809,20 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "handlebars"
+version = "4.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "faa67bab9ff362228eb3d00bd024a4965d8231bbb7921167f0cfa66c6626b225"
+dependencies = [
+ "log",
+ "pest",
+ "pest_derive",
+ "serde",
+ "serde_json",
+ "thiserror 1.0.69",
+]
+
 [[package]]
 name = "hashbrown"
 version = "0.12.3"
@@ -1312,7 +1326,7 @@ dependencies = [
  "serde_json",
  "strum",
  "strum_macros",
- "thiserror",
+ "thiserror 1.0.69",
 ]
 
 [[package]]
@@ -1356,6 +1370,51 @@ version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
+[[package]]
+name = "pest"
+version = "2.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b7cafe60d6cf8e62e1b9b2ea516a089c008945bb5a275416789e7db0bc199dc"
+dependencies = [
+ "memchr",
+ "thiserror 2.0.11",
+ "ucd-trie",
+]
+
+[[package]]
+name = "pest_derive"
+version = "2.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "816518421cfc6887a0d62bf441b6ffb4536fcc926395a69e1a85852d4363f57e"
+dependencies = [
+ "pest",
+ "pest_generator",
+]
+
+[[package]]
+name = "pest_generator"
+version = "2.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d1396fd3a870fc7838768d171b4616d5c91f6cc25e377b673d714567d99377b"
+dependencies = [
+ "pest",
+ "pest_meta",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "pest_meta"
+version = "2.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e1e58089ea25d717bfd31fb534e4f3afcc2cc569c70de3e239778991ea3b7dea"
+dependencies = [
+ "once_cell",
+ "pest",
+ "sha2",
+]
+
 [[package]]
 name = "petgraph"
 version = "0.6.5"
@@ -1512,6 +1571,10 @@ dependencies = [
  "prost",
 ]
 
+[[package]]
+name = "provider"
+version = "0.0.1"
+
 [[package]]
 name = "quote"
 version = "1.0.38"
@@ -1697,6 +1760,7 @@ version = "0.1.0"
 dependencies = [
  "containerd-client",
  "env_logger",
+ "handlebars",
  "hex",
  "log",
  "my-workspace-hack",
@@ -1859,7 +1923,16 @@ version = "1.0.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
 dependencies = [
- "thiserror-impl",
+ "thiserror-impl 1.0.69",
+]
+
+[[package]]
+name = "thiserror"
+version = "2.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d452f284b73e6d76dd36758a0c8684b1d5be31f92b89d07fd5822175732206fc"
+dependencies = [
+ "thiserror-impl 2.0.11",
 ]
 
 [[package]]
@@ -1873,6 +1946,17 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "thiserror-impl"
+version = "2.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26afc1baea8a989337eeb52b6e72a039780ce45c3edfcc9c5b9d112feeb173c2"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "time"
 version = "0.3.37"
@@ -2101,6 +2185,12 @@ version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825"
 
+[[package]]
+name = "ucd-trie"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971"
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.15"

crates/provider/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "provider"
+edition = "2024"
+version.workspace = true
+authors.workspace = true
+
+[dependencies]

crates/provider/src/lib.rs

@@ -0,0 +1,23 @@
+pub mod config;
+pub mod handler;
+pub mod types;
+pub mod httputils;
+pub mod proxy;
+pub mod auth;
+pub mod logs;
+pub mod metrics;
+
+pub fn add(left: u64, right: u64) -> u64 {
+    left + right
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn it_works() {
+        let result = add(2, 2);
+        assert_eq!(result, 4);
+    }
+}

crates/provider/src/types/config.rs

@@ -0,0 +1,58 @@
+use std::time::Duration;
+
+const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
+const DEFAULT_MAX_IDLE_CONNS: usize = 1024;
+
+pub struct FaasHandler<S> {
+    pub list_namespaces: S,
+    pub mutate_namespace: S,
+    pub function_proxy: S,
+    pub function_lister: S,
+    pub deploy_function: S,
+    pub update_function: S,
+    pub delete_function: S,
+    pub function_status: S,
+    pub scale_function: S,
+    pub secrets: S,
+    pub logs: S,
+    pub health: Option<S>,
+    pub info: S,
+    pub telemetry: S,
+}
+
+pub struct FaaSConfig {
+    pub tcp_port: Option<u16>,
+    pub read_timeout: Duration,
+    pub write_timeout: Duration,
+    pub enable_health: bool,
+    pub enable_basic_auth: bool,
+    pub secret_mount_path: String,
+    pub max_idle_conns: usize,
+    pub max_idle_conns_per_host: usize,
+}
+
+impl FaaSConfig {
+    pub fn get_read_timeout(&self) -> Duration {
+        if self.read_timeout <= Duration::from_secs(0) {
+            DEFAULT_READ_TIMEOUT
+        } else {
+            self.read_timeout
+        }
+    }
+
+    pub fn get_max_idle_conns(&self) -> usize {
+        if self.max_idle_conns < 1 {
+            DEFAULT_MAX_IDLE_CONNS
+        } else {
+            self.max_idle_conns
+        }
+    }
+
+    pub fn get_max_idle_conns_per_host(&self) -> usize {
+        if self.max_idle_conns_per_host < 1 {
+            self.get_max_idle_conns()
+        } else {
+            self.max_idle_conns_per_host
+        }
+    }
+}

crates/provider/src/types/mod.rs

@@ -0,0 +1 @@
+pub mod config;

crates/service/Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-containerd-client = "0.6"
+containerd-client = "0.8"
 tokio = { version = "1", features = ["full"] }
 tonic = "0.12"
 serde = { version = "1.0", features = ["derive"] }
@@ -16,3 +16,4 @@ oci-spec =  "0.6"
 sha2 = "0.10"
 hex = "0.4"
 my-workspace-hack = { version = "0.1", path = "../my-workspace-hack" }
+handlebars= "4.1.0"     

crates/service/src/lib.rs

@@ -1,33 +1,37 @@
 pub mod spec;
+pub mod systemd;
 
 use containerd_client::{
     Client,
     services::v1::{
         Container, CreateContainerRequest, CreateTaskRequest, DeleteContainerRequest,
         DeleteTaskRequest, GetImageRequest, KillRequest, ListContainersRequest, ListTasksRequest,
-        ReadContentRequest, StartRequest, WaitRequest,
+        ReadContentRequest, StartRequest, TransferOptions, TransferRequest, WaitRequest,
         container::Runtime,
         snapshots::{MountsRequest, PrepareSnapshotRequest},
     },
+    to_any,
     tonic::Request,
-    types::Mount,
+    types::{
+        Mount, Platform,
+        transfer::{ImageStore, OciRegistry, UnpackConfiguration},
+    },
     with_namespace,
 };
 use oci_spec::image::{Arch, ImageConfiguration, ImageIndex, ImageManifest, MediaType, Os};
 use prost_types::Any;
 use sha2::{Digest, Sha256};
-use spec::generate_spec;
+use spec::{DEFAULT_NAMESPACE, generate_spec};
 use std::{
     fs,
     sync::{Arc, Mutex},
     time::Duration,
+    vec,
 };
 use tokio::time::timeout;
 
 // config.json,dockerhub密钥
 // const DOCKER_CONFIG_DIR: &str = "/var/lib/faasd/.docker/";
-// 命名空间（容器的）
-const NAMESPACE: &str = "default";
 
 type Err = Box<dyn std::error::Error>;
 
@@ -110,10 +114,9 @@ impl Service {
     }
 
     pub async fn remove_container(&self, cid: &str, ns: &str) -> Result<(), Err> {
-        let namespace = match ns {
-            "" => NAMESPACE,
-            _ => ns,
-        };
+        let namespace = self.check_namespace(ns);
+        let namespace = namespace.as_str();
+
         let c = self.client.lock().unwrap();
         let request = ListContainersRequest {
             ..Default::default()
@@ -236,10 +239,9 @@ impl Service {
     }
 
     pub async fn kill_task(&self, cid: String, ns: &str) -> Result<(), Err> {
-        let namespace = match ns {
-            "" => NAMESPACE,
-            _ => ns,
-        };
+        let namespace = self.check_namespace(ns);
+        let namespace = namespace.as_str();
+
         let mut c = self.client.lock().unwrap().tasks();
         let kill_request = KillRequest {
             container_id: cid.to_string(),
@@ -260,10 +262,9 @@ impl Service {
         todo!()
     }
     pub async fn delete_task(&self, cid: &str, ns: &str) {
-        let namespace = match ns {
-            "" => NAMESPACE,
-            _ => ns,
-        };
+        let namespace = self.check_namespace(ns);
+        let namespace = namespace.as_str();
+
         let mut c = self.client.lock().unwrap().tasks();
         let time_out = Duration::from_secs(30);
         let wait_result = timeout(time_out, async {
@@ -314,10 +315,9 @@ impl Service {
     }
 
     pub async fn get_container_list(&self, ns: &str) -> Result<Vec<String>, tonic::Status> {
-        let namespace = match ns {
-            "" => NAMESPACE,
-            _ => ns,
-        };
+        let namespace = self.check_namespace(ns);
+        let namespace = namespace.as_str();
+
         let mut c = self.client.lock().unwrap().containers();
 
         let request = ListContainersRequest {
@@ -340,13 +340,86 @@ impl Service {
         todo!()
     }
 
-    pub fn prepare_image(&self) {
-        todo!()
+    pub async fn prepare_image(
+        &self,
+        image_name: &str,
+        ns: &str,
+        always_pull: bool,
+    ) -> Result<(), Err> {
+        if always_pull {
+            self.pull_image(image_name, ns).await?;
+        } else {
+            let namespace = self.check_namespace(ns);
+            let namespace = namespace.as_str();
+            let mut c = self.client.lock().unwrap().images();
+            let req = GetImageRequest {
+                name: image_name.to_string(),
+            };
+            let resp = c
+                .get(with_namespace!(req, namespace))
+                .await
+                .map_err(|e| {
+                    eprintln!(
+                        "Failed to get the config of {} in namespace {}: {}",
+                        image_name, namespace, e
+                    );
+                    e
+                })
+                .ok()
+                .unwrap()
+                .into_inner();
+            if resp.image.is_none() {
+                self.pull_image(image_name, ns).await?;
+            }
+        }
+        Ok(())
     }
-    pub fn pull_image(&self) {
-        todo!()
+
+    pub async fn pull_image(&self, image_name: &str, ns: &str) -> Result<(), Err> {
+        let namespace = self.check_namespace(ns);
+        let namespace = namespace.as_str();
+
+        let mut c = self.client.lock().unwrap().transfer();
+        let source = OciRegistry {
+            reference: image_name.to_string(),
+            resolver: Default::default(),
+        };
+        // 这里先写死linux amd64
+        let platform = Platform {
+            os: "linux".to_string(),
+            architecture: "amd64".to_string(),
+            ..Default::default()
+        };
+        let dest = ImageStore {
+            name: image_name.to_string(),
+            platforms: vec![platform.clone()],
+            unpacks: vec![UnpackConfiguration {
+                platform: Some(platform),
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let anys = to_any(&source);
+        let anyd = to_any(&dest);
+
+        let req = TransferRequest {
+            source: Some(anys),
+            destination: Some(anyd),
+            options: Some(TransferOptions {
+                ..Default::default()
+            }),
+        };
+        c.transfer(with_namespace!(req, namespace))
+            .await
+            .expect(&format!(
+                "Unable to transfer image {} to namespace {}",
+                image_name, namespace
+            ));
+        Ok(())
     }
-    /// 获取resolver，验证用，后面拉取镜像可能会用到
+
+    // 不用这个也能拉取镜像？
     pub fn get_resolver(&self) {
         todo!()
     }
@@ -518,6 +591,13 @@ impl Service {
         }
         Ok(ret)
     }
+
+    fn check_namespace(&self, ns: &str) -> String {
+        match ns {
+            "" => DEFAULT_NAMESPACE.to_string(),
+            _ => ns.to_string(),
+        }
+    }
 }
 //容器是容器，要先启动，然后才能运行任务
 //要想删除一个正在运行的Task，必须先kill掉这个task，然后才能删除。

crates/service/src/systemd.rs

@@ -0,0 +1,80 @@
+use crate::Err;
+use handlebars::Handlebars;
+use std::{collections::HashMap, fs::File, io::Write, path::Path};
+
+pub struct Systemd;
+
+impl Systemd {
+    pub fn enable(unit: String) -> Result<(), Err> {
+        let output = std::process::Command::new("systemctl")
+            .arg("enable")
+            .arg(&unit)
+            .output()?;
+        if !output.status.success() {
+            return Err(Box::new(std::io::Error::new(
+                std::io::ErrorKind::Other,
+                format!(
+                    "Failed to enable unit {}: {}",
+                    unit,
+                    String::from_utf8_lossy(&output.stderr)
+                ),
+            )));
+        }
+        Ok(())
+    }
+
+    pub fn start(unit: String) -> Result<(), Err> {
+        let output = std::process::Command::new("systemctl")
+            .arg("start")
+            .arg(&unit)
+            .output()?;
+        if !output.status.success() {
+            return Err(Box::new(std::io::Error::new(
+                std::io::ErrorKind::Other,
+                format!(
+                    "Failed to start unit {}: {}",
+                    unit,
+                    String::from_utf8_lossy(&output.stderr)
+                ),
+            )));
+        }
+        Ok(())
+    }
+
+    pub fn daemon_reload() -> Result<(), Err> {
+        let output = std::process::Command::new("systemctl")
+            .arg("daemon-reload")
+            .output()?;
+        if !output.status.success() {
+            return Err(Box::new(std::io::Error::new(
+                std::io::ErrorKind::Other,
+                format!(
+                    "Failed to reload systemd daemon: {}",
+                    String::from_utf8_lossy(&output.stderr)
+                ),
+            )));
+        }
+        Ok(())
+    }
+
+    pub fn install_unit(name: String, tokens: HashMap<String, String>) -> Result<(), Err> {
+        if tokens.get("Cwd").map_or(true, |v| v.is_empty()) {
+            return Err("key Cwd expected in tokens parameter".into());
+        }
+
+        let tmpl_name = format!("./hack/{}.service", name);
+        let mut handlebars = Handlebars::new();
+        handlebars.register_template_file("template", &tmpl_name)?;
+
+        let rendered = handlebars.render("template", &tokens)?;
+        Self::write_unit(&format!("{}.service", name), rendered.as_bytes())?;
+        Ok(())
+    }
+
+    pub fn write_unit(name: &str, content: &[u8]) -> Result<(), Err> {
+        let path = Path::new("/lib/systemd/system").join(name);
+        let mut file = File::create(path)?;
+        file.write_all(content)?;
+        Ok(())
+    }
+}