2019-10 Kubeflow summit/106 - Creating components from command-line programs/106 - Creating components from command-line programs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating components from command-line programs\n",
    "\n",
    "This tutorial shows how to quicly author a set of reusable components based on existing command-line programs.\n",
    "\n",
    "The components are used to compose a pipeline.\n",
    "\n",
    "In this example the goal of the pipeline is to dynamically build a container image from a directory in GIT repository.\n",
    "\n",
    "The main program is the Kaniko executor that can take context archive file from a Google Cloud Storage location and build+push a container image. But to run that program we first need to get the files from GIT, select the needed files, archive them and upload to Google Cloud Storage.\n",
    "\n",
    "Pipeline steps:\n",
    "\n",
    "* Get files from GIT. (`git clone`)\n",
    "* Choose needed files. (`cp`)\n",
    "* Archive files. (`tar`)\n",
    "* Upload archive. (`gsutil cp`)\n",
    "* Build container image. (Kaniko)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Install Kubeflow Pipelines SDK\n",
    "!PIP_DISABLE_PIP_VERSION_CHECK=1 pip3 install 'kfp>=0.1.32.2' --quiet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# GCP project ID which will be used to push the new image\n",
    "GCP_PROJECT_ID='...'\n",
    "target_image_name = 'gcr.io/' + GCP_PROJECT_ID + '/tmp-kfp-106-build-image:tag1'\n",
    "\n",
    "GCP_STORAGE_DIR = 'gs://<bucket-name>/tmp_kfp_106_build_image/'\n",
    "\n",
    "import kfp\n",
    "image_context_scratch_uri = GCP_STORAGE_DIR + kfp.dsl.RUN_ID_PLACEHOLDER + '.tar.gz'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initializing the client\n",
    "client = kfp.Client()\n",
    "\n",
    "# ! Use kfp.Client(host='https://xxxxx.notebooks.googleusercontent.com/') if working from GCP notebooks (or local notebooks)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "\n",
    "from kfp.components import InputPath, OutputPath, load_component_from_file, load_component_from_url"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Path('git_clone.component.yaml').write_text('''\\\n",
    "name: Git clone\n",
    "inputs:\n",
    "- {name: Repo URI, type: URI}\n",
    "- {name: Branch, type: String, default: 'master'}\n",
    "outputs:\n",
    "- {name: Repo dir, type: Directory}\n",
    "implementation:\n",
    "  container:\n",
    "    image: alpine/git\n",
    "    command:\n",
    "    - git\n",
    "    - clone\n",
    "    - --depth=1\n",
    "    - --branch\n",
    "    - inputValue: Branch\n",
    "    - inputValue: Repo URI\n",
    "    - outputPath: Repo dir\n",
    "'''\n",
    ")\n",
    "\n",
    "git_clone_op = load_component_from_file('git_clone.component.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Path('get_subdir.component.yaml').write_text('''\\\n",
    "name: Get subdirectory\n",
    "description: Get subdirectory items (files or directories).\n",
    "inputs:\n",
    "- {name: Directory, type: Directory}\n",
    "- {name: Subpath, type: String}\n",
    "outputs:\n",
    "- {name: Subdir, type: Directory}\n",
    "implementation:\n",
    "  container:\n",
    "    image: alpine\n",
    "    command:\n",
    "    - sh\n",
    "    - -ex\n",
    "    - -c\n",
    "    - |\n",
    "      mkdir -p \"$(dirname \"$2\")\"\n",
    "      cp -r \"$0/$1\" \"$2\"\n",
    "    - inputPath: Directory\n",
    "    - inputValue: Subpath\n",
    "    - outputPath: Subdir\n",
    "'''\n",
    ")\n",
    "\n",
    "get_subdir_op = load_component_from_file('get_subdir.component.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Path('tar_gzip.component.yaml').write_text('''\\\n",
    "name: Compress directory using TAR GZIP\n",
    "inputs:\n",
    "- {name: Data, type: Directory}\n",
    "outputs:\n",
    "- {name: Gzipped data, type: GzippedTar}\n",
    "implementation:\n",
    "  container:\n",
    "    image: alpine\n",
    "    command:\n",
    "    - sh\n",
    "    - -ex\n",
    "    - -c\n",
    "    - |\n",
    "      mkdir -p \"$(dirname \"$1\")\"\n",
    "      cd \"$0\"\n",
    "      tar c -z -f \"$1\" .\n",
    "    - inputPath: Data\n",
    "    - outputPath: Gzipped data\n",
    "'''\n",
    ")\n",
    "\n",
    "tar_gzip_op = load_component_from_file('tar_gzip.component.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Path('google_cloud_storage_upload.component.yaml').write_text('''\\\n",
    "name: Upload to GCS\n",
    "inputs:\n",
    "- {name: Data}\n",
    "- {name: GCS path, type: URI}\n",
    "outputs:\n",
    "- {name: GCS path, type: URI}\n",
    "implementation:\n",
    "  container:\n",
    "    image: google/cloud-sdk\n",
    "    command:\n",
    "    - sh\n",
    "    - -ex\n",
    "    - -c\n",
    "    - |\n",
    "      gcloud auth activate-service-account --key-file=\"${GOOGLE_APPLICATION_CREDENTIALS}\"\n",
    "      gsutil cp \"$0\" \"$1\"\n",
    "      mkdir -p \"$(dirname \"$2\")\"\n",
    "      echo \"$1\" > \"$2\"\n",
    "    - inputPath: Data\n",
    "    - inputValue: GCS path\n",
    "    - outputPath: GCS path\n",
    "'''\n",
    ")\n",
    "\n",
    "upload_to_gcs_op = load_component_from_file('google_cloud_storage_upload.component.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Path('containers_build_image_from_context_uri.component.yaml').write_text('''\\\n",
    "name: Build container image\n",
    "inputs:\n",
    "- {name: Context archive URI, type: URI}\n",
    "- {name: Target image name, type: String}\n",
    "outputs:\n",
    "- {name: Image digest, type: String}\n",
    "implementation:\n",
    "  container:\n",
    "    image: gcr.io/kaniko-project/executor@sha256:78d44ec4e9cb5545d7f85c1924695c89503ded86a59f92c7ae658afa3cff5400\n",
    "    command:\n",
    "    - /kaniko/executor\n",
    "    - --cache=true\n",
    "    - --dockerfile\n",
    "    - Dockerfile\n",
    "    - --context\n",
    "    - inputValue: Context archive URI\n",
    "    - --destination\n",
    "    - inputValue: Target image name\n",
    "    - --digest-file\n",
    "    - /tmp/digest.txt\n",
    "    fileOutputs:\n",
    "      Image digest: /tmp/digest.txt\n",
    "'''\n",
    ")\n",
    "\n",
    "build_image_from_context_uri_op = load_component_from_file('containers_build_image_from_context_uri.component.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print_component_text = '''\n",
    "name: Print text\n",
    "inputs:\n",
    "- {name: Text}\n",
    "implementation:\n",
    "  container:\n",
    "    image: alpine\n",
    "    command:\n",
    "    - cat\n",
    "    - inputPath: Text\n",
    "'''\n",
    "print_op = kfp.components.load_component_from_text(print_component_text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_image_with_staging_pipeline(\n",
    "    target_image_name,\n",
    "    staging_gcs_path='gs://avolkov/tmp_git_pipeline/tmp123.tgz',\n",
    "    repo_uri='https://github.com/kubeflow/pipelines.git',\n",
    "    repo_subpath='components/sample/keras/train_classifier',\n",
    "):\n",
    "    git_clone_task = git_clone_op(\n",
    "        repo_uri=repo_uri,\n",
    "    )\n",
    "\n",
    "    get_subdir_task = get_subdir_op(\n",
    "        directory=git_clone_task.output,\n",
    "        subpath=repo_subpath,\n",
    "    )\n",
    "\n",
    "    tar_gzip_task = tar_gzip_op(\n",
    "        data=get_subdir_task.output,\n",
    "    )\n",
    "\n",
    "    upload_to_gcs_task = upload_to_gcs_op(\n",
    "        data=tar_gzip_task.output,\n",
    "        gcs_path=staging_gcs_path,\n",
    "    )\n",
    "\n",
    "    build_image_from_context_uri_task = build_image_from_context_uri_op(\n",
    "        context_archive_uri=upload_to_gcs_task.output,\n",
    "        target_image_name=target_image_name,\n",
    "    )\n",
    "\n",
    "    print_op(\n",
    "        build_image_from_context_uri_task.outputs['image_digest']\n",
    "    )\n",
    "    \n",
    "\n",
    "# Adding GCP credential secrets\n",
    "# Needed to get permissions in Kubeflow < 0.7\n",
    "from kfp import gcp\n",
    "pipeline_conf=kfp.dsl.PipelineConf()\n",
    "pipeline_conf.add_op_transformer(gcp.use_gcp_secret('user-gcp-sa'))\n",
    "\n",
    "client.create_run_from_pipeline_func(\n",
    "    build_image_with_staging_pipeline,\n",
    "    arguments={\n",
    "        'target_image_name': target_image_name,\n",
    "        'staging_gcs_path': image_context_scratch_uri,\n",
    "    },\n",
    "    pipeline_conf=pipeline_conf,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Modifying the \"build image\" component to remove the usage of staging location in Google Cloud Storage.\n",
    "After the pipeline was completed, it was discovered that the Kaniko builder can build images from local context directory\n",
    "\n",
    "This can greatly simplify our pipeline and remove the need for the staging Google Cloud Storage location.\n",
    "\n",
    "Let's rewrite the component to accept in-system data instead of a Google Cloud Storage URI."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Path('containers_build_image_from_context.component.yaml').write_text('''\\\n",
    "name: Build container image\n",
    "inputs:\n",
    "- {name: Context}\n",
    "- {name: Target image name, type: String}\n",
    "outputs:\n",
    "- {name: Image digest, type: String}\n",
    "implementation:\n",
    "  container:\n",
    "    image: gcr.io/kaniko-project/executor@sha256:78d44ec4e9cb5545d7f85c1924695c89503ded86a59f92c7ae658afa3cff5400\n",
    "    command:\n",
    "    - /kaniko/executor\n",
    "    - --cache=true\n",
    "    - --dockerfile\n",
    "    - Dockerfile\n",
    "    - --context\n",
    "    - inputPath: Context\n",
    "    - --destination\n",
    "    - inputValue: Target image name\n",
    "    - --digest-file\n",
    "    - /tmp/digest.txt\n",
    "    fileOutputs:\n",
    "      Image digest: /tmp/digest.txt\n",
    "'''\n",
    ")\n",
    "\n",
    "build_image_op = load_component_from_file('containers_build_image_from_context.component.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_image_pipeline(\n",
    "    target_image_name,\n",
    "    repo_uri='https://github.com/kubeflow/pipelines.git',\n",
    "    repo_subpath='components/sample/keras/train_classifier',\n",
    "):\n",
    "    git_clone_task = git_clone_op(\n",
    "        repo_uri=repo_uri,\n",
    "    )\n",
    "\n",
    "    get_subdir_task = get_subdir_op(\n",
    "        directory=git_clone_task.output,\n",
    "        subpath=repo_subpath,\n",
    "    )\n",
    "\n",
    "    build_image_task = build_image_op(\n",
    "        context=get_subdir_task.output,\n",
    "        target_image_name=target_image_name,\n",
    "    )\n",
    "\n",
    "    print_op(\n",
    "        build_image_task.outputs['image_digest']\n",
    "    )\n",
    "\n",
    "\n",
    "# Adding GCP credential secrets\n",
    "# Needed to get permissions in Kubeflow < 0.7\n",
    "from kfp import gcp\n",
    "pipeline_conf=kfp.dsl.PipelineConf()\n",
    "pipeline_conf.add_op_transformer(gcp.use_gcp_secret('user-gcp-sa'))\n",
    "\n",
    "client.create_run_from_pipeline_func(\n",
    "    build_image_pipeline,\n",
    "    arguments={'target_image_name': target_image_name},\n",
    "    pipeline_conf=pipeline_conf,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Exercise 1**: The build image component returns only the hash digest of the container image. Create a new component that can merge the target image name with the hash digest to get a full image name with digest."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}