创建基本文件夹

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,18 @@
+pub mod config;
+pub mod handler;
+pub mod types;
+
+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(())
+    }
+}