Gladier_Flows_Tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Gladier Flows Tutorial\n",
    "### Gladier: The Globus Architecture for Data-Intensive Experimental Research.\n",
    "\n",
    "Gladier is a programmable data capture, storage, and analysis architecture for experimental facilities. The architecture leverages a data and computing substrate based on agents deployed across computer and storage systems at APS, ALCF, and elsewhere, all managed by cloud-hosted Globus services. In particular, we leverage [Globus Connect](https://www.globus.org/globus-connect)\n",
    "and [Globus Compute](https://globus-compute.readthedocs.io/en/latest/) agents to facilitate secure, reliable remote data and computation and employ the [Globus Flows](https://www.globus.org/platform/services/flows) platform to orchestrate distributed data management tasks into reliable pipelines.\n",
    "\n",
    "## Gladier Toolkit\n",
    "The Gladier toolkit provides tools and capabilities to simplify and accelerate the development of these automations. The toolkit manages the dynamic creation of flows, automatically registers Globus Compute functions, and assists in validating inputs. \n",
    "\n",
    "Here we demonstrate how the Gladier toolkit can be used to let anyone create a simple, yet powerful client to automate data management tasks.\n",
    "\n",
    "While not necessary to use this notebook, the Gladier toolkit is available on pypi and can be installed with:\n",
    "\n",
    "    $ pip install gladier\n",
    "\n",
    "Documentation is available [here.](https://gladier.readthedocs.io/en/latest/index.html)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# General Imports\n",
    "import pprint\n",
    "import json\n",
    "import os\n",
    "import pathlib\n",
    "import random\n",
    "\n",
    "# Gladier Imports\n",
    "from gladier import GladierBaseClient, GladierBaseTool, generate_flow_definition\n",
    "\n",
    "# Set this so Gladier knows it should login on a remote system\n",
    "os.environ['SSH_TTY'] = 'JUPYTERHUB_REMOTE'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Globus Search\n",
    "\n",
    "For our little experiment here, we will be publishing data to Globus Search so we can later display it in a portal. We need to setup a Globus Search index in order to do the publishing step below.\n",
    "\n",
    "See the \"Metadata Search and Discovery\" for more information on Globus Search"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pickle, base64, os, pprint, globus_sdk\n",
    "\n",
    "# Name of the search index to use for this notebook\n",
    "index_name = 'gladier-tutorial'\n",
    "\n",
    "# Load a search client using a token from the Jupyterhub login\n",
    "data = pickle.loads(base64.b64decode(os.getenv('GLOBUS_DATA')))\n",
    "search_token = data['tokens']['search.api.globus.org']['access_token']\n",
    "search = globus_sdk.SearchClient(authorizer=globus_sdk.AccessTokenAuthorizer(search_token))\n",
    "\n",
    "# Fetch all indices to which the user has access\n",
    "indices = [si for si in search.get(\"/v1/index_list\").data['index_list']\n",
    "           if si['is_trial'] \n",
    "           and si['display_name'] == index_name\n",
    "           and 'owner' in si['permissions']\n",
    "          ]\n",
    "\n",
    "# If an index was found with the criteria above, re-use it\n",
    "# Otherwise, create a new index\n",
    "if indices:\n",
    "    tutorial_index = indices[0]\n",
    "    print('Found existing trial index!')\n",
    "else:\n",
    "    tutorial_index = search.create_index(\n",
    "        display_name=index_name,\n",
    "        description=\"A trial index for use in Globus tutorials\"\n",
    "    ).data\n",
    "    print('Index created for use in Globus tutorials.')\n",
    "\n",
    "search_index = tutorial_index['id']\n",
    "\n",
    "print(tutorial_index['display_name'])\n",
    "print(tutorial_index['description'])\n",
    "print(search_index)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Gladier Tools\n",
    "\n",
    "Gladier Tools are the glue that holds together Globus Flows and Compute Functions. Tools bundle everything the Compute Function needs to run, so the Glaider Client can register the function, check the requirements, and run it inside the flow.\n",
    "\n",
    "We need three Flow States below to run our full experiment:\n",
    "\n",
    "1. RunExperiment -- A function which will do the experimental work.\n",
    "2. GatherMetadata -- A function to gather results of the experiment.\n",
    "3. PublishMetadata -- A flow state to ingest the metadata into Globus Search.\n",
    "\n",
    "The first two Gladier Tools are Compute Functions, and use the `@generate_flow_definition` decorator to create the flow state for each function. The final publication state does not use the decorator, and instead uses a static flow definition. Gladier will chain each of these together into one single flow, run one after the other.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def run_experiment(**data):\n",
    "    \"\"\"Run an 'experiment' on our remote execution environment.\"\"\"\n",
    "    import pathlib\n",
    "    import random\n",
    "    experiment = pathlib.Path(data['experiment'])\n",
    "    # Say hello a bunch of times\n",
    "    experiment.write_text('Hello Gladier!' * random.randint(1, 100))\n",
    "    return str(experiment)\n",
    "\n",
    "\n",
    "@generate_flow_definition\n",
    "class RunExperiment(GladierBaseTool):\n",
    "    compute_functions = [run_experiment]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def gather_metadata(**data):\n",
    "    import pathlib\n",
    "    import hashlib\n",
    "    from datetime import datetime\n",
    "    experiment = pathlib.Path(data['experiment'])  \n",
    "    search_document = {\n",
    "        'search_index': data['search_index'],\n",
    "        'content': {    \n",
    "            'dc': {\n",
    "                'creators': [{'creatorName': 'ALCF Researcher'}],\n",
    "                'dates': [{'date': datetime.now().isoformat(),\n",
    "                           'dateType': 'Created'}],\n",
    "                'formats': ['text/plain'],\n",
    "                'publicationYear': '2021',\n",
    "                'publisher': 'ALCF Researcher',\n",
    "                'resourceType': {'resourceType': 'Dataset',\n",
    "                                 'resourceTypeGeneral': 'Dataset'},\n",
    "                'subjects': [{'subject': 'Globus'}, {'subject': 'Flows'}, {'subject': 'ALCF'}],\n",
    "                'titles': [{'title': f'My Experiment {experiment.name}'}],\n",
    "                'version': 1,\n",
    "                \n",
    "            },\n",
    "            'files': [{\n",
    "                'filename': experiment.name,\n",
    "                'length': experiment.stat().st_size,\n",
    "                'mime_type': 'text/plain',\n",
    "                'md5': hashlib.md5(experiment.read_bytes()).hexdigest(),\n",
    "                'sha256': hashlib.sha256(experiment.read_bytes()).hexdigest(),\n",
    "            }],\n",
    "        },\n",
    "        'subject': data['subject'],\n",
    "        'visible_to': ['public'],\n",
    "    }\n",
    "    # Clean up our 'experiment' file.\n",
    "    experiment.unlink()\n",
    "    return search_document\n",
    "\n",
    "\n",
    "@generate_flow_definition\n",
    "class GatherMetadata(GladierBaseTool):\n",
    "    compute_functions = [gather_metadata]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class PublishMetadata(GladierBaseTool):\n",
    "    flow_definition = {\n",
    "        'Comment': 'Publish metadata to Globus Search, with data from the result.',\n",
    "        'StartAt': 'PublishMetadata',\n",
    "        'States': {\n",
    "            'PublishMetadata': {\n",
    "                'Comment': 'Ingest a Globus Search document',\n",
    "                'Type': 'Action',\n",
    "                'ActionUrl': 'https://actions.globus.org/search/ingest',\n",
    "                'ExceptionOnActionFailure': True,\n",
    "                'InputPath': '$.GatherMetadata.details.result[0]',\n",
    "                'ResultPath': '$.PublishMetadata',\n",
    "                'WaitTime': 300,\n",
    "                'End': True\n",
    "            }\n",
    "        },\n",
    "    }\n",
    "    compute_functions = []\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Gladier Clients\n",
    "\n",
    "Gladier Clients manage a collection of Glaider Tools and a Globus Flow to link them together into a pipeline. Clients handle both registering Compute Functions for each tool and registering the flow to orchestrate each tool's execution. The checksum of the flows and Compute Functions are checked prior to each invocation to ensure they are always up-to-date. Further, the client checks the necessary inputs to each tool are present before the flow is invoked.\n",
    "\n",
    "Once a tool has been created it can be imported and used by a client. The client can then dynamically create a flow using the list of tools.\n",
    "\n",
    "Here we define an `ExampleClient` and specify the `FileSize` tool. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@generate_flow_definition\n",
    "class MyGladierClient(GladierBaseClient):\n",
    "    gladier_tools = [\n",
    "        RunExperiment,\n",
    "        GatherMetadata,\n",
    "        PublishMetadata,\n",
    "    ]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `@generate_flow_definition` annotation prompts the client to dynamically create a Flow to serially combine each tool used by the client. The resulting flow definition is then saved and can be inspected.\n",
    "\n",
    "More information on flow generation can be found [here.](https://gladier.readthedocs.io/en/latest/gladier/flow_generation.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Flow Input\n",
    "\n",
    "As you can see from the flow definition the input arguments for the tool have been dynamically defined. In this case, the `FileSize` tool requires a `compute_endpoint`, `file_size_function_id` and the entire `input` document is passed as the function payload. These values can be overridden in the flow or defined in the Tool definition.\n",
    "\n",
    "It is important to note that the compute function id, `file_size_function_id` is automatically populated by the Client at runtime. This allows the client to check whether the function definition has changed and re-register the function with Globus Compute if necessary. As such, you do not need to specify the function id as input to the flow.\n",
    "\n",
    "Here we define the input to include a pathname for the tool to act on and a public Globus Compute Endpoint to perform the execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "experiment = pathlib.Path(f'/tmp/file_{random.randint(1, 1000000)}.txt')\n",
    "subject = pathlib.Path('https://example.com/my-gladier-experiment') / experiment.name\n",
    "\n",
    "flow_input = {\n",
    "    'input': {\n",
    "        'experiment': str(experiment),\n",
    "        'subject': str(subject),\n",
    "        'compute_endpoint': '4b116d3c-1703-4f8f-9f6f-39921e5864df',\n",
    "        'search_index': search_index,\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check Existing Flows\n",
    "\n",
    "For new users, the flows service only allows one flow deployed per-person. If you have deployed a flow before, you can delete it using the commented line below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_gladier_client = MyGladierClient()\n",
    "for flow in my_gladier_client.flows_manager.flows_client.list_flows()['flows']:\n",
    "    print(f\"{flow['title']}: {flow['id']}\")\n",
    "    \n",
    "# You can delete flows you don't want with the following line\n",
    "# my_gladier_client.flows_client.delete_flow('')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Running the flow\n",
    "\n",
    "Now input has been created we can use the client to start and monitor the flow.\n",
    "\n",
    "This will prompt you to authenticate and grant permission to the flow to run a Globus Compute function on your behalf. You may need to login twice, once for the static scopes required by Gladier. Second, for the new flow you have just deployed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_gladier_client = MyGladierClient()\n",
    "flow = my_gladier_client.run_flow(flow_input=flow_input, label=f'Experiment {experiment.name}')\n",
    "print(f\"https://app.globus.org/flows/{flow['flow_id']}/runs/{flow['run_id']}\")\n",
    "my_gladier_client.progress(flow['run_id'])\n",
    "my_gladier_client.get_status(flow['run_id'])"
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "29c7d61c0b705a1e0ee1ff93d2482337e8314a272648cd14c87bbb9c779684f0"
  },
  "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}