feat(sdk): Add pipeline_task_name to PipelineTaskFinalStatus

sdk/python/kfp/compiler_cli_tests/test_data/pipeline_with_task_final_status.py

@@ -20,14 +20,12 @@
 from kfp.dsl import PipelineTaskFinalStatus
 
 
-@component(
-    # TODO: remove once the kfp executor change is released.
-    kfp_package_path='git+https://github.com/kubeflow/pipelines.git@master#subdirectory=sdk/python',
-)
+@component
 def exit_op(user_input: str, status: PipelineTaskFinalStatus):
     """Checks pipeline run status."""
     print('Pipeline status: ', status.state)
     print('Job resource name: ', status.pipeline_job_resource_name)
+    print('Pipeline task name: ', status.pipeline_task_name)
     print('Error code: ', status.error_code)
     print('Error message: ', status.error_message)
 

sdk/python/kfp/compiler_cli_tests/test_data/pipeline_with_task_final_status.yaml

@@ -4,9 +4,14 @@ sdkVersion: kfp-2.0.0-alpha.1
 schemaVersion: 2.1.0
 deploymentSpec:
   executors:
-    exec-print-op:
+    exec-exit-op:
       container:
         image: python:3.7
+        args:
+        - --executor_input
+        - '{{$}}'
+        - --function_to_execute
+        - exit_op
         command:
         - sh
         - -c
@@ -24,16 +29,19 @@ deploymentSpec:
 
           '
         - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
-          \ *\n\ndef print_op(message: str):\n    \"\"\"Prints a message.\"\"\"\n\
-          \    print(message)\n\n"
+          \ *\n\ndef exit_op(user_input: str, status: PipelineTaskFinalStatus):\n\
+          \    \"\"\"Checks pipeline run status.\"\"\"\n    print('Pipeline status:\
+          \ ', status.state)\n    print('Job resource name: ', status.pipeline_job_resource_name)\n\
+          \    print('Pipeline task name: ', status.pipeline_task_name)\n    print('Error\
+          \ code: ', status.error_code)\n    print('Error message: ', status.error_message)\n\
+          \n"
+    exec-fail-op:
+      container:
         args:
         - --executor_input
         - '{{$}}'
         - --function_to_execute
-        - print_op
-    exec-fail-op:
-      container:
-        image: python:3.7
+        - fail_op
         command:
         - sh
         - -c
@@ -53,20 +61,15 @@ deploymentSpec:
         - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
           \ *\n\ndef fail_op(message: str):\n    \"\"\"Fails.\"\"\"\n    import sys\n\
           \    print(message)\n    sys.exit(1)\n\n"
-        args:
-        - --executor_input
-        - '{{$}}'
-        - --function_to_execute
-        - fail_op
-    exec-exit-op:
-      container:
         image: python:3.7
+    exec-print-op:
+      container:
         command:
         - sh
         - -c
         - "\nif ! [ -x \"$(command -v pip)\" ]; then\n    python3 -m ensurepip ||\
           \ python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1\
-          \ python3 -m pip install --quiet     --no-warn-script-location 'git+https://github.com/kubeflow/pipelines.git@master#subdirectory=sdk/python'\
+          \ python3 -m pip install --quiet     --no-warn-script-location 'kfp==2.0.0-alpha.1'\
           \ && \"$0\" \"$@\"\n"
         - sh
         - -ec
@@ -78,16 +81,14 @@ deploymentSpec:
 
           '
         - "\nimport kfp\nfrom kfp import dsl\nfrom kfp.dsl import *\nfrom typing import\
-          \ *\n\ndef exit_op(user_input: str, status: PipelineTaskFinalStatus):\n\
-          \    \"\"\"Checks pipeline run status.\"\"\"\n    print('Pipeline status:\
-          \ ', status.state)\n    print('Job resource name: ', status.pipeline_job_resource_name)\n\
-          \    print('Error code: ', status.error_code)\n    print('Error message:\
-          \ ', status.error_message)\n\n"
+          \ *\n\ndef print_op(message: str):\n    \"\"\"Prints a message.\"\"\"\n\
+          \    print(message)\n\n"
         args:
         - --executor_input
         - '{{$}}'
         - --function_to_execute
-        - exit_op
+        - print_op
+        image: python:3.7
 components:
   comp-exit-handler-1:
     inputDefinitions:
@@ -125,12 +126,6 @@ components:
         message:
           parameterType: STRING
     executorLabel: exec-print-op
-  comp-fail-op:
-    inputDefinitions:
-      parameters:
-        message:
-          parameterType: STRING
-    executorLabel: exec-fail-op
   comp-exit-op:
     inputDefinitions:
       parameters:
@@ -139,6 +134,12 @@ components:
         status:
           parameterType: STRUCT
     executorLabel: exec-exit-op
+  comp-fail-op:
+    inputDefinitions:
+      parameters:
+        message:
+          parameterType: STRING
+    executorLabel: exec-fail-op
 root:
   inputDefinitions:
     parameters:
@@ -147,15 +148,6 @@ root:
         defaultValue: Hello World!
   dag:
     tasks:
-      exit-handler-1:
-        taskInfo:
-          name: my-pipeline
-        inputs:
-          parameters:
-            pipelinechannel--message:
-              componentInputParameter: message
-        componentRef:
-          name: comp-exit-handler-1
       exit-op:
         taskInfo:
           name: exit-op
@@ -174,3 +166,12 @@ root:
           name: comp-exit-op
         triggerPolicy:
           strategy: ALL_UPSTREAM_TASKS_COMPLETED
+      exit-handler-1:
+        taskInfo:
+          name: my-pipeline
+        inputs:
+          parameters:
+            pipelinechannel--message:
+              componentInputParameter: message
+        componentRef:
+          name: comp-exit-handler-1

sdk/python/kfp/components/executor.py

@@ -269,6 +269,7 @@ def execute(self):
                     state=value.get('state'),
                     pipeline_job_resource_name=value.get(
                         'pipelineJobResourceName'),
+                    pipeline_task_name=value.get('pipelineTaskName'),
                     error_code=value.get('error').get('code', None),
                     error_message=value.get('error').get('message', None),
                 )

sdk/python/kfp/components/executor_test.py

@@ -652,7 +652,7 @@ def test_function_with_pipeline_task_final_status(self):
     {
       "inputs": {
         "parameterValues": {
-          "status": {"error":{"code":9,"message":"The DAG failed because some tasks failed. The failed tasks are: [fail-op]."},"pipelineJobResourceName":"projects/123/locations/us-central1/pipelineJobs/pipeline-456","state":"FAILED"}
+          "status": {"error":{"code":9,"message":"The DAG failed because some tasks failed. The failed tasks are: [fail-op]."},"pipelineJobResourceName":"projects/123/locations/us-central1/pipelineJobs/pipeline-456", "pipelineTaskName": "upstream-task", "state":"FAILED"}
         }
       },
       "outputs": {
@@ -669,6 +669,7 @@ def test_function_with_pipeline_task_final_status(self):
         def test_func(status: PipelineTaskFinalStatus) -> str:
             return (f'Pipeline status: {status.state}\n'
                     f'Job resource name: {status.pipeline_job_resource_name}\n'
+                    f'Pipeline task name: {status.pipeline_task_name}\n'
                     f'Error code: {status.error_code}\n'
                     f'Error message: {status.error_message}')
 
@@ -678,12 +679,13 @@ def test_func(status: PipelineTaskFinalStatus) -> str:
             output_metadata = json.loads(f.read())
         self.assertDictEqual(
             output_metadata, {
-                "parameterValues": {
-                    "Output":
-                        "Pipeline status: FAILED\n"
-                        "Job resource name: projects/123/locations/us-central1/pipelineJobs/pipeline-456\n"
-                        "Error code: 9\n"
-                        "Error message: The DAG failed because some tasks failed. The failed tasks are: [fail-op]."
+                'parameterValues': {
+                    'Output':
+                        'Pipeline status: FAILED\n'
+                        'Job resource name: projects/123/locations/us-central1/pipelineJobs/pipeline-456\n'
+                        'Pipeline task name: upstream-task\n'
+                        'Error code: 9\n'
+                        'Error message: The DAG failed because some tasks failed. The failed tasks are: [fail-op].'
                 },
             })
 

sdk/python/kfp/components/task_final_status.py

@@ -29,6 +29,7 @@ class PipelineTaskFinalStatus:
             'SUCCEEDED', 'FAILED' or 'CANCELLED'.
         pipeline_job_resource_name: The pipeline job resource name, in the format
             of `projects/{project}/locations/{location}/pipelineJobs/{pipeline_job}`.
+        pipeline_task_name: The pipeline task that produces this status.
         error_code: In case of error, the oogle.rpc.Code
             https://github.com/googleapis/googleapis/blob/master/google/rpc/code.proto
             If state is 'SUCCEEDED', this is None.
@@ -37,5 +38,6 @@ class PipelineTaskFinalStatus:
     """
     state: str
     pipeline_job_resource_name: str
+    pipeline_task_name: str
     error_code: Optional[int]
     error_message: Optional[str]

sdk/python/requirements.in

@@ -16,7 +16,7 @@ google-cloud-storage>=1.20.0,<2
 # NOTE: Maintainers, please do not require google-auth>=2.x.x
 # Until this issue is closed
 # https://github.com/googleapis/google-cloud-python/issues/10566
-kfp-pipeline-spec>=0.1.13,<0.2.0
+kfp-pipeline-spec>=0.1.14,<0.2.0
 # Update the upper version whenever a new major version of the
 # kfp-server-api package is released.
 # Update the lower version when kfp sdk depends on new apis/fields in

sdk/python/requirements.txt

@@ -55,7 +55,7 @@ idna==3.2
     # via requests
 jsonschema==3.2.0
     # via -r requirements.in
-kfp-pipeline-spec==0.1.13
+kfp-pipeline-spec==0.1.14
     # via -r requirements.in
 kfp-server-api==2.0.0a0
     # via -r requirements.in