Run_Flow.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "fbffcd67",
   "metadata": {},
   "source": [
    "Now get an instance of the flows client.  You will be asked to validate credentials with the Globus service.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "9d902a0b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "\n",
    "import globus_sdk\n",
    "\n",
    "from utils import get_specific_flow_client\n",
    "# fc = get_flows_client()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "680519c8",
   "metadata": {},
   "source": [
    "Next, create a flow.  You will again be asked to validate credentials with the globus service."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b146f619",
   "metadata": {},
   "source": [
    "## Run the Flow\n",
    "\n",
    "Now we need to create a set of inputs for the flow, that will follow the json structure below.  The key elements that are needed are:\n",
    "1. The Globus compute endpoint id\n",
    "2. The Globus compute function id\n",
    "3. Inputs to the Globus compute function\n",
    "4. The endpoint and path from which to transfer data\n",
    "5. The endpoint and path to which to transfer data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ca578c2",
   "metadata": {},
   "source": [
    "The user should choose a destination path and endpoint that they have access to.  As a sample endpoint, below the ALCF /home space is used.  To use this endpoint, the user should set a path in their home space (but remove the leading `/home` so that it will appear like `/csimpson/als_example`)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "6056aeb2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# lcfhome_transfer_endpoint_id = \"9032dd3a-e841-4687-a163-2720da731b5b\"\n",
    "nersc_transfer_endpoint_id = \"6bdc7956-fc0f-4ad2-989c-7aa5ee643a79\"\n",
    "destination_path_on_alcfhome = \"/global/cfs/cdirs/als/data_mover/iribeta/\" # Note that paths for transfers on the home endpoint should remove the leading /home\n",
    "reconstruction_func = \"91ce59ba-cbf1-45e3-b4f3-c86aeaacda42\"\n",
    "polaris_endpoint_id = \"b33b3f5f-eb14-4bbd-89ed-9333a6ece6b6\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5ea4510b",
   "metadata": {},
   "source": [
    "The example problem is setup in a directory on the Eagle filesystem at ALCF, so that endpoint and path is used as the source of the data.  All the endpoints and functions are added to a `flow_input`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "aee0f949",
   "metadata": {},
   "outputs": [],
   "source": [
    "eagle_transfer_endpoint_id = \"05d2c76a-e867-4f67-aa57-76edeb0beda0\"\n",
    "source_path_on_eagle = \"/IRIBeta/als/example\" # Note that paths for transfers on the eagle endpoint should remove the leading /eagle\n",
    "\n",
    "function_inputs = {\"rundir\": \"/eagle/IRIBeta/als/example\"}\n",
    "\n",
    "flow_input = {\n",
    "    \"input\": {\n",
    "      \"source\": {\n",
    "        \"id\": eagle_transfer_endpoint_id,\n",
    "        \"path\": source_path_on_eagle\n",
    "      },\n",
    "      \"destination\": {\n",
    "        \"id\": nersc_transfer_endpoint_id,\n",
    "        \"path\": destination_path_on_alcfhome\n",
    "      },\n",
    "      \"recursive_tx\": True,\n",
    "      \"compute_endpoint_id\": polaris_endpoint_id,\n",
    "      \"compute_function_id\": reconstruction_func,\n",
    "      \"compute_function_kwargs\": function_inputs\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "8dc12cc8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# collection_ids should contain all the transfer endpoint ids involved in the flow\n",
    "collection_ids = [flow_input[\"input\"][\"source\"][\"id\"], flow_input[\"input\"][\"destination\"][\"id\"]]\n",
    "\n",
    "run_client = get_specific_flow_client(reconstruction_func, collection_ids=collection_ids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "2635cf91",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Flow action started with id: a618a5a2-e5ff-4339-9ec7-260abc98d289\n",
      "Monitor your flow here: https://app.globus.org/runs/a618a5a2-e5ff-4339-9ec7-260abc98d289\n",
      "Flow status: FAILED\n",
      "Final status: FAILED\n"
     ]
    }
   ],
   "source": [
    "flow_action = run_client.run_flow(flow_input, label=\"ALS run\", tags=[\"demo\", \"als\", \"tomopy\"])\n",
    "flow_run_id = flow_action['action_id']\n",
    "\n",
    "print(f'Flow action started with id: {flow_run_id}')\n",
    "\n",
    "print(f\"Monitor your flow here: https://app.globus.org/runs/{flow_run_id}\")\n",
    "import time\n",
    "flow_status = flow_action['status']\n",
    "while flow_status in ['ACTIVE', 'INACTIVE']:\n",
    "    time.sleep(10)\n",
    "    flow_action = fc.get_run(flow_run_id)\n",
    "    flow_status = flow_action['status']\n",
    "    print(f'Flow status: {flow_status}')\n",
    "print(f'Final status: {flow_status}')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}