Rename pipe crd into function

Author: HoKim98

Cargo.toml

@@ -15,6 +15,7 @@ members = [
     "dash/pipe/connectors/storage",
     "dash/pipe/connectors/webcam",          # exclude(alpine)
     "dash/pipe/functions/ai",               # exclude(alpine)
+    "dash/pipe/functions/ai/plugin",
     "dash/pipe/functions/identity",
     "dash/pipe/functions/performance-test",
     "dash/pipe/functions/python",           # exclude(alpine)

ark/core/k8s/src/data.rs

@@ -23,7 +23,9 @@ impl FromStr for Name {
     type Err = Error;
 
     fn from_str(name: &str) -> Result<Self, <Self as FromStr>::Err> {
-        let re = Regex::new(r"^[a-z][a-z0-9_-]*[a-z0-9]?$")?;
+        let field = Self::RE_FIELD;
+        let fields = format!(r"^{field}(\.{field})*$");
+        let re = Regex::new(&fields)?;
         if re.is_match(name) {
             Ok(Self(name.into()))
         } else {
@@ -38,13 +40,6 @@ impl From<Name> for String {
     }
 }
 
-// #[cfg(feature = "nats")]
-// impl From<Name> for ::async_nats::Subject {
-//     fn from(value: Name) -> Self {
-//         value.0.into()
-//     }
-// }
-
 impl Borrow<str> for Name {
     fn borrow(&self) -> &str {
         &self.0
@@ -143,6 +138,14 @@ impl<'de> Deserialize<'de> for Name {
     }
 }
 
+impl Name {
+    const RE_FIELD: &str = r"[a-z]([a-z0-9_-]*[a-z0-9])?";
+
+    pub fn storage(&self) -> &str {
+        self.0.split('.').next().unwrap()
+    }
+}
+
 #[derive(Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[serde(transparent)]
 pub struct Url(pub ::url::Url);

dash/api/src/pipe.rs dash/api/src/function.rsdash/api/src/pipe.rs renamed to dash/api/src/function.rs

@@ -3,7 +3,7 @@ use chrono::{DateTime, Utc};
 use kube::CustomResource;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
-use straw_api::pipe::StrawPipe;
+use straw_api::function::{StrawFunction, StrawFunctionType};
 use strum::{Display, EnumString};
 
 use crate::model::ModelFieldsNativeSpec;
@@ -12,15 +12,15 @@ use crate::model::ModelFieldsNativeSpec;
 #[kube(
     group = "dash.ulagbulag.io",
     version = "v1alpha1",
-    kind = "Pipe",
-    struct = "PipeCrd",
-    status = "PipeStatus",
-    shortname = "pi",
+    kind = "Function",
+    struct = "FunctionCrd",
+    status = "FunctionStatus",
+    shortname = "f",
     namespaced,
     printcolumn = r#"{
         "name": "state",
         "type": "string",
-        "description": "state of the pipe",
+        "description": "state of the function",
         "jsonPath": ".status.state"
     }"#,
     printcolumn = r#"{
@@ -37,32 +37,67 @@ use crate::model::ModelFieldsNativeSpec;
     }"#
 )]
 #[serde(rename_all = "camelCase")]
-pub struct PipeSpec<Spec = Name, Exec = PipeExec> {
+pub struct FunctionSpec<Spec = Name, Exec = FunctionExec> {
     pub input: Spec,
     pub output: Spec,
     #[serde(default, flatten)]
     pub exec: Exec,
+    #[serde(rename = "type")]
+    pub type_: StrawFunctionType,
+    pub volatility: FunctionVolatility,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "camelCase")]
-pub enum PipeExec {
+pub enum FunctionExec {
     Placeholder {},
-    Straw(StrawPipe),
+    Straw(StrawFunction),
 }
 
-impl Default for PipeExec {
+impl Default for FunctionExec {
     fn default() -> Self {
         Self::Placeholder {}
     }
 }
 
+#[derive(
+    Copy,
+    Clone,
+    Debug,
+    Display,
+    EnumString,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Hash,
+    Serialize,
+    Deserialize,
+    JsonSchema,
+)]
+pub enum FunctionVolatility {
+    /// Immutable function.
+    /// If the same input value is received during execution of the function,
+    /// the same output value is always returned.
+    Immutable,
+    /// Stable function.
+    /// If the same input value is received during execution of the function,
+    /// the output value may change.
+    /// However, if the same input value is input with the same timestamp,
+    /// the same output value is always returned.
+    Stable,
+    /// Volatile function.
+    /// This can produce different results every time
+    /// even if the same input value is input.
+    Volatile,
+}
+
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "camelCase")]
-pub struct PipeStatus {
+pub struct FunctionStatus {
     #[serde(default)]
-    pub state: PipeState,
-    pub spec: Option<PipeSpec<ModelFieldsNativeSpec>>,
+    pub state: FunctionState,
+    pub spec: Option<FunctionSpec<ModelFieldsNativeSpec>>,
     pub last_updated: DateTime<Utc>,
 }
 
@@ -82,7 +117,7 @@ pub struct PipeStatus {
     Deserialize,
     JsonSchema,
 )]
-pub enum PipeState {
+pub enum FunctionState {
     #[default]
     Pending,
     Ready,

dash/api/src/lib.rs

@@ -1,7 +1,7 @@
+pub mod function;
 pub mod job;
 pub mod model;
 pub mod model_storage_binding;
-pub mod pipe;
 pub mod storage;
 pub mod task;
 

dash/controller/src/ctx/pipe.rs dash/controller/src/ctx/function.rsdash/controller/src/ctx/pipe.rs renamed to dash/controller/src/ctx/function.rs

@@ -5,8 +5,8 @@ use ark_core_k8s::manager::Manager;
 use async_trait::async_trait;
 use chrono::Utc;
 use dash_api::{
+    function::{FunctionCrd, FunctionSpec, FunctionState, FunctionStatus},
     model::ModelFieldsNativeSpec,
-    pipe::{PipeCrd, PipeSpec, PipeState, PipeStatus},
 };
 use kube::{
     api::{Patch, PatchParams},
@@ -16,14 +16,14 @@ use kube::{
 use serde_json::json;
 use tracing::{info, warn};
 
-use crate::validator::pipe::PipeValidator;
+use crate::validator::function::FunctionValidator;
 
 #[derive(Default)]
 pub struct Ctx {}
 
 #[async_trait]
 impl ::ark_core_k8s::manager::Ctx for Ctx {
-    type Data = PipeCrd;
+    type Data = FunctionCrd;
 
     const NAME: &'static str = crate::consts::NAME;
     const NAMESPACE: &'static str = ::dash_api::consts::NAMESPACE;
@@ -45,24 +45,24 @@ impl ::ark_core_k8s::manager::Ctx for Ctx {
             .map(|status| status.state)
             .unwrap_or_default()
         {
-            PipeState::Pending => {
-                let validator = PipeValidator {
+            FunctionState::Pending => {
+                let validator = FunctionValidator {
                     namespace: &namespace,
                     kube: &manager.kube,
                 };
-                match validator.validate_pipe(data.spec.clone()).await {
+                match validator.validate_function(data.spec.clone()).await {
                     Ok(spec) => {
                         Self::update_spec_or_requeue(&namespace, &manager.kube, &name, spec).await
                     }
                     Err(e) => {
-                        warn!("failed to validate pipe: {name:?}: {e}");
+                        warn!("failed to validate function: {name:?}: {e}");
                         Ok(Action::requeue(
                             <Self as ::ark_core_k8s::manager::Ctx>::FALLBACK,
                         ))
                     }
                 }
             }
-            PipeState::Ready => {
+            FunctionState::Ready => {
                 // TODO: implement to finding changes
                 Ok(Action::await_change())
             }
@@ -75,17 +75,17 @@ impl Ctx {
         namespace: &str,
         kube: &Client,
         name: &str,
-        spec: PipeSpec<ModelFieldsNativeSpec>,
+        spec: FunctionSpec<ModelFieldsNativeSpec>,
     ) -> Result<Action, Error> {
         match Self::update_spec(namespace, kube, name, spec).await {
             Ok(()) => {
-                info!("pipe is ready: {namespace}/{name}");
+                info!("function is ready: {namespace}/{name}");
                 Ok(Action::requeue(
                     <Self as ::ark_core_k8s::manager::Ctx>::FALLBACK,
                 ))
             }
             Err(e) => {
-                warn!("failed to update pipe state ({namespace}/{name}): {e}");
+                warn!("failed to update function state ({namespace}/{name}): {e}");
                 Ok(Action::requeue(
                     <Self as ::ark_core_k8s::manager::Ctx>::FALLBACK,
                 ))
@@ -97,7 +97,7 @@ impl Ctx {
         namespace: &str,
         kube: &Client,
         name: &str,
-        spec: PipeSpec<ModelFieldsNativeSpec>,
+        spec: FunctionSpec<ModelFieldsNativeSpec>,
     ) -> Result<()> {
         let api = Api::<<Self as ::ark_core_k8s::manager::Ctx>::Data>::namespaced(
             kube.clone(),
@@ -108,8 +108,8 @@ impl Ctx {
         let patch = Patch::Merge(json!({
             "apiVersion": crd.api_version,
             "kind": crd.kind,
-            "status": PipeStatus {
-                state: PipeState::Ready,
+            "status": FunctionStatus {
+                state: FunctionState::Ready,
                 spec: Some(spec),
                 last_updated: Utc::now(),
             },

dash/controller/src/ctx/mod.rs

@@ -1,6 +1,6 @@
+pub mod function;
 pub mod job;
 pub mod model;
 pub mod model_storage_binding;
-pub mod pipe;
 pub mod storage;
 pub mod task;

dash/controller/src/main.rs

@@ -11,11 +11,11 @@ pub(crate) mod consts {
 #[tokio::main]
 async fn main() {
     join!(
-        self::ctx::task::Ctx::spawn_crd(),
+        self::ctx::function::Ctx::spawn_crd(),
         self::ctx::job::Ctx::spawn_crd(),
         self::ctx::model::Ctx::spawn_crd(),
         self::ctx::model_storage_binding::Ctx::spawn_crd(),
-        self::ctx::pipe::Ctx::spawn_crd(),
         self::ctx::storage::Ctx::spawn_crd(),
+        self::ctx::task::Ctx::spawn_crd(),
     );
 }

dash/controller/src/validator/function.rs

@@ -0,0 +1,108 @@
+use anyhow::Result;
+use ark_core_k8s::data::Name;
+use dash_api::{
+    function::{FunctionExec, FunctionSpec},
+    model::{
+        ModelFieldKindExtendedSpec, ModelFieldKindSpec, ModelFieldSpec, ModelFieldsNativeSpec,
+    },
+};
+use dash_provider::storage::KubernetesStorageClient;
+use kube::Client;
+use straw_api::{
+    function::{StrawFunction, StrawFunctionType},
+    plugin::PluginContext,
+};
+use straw_provider::StrawSession;
+
+use super::model::ModelValidator;
+
+pub struct FunctionValidator<'namespace, 'kube> {
+    pub namespace: &'namespace str,
+    pub kube: &'kube Client,
+}
+
+impl<'namespace, 'kube> FunctionValidator<'namespace, 'kube> {
+    pub async fn validate_function(
+        &self,
+        spec: FunctionSpec,
+    ) -> Result<FunctionSpec<ModelFieldsNativeSpec>> {
+        let FunctionSpec {
+            input,
+            output,
+            exec,
+            type_,
+            volatility,
+        } = spec;
+
+        let models = Models { input, output };
+
+        let model_validator = ModelValidator {
+            kubernetes_storage: KubernetesStorageClient {
+                namespace: self.namespace,
+                kube: self.kube,
+            },
+        };
+        let validate_model = |name| async {
+            model_validator
+                .validate_fields(vec![ModelFieldSpec {
+                    name: "/".into(),
+                    kind: ModelFieldKindSpec::Extended(ModelFieldKindExtendedSpec::Model { name }),
+                    attribute: Default::default(),
+                }])
+                .await
+        };
+
+        Ok(FunctionSpec {
+            input: validate_model(models.input.clone().into()).await?,
+            output: validate_model(models.output.clone().into()).await?,
+            exec: self.validate_exec(type_, exec, models).await?,
+            type_,
+            volatility,
+        })
+    }
+
+    async fn validate_exec(
+        &self,
+        type_: StrawFunctionType,
+        exec: FunctionExec,
+        models: Models,
+    ) -> Result<FunctionExec> {
+        match exec {
+            FunctionExec::Placeholder {} => Ok(exec),
+            FunctionExec::Straw(exec) => self
+                .validate_exec_straw(type_, exec, models)
+                .await
+                .map(FunctionExec::Straw),
+        }
+    }
+
+    async fn validate_exec_straw(
+        &self,
+        type_: StrawFunctionType,
+        function: StrawFunction,
+        models: Models,
+    ) -> Result<StrawFunction> {
+        if type_ == StrawFunctionType::Pipe {
+            self.validate_model_storage_binding(&models).await?;
+        }
+
+        let ctx = PluginContext::new(type_, Some(models.input), Some(models.output));
+        let session = StrawSession::new(self.kube.clone(), Some(self.namespace.into()));
+        session.create(&ctx, &function).await.map(|()| function)
+    }
+
+    async fn validate_model_storage_binding(&self, models: &Models) -> Result<()> {
+        let client = KubernetesStorageClient {
+            namespace: self.namespace,
+            kube: self.kube,
+        };
+        client.ensure_model_storage_binding(&models.input).await?;
+        client.ensure_model_storage_binding(&models.output).await?;
+        Ok(())
+    }
+}
+
+struct Models {
+    input: Name,
+    output: Name,
+}