Specifying artifact regististry images with tags fails

[khaerensml6]

Creating a PipelineJob based on a artifact registry URI using a tag instead of a version raises an internal server error.
This is pretty annoying.

to be clear:

• Using an artifact with sha256 version number works:
  https://europe-west1-kfp.pkg.dev/my-project/my-project-kfp-template-repository/my-pipeline/sha256:3456789874334a...

• Using a tag gives a 500 internal server error:
  https://europe-west1-kfp.pkg.dev/my-project/my-project-kfp-template-repository/my-pipeline/my-tag

• Using a tag that doesn't exist gives a 404 error when creating the PipelineJob
  https://europe-west1-kfp.pkg.dev/my-project/my-project-kfp-template-repository/my-pipeline/my-non-existent-tag

All of this combined makes it seem like there's a bug for executing tagged artifacts.

Environment details

• OS type and version: Ubuntu 22.04.2 LTS
• Python version: Python 3.9.16
• pip version: pip 22.0.4
• google-cloud-aiplatform version: 1.24.1

Steps to reproduce

1. Create a PipelineJob with a artifact repository URI using a tag instead of the hash
2. Run the pipeline job

Code example

This fails with a 500 internal server error on the "run" call.

    from google.cloud import aiplatform as aip

    service_account = ... 
    pipeline_name = ...
    job_id = f"{pipeline_name}-{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"

    compiled_job = f"https://{region}-kfp.pkg.dev/{PROJECTID}/{REPOSITORY_NAME}/{PIPELINE_NAME}/{TAG}"
    pipeline_job = aip.PipelineJob(
        display_name="test-name",
        job_id=job_id,
        template_path=compiled_job,
    )

    pipeline_job.run(network=None,
                     service_account=service_account,
                     sync=True)

This runs without problem:

    from google.cloud import aiplatform as aip

    service_account = ... 
    pipeline_name = ...
    job_id = f"{pipeline_name}-{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"

    compiled_job = f"https://{region}-kfp.pkg.dev/{PROJECTID}/{REPOSITORY_NAME}/{PIPELINE_NAME}/sha:...."
    pipeline_job = aip.PipelineJob(
        display_name="test-name",
        job_id=job_id,
        template_path=compiled_job,
    )

    pipeline_job.run(network=None,
                     service_account=service_account,
                     sync=True)

Stack trace

Traceback (most recent call last):
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/api_core/grpc_helpers.py", line 72, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/grpc/_channel.py", line 1030, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/grpc/_channel.py", line 910, in _end_unary_response_blocking
    raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.INTERNAL
        details = "Internal error encountered."
        debug_error_string = "UNKNOWN:Error received from peer ipv6:%5B2a00:1450:400e:800::200a%5D:443 {created_time:"2023-05-09T00:09:30.434205595+02:00", grpc_status:13, grpc_message:"Internal error encountered."}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/.../Desktop/test/mlpipelines/monthly_pipeline.py", line 98, in <module>
    pipeline_job.run(network=None,
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/cloud/aiplatform/pipeline_jobs.py", line 314, in run
    self._run(
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/cloud/aiplatform/base.py", line 814, in wrapper
    return method(*args, **kwargs)
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/cloud/aiplatform/pipeline_jobs.py", line 345, in _run
    self.submit(
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/cloud/aiplatform/pipeline_jobs.py", line 419, in submit
    self._gca_resource = self.api_client.create_pipeline_job(
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/cloud/aiplatform_v1/services/pipeline_service/client.py", line 1347, in create_pipeline_job
    response = rpc(
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/api_core/gapic_v1/method.py", line 113, in __call__
    return wrapped_func(*args, **kwargs)
  File "/home/.../Desktop/test/venv/lib/python3.9/site-packages/google/api_core/grpc_helpers.py", line 74, in error_remapped_callable
    raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.InternalServerError: 500 Internal error encountered.

[matthew29tang]

Thanks for the detailed report! I've filed this as an internal bug and I'll get back to you when I have further updates about this.

[xRagnorokx]

For anyone finding this via google. I encountered a very similar error to this when trying to run a template as a pipeline job.

Turns out that I was not specifying the service account in the job submit call. Adding the service account to that call (i.e job.submit(service_account=pipeline_service_account)) fixed the issue for me.

[kab840]

I have also similar problem.
In my case, the service account for setting to job.submit needs to have roles/artifactregistry.reader to the target artifact registry for uploading pipeline template.

Anyway in my understanding, the service account called in job.submit is for executing vertex ai but for call pipeline template.

[ghost]

Any update on this? I seem to have the same issue

[ghost]

BTW downloading the pipeline YAML from Artifact Registry via tag works fine using the KFP SDK registry functions

[ghost]

Similarly it works fine using a curl request as per the docs

[ghost]

A workaround for anyone with the same issue - first use the KFP SDK to resolve the tag to an exact version, then pass the exact version as template_path:

import re
from kfp.registry import RegistryClient
from google.cloud import aiplatform

_VALID_AR_URL = re.compile(
    r"https://([\w\-]+)-kfp\.pkg\.dev/([\w\-]+)/([\w\-]+)/([\w\-]+)/([\w\-.]+)",
    re.IGNORECASE,
)

template_path = f"https://{region}-kfp.pkg.dev/{PROJECTID}/{REPOSITORY_NAME}/{PIPELINE_NAME}/{TAG}"

match = _VALID_AR_URL.match(template_path)
if match and "sha256:" not in template_path:
    region = match.group(1)
    project = match.group(2)
    repo = match.group(3)
    package_name = match.group(4)
    tag = match.group(5)
    host = f"https://{region}-kfp.pkg.dev/{project}/{repo}"
    client = RegistryClient(host=host)
    metadata = client.get_tag(package_name, tag)
    version = metadata["version"][metadata["version"].find("sha256:") :]
    template_path = f"{host}/{package_name}/{version}"

# Instantiate PipelineJob object
pl = aiplatform.pipeline_jobs.PipelineJob(
    template_path=template_path,
    ...
)