Repository: fluid-cloudnative/fluid
Pull request #3695 (2024-01-16 13:43:40)

Add a regex to validate the resource before sending the http request in the vineyard Runtime


### Ⅰ. Describe what this PR does

* Add a regex to validate the resource before sending the http request.
* Add a timeout for http get request.

### Ⅱ. Does this pull request fix one issue?
<!--If so, add "fixes #xxxx" so that the issue will be closed when this PR is merged (for example, "fixes #15" to close Issue #15). Otherwise, add "NONE" -->
fixes #XXXX

### Ⅲ. List the added test cases (unit test/integration test) if any, please explain if no tests are needed.


### Ⅳ. Describe how to verify it


### Ⅴ. Special notes for reviews

fluid-e2e-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please assign zwwhdls for approval by writing `/assign @zwwhdls` in a comment. For more information see:[The Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=fluid-cloudnative%2Ffluid).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/fluid-cloudnative/fluid/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["zwwhdls"]} -->
fluid-e2e-bot[bot]: Hi @dashanji. Thanks for your PR.

I'm waiting for a [fluid-cloudnative](https://github.com/orgs/fluid-cloudnative/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/fluid-cloudnative/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=fluid-cloudnative%2Ffluid).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695) **Quality Gate passed**  
Kudos, no new issues were introduced!

[0 New issues](https://sonarcloud.io/project/issues?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
No data about Coverage  
[21.9% Duplication on New Code](https://sonarcloud.io/component_measures?id=fluid-cloudnative_fluid&pullRequest=3695&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695)


sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695) **Quality Gate passed**  
Kudos, no new issues were introduced!

[0 New issues](https://sonarcloud.io/project/issues?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
No data about Coverage  
[21.9% Duplication on New Code](https://sonarcloud.io/component_measures?id=fluid-cloudnative_fluid&pullRequest=3695&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695)


sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695) **Quality Gate passed**  
Kudos, no new issues were introduced!

[0 New issues](https://sonarcloud.io/project/issues?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
No data about Coverage  
[21.9% Duplication on New Code](https://sonarcloud.io/component_measures?id=fluid-cloudnative_fluid&pullRequest=3695&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695)


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/fluid-cloudnative/fluid/pull/3695?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative) Report
Attention: `4 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`7b0a6b1`)](https://app.codecov.io/gh/fluid-cloudnative/fluid/commit/7b0a6b1cce5448fd2a3afa1ed32aeb4140d0f75c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative) 64.46% compared to head [(`c310e2d`)](https://app.codecov.io/gh/fluid-cloudnative/fluid/pull/3695?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative) 64.46%.

> :exclamation: Current head c310e2d differs from pull request most recent head aadca21. Consider uploading reports for the commit aadca21 to get more accurate results

| [Files](https://app.codecov.io/gh/fluid-cloudnative/fluid/pull/3695?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative) | Patch % | Lines |
|---|---|---|
| [pkg/ddc/vineyard/operations/base.go](https://app.codecov.io/gh/fluid-cloudnative/fluid/pull/3695?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative#diff-cGtnL2RkYy92aW5leWFyZC9vcGVyYXRpb25zL2Jhc2UuZ28=) | 75.00% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/fluid-cloudnative/fluid/pull/3695?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative) |

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #3695   +/-   ##
=======================================
  Coverage   64.46%   64.46%           
=======================================
  Files         471      471           
  Lines       28153    28167   +14     
=======================================
+ Hits        18149    18159   +10     
- Misses       7848     7850    +2     
- Partials     2156     2158    +2     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/fluid-cloudnative/fluid/pull/3695?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fluid-cloudnative).

sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695) **Quality Gate passed**  
Kudos, no new issues were introduced!

[0 New issues](https://sonarcloud.io/project/issues?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=fluid-cloudnative_fluid&pullRequest=3695&resolved=false&inNewCodePeriod=true)  
No data about Coverage  
[25.3% Duplication on New Code](https://sonarcloud.io/component_measures?id=fluid-cloudnative_fluid&pullRequest=3695&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=fluid-cloudnative_fluid&pullRequest=3695)



Patch:
diff --git a/integration/kubeflow/README.md b/integration/kubeflow/README.md
new file mode 100644
index 00000000..da1e9fe6
--- /dev/null
+++ b/integration/kubeflow/README.md
@@ -0,0 +1,37 @@
+# Demo - Using FLuid in Kubeflow Pipelines v2
+This is a demo that wraps Fluid'operations as KFP v2 components to complete the processes of dataset creation, runtime creation, and cache preheating to accelerate model training (a simple CNN model for [Fashion MNIST](https://www.kaggle.com/datasets/zalando-research/fashionmnist)).
+
+## Prerequisites
+
+### Installation 
+- [Fluid](https://github.com/fluid-cloudnative/fluid)
+
+- [Kubeflow Pipelines v2](https://www.kubeflow.org/docs/components/pipelines/v2/)
+
+Please refer to the [Fluid installation guide](https://github.com/fluid-cloudnative/fluid/blob/master/docs/zh/userguide/install.md) and [KFP v2 installation guide](https://www.kubeflow.org/docs/components/pipelines/v2/installation/quickstart/) to complete the installation of Fluid and KFP v2.
+
+### Dataset
+- [Fashion MNIST](https://www.kaggle.com/datasets/zalando-research/fashionmnist)
+
+You should upload `fashion-mnist_train.csv` and `fashion-mnist_test.csv` to your Amazon S3 (or any S3-compatible storage, such as [MinIO](https://min.io/)) and deploy the [s3-secret.yaml](./s3-secret.yaml) to provide the access key.
+
+### RBAC
+Because KFP components require access or modification permissions to Fluid resources, it is necessary to deploy [rbac.yaml](./rbac.yaml) in advance to grant permissions.
+
+
+## Demo
+The sample is composed of dataset creation, alluxioruntime creation, data  preloading ,and model training.
+
+The simple pipeline provides the following parameters:
+- batch_size: int
+- dataset_name: str
+- epochs: int
+- learning_rate: float
+- mount_point: str
+- mount_s3_endpoint: str
+- mount_s3_region: str
+- namespace: str (For now, this value should be the namespace in which you deploy your KFP)
+
+If you want to run the sample, you can upload the [train-cnn-for-fashion-mnist-pipline.yaml](./pipline-yaml/train-cnn-for-fashion-mnist-pipline.yaml) to your pipeline dashboard UI and fill in these parameters.
+
+Most importantly, this is just an example of packaging Fluid operations into KFP components, and users need to develop KFP components according to their own needs.
\ No newline at end of file
diff --git a/integration/kubeflow/component-yaml/cleanup-dataset-and-alluxioruntime.yaml b/integration/kubeflow/component-yaml/cleanup-dataset-and-alluxioruntime.yaml
new file mode 100644
index 00000000..f74a28db
--- /dev/null
+++ b/integration/kubeflow/component-yaml/cleanup-dataset-and-alluxioruntime.yaml
@@ -0,0 +1,77 @@
+# PIPELINE DEFINITION
+# Name: cleanup-dataset-and-alluxio-runtime
+# Inputs:
+#    dataset_name: str
+#    namespace: str
+components:
+  comp-cleanup-dataset-and-alluxio-runtime:
+    executorLabel: exec-cleanup-dataset-and-alluxio-runtime
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+deploymentSpec:
+  executors:
+    exec-cleanup-dataset-and-alluxio-runtime:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - cleanup_dataset_and_alluxio_runtime
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef cleanup_dataset_and_alluxio_runtime(dataset_name: str, namespace:\
+          \ str):\n    import logging\n    from fluid import FluidClient\n\n    fluid_client\
+          \ = FluidClient()\n    fluid_client.delete_runtime(name=dataset_name, namespace=namespace,\
+          \ runtime_type=\"alluxio\", wait_until_cleaned_up=True)\n    fluid_client.delete_dataset(name=dataset_name,\
+          \ namespace=namespace, wait_until_cleaned_up=True)\n\n    logging.info(f\"\
+          Cleanup Dataset and AlluxioRuntime \\\"{namespace}/{dataset_name}\\\" successfully!\"\
+          )\n\n"
+        image: python:3.7
+pipelineInfo:
+  name: cleanup-dataset-and-alluxio-runtime
+root:
+  dag:
+    tasks:
+      cleanup-dataset-and-alluxio-runtime:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-cleanup-dataset-and-alluxio-runtime
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: cleanup-dataset-and-alluxio-runtime
+  inputDefinitions:
+    parameters:
+      dataset_name:
+        parameterType: STRING
+      namespace:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.4.0
diff --git a/integration/kubeflow/component-yaml/cleanup-preheat-operation.yaml b/integration/kubeflow/component-yaml/cleanup-preheat-operation.yaml
new file mode 100644
index 00000000..25dfa45b
--- /dev/null
+++ b/integration/kubeflow/component-yaml/cleanup-preheat-operation.yaml
@@ -0,0 +1,75 @@
+# PIPELINE DEFINITION
+# Name: cleanup-preheat-operation
+# Inputs:
+#    dataset_name: str
+#    namespace: str
+components:
+  comp-cleanup-preheat-operation:
+    executorLabel: exec-cleanup-preheat-operation
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+deploymentSpec:
+  executors:
+    exec-cleanup-preheat-operation:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - cleanup_preheat_operation
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef cleanup_preheat_operation(dataset_name: str, namespace: str):\n\
+          \    import logging\n    from fluid import FluidClient\n\n    fluid_client\
+          \ = FluidClient()\n    fluid_client.delete_data_operation(name=\"%s-loader\"\
+          \ % dataset_name, data_op_type=\"dataload\", namespace=namespace)\n    logging.info(\"\
+          Cleanup preheat dataset operation successfully!\")\n\n"
+        image: python:3.7
+pipelineInfo:
+  name: cleanup-preheat-operation
+root:
+  dag:
+    tasks:
+      cleanup-preheat-operation:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-cleanup-preheat-operation
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: cleanup-preheat-operation
+  inputDefinitions:
+    parameters:
+      dataset_name:
+        parameterType: STRING
+      namespace:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.4.0
diff --git a/integration/kubeflow/component-yaml/create-alluxioruntime.yaml b/integration/kubeflow/component-yaml/create-alluxioruntime.yaml
new file mode 100644
index 00000000..953db9f2
--- /dev/null
+++ b/integration/kubeflow/component-yaml/create-alluxioruntime.yaml
@@ -0,0 +1,85 @@
+# PIPELINE DEFINITION
+# Name: create-alluxio-runtime
+# Inputs:
+#    dataset_name: str
+#    namespace: str
+components:
+  comp-create-alluxio-runtime:
+    executorLabel: exec-create-alluxio-runtime
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+deploymentSpec:
+  executors:
+    exec-create-alluxio-runtime:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - create_alluxio_runtime
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef create_alluxio_runtime(dataset_name: str, namespace: str):\n\
+          \    import logging\n    from fluid import AlluxioRuntime, AlluxioRuntimeSpec,\
+          \ models, FluidClient\n    from kubernetes import client as k8s_client\n\
+          \n    fluid_client = FluidClient()\n\n    FLUID_GROUP = \"data.fluid.io\"\
+          \n    FLUID_VERSION = \"v1alpha1\"\n\n    replicas = 1\n\n    # This is\
+          \ the simplest configuration for AlluxioRuntime, you can change the AlluxioRuntime\
+          \ according to your needs\n    alluxio_runtime = AlluxioRuntime(\n     \
+          \   api_version=\"%s/%s\" % (FLUID_GROUP, FLUID_VERSION),\n        kind=\"\
+          AlluxioRuntime\",\n        metadata=k8s_client.V1ObjectMeta(\n         \
+          \   name=dataset_name,\n            namespace=namespace\n        ),\n  \
+          \      spec=AlluxioRuntimeSpec(\n            replicas=replicas,\n      \
+          \      tieredstore=models.TieredStore([models.Level('0.95', '0.7', 'MEM',\
+          \ '/dev/shm', '2Gi', volume_type=None)])\n        )\n    )\n\n    fluid_client.create_runtime(alluxio_runtime)\n\
+          \n\n    logging.info(f\"Runtime \\\"{alluxio_runtime.metadata.namespace}/{alluxio_runtime.metadata.name}\\\
+          \" created successfully\")\n\n"
+        image: python:3.7
+pipelineInfo:
+  name: create-alluxio-runtime
+root:
+  dag:
+    tasks:
+      create-alluxio-runtime:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-create-alluxio-runtime
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: create-alluxio-runtime
+  inputDefinitions:
+    parameters:
+      dataset_name:
+        parameterType: STRING
+      namespace:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.4.0
diff --git a/integration/kubeflow/component-yaml/create-s3-dataset.yaml b/integration/kubeflow/component-yaml/create-s3-dataset.yaml
new file mode 100644
index 00000000..e01b47b3
--- /dev/null
+++ b/integration/kubeflow/component-yaml/create-s3-dataset.yaml
@@ -0,0 +1,121 @@
+# PIPELINE DEFINITION
+# Name: create-s3-dataset
+# Inputs:
+#    dataset_name: str
+#    mount_point: str
+#    mount_s3_endpoint: str
+#    mount_s3_region: str
+#    namespace: str
+components:
+  comp-create-s3-dataset:
+    executorLabel: exec-create-s3-dataset
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        mount_point:
+          parameterType: STRING
+        mount_s3_endpoint:
+          parameterType: STRING
+        mount_s3_region:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+deploymentSpec:
+  executors:
+    exec-create-s3-dataset:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - create_s3_dataset
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef create_s3_dataset(dataset_name: str, namespace: str, mount_point:\
+          \ str, mount_s3_endpoint: str, mount_s3_region: str):\n    import logging\n\
+          \    import fluid\n    from kubernetes import client\n    fluid_client =\
+          \ fluid.FluidClient()\n\n    FLUID_GROUP = \"data.fluid.io\"\n    FLUID_VERSION\
+          \ = \"v1alpha1\"\n\n    # This is an sample which use some pre-defined options.\n\
+          \    # Users can change these code customily\n    dataset = fluid.Dataset(\n\
+          \        api_version=\"%s/%s\" % (FLUID_GROUP, FLUID_VERSION),\n       \
+          \ kind=\"Dataset\",\n        metadata=client.V1ObjectMeta(\n           \
+          \ name=dataset_name,\n            namespace=namespace\n        ),\n    \
+          \    spec=fluid.DatasetSpec(\n            mounts=[\n                fluid.Mount(\n\
+          \                    mount_point=mount_point,\n                    name=dataset_name,\n\
+          \                    options={\n                        \"alluxio.underfs.s3.endpoint\"\
+          : mount_s3_endpoint,\n                        \"alluxio.underfs.s3.endpoint.region\"\
+          : mount_s3_region,\n                        \"alluxio.underfs.s3.disable.dns.buckets\"\
+          : \"true\",\n                        \"alluxio.underfs.s3.disable.inherit.acl\"\
+          : \"false\"\n                    },\n                    encrypt_options=[\n\
+          \                        {\n                            \"name\": \"aws.accessKeyId\"\
+          ,\n                            \"valueFrom\": {\n                      \
+          \        \"secretKeyRef\": {\n                                \"name\":\
+          \ \"s3-secret\",\n                                \"key\": \"aws.accessKeyId\"\
+          \n                              }\n                            }\n     \
+          \                   },\n                        {\n                    \
+          \        \"name\": \"aws.secretKey\",\n                            \"valueFrom\"\
+          : {\n                              \"secretKeyRef\": {\n               \
+          \                 \"name\": \"s3-secret\",\n                           \
+          \     \"key\": \"aws.secretKey\"\n                              }\n    \
+          \                        }\n                        }\n                \
+          \    ]\n                )\n            ]\n        )\n    )\n\n    fluid_client.create_dataset(dataset)\n\
+          \n    logging.info(f\"Dataset \\\"{dataset.metadata.namespace}/{dataset.metadata.name}\\\
+          \" created successfully\")\n\n"
+        image: python:3.7
+pipelineInfo:
+  name: create-s3-dataset
+root:
+  dag:
+    tasks:
+      create-s3-dataset:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-create-s3-dataset
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            mount_point:
+              componentInputParameter: mount_point
+            mount_s3_endpoint:
+              componentInputParameter: mount_s3_endpoint
+            mount_s3_region:
+              componentInputParameter: mount_s3_region
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: create-s3-dataset
+  inputDefinitions:
+    parameters:
+      dataset_name:
+        parameterType: STRING
+      mount_point:
+        parameterType: STRING
+      mount_s3_endpoint:
+        parameterType: STRING
+      mount_s3_region:
+        parameterType: STRING
+      namespace:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.4.0
diff --git a/integration/kubeflow/component-yaml/preheat-dataset.yaml b/integration/kubeflow/component-yaml/preheat-dataset.yaml
new file mode 100644
index 00000000..145adb92
--- /dev/null
+++ b/integration/kubeflow/component-yaml/preheat-dataset.yaml
@@ -0,0 +1,82 @@
+# PIPELINE DEFINITION
+# Name: preheat-dataset
+# Inputs:
+#    dataset_name: str
+#    namespace: str
+components:
+  comp-preheat-dataset:
+    executorLabel: exec-preheat-dataset
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+deploymentSpec:
+  executors:
+    exec-preheat-dataset:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - preheat_dataset
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef preheat_dataset(dataset_name: str, namespace: str):\n    import\
+          \ logging\n    from fluid import DataLoad, DataLoadSpec, FluidClient\n \
+          \   from kubernetes import client as k8s_client\n\n    fluid_client = FluidClient()\n\
+          \n    FLUID_GROUP = \"data.fluid.io\"\n    FLUID_VERSION = \"v1alpha1\"\n\
+          \n    dataload = DataLoad(\n        api_version=\"%s/%s\" % (FLUID_GROUP,\
+          \ FLUID_VERSION),\n        kind=\"DataLoad\",\n        metadata=k8s_client.V1ObjectMeta(\n\
+          \            name=\"%s-loader\" % dataset_name,\n            namespace=namespace\n\
+          \        ),\n        spec=DataLoadSpec(\n            dataset={\n       \
+          \         \"name\": dataset_name,\n                \"namespace\": namespace\n\
+          \            }\n        )\n    )\n\n    fluid_client.create_data_operation(data_op=dataload,\
+          \ wait=True)\n\n    logging.info(f\"Load Dataset \\\"{namespace}/{dataset_name}\\\
+          \"  successfully\")\n\n"
+        image: python:3.7
+pipelineInfo:
+  name: preheat-dataset
+root:
+  dag:
+    tasks:
+      preheat-dataset:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-preheat-dataset
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: preheat-dataset
+  inputDefinitions:
+    parameters:
+      dataset_name:
+        parameterType: STRING
+      namespace:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.4.0
diff --git a/integration/kubeflow/fluid_components.py b/integration/kubeflow/fluid_components.py
new file mode 100644
index 00000000..ed3f494f
--- /dev/null
+++ b/integration/kubeflow/fluid_components.py
@@ -0,0 +1,168 @@
+# Copyright 2024 The Fluid Authors.
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from kfp import dsl, compiler
+
+# Create a Fluid dataset which contains data in S3.
+@dsl.component(packages_to_install=['git+https://github.com/fluid-cloudnative/fluid-client-python.git'])
+def create_s3_dataset(dataset_name: str, namespace: str, mount_point: str, mount_s3_endpoint: str, mount_s3_region: str):
+    import logging
+    import fluid
+    from kubernetes import client
+    fluid_client = fluid.FluidClient()
+
+    FLUID_GROUP = "data.fluid.io"
+    FLUID_VERSION = "v1alpha1"
+
+    # This is an sample which use some pre-defined options.
+    # Users can change these code customily
+    dataset = fluid.Dataset(
+        api_version="%s/%s" % (FLUID_GROUP, FLUID_VERSION),
+        kind="Dataset",
+        metadata=client.V1ObjectMeta(
+            name=dataset_name,
+            namespace=namespace
+        ),
+        spec=fluid.DatasetSpec(
+            mounts=[
+                fluid.Mount(
+                    mount_point=mount_point,
+                    name=dataset_name,
+                    options={
+                        "alluxio.underfs.s3.endpoint": mount_s3_endpoint,
+                        "alluxio.underfs.s3.endpoint.region": mount_s3_region,
+                        "alluxio.underfs.s3.disable.dns.buckets": "true",
+                        "alluxio.underfs.s3.disable.inherit.acl": "false"
+                    },
+                    encrypt_options=[
+                        {
+                            "name": "aws.accessKeyId",
+                            "valueFrom": {
+                              "secretKeyRef": {
+                                "name": "s3-secret",
+                                "key": "aws.accessKeyId"
+                              }
+                            }
+                        },
+                        {
+                            "name": "aws.secretKey",
+                            "valueFrom": {
+                              "secretKeyRef": {
+                                "name": "s3-secret",
+                                "key": "aws.secretKey"
+                              }
+                            }
+                        }
+                    ]
+                )
+            ]
+        )
+    )
+
+    fluid_client.create_dataset(dataset)
+    
+    logging.info(f"Dataset \"{dataset.metadata.namespace}/{dataset.metadata.name}\" created successfully")
+
+# Deploy a simple AlluxioRuntime
+@dsl.component(packages_to_install=['git+https://github.com/fluid-cloudnative/fluid-client-python.git'])
+def create_alluxio_runtime(dataset_name: str, namespace: str):
+    import logging
+    from fluid import AlluxioRuntime, AlluxioRuntimeSpec, models, FluidClient
+    from kubernetes import client as k8s_client
+
+    fluid_client = FluidClient()
+
+    FLUID_GROUP = "data.fluid.io"
+    FLUID_VERSION = "v1alpha1"
+
+    replicas = 1
+
+    # This is the simplest configuration for AlluxioRuntime, you can change the AlluxioRuntime according to your needs
+    alluxio_runtime = AlluxioRuntime(
+        api_version="%s/%s" % (FLUID_GROUP, FLUID_VERSION),
+        kind="AlluxioRuntime",
+        metadata=k8s_client.V1ObjectMeta(
+            name=dataset_name,
+            namespace=namespace
+        ),
+        spec=AlluxioRuntimeSpec(
+            replicas=replicas,
+            tieredstore=models.TieredStore([models.Level('0.95', '0.7', 'MEM', '/dev/shm', '2Gi', volume_type=None)])
+        )
+    )
+
+    fluid_client.create_runtime(alluxio_runtime)
+
+
+    logging.info(f"Runtime \"{alluxio_runtime.metadata.namespace}/{alluxio_runtime.metadata.name}\" created successfully")
+
+# Preheat the dataset with specific dataset name and namespace
+@dsl.component(packages_to_install=['git+https://github.com/fluid-cloudnative/fluid-client-python.git'])
+def preheat_dataset(dataset_name: str, namespace: str):
+    import logging
+    from fluid import DataLoad, DataLoadSpec, FluidClient
+    from kubernetes import client as k8s_client
+    
+    fluid_client = FluidClient()
+
+    FLUID_GROUP = "data.fluid.io"
+    FLUID_VERSION = "v1alpha1"
+
+    dataload = DataLoad(
+        api_version="%s/%s" % (FLUID_GROUP, FLUID_VERSION),
+        kind="DataLoad",
+        metadata=k8s_client.V1ObjectMeta(
+            name="%s-loader" % dataset_name,
+            namespace=namespace
+        ),
+        spec=DataLoadSpec(
+            dataset={
+                "name": dataset_name,
+                "namespace": namespace
+            }
+        )
+    )
+    
+    fluid_client.create_data_operation(data_op=dataload, wait=True)
+    
+    logging.info(f"Load Dataset \"{namespace}/{dataset_name}\"  successfully")
+
+# Cleanup the dataset along with the corresponding alluxioruntime
+@dsl.component(packages_to_install=['git+https://github.com/fluid-cloudnative/fluid-client-python.git'])
+def cleanup_dataset_and_alluxio_runtime(dataset_name: str, namespace: str):
+    import logging
+    from fluid import FluidClient
+    
+    fluid_client = FluidClient()
+    fluid_client.delete_runtime(name=dataset_name, namespace=namespace, runtime_type="alluxio", wait_until_cleaned_up=True)
+    fluid_client.delete_dataset(name=dataset_name, namespace=namespace, wait_until_cleaned_up=True)
+
+    logging.info(f"Cleanup Dataset and AlluxioRuntime \"{namespace}/{dataset_name}\" successfully!")
+
+# Cleanup the preheat dataset operation
+@dsl.component(packages_to_install=['git+https://github.com/fluid-cloudnative/fluid-client-python.git'])
+def cleanup_preheat_operation(dataset_name: str, namespace: str):
+    import logging
+    from fluid import FluidClient
+    
+    fluid_client = FluidClient()
+    fluid_client.delete_data_operation(name="%s-loader" % dataset_name, data_op_type="dataload", namespace=namespace)
+    logging.info("Cleanup preheat dataset operation successfully!")
+
+# Re-run this file when you changed code above to re-generate components' yaml file.
+compiler.Compiler().compile(create_s3_dataset, "./component-yaml/create-s3-dataset.yaml")
+compiler.Compiler().compile(create_alluxio_runtime, "./component-yaml/create-alluxioruntime.yaml")
+compiler.Compiler().compile(preheat_dataset, "./component-yaml/preheat-dataset.yaml")
+compiler.Compiler().compile(cleanup_dataset_and_alluxio_runtime, "./component-yaml/cleanup-dataset-and-alluxioruntime.yaml")
+compiler.Compiler().compile(cleanup_preheat_operation, "./component-yaml/cleanup-preheat-operation.yaml")
\ No newline at end of file
diff --git a/integration/kubeflow/pipline-yaml/train-cnn-for-fashion-mnist-pipline.yaml b/integration/kubeflow/pipline-yaml/train-cnn-for-fashion-mnist-pipline.yaml
new file mode 100644
index 00000000..a649f9b1
--- /dev/null
+++ b/integration/kubeflow/pipline-yaml/train-cnn-for-fashion-mnist-pipline.yaml
@@ -0,0 +1,519 @@
+# PIPELINE DEFINITION
+# Name: train-cnn-for-fashion-mnist
+# Inputs:
+#    batch_size: int
+#    dataset_name: str
+#    epochs: int
+#    learning_rate: float
+#    mount_point: str
+#    mount_s3_endpoint: str
+#    mount_s3_region: str
+#    namespace: str
+components:
+  comp-cleanup-dataset-and-alluxio-runtime:
+    executorLabel: exec-cleanup-dataset-and-alluxio-runtime
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+  comp-cleanup-preheat-operation:
+    executorLabel: exec-cleanup-preheat-operation
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+  comp-create-alluxio-runtime:
+    executorLabel: exec-create-alluxio-runtime
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+  comp-create-s3-dataset:
+    executorLabel: exec-create-s3-dataset
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        mount_point:
+          parameterType: STRING
+        mount_s3_endpoint:
+          parameterType: STRING
+        mount_s3_region:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+  comp-preheat-dataset:
+    executorLabel: exec-preheat-dataset
+    inputDefinitions:
+      parameters:
+        dataset_name:
+          parameterType: STRING
+        namespace:
+          parameterType: STRING
+  comp-train-simple-cnn:
+    executorLabel: exec-train-simple-cnn
+    inputDefinitions:
+      parameters:
+        batch_size:
+          parameterType: NUMBER_INTEGER
+        data_root_path:
+          parameterType: STRING
+        dataset_name:
+          parameterType: STRING
+        epochs:
+          parameterType: NUMBER_INTEGER
+        learning_rate:
+          parameterType: NUMBER_DOUBLE
+deploymentSpec:
+  executors:
+    exec-cleanup-dataset-and-alluxio-runtime:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - cleanup_dataset_and_alluxio_runtime
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef cleanup_dataset_and_alluxio_runtime(dataset_name: str, namespace:\
+          \ str):\n    import logging\n    from fluid import FluidClient\n\n    fluid_client\
+          \ = FluidClient()\n    fluid_client.delete_runtime(name=dataset_name, namespace=namespace,\
+          \ runtime_type=\"alluxio\", wait_until_cleaned_up=True)\n    fluid_client.delete_dataset(name=dataset_name,\
+          \ namespace=namespace, wait_until_cleaned_up=True)\n\n    logging.info(f\"\
+          Cleanup Dataset and AlluxioRuntime \\\"{namespace}/{dataset_name}\\\" successfully!\"\
+          )\n\n"
+        image: python:3.7
+    exec-cleanup-preheat-operation:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - cleanup_preheat_operation
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef cleanup_preheat_operation(dataset_name: str, namespace: str):\n\
+          \    import logging\n    from fluid import FluidClient\n\n    fluid_client\
+          \ = FluidClient()\n    fluid_client.delete_data_operation(name=\"%s-loader\"\
+          \ % dataset_name, data_op_type=\"dataload\", namespace=namespace)\n    logging.info(\"\
+          Cleanup preheat dataset operation successfully!\")\n\n"
+        image: python:3.7
+    exec-create-alluxio-runtime:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - create_alluxio_runtime
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef create_alluxio_runtime(dataset_name: str, namespace: str):\n\
+          \    import logging\n    from fluid import AlluxioRuntime, AlluxioRuntimeSpec,\
+          \ models, FluidClient\n    from kubernetes import client as k8s_client\n\
+          \n    fluid_client = FluidClient()\n\n    FLUID_GROUP = \"data.fluid.io\"\
+          \n    FLUID_VERSION = \"v1alpha1\"\n\n    replicas = 1\n\n    # This is\
+          \ the simplest configuration for AlluxioRuntime, you can change the AlluxioRuntime\
+          \ according to your needs\n    alluxio_runtime = AlluxioRuntime(\n     \
+          \   api_version=\"%s/%s\" % (FLUID_GROUP, FLUID_VERSION),\n        kind=\"\
+          AlluxioRuntime\",\n        metadata=k8s_client.V1ObjectMeta(\n         \
+          \   name=dataset_name,\n            namespace=namespace\n        ),\n  \
+          \      spec=AlluxioRuntimeSpec(\n            replicas=replicas,\n      \
+          \      tieredstore=models.TieredStore([models.Level('0.95', '0.7', 'MEM',\
+          \ '/dev/shm', '2Gi', volume_type=None)])\n        )\n    )\n\n    fluid_client.create_runtime(alluxio_runtime)\n\
+          \n\n    logging.info(f\"Runtime \\\"{alluxio_runtime.metadata.namespace}/{alluxio_runtime.metadata.name}\\\
+          \" created successfully\")\n\n"
+        image: python:3.7
+    exec-create-s3-dataset:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - create_s3_dataset
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef create_s3_dataset(dataset_name: str, namespace: str, mount_point:\
+          \ str, mount_s3_endpoint: str, mount_s3_region: str):\n    import logging\n\
+          \    import fluid\n    from kubernetes import client\n    fluid_client =\
+          \ fluid.FluidClient()\n\n    FLUID_GROUP = \"data.fluid.io\"\n    FLUID_VERSION\
+          \ = \"v1alpha1\"\n\n    # This is an sample which use some pre-defined options.\n\
+          \    # Users can change these code customily\n    dataset = fluid.Dataset(\n\
+          \        api_version=\"%s/%s\" % (FLUID_GROUP, FLUID_VERSION),\n       \
+          \ kind=\"Dataset\",\n        metadata=client.V1ObjectMeta(\n           \
+          \ name=dataset_name,\n            namespace=namespace\n        ),\n    \
+          \    spec=fluid.DatasetSpec(\n            mounts=[\n                fluid.Mount(\n\
+          \                    mount_point=mount_point,\n                    name=dataset_name,\n\
+          \                    options={\n                        \"alluxio.underfs.s3.endpoint\"\
+          : mount_s3_endpoint,\n                        \"alluxio.underfs.s3.endpoint.region\"\
+          : mount_s3_region,\n                        \"alluxio.underfs.s3.disable.dns.buckets\"\
+          : \"true\",\n                        \"alluxio.underfs.s3.disable.inherit.acl\"\
+          : \"false\"\n                    },\n                    encrypt_options=[\n\
+          \                        {\n                            \"name\": \"aws.accessKeyId\"\
+          ,\n                            \"valueFrom\": {\n                      \
+          \        \"secretKeyRef\": {\n                                \"name\":\
+          \ \"s3-secret\",\n                                \"key\": \"aws.accessKeyId\"\
+          \n                              }\n                            }\n     \
+          \                   },\n                        {\n                    \
+          \        \"name\": \"aws.secretKey\",\n                            \"valueFrom\"\
+          : {\n                              \"secretKeyRef\": {\n               \
+          \                 \"name\": \"s3-secret\",\n                           \
+          \     \"key\": \"aws.secretKey\"\n                              }\n    \
+          \                        }\n                        }\n                \
+          \    ]\n                )\n            ]\n        )\n    )\n\n    fluid_client.create_dataset(dataset)\n\
+          \n    logging.info(f\"Dataset \\\"{dataset.metadata.namespace}/{dataset.metadata.name}\\\
+          \" created successfully\")\n\n"
+        image: python:3.7
+    exec-preheat-dataset:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - preheat_dataset
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'git+https://github.com/fluid-cloudnative/fluid-client-python.git'\
+          \ && \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef preheat_dataset(dataset_name: str, namespace: str):\n    import\
+          \ logging\n    from fluid import DataLoad, DataLoadSpec, FluidClient\n \
+          \   from kubernetes import client as k8s_client\n\n    fluid_client = FluidClient()\n\
+          \n    FLUID_GROUP = \"data.fluid.io\"\n    FLUID_VERSION = \"v1alpha1\"\n\
+          \n    dataload = DataLoad(\n        api_version=\"%s/%s\" % (FLUID_GROUP,\
+          \ FLUID_VERSION),\n        kind=\"DataLoad\",\n        metadata=k8s_client.V1ObjectMeta(\n\
+          \            name=\"%s-loader\" % dataset_name,\n            namespace=namespace\n\
+          \        ),\n        spec=DataLoadSpec(\n            dataset={\n       \
+          \         \"name\": dataset_name,\n                \"namespace\": namespace\n\
+          \            }\n        )\n    )\n\n    fluid_client.create_data_operation(data_op=dataload,\
+          \ wait=True)\n\n    logging.info(f\"Load Dataset \\\"{namespace}/{dataset_name}\\\
+          \"  successfully\")\n\n"
+        image: python:3.7
+    exec-train-simple-cnn:
+      container:
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - train_simple_cnn
+        command:
+        - sh
+        - -c
+        - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
+          \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
+          \ python3 -m pip install --quiet --no-warn-script-location 'kfp==2.4.0'\
+          \ '--no-deps' 'typing-extensions>=3.7.4,<5; python_version<\"3.9\"'  &&\
+          \  python3 -m pip install --quiet --no-warn-script-location 'pandas' &&\
+          \ \"$0\" \"$@\"\n"
+        - sh
+        - -ec
+        - 'program_path=$(mktemp -d)
+
+
+          printf "%s" "$0" > "$program_path/ephemeral_component.py"
+
+          _KFP_RUNTIME=true python3 -m kfp.dsl.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"
+
+          '
+        - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
+          \ *\n\ndef train_simple_cnn(dataset_name: str, data_root_path: str, batch_size:\
+          \ int, epochs: int, learning_rate: float):\n    import logging\n\n    logging.info(\"\
+          Training a simple CNN...\")\n\n    import os\n    import pandas as pd\n\
+          \    import torch\n    import torch.nn as nn\n    import torch.optim as\
+          \ optim\n    import torch.nn.init as init\n    from torchvision import transforms\n\
+          \    from torch.utils.data import DataLoader, Dataset, random_split   \n\
+          \n    dataset_path = os.path.join(data_root_path, dataset_name)\n    logging.info(\"\
+          Dataset file location: \" + dataset_path)\n\n    # Check Dataset Path\n\
+          \    logging.info(\"Check dataset\")\n    for dirname, _, filenames in os.walk(dataset_path):\n\
+          \        for filename in filenames:\n            logging.info(os.path.join(dirname,\
+          \ filename))\n\n    # Prepare Data\n    logging.info(\"Start load dataset...\"\
+          )\n    train_data=pd.read_csv(f\"{dataset_path}/fashion-mnist_train.csv\"\
+          )\n    test_data=pd.read_csv(f\"{dataset_path}/fashion-mnist_test.csv\"\
+          )\n    logging.info(\"Load dataset successfully!\")\n\n    class CustomDataset(Dataset):\n\
+          \n        def __init__(self,dataframe,transform=None):\n            self.dataframe=dataframe\n\
+          \            self.transform=transform\n\n\n        def __len__(self):\n\n\
+          \            return len(self.dataframe)\n\n        def __getitem__(self,idx):\n\
+          \            label = self.dataframe.iloc[idx, 0]\n            image_data\
+          \ = self.dataframe.iloc[idx, 1:].values.astype('uint8').reshape((28, 28,\
+          \ 1))\n\n            if(self.transform):\n                image=self.transform(image_data)\n\
+          \n            return image,label\n\n    transform = transforms.Compose([transforms.ToTensor()])\n\
+          \n    train_dataset=CustomDataset(train_data,transform=transform)\n    test_dataset=CustomDataset(test_data,\
+          \ transform=transform)\n\n    train_size=int(0.8*len(train_dataset))\n \
+          \   valid_size=len(train_dataset)-train_size\n\n    train_dataset,valid_dataset=random_split(train_dataset,[train_size,valid_size])\n\
+          \    train_loader=DataLoader(train_dataset,batch_size=batch_size,shuffle=True)\n\
+          \    valid_loader=DataLoader(valid_dataset,batch_size=batch_size)\n    test_loader=DataLoader(test_dataset,batch_size=batch_size)\n\
+          \n    # Build a simple CNN\n    class CNN(nn.Module):\n\n        def __init__(self,num_classes):\n\
+          \            super(CNN, self).__init__()\n            self.feature = nn.Sequential(\n\
+          \                nn.Conv2d(1,24,kernel_size=3,padding=1),\n            \
+          \    nn.ReLU(inplace=True),\n                nn.MaxPool2d(kernel_size=2),\n\
+          \                nn.Conv2d(24,128,kernel_size=3,padding=1),\n          \
+          \      nn.ReLU(inplace=True),\n                nn.MaxPool2d(kernel_size=2)\n\
+          \            )\n            self.classifier = nn.Sequential(\n         \
+          \       nn.Linear(128*7*7,48),\n                nn.ReLU(inplace=True),\n\
+          \                nn.Linear(48,num_classes)\n            )\n\n        def\
+          \ forward(self,x):\n            x = self.feature(x)\n            x = x.view(x.size(0),-1)\n\
+          \            x = self.classifier(x)\n            return x \n\n    # Train\
+          \ this CNN\n    def train(model,train_loader,optimizer,criterion, device):\n\
+          \        model.train()\n        train_loss=0\n        correct=0\n      \
+          \  total=0\n\n        for images,labels in train_loader:\n            images,labels\
+          \ =images.to(device),labels.to(device)\n\n            optimizer.zero_grad()\n\
+          \            outputs=model(images)\n            loss=criterion(outputs,labels)\n\
+          \            loss.backward()\n            optimizer.step()\n\n         \
+          \   train_loss+=loss.item()\n            _,predicted=outputs.max(1)\n  \
+          \          total+=labels.size(0)\n            correct+=predicted.eq(labels).sum().item()\n\
+          \n        train_accuracy=100*correct/total\n        train_loss/=len(train_loader)\n\
+          \        return train_loss,train_accuracy\n\n    def validate(model,valid_loader,criterion,device):\n\
+          \        model.eval()\n        val_loss=0\n        correct=0\n        total=0\n\
+          \n        with torch.no_grad():\n            for images,labels in valid_loader:\n\
+          \                images,labels=images.to(device),labels.to(device)\n\n \
+          \               outputs=model(images)\n                loss=criterion(outputs,labels)\n\
+          \n                val_loss+=loss.item()\n                _,predicted=outputs.max(1)\n\
+          \                total+=labels.size(0)\n                correct+=predicted.eq(labels).sum().item()\n\
+          \n            val_accuracy = 100.0 * correct / total\n            val_loss\
+          \ /= len(valid_loader)\n        return val_loss, val_accuracy\n\n    def\
+          \ test(model, test_loader, device):\n        model.eval()\n        correct=0\n\
+          \        total=0\n\n        with torch.no_grad():\n            for images,labels\
+          \ in test_loader:\n                images,labels=images.to(device),labels.to(device)\n\
+          \n                outputs=model(images)\n                _,predicted=outputs.max(1)\n\
+          \                total+=labels.size(0)\n                correct+=predicted.eq(labels).sum().item()\n\
+          \n            test_accuracy = 100.0 * correct / total\n\n        return\
+          \ test_accuracy\n\n    device = torch.device(\"cuda\" if torch.cuda.is_available()\
+          \ else \"cpu\")\n    cnn = CNN(10).to(device)\n\n    criterion = nn.CrossEntropyLoss()\n\
+          \    optimizer = optim.Adam(cnn.parameters(),lr=learning_rate)\n\n    train_accuracy=[]\n\
+          \    validation_accuracy=[]\n    train_losses=[]\n    validation_losses=[]\n\
+          \n    logging.info(\"Begin training...\")\n    for epoch in range(epochs):\n\
+          \        train_loss, train_acc = train(cnn, train_loader, optimizer, criterion,\
+          \ device)\n        val_loss, val_acc = validate(cnn, valid_loader, criterion,\
+          \ device)\n\n        train_accuracy.append(train_acc)\n        validation_accuracy.append(val_acc)\n\
+          \        train_losses.append(train_loss)\n        validation_losses.append(val_loss)\n\
+          \n\n        logging.info(f\"Epoch {epoch+1}/{epochs}: Train Loss: {train_loss:.4f},\
+          \ Validation Loss: {val_loss:.4f} Train Accuracy: {train_acc:.2f}%, Validation\
+          \ Accuracy: {val_acc:.2f}%\")\n\n    logging.info(\"Test the CNN...\")\n\
+          \    test_acc = test(cnn, test_loader, device)\n    logging.info(f\"Final\
+          \ Test Accuracy: {test_acc:.2f}%\")\n\n"
+        image: bitnami/pytorch:2.1.2
+pipelineInfo:
+  name: train-cnn-for-fashion-mnist
+root:
+  dag:
+    tasks:
+      cleanup-dataset-and-alluxio-runtime:
+        cachingOptions: {}
+        componentRef:
+          name: comp-cleanup-dataset-and-alluxio-runtime
+        dependentTasks:
+        - train-simple-cnn
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: cleanup-dataset-and-alluxio-runtime
+        triggerPolicy:
+          strategy: ALL_UPSTREAM_TASKS_COMPLETED
+      cleanup-preheat-operation:
+        cachingOptions: {}
+        componentRef:
+          name: comp-cleanup-preheat-operation
+        dependentTasks:
+        - train-simple-cnn
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: cleanup-preheat-operation
+        triggerPolicy:
+          strategy: ALL_UPSTREAM_TASKS_COMPLETED
+      create-alluxio-runtime:
+        cachingOptions: {}
+        componentRef:
+          name: comp-create-alluxio-runtime
+        dependentTasks:
+        - create-s3-dataset
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: create-alluxio-runtime
+      create-s3-dataset:
+        cachingOptions: {}
+        componentRef:
+          name: comp-create-s3-dataset
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            mount_point:
+              componentInputParameter: mount_point
+            mount_s3_endpoint:
+              componentInputParameter: mount_s3_endpoint
+            mount_s3_region:
+              componentInputParameter: mount_s3_region
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: create-s3-dataset
+      preheat-dataset:
+        cachingOptions: {}
+        componentRef:
+          name: comp-preheat-dataset
+        dependentTasks:
+        - create-alluxio-runtime
+        inputs:
+          parameters:
+            dataset_name:
+              componentInputParameter: dataset_name
+            namespace:
+              componentInputParameter: namespace
+        taskInfo:
+          name: preheat-dataset
+      train-simple-cnn:
+        cachingOptions: {}
+        componentRef:
+          name: comp-train-simple-cnn
+        dependentTasks:
+        - preheat-dataset
+        inputs:
+          parameters:
+            batch_size:
+              componentInputParameter: batch_size
+            data_root_path:
+              runtimeValue:
+                constant: /datasets
+            dataset_name:
+              componentInputParameter: dataset_name
+            epochs:
+              componentInputParameter: epochs
+            learning_rate:
+              componentInputParameter: learning_rate
+        taskInfo:
+          name: train-simple-cnn
+  inputDefinitions:
+    parameters:
+      batch_size:
+        parameterType: NUMBER_INTEGER
+      dataset_name:
+        parameterType: STRING
+      epochs:
+        parameterType: NUMBER_INTEGER
+      learning_rate:
+        parameterType: NUMBER_DOUBLE
+      mount_point:
+        parameterType: STRING
+      mount_s3_endpoint:
+        parameterType: STRING
+      mount_s3_region:
+        parameterType: STRING
+      namespace:
+        parameterType: STRING
+schemaVersion: 2.1.0
+sdkVersion: kfp-2.4.0
+---
+platforms:
+  kubernetes:
+    deploymentSpec:
+      executors:
+        exec-train-simple-cnn:
+          pvcMount:
+          - componentInputParameter: dataset_name
+            mountPath: /datasets
diff --git a/integration/kubeflow/rbac.yaml b/integration/kubeflow/rbac.yaml
new file mode 100644
index 00000000..31339108
--- /dev/null
+++ b/integration/kubeflow/rbac.yaml
@@ -0,0 +1,27 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: kubeflow-fluid 
+rules:
+  - apiGroups:
+      - data.fluid.io
+    resources:
+      - datasets
+      - datasets/status
+      - alluxioruntimes
+      - alluxioruntimes/status
+    verbs:
+      - '*'
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: kubeflow-fluid-clusterrolebinding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: kubeflow-fluid 
+subjects:
+  - kind: ServiceAccount
+    name: pipeline-runner
+    namespace: kubeflow
\ No newline at end of file
diff --git a/integration/kubeflow/s3-secret.yaml b/integration/kubeflow/s3-secret.yaml
new file mode 100644
index 00000000..1c656157
--- /dev/null
+++ b/integration/kubeflow/s3-secret.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: s3-secret
+  namespace: <TARGET-NAMESPACE>
+stringData:
+  aws.accessKeyId: <YOUR-ACCESS-KEY-ID>
+  aws.secretKey: <YOUR-SECRET-KEY>
\ No newline at end of file
diff --git a/integration/kubeflow/simple_cnn_sample.py b/integration/kubeflow/simple_cnn_sample.py
new file mode 100644
index 00000000..491d9f33
--- /dev/null
+++ b/integration/kubeflow/simple_cnn_sample.py
@@ -0,0 +1,246 @@
+# Copyright 2024 The Fluid Authors.
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from kfp import dsl, components, kubernetes, compiler
+
+# Load Fluid components
+create_s3_dataset = components.load_component_from_file('./component-yaml/create-s3-dataset.yaml')
+create_alluxio_runtime = components.load_component_from_file('./component-yaml/create-alluxioruntime.yaml')
+preheat_dataset = components.load_component_from_file('./component-yaml/preheat-dataset.yaml')
+cleanup_preheat_operation = components.load_component_from_file('./component-yaml/cleanup-preheat-operation.yaml')
+cleanup_dataset_and_alluxio_runtime = components.load_component_from_file('./component-yaml/cleanup-dataset-and-alluxioruntime.yaml')
+
+# The component to train a simple CNN with FashionMNIST
+# In production environment, you'd better not to use packages_to_install,
+# since the KFP SDK will install these dependencies at task runtime.
+# More details in https://www.kubeflow.org/docs/components/pipelines/v2/components/containerized-python-components/
+@dsl.component(
+    base_image='bitnami/pytorch:2.1.2',
+    packages_to_install=['pandas']
+)
+def train_simple_cnn(dataset_name: str, data_root_path: str, batch_size: int, epochs: int, learning_rate: float):
+    import logging
+
+    logging.info("Training a simple CNN...")
+
+    import os
+    import pandas as pd
+    import torch
+    import torch.nn as nn
+    import torch.optim as optim
+    import torch.nn.init as init
+    from torchvision import transforms
+    from torch.utils.data import DataLoader, Dataset, random_split   
+
+    dataset_path = os.path.join(data_root_path, dataset_name)
+    logging.info("Dataset file location: " + dataset_path)
+
+    # Check Dataset Path
+    logging.info("Check dataset")
+    for dirname, _, filenames in os.walk(dataset_path):
+        for filename in filenames:
+            logging.info(os.path.join(dirname, filename))
+    
+    # Prepare Data
+    logging.info("Start load dataset...")
+    train_data=pd.read_csv(f"{dataset_path}/fashion-mnist_train.csv")
+    test_data=pd.read_csv(f"{dataset_path}/fashion-mnist_test.csv")
+    logging.info("Load dataset successfully!")
+
+    class CustomDataset(Dataset):
+
+        def __init__(self,dataframe,transform=None):
+            self.dataframe=dataframe
+            self.transform=transform
+
+
+        def __len__(self):
+
+            return len(self.dataframe)
+    
+        def __getitem__(self,idx):
+            label = self.dataframe.iloc[idx, 0]
+            image_data = self.dataframe.iloc[idx, 1:].values.astype('uint8').reshape((28, 28, 1))
+
+            if(self.transform):
+                image=self.transform(image_data)
+
+            return image,label
+    
+    transform = transforms.Compose([transforms.ToTensor()])
+
+    train_dataset=CustomDataset(train_data,transform=transform)
+    test_dataset=CustomDataset(test_data, transform=transform)
+    
+    train_size=int(0.8*len(train_dataset))
+    valid_size=len(train_dataset)-train_size
+
+    train_dataset,valid_dataset=random_split(train_dataset,[train_size,valid_size])
+    train_loader=DataLoader(train_dataset,batch_size=batch_size,shuffle=True)
+    valid_loader=DataLoader(valid_dataset,batch_size=batch_size)
+    test_loader=DataLoader(test_dataset,batch_size=batch_size)
+    
+    # Build a simple CNN
+    class CNN(nn.Module):
+
+        def __init__(self,num_classes):
+            super(CNN, self).__init__()
+            self.feature = nn.Sequential(
+                nn.Conv2d(1,24,kernel_size=3,padding=1),
+                nn.ReLU(inplace=True),
+                nn.MaxPool2d(kernel_size=2),
+                nn.Conv2d(24,128,kernel_size=3,padding=1),
+                nn.ReLU(inplace=True),
+                nn.MaxPool2d(kernel_size=2)
+            )
+            self.classifier = nn.Sequential(
+                nn.Linear(128*7*7,48),
+                nn.ReLU(inplace=True),
+                nn.Linear(48,num_classes)
+            )
+
+        def forward(self,x):
+            x = self.feature(x)
+            x = x.view(x.size(0),-1)
+            x = self.classifier(x)
+            return x 
+    
+    # Train this CNN
+    def train(model,train_loader,optimizer,criterion, device):
+        model.train()
+        train_loss=0
+        correct=0
+        total=0
+
+        for images,labels in train_loader:
+            images,labels =images.to(device),labels.to(device)
+
+            optimizer.zero_grad()
+            outputs=model(images)
+            loss=criterion(outputs,labels)
+            loss.backward()
+            optimizer.step()
+
+            train_loss+=loss.item()
+            _,predicted=outputs.max(1)
+            total+=labels.size(0)
+            correct+=predicted.eq(labels).sum().item()
+
+        train_accuracy=100*correct/total
+        train_loss/=len(train_loader)
+        return train_loss,train_accuracy
+    
+    def validate(model,valid_loader,criterion,device):
+        model.eval()
+        val_loss=0
+        correct=0
+        total=0
+
+        with torch.no_grad():
+            for images,labels in valid_loader:
+                images,labels=images.to(device),labels.to(device)
+
+                outputs=model(images)
+                loss=criterion(outputs,labels)
+
+                val_loss+=loss.item()
+                _,predicted=outputs.max(1)
+                total+=labels.size(0)
+                correct+=predicted.eq(labels).sum().item()
+
+            val_accuracy = 100.0 * correct / total
+            val_loss /= len(valid_loader)
+        return val_loss, val_accuracy
+
+    def test(model, test_loader, device):
+        model.eval()
+        correct=0
+        total=0
+        
+        with torch.no_grad():
+            for images,labels in test_loader:
+                images,labels=images.to(device),labels.to(device)
+                
+                outputs=model(images)
+                _,predicted=outputs.max(1)
+                total+=labels.size(0)
+                correct+=predicted.eq(labels).sum().item()
+
+            test_accuracy = 100.0 * correct / total
+        
+        return test_accuracy
+    
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    cnn = CNN(10).to(device)
+
+    criterion = nn.CrossEntropyLoss()
+    optimizer = optim.Adam(cnn.parameters(),lr=learning_rate)
+
+    train_accuracy=[]
+    validation_accuracy=[]
+    train_losses=[]
+    validation_losses=[]
+
+    logging.info("Begin training...")
+    for epoch in range(epochs):
+        train_loss, train_acc = train(cnn, train_loader, optimizer, criterion, device)
+        val_loss, val_acc = validate(cnn, valid_loader, criterion, device)
+
+        train_accuracy.append(train_acc)
+        validation_accuracy.append(val_acc)
+        train_losses.append(train_loss)
+        validation_losses.append(val_loss)
+
+
+        logging.info(f"Epoch {epoch+1}/{epochs}: Train Loss: {train_loss:.4f}, Validation Loss: {val_loss:.4f} Train Accuracy: {train_acc:.2f}%, Validation Accuracy: {val_acc:.2f}%")
+    
+    logging.info("Test the CNN...")
+    test_acc = test(cnn, test_loader, device)
+    logging.info(f"Final Test Accuracy: {test_acc:.2f}%")
+
+@dsl.pipeline(name='train-cnn-for-fashion-mnist')
+def train_cnn_for_fashion_mnist_pipeline(dataset_name: str, namespace: str, mount_s3_endpoint: str, mount_s3_region: str, mount_point: str, batch_size: int, epochs: int, learning_rate: float):
+    # dataset's mount path when training
+    mount_path = '/datasets'
+    # prepare dataset
+    create_dataset_op = create_s3_dataset(dataset_name=dataset_name, namespace=namespace, mount_point=mount_point, mount_s3_endpoint=mount_s3_endpoint, mount_s3_region=mount_s3_region)
+    create_alluxio_runtime_op = create_alluxio_runtime(dataset_name=dataset_name, namespace=namespace)
+    preheat_dataset_op = preheat_dataset(dataset_name=dataset_name, namespace=namespace)
+    # disable caching
+    create_dataset_op.set_caching_options(False)
+    create_alluxio_runtime_op.set_caching_options(False)
+    preheat_dataset_op.set_caching_options(False)
+    # train cnn
+    train_simple_cnn_op = train_simple_cnn(dataset_name=dataset_name, data_root_path=mount_path, batch_size=batch_size, epochs=epochs, learning_rate=learning_rate)
+    train_simple_cnn_op.set_caching_options(False)
+    # mount dataset pvc to training component
+    kubernetes.mount_pvc(task=train_simple_cnn_op, pvc_name=dataset_name, mount_path=mount_path)
+    # define components' dependence
+    create_alluxio_runtime_op.after(create_dataset_op)
+    preheat_dataset_op.after(create_alluxio_runtime_op)
+    train_simple_cnn_op.after(preheat_dataset_op)
+    # cleanup dataset and preheat operation
+    cleanup_dataset_and_alluxio_runtime_op = cleanup_dataset_and_alluxio_runtime(dataset_name=dataset_name, namespace=namespace)
+    cleanup_dataset_and_alluxio_runtime_op.set_caching_options(False)
+    cleanup_dataset_and_alluxio_runtime_op.after(train_simple_cnn_op)
+    cleanup_dataset_and_alluxio_runtime_op.ignore_upstream_failure()
+    cleanup_preheat_operation_op = cleanup_preheat_operation(dataset_name=dataset_name, namespace=namespace)
+    cleanup_preheat_operation_op.set_caching_options(False)
+    cleanup_preheat_operation_op.after(train_simple_cnn_op)
+    cleanup_preheat_operation_op.ignore_upstream_failure()
+    
+
+# Compile into IR YAML
+# Re-run this file to re-generate the yaml file when you changed the code above.
+compiler.Compiler().compile(train_cnn_for_fashion_mnist_pipeline, './pipline-yaml/train-cnn-for-fashion-mnist-pipline.yaml')
\ No newline at end of file
diff --git a/pkg/csi/plugins/nodeserver.go b/pkg/csi/plugins/nodeserver.go
index 7b871571..3363a0ec 100644
--- a/pkg/csi/plugins/nodeserver.go
+++ b/pkg/csi/plugins/nodeserver.go
@@ -161,7 +161,10 @@ func (ns *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 	} else {
 		args = append(args, mountPath, targetPath)
 	}
-	command := exec.Command("mount", args...)
+	command, err := utils.SimpleCommand("mount", args...)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
 
 	glog.V(3).Infof("NodePublishVolume: exec command %v", command)
 	stdoutStderr, err := command.CombinedOutput()
@@ -508,7 +511,10 @@ func checkMountInUse(volumeName string) (bool, error) {
 
 	// TODO: refer to https://github.com/kubernetes-sigs/alibaba-cloud-csi-driver/blob/4fcb743220371de82d556ab0b67b08440b04a218/pkg/oss/utils.go#L72
 	// for a better implementation
-	command := exec.Command("/usr/local/bin/check_bind_mounts.sh", volumeName)
+	command, err := utils.SimpleCommand("/usr/local/bin/check_bind_mounts.sh", volumeName)
+	if err != nil {
+		return inUse, err
+	}
 	glog.Infoln(command)
 
 	stdoutStderr, err := command.CombinedOutput()
diff --git a/pkg/ddc/alluxio/operations/base.go b/pkg/ddc/alluxio/operations/base.go
index 111a7a8e..9fa4107a 100644
--- a/pkg/ddc/alluxio/operations/base.go
+++ b/pkg/ddc/alluxio/operations/base.go
@@ -515,6 +515,11 @@ func (a AlluxioFileUtils) exec(command []string, verbose bool) (stdout string, s
 
 // execWithoutTimeout
 func (a AlluxioFileUtils) execWithoutTimeout(command []string, verbose bool) (stdout string, stderr string, err error) {
+	err = utils.ValidateCommandSlice(command)
+	if err != nil {
+		return
+	}
+
 	stdout, stderr, err = kubeclient.ExecCommandInContainer(a.podName, a.container, a.namespace, command)
 	if err != nil {
 		a.log.Info("Stdout", "Command", command, "Stdout", stdout)
diff --git a/pkg/ddc/goosefs/operations/base.go b/pkg/ddc/goosefs/operations/base.go
index dfdb5183..905fb01b 100644
--- a/pkg/ddc/goosefs/operations/base.go
+++ b/pkg/ddc/goosefs/operations/base.go
@@ -25,6 +25,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
 	"github.com/go-logr/logr"
 )
@@ -501,6 +502,11 @@ func (a GooseFSFileUtils) exec(command []string, verbose bool) (stdout string, s
 
 // execWithoutTimeout
 func (a GooseFSFileUtils) execWithoutTimeout(command []string, verbose bool) (stdout string, stderr string, err error) {
+	err = utils.ValidateCommandSlice(command)
+	if err != nil {
+		return
+	}
+
 	stdout, stderr, err = kubeclient.ExecCommandInContainer(a.podName, a.container, a.namespace, command)
 	if err != nil {
 		a.log.Info("Stdout", "Command", command, "Stdout", stdout)
diff --git a/pkg/ddc/jindo/operations/base.go b/pkg/ddc/jindo/operations/base.go
index 70ee279f..f4419f0c 100644
--- a/pkg/ddc/jindo/operations/base.go
+++ b/pkg/ddc/jindo/operations/base.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
 	"github.com/go-logr/logr"
 )
@@ -66,6 +67,11 @@ func (a JindoFileUtils) exec(command []string, verbose bool) (stdout string, std
 
 // execWithoutTimeout
 func (a JindoFileUtils) execWithoutTimeout(command []string, verbose bool) (stdout string, stderr string, err error) {
+	err = utils.ValidateCommandSlice(command)
+	if err != nil {
+		return
+	}
+
 	stdout, stderr, err = kubeclient.ExecCommandInContainer(a.podName, a.container, a.namespace, command)
 	if err != nil {
 		a.log.Info("Stdout", "Command", command, "Stdout", stdout)
diff --git a/pkg/ddc/jindofsx/operations/base.go b/pkg/ddc/jindofsx/operations/base.go
index 1d19bfe7..28051041 100644
--- a/pkg/ddc/jindofsx/operations/base.go
+++ b/pkg/ddc/jindofsx/operations/base.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
 	"github.com/go-logr/logr"
 )
@@ -87,6 +88,11 @@ func (a JindoFileUtils) execWithTimeOut(command []string, verbose bool, second i
 
 // execWithoutTimeout
 func (a JindoFileUtils) execWithoutTimeout(command []string, verbose bool) (stdout string, stderr string, err error) {
+	err = utils.ValidateCommandSlice(command)
+	if err != nil {
+		return
+	}
+
 	stdout, stderr, err = kubeclient.ExecCommandInContainer(a.podName, a.container, a.namespace, command)
 	if err != nil {
 		a.log.Info("Stdout", "Command", command, "Stdout", stdout)
diff --git a/pkg/ddc/juicefs/operations/base.go b/pkg/ddc/juicefs/operations/base.go
index 5e50b488..d32a8850 100644
--- a/pkg/ddc/juicefs/operations/base.go
+++ b/pkg/ddc/juicefs/operations/base.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/go-logr/logr"
 
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
 )
 
@@ -325,6 +326,12 @@ func (j JuiceFileUtils) exec(command []string) (stdout string, stderr string, er
 
 // execWithoutTimeout
 func (j JuiceFileUtils) execWithoutTimeout(command []string) (stdout string, stderr string, err error) {
+	// validate the pipe command with white list
+	err = utils.ValidateCommandSlice(command)
+	if err != nil {
+		return
+	}
+
 	stdout, stderr, err = kubeclient.ExecCommandInContainer(j.podName, j.container, j.namespace, command)
 	if err != nil {
 		j.log.Info("Stdout", "Command", command, "Stdout", stdout)
diff --git a/pkg/ddc/thin/operations/base.go b/pkg/ddc/thin/operations/base.go
index 9359afc5..21b4c70f 100644
--- a/pkg/ddc/thin/operations/base.go
+++ b/pkg/ddc/thin/operations/base.go
@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
 	"github.com/go-logr/logr"
 )
@@ -155,6 +156,12 @@ func (t ThinFileUtils) exec(command []string, verbose bool) (stdout string, stde
 
 // execWithoutTimeout
 func (t ThinFileUtils) execWithoutTimeout(command []string, verbose bool) (stdout string, stderr string, err error) {
+	// validate the pipe command with white list
+	err = utils.ValidateCommandSlice(command)
+	if err != nil {
+		return
+	}
+
 	stdout, stderr, err = kubeclient.ExecCommandInContainer(t.podName, t.container, t.namespace, command)
 	if err != nil {
 		t.log.Info("Stdout", "Command", command, "Stdout", stdout)
diff --git a/pkg/ddc/vineyard/operations/base.go b/pkg/ddc/vineyard/operations/base.go
index a445560e..cab367de 100644
--- a/pkg/ddc/vineyard/operations/base.go
+++ b/pkg/ddc/vineyard/operations/base.go
@@ -15,10 +15,15 @@ package operations
 
 import (
 	"fmt"
-	"io/ioutil"
+	"io"
+	"net"
 	"net/http"
+	"net/url"
+	"strconv"
+	"time"
 
 	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/util/validation"
 )
 
 type VineyardFileUtils struct {
@@ -39,23 +44,44 @@ func NewVineyardFileUtils(podNamePrefix string, port int32, replicas int32, name
 	}
 }
 
+func validateResourceName(name string) error {
+	if errs := validation.IsDNS1035Label(name); len(errs) > 0 {
+		return fmt.Errorf("invalid DNS-1035 label: %s", name)
+	}
+	return nil
+}
+
 // Get summary info of the Vineyard Engine
 func (a VineyardFileUtils) ReportSummary() (summary []string, err error) {
 	var resp *http.Response
 	var body []byte
+	if err := validateResourceName(a.podNamePrefix); err != nil {
+		return nil, err
+	}
+	if err := validateResourceName(a.namespace); err != nil {
+		return nil, err
+	}
+	client := &http.Client{
+		Timeout: 20 * time.Second,
+	}
 	for i := int32(0); i < a.replicas; i++ {
-		url := fmt.Sprintf("http://%s-%d.%s-svc.%s.svc.cluster.local:%d/metrics", a.podNamePrefix, i, a.podNamePrefix, a.namespace, a.port)
+		u := url.URL{
+			Scheme: "http",
+			Host: net.JoinHostPort(fmt.Sprintf("%s-%d.%s-svc.%s.svc", a.podNamePrefix, i,
+				a.podNamePrefix, a.namespace), strconv.Itoa(int(a.port))),
+			Path: "/metrics",
+		}
 
-		resp, err = http.Get(url)
+		resp, err = client.Get(u.String())
 		if err != nil {
-			err = fmt.Errorf("failed to get metrics from %s, error: %v", url, err)
+			err = fmt.Errorf("failed to get metrics from %s, error: %v", u.String(), err)
 			return summary, err
 		}
 		defer resp.Body.Close()
 
-		body, err = ioutil.ReadAll(resp.Body)
+		body, err = io.ReadAll(resp.Body)
 		if err != nil {
-			err = fmt.Errorf("failed to read response body from %s, error: %v", url, err)
+			err = fmt.Errorf("failed to read response body from %s, error: %v", u.String(), err)
 			return
 		}
 
diff --git a/pkg/utils/exec.go b/pkg/utils/exec.go
new file mode 100644
index 00000000..ae05eec1
--- /dev/null
+++ b/pkg/utils/exec.go
@@ -0,0 +1,86 @@
+/*
+Copyright 2024 The Fluid Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"fmt"
+	"os/exec"
+	"path/filepath"
+	"strings"
+)
+
+func init() {
+	allowPathlist = buildPathList(allowPathlist)
+}
+
+// allowPathlist of safe commands
+var allowPathlist = map[string]bool{
+	// "helm":    true,
+	"kubectl":  true,
+	"ddc-helm": true,
+	// add other commands as needed
+}
+
+// illegalChars to check
+// var illegalChars = []string{"&", "|", ";", "$", "'", "`", "(", ")", ">>"}
+var illegalChars = []rune{'&', '|', ';', '$', '\'', '`', '(', ')', '>'}
+
+// buildPathList is a function that builds a map of paths for the given pathList.
+func buildPathList(pathList map[string]bool) (targetPath map[string]bool) {
+	targetPath = make(map[string]bool)
+	for name, enabled := range pathList {
+		if filepath.Base(name) == name {
+			path, err := exec.LookPath(name)
+			if err != nil {
+				log.Info("Failed to find path %s due to %v", path, err)
+
+			} else {
+				targetPath[path] = enabled
+			}
+		}
+		targetPath[name] = enabled
+	}
+	return
+}
+
+// SimpleCommand checks the args before creating *exec.Cmd
+func SimpleCommand(name string, arg ...string) (cmd *exec.Cmd, err error) {
+	if allowPathlist[name] {
+		cmd = exec.Command(name, arg...)
+	} else {
+		err = checkCommandArgs(arg...)
+		if err != nil {
+			return
+		}
+		cmd = exec.Command(name, arg...)
+	}
+
+	return
+}
+
+// CheckCommandArgs is check string is valid in args
+func checkCommandArgs(arg ...string) (err error) {
+	for _, value := range arg {
+		for _, illegalChar := range illegalChars {
+			if strings.ContainsRune(value, illegalChar) {
+				return fmt.Errorf("args %s has illegal access with illegalChar %c", value, illegalChar)
+			}
+		}
+	}
+
+	return
+}
diff --git a/pkg/utils/helm/helm.go b/pkg/utils/helm/helm.go
index 4f8d5acb..3be00a45 100644
--- a/pkg/utils/helm/helm.go
+++ b/pkg/utils/helm/helm.go
@@ -54,64 +54,6 @@ func GenerateValueFile(values interface{}) (valueFileName string, err error) {
 	return valueFileName, err
 }
 
-// GenerateHelmTemplate generates helm template without tiller: helm template -f values.yaml chart_name
-// Exec /usr/local/bin/helm, [template -f /tmp/values313606961 --namespace default --name hj /charts/tf-horovod]
-// returns generated template file: templateFileName
-func GenerateHelmTemplate(name string, namespace string, valueFileName string, chartName string, options ...string) (templateFileName string, err error) {
-	tempName := fmt.Sprintf("%s.yaml", name)
-	templateFile, err := os.CreateTemp("", tempName)
-	if err != nil {
-		return templateFileName, err
-	}
-	templateFileName = templateFile.Name()
-
-	binary, err := exec.LookPath(helmCmd[0])
-	if err != nil {
-		return templateFileName, err
-	}
-
-	// 3. check if the chart file exists
-	// if strings.HasPrefix(chartName, "/") {
-	if _, err = os.Stat(chartName); err != nil {
-		return templateFileName, err
-	}
-	// }
-
-	// 4. prepare the arguments
-	args := []string{binary, "template", "-f", valueFileName,
-		"--namespace", namespace,
-		"--name", name,
-	}
-	if len(options) != 0 {
-		args = append(args, options...)
-	}
-
-	args = append(args, []string{chartName, ">", templateFileName}...)
-
-	log.V(1).Info("Exec bash -c ", "cmd", args)
-
-	// return syscall.Exec(cmd, args, env)
-	// 5. execute the command
-	log.V(1).Info("Generating template", "args", args)
-	cmd := exec.Command("bash", "-c", strings.Join(args, " "))
-	// cmd.Env = env
-	out, err := cmd.CombinedOutput()
-	fmt.Printf("%s", string(out))
-	if err != nil {
-		return templateFileName, fmt.Errorf("failed to execute %s, %v with %v", binary, args, err)
-	}
-
-	// // 6. clean up the value file if needed
-	// if log.GetLevel() != log.DebugLevel {
-	// 	err = os.Remove(valueFileName)
-	// 	if err != nil {
-	// 		log.Warnf("Failed to delete %s due to %v", valueFileName, err)
-	// 	}
-	// }
-
-	return templateFileName, nil
-}
-
 // GetChartVersion checks the chart version by given the chart directory
 // helm inspect chart /charts/tf-horovod
 func GetChartVersion(chart string) (version string, err error) {
@@ -132,7 +74,11 @@ func GetChartVersion(chart string) (version string, err error) {
 		"|", "grep", "version:"}
 	log.V(1).Info("Exec bash -c", "args", args)
 
-	cmd := exec.Command("bash", "-c", strings.Join(args, " "))
+	// cmd := exec.Command("bash", "-c", strings.Join(args, " "))
+	cmd, err := utils.PipeCommand("bash", "-c", strings.Join(args, " "))
+	if err != nil {
+		return "", err
+	}
 	out, err := cmd.Output()
 	if err != nil {
 		return "", err
diff --git a/pkg/utils/helm/utils.go b/pkg/utils/helm/utils.go
index ae25b393..2e853554 100644
--- a/pkg/utils/helm/utils.go
+++ b/pkg/utils/helm/utils.go
@@ -55,7 +55,10 @@ func InstallRelease(name string, namespace string, valueFile string, chartName s
 
 	// return syscall.Exec(cmd, args, env)
 	// 5. execute the command
-	cmd := exec.Command(binary, args...)
+	cmd, err := utils.SimpleCommand(binary, args...)
+	if err != nil {
+		return err
+	}
 	log.Info("Exec", "command", cmd.String())
 	// cmd.Env = env
 	out, err := cmd.CombinedOutput()
@@ -82,7 +85,10 @@ func CheckRelease(name, namespace string) (exist bool, err error) {
 		return exist, err
 	}
 
-	cmd := exec.Command(helmCmd[0], "status", name, "-n", namespace)
+	cmd, err := utils.SimpleCommand(helmCmd[0], "status", name, "-n", namespace)
+	if err != nil {
+		return exist, err
+	}
 	// support multiple cluster management
 	// if types.KubeConfig != "" {
 	// 	cmd.Env = append(cmd.Env, fmt.Sprintf("KUBECONFIG=%s", types.KubeConfig))
@@ -147,7 +153,10 @@ func DeleteRelease(name, namespace string) error {
 	}
 
 	args := []string{"uninstall", name, "-n", namespace}
-	cmd := exec.Command(binary, args...)
+	cmd, err := utils.SimpleCommand(binary, args...)
+	if err != nil {
+		return err
+	}
 	log.Info("Exec", "command", cmd.String())
 	// env := os.Environ()
 	// if types.KubeConfig != "" {
@@ -171,7 +180,10 @@ func ListReleases(namespace string) (releases []string, err error) {
 		return releases, err
 	}
 
-	cmd := exec.Command(helmCmd[0], "list", "-q", "-n", namespace)
+	cmd, err := utils.SimpleCommand(helmCmd[0], "list", "-q", "-n", namespace)
+	if err != nil {
+		return releases, err
+	}
 	// support multiple cluster management
 	// if types.KubeConfig != "" {
 	// 	cmd.Env = append(cmd.Env, fmt.Sprintf("KUBECONFIG=%s", types.KubeConfig))
@@ -191,7 +203,10 @@ func ListReleaseMap(namespace string) (releaseMap map[string]string, err error)
 		return releaseMap, err
 	}
 
-	cmd := exec.Command(helmCmd[0], "list", "-n", namespace)
+	cmd, err := utils.SimpleCommand(helmCmd[0], "list", "-n", namespace)
+	if err != nil {
+		return releaseMap, err
+	}
 	// // support multiple cluster management
 	// if types.KubeConfig != "" {
 	// 	cmd.Env = append(cmd.Env, fmt.Sprintf("KUBECONFIG=%s", types.KubeConfig))
@@ -225,7 +240,10 @@ func ListAllReleasesWithDetail(namespace string) (releaseMap map[string][]string
 		return releaseMap, err
 	}
 
-	cmd := exec.Command(helmCmd[0], "list", "--all", "-n", namespace)
+	cmd, err := utils.SimpleCommand(helmCmd[0], "list", "--all", "-n", namespace)
+	if err != nil {
+		return releaseMap, err
+	}
 	// support multiple cluster management
 	// if types.KubeConfig != "" {
 	// 	cmd.Env = append(cmd.Env, fmt.Sprintf("KUBECONFIG=%s", types.KubeConfig))
diff --git a/pkg/utils/home.go b/pkg/utils/home.go
index f346dc87..0c557e52 100644
--- a/pkg/utils/home.go
+++ b/pkg/utils/home.go
@@ -20,7 +20,6 @@ import (
 	"bytes"
 	"errors"
 	"os"
-	"os/exec"
 	"os/user"
 	"runtime"
 	"strings"
@@ -45,26 +44,32 @@ func Home() (string, error) {
 	return homeUnix()
 }
 
-func homeUnix() (string, error) {
+func homeUnix() (home string, err error) {
 	// First prefer the HOME environmental variable
-	if home := os.Getenv("HOME"); home != "" {
+	home = os.Getenv("HOME")
+	if home != "" {
 		return home, nil
 	}
 
 	// If that fails, try the shell
 	var stdout bytes.Buffer
-	cmd := exec.Command("sh", "-c", "eval echo ~$USER")
+	cmd, err := SimpleCommand("sh", "-c", "eval echo ~$USER")
+	if err != nil {
+		return
+	}
 	cmd.Stdout = &stdout
-	if err := cmd.Run(); err != nil {
-		return "", err
+	err = cmd.Run()
+	if err != nil {
+		return
 	}
 
-	result := strings.TrimSpace(stdout.String())
-	if result == "" {
-		return "", errors.New("blank output when reading home directory")
+	home = strings.TrimSpace(stdout.String())
+	if home == "" {
+		err = errors.New("blank output when reading home directory")
+		return
 	}
 
-	return result, nil
+	return
 }
 
 func homeWindows() (string, error) {
diff --git a/pkg/utils/kubectl/configmap.go b/pkg/utils/kubectl/configmap.go
index ecce9835..50386499 100644
--- a/pkg/utils/kubectl/configmap.go
+++ b/pkg/utils/kubectl/configmap.go
@@ -21,6 +21,7 @@ import (
 	"os/exec"
 	"strings"
 
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/go-logr/logr"
 	ctrl "sigs.k8s.io/controller-runtime"
 )
@@ -51,41 +52,6 @@ func CreateConfigMapFromFile(name string, key, fileName string, namespace string
 	return
 }
 
-// SaveConfigMapToFile saves the key of configMap into a file
-func SaveConfigMapToFile(name string, key string, namespace string) (fileName string, err error) {
-	binary, err := exec.LookPath(kubectlCmd[0])
-	if err != nil {
-		return "", err
-	}
-
-	file, err := os.CreateTemp(os.TempDir(), name)
-	if err != nil {
-		log.Error(err, "failed to create tmp file", "tmpFile", file.Name())
-		return fileName, err
-	}
-	fileName = file.Name()
-
-	args := []string{binary, "get", "configmap", name,
-		"--namespace", namespace,
-		fmt.Sprintf("-o=jsonpath='{.data.%s}'", key),
-		">", fileName}
-
-	log.V(1).Info("exec", "cmd", strings.Join(args, " "))
-
-	cmd := exec.Command("bash", "-c", strings.Join(args, " "))
-	// env := os.Environ()
-	// if types.KubeConfig != "" {
-	// 	env = append(env, fmt.Sprintf("KUBECONFIG=%s", types.KubeConfig))
-	// }
-	out, err := cmd.Output()
-	fmt.Printf("%s", string(out))
-
-	if err != nil {
-		return fileName, fmt.Errorf("failed to execute %s, %v with %v", "kubectl", args, err)
-	}
-	return fileName, err
-}
-
 // kubectl executes command with arguments (string array)
 func kubectl(args []string) ([]byte, error) {
 	binary, err := exec.LookPath(kubectlCmd[0])
@@ -103,7 +69,10 @@ func kubectl(args []string) ([]byte, error) {
 
 	// return syscall.Exec(cmd, args, env)
 	// 2. execute the command
-	cmd := exec.Command(binary, args...)
+	cmd, err := utils.SimpleCommand(binary, args...)
+	if err != nil {
+		return nil, err
+	}
 	// cmd.Env = env
 	return cmd.CombinedOutput()
 }
diff --git a/pkg/utils/mount.go b/pkg/utils/mount.go
index 51a33387..480a8300 100644
--- a/pkg/utils/mount.go
+++ b/pkg/utils/mount.go
@@ -41,13 +41,16 @@ func GetMountRoot() (string, error) {
 	return mountRoot, nil
 }
 
-func CheckMountReadyAndSubPathExist(fluidPath string, mountType string, subPath string) error {
+func CheckMountReadyAndSubPathExist(fluidPath string, mountType string, subPath string) (err error) {
 	glog.Infof("Try to check if the mount target %s is ready", fluidPath)
 	if fluidPath == "" {
 		return errors.New("target is not specified for checking the mount")
 	}
 	args := []string{fluidPath, mountType, subPath}
-	command := exec.Command("/usr/local/bin/check_mount.sh", args...)
+	command, err := SimpleCommand("/usr/local/bin/check_mount.sh", args...)
+	if err != nil {
+		return
+	}
 	glog.Infoln(command)
 	stdoutStderr, err := command.CombinedOutput()
 	glog.Infoln(string(stdoutStderr))
diff --git a/pkg/utils/shell_pipes.go b/pkg/utils/shell_pipes.go
new file mode 100644
index 00000000..f79f016a
--- /dev/null
+++ b/pkg/utils/shell_pipes.go
@@ -0,0 +1,156 @@
+/*
+Copyright 2024 The Fluid Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"fmt"
+	"os/exec"
+	"strings"
+)
+
+func PipeCommand(name string, arg ...string) (cmd *exec.Cmd, err error) {
+	// prepare the slice for ValidatePipeCommandSlice
+	var commands []string
+	commands = append(commands, name)
+	commands = append(commands, arg...)
+
+	// validate commands
+	err = ValidatePipeCommandSlice(commands)
+	if err != nil {
+		return nil, err
+	}
+
+	return exec.Command(name, arg...), nil
+}
+
+// ValidateCommandSlice takes in a slice of shell commands and returns an error if any are invalid.
+// The function looks specifically for pipe commands (i.e., commands that contain a '|').
+// If a pipe command is found in the slice, ValidatePipeCommandSlice is called for further validation.
+func ValidateCommandSlice(shellCommandSlice []string) (err error) {
+	isPossiblePipeCommand := false
+
+	for _, command := range shellCommandSlice {
+		if strings.Contains(command, "|") {
+			isPossiblePipeCommand = true
+		}
+	}
+
+	if isPossiblePipeCommand {
+		err = ValidatePipeCommandSlice(shellCommandSlice)
+	} // else {
+	// 	// Todo: need handle no PossiblePipeCommand
+	// }
+	return
+}
+
+func ValidatePipeCommandSlice(shellCommandSlice []string) (err error) {
+	// Make sure the shell command is allowed
+	var AllowedShellCommands = map[string]bool{
+		"bash -c": true,
+		"sh -c":   true,
+	}
+
+	// check if shellCommandSlice has enough arguments
+	if len(shellCommandSlice) < 3 {
+		return fmt.Errorf("insufficient arguments. Expected at least 3, received %d", len(shellCommandSlice))
+	}
+	// We assume -c always directly follows the shell command
+	shellCommand := strings.Join(strings.Fields(shellCommandSlice[0]+" "+shellCommandSlice[1]), " ")
+	if _, ok := AllowedShellCommands[shellCommand]; !ok {
+		return fmt.Errorf("unknown shell command: %s", shellCommand)
+	}
+
+	for _, command := range shellCommandSlice[2:] {
+		if err := ValidateShellPipeString(command); err != nil {
+			return err
+		}
+	}
+	return
+}
+
+// ValidateShellPipeString function checks whether the input command string is safe to execute.
+// It checks whether all parts of a pipeline command start with any command prefixes defined in AllowedCommands
+// It also checks for any illegal sequences that may lead to command injection attack.
+func ValidateShellPipeString(command string) error {
+	// Define illegal sequences that may lead to command injection attack
+	illegalSequences := []string{"&", ";", "$", "'", "`", "(", ")", "||", ">>"}
+	// Separate parts of pipeline command
+	pipelineCommands := strings.Split(command, "|")
+
+	// AllowedCommands is a global map that contains all allowed command prefixes.
+	var AllowedCommands = map[string]bool{
+		"ls":      false,
+		"df":      false,
+		"mount":   false,
+		"alluxio": false,
+		"goosefs": false,
+		"kubectl": false,
+		"helm":    false,
+	}
+
+	// AllowedPipeCommands is a map that contains all allowed pipe command prefixes.
+	var allowedPipeCommands = map[string]bool{
+		"grep":  false, // false means partial match
+		"wc -l": true,  // true means full match (wc -l is exactly the allowed command)
+		// Add more commands as you see fit
+	}
+
+	// Check each part of pipeline command
+	for i, cmd := range pipelineCommands {
+		cmd = strings.Join(
+			strings.Fields(
+				strings.TrimSpace(cmd)), " ")
+
+		if i > 0 {
+			// Check whether command starts with any allowed command prefix
+			validCmd := isValidCommand(cmd, allowedPipeCommands)
+
+			// If none of the allowed command prefix is found, throw error
+			if !validCmd {
+				return fmt.Errorf("full pipeline command not supported: part %d contains unsupported command '%s', the whole command %s", i+1, cmd, command)
+			}
+		} else {
+			validCmd := isValidCommand(cmd, AllowedCommands)
+			// If none of the allowed command prefix is found, throw error
+			if !validCmd {
+				return fmt.Errorf("full pipeline command not supported: part %d contains unsupported command '%s', the whole command %s", i+1, cmd, command)
+			}
+		}
+
+		// Check for illegal sequences in command
+		for _, illegalSeq := range illegalSequences {
+			if strings.Contains(cmd, illegalSeq) {
+				return fmt.Errorf("unsafe pipeline command %s, illegal sequence detected: %s in part %d: '%s'", command, illegalSeq, i+1, cmd)
+			}
+		}
+	}
+
+	// If no error found, return nil
+	return nil
+}
+
+// Defining a function to check if the command is valid
+func isValidCommand(cmd string, allowedCommands map[string]bool) bool {
+	for cmdPrefix, exactMatch := range allowedCommands {
+		if exactMatch && cmd == cmdPrefix {
+			return true
+		} else if !exactMatch && strings.HasPrefix(cmd, cmdPrefix) {
+			return true
+		}
+	}
+	return false
+}


Test patch:
diff --git a/pkg/ddc/goosefs/hcfs_test.go b/pkg/ddc/goosefs/hcfs_test.go
index 40a0ffb0..66a44aab 100644
--- a/pkg/ddc/goosefs/hcfs_test.go
+++ b/pkg/ddc/goosefs/hcfs_test.go
@@ -97,7 +97,10 @@ func TestGetHCFSStatus(t *testing.T) {
 		t.Fatal(err.Error())
 	}
 	engine := newGooseFSEngineHCFS(fakeClient, "hbase", "fluid")
-	out, _ := engine.GetHCFSStatus()
+	out, err := engine.GetHCFSStatus()
+	if err != nil {
+		t.Fatal(err.Error())
+	}
 	wrappedUnhook()
 	status := &v1alpha1.HCFSStatus{
 		Endpoint:                    "goosefs://hbase-master-0.fluid:2333",
diff --git a/pkg/ddc/juicefs/data_load_test.go b/pkg/ddc/juicefs/data_load_test.go
index d6929fcf..88f4a13d 100644
--- a/pkg/ddc/juicefs/data_load_test.go
+++ b/pkg/ddc/juicefs/data_load_test.go
@@ -18,13 +18,14 @@ package juicefs
 
 import (
 	"errors"
-	"github.com/fluid-cloudnative/fluid/pkg/common"
 	"os"
 	"path/filepath"
 	"reflect"
 	"strings"
 	"testing"
 
+	"github.com/fluid-cloudnative/fluid/pkg/common"
+
 	"github.com/brahma-adshonor/gohook"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
@@ -621,7 +622,7 @@ func TestJuiceFSEngine_CheckExistenceOfPath(t *testing.T) {
 	}
 	notExist, err = engine.CheckExistenceOfPath(targetDataload)
 	if !(err == nil && notExist == false) {
-		t.Errorf("fail to exec the function")
+		t.Errorf("fail to exec the function due to %v", err)
 	}
 	wrappedUnhook()
 }
diff --git a/pkg/ddc/vineyard/operations/base_test.go b/pkg/ddc/vineyard/operations/base_test.go
index 4b8e4ba9..274b9d05 100644
--- a/pkg/ddc/vineyard/operations/base_test.go
+++ b/pkg/ddc/vineyard/operations/base_test.go
@@ -19,32 +19,33 @@ import (
 	"net/http"
 	"reflect"
 	"testing"
+	"time"
 
-	"github.com/brahma-adshonor/gohook"
+	. "github.com/agiledragon/gomonkey/v2"
 	"github.com/go-logr/logr"
 )
 
-func mockHttpGet(url string) (*http.Response, error) {
-	if url == "http://vineyard-0.vineyard-svc.default.svc.cluster.local:8080/metrics" {
-		return &http.Response{
-			StatusCode: http.StatusOK,
-			Body:       ioutil.NopCloser(bytes.NewBufferString("metric for pod 1")),
-		}, nil
-	}
-	if url == "http://vineyard-1.vineyard-svc.default.svc.cluster.local:8080/metrics" {
-		return &http.Response{
-			StatusCode: http.StatusOK,
-			Body:       ioutil.NopCloser(bytes.NewBufferString("metric for pod 2")),
-		}, nil
-	}
-	return nil, nil
-}
-
 func TestVineyardFileUtils_ReportSummary(t *testing.T) {
-	err := gohook.Hook(http.Get, mockHttpGet, nil)
-	if err != nil {
-		t.Fatalf("could not hook http.Get: %v", err)
+	client := &http.Client{
+		Timeout: 20 * time.Second,
 	}
+	patch := ApplyMethod(reflect.TypeOf(client), "Get",
+		func(_ *http.Client, url string) (resp *http.Response, err error) {
+			if url == "http://vineyard-0.vineyard-svc.default.svc:8080/metrics" {
+				return &http.Response{
+					StatusCode: http.StatusOK,
+					Body:       ioutil.NopCloser(bytes.NewBufferString("metric for pod 1")),
+				}, nil
+			}
+			if url == "http://vineyard-1.vineyard-svc.default.svc:8080/metrics" {
+				return &http.Response{
+					StatusCode: http.StatusOK,
+					Body:       ioutil.NopCloser(bytes.NewBufferString("metric for pod 2")),
+				}, nil
+			}
+			return nil, nil
+		})
+	defer patch.Reset()
 
 	podNamePrefix := "vineyard"
 	port := int32(8080)
@@ -59,8 +60,11 @@ func TestVineyardFileUtils_ReportSummary(t *testing.T) {
 		t.Errorf("VineyardFileUtils.ReportSummary() got = %v, want %v, err = %v", got, expected, err)
 	}
 
-	err = gohook.UnHook(http.Get)
-	if err != nil {
-		t.Fatalf("could not unhook http.Get: %v", err)
+	// test the potenial risk args
+	podNamePrefix = "curl -d @/etc/passwd 127.0.0.1 -o-"
+	vineyard = NewVineyardFileUtils(podNamePrefix, port, replicas, namespace, mockLogger)
+	_, err = vineyard.ReportSummary()
+	if err == nil {
+		t.Errorf("Expect error, get nil. as the string %s can't pass the regex validation. err = %v", podNamePrefix, err)
 	}
 }
diff --git a/pkg/utils/exec_test.go b/pkg/utils/exec_test.go
new file mode 100644
index 00000000..036cb4a6
--- /dev/null
+++ b/pkg/utils/exec_test.go
@@ -0,0 +1,188 @@
+/*
+Copyright 2024 The Fluid Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"fmt"
+	"os/exec"
+	"reflect"
+	"testing"
+
+	"github.com/brahma-adshonor/gohook"
+)
+
+func TestCheckCommandArgs(t *testing.T) {
+	type args struct {
+		arg []string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "Test with illegal arguments",
+			args: args{
+				arg: []string{"ls", "|", "grep go"},
+			},
+			wantErr: true,
+		},
+		{
+			name: "Test with legal arguments",
+			args: args{
+				arg: []string{"ls"},
+			},
+			wantErr: false,
+		}, {
+			name: "Test with legal arguments2",
+			args: args{
+				arg: []string{"echo test > /dev/null"},
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := checkCommandArgs(tt.args.arg...); (err != nil) != tt.wantErr {
+				t.Errorf("checkCommandArgs() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestSimpleCommand(t *testing.T) {
+	type args struct {
+		name string
+		arg  []string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantCmd *exec.Cmd
+		wantErr bool
+	}{
+		{
+			name: "Allow path list",
+			args: args{
+				name: "kubectl",
+				arg:  []string{"Hello", "World"},
+			},
+			wantCmd: &exec.Cmd{
+				Path: "echo",
+				Args: []string{"Hello", "World"},
+			},
+			wantErr: false,
+		}, {
+			name: "Valid command arguments",
+			args: args{
+				name: "echo",
+				arg:  []string{"Hello", "World"},
+			},
+			wantCmd: &exec.Cmd{
+				Path: "echo",
+				Args: []string{"Hello", "World"},
+			},
+			wantErr: false,
+		},
+		{
+			name: "Invalid command arguments",
+			args: args{
+				name: "echo",
+				arg:  []string{"Hello", "World&"},
+			},
+			wantCmd: nil,
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cmd, err := SimpleCommand(tt.args.name, tt.args.arg...)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("SimpleCommand() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if err != nil {
+				return
+			}
+			tt.wantCmd = exec.Command(tt.args.name, tt.args.arg...)
+			if !reflect.DeepEqual(tt.wantCmd.Args, cmd.Args) {
+				t.Errorf("SimpleCommand() = %v, want %v", tt.args.arg, cmd.Args)
+			}
+			if !reflect.DeepEqual(tt.wantCmd.Path, cmd.Path) {
+				t.Errorf("SimpleCommand() = %v, want %v", tt.args.arg, cmd.Args)
+			}
+		})
+	}
+}
+
+func Test_buildPathList(t *testing.T) {
+	type args struct {
+		pathList map[string]bool
+	}
+	tests := []struct {
+		name             string
+		args             args
+		mockLookpathFunc func(file string) (string, error)
+		want             map[string]bool
+	}{
+		{
+			name: "Test with command 'kubectl'",
+			args: args{
+				pathList: map[string]bool{"kubectl": true},
+			},
+			mockLookpathFunc: func(file string) (string, error) {
+				return "/path/to/" + file, nil // Mocked path
+			},
+			want: map[string]bool{"kubectl": true, "/path/to/kubectl": true}, // assuming '/path/to/kubectl' is the path of the 'kubectl' command
+		},
+		{
+			name: "Test with nonexistent command",
+			args: args{
+				pathList: map[string]bool{"nonexistent": true},
+			}, mockLookpathFunc: func(file string) (string, error) {
+				return "", fmt.Errorf("Failed to find path")
+			},
+			want: map[string]bool{"nonexistent": true}, // as 'nonexistent' command does not exist, so the result should be same as initial
+		},
+		{
+			name: "Test with full path command",
+			args: args{
+				pathList: map[string]bool{"/usr/local/bin/kubectl": true},
+			},
+			mockLookpathFunc: func(file string) (string, error) {
+				return "/path/to/" + file, nil // Mocked path
+			},
+			want: map[string]bool{"/usr/local/bin/kubectl": true}, // since '/usr/local/bin/kubectl' command already has full path, so the result should be same as initial
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := gohook.Hook(exec.LookPath, tt.mockLookpathFunc, nil)
+			if err != nil {
+				t.Fatalf("failed to hook function: %v", err)
+			}
+			got := buildPathList(tt.args.pathList)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("buildPathList() = %v, want %v", got, tt.want)
+			}
+			_ = gohook.UnHook(tt.mockLookpathFunc)
+
+		})
+	}
+}
diff --git a/pkg/utils/helm/helm_test.go b/pkg/utils/helm/helm_test.go
index 5d4068cf..35e9dc8a 100644
--- a/pkg/utils/helm/helm_test.go
+++ b/pkg/utils/helm/helm_test.go
@@ -17,99 +17,9 @@ func TestGenerateValueFile(t *testing.T) {
 	}
 }
 
-func TestGenerateHelmTemplate(t *testing.T) {
-	CombinedOutputCommon := func(cmd *exec.Cmd) ([]byte, error) {
-		return []byte("test-output"), nil
-	}
-	CombinedOutputErr := func(cmd *exec.Cmd) ([]byte, error) {
-		return nil, errors.New("fail to run the command")
-	}
-	StatCommon := func(name string) (os.FileInfo, error) {
-		return nil, nil
-	}
-	StatErr := func(name string) (os.FileInfo, error) {
-		return nil, errors.New("fail to run the command")
-	}
-	LookPathCommon := func(file string) (string, error) {
-		return "test-path", nil
-	}
-	LookPathErr := func(file string) (string, error) {
-		return "", errors.New("fail to run the command")
-	}
-
-	wrappedUnhookCombinedOutput := func() {
-		err := gohook.UnHook((*exec.Cmd).CombinedOutput)
-		if err != nil {
-			t.Fatal(err.Error())
-		}
-	}
-	wrappedUnhookStat := func() {
-		err := gohook.UnHook(os.Stat)
-		if err != nil {
-			t.Fatal(err.Error())
-		}
-	}
-	wrappedUnhookLookPath := func() {
-		err := gohook.UnHook(exec.LookPath)
-		if err != nil {
-			t.Fatal(err.Error())
-		}
-	}
-
-	err := gohook.Hook(exec.LookPath, LookPathErr, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	_, err = GenerateHelmTemplate("fluid", "default", "testValueFile", "testChartName")
-	if err == nil {
-		t.Errorf("fail to catch the error")
-	}
-	wrappedUnhookLookPath()
-
-	err = gohook.Hook(exec.LookPath, LookPathCommon, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	err = gohook.Hook(os.Stat, StatErr, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	_, err = GenerateHelmTemplate("fluid", "default", "testValueFile", "testChartName")
-	if err == nil {
-		t.Errorf("fail to catch the error")
-	}
-	wrappedUnhookStat()
-
-	err = gohook.Hook(os.Stat, StatCommon, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	err = gohook.Hook((*exec.Cmd).CombinedOutput, CombinedOutputErr, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	_, err = GenerateHelmTemplate("fluid", "default", "testValueFile", "testChartName")
-	if err == nil {
-		t.Errorf("fail to catch the error")
-	}
-	wrappedUnhookCombinedOutput()
-
-	err = gohook.Hook((*exec.Cmd).CombinedOutput, CombinedOutputCommon, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	_, err = GenerateHelmTemplate("fluid", "default", "testValueFile", "testChartName")
-	if err != nil {
-		t.Errorf("fail to exec the function")
-	}
-	wrappedUnhookCombinedOutput()
-	wrappedUnhookStat()
-	wrappedUnhookLookPath()
-}
-
 func TestGetChartVersion(t *testing.T) {
 	LookPathCommon := func(file string) (string, error) {
-		return "test-path", nil
+		return "helm", nil
 	}
 	LookPathErr := func(file string) (string, error) {
 		return "", errors.New("fail to run the command")
@@ -190,10 +100,10 @@ func TestGetChartVersion(t *testing.T) {
 	}
 	version, err := GetChartVersion("fluid:v0.6.0")
 	if err != nil {
-		t.Errorf("fail to exec the function")
+		t.Errorf("fail to exec the function due to %v", err)
 	}
 	if version != "v0.6.0" {
-		t.Errorf("fail to get the version of the helm")
+		t.Errorf("fail to get the version of the helm due to %v", err)
 	}
 	wrappedUnhookOutput()
 	wrappedUnhookStat()
diff --git a/pkg/utils/helm/utils_test.go b/pkg/utils/helm/utils_test.go
index 136e35d0..8a8b7164 100644
--- a/pkg/utils/helm/utils_test.go
+++ b/pkg/utils/helm/utils_test.go
@@ -102,6 +102,12 @@ func TestInstallRelease(t *testing.T) {
 	}
 	wrappedUnhookCombinedOutput()
 
+	badValue := "test$bad"
+	err = InstallRelease("fluid", badValue, "testValueFile", "/chart/fluid")
+	if err == nil {
+		t.Errorf("fail to catch the error of %s", badValue)
+	}
+
 	err = gohook.Hook((*exec.Cmd).CombinedOutput, CombinedOutputCommon, nil)
 	if err != nil {
 		t.Fatal(err.Error())
@@ -175,6 +181,12 @@ func TestCheckRelease(t *testing.T) {
 	}
 	wrappedUnhookStart()
 
+	badValue := "test$bad"
+	_, err = CheckRelease("fluid", badValue)
+	if err == nil {
+		t.Errorf("fail to catch the error of %s", badValue)
+	}
+
 	err = gohook.Hook((*exec.Cmd).Start, StartCommon, nil)
 	if err != nil {
 		t.Fatal(err.Error())
@@ -242,6 +254,12 @@ func TestDeleteRelease(t *testing.T) {
 		t.Errorf("fail to catch the error")
 	}
 	wrappedUnhookOutput()
+	// test check illegal arguements
+	badValue := "test$bad"
+	err = DeleteRelease("fluid", badValue)
+	if err == nil {
+		t.Errorf("fail to catch the error of %s", badValue)
+	}
 
 	err = gohook.Hook((*exec.Cmd).Output, OutputCommon, nil)
 	if err != nil {
@@ -319,6 +337,11 @@ func TestListReleases(t *testing.T) {
 	}
 	wrappedUnhookOutput()
 	wrappedUnhookLookPath()
+
+	_, err = ListReleases("def$ault")
+	if err == nil {
+		t.Errorf("fail to catch the error")
+	}
 }
 
 func TestListReleaseMap(t *testing.T) {
@@ -385,6 +408,11 @@ func TestListReleaseMap(t *testing.T) {
 	}
 	wrappedUnhookOutput()
 	wrappedUnhookLookPath()
+
+	_, err = ListReleaseMap("def$ault")
+	if err == nil {
+		t.Errorf("fail to catch the error")
+	}
 }
 
 func TestListAllReleasesWithDetail(t *testing.T) {
@@ -451,6 +479,11 @@ func TestListAllReleasesWithDetail(t *testing.T) {
 	}
 	wrappedUnhookOutput()
 	wrappedUnhookLookPath()
+
+	_, err = ListAllReleasesWithDetail("def$ault")
+	if err == nil {
+		t.Errorf("fail to catch the error")
+	}
 }
 
 func TestDeleteReleaseIfExists(t *testing.T) {
diff --git a/pkg/utils/kubectl/configmap_test.go b/pkg/utils/kubectl/configmap_test.go
index 56db555d..b36dd600 100644
--- a/pkg/utils/kubectl/configmap_test.go
+++ b/pkg/utils/kubectl/configmap_test.go
@@ -72,68 +72,6 @@ func TestCreateConfigMapFromFile(t *testing.T) {
 	wrappedUnhookStat()
 }
 
-func TestSaveConfigMapToFile(t *testing.T) {
-	LookPathCommon := func(file string) (string, error) {
-		return "test-path", nil
-	}
-	LookPathErr := func(file string) (string, error) {
-		return "", errors.New("fail to run the command")
-	}
-	OutputCommon := func(cmd *exec.Cmd) ([]byte, error) {
-		return []byte("output"), nil
-	}
-	OutputErr := func(cmd *exec.Cmd) ([]byte, error) {
-		return nil, errors.New("fail to run the command")
-	}
-
-	wrappedUnhookLookPath := func() {
-		err := gohook.UnHook(exec.LookPath)
-		if err != nil {
-			t.Fatal(err.Error())
-		}
-	}
-	wrappedUnhookOutput := func() {
-		err := gohook.UnHook((*exec.Cmd).Output)
-		if err != nil {
-			t.Fatal(err.Error())
-		}
-	}
-	err := gohook.Hook(exec.LookPath, LookPathErr, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	_, err = SaveConfigMapToFile("hbase", "data", "default")
-	if err == nil {
-		t.Errorf("check failure, want err, get nil")
-	}
-	wrappedUnhookLookPath()
-
-	err = gohook.Hook(exec.LookPath, LookPathCommon, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	err = gohook.Hook((*exec.Cmd).Output, OutputErr, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	_, err = SaveConfigMapToFile("hbase", "data", "default")
-	if err == nil {
-		t.Errorf("check failure, want err, get nil")
-	}
-	wrappedUnhookOutput()
-
-	err = gohook.Hook((*exec.Cmd).Output, OutputCommon, nil)
-	if err != nil {
-		t.Fatal(err.Error())
-	}
-	_, err = SaveConfigMapToFile("hbase", "data", "default")
-	if err != nil {
-		t.Errorf("check failure, want nil, get err %v", err)
-	}
-	wrappedUnhookOutput()
-	wrappedUnhookLookPath()
-}
-
 func TestKubectl(t *testing.T) {
 	CombinedOutputCommon := func(cmd *exec.Cmd) ([]byte, error) {
 		return []byte("test-CombineOutput"), nil
diff --git a/pkg/utils/mount_test.go b/pkg/utils/mount_test.go
index 97e59b52..bba791eb 100644
--- a/pkg/utils/mount_test.go
+++ b/pkg/utils/mount_test.go
@@ -101,6 +101,10 @@ func TestCheckMountReady(t *testing.T) {
 			err := CheckMountReadyAndSubPathExist("", "test", "")
 			So(err, ShouldNotBeNil)
 		})
+		Convey("illegal subpath", func() {
+			err := CheckMountReadyAndSubPathExist("/test", "test", "$(echo)")
+			So(err, ShouldNotBeNil)
+		})
 	})
 }
 
diff --git a/pkg/utils/shell_pipes_test.go b/pkg/utils/shell_pipes_test.go
new file mode 100644
index 00000000..430f86c0
--- /dev/null
+++ b/pkg/utils/shell_pipes_test.go
@@ -0,0 +1,132 @@
+package utils
+
+import (
+	"os/exec"
+	"reflect"
+	"testing"
+)
+
+/*
+Copyright 2024 The Fluid Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+func TestValidateShellPipeString(t *testing.T) {
+	type args struct {
+		command string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{name: "valid command with grep", args: args{command: "echo hello world | grep hello"}, wantErr: true},
+		{name: "valid command with wc -l", args: args{command: "kubectl hello world | wc -l"}, wantErr: false},
+		{name: "invalid command with xyz", args: args{command: "echo hello world | xyz"}, wantErr: true},
+		{name: "illegal sequence in command with &", args: args{command: "echo hello world & echo y"}, wantErr: true},
+		{name: "illegal sequence in command with ;", args: args{command: "ls ; echo y"}, wantErr: true},
+		{name: "command with $", args: args{command: "kubectl $HOME"}, wantErr: true},
+		{name: "command with absolute path", args: args{command: "ls /etc"}, wantErr: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := ValidateShellPipeString(tt.args.command); (err != nil) != tt.wantErr {
+				t.Errorf("Testcase '%s' ValidateShellPipeString() error = %v, wantErr %v", tt.name, err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestPipeCommand(t *testing.T) {
+	type args struct {
+		name string
+		arg  []string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantCmd *exec.Cmd
+		wantErr bool
+	}{
+		{name: "valid simple command", args: args{name: "bash", arg: []string{"-c", "ls"}}, wantCmd: exec.Command("bash", "-c", "ls"), wantErr: false},
+		{name: "insufficient arguments", args: args{name: "bash", arg: []string{"-c"}}, wantCmd: nil, wantErr: true},
+		{name: "unknown shell command", args: args{name: "zsh", arg: []string{"-c", "ls"}}, wantCmd: nil, wantErr: true},
+		{name: "valid piped command", args: args{name: "bash", arg: []string{"-c", "ls | grep something"}}, wantCmd: exec.Command("bash", "-c", "ls | grep something"), wantErr: false},
+		{name: "invalid piped command", args: args{name: "bash", arg: []string{"-c", "ls | random-command"}}, wantCmd: nil, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotCmd, err := PipeCommand(tt.args.name, tt.args.arg...)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Testcase '%s': PipeCommand()  error = %v, wantErr %v", tt.name, err, tt.wantErr)
+				return
+			}
+			if gotCmd != nil && !reflect.DeepEqual(gotCmd.Path, tt.wantCmd.Path) {
+				t.Errorf("Testcase '%s': PipeCommand()  = %v, want %v", tt.name, gotCmd, tt.wantCmd)
+			}
+			if gotCmd != nil && !reflect.DeepEqual(gotCmd.Args, tt.wantCmd.Args) {
+				t.Errorf("Testcase '%s': PipeCommand()   = %v, want %v", tt.name, gotCmd, tt.wantCmd)
+			}
+		})
+	}
+}
+
+func TestValidatePipeCommandSlice(t *testing.T) {
+	type args struct {
+		shellCommandSlice []string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{name: "valid bash command", args: args{shellCommandSlice: []string{"bash", "-c", "ls"}}, wantErr: false},
+		{name: "valid sh command", args: args{shellCommandSlice: []string{"sh", "-c", "ls"}}, wantErr: false},
+		{name: "unknown shell command", args: args{shellCommandSlice: []string{"zsh", "-c", "ls"}}, wantErr: true},
+		{name: "invalid bash command", args: args{shellCommandSlice: []string{"bash", "-c", "wrong_command"}}, wantErr: true},
+		{name: "insufficient arguments", args: args{shellCommandSlice: []string{"bash", "-c"}}, wantErr: true},
+		{name: "empty command slice", args: args{shellCommandSlice: []string{}}, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := ValidatePipeCommandSlice(tt.args.shellCommandSlice); (err != nil) != tt.wantErr {
+				t.Errorf("Testcase '%s': ValidatePipeCommandSlice() error = %v, wantErr %v", tt.name, err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestIsValidCommand(t *testing.T) {
+	type args struct {
+		cmd             string
+		allowedCommands map[string]bool
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{name: "valid bash command", args: args{cmd: "bash", allowedCommands: map[string]bool{"bash": true}}, want: true},
+		{name: "valid sh command", args: args{cmd: "sh", allowedCommands: map[string]bool{"bash": true, "sh": true}}, want: true},
+		{name: "invalid zsh command", args: args{cmd: "zsh", allowedCommands: map[string]bool{"bash": true}}, want: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isValidCommand(tt.args.cmd, tt.args.allowedCommands); got != tt.want {
+				t.Errorf("Testcase '%s': isValidCommand() = %v, want %v", tt.name, got, tt.want)
+			}
+		})
+	}
+}
\ No newline at end of file