cloud run job unique names

prefect_gcp/cloud_run.py

@@ -27,6 +27,7 @@
     ```
 
 """
+
 from __future__ import annotations
 
 import json
@@ -59,6 +60,8 @@
 
 from prefect_gcp.credentials import GcpCredentials
 
+JOB_NAME_MAX_LENGTH = 30
+
 
 class Job(BaseModel):
     """

prefect_gcp/models/cloud_run_v2.py

@@ -11,6 +11,8 @@
 else:
     from pydantic import BaseModel
 
+JOB_V2_NAME_MAX_LENGTH = 30
+
 
 class JobV2(BaseModel):
     """

prefect_gcp/workers/cloud_run.py

@@ -145,6 +145,7 @@
 import shlex
 import time
 from typing import TYPE_CHECKING, Any, Dict, Optional
+from uuid import uuid4
 
 import anyio
 import googleapiclient
@@ -170,7 +171,7 @@
 else:
     from pydantic import Field, validator
 
-from prefect_gcp.cloud_run import Execution, Job
+from prefect_gcp.cloud_run import JOB_NAME_MAX_LENGTH, Execution, Job
 from prefect_gcp.credentials import GcpCredentials
 
 if TYPE_CHECKING:
@@ -338,7 +339,11 @@ def _populate_name_if_not_present(self):
         """Adds the flow run name to the job if one is not already provided."""
         try:
             if "name" not in self.job_body["metadata"]:
-                self.job_body["metadata"]["name"] = self.name
+                base_job_name = self.name.lower()
+                if len(base_job_name) > JOB_NAME_MAX_LENGTH:
+                    base_job_name = base_job_name[:JOB_NAME_MAX_LENGTH]
+                job_name = f"{base_job_name}-{uuid4().hex}"
+                self.job_body["metadata"]["name"] = job_name
         except KeyError:
             raise ValueError("Unable to verify name due to invalid job body template.")
 

prefect_gcp/workers/cloud_run_v2.py

@@ -2,6 +2,7 @@
 import shlex
 import time
 from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional
+from uuid import uuid4
 
 from anyio.abc import TaskStatus
 from google.api_core.client_options import ClientOptions
@@ -29,7 +30,12 @@
     from pydantic import Field, validator
 
 from prefect_gcp.credentials import GcpCredentials
-from prefect_gcp.models.cloud_run_v2 import CloudRunJobV2Result, ExecutionV2, JobV2
+from prefect_gcp.models.cloud_run_v2 import (
+    JOB_V2_NAME_MAX_LENGTH,
+    CloudRunJobV2Result,
+    ExecutionV2,
+    JobV2,
+)
 
 if TYPE_CHECKING:
     from prefect.client.schemas import FlowRun
@@ -125,6 +131,7 @@ class CloudRunWorkerJobV2Configuration(BaseJobConfiguration):
             "complete before raising an exception."
         ),
     )
+    job_name: str = Field(default=None, description="The name of the Cloud Run job.")
 
     @property
     def project(self) -> str:
@@ -136,20 +143,6 @@ def project(self) -> str:
         """
         return self.credentials.project
 
-    @property
-    def job_name(self):
-        """
-        Returns the job name, if it does not exist, it creates it.
-        """
-        pre_trim_cr_job_name = f"prefect-{self.name}"
-
-        if len(pre_trim_cr_job_name) > 40:
-            pre_trim_cr_job_name = pre_trim_cr_job_name[:40]
-
-        pre_trim_cr_job_name = pre_trim_cr_job_name.rstrip("-")
-
-        return pre_trim_cr_job_name
-
     def prepare_for_flow_run(
         self,
         flow_run: "FlowRun",
@@ -175,6 +168,7 @@ def prepare_for_flow_run(
         )
 
         self._populate_env()
+        self._populate_name()
         self._populate_or_format_command()
         self._format_args_if_present()
         self._populate_image_if_not_present()
@@ -194,6 +188,14 @@ def _populate_env(self):
 
         self.job_body["template"]["template"]["containers"][0]["env"] = envs
 
+    def _populate_name(self):
+        """Create a unique and valid job name."""
+        base_job_name = self.name.lower()
+        if len(base_job_name) > JOB_V2_NAME_MAX_LENGTH:
+            base_job_name = base_job_name[:JOB_V2_NAME_MAX_LENGTH]
+        job_name = f"{base_job_name}-{uuid4().hex}"
+        self.job_name = job_name
+
     def _populate_image_if_not_present(self):
         """
         Populates the job body with the image if not present.