[ML][Pipelines] Singularity: bypass operations to Singularity compute (Azure#28576)

* test: add test for Singularity compute in pipeline job

* skip get asset id/compute operation when Singularity

* test: add recording for newly added test

* test: refine test

Author: zhengfeiwang

sdk/ml/azure-ai-ml/azure/ai/ml/_utils/_arm_id_utils.py

@@ -19,6 +19,7 @@
     PROVIDER_RESOURCE_ID_WITH_VERSION,
     REGISTRY_URI_REGEX_FORMAT,
     REGISTRY_VERSION_PATTERN,
+    SINGULARITY_ID_FORMAT,
 )
 from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException
 
@@ -325,6 +326,12 @@ def is_registry_id_for_resource(name: Any) -> bool:
     return False
 
 
+def is_singularity_id_for_resource(name: Any) -> bool:
+    if isinstance(name, str) and re.match(SINGULARITY_ID_FORMAT, name, re.IGNORECASE):
+        return True
+    return False
+
+
 def get_arm_id_with_version(
     operation_scope: OperationScope,
     provider_name: str,

sdk/ml/azure-ai-ml/azure/ai/ml/constants/_common.py

@@ -35,6 +35,9 @@
 PROVIDER_RESOURCE_ID_WITH_VERSION = (
     "/subscriptions/{}/resourceGroups/{}/providers/Microsoft.MachineLearningServices/workspaces/{}/{}/{}/versions/{}"
 )
+SINGULARITY_ID_FORMAT = (
+    "/subscriptions/.*/resourceGroups/.*/providers/Microsoft.MachineLearningServices/virtualclusters/.*"
+)
 ASSET_ID_FORMAT = "azureml://locations/{}/workspaces/{}/{}/{}/versions/{}"
 VERSIONED_RESOURCE_NAME = "{}:{}"
 LABELLED_RESOURCE_NAME = "{}@{}"

sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py

@@ -105,7 +105,12 @@
 from ._job_ops_helper import get_git_properties, get_job_output_uris_from_dataplane, stream_logs_until_completion
 from ._local_job_invoker import is_local_run, start_run_if_local
 from ._model_dataplane_operations import ModelDataplaneOperations
-from ._operation_orchestrator import OperationOrchestrator, is_ARM_id_for_resource, is_registry_id_for_resource
+from ._operation_orchestrator import (
+    OperationOrchestrator,
+    is_ARM_id_for_resource,
+    is_registry_id_for_resource,
+    is_singularity_id_for_resource,
+)
 from ._run_operations import RunOperations
 
 try:
@@ -331,6 +336,9 @@ def _try_get_compute_arm_id(self, compute: Union[Compute, str]):
             return compute
 
         if compute is not None:
+            if is_singularity_id_for_resource(compute):
+                # Singularity compute, skip try to get operation
+                return compute
             if is_ARM_id_for_resource(compute, resource_type=AzureMLResourceType.COMPUTE):
                 # compute is not a sub-workspace resource
                 compute_name = compute.split("/")[-1]

sdk/ml/azure-ai-ml/azure/ai/ml/operations/_operation_orchestrator.py

@@ -18,6 +18,7 @@
     get_arm_id_with_version,
     is_ARM_id_for_resource,
     is_registry_id_for_resource,
+    is_singularity_id_for_resource,
     parse_name_label,
     parse_prefixed_name_version,
 )
@@ -124,6 +125,7 @@ def get_asset_arm_id(
             asset is None
             or is_ARM_id_for_resource(asset, azureml_type, sub_workspace_resource)
             or is_registry_id_for_resource(asset)
+            or is_singularity_id_for_resource(asset)
         ):
             return asset
         if isinstance(asset, str):

sdk/ml/azure-ai-ml/tests/pipeline_job/e2etests/test_pipeline_job.py

@@ -1344,6 +1344,26 @@ def test_pipeline_node_with_default_component(self, client: MLClient, randstr: C
             == "microsoftsamples_command_component_basic@default"
         )
 
+    def test_pipeline_job_with_singularity_compute(self, client: MLClient, randstr: Callable[[str], str]):
+        params_override = [{"name": randstr("job_name")}]
+        pipeline_job: PipelineJob = load_job(
+            "./tests/test_configs/pipeline_jobs/helloworld_pipeline_job_with_singularity_compute.yml",
+            params_override=params_override,
+        )
+
+        singularity_compute_id = (
+            f"/subscriptions/{client.subscription_id}/resourceGroups/{client.resource_group_name}/"
+            f"providers/Microsoft.MachineLearningServices/virtualclusters/SingularityTestVC"
+        )
+        pipeline_job.settings.default_compute = singularity_compute_id
+        pipeline_job.jobs["hello_job"].compute = singularity_compute_id
+
+        assert pipeline_job._customized_validate().passed is True
+
+        created_pipeline_job: PipelineJob = assert_job_cancel(pipeline_job, client)
+        assert created_pipeline_job.settings.default_compute == singularity_compute_id
+        assert created_pipeline_job.jobs["hello_job"].compute == singularity_compute_id
+
 
 @pytest.mark.usefixtures("enable_pipeline_private_preview_features")
 @pytest.mark.e2etest