Default ttl seconds after finished to 3days after data flow completes

TrafalgarZZZ · TrafalgarZZZ · commit 8f9d64ea5c2f · 2024-03-07T14:20:03.000+08:00
Signed-off-by: trafalgarzzz &lt;trafalgarz@outlook.com&gt;
diff --git a/examples/kubernetes/create_dataset_and_runtime.py b/examples/kubernetes/create_dataset_and_runtime.py
@@ -19,19 +19,13 @@ def init_logger(logger_name, logger_level):
     logger.setLevel(logger_level)
 
 
-# Example for data experts like data scientists and data engineers.
-def main():
-    global logger
-    init_logger("fluidsdk", logging.INFO)
-
-    client_config = ClientConfig()
-    fluid_client = FluidClient(client_config)
-
+def create_dataset_with_alluxio(fluid_client: FluidClient):
     dataset_name = "demo"
     try:
+        # Mounting WebUFS to Alluxio
         fluid_client.create_dataset(dataset_name, "hbase", "https://mirrors.bit.edu.cn/apache/hbase/stable/", "/hbase")
     except Exception as e:
-        raise RuntimeError("f""Failed to create dataset: {e}")
+        raise RuntimeError(f"Failed to create dataset: {e}")
 
     logger.info(f"Dataset \"{dataset_name}\" created successfully")
 
@@ -45,6 +39,100 @@ def main():
     logger.info(f"AlluxioRuntime created and bound to dataset \"{dataset_name}\", cache engine is now ready")
 
 
+def create_dataset_with_jindofs(fluid_client: FluidClient):
+    dataset_name = "demo"
+    try:
+        # Mounting OSS Bucket to Jindo
+        fluid_client.create_dataset(dataset_name,
+                                    "mybucket", "oss://mybucket/subdir", "/",
+                                    options={"fs.oss.endpoint": "oss-cn-beijing-internal.aliyuncs.com"},
+                                    cred_secret_name="access-key",
+                                    cred_secret_options={
+                                        "fs.oss.accessKeyId": "fs.oss.accessKeyId",
+                                        "fs.oss.accessKeySecret": "fs.oss.accessKeySecret"
+                                    })
+    except Exception as e:
+        raise RuntimeError("f""Failed to create dataset: {e}")
+
+    logger.info(f"Dataset \"{dataset_name}\" created successfully")
+
+    try:
+        dataset = fluid_client.get_dataset(dataset_name)
+    except Exception as e:
+        raise RuntimeError(f"Failed to get dataset: {e}")
+
+    logger.info(f"Binding JindoRuntime to dataset \"{dataset_name}\"...")
+    dataset.bind_runtime(runtime_type="jindo", replicas=1, cache_medium="MEM", cache_capacity_GiB=2, wait=True)
+    logger.info(f"JindoRuntime created and bound to dataset \"{dataset_name}\", cache engine is now ready")
+
+
+def create_dataset_with_juicefs(fluid_client: FluidClient):
+    dataset_name = "demo"
+    try:
+        # Setting minio as JuiceFS's backend storage and redis as JuiceFS's meta server
+        fluid_client.create_dataset(dataset_name,
+                                    "minio", "juicefs:///", "/",
+                                    options={"bucket": "http://minio:9000/minio/test", "storage": "minio"},
+                                    cred_secret_name="jfs-secret",
+                                    cred_secret_options={
+                                        "metaurl": "metaurl",
+                                        "access-key": "access-key",
+                                        "secret-key": "secret-key"
+                                    })
+    except Exception as e:
+        raise RuntimeError("f""Failed to create dataset: {e}")
+
+    logger.info(f"Dataset \"{dataset_name}\" created successfully")
+
+    try:
+        dataset = fluid_client.get_dataset(dataset_name)
+    except Exception as e:
+        raise RuntimeError(f"Failed to get dataset: {e}")
+
+    logger.info(f"Binding JuiceFSRuntime to dataset \"{dataset_name}\"...")
+    dataset.bind_runtime(runtime_type="juicefs", replicas=1, cache_medium="MEM", cache_capacity_GiB=2, wait=True)
+    logger.info(f"JuiceFSRuntime created and bound to dataset \"{dataset_name}\", cache engine is now ready")
+
+
+def create_dataset_with_vineyard(fluid_client: FluidClient):
+    dataset_name = "vineyard"
+    try:
+        fluid_client.create_dataset(dataset_name)
+    except Exception as e:
+        raise RuntimeError("f""Failed to create dataset: {e}")
+
+    logger.info(f"Dataset \"{dataset_name}\" created successfully")
+
+    try:
+        dataset = fluid_client.get_dataset(dataset_name)
+    except Exception as e:
+        raise RuntimeError(f"Failed to get dataset: {e}")
+
+    logger.info(f"Binding VineyardRuntime to dataset \"{dataset_name}\"...")
+    dataset.bind_runtime(runtime_type="vineyard", replicas=1, cache_medium="MEM", cache_capacity_GiB=2, wait=True)
+    logger.info(f"VineyardRuntime created and bound to dataset \"{dataset_name}\", cache engine is now ready")
+
+
+# Examples for data experts like data scientists and data engineers.
+def main():
+    global logger
+    init_logger("fluidsdk", logging.INFO)
+
+    client_config = ClientConfig()
+    fluid_client = FluidClient(client_config)
+
+    cases = {
+        "alluxio": create_dataset_with_alluxio,
+        "jindofs": create_dataset_with_jindofs,
+        "juicefs": create_dataset_with_juicefs,
+        "vineyard": create_dataset_with_vineyard
+    }
+
+    # Change case_name to play with different cache engines
+    case_name = "alluxio"
+    cases[case_name](fluid_client)
+
+
 # Example for Kubernetes experts who is familiar with YAML-like APIs.
 def main_k8s_client():
     global logger
@@ -90,7 +178,7 @@ def main_k8s_client():
                 levels=[
                     models.Level(
                         mediumtype="MEM",
-                        volume_type="hostPath",
+                        volume_type="emptyDir",
                         path="/dev/shm",
                         quota="2Gi",
                         high="0.95",
diff --git a/examples/kubernetes/fluid_dataflow.py b/examples/kubernetes/fluid_dataflow.py
@@ -16,7 +16,7 @@
 
 from fluid import constants
 from fluid import models
-from fluid import FluidK8sClient
+from fluid import FluidClient, ClientConfig
 
 logger = logging.getLogger("fluidsdk")
 stream_handler = logging.StreamHandler(sys.stdout)
@@ -71,7 +71,7 @@ def build_data_processor():
 
 def main():
     # Initialize Fluid client
-    fluid_client = FluidK8sClient()
+    fluid_client = FluidClient(ClientConfig())
     # Get a bound dataset for later operations and dataflow
     dataset = fluid_client.get_dataset("demo-dataset")
 
@@ -85,12 +85,14 @@ def main():
         print(op_status)
 
     # Flow example 3: migrate some data to the dataset, preload it and finally process it.
-    # Then wait until the flow is completed
-    flow3 = dataset.migrate("/ossdata", constants.DATA_MIGRATE_DIRECTION_FROM,
-                            models.ExternalStorage(uri="oss://my-bucket/my-data",
-                                                   encrypt_options=get_encrypt_options())) \
-        .preload("/ossdata").process(dataset_mountpath="/data", processor=build_data_processor()).run(
-        run_id="testflow3")
+    # The following code waits until the flow is completed.
+    # After 24 hours, all the operations in the flow will be automatically garbage collected.
+    flow3 = dataset \
+        .migrate("/ossdata", constants.DATA_MIGRATE_DIRECTION_FROM,
+                 models.ExternalStorage(uri="oss://my-bucket/my-data", encrypt_options=get_encrypt_options())) \
+        .preload("/ossdata") \
+        .process(dataset_mountpath="/data", processor=build_data_processor()) \
+        .run(run_id="testflow3", ttl_seconds_after_finished=86400)
     flow3.wait()
 
 
diff --git a/fluid/api/fluid_dataflow.py b/fluid/api/fluid_dataflow.py
@@ -208,14 +208,16 @@ def process(self, dataset_mountpath, processor, sub_path=None):
         self.flow_ops.append(op)
         return self
 
-    def run(self, run_id):
+    def run(self, run_id, ttl_seconds_after_finished=constants.DEFAULT_DATAFLOW_TIME_TO_LIVE):
         """
         Execute the dataflow by submitting all the defined operations to the cluster.
 
         Parameters
         ----------
         run_id: str
             The run id of the dataflow.
+        ttl_seconds_after_finished: int
+            Seconds to live after finished for each operation in the dataflow.
 
         Returns
         -------
@@ -246,6 +248,7 @@ def run(self, run_id):
                     namespace=last_step.metadata.namespace,
                     kind=last_step.kind
                 )
+            op.spec.ttl_seconds_after_finished = ttl_seconds_after_finished
             self.fluid_client.create_data_operation(op)
             last_step = op
         return FlowHandle(run_id, steps_to_execute)
diff --git a/fluid/api/fluid_k8s_client.py b/fluid/api/fluid_k8s_client.py
@@ -140,7 +140,7 @@ def create_data_operation(self, data_op: constants.DATA_OPERATION_MODELS_TYPE, n
         if wait:
             self.wait_data_operation_completed(data_op.metadata.name, data_op.kind, namespace)
 
-    def get_dataset(self, name, namespace=None, timeout=constants.DEFAULT_TIMEOUT) -> models.Dataset:
+    def get_dataset(self, name, namespace=None, timeout=constants.DEFAULT_API_REQUEST_TIMEOUT) -> models.Dataset:
         namespace = namespace or self.namespace
 
         try:
@@ -168,7 +168,7 @@ def get_dataset(self, name, namespace=None, timeout=constants.DEFAULT_TIMEOUT) -
         return dataset
 
     def get_runtime(self, name, runtime_type=None, namespace=None,
-                    timeout=constants.DEFAULT_TIMEOUT) -> constants.RUNTIME_MODELS_TYPE:
+                    timeout=constants.DEFAULT_API_REQUEST_TIMEOUT) -> constants.RUNTIME_MODELS_TYPE:
         namespace = namespace or self.namespace
         runtime_kind = utils.infer_runtime_kind(runtime_type) or self.default_runtime_kind
         if runtime_kind is None:
@@ -199,7 +199,7 @@ def get_runtime(self, name, runtime_type=None, namespace=None,
 
         return runtime
 
-    def get_data_operation(self, name, data_op_type, namespace=None, timeout=constants.DEFAULT_TIMEOUT):
+    def get_data_operation(self, name, data_op_type, namespace=None, timeout=constants.DEFAULT_API_REQUEST_TIMEOUT):
         namespace = namespace or self.namespace
         data_op_kind = utils.infer_data_operation_kind(data_op_type)
         if data_op_kind is None:
@@ -280,7 +280,8 @@ def wait_dataset_bound(self, name, namespace=None, poll_timeout=constants.DEFAUL
         if poll >= poll_timeout:
             raise TimeoutError(f"TimeoutError: Timed out when waiting dataset \"{namespace}/{name}\" bound")
 
-    def delete_dataset(self, name, namespace=None, wait_until_cleaned_up=False, timeout=constants.DEFAULT_TIMEOUT,
+    def delete_dataset(self, name, namespace=None, wait_until_cleaned_up=False,
+                       timeout=constants.DEFAULT_API_REQUEST_TIMEOUT,
                        **kwargs):
         namespace = namespace or self.namespace
 
@@ -330,7 +331,7 @@ def delete_dataset(self, name, namespace=None, wait_until_cleaned_up=False, time
                                 wait_until_cleaned_up=wait_until_cleaned_up, timeout=timeout, **kwargs)
 
     def delete_runtime(self, name, runtime_type=None, namespace=None, wait_until_cleaned_up=False,
-                       timeout=constants.DEFAULT_TIMEOUT, **kwargs):
+                       timeout=constants.DEFAULT_API_REQUEST_TIMEOUT, **kwargs):
         namespace = namespace or self.namespace
         runtime_kind = utils.infer_runtime_kind(runtime_type) or self.default_runtime_kind
         if runtime_kind is None:
diff --git a/fluid/constants/constants.py b/fluid/constants/constants.py
@@ -16,11 +16,13 @@
 
 from fluid import models
 
-DEFAULT_TIMEOUT = 300
-
+# ------ Common Contants -------
+DEFAULT_API_REQUEST_TIMEOUT = 300
 DEFAULT_POLL_TIMEOUT = 600
 DEFAULT_POLL_INTERVAL = 3
+DEFAULT_DATAFLOW_TIME_TO_LIVE = 86400 * 3  # 3 days
 
+# ------ Fluid Types -------
 GROUP = "data.fluid.io"
 VERSION = "v1alpha1"
 API_VERSION = f"{GROUP}/{VERSION}"
@@ -125,5 +127,6 @@
 for kind, params in DATA_OPERATION_PARAMETERS.items():
     FLUID_CRD_PARAMETERS[kind] = params
 
+# ------ DataMigrate Constants ------
 DATA_MIGRATE_DIRECTION_FROM = "from"
 DATA_MIGRATE_DIRECTION_TO = "to"
