MLOps/Pipelines/Vertex AI Pipelines - Notifications.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "14fcf806",
   "metadata": {},
   "source": [
    "![tracker](https://us-central1-vertex-ai-mlops-369716.cloudfunctions.net/pixel-tracking?path=statmike%2Fvertex-ai-mlops%2FMLOps%2FPipelines&file=Vertex+AI+Pipelines+-+Notifications.ipynb)\n",
    "<!--- header table --->\n",
    "<table align=\"left\">\n",
    "  <td style=\"text-align: center\">\n",
    "    <a href=\"https://colab.research.google.com/github/statmike/vertex-ai-mlops/blob/main/MLOps/Pipelines/Vertex%20AI%20Pipelines%20-%20Notifications.ipynb\">\n",
    "      <img src=\"https://cloud.google.com/ml-engine/images/colab-logo-32px.png\" alt=\"Google Colaboratory logo\">\n",
    "      <br>Run in<br>Colab\n",
    "    </a>\n",
    "  </td>\n",
    "  <td style=\"text-align: center\">\n",
    "    <a href=\"https://console.cloud.google.com/vertex-ai/colab/import/https%3A%2F%2Fraw.githubusercontent.com%2Fstatmike%2Fvertex-ai-mlops%2Fmain%2FMLOps%2FPipelines%2FVertex%2520AI%2520Pipelines%2520-%2520Notifications.ipynb\">\n",
    "      <img width=\"32px\" src=\"https://lh3.googleusercontent.com/JmcxdQi-qOpctIvWKgPtrzZdJJK-J3sWE1RsfjZNwshCFgE_9fULcNpuXYTilIR2hjwN\" alt=\"Google Cloud Colab Enterprise logo\">\n",
    "      <br>Run in<br>Colab Enterprise\n",
    "    </a>\n",
    "  </td>      \n",
    "  <td style=\"text-align: center\">\n",
    "    <a href=\"https://github.com/statmike/vertex-ai-mlops/blob/main/MLOps/Pipelines/Vertex%20AI%20Pipelines%20-%20Notifications.ipynb\">\n",
    "      <img src=\"https://cloud.google.com/ml-engine/images/github-logo-32px.png\" alt=\"GitHub logo\">\n",
    "      <br>View on<br>GitHub\n",
    "    </a>\n",
    "  </td>\n",
    "  <td style=\"text-align: center\">\n",
    "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://raw.githubusercontent.com/statmike/vertex-ai-mlops/main/MLOps/Pipelines/Vertex%20AI%20Pipelines%20-%20Notifications.ipynb\">\n",
    "      <img src=\"https://lh3.googleusercontent.com/UiNooY4LUgW_oTvpsNhPpQzsstV5W8F7rYgxgGBD85cWJoLmrOzhVs_ksK_vgx40SHs7jCqkTkCk=e14-rj-sc0xffffff-h130-w32\" alt=\"Vertex AI logo\">\n",
    "      <br>Open in<br>Vertex AI Workbench\n",
    "    </a>\n",
    "  </td>\n",
    "</table>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "804e7893",
   "metadata": {},
   "source": [
    "---\n",
    "This is part of a [series of notebook based workflows](./readme.md) that teach all the ways to use pipelines within Vertex AI. The suggested order and description/reason is:\n",
    "\n",
    "|Link To Section|Notebook Workflow|Description|\n",
    "|---|---|---|\n",
    "||[Vertex AI Pipelines - Start Here](./Vertex%20AI%20Pipelines%20-%20Start%20Here.ipynb)|What are pipelines? Start here to go from code to pipeline and see it in action.|\n",
    "||[Vertex AI Pipelines - Introduction](./Vertex%20AI%20Pipelines%20-%20Introduction.ipynb)|Introduction to pipelines with the console and Vertex AI SDK|\n",
    "||[Vertex AI Pipelines - Components](./Vertex%20AI%20Pipelines%20-%20Components.ipynb)|An introduction to all the ways to create pipeline components from your code|\n",
    "||[Vertex AI Pipelines - IO](./Vertex%20AI%20Pipelines%20-%20IO.ipynb)|An overview of all the type of inputs and outputs for pipeline components|\n",
    "||[Vertex AI Pipelines - Control](./Vertex%20AI%20Pipelines%20-%20Control.ipynb)|An overview of controlling the flow of exectution for pipelines|\n",
    "||[Vertex AI Pipelines - Secret Manager](./Vertex%20AI%20Pipelines%20-%20Secret%20Manager.ipynb)|How to pass sensitive information to pipelines and components|\n",
    "||[Vertex AI Pipelines - GCS Read and Write](./Vertex%20AI%20Pipelines%20-%20GCS%20Read%20and%20Write.ipynb)|How to read/write to GCS from components, including container components.|\n",
    "||[Vertex AI Pipelines - Scheduling](./Vertex%20AI%20Pipelines%20-%20Scheduling.ipynb)|How to schedule pipeline execution|\n",
    "|_**This Notebook**_|[Vertex AI Pipelines - Notifications](./Vertex%20AI%20Pipelines%20-%20Notifications.ipynb)|How to send email notification of pipeline status.|\n",
    "||[Vertex AI Pipelines - Management](./Vertex%20AI%20Pipelines%20-%20Management.ipynb)|Managing, Reusing, and Storing pipelines and components|\n",
    "||[Vertex AI Pipelines - Testing](./Vertex%20AI%20Pipelines%20-%20Testing.ipynb)|Strategies for testing components and pipeliens locally and remotely to aide development.|\n",
    "||[Vertex AI Pipelines - Managing Pipeline Jobs](./Vertex%20AI%20Pipelines%20-%20Managing%20Pipeline%20Jobs.ipynb)|Manage runs of pipelines in an environment: list, check status, filtered list, cancel and delete jobs.|\n",
    "\n",
    "\n",
    "To discover these notebooks as part of an introduction to MLOps orchestration [start here](./readme.md).  To read more about MLOps also check out [the parent folder](../readme.md).\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "35fc3f49-af20-4c31-b7d7-62b62da89677",
   "metadata": {},
   "source": [
    "# Vertex AI Pipelines - Notifications\n",
    "\n",
    "Send an email notification based on pipeline execution:\n",
    "\n",
    "This workflow covers how to send email notifications based on pipelines execution.\n",
    "- pre-built component to send email notificaitons on completion - any/all status\n",
    "- how to gather pipeline status on exit\n",
    "    - custom built component to send emails conditional on pipelines status at exit"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a9eb31c4-5ad5-48d8-8fee-3906611c7c67",
   "metadata": {
    "id": "od_UkDpvRmgD",
    "tags": []
   },
   "source": [
    "---\n",
    "## Colab Setup\n",
    "\n",
    "To run this notebook in Colab run the cells in this section.  Otherwise, skip this section.\n",
    "\n",
    "This cell will authenticate to GCP (follow prompts in the popup)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "82fb9f68-0848-45b4-b34a-92a5cf742b47",
   "metadata": {
    "executionInfo": {
     "elapsed": 195,
     "status": "ok",
     "timestamp": 1683726184843,
     "user": {
      "displayName": "Mike Henderson",
      "userId": "07691629187611687318"
     },
     "user_tz": 240
    },
    "id": "8UO9FnqyKBlF",
    "tags": []
   },
   "outputs": [],
   "source": [
    "PROJECT_ID = 'statmike-mlops-349915' # replace with project ID"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "01bc0f83-ee97-4b2c-bdc4-52c5b6de8f89",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "executionInfo": {
     "elapsed": 68869,
     "status": "ok",
     "timestamp": 1683726253709,
     "user": {
      "displayName": "Mike Henderson",
      "userId": "07691629187611687318"
     },
     "user_tz": 240
    },
    "id": "N98-KK7LRkjm",
    "outputId": "09ec5008-0def-4e1a-c349-c598ee752f78",
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Not a Colab Environment\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    from google.colab import auth\n",
    "    auth.authenticate_user()\n",
    "    !gcloud config set project {PROJECT_ID}\n",
    "    print('Colab authorized to GCP')\n",
    "except Exception:\n",
    "    print('Not a Colab Environment')\n",
    "    pass"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e207b4a-02a4-4954-96c6-c629e544cf52",
   "metadata": {},
   "source": [
    "---\n",
    "## Installs\n",
    "\n",
    "The list `packages` contains tuples of package import names and install names.  If the import name is not found then the install name is used to install quitely for the current user."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "3149d910-2399-43e5-ae4d-6c72b1bd6294",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# tuples of (import name, install name, min_version)\n",
    "packages = [\n",
    "    ('google.cloud.aiplatform', 'google-cloud-aiplatform', '1.51.0'),\n",
    "    ('google_cloud_pipeline_components', 'google-cloud-pipeline-components'),\n",
    "    ('kfp', 'kfp'),\n",
    "    ('google.cloud.pubsub', 'google-cloud-pubsub'),\n",
    "]\n",
    "\n",
    "import importlib\n",
    "install = False\n",
    "for package in packages:\n",
    "    if not importlib.util.find_spec(package[0]):\n",
    "        print(f'installing package {package[1]}')\n",
    "        install = True\n",
    "        !pip install {package[1]} -U -q --user\n",
    "    elif len(package) == 3:\n",
    "        if importlib.metadata.version(package[0]) < package[2]:\n",
    "            print(f'updating package {package[1]}')\n",
    "            install = True\n",
    "            !pip install {package[1]} -U -q --user"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7c106b1a-1e3b-444d-bbda-e97fe12004b0",
   "metadata": {},
   "source": [
    "### API Enablement"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "0ffc453c-53f6-46e2-87bb-ed2a3ea398b6",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "!gcloud services enable aiplatform.googleapis.com\n",
    "!gcloud services enable pubsub.googleapis.com"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30153c0e-8c90-4991-bf15-27046f0522f3",
   "metadata": {},
   "source": [
    "### Restart Kernel (If Installs Occured)\n",
    "\n",
    "After a kernel restart the code submission can start with the next cell after this one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "050ac1c2-17c6-45e8-b144-ff72342a72a0",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "if install:\n",
    "    import IPython\n",
    "    app = IPython.Application.instance()\n",
    "    app.kernel.do_shutdown(True)\n",
    "    IPython.display.display(IPython.display.Markdown(\"\"\"<div class=\\\"alert alert-block alert-warning\\\">\n",
    "        <b>⚠️ The kernel is going to restart. Please wait until it is finished before continuing to the next step. The previous cells do not need to be run again⚠️</b>\n",
    "        </div>\"\"\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "76a43222-36e7-455e-9e49-0da5afdc75a9",
   "metadata": {
    "id": "appt8-yVRtJ1"
   },
   "source": [
    "---\n",
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6d752b49-d0ae-4a52-ab31-68883f911320",
   "metadata": {
    "id": "63mx2EozRxFP"
   },
   "source": [
    "Inputs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "f70bb8f2-49f3-43a5-af47-20b14df69ddf",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 35
    },
    "executionInfo": {
     "elapsed": 2124,
     "status": "ok",
     "timestamp": 1683726390544,
     "user": {
      "displayName": "Mike Henderson",
      "userId": "07691629187611687318"
     },
     "user_tz": 240
    },
    "id": "xzcoXjM5Rky5",
    "outputId": "b3bdcbc1-70d5-472e-aea2-42c74a42efde",
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'statmike-mlops-349915'"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "project = !gcloud config get-value project\n",
    "PROJECT_ID = project[0]\n",
    "PROJECT_ID"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "747b9f97-bf67-4d11-ac0d-1227e7cd49ff",
   "metadata": {
    "executionInfo": {
     "elapsed": 2,
     "status": "ok",
     "timestamp": 1683726390712,
     "user": {
      "displayName": "Mike Henderson",
      "userId": "07691629187611687318"
     },
     "user_tz": 240
    },
    "id": "IxWrFtqYMfku",
    "tags": []
   },
   "outputs": [],
   "source": [
    "REGION = 'us-central1'\n",
    "SERIES = 'mlops'\n",
    "EXPERIMENT = 'pipeline-notifications'\n",
    "\n",
    "# gcs bucket\n",
    "GCS_BUCKET = PROJECT_ID"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ac559644-8a22-4e41-8b9e-c1906f9e3b12",
   "metadata": {
    "id": "LuajVwCiO6Yg"
   },
   "source": [
    "Packages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "cb368b33-6206-45c9-b924-0a2aed91d986",
   "metadata": {
    "executionInfo": {
     "elapsed": 17761,
     "status": "ok",
     "timestamp": 1683726409304,
     "user": {
      "displayName": "Mike Henderson",
      "userId": "07691629187611687318"
     },
     "user_tz": 240
    },
    "id": "LVC7zzSLRk2C",
    "tags": []
   },
   "outputs": [],
   "source": [
    "import os, time, importlib\n",
    "from typing import NamedTuple\n",
    "\n",
    "from google.cloud import aiplatform\n",
    "from google.cloud import pubsub_v1\n",
    "import kfp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "f21ae861-f7ac-449f-90c2-0e4dd9426902",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'2.12.1'"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kfp.__version__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "8c6215b3-7623-4eb9-8acd-a0039621df5c",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'1.78.0'"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "aiplatform.__version__"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "41cba571-1161-4c47-9d59-641bbc92f9e4",
   "metadata": {
    "id": "EyAVFG9TO9H-"
   },
   "source": [
    "Clients"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "f4f561a5-ab73-4d95-abd2-637bc994e672",
   "metadata": {
    "executionInfo": {
     "elapsed": 19,
     "status": "ok",
     "timestamp": 1683726409306,
     "user": {
      "displayName": "Mike Henderson",
      "userId": "07691629187611687318"
     },
     "user_tz": 240
    },
    "id": "L0RPE13LOZce",
    "tags": []
   },
   "outputs": [],
   "source": [
    "# vertex ai clients\n",
    "aiplatform.init(project = PROJECT_ID, location = REGION)\n",
    "\n",
    "# pubsub client\n",
    "pubsub_pubclient = pubsub_v1.PublisherClient()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "45b5e52e-de6e-48a7-9784-0cd1fe40ff9e",
   "metadata": {},
   "source": [
    "parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "a3ad3776-8c47-4cf2-9885-02456d0aba43",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "DIR = f\"temp/{SERIES}-{EXPERIMENT}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "041c7d80-736e-49ed-a205-07e79d2bd3b3",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'1026793852137-compute@developer.gserviceaccount.com'"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "SERVICE_ACCOUNT = !gcloud config list --format='value(core.account)' \n",
    "SERVICE_ACCOUNT = SERVICE_ACCOUNT[0]\n",
    "SERVICE_ACCOUNT"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "82ccd5b6-a445-46fe-ac97-48454fffffbb",
   "metadata": {},
   "source": [
    "environment:\n",
    "- make a local folder for temporary storage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "a0a31627-b190-4c9b-868e-bd162bb3457a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "if not os.path.exists(DIR):\n",
    "    os.makedirs(DIR)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2b451f1f-0d58-409e-bf08-25d52321e2d8",
   "metadata": {},
   "source": [
    "---\n",
    "## Example Components\n",
    "\n",
    "Components that:\n",
    "- generate coin flips with `flip_coin`\n",
    "    - by default it returns flip of a single coin as 'H' or 'T'\n",
    "    - optional input parameter of `num_coins` can be set to number of coins to retrive a string of flips, like 2 => 'HT'\n",
    "- generate dice rolls with `roll_dice`\n",
    "    - by default it returns the face number [1, 6] from a single die roll\n",
    "    - optionn input parameter of `num_dice` an be set to number of dice to retrieve a sum of rolls, like 2 => [2, 12]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "56e31bc5-9a7b-4399-a9ef-d78a5861102c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "@kfp.dsl.component(base_image = 'python:3.10')\n",
    "def flip_coins(num_coins: int = 1) -> str:\n",
    "    import random\n",
    "    flipmap = ['T', 'H']\n",
    "    flips = [flipmap[random.randint(0, 1)] for n in range(num_coins)]\n",
    "    return ''.join(flips)\n",
    "\n",
    "@kfp.dsl.component(base_image = 'python:3.10')\n",
    "def roll_dice(num_dice: int = 1) -> int:\n",
    "    import random\n",
    "    result = sum([random.randint(1,6) for n in range(num_dice)])\n",
    "    return result"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b2734967-9f85-415f-952f-dac2c29380c2",
   "metadata": {},
   "source": [
    "---\n",
    "## Notifications: With Pre-built Components\n",
    "\n",
    "Use pre-built component to send and email on pipeline exit.  This will send email for all pipeline runs it is incorporated in.  For an approach to customize this and only send email for pipeliens that end in specific states see the customized approach later in this workflow.\n",
    "\n",
    "- https://cloud.google.com/vertex-ai/docs/pipelines/email-notifications\n",
    "- https://cloud.google.com/vertex-ai/docs/pipelines/gcpc-list#emailnotification_components\n",
    "- https://google-cloud-pipeline-components.readthedocs.io/en/google-cloud-pipeline-components-2.14.1/api/v1/vertex_notification_email.html#v1.vertex_notification_email.VertexNotificationEmailOp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e5f49c7d",
   "metadata": {},
   "source": [
    "### Load Pre-Built Component"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "16782493-e333-4c36-9dbd-675824852247",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from google_cloud_pipeline_components.v1.vertex_notification_email import VertexNotificationEmailOp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2ae9ad98",
   "metadata": {},
   "source": [
    "### Build Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "50cba01c-325e-4dff-b0d1-1eff14f60f09",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "@kfp.dsl.pipeline(\n",
    "    name = f'{SERIES}-{EXPERIMENT}-notify',\n",
    "    pipeline_root = f'gs://{GCS_BUCKET}/{SERIES}/{EXPERIMENT}/pipeline_root'\n",
    ")\n",
    "def notify_pipeline():\n",
    "    \n",
    "    task_1 = roll_dice()\n",
    "    notify_complete = VertexNotificationEmailOp(recipients = ['statmike@google.com'])\n",
    "    \n",
    "    with kfp.dsl.ExitHandler(exit_task = notify_complete):\n",
    "        task_2 = flip_coins(num_coins = task_1.output)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3865bae1-29ff-4955-92b5-6075e09d5ade",
   "metadata": {},
   "source": [
    "### Compile Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "c3230c96-5129-4d33-ba2d-ed096f2e72a7",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "kfp.compiler.Compiler().compile(\n",
    "    pipeline_func = notify_pipeline,\n",
    "    package_path = f'{DIR}/{SERIES}-{EXPERIMENT}-notify.yaml'\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6db92649-ea4b-4bd7-a42e-4e7b23f435f9",
   "metadata": {},
   "source": [
    "### Create Pipeline Job (With Vertex AI SDK)\n",
    "\n",
    "The compiled pipeline file can be submitted for running with the console or the SDK (shown here).  Check out the details in the documentation [here](https://cloud.google.com/vertex-ai/docs/pipelines/run-pipeline#create_a_pipeline_run) for an overview with the console.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "cb9514a4-ef8f-4e9c-9a3a-f67dcd2f4d5e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "pipeline_job = aiplatform.PipelineJob(\n",
    "    display_name = f\"{SERIES}-{EXPERIMENT}-notify\",\n",
    "    template_path = f\"{DIR}/{SERIES}-{EXPERIMENT}-notify.yaml\",\n",
    "    pipeline_root = f'gs://{GCS_BUCKET}/{SERIES}/{EXPERIMENT}/pipeline_root',\n",
    "    enable_caching = None # True (enabled), False (disable), None (defer to component level caching) \n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eda6ff5e-1dab-4833-a26f-efc215e4489c",
   "metadata": {},
   "source": [
    "### Submit Pipeline Job (On Vertex AI Pipelines)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "59475843-859a-4afc-8866-8c350f40c46f",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating PipelineJob\n",
      "PipelineJob created. Resource name: projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-20250310194004\n",
      "To use this PipelineJob in another session:\n",
      "pipeline_job = aiplatform.PipelineJob.get('projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-20250310194004')\n",
      "View Pipeline Job:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/mlops-pipeline-notifications-notify-20250310194004?project=1026793852137\n"
     ]
    }
   ],
   "source": [
    "response = pipeline_job.submit(\n",
    "    service_account = SERVICE_ACCOUNT\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "b55ec202-c722-4eeb-afd9-f5dcc34f5c5f",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The Dashboard can be viewed here:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/mlops-pipeline-notifications-notify-20250310194004?project=1026793852137\n"
     ]
    }
   ],
   "source": [
    "print(f'The Dashboard can be viewed here:\\n{pipeline_job._dashboard_uri()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "c1dc7159-3194-4704-9b6b-aad4188d997a",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-20250310194004 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-20250310194004 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-20250310194004 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob run completed. Resource name: projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-20250310194004\n"
     ]
    }
   ],
   "source": [
    "pipeline_job.wait()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "57375f12",
   "metadata": {},
   "source": [
    "### Review In Console\n",
    "\n",
    "**Pipeline Job:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/pipeline_email_prebuilt.png\" width=\"75%\">\n",
    "<p>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d4ac49a5",
   "metadata": {},
   "source": [
    "### Review The Email Notification\n",
    "\n",
    "<p align=\"center\">\n",
    "    <img src=\"../../architectures/notebooks/mlops/pipelines/notifications/email_prebuilt.png\" width=\"75%\">\n",
    "<p>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f489ede8-47d5-4f7d-bf57-079ee2fcbbef",
   "metadata": {},
   "source": [
    "---\n",
    "## Gather Pipeline Final Status Details\n",
    "\n",
    "To conditionally send an email we need to capture the pipelines final status and then conditionaly execute an exit task.  This section covers how to capture the pipelines final status details to use with this approach.\n",
    "\n",
    "- https://kubeflow-pipelines.readthedocs.io/en/stable/source/dsl.html#kfp.dsl.PipelineTaskFinalStatus\n",
    "- https://kubeflow-pipelines.readthedocs.io/en/stable/source/dsl.html#kfp.dsl.PipelineTaskFinalStatus"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3eb6a801",
   "metadata": {},
   "source": [
    "### Build Components"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "430154af-d2fc-4cd9-be3a-7e49ebebace9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "@kfp.dsl.component(base_image = 'python:3.10')\n",
    "def exit_op(status: kfp.dsl.PipelineTaskFinalStatus) -> dict:\n",
    "    response = status.__dict__\n",
    "    return response"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "465a5f01-79a5-4e76-875a-62d0b857893b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "@kfp.dsl.component(base_image = 'python:3.10')\n",
    "def force_fail():\n",
    "    import sys\n",
    "    sys.exit(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c759c9ba",
   "metadata": {},
   "source": [
    "### Build Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "f87507aa-2b82-4c3e-875e-8d3b15056fa2",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "@kfp.dsl.pipeline(\n",
    "    name = f'{SERIES}-{EXPERIMENT}-status',\n",
    "    pipeline_root = f'gs://{GCS_BUCKET}/{SERIES}/{EXPERIMENT}/pipeline_root'\n",
    ")\n",
    "def status_pipeline():\n",
    "    \n",
    "    task_status = exit_op()\n",
    "    with kfp.dsl.ExitHandler(exit_task = task_status):\n",
    "        task_1 = roll_dice()\n",
    "        task_fail = force_fail().after(task_1)\n",
    "        task_2 = flip_coins(num_coins = task_1.output)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "69a958d2-a804-4a4b-a3e9-877c16846d19",
   "metadata": {},
   "source": [
    "### Compile Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "b7af3ff4-e42b-457d-b8e1-eccda8ea6ce3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "kfp.compiler.Compiler().compile(\n",
    "    pipeline_func = status_pipeline,\n",
    "    package_path = f'{DIR}/{SERIES}-{EXPERIMENT}-status.yaml'\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "de9891bd-e4a8-469e-99e7-349402bf5855",
   "metadata": {},
   "source": [
    "### Create Pipeline Job (With Vertex AI SDK)\n",
    "\n",
    "The compiled pipeline file can be submitted for running with the console or the SDK (shown here).  Check out the details in the documentation [here](https://cloud.google.com/vertex-ai/docs/pipelines/run-pipeline#create_a_pipeline_run) for an overview with the console.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "bc0350eb-3c6b-4162-98ea-326da22a467a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "pipeline_job = aiplatform.PipelineJob(\n",
    "    display_name = f\"{SERIES}-{EXPERIMENT}-status\",\n",
    "    template_path = f\"{DIR}/{SERIES}-{EXPERIMENT}-status.yaml\",\n",
    "    pipeline_root = f'gs://{GCS_BUCKET}/{SERIES}/{EXPERIMENT}/pipeline_root',\n",
    "    enable_caching = None # True (enabled), False (disable), None (defer to component level caching) \n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "37ee4bb2-8495-45d9-8eaf-b71242f225c0",
   "metadata": {},
   "source": [
    "### Submit Pipeline Job (On Vertex AI Pipelines)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "b1ee99b9-c432-4a9b-8175-638177c67ea2",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating PipelineJob\n",
      "PipelineJob created. Resource name: projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-status-20250310210611\n",
      "To use this PipelineJob in another session:\n",
      "pipeline_job = aiplatform.PipelineJob.get('projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-status-20250310210611')\n",
      "View Pipeline Job:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/mlops-pipeline-notifications-status-20250310210611?project=1026793852137\n"
     ]
    }
   ],
   "source": [
    "response = pipeline_job.submit(\n",
    "    service_account = SERVICE_ACCOUNT\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "da7d9d62-986e-4485-9bfa-c5461dc7677c",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The Dashboard can be viewed here:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/mlops-pipeline-notifications-status-20250310210611?project=1026793852137\n"
     ]
    }
   ],
   "source": [
    "print(f'The Dashboard can be viewed here:\\n{pipeline_job._dashboard_uri()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "bb5de43d-ed73-4b37-916b-afef187670b8",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-status-20250310210611 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-status-20250310210611 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-status-20250310210611 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-status-20250310210611 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-status-20250310210611 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "RuntimeError was raised: Job failed with:\n",
      "code: 9\n",
      "message: \"  The DAG failed because some tasks failed. The failed tasks are: [exit-handler-1].; Job (project_id = statmike-mlops-349915, job_id = 3490294246697598976) is failed due to the above error.; Failed to handle the job: {project_number = 1026793852137, job_id = 3490294246697598976}\"\n",
      "\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    pipeline_job.wait()\n",
    "except Exception as err:\n",
    "    print(f\"{type(err).__name__} was raised: {err}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5e478c5-a0f2-4fa4-94dd-e1bfef4c6fc9",
   "metadata": {},
   "source": [
    "### Retrieve all runs to dataframe:\n",
    "- SDK Refrence: [`aiplatform.get_pipeline_df`](https://cloud.google.com/python/docs/reference/aiplatform/latest/google.cloud.aiplatform#google_cloud_aiplatform_get_pipeline_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "b1f6e3b7-1989-422d-9d01-ea4d31d20479",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>pipeline_name</th>\n",
       "      <th>run_name</th>\n",
       "      <th>param.vmlmd_lineage_integration</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>mlops-pipeline-notifications-status</td>\n",
       "      <td>mlops-pipeline-notifications-status-2025031021...</td>\n",
       "      <td>{'pipeline_run_component': {'location_id': 'us...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>mlops-pipeline-notifications-status</td>\n",
       "      <td>mlops-pipeline-notifications-status-2024061318...</td>\n",
       "      <td>{'pipeline_run_component': {'location_id': 'us...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>mlops-pipeline-notifications-status</td>\n",
       "      <td>mlops-pipeline-notifications-status-2024061318...</td>\n",
       "      <td>{'pipeline_run_component': {'location_id': 'us...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>mlops-pipeline-notifications-status</td>\n",
       "      <td>mlops-pipeline-notifications-status-2024061317...</td>\n",
       "      <td>{'pipeline_run_component': {'parent_task_names...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>mlops-pipeline-notifications-status</td>\n",
       "      <td>mlops-pipeline-notifications-status-2024061317...</td>\n",
       "      <td>{'pipeline_run_component': {'parent_task_names...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                         pipeline_name  \\\n",
       "0  mlops-pipeline-notifications-status   \n",
       "1  mlops-pipeline-notifications-status   \n",
       "2  mlops-pipeline-notifications-status   \n",
       "3  mlops-pipeline-notifications-status   \n",
       "4  mlops-pipeline-notifications-status   \n",
       "\n",
       "                                            run_name  \\\n",
       "0  mlops-pipeline-notifications-status-2025031021...   \n",
       "1  mlops-pipeline-notifications-status-2024061318...   \n",
       "2  mlops-pipeline-notifications-status-2024061318...   \n",
       "3  mlops-pipeline-notifications-status-2024061317...   \n",
       "4  mlops-pipeline-notifications-status-2024061317...   \n",
       "\n",
       "                     param.vmlmd_lineage_integration  \n",
       "0  {'pipeline_run_component': {'location_id': 'us...  \n",
       "1  {'pipeline_run_component': {'location_id': 'us...  \n",
       "2  {'pipeline_run_component': {'location_id': 'us...  \n",
       "3  {'pipeline_run_component': {'parent_task_names...  \n",
       "4  {'pipeline_run_component': {'parent_task_names...  "
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "aiplatform.get_pipeline_df(pipeline = f'{SERIES}-{EXPERIMENT}-status')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "4922d1b5-d995-4809-b868-27453c30b1a6",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "mlops-pipeline-notifications-status-20250310210611 State.FAILED\n",
      "exit-handler-1 State.FAILED\n",
      "force-fail State.FAILED\n",
      "exit-op State.SUCCEEDED\n",
      "roll-dice State.SUCCEEDED\n",
      "flip-coins State.SUCCEEDED\n"
     ]
    }
   ],
   "source": [
    "for task in pipeline_job.task_details:\n",
    "    print(task.task_name, task.state)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e5694070-dacf-4e70-99c3-d1b10f792ea6",
   "metadata": {},
   "source": [
    "---\n",
    "## Notifications: Use Pipeline Status To Conditionally Send Email\n",
    "\n",
    "Conditionally send an email when pipeline jobs have a status of 'FAILED':\n",
    "\n",
    "Idea:\n",
    "- Create Pub/Sub Topic\n",
    "- Create Pipeline Component that sends subject and body to Pub/Sub Topic\n",
    "- Create Application Integration that listens to Pub/Sub Topic and Sends Emails\n",
    "- This can be reused by many/all pipelines\n",
    "\n",
    "Motivation: https://cloud.google.com/application-integration/docs/listen-pub-sub-topic-send-email\n",
    "\n",
    "Notes:\n",
    "This is one approach.  It could be modified to use an SMTP service to send email or use a service like SendGrid.  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e7df42cb-c675-4bd2-8720-b8e8a9bc3fb3",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Create Pub/Sub Topic\n",
    "\n",
    "The main concepts:\n",
    "- Topic - a feed of messages\n",
    "     - Publish - send a new message to a topic\n",
    "     - Subscription - receive messages that arrive on topic\n",
    "          - Push - the subscriber has new messages pushed to it\n",
    "          - Pull - the subscriber request new messages by pulling them\n",
    "          \n",
    "In this example, a topic will be set up for receiving new event entries for the tracking pixel.  Publishing a new message to this topic will trigger a data load to BigQuery by the Cloud Function (setup below).  The Cloud Funtion will have a push subscription to the topic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "cd6b6658-359b-4d4f-8d2d-c402da54031b",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'mlops-pipeline-notifications'"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "PUBSUB_TOPIC = f'{SERIES}-{EXPERIMENT}'\n",
    "PUBSUB_TOPIC"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "7cd1f609-deaf-42c9-acde-21ddd1089553",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "projects/statmike-mlops-349915/topics/mlops-pipeline-notifications\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    topic = pubsub_pubclient.get_topic(\n",
    "        topic = pubsub_pubclient.topic_path(PROJECT_ID, PUBSUB_TOPIC)\n",
    "    )\n",
    "    print(topic.name)\n",
    "except Exception:\n",
    "    topic = pubsub_pubclient.create_topic(\n",
    "        name = pubsub_pubclient.topic_path(PROJECT_ID, PUBSUB_TOPIC)\n",
    "    )\n",
    "    print(topic.name)   "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8bf3050-dfe7-4d32-b02f-149ce840f52a",
   "metadata": {},
   "source": [
    "### Build Component For Exit Handling\n",
    "\n",
    "- Needs to receive `fp.dsl.PipelineTaskFinalStatus`\n",
    "- prepare strings for email subject and body\n",
    "- send email content to pub/sub topic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "id": "8c57b942-581b-44df-ac2e-7ebb738191f7",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "@kfp.dsl.component(\n",
    "    base_image = 'python:3.10',\n",
    "    packages_to_install = ['google-cloud-pubsub', 'google-cloud-aiplatform']\n",
    ")\n",
    "def email_on_failure(recipients: list, status: kfp.dsl.PipelineTaskFinalStatus) -> dict:\n",
    "    response = status.__dict__\n",
    "\n",
    "    # if failure, send email\n",
    "    if status.state == 'FAILED':\n",
    "\n",
    "        # retrieve pipeline:\n",
    "        from google.cloud import aiplatform\n",
    "        aiplatform.init(\n",
    "            project = status.pipeline_job_resource_name.split('/')[1],\n",
    "            location = status.pipeline_job_resource_name.split('/')[3]\n",
    "        )\n",
    "        pipeline_job = aiplatform.PipelineJob.get(resource_name = status.pipeline_job_resource_name)\n",
    "\n",
    "        # prepare email subject and body:\n",
    "        email = dict(\n",
    "            recipients = ','.join(recipients),\n",
    "            subject = f'Vertex AI Pipeline Job Failed: {pipeline_job.name}',\n",
    "            body = (\n",
    "f'''Hello Vertex AI Pipelines Reviewer,\n",
    "\n",
    "Vertex AI Pipelines Job \"{pipeline_job.display_name}\" experienced a failure.\n",
    "\n",
    "Additional Details:\n",
    "- Project: {status.pipeline_job_resource_name.split('/')[1]}\n",
    "- Pipeline Name: {pipeline_job.display_name}\n",
    "- Pipeline job ID: {pipeline_job.name}\n",
    "- Start Time: {pipeline_job.create_time.isoformat(timespec='microseconds')}\n",
    "\n",
    "To view this pipeline job in Cloud Console, use the following link:\n",
    "https://console.cloud.google.com/vertex-ai/locations/{status.pipeline_job_resource_name.split('/')[3]}/pipelines/runs/{pipeline_job.name}?project={status.pipeline_job_resource_name.split('/')[1]}\n",
    "\n",
    "Sincerely,\n",
    "The MLOps Team\n",
    "''')\n",
    "        )\n",
    "\n",
    "        # send email details to pub/sub\n",
    "        from google.cloud import pubsub_v1\n",
    "        import json\n",
    "        pubsub_pubclient = pubsub_v1.PublisherClient()\n",
    "        message = json.dumps(email).encode('utf-8')\n",
    "        future = pubsub_pubclient.publish(\n",
    "            'projects/statmike-mlops-349915/topics/mlops-pipeline-notifications', # hardcoded, might be better to parameterize\n",
    "            message,\n",
    "            trigger = 'manual'\n",
    "        )\n",
    "        \n",
    "        # add email and pub/sub future to the component response\n",
    "        response = response | email | dict(future = future.result())\n",
    "    \n",
    "    return response"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a297b897-670a-4c7a-bfeb-f20a8a3239c5",
   "metadata": {},
   "source": [
    "### Create Application Integration To Send Emails\n",
    "\n",
    "These steps are conducted in the Application Integration Cloud Console Page:\n",
    "- [Go to Application Integration](https://console.cloud.google.com/integrations)\n",
    "    - In the navigation menu, select Overview\n",
    "        - Setup Application Integrations\n",
    "        - Select Region, 'us-central1' for this workflow\n",
    "        - Click 'Quick Setup'\n",
    "    - In the navigation menu, select Integrations\n",
    "        - Select 'Create Integration'\n",
    "        - Enter a name: 'mlops-notifications'\n",
    "        - Click 'Create'\n",
    "    - Build the integration with a Trigger and Tasks:\n",
    "        - Trigger: In the editor, select 'Triggers' then 'Cloud Pub/Sub Trigger'\n",
    "            - Click to place the element on the designer surface\n",
    "            - Configure the Cloud Pub/Sub trigger:\n",
    "                - Click the element on the designer surface\n",
    "                - Enter the Pub/Sub topic name in the right pane\n",
    "                - Select a service account\n",
    "                - Click to grant roles to service account\n",
    "        - Task 1: Map the pub/sub messages data to JSON\n",
    "            - In the editor, select 'Tasks' then 'Data Mapping'\n",
    "            - Click to place the element on the designer surface (below the Cloud Pub/Sub Trigger)\n",
    "            - Click the 'Data Mapping' element on the designer surface\n",
    "            - Click 'Open Data Mapping Editor' on the right pane\n",
    "                - Expand the `CloudPubSubMessage` JSON variable in the 'Local Variables' list on the left pane\n",
    "                - Click and drag `CloudPubSUbMessaage.data` to the 'Input' row as the first element\n",
    "        - Task 2: Map the JSON to Variables for recipients, subject, body\n",
    "        - Task 3: Send An Email\n",
    "    - Publish the integration\n",
    "\n",
    " Screenshot Overviews of the process above:\n",
    "\n",
    "**Overall Integration Flow:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/app_int_overall.png\" width=\"75%\">\n",
    "<p>\n",
    "\n",
    "**Create Pub/Sub Trigger:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/app_int_trigger.png\" width=\"75%\">\n",
    "<p>\n",
    "\n",
    "**Map Pub/Sub To JSON:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/app_int_map1.png\" width=\"75%\">\n",
    "<p>\n",
    "\n",
    "**Map JSON to Variables:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/app_int_map2.png\" width=\"75%\">\n",
    "<p>\n",
    "\n",
    "**Send Email:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/app_int_sendemail.png\" width=\"75%\">\n",
    "<p>\n",
    "\n",
    "**Publish Integration:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/app_int_publish.png\" width=\"75%\">\n",
    "<p>    \n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "111d3d27-3004-428d-9bcd-061f85c4235f",
   "metadata": {},
   "source": [
    "### Build Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "id": "388473c4-080c-42a5-89f1-b75338477170",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "@kfp.dsl.pipeline(\n",
    "    name = f'{SERIES}-{EXPERIMENT}-notify-on_failure',\n",
    "    pipeline_root = f'gs://{GCS_BUCKET}/{SERIES}/{EXPERIMENT}/pipeline_root'\n",
    ")\n",
    "def email_pipeline(emails: list):\n",
    "    \n",
    "    fail_status = email_on_failure(recipients = emails)\n",
    "    with kfp.dsl.ExitHandler(exit_task = fail_status):\n",
    "        task_1 = roll_dice()\n",
    "        task_fail = force_fail().after(task_1)\n",
    "        task_2 = flip_coins(num_coins = task_1.output)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "beef7557-3caa-4ddc-b295-f15875cd6084",
   "metadata": {},
   "source": [
    "### Compile Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "id": "4b3800d5-616d-4382-ad12-ab8cab826145",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "kfp.compiler.Compiler().compile(\n",
    "    pipeline_func = email_pipeline,\n",
    "    package_path = f'{DIR}/{SERIES}-{EXPERIMENT}-notify-on_failure.yaml'\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9d3173ae-4e69-4726-84f8-7903bff0a284",
   "metadata": {},
   "source": [
    "### Create Pipeline Job (With Vertex AI SDK)\n",
    "\n",
    "The compiled pipeline file can be submitted for running with the console or the SDK (shown here).  Check out the details in the documentation [here](https://cloud.google.com/vertex-ai/docs/pipelines/run-pipeline#create_a_pipeline_run) for an overview with the console.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "id": "00e61670-38e3-4222-a31b-429bba7e07dc",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "pipeline_job = aiplatform.PipelineJob(\n",
    "    display_name = f\"{SERIES}-{EXPERIMENT}-notify-on_failure\",\n",
    "    template_path = f\"{DIR}/{SERIES}-{EXPERIMENT}-notify-on_failure.yaml\",\n",
    "    parameter_values = dict(\n",
    "        emails = ['statmike@google.com']\n",
    "    ),\n",
    "    pipeline_root = f'gs://{GCS_BUCKET}/{SERIES}/{EXPERIMENT}/pipeline_root',\n",
    "    enable_caching = None # True (enabled), False (disable), None (defer to component level caching) \n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "71bb372d-0e38-4e14-b292-44ed714b6f71",
   "metadata": {},
   "source": [
    "### Submit Pipeline Job (On Vertex AI Pipelines)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "id": "d1e56b58-1f3d-4e0a-9b89-1359b8f407fb",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating PipelineJob\n",
      "PipelineJob created. Resource name: projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-on-failure-20240614011200\n",
      "To use this PipelineJob in another session:\n",
      "pipeline_job = aiplatform.PipelineJob.get('projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-on-failure-20240614011200')\n",
      "View Pipeline Job:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/mlops-pipeline-notifications-notify-on-failure-20240614011200?project=1026793852137\n"
     ]
    }
   ],
   "source": [
    "response = pipeline_job.submit(\n",
    "    service_account = SERVICE_ACCOUNT\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "id": "150620b2-9c0d-4a57-8b59-64b5342de500",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The Dashboard can be viewed here:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/mlops-pipeline-notifications-notify-on-failure-20240614011200?project=1026793852137\n"
     ]
    }
   ],
   "source": [
    "print(f'The Dashboard can be viewed here:\\n{pipeline_job._dashboard_uri()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "id": "bc878d7a-3a4b-4588-a6c9-26960c4a76d7",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-on-failure-20240614011200 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-on-failure-20240614011200 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-on-failure-20240614011200 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "PipelineJob projects/1026793852137/locations/us-central1/pipelineJobs/mlops-pipeline-notifications-notify-on-failure-20240614011200 current state:\n",
      "PipelineState.PIPELINE_STATE_RUNNING\n",
      "RuntimeError was raised: Job failed with:\n",
      "code: 9\n",
      "message: \"The DAG failed because some tasks failed. The failed tasks are: [exit-handler-1].; Job (project_id = statmike-mlops-349915, job_id = 1265031283587678208) is failed due to the above error.; Failed to handle the job: {project_number = 1026793852137, job_id = 1265031283587678208}\"\n",
      "\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    pipeline_job.wait()\n",
    "except Exception as err:\n",
    "    print(f\"{type(err).__name__} was raised: {err}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67e768c4",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Review In Console\n",
    "\n",
    "**Pipeline Job:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/pipeline_email_custom.png\" width=\"75%\">\n",
    "<p>\n",
    "\n",
    "**Application Integration Logs:**\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/app_int_log.png\" width=\"75%\">\n",
    "<p>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8fc4d809-2ab7-4803-9f0b-2cc2d17c51d9",
   "metadata": {},
   "source": [
    "### Retrieve all runs to dataframe:\n",
    "- SDK Refrence: [`aiplatform.get_pipeline_df`](https://cloud.google.com/python/docs/reference/aiplatform/latest/google.cloud.aiplatform#google_cloud_aiplatform_get_pipeline_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "id": "acb4f07e-51d0-403c-8247-3b5d3d4ea6e9",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>pipeline_name</th>\n",
       "      <th>run_name</th>\n",
       "      <th>param.vmlmd_lineage_integration</th>\n",
       "      <th>param.input:emails</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure</td>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure...</td>\n",
       "      <td>{'pipeline_run_component': {'task_name': 'mlop...</td>\n",
       "      <td>[statmike@google.com]</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure</td>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure...</td>\n",
       "      <td>{'pipeline_run_component': {'parent_task_names...</td>\n",
       "      <td>[statmike@google.com]</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure</td>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure...</td>\n",
       "      <td>{'pipeline_run_component': {'location_id': 'us...</td>\n",
       "      <td>[statmike@google.com]</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure</td>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure...</td>\n",
       "      <td>{'pipeline_run_component': {'task_name': 'mlop...</td>\n",
       "      <td>[statmike@google.com]</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure</td>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure...</td>\n",
       "      <td>{'pipeline_run_component': {'parent_task_names...</td>\n",
       "      <td>[statmike@google.com]</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure</td>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure...</td>\n",
       "      <td>{'pipeline_run_component': {'project_id': 'sta...</td>\n",
       "      <td>[statmike@google.com]</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure</td>\n",
       "      <td>mlops-pipeline-notifications-notify-on-failure...</td>\n",
       "      <td>{'pipeline_run_component': {'task_name': 'mlop...</td>\n",
       "      <td>[statmike@google.com]</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                    pipeline_name  \\\n",
       "0  mlops-pipeline-notifications-notify-on-failure   \n",
       "1  mlops-pipeline-notifications-notify-on-failure   \n",
       "2  mlops-pipeline-notifications-notify-on-failure   \n",
       "3  mlops-pipeline-notifications-notify-on-failure   \n",
       "4  mlops-pipeline-notifications-notify-on-failure   \n",
       "5  mlops-pipeline-notifications-notify-on-failure   \n",
       "6  mlops-pipeline-notifications-notify-on-failure   \n",
       "\n",
       "                                            run_name  \\\n",
       "0  mlops-pipeline-notifications-notify-on-failure...   \n",
       "1  mlops-pipeline-notifications-notify-on-failure...   \n",
       "2  mlops-pipeline-notifications-notify-on-failure...   \n",
       "3  mlops-pipeline-notifications-notify-on-failure...   \n",
       "4  mlops-pipeline-notifications-notify-on-failure...   \n",
       "5  mlops-pipeline-notifications-notify-on-failure...   \n",
       "6  mlops-pipeline-notifications-notify-on-failure...   \n",
       "\n",
       "                     param.vmlmd_lineage_integration     param.input:emails  \n",
       "0  {'pipeline_run_component': {'task_name': 'mlop...  [statmike@google.com]  \n",
       "1  {'pipeline_run_component': {'parent_task_names...  [statmike@google.com]  \n",
       "2  {'pipeline_run_component': {'location_id': 'us...  [statmike@google.com]  \n",
       "3  {'pipeline_run_component': {'task_name': 'mlop...  [statmike@google.com]  \n",
       "4  {'pipeline_run_component': {'parent_task_names...  [statmike@google.com]  \n",
       "5  {'pipeline_run_component': {'project_id': 'sta...  [statmike@google.com]  \n",
       "6  {'pipeline_run_component': {'task_name': 'mlop...  [statmike@google.com]  "
      ]
     },
     "execution_count": 122,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "aiplatform.get_pipeline_df(pipeline = f'{SERIES}-{EXPERIMENT}-notify-on-failure')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "id": "f40e3911-2b66-44ea-aa0f-d0320c01f703",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "flip-coins State.SKIPPED\n",
      "mlops-pipeline-notifications-notify-on-failure-20240614011200 State.FAILED\n",
      "roll-dice State.SKIPPED\n",
      "email-on-failure State.SUCCEEDED\n",
      "exit-handler-1 State.FAILED\n",
      "force-fail State.FAILED\n"
     ]
    }
   ],
   "source": [
    "for task in pipeline_job.task_details:\n",
    "    print(task.task_name, task.state)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30d274c3-c628-48fd-9038-10cacecc027f",
   "metadata": {},
   "source": [
    "### Review The Email Notification\n",
    "\n",
    "<p align=\"center\">\n",
    "    <img src=\"../resources/images/screenshots/pipelines/notifications/email_custom.png\" width=\"75%\">\n",
    "<p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1bfcbd2d-57bd-49c7-8ffa-08d9d89aca0e",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "821c8a00-da9e-4438-8642-f975cc85e017",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ae73aadb-d156-4fcd-9244-4ccecbc64b13",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fd7d52a0-0194-4844-b73f-9a71ae19c857",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "environment": {
   "kernel": "conda-base-py",
   "name": "workbench-notebooks.m128",
   "type": "gcloud",
   "uri": "us-docker.pkg.dev/deeplearning-platform-release/gcr.io/workbench-notebooks:m128"
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel) (Local)",
   "language": "python",
   "name": "conda-base-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}