tutorials/demo_notebooks/demo_pipeline/demo-pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "# Demo KFP pipeline"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Install requirements:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "%%bash\n",
    "\n",
    "pip install kfp~=1.8.14"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Imports:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "import warnings\n",
    "warnings.filterwarnings(\"ignore\")\n",
    "\n",
    "import kfp\n",
    "import kfp.dsl as dsl\n",
    "from kfp.aws import use_aws_secret\n",
    "from kfp.v2.dsl import (\n",
    "    component,\n",
    "    Input,\n",
    "    Output,\n",
    "    Dataset,\n",
    "    Metrics,\n",
    "    Artifact,\n",
    "    Model\n",
    ")"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "## 1. Connect to client\n",
    "\n",
    "The default way of accessing Kubeflow is via port-forward. This enables you to get started quickly without imposing any requirements on your environment. Run the following to port-forward Istio's Ingress-Gateway to local port `8080`:\n",
    "\n",
    "```sh\n",
    "kubectl port-forward svc/istio-ingressgateway -n istio-system 8080:80\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "source": [
    "import re\n",
    "import requests\n",
    "from urllib.parse import urlsplit\n",
    "\n",
    "def get_istio_auth_session(url: str, username: str, password: str) -> dict:\n",
    "    \"\"\"\n",
    "    Determine if the specified URL is secured by Dex and try to obtain a session cookie.\n",
    "    WARNING: only Dex `staticPasswords` and `LDAP` authentication are currently supported\n",
    "             (we default default to using `staticPasswords` if both are enabled)\n",
    "\n",
    "    :param url: Kubeflow server URL, including protocol\n",
    "    :param username: Dex `staticPasswords` or `LDAP` username\n",
    "    :param password: Dex `staticPasswords` or `LDAP` password\n",
    "    :return: auth session information\n",
    "    \"\"\"\n",
    "    # define the default return object\n",
    "    auth_session = {\n",
    "        \"endpoint_url\": url,    # KF endpoint URL\n",
    "        \"redirect_url\": None,   # KF redirect URL, if applicable\n",
    "        \"dex_login_url\": None,  # Dex login URL (for POST of credentials)\n",
    "        \"is_secured\": None,     # True if KF endpoint is secured\n",
    "        \"session_cookie\": None  # Resulting session cookies in the form \"key1=value1; key2=value2\"\n",
    "    }\n",
    "\n",
    "    # use a persistent session (for cookies)\n",
    "    with requests.Session() as s:\n",
    "\n",
    "        ################\n",
    "        # Determine if Endpoint is Secured\n",
    "        ################\n",
    "        resp = s.get(url, allow_redirects=True)\n",
    "        if resp.status_code != 200:\n",
    "            raise RuntimeError(\n",
    "                f\"HTTP status code '{resp.status_code}' for GET against: {url}\"\n",
    "            )\n",
    "\n",
    "        auth_session[\"redirect_url\"] = resp.url\n",
    "\n",
    "        # if we were NOT redirected, then the endpoint is UNSECURED\n",
    "        if len(resp.history) == 0:\n",
    "            auth_session[\"is_secured\"] = False\n",
    "            return auth_session\n",
    "        else:\n",
    "            auth_session[\"is_secured\"] = True\n",
    "\n",
    "        ################\n",
    "        # Get Dex Login URL\n",
    "        ################\n",
    "        redirect_url_obj = urlsplit(auth_session[\"redirect_url\"])\n",
    "\n",
    "        # if we are at `/auth?=xxxx` path, we need to select an auth type\n",
    "        if re.search(r\"/auth$\", redirect_url_obj.path):\n",
    "\n",
    "            #######\n",
    "            # TIP: choose the default auth type by including ONE of the following\n",
    "            #######\n",
    "\n",
    "            # OPTION 1: set \"staticPasswords\" as default auth type\n",
    "            redirect_url_obj = redirect_url_obj._replace(\n",
    "                path=re.sub(r\"/auth$\", \"/auth/local\", redirect_url_obj.path)\n",
    "            )\n",
    "            # OPTION 2: set \"ldap\" as default auth type\n",
    "            # redirect_url_obj = redirect_url_obj._replace(\n",
    "            #     path=re.sub(r\"/auth$\", \"/auth/ldap\", redirect_url_obj.path)\n",
    "            # )\n",
    "\n",
    "        # if we are at `/auth/xxxx/login` path, then no further action is needed (we can use it for login POST)\n",
    "        if re.search(r\"/auth/.*/login$\", redirect_url_obj.path):\n",
    "            auth_session[\"dex_login_url\"] = redirect_url_obj.geturl()\n",
    "\n",
    "        # else, we need to be redirected to the actual login page\n",
    "        else:\n",
    "            # this GET should redirect us to the `/auth/xxxx/login` path\n",
    "            resp = s.get(redirect_url_obj.geturl(), allow_redirects=True)\n",
    "            if resp.status_code != 200:\n",
    "                raise RuntimeError(\n",
    "                    f\"HTTP status code '{resp.status_code}' for GET against: {redirect_url_obj.geturl()}\"\n",
    "                )\n",
    "\n",
    "            # set the login url\n",
    "            auth_session[\"dex_login_url\"] = resp.url\n",
    "\n",
    "        ################\n",
    "        # Attempt Dex Login\n",
    "        ################\n",
    "        resp = s.post(\n",
    "            auth_session[\"dex_login_url\"],\n",
    "            data={\"login\": username, \"password\": password},\n",
    "            allow_redirects=True\n",
    "        )\n",
    "        if len(resp.history) == 0:\n",
    "            raise RuntimeError(\n",
    "                f\"Login credentials were probably invalid - \"\n",
    "                f\"No redirect after POST to: {auth_session['dex_login_url']}\"\n",
    "            )\n",
    "\n",
    "        # store the session cookies in a \"key1=value1; key2=value2\" string\n",
    "        auth_session[\"session_cookie\"] = \"; \".join([f\"{c.name}={c.value}\" for c in s.cookies])\n",
    "\n",
    "    return auth_session"
   ],
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "source": [
    "import kfp\n",
    "\n",
    "KUBEFLOW_ENDPOINT = \"http://localhost:8080\"\n",
    "KUBEFLOW_USERNAME = \"user@example.com\"\n",
    "KUBEFLOW_PASSWORD = \"12341234\"\n",
    "\n",
    "auth_session = get_istio_auth_session(\n",
    "    url=KUBEFLOW_ENDPOINT,\n",
    "    username=KUBEFLOW_USERNAME,\n",
    "    password=KUBEFLOW_PASSWORD\n",
    ")\n",
    "\n",
    "client = kfp.Client(host=f\"{KUBEFLOW_ENDPOINT}/pipeline\", cookies=auth_session[\"session_cookie\"])\n",
    "# print(client.list_experiments())"
   ],
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "## 2. Components\n",
    "\n",
    "There are different ways to define components in KFP. Here, we use the **@component** decorator to define the components as Python function-based components.\n",
    "\n",
    "The **@component** annotation converts the function into a factory function that creates pipeline steps that execute this function. This example also specifies the base container image to run you component in."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Pull data component:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "@component(\n",
    "    base_image=\"python:3.10\",\n",
    "    packages_to_install=[\"numpy~=1.26.4\", \"pandas~=1.4.2\"],\n",
    "    output_component_file='components/pull_data_component.yaml',\n",
    ")\n",
    "def pull_data(url: str, data: Output[Dataset]):\n",
    "    \"\"\"\n",
    "    Pull data component.\n",
    "    \"\"\"\n",
    "    import pandas as pd\n",
    "\n",
    "    df = pd.read_csv(url, sep=\";\")\n",
    "    df.to_csv(data.path, index=None)"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Preprocess component:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "@component(\n",
    "    base_image=\"python:3.10\",\n",
    "    packages_to_install=[\"numpy~=1.26.4\", \"pandas~=1.4.2\", \"scikit-learn~=1.0.2\"],\n",
    "    output_component_file='components/preprocess_component.yaml',\n",
    ")\n",
    "def preprocess(\n",
    "    data: Input[Dataset],\n",
    "    scaler_out: Output[Artifact],\n",
    "    train_set: Output[Dataset],\n",
    "    test_set: Output[Dataset],\n",
    "    target: str = \"quality\",\n",
    "):\n",
    "    \"\"\"\n",
    "    Preprocess component.\n",
    "    \"\"\"\n",
    "    import pandas as pd\n",
    "    import pickle\n",
    "    from sklearn.model_selection import train_test_split\n",
    "    from sklearn.preprocessing import StandardScaler\n",
    "\n",
    "    data = pd.read_csv(data.path)\n",
    "\n",
    "    # Split the data into training and test sets. (0.75, 0.25) split.\n",
    "    train, test = train_test_split(data)\n",
    "\n",
    "    scaler = StandardScaler()\n",
    "\n",
    "    train[train.drop(target, axis=1).columns] = scaler.fit_transform(train.drop(target, axis=1))\n",
    "    test[test.drop(target, axis=1).columns] = scaler.transform(test.drop(target, axis=1))\n",
    "\n",
    "    with open(scaler_out.path, 'wb') as fp:\n",
    "        pickle.dump(scaler, fp, pickle.HIGHEST_PROTOCOL)\n",
    "\n",
    "    train.to_csv(train_set.path, index=None)\n",
    "    test.to_csv(test_set.path, index=None)"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Train component:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "from typing import NamedTuple\n",
    "\n",
    "@component(\n",
    "    base_image=\"python:3.10\",\n",
    "    packages_to_install=[\"numpy~=1.26.4\", \"pandas~=1.4.2\", \"scikit-learn~=1.0.2\", \"mlflow~=2.4.1\", \"boto3~=1.21.0\"],\n",
    "    output_component_file='components/train_component.yaml',\n",
    ")\n",
    "def train(\n",
    "    train_set: Input[Dataset],\n",
    "    test_set: Input[Dataset],\n",
    "    saved_model: Output[Model],\n",
    "    mlflow_experiment_name: str,\n",
    "    mlflow_tracking_uri: str,\n",
    "    mlflow_s3_endpoint_url: str,\n",
    "    model_name: str,\n",
    "    alpha: float,\n",
    "    l1_ratio: float,\n",
    "    target: str = \"quality\",\n",
    ") -> NamedTuple(\"Output\", [('storage_uri', str), ('run_id', str),]):\n",
    "    \"\"\"\n",
    "    Train component.\n",
    "    \"\"\"\n",
    "    import numpy as np\n",
    "    import pandas as pd\n",
    "    from sklearn.linear_model import ElasticNet\n",
    "    from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score\n",
    "    import mlflow\n",
    "    import mlflow.sklearn\n",
    "    import os\n",
    "    import logging\n",
    "    import pickle\n",
    "    from collections import namedtuple\n",
    "\n",
    "    logging.basicConfig(level=logging.INFO)\n",
    "    logger = logging.getLogger(__name__)\n",
    "\n",
    "    def eval_metrics(actual, pred):\n",
    "        rmse = np.sqrt(mean_squared_error(actual, pred))\n",
    "        mae = mean_absolute_error(actual, pred)\n",
    "        r2 = r2_score(actual, pred)\n",
    "        return rmse, mae, r2\n",
    "\n",
    "    os.environ['MLFLOW_S3_ENDPOINT_URL'] = mlflow_s3_endpoint_url\n",
    "\n",
    "    # load data\n",
    "    train = pd.read_csv(train_set.path)\n",
    "    test = pd.read_csv(test_set.path)\n",
    "\n",
    "    # The predicted column is \"quality\" which is a scalar from [3, 9]\n",
    "    train_x = train.drop([target], axis=1)\n",
    "    test_x = test.drop([target], axis=1)\n",
    "    train_y = train[[target]]\n",
    "    test_y = test[[target]]\n",
    "\n",
    "    logger.info(f\"Using MLflow tracking URI: {mlflow_tracking_uri}\")\n",
    "    mlflow.set_tracking_uri(mlflow_tracking_uri)\n",
    "\n",
    "    logger.info(f\"Using MLflow experiment: {mlflow_experiment_name}\")\n",
    "    mlflow.set_experiment(mlflow_experiment_name)\n",
    "\n",
    "    with mlflow.start_run() as run:\n",
    "\n",
    "        run_id = run.info.run_id\n",
    "        logger.info(f\"Run ID: {run_id}\")\n",
    "\n",
    "        model = ElasticNet(alpha=alpha, l1_ratio=l1_ratio, random_state=42)\n",
    "\n",
    "        logger.info(\"Fitting model...\")\n",
    "        model.fit(train_x, train_y)\n",
    "\n",
    "        logger.info(\"Predicting...\")\n",
    "        predicted_qualities = model.predict(test_x)\n",
    "\n",
    "        (rmse, mae, r2) = eval_metrics(test_y, predicted_qualities)\n",
    "\n",
    "        logger.info(\"Elasticnet model (alpha=%f, l1_ratio=%f):\" % (alpha, l1_ratio))\n",
    "        logger.info(\"  RMSE: %s\" % rmse)\n",
    "        logger.info(\"  MAE: %s\" % mae)\n",
    "        logger.info(\"  R2: %s\" % r2)\n",
    "\n",
    "        logger.info(\"Logging parameters to MLflow\")\n",
    "        mlflow.log_param(\"alpha\", alpha)\n",
    "        mlflow.log_param(\"l1_ratio\", l1_ratio)\n",
    "        mlflow.log_metric(\"rmse\", rmse)\n",
    "        mlflow.log_metric(\"r2\", r2)\n",
    "        mlflow.log_metric(\"mae\", mae)\n",
    "\n",
    "        # save model to mlflow\n",
    "        logger.info(\"Logging trained model\")\n",
    "        mlflow.sklearn.log_model(\n",
    "            model,\n",
    "            model_name,\n",
    "            registered_model_name=\"ElasticnetWineModel\",\n",
    "            serialization_format=\"pickle\"\n",
    "        )\n",
    "\n",
    "        logger.info(\"Logging predictions artifact to MLflow\")\n",
    "        np.save(\"predictions.npy\", predicted_qualities)\n",
    "        mlflow.log_artifact(\n",
    "        local_path=\"predictions.npy\", artifact_path=\"predicted_qualities/\"\n",
    "        )\n",
    "\n",
    "        # save model as KFP artifact\n",
    "        logging.info(f\"Saving model to: {saved_model.path}\")\n",
    "        with open(saved_model.path, 'wb') as fp:\n",
    "            pickle.dump(model, fp, pickle.HIGHEST_PROTOCOL)\n",
    "\n",
    "        # prepare output\n",
    "        output = namedtuple('Output', ['storage_uri', 'run_id'])\n",
    "\n",
    "        # return str(mlflow.get_artifact_uri())\n",
    "        return output(mlflow.get_artifact_uri(), run_id)"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Evaluate component:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "@component(\n",
    "    base_image=\"python:3.10\",\n",
    "    packages_to_install=[\"numpy\", \"mlflow~=2.4.1\"],\n",
    "    output_component_file='components/evaluate_component.yaml',\n",
    ")\n",
    "def evaluate(\n",
    "    run_id: str,\n",
    "    mlflow_tracking_uri: str,\n",
    "    threshold_metrics: dict\n",
    ") -> bool:\n",
    "    \"\"\"\n",
    "    Evaluate component: Compares metrics from training with given thresholds.\n",
    "\n",
    "    Args:\n",
    "        run_id (string):  MLflow run ID\n",
    "        mlflow_tracking_uri (string): MLflow tracking URI\n",
    "        threshold_metrics (dict): Minimum threshold values for each metric\n",
    "    Returns:\n",
    "        Bool indicating whether evaluation passed or failed.\n",
    "    \"\"\"\n",
    "    from mlflow.tracking import MlflowClient\n",
    "    import logging\n",
    "\n",
    "    logging.basicConfig(level=logging.INFO)\n",
    "    logger = logging.getLogger(__name__)\n",
    "\n",
    "    client = MlflowClient(tracking_uri=mlflow_tracking_uri)\n",
    "    info = client.get_run(run_id)\n",
    "    training_metrics = info.data.metrics\n",
    "\n",
    "    logger.info(f\"Training metrics: {training_metrics}\")\n",
    "\n",
    "    # compare the evaluation metrics with the defined thresholds\n",
    "    for key, value in threshold_metrics.items():\n",
    "        if key not in training_metrics or training_metrics[key] > value:\n",
    "            logger.error(f\"Metric {key} failed. Evaluation not passed!\")\n",
    "            return False\n",
    "    return True"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Deploy model component:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "@component(\n",
    "    base_image=\"python:3.9\",\n",
    "    packages_to_install=[\"kserve==0.11.0\"],\n",
    "    output_component_file='components/deploy_model_component.yaml',\n",
    ")\n",
    "def deploy_model(model_name: str, storage_uri: str):\n",
    "    \"\"\"\n",
    "    Deploy the model as an inference service with Kserve.\n",
    "    \"\"\"\n",
    "    import logging\n",
    "    from kubernetes import client\n",
    "    from kserve import KServeClient\n",
    "    from kserve import constants\n",
    "    from kserve import utils\n",
    "    from kserve import V1beta1InferenceService\n",
    "    from kserve import V1beta1InferenceServiceSpec\n",
    "    from kserve import V1beta1PredictorSpec\n",
    "    from kserve import V1beta1SKLearnSpec\n",
    "    from kubernetes.client import V1ResourceRequirements\n",
    "\n",
    "    logging.basicConfig(level=logging.INFO)\n",
    "    logger = logging.getLogger(__name__)\n",
    "\n",
    "    model_uri = f\"{storage_uri}/{model_name}\"\n",
    "    logger.info(f\"MODEL URI: {model_uri}\")\n",
    "\n",
    "    namespace = utils.get_default_target_namespace()\n",
    "    kserve_version='v1beta1'\n",
    "    api_version = constants.KSERVE_GROUP + '/' + kserve_version\n",
    "\n",
    "    isvc = V1beta1InferenceService(\n",
    "        api_version = api_version,\n",
    "        kind = constants.KSERVE_KIND,\n",
    "        metadata = client.V1ObjectMeta(\n",
    "            name = model_name,\n",
    "            namespace = namespace,\n",
    "            annotations = {'sidecar.istio.io/inject':'false'}\n",
    "        ),\n",
    "        spec = V1beta1InferenceServiceSpec(\n",
    "            predictor=V1beta1PredictorSpec(\n",
    "                service_account_name=\"kserve-sa\",\n",
    "                min_replicas=1,\n",
    "                max_replicas = 1,\n",
    "                sklearn=V1beta1SKLearnSpec(\n",
    "                    storage_uri=model_uri,\n",
    "                    resources=V1ResourceRequirements(\n",
    "                        requests={\"cpu\": \"100m\", \"memory\": \"512Mi\"},\n",
    "                        limits={\"cpu\": \"300m\", \"memory\": \"512Mi\"}\n",
    "                    )\n",
    "                ),\n",
    "            )\n",
    "        )\n",
    "    )\n",
    "    KServe = KServeClient()\n",
    "    KServe.create(isvc)"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Inference component:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "@component(\n",
    "    base_image=\"python:3.9\",  # kserve on python 3.10 comes with a dependency that fails to get installed\n",
    "    packages_to_install=[\"kserve==0.11.0\", \"scikit-learn~=1.0.2\"],\n",
    "    output_component_file='components/inference_component.yaml',\n",
    ")\n",
    "def inference(\n",
    "    model_name: str,\n",
    "    scaler_in: Input[Artifact]\n",
    "):\n",
    "    \"\"\"\n",
    "    Test inference.\n",
    "    \"\"\"\n",
    "    from kserve import KServeClient\n",
    "    import requests\n",
    "    import pickle\n",
    "    import logging\n",
    "    from kserve import utils\n",
    "    from urllib.parse import urlsplit\n",
    "    import re\n",
    "    \n",
    "    logging.basicConfig(level=logging.INFO)\n",
    "    logger = logging.getLogger(__name__)\n",
    "    \n",
    "    def get_istio_auth_session(url: str, username: str, password: str) -> dict:\n",
    "        \"\"\"\n",
    "        Determine if the specified URL is secured by Dex and try to obtain a session cookie.\n",
    "        WARNING: only Dex `staticPasswords` and `LDAP` authentication are currently supported\n",
    "                 (we default default to using `staticPasswords` if both are enabled)\n",
    "    \n",
    "        :param url: Kubeflow server URL, including protocol\n",
    "        :param username: Dex `staticPasswords` or `LDAP` username\n",
    "        :param password: Dex `staticPasswords` or `LDAP` password\n",
    "        :return: auth session information\n",
    "        \"\"\"\n",
    "        # define the default return object\n",
    "        auth_session = {\n",
    "            \"endpoint_url\": url,    # KF endpoint URL\n",
    "            \"redirect_url\": None,   # KF redirect URL, if applicable\n",
    "            \"dex_login_url\": None,  # Dex login URL (for POST of credentials)\n",
    "            \"is_secured\": None,     # True if KF endpoint is secured\n",
    "            \"session_cookie\": None  # Resulting session cookies in the form \"key1=value1; key2=value2\"\n",
    "        }\n",
    "    \n",
    "        # use a persistent session (for cookies)\n",
    "        with requests.Session() as s:\n",
    "    \n",
    "            ################\n",
    "            # Determine if Endpoint is Secured\n",
    "            ################\n",
    "            resp = s.get(url, allow_redirects=True)\n",
    "            if resp.status_code != 200:\n",
    "                raise RuntimeError(\n",
    "                    f\"HTTP status code '{resp.status_code}' for GET against: {url}\"\n",
    "                )\n",
    "    \n",
    "            auth_session[\"redirect_url\"] = resp.url\n",
    "    \n",
    "            # if we were NOT redirected, then the endpoint is UNSECURED\n",
    "            if len(resp.history) == 0:\n",
    "                auth_session[\"is_secured\"] = False\n",
    "                return auth_session\n",
    "            else:\n",
    "                auth_session[\"is_secured\"] = True\n",
    "    \n",
    "            ################\n",
    "            # Get Dex Login URL\n",
    "            ################\n",
    "            redirect_url_obj = urlsplit(auth_session[\"redirect_url\"])\n",
    "    \n",
    "            # if we are at `/auth?=xxxx` path, we need to select an auth type\n",
    "            if re.search(r\"/auth$\", redirect_url_obj.path):\n",
    "    \n",
    "                #######\n",
    "                # TIP: choose the default auth type by including ONE of the following\n",
    "                #######\n",
    "    \n",
    "                # OPTION 1: set \"staticPasswords\" as default auth type\n",
    "                redirect_url_obj = redirect_url_obj._replace(\n",
    "                    path=re.sub(r\"/auth$\", \"/auth/local\", redirect_url_obj.path)\n",
    "                )\n",
    "                # OPTION 2: set \"ldap\" as default auth type\n",
    "                # redirect_url_obj = redirect_url_obj._replace(\n",
    "                #     path=re.sub(r\"/auth$\", \"/auth/ldap\", redirect_url_obj.path)\n",
    "                # )\n",
    "    \n",
    "            # if we are at `/auth/xxxx/login` path, then no further action is needed (we can use it for login POST)\n",
    "            if re.search(r\"/auth/.*/login$\", redirect_url_obj.path):\n",
    "                auth_session[\"dex_login_url\"] = redirect_url_obj.geturl()\n",
    "    \n",
    "            # else, we need to be redirected to the actual login page\n",
    "            else:\n",
    "                # this GET should redirect us to the `/auth/xxxx/login` path\n",
    "                resp = s.get(redirect_url_obj.geturl(), allow_redirects=True)\n",
    "                if resp.status_code != 200:\n",
    "                    raise RuntimeError(\n",
    "                        f\"HTTP status code '{resp.status_code}' for GET against: {redirect_url_obj.geturl()}\"\n",
    "                    )\n",
    "    \n",
    "                # set the login url\n",
    "                auth_session[\"dex_login_url\"] = resp.url\n",
    "    \n",
    "            ################\n",
    "            # Attempt Dex Login\n",
    "            ################\n",
    "            resp = s.post(\n",
    "                auth_session[\"dex_login_url\"],\n",
    "                data={\"login\": username, \"password\": password},\n",
    "                allow_redirects=True\n",
    "            )\n",
    "            if len(resp.history) == 0:\n",
    "                raise RuntimeError(\n",
    "                    f\"Login credentials were probably invalid - \"\n",
    "                    f\"No redirect after POST to: {auth_session['dex_login_url']}\"\n",
    "                )\n",
    "    \n",
    "            # store the session cookies in a \"key1=value1; key2=value2\" string\n",
    "            auth_session[\"session_cookie\"] = \"; \".join([f\"{c.name}={c.value}\" for c in s.cookies])\n",
    "    \n",
    "        return auth_session\n",
    "    \n",
    "    KUBEFLOW_ENDPOINT = \"http://istio-ingressgateway.istio-system.svc.cluster.local:80\"\n",
    "    KUBEFLOW_USERNAME = \"user@example.com\"\n",
    "    KUBEFLOW_PASSWORD = \"12341234\"\n",
    "    \n",
    "    auth_session = get_istio_auth_session(\n",
    "    url=KUBEFLOW_ENDPOINT,\n",
    "    username=KUBEFLOW_USERNAME,\n",
    "    password=KUBEFLOW_PASSWORD,\n",
    "    )\n",
    "    TOKEN = auth_session[\"session_cookie\"].replace(\"authservice_session=\", \"\")\n",
    "    print(\"Token:\", TOKEN)\n",
    "\n",
    "    namespace = utils.get_default_target_namespace()\n",
    "\n",
    "    input_sample = [[5.6, 0.54, 0.04, 1.7, 0.049, 5, 13, 0.9942, 3.72, 0.58, 11.4],\n",
    "                    [11.3, 0.34, 0.45, 2, 0.082, 6, 15, 0.9988, 2.94, 0.66, 9.2]]\n",
    "\n",
    "    logger.info(f\"Loading standard scaler from: {scaler_in.path}\")\n",
    "    with open(scaler_in.path, 'rb') as fp:\n",
    "        scaler = pickle.load(fp)\n",
    "\n",
    "    logger.info(f\"Standardizing sample: {scaler_in.path}\")\n",
    "    input_sample = scaler.transform(input_sample)\n",
    "\n",
    "    # get inference service\n",
    "    KServe = KServeClient()\n",
    "\n",
    "    # wait for deployment to be ready\n",
    "    KServe.get(model_name, namespace=namespace, watch=True, timeout_seconds=120)\n",
    "\n",
    "    inference_service = KServe.get(model_name, namespace=namespace)\n",
    "    logger.info(f\"inference_service: {inference_service}\")\n",
    "\n",
    "    is_url = f\"http://istio-ingressgateway.istio-system.svc.cluster.local:80/v1/models/{model_name}:predict\"\n",
    "    header = {\"Host\": f\"{model_name}.{namespace}.example.com\"}\n",
    "    \n",
    "    logger.info(f\"\\nInference service status:\\n{inference_service['status']}\")\n",
    "    logger.info(f\"\\nInference service URL:\\n{is_url}\\n\")\n",
    "\n",
    "    inference_input = {\n",
    "        'instances': input_sample.tolist()\n",
    "    }\n",
    "    response = requests.post(\n",
    "        is_url,\n",
    "        json=inference_input,\n",
    "        headers=header,\n",
    "        cookies={\"authservice_session\": TOKEN}\n",
    "        \n",
    "    )\n",
    "    if response.status_code != 200:\n",
    "        raise RuntimeError(f\"HTTP status code '{response.status_code}': {response.json()}\")\n",
    "    \n",
    "    logger.info(f\"\\nPrediction response:\\n{response.json()}\\n\")"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "## 3. Pipeline\n",
    "\n",
    "Pipeline definition:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "@dsl.pipeline(\n",
    "      name='demo-pipeline',\n",
    "      description='An example pipeline that performs addition calculations.',\n",
    ")\n",
    "def pipeline(\n",
    "    url: str,\n",
    "    target: str,\n",
    "    mlflow_experiment_name: str,\n",
    "    mlflow_tracking_uri: str,\n",
    "    mlflow_s3_endpoint_url: str,\n",
    "    model_name: str,\n",
    "    alpha: float,\n",
    "    l1_ratio: float,\n",
    "    threshold_metrics: dict,\n",
    "):\n",
    "    pull_task = pull_data(url=url)\n",
    "\n",
    "    preprocess_task = preprocess(data=pull_task.outputs[\"data\"])\n",
    "\n",
    "    train_task = train(\n",
    "        train_set=preprocess_task.outputs[\"train_set\"],\n",
    "        test_set=preprocess_task.outputs[\"test_set\"],\n",
    "        target=target,\n",
    "        mlflow_experiment_name=mlflow_experiment_name,\n",
    "        mlflow_tracking_uri=mlflow_tracking_uri,\n",
    "        mlflow_s3_endpoint_url=mlflow_s3_endpoint_url,\n",
    "        model_name=model_name,\n",
    "        alpha=alpha,\n",
    "        l1_ratio=l1_ratio\n",
    "    )\n",
    "    train_task.apply(use_aws_secret(secret_name=\"aws-secret\"))\n",
    "\n",
    "    evaluate_trask = evaluate(\n",
    "        run_id=train_task.outputs[\"run_id\"],\n",
    "        mlflow_tracking_uri=mlflow_tracking_uri,\n",
    "        threshold_metrics=threshold_metrics\n",
    "    )\n",
    "\n",
    "    eval_passed = evaluate_trask.output\n",
    "\n",
    "    with dsl.Condition(eval_passed == \"true\"):\n",
    "        deploy_model_task = deploy_model(\n",
    "            model_name=model_name,\n",
    "            storage_uri=train_task.outputs[\"storage_uri\"],\n",
    "        )\n",
    "\n",
    "        inference_task = inference(\n",
    "            model_name=model_name,\n",
    "            scaler_in=preprocess_task.outputs[\"scaler_out\"]\n",
    "        )\n",
    "        inference_task.after(deploy_model_task)"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Pipeline arguments:"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "# Specify pipeline argument values\n",
    "\n",
    "eval_threshold_metrics = {'rmse': 0.9, 'r2': 0.3, 'mae': 0.8}\n",
    "\n",
    "arguments = {\n",
    "    \"url\": \"http://archive.ics.uci.edu/ml/machine-learning-databases/wine-quality/winequality-red.csv\",\n",
    "    \"target\": \"quality\",\n",
    "    \"mlflow_tracking_uri\": \"http://mlflow.mlflow.svc.cluster.local:5000\",\n",
    "    \"mlflow_s3_endpoint_url\": \"http://mlflow-minio-service.mlflow.svc.cluster.local:9000\",\n",
    "    \"mlflow_experiment_name\": \"demo-notebook\",\n",
    "    \"model_name\": \"wine-quality\",\n",
    "    \"alpha\": 0.5,\n",
    "    \"l1_ratio\": 0.5,\n",
    "    \"threshold_metrics\": eval_threshold_metrics\n",
    "}"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "## 4. Submit run"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "run_name = \"demo-run\"\n",
    "experiment_name = \"demo-experiment\"\n",
    "\n",
    "client.create_run_from_pipeline_func(\n",
    "    pipeline_func=pipeline,\n",
    "    run_name=run_name,\n",
    "    experiment_name=experiment_name,\n",
    "    arguments=arguments,\n",
    "    mode=kfp.dsl.PipelineExecutionMode.V2_COMPATIBLE,\n",
    "    enable_caching=False,\n",
    "    namespace=\"kubeflow-user-example-com\"\n",
    ")"
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "## 5. Check run"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "### Kubeflow Pipelines UI\n",
    "\n",
    "The default way of accessing Kubeflow is via port-forward. This enables you to get started quickly without imposing any requirements on your environment. Run the following to port-forward Istio's Ingress-Gateway to local port `8080`:\n",
    "\n",
    "```sh\n",
    "kubectl port-forward svc/istio-ingressgateway -n istio-system 8080:80\n",
    "```\n",
    "\n",
    "After running the command, you can access the Kubeflow Central Dashboard by doing the following:\n",
    "\n",
    "1. Open your browser and visit [http://localhost:8080/](http://localhost:8080/). You should get the Dex login screen.\n",
    "2. Login with the default user's credential. The default email address is `user@example.com` and the default password is `12341234`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "### MLFlow UI\n",
    "\n",
    "To access MLFlow UI, open a terminal and forward a local port to MLFlow server:\n",
    "\n",
    "<br>\n",
    "\n",
    "```bash\n",
    "kubectl -n mlflow port-forward svc/mlflow 5000:5000\n",
    "```\n",
    "\n",
    "<br>\n",
    "\n",
    "Now MLFlow's UI should be reachable at [`http://localhost:5000`](http://localhost:5000)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "## 6. Check deployed model\n",
    "\n",
    "```bash\n",
    "# get inference services\n",
    "kubectl -n kubeflow-user-example-com get inferenceservice\n",
    "\n",
    "# get deployed model pods\n",
    "kubectl -n kubeflow-user-example-com get pods\n",
    "\n",
    "# delete inference service\n",
    "kubectl -n kubeflow-user-example-com delete inferenceservice wine-quality\n",
    "```\n",
    "<br>\n",
    "\n",
    "If something goes wrong, check the logs with:\n",
    "\n",
    "<br>\n",
    "\n",
    "```bash\n",
    "kubectl logs -n kubeflow-user-example-com <pod-name> kserve-container\n",
    "\n",
    "kubectl logs -n kubeflow-user-example-com <pod-name> queue-proxy\n",
    "\n",
    "kubectl logs -n kubeflow-user-example-com <pod-name> storage-initializer\n",
    "```\n"
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 7. Troubleshooting\n",
    "\n",
    "If the inference isn't working, try to patch the knative-serving config-domain:\n",
    "\n",
    "\n",
    "\n",
    "```bash\n",
    "kubectl patch cm config-domain --patch '{\"data\":{\"example.com\":\"\"}}' -n knative-serving\n",
    "```"
   ],
   "metadata": {
    "collapsed": false
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "iml4e",
   "language": "python",
   "name": "python3"
  },
  "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.8.15 (default, Nov 24 2022, 08:57:44) \n[Clang 14.0.6 ]"
  },
  "vscode": {
   "interpreter": {
    "hash": "2976e1db094957a35b33d12f80288a268286b510a60c0d029aa085f0b10be691"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}