enh: rename pipeline UI (#187)

Author: kouyoumart

deployer/cli.py

@@ -227,6 +227,15 @@ def deploy(  # noqa: C901
             "Defaults to '{pipeline_name}-experiment'.",
         ),
     ] = None,
+    run_name: Annotated[
+        Optional[str],
+        typer.Option(
+            "--run-name",
+            "-rn",
+            help="The pipeline's run name. Displayed in the UI."
+            "Defaults to '{pipeline_name}-{tags}-%Y%m%d%H%M%S'.",
+        ),
+    ] = None,
     skip_validation: Annotated[
         bool,
         typer.Option(
@@ -276,6 +285,7 @@ def deploy(  # noqa: C901
             staging_bucket_name=vertex_settings.VERTEX_STAGING_BUCKET_NAME,
             service_account=vertex_settings.VERTEX_SERVICE_ACCOUNT,
             pipeline_name=pipeline_name,
+            run_name=run_name,
             pipeline_func=pipeline_func,
             gar_location=vertex_settings.GAR_LOCATION,
             gar_repo_id=vertex_settings.GAR_PIPELINES_REPO_ID,

deployer/constants.py

@@ -1,3 +1,4 @@
+import re
 from pathlib import Path
 
 TEMPLATES_PATH = Path(__file__).parent / "_templates"
@@ -64,3 +65,5 @@
     "you can add the following flags to the deploy command if not set in your config:\n"
     "--schedule --cron=cron_expression --scheduler-timezone=IANA_time_zone\n"
 )
+
+VALID_RUN_NAME_PATTERN = re.compile("^[a-z][-a-z0-9]{0,127}$", re.IGNORECASE)

deployer/pipeline_deployer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+from datetime import datetime
 from pathlib import Path
 from typing import Callable, List, Optional
 
@@ -11,6 +12,7 @@
 from loguru import logger
 from requests import HTTPError
 
+from deployer import constants
 from deployer.utils.exceptions import (
     MissingGoogleArtifactRegistryHostError,
     TagNotFoundError,
@@ -24,6 +26,7 @@ def __init__(
         self,
         pipeline_name: str,
         pipeline_func: Callable,
+        run_name: Optional[str] = None,
         project_id: Optional[str] = None,
         region: Optional[str] = None,
         staging_bucket_name: Optional[str] = None,
@@ -39,6 +42,7 @@ def __init__(
         self.service_account = service_account
 
         self.pipeline_name = pipeline_name
+        self.run_name = run_name
         self.pipeline_func = pipeline_func
 
         self.gar_location = gar_location
@@ -106,6 +110,26 @@ def _check_experiment_name(self, experiment_name: Optional[str] = None) -> str:
 
         return experiment_name
 
+    def _check_run_name(self, tag: Optional[str] = None) -> None:
+        """Each run name (job_id) must be unique.
+        We thus always add a timestamp to ensure uniqueness.
+        """
+        now_str = datetime.now().strftime("%Y%m%d-%H%M%S")
+        if self.run_name is None:
+            self.run_name = f"{self.pipeline_name}"
+            if tag:
+                self.run_name += f"-{tag}"
+
+        self.run_name = self.run_name.replace("_", "-")
+        self.run_name += f"-{now_str}"
+
+        if not constants.VALID_RUN_NAME_PATTERN.match(self.run_name):
+            raise ValueError(
+                f"Run name {self.run_name} does not match the pattern"
+                f" {constants.VALID_RUN_NAME_PATTERN.pattern}"
+            )
+        logger.debug(f"run_name is: {self.run_name}")
+
     def _create_pipeline_job(
         self,
         template_path: str,
@@ -139,6 +163,7 @@ def _create_pipeline_job(
         """  # noqa: E501
         job = aiplatform.PipelineJob(
             display_name=self.pipeline_name,
+            job_id=self.run_name,
             template_path=template_path,
             pipeline_root=self.staging_bucket_uri,
             location=self.region,
@@ -210,7 +235,7 @@ def run(
             tag (str, optional): Tag of the pipeline template. Defaults to None.
         """  # noqa: E501
         experiment_name = self._check_experiment_name(experiment_name)
-
+        self._check_run_name(tag=tag)
         template_path = self._get_template_path(tag)
 
         logger.debug(

deployer/settings.py

@@ -30,6 +30,7 @@ class _DeployerDeploySettings(CustomBaseModel):
     config_name: Optional[str] = None
     enable_caching: Optional[bool] = None
     experiment_name: Optional[str] = None
+    run_name: Optional[str] = None
     skip_validation: bool = True
 
 

docs/CLI_REFERENCE.md

@@ -123,6 +123,7 @@ $ vertex-deployer deploy [OPTIONS] PIPELINE_NAMES...
 * `-cn, --config-name TEXT`: Name of the json/py file with parameter values and input artifacts to use when running the pipeline. It must be in the pipeline config dir. e.g. `config_dev.json` for `./vertex/configs/{pipeline-name}/config_dev.json`.
 * `-ec, --enable-caching / -nec, --no-cache`: Whether to turn on caching for the run.If this is not set, defaults to the compile time settings, which are True for alltasks by default, while users may specify different caching options for individualtasks. If this is set, the setting applies to all tasks in the pipeline.Overrides the compile time settings. Defaults to None.
 * `-en, --experiment-name TEXT`: The name of the experiment to run the pipeline in.Defaults to '{pipeline_name}-experiment'.
+* `-rn, --run-name TEXT`: The pipeline's run name. Displayed in the UI.Defaults to '{pipeline_name}-{tags}-%Y%m%d%H%M%S'.
 * `-y, --skip-validation / -n, --no-skip`: Whether to continue without user validation of the settings.  [default: skip-validation]
 * `--help`: Show this message and exit.
 

tests/integration_tests/test_init.py

@@ -60,6 +60,7 @@ def test_init_command_with_user_input(tmp_path):
                 "",
                 "",
                 "",
+                "",
                 "y",
                 "json",
                 "",