scripts/notebooks/pybind11_examples/example_ControlPlane.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Python examples of Control Plane bindings\n",
    "\n",
    "*This notebook is developed and test on the FABRIC sender node.*\n",
    "\n",
    "First import the library \"e2sar_py\". Make sure the module is included in the path."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "\n",
    "## IMPORTANT: Update the path to your built Python module\n",
    "sys.path.append(\n",
    "    '/home/ubuntu/dev-e2sar/build/src/pybind')\n",
    "\n",
    "import e2sar_py\n",
    "# print(e2sar_py.get_version())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# print(dir(e2sar_py.ControlPlane))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Helper classes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### \"Timestamp\" class\n",
    "\n",
    "The Python `e2sar_py.ControlPlane.Timestamp` class serves as a bridge to the C++ `google::protobuf::Timestamp` class. The following code illustrates how to use it. Note that the object is printed as a string."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Defaulting to user installation because normal site-packages is not writeable\n",
      "Requirement already satisfied: protobuf in /home/ubuntu/.local/lib/python3.10/site-packages (5.28.1)\n",
      "Note: you may need to restart the kernel to use updated packages.\n",
      "Timestamp: 2024-10-02T04:14:25.698545Z, seconds = 1727842465, nanos = 698545000\n"
     ]
    }
   ],
   "source": [
    "%pip install protobuf\n",
    "\n",
    "'''Cannot direct map Python protobuf TimeStamp to C++ protobuf TimeStamp So we created our own.'''\n",
    "\n",
    "from google.protobuf.timestamp_pb2 import Timestamp as gts\n",
    "from e2sar_py.ControlPlane import Timestamp\n",
    "\n",
    "\n",
    "def get_currtimestamp_from_gts() -> e2sar_py.ControlPlane.Timestamp:\n",
    "    g_ts = gts()\n",
    "    g_ts.GetCurrentTime()\n",
    "    curr_ts = Timestamp()\n",
    "    curr_ts.set_seconds(g_ts.seconds)\n",
    "    curr_ts.set_nanos(g_ts.nanos)\n",
    "    return curr_ts\n",
    "\n",
    "ts = get_currtimestamp_from_gts()\n",
    "print(f\"Timestamp: {ts}, seconds = {ts.get_seconds()}, nanos = {ts.get_nanos()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### \"LBWorkerStatus\" class\n",
    "\n",
    "The following code block demonstrates how to manipulate the `ControlPlane.LBWorkerStatus` class. The `last_updated` attribute is a string."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Worker name: Worker1\n",
      "Worker fill percent: 75.5\n",
      "Worker control signal: 0.8999999761581421\n",
      "Worker slots assigned: 10\n",
      "Worker last updated: 2024-10-02T04:14:25.698545Z\n"
     ]
    }
   ],
   "source": [
    "worker = e2sar_py.ControlPlane.LBWorkerStatus(\"Worker1\", 75.5, 0.9, 10, ts)\n",
    "\n",
    "print(f\"Worker name: {worker.name}\")\n",
    "print(f\"Worker fill percent: {worker.fill_percent}\")\n",
    "print(f\"Worker control signal: {worker.control_signal}\")\n",
    "print(f\"Worker slots assigned: {worker.slots_assigned}\")\n",
    "print(f\"Worker last updated: {worker.last_updated}\")\n",
    "\n",
    "assert(worker.last_updated == str(ts))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### \"LBStatus\" class\n",
    "The attributes `timestamp` and `expiresAt` are bound as Python strings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Timestamp:  2024-10-02T04:14:25.698545Z\n",
      "Expires at:  2024-10-02T05:14:25.698545Z\n"
     ]
    }
   ],
   "source": [
    "ip_list = [\"192.168.100.1\", \"192.168.100.2\"]\n",
    "\n",
    "# Set the expire timestamp\n",
    "expire_ts = Timestamp()  # DO NOT decalre as \"expire_ts = ts\"\n",
    "expire_ts.set_seconds(ts.get_seconds() + 3600)  # 1 hr\n",
    "expire_ts.set_nanos(ts.get_nanos()) \n",
    "\n",
    "# Create an LBStatus object with empty WorkerStatus list\n",
    "status = e2sar_py.ControlPlane.LBStatus(ts, 1234, 5678, [], ip_list, expire_ts)\n",
    "\n",
    "# Access members\n",
    "assert(status.timestamp == str(ts))\n",
    "assert(status.currentEpoch == 1234)\n",
    "assert(status.currentPredictedEventNumber == 5678)\n",
    "assert(status.senderAddressList == ip_list)\n",
    "assert(status.workerStatusList == [])\n",
    "assert(status.expiresAt == str(expire_ts))\n",
    "\n",
    "print(\"Timestamp: \", status.timestamp)\n",
    "print(\"Expires at: \", status.expiresAt)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### \"OverviewEntry\" class\n",
    "\n",
    "The attributes of this class are read-only. Its usage will be demonstrated in `LBMLiveTest6`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## \"LBManager\" class"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# print(dir(e2sar_py.ControlPlane.LBManager))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Examples with a mock EJFAT Load Balancer\n",
    "\n",
    "This section requires a functional (mock) load balancer. If you are launching with a FABRIC slice, ensure the `udplbd` container is running on the FABRIC `cpnode` before executing the code blocks.\n",
    "\n",
    "The Python code blocks below replicate the tests performed in the C++ `e2sar_lbcp_live_test`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "EJFAT_URI=ejfats://udplbd@192.168.0.3:18347/lb/1?data=127.0.0.1&sync=192.168.88.199:1234\n"
     ]
    }
   ],
   "source": [
    "# Verify env variable EJFAT_URI is set\n",
    "import os\n",
    "\n",
    "# This one works on the FABRIC sender node\n",
    "FABRIC_CP_URI=\"ejfats://udplbd@192.168.0.3:18347/lb/1?data=127.0.0.1&sync=192.168.88.199:1234\"\n",
    "\n",
    "os.environ['EJFAT_URI'] = FABRIC_CP_URI\n",
    "\n",
    "print(f\"EJFAT_URI={os.environ['EJFAT_URI']}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create an EjfatURI object\n",
    "uri_rt_obj = e2sar_py.EjfatURI.get_from_env()\n",
    "assert(uri_rt_obj.has_error() == False)  # make sure the EjfatURI object is created successfully"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest1\n",
    "\n",
    "Create a `LBManager` object, reserve the LB for 1 hour, and then free the LB."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ejfats://udplbd@192.168.0.3:18347/lb/1?sync=192.168.88.199:1234&data=127.0.0.1\n"
     ]
    }
   ],
   "source": [
    "# Create a LBManager object\n",
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "print(lbm.get_uri())  # Make sure the uri agrees with the env var"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Reserve a LB with duration in seconds**\n",
    "\n",
    "The Python `datetime.timedelta` package is very convenient to get durations. For simplicity, we create a C++ method `LBManager::reserveLB` based on duration in seconds."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Create LBManager obj succeeded:  0\n"
     ]
    }
   ],
   "source": [
    "from datetime import timedelta\n",
    "\n",
    "# Reserve for 1 hour\n",
    "duration_seconds = timedelta(hours=1).total_seconds()\n",
    "ip_list = [\"192.168.100.1\", \"192.168.100.2\"]\n",
    "\n",
    "try:\n",
    "    res = lbm.reserve_lb_in_seconds(\n",
    "        lb_id=\"ejfat\", seconds=duration_seconds, senders=ip_list)\n",
    "    if res.has_error():\n",
    "        print(\"Reserve LB by seconds error: \", res.error())\n",
    "    else:\n",
    "        print(\"Create LBManager obj succeeded: \", res.value())  # the n-th active LBManager\n",
    "except RuntimeError as e:\n",
    "    print(\"Caught runtime error:\", e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you see a \"failed to reserve load balancer: a maximum of 4 load balancers are supported\" error, restart the `udplbd` container on the FABRIC `cpnode` and try again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Instance token: 955d427b68d84c56e0f197dbcb80ff5217b21c7ff13e5d4334ec33dd31fe177f\n"
     ]
    }
   ],
   "source": [
    "# Some basic verification\n",
    "rt_uri = lbm.get_uri()\n",
    "assert rt_uri.has_sync_addr() == True\n",
    "assert rt_uri.has_data_addr() == True\n",
    "assert rt_uri.get_instance_token().value() != \"\"\n",
    "print(f\"Instance token: {rt_uri.get_instance_token().value()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Free LB\n",
    "res = lbm.free_lb()\n",
    "if res.has_error():\n",
    "    print(\"Free LB error: \", res.error())\n",
    "assert res.value() == 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Create LBManager obj succeeded:  0\n"
     ]
    }
   ],
   "source": [
    "# Create the LBManager object and reserve for 1 hour\n",
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "\n",
    "try:\n",
    "    res = lbm.reserve_lb_in_seconds(\n",
    "        lb_id=\"ejfat\", seconds=duration_seconds, senders=ip_list)\n",
    "    if res.has_error():\n",
    "        print(\"Reserve LB by seconds error: \", res.error())\n",
    "    else:\n",
    "        print(\"Create LBManager obj succeeded: \", res.value())\n",
    "except RuntimeError as e:\n",
    "    print(\"Caught runtime error:\", e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sync IPv4 addr is:  192.168.0.3\n",
      "LB id is:  2\n"
     ]
    }
   ],
   "source": [
    "sync_addr_1 = lbm.get_uri().get_sync_addr().value()\n",
    "print(\"Sync IPv4 addr is: \", sync_addr_1[0])\n",
    "lb_id = lbm.get_uri().lb_id\n",
    "print(\"LB id is: \", lb_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ejfats://udplbd@192.168.0.3:18347/lb/1?sync=192.168.88.199:1234&data=127.0.0.1\n"
     ]
    }
   ],
   "source": [
    "# Creat a new EjfatURI obj from env var\n",
    "new_uri_rt = e2sar_py.EjfatURI.get_from_env()\n",
    "if new_uri_rt.has_error():\n",
    "    print(\"Return EjfatURI from env var error: \", new_uri_rt.error())\n",
    "print(new_uri_rt.value())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create another LBManager object based on the new EjfatURI obj\n",
    "lbm_new = e2sar_py.ControlPlane.LBManager(new_uri_rt.value(), False)\n",
    "\n",
    "# LB ID is an integer. Make sure to convert it to a string\n",
    "res = lbm_new.get_lb_by_id(str(lb_id))\n",
    "if res.has_error():\n",
    "    print(\"Get LB by id error: \", res.error())\n",
    "assert res.has_error() == False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "New sync addr is:  192.168.0.3\n"
     ]
    }
   ],
   "source": [
    "# Ensure the 2 LBManager objects have the same sync address\n",
    "sync_addr_2 = lbm_new.get_uri().get_sync_addr().value()\n",
    "print(\"New sync addr is: \", sync_addr_2[0])\n",
    "assert str(sync_addr_1[0]) == str(sync_addr_2[0])\n",
    "assert sync_addr_1[1] == sync_addr_2[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Free LB succeeded!\n"
     ]
    }
   ],
   "source": [
    "def free_lbmanager(lb_obj : e2sar_py.ControlPlane.LBManager):\n",
    "    res = lb_obj.free_lb()\n",
    "    if res.has_error():\n",
    "        print(f\"Free LB error: \", res.error())\n",
    "    assert res.value() == 0\n",
    "    print(\"Free LB succeeded!\")\n",
    "\n",
    "free_lbmanager(lbm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest2_1\n",
    "\n",
    "Create a `LBManager` object with empty senders list. Add and remove the senders later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Create LBManager obj succeeded:  0\n"
     ]
    }
   ],
   "source": [
    "# Create the LBManager object from an empty sender list\n",
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "\n",
    "try:\n",
    "    res = lbm.reserve_lb_in_seconds(\n",
    "        lb_id=\"ejfat\", seconds=duration_seconds, senders=[])\n",
    "    if res.has_error():\n",
    "        print(\"Reserve LB by seconds error: \", res.error())\n",
    "        free_lbmanager(lbm)\n",
    "    else:\n",
    "        print(\"Create LBManager obj succeeded: \", res.value())\n",
    "except RuntimeError as e:\n",
    "    print(\"Caught runtime error:\", e)\n",
    "\n",
    "assert lbm.get_uri().has_sync_addr() == True\n",
    "assert lbm.get_uri().has_data_addr() == True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add senders list\n",
    "res = lbm.add_senders(ip_list)\n",
    "if res.has_error():\n",
    "    print(\"Add senders to LBManager error: \", res.error())\n",
    "    free_lbmanager(lbm)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "LB ID is:  3\n"
     ]
    }
   ],
   "source": [
    "# Create a new LBManager object\n",
    "lb_id = lbm.get_uri().lb_id\n",
    "print(\"LB ID is: \", lb_id)\n",
    "\n",
    "lbm_new = e2sar_py.ControlPlane.LBManager(new_uri_rt.value(), False)\n",
    "\n",
    "# LB ID is an integer. Make sure to convert it to a string\n",
    "res = lbm_new.get_lb_by_id(str(lb_id))\n",
    "if res.has_error():\n",
    "    print(\"Get LB by id error: \", res.error())\n",
    "assert res.has_error() == False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compare the 2 objs' sync addresses\n",
    "sync_addr_1 = lbm.get_uri().get_sync_addr().value()\n",
    "sync_addr_2 = lbm_new.get_uri().get_sync_addr().value()\n",
    "assert(str(sync_addr_1[0]) == str(sync_addr_2[0]))\n",
    "assert(sync_addr_1[1] == sync_addr_2[1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Remove senders from the LBManager\n",
    "res = lbm.remove_senders(ip_list)\n",
    "if res.has_error():\n",
    "    print(\"Remove senders from LBManager error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "assert(res.has_error() == False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Free LB succeeded!\n"
     ]
    }
   ],
   "source": [
    "free_lbmanager(lbm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest3\n",
    "\n",
    "Register and unregister worker."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Create LBManager obj succeeded:  0\n"
     ]
    }
   ],
   "source": [
    "# Create a LBManager obj and reserve for 1 hr\n",
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "\n",
    "try:\n",
    "    res = lbm.reserve_lb_in_seconds(\n",
    "        lb_id=\"ejfat\", seconds=duration_seconds, senders=ip_list)\n",
    "    if res.has_error():\n",
    "        print(\"Reserve LB by seconds error: \", res.error())\n",
    "        free_lbmanager(lbm)\n",
    "    else:\n",
    "        print(\"Create LBManager obj succeeded: \", res.value())\n",
    "except RuntimeError as e:\n",
    "    print(\"Caught runtime error:\", e)\n",
    "\n",
    "assert lbm.get_uri().has_sync_addr() == True\n",
    "assert lbm.get_uri().has_data_addr() == True"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Every registered worker is required to send its state every 100ms; otherwise, it will be automatically deregistered. A single code block ensures that `register_worker`, `send_state`, and `deregister_worker` all execute within the 100ms timeframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Register worker succeeded\n",
      "\n",
      "Send state succeeded\n",
      "\n",
      "Session id is:  1\n",
      "Session token is:  9f5dbfd36e86f520eea8a808faf79129a709f896df3a1123f8bed664c400c1d4\n",
      "\n",
      "Deregister worker succeeded\n"
     ]
    }
   ],
   "source": [
    "# Register worker\n",
    "res = lbm.register_worker(\n",
    "    \"my_node\",  # node name\n",
    "    (e2sar_py.IPAddress.from_string(\"192.168.101.5\"), 10000),  # Ip_addr\n",
    "    0.5,  # weight\n",
    "    10,  # source_count\n",
    "    1.0,  # min_factor\n",
    "    1.0   # max_factor\n",
    ")\n",
    "if res.has_error():\n",
    "    print(\"Register worker error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "assert(res.value() == 0)\n",
    "print(\"Register worker succeeded\")\n",
    "\n",
    "# Send state\n",
    "print()\n",
    "res = lbm.send_state(0.8, 1.0, True)\n",
    "if res.has_error():\n",
    "    print(\"Send state error: \", res.error())\n",
    "assert(res.value() == 0)\n",
    "print(\"Send state succeeded\")\n",
    "\n",
    "# Verifications\n",
    "print()\n",
    "session_id = lbm.get_uri().session_id\n",
    "print(\"Session id is: \", session_id)\n",
    "session_token = lbm.get_uri().get_session_token().value()\n",
    "print(\"Session token is: \", session_token)\n",
    "\n",
    "# Deregister\n",
    "print()\n",
    "res = lbm.deregister_worker()\n",
    "if res.has_error():\n",
    "    print(\"Deregister worker error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "assert(res.value() == 0)\n",
    "print(\"Deregister worker succeeded\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Free LB succeeded!\n"
     ]
    }
   ],
   "source": [
    "free_lbmanager(lbm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest4\n",
    "\n",
    "Examples with get LB status."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Create LBManager obj succeeded:  0\n"
     ]
    }
   ],
   "source": [
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "\n",
    "try:\n",
    "    res = lbm.reserve_lb_in_seconds(\n",
    "        lb_id=\"ejfat\", seconds=duration_seconds, senders=ip_list)\n",
    "    if res.has_error():\n",
    "        print(\"Reserve LB by seconds error: \", res.error())\n",
    "        free_lbmanager(lbm)\n",
    "    else:\n",
    "        print(\"Create LBManager obj succeeded: \", res.value())\n",
    "except RuntimeError as e:\n",
    "    print(\"Caught runtime error:\", e)\n",
    "\n",
    "assert lbm.get_uri().has_sync_addr() == True\n",
    "assert lbm.get_uri().has_data_addr() == True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Register worker succeeded\n",
      "\n",
      "Send state for 25 times\n",
      "\n",
      "Get LB status succeeded:  <e2sar_py.ControlPlane.LoadBalancerStatusReply object at 0x7f9a0e44bab0>\n"
     ]
    }
   ],
   "source": [
    "# Register a worker and send states for 25 times\n",
    "res = lbm.register_worker(\n",
    "    \"my_node\",  # node name\n",
    "    (e2sar_py.IPAddress.from_string(\"192.168.101.5\"), 10000),  # ip_addr\n",
    "    0.5,  # weight\n",
    "    10,  # source_count\n",
    "    1.0,  # min_factor\n",
    "    1.0   # max_factor\n",
    ")\n",
    "if res.has_error():\n",
    "    print(\"Register worker error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "assert(res.value() == 0)\n",
    "print(\"Register worker succeeded\")\n",
    "\n",
    "import time\n",
    "print(\"\\nSend state for 25 times\")\n",
    "for i in range(25):\n",
    "    res = lbm.send_state(0.8, 1.0, True)\n",
    "    if res.has_error():\n",
    "        print(f\"{i + 1}-th send state error: \", res.error())\n",
    "    assert(res.value() == 0)\n",
    "    # print(f\"{i + 1}-th send state succeeded\")\n",
    "    time.sleep(0.1)  # sleep for 100ms\n",
    "\n",
    "print()\n",
    "status_res = lbm.get_lb_status()\n",
    "if (status_res is None):\n",
    "    print(\"Get LB status error\")\n",
    "print(\"Get LB status succeeded: \", status_res)\n",
    "\n",
    "# Ensure the returned senders match the fact\n",
    "res = lbm.get_sender_addresses(status_res)\n",
    "assert(res == ip_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Worker status:\n",
      "    name: my_node\n",
      "    fill_percent: 0.800000011920929\n",
      "    control_signal: 1.0\n",
      "    slots_assigned: 512\n",
      "    last_updated: 2024-10-02T04:18:05.888130752Z\n"
     ]
    }
   ],
   "source": [
    "# Get worker status\n",
    "workers = lbm.get_worker_statuses(status_res)\n",
    "\n",
    "assert len(workers) == 1\n",
    "\n",
    "def print_workerstatus(w : e2sar_py.ControlPlane.WorkerStatus):\n",
    "    print(f\"Worker status:\")\n",
    "    print(f\"    name: {w.get_name()}\")\n",
    "    print(f\"    fill_percent: {w.get_fill_percent()}\")\n",
    "    print(f\"    control_signal: {w.get_control_signal()}\")\n",
    "    print(f\"    slots_assigned: {w.get_slots_assigned()}\")\n",
    "    print(f\"    last_updated: {w.get_last_updated()}\")\n",
    "\n",
    "# print a worker's status\n",
    "print_workerstatus(workers[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert(workers[0].get_name() == \"my_node\")\n",
    "\n",
    "DELTA = 0.000001\n",
    "assert(abs(workers[0].get_fill_percent() - 0.8) < DELTA)\n",
    "assert(abs(workers[0].get_control_signal() - 1) < DELTA)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Timestamp:  2024-10-02T04:18:05.990769793Z\n",
      "expiresAt:  2024-10-02T05:18:03Z\n",
      "currentEpoch:  3\n",
      "currentPredictedEventNumber:  9223372036854775808\n",
      "workerStatusList:  [<e2sar_py.ControlPlane.WorkerStatus object at 0x7f9a343dde70>]\n"
     ]
    }
   ],
   "source": [
    "lb_status = lbm.as_lb_status(status_res)\n",
    "dir(lb_status)\n",
    "assert(lb_status.senderAddressList == ip_list)\n",
    "assert(len(lb_status.workerStatusList) == 1)\n",
    "print(\"Timestamp: \", lb_status.timestamp)\n",
    "print(\"expiresAt: \", lb_status.expiresAt)\n",
    "print(\"currentEpoch: \", lb_status.currentEpoch)\n",
    "print(\"currentPredictedEventNumber: \", lb_status.currentPredictedEventNumber)\n",
    "print(\"workerStatusList: \", lb_status.workerStatusList)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "DELTA = 0.000001\n",
    "\n",
    "assert(lb_status.workerStatusList[0].get_name() == \"my_node\")\n",
    "assert(abs(lb_status.workerStatusList[0].get_fill_percent() - 0.8) < DELTA)\n",
    "assert(abs(lb_status.workerStatusList[0].get_control_signal() - 1) < DELTA)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deregister worker succeeded\n",
      "Free LB succeeded!\n"
     ]
    }
   ],
   "source": [
    "res = lbm.deregister_worker()\n",
    "if res.has_error():\n",
    "    print(\"Deregister worker error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "else:\n",
    "    print(\"Deregister worker succeeded\")\n",
    "\n",
    "free_lbmanager(lbm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest4_1\n",
    "\n",
    "Reserve with empty sender list first, register workers, add senders, get status, unregister workers, get status and free the LB."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Create LBManager obj succeeded:  0\n",
      "Register worker succeeded\n",
      "\n",
      "Session id is:  3\n",
      "Session token is:  68cb8d9db7a4b367d4d57572a785c1d68aca22c3331fa52d4fdd1e2304c5c9ce\n",
      "\n",
      "Send state for 25 times\n",
      "Add senders succeeded.\n",
      "\n",
      "Get LB status succeeded:  <e2sar_py.ControlPlane.LoadBalancerStatusReply object at 0x7f9a343ce9b0>\n",
      "Remove senders succeeded.\n",
      "\n",
      "Get LB status succeeded:  <e2sar_py.ControlPlane.LoadBalancerStatusReply object at 0x7f9a0e5ee1f0>\n",
      "\n",
      "Deregister worker succeeded\n",
      "Free LB succeeded!\n"
     ]
    }
   ],
   "source": [
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "\n",
    "try:\n",
    "    res = lbm.reserve_lb_in_seconds(\n",
    "        lb_id=\"ejfat\", seconds=duration_seconds, senders=[])\n",
    "    if res.has_error():\n",
    "        print(\"Reserve LB by seconds error: \", res.error())\n",
    "        free_lbmanager(lbm)\n",
    "    else:\n",
    "        print(\"Create LBManager obj succeeded: \", res.value())\n",
    "except RuntimeError as e:\n",
    "    print(\"Caught runtime error:\", e)\n",
    "\n",
    "assert lbm.get_uri().has_sync_addr() == True\n",
    "assert lbm.get_uri().has_data_addr() == True\n",
    "\n",
    "res = lbm.register_worker(\n",
    "    \"my_node\",  # node name\n",
    "    (e2sar_py.IPAddress.from_string(\"192.168.101.5\"), 10000),  # ip_addr\n",
    "    0.5,  # weight\n",
    "    10,  # source_count\n",
    "    1.0,  # min_factor\n",
    "    1.0   # max_factor\n",
    ")\n",
    "if res.has_error():\n",
    "    print(\"Register worker error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "assert(res.value() == 0)\n",
    "print(\"Register worker succeeded\")\n",
    "\n",
    "print()\n",
    "session_id = lbm.get_uri().session_id\n",
    "print(\"Session id is: \", session_id)\n",
    "session_token = lbm.get_uri().get_session_token().value()\n",
    "print(\"Session token is: \", session_token)\n",
    "\n",
    "print(\"\\nSend state for 25 times\")\n",
    "for i in range(25):\n",
    "    res = lbm.send_state(0.8, 1.0, True)\n",
    "    if res.has_error():\n",
    "        print(f\"{i + 1}-th send state error: \", res.error())\n",
    "    assert(res.value() == 0)\n",
    "    # print(f\"{i + 1}-th send state succeeded\")\n",
    "    time.sleep(0.1)  # sleep for 100ms\n",
    "\n",
    "# Add senders list\n",
    "res = lbm.add_senders(ip_list)\n",
    "if res.has_error():\n",
    "    print(\"Add senders to LBManager error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "print(\"Add senders succeeded.\")\n",
    "\n",
    "# Get LB Status\n",
    "print()\n",
    "status_res = lbm.get_lb_status()\n",
    "if (status_res is None):\n",
    "    print(\"Get LB status error\")\n",
    "print(\"Get LB status succeeded: \", status_res)\n",
    "\n",
    "res = lbm.get_sender_addresses(status_res)\n",
    "assert(res == ip_list)\n",
    "workers = lbm.get_worker_statuses(status_res)\n",
    "assert len(workers) == 1\n",
    "assert(workers[0].get_name() == \"my_node\")\n",
    "assert(abs(workers[0].get_fill_percent() - 0.8) < DELTA)\n",
    "assert(abs(workers[0].get_control_signal() - 1) < DELTA)\n",
    "\n",
    "# Remove senders\n",
    "res = lbm.remove_senders(ip_list)\n",
    "if res.has_error():\n",
    "    print(\"Remove senders from LBManager error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "print(\"Remove senders succeeded.\")\n",
    "\n",
    "print()\n",
    "status_res = lbm.get_lb_status()\n",
    "if (status_res is None):\n",
    "    print(\"Get LB status error\")\n",
    "print(\"Get LB status succeeded: \", status_res)\n",
    "res = lbm.get_sender_addresses(status_res)\n",
    "assert(res == [])\n",
    "\n",
    "print()\n",
    "res = lbm.deregister_worker()\n",
    "if res.has_error():\n",
    "    print(\"Deregister worker error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "else:\n",
    "    print(\"Deregister worker succeeded\")\n",
    "\n",
    "free_lbmanager(lbm)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest5\n",
    "Get version tuple."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "LBManager version:  ('ef70f7fb3a537eebe83132fc5d11f44512c8691e', 'v0.3.1', 'v0.3.0')\n"
     ]
    }
   ],
   "source": [
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "print(\"LBManager version: \", lbm.get_version())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Do not need to free the LB since no reservation is involved."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### LBMLiveTest6\n",
    "\n",
    "Demonstrate the usage of getting the overview of the LB."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Create LBManager obj succeeded:  0\n",
      "Register worker succeeded\n",
      "\n",
      "Session id is:  4\n",
      "Session token is:  8800e7562f0a9a4b78bff20d8a20a9047201faaee9658c65e41e57b8e3fe41af\n",
      "\n",
      "Send state for 25 times\n",
      "\n",
      "Get LB status succeeded:  <e2sar_py.ControlPlane.LoadBalancerStatusReply object at 0x7f9a0e4391f0>\n",
      "Sender addresses: ['192.168.20.1', '192.168.20.2']\n",
      "Current Epoch: 2\n",
      "Current predicted Event No: 9223372036854775808\n",
      "Workers: [<e2sar_py.ControlPlane.WorkerStatus object at 0x7f9a0e5ee1f0>]\n",
      "Timestamp: 2024-10-02T04:18:11.402769466Z\n",
      "expiresAt: 2024-10-02T05:18:08Z\n",
      "\n",
      "Send state for 25 times\n",
      "\n",
      "LB id:  7\n",
      "1\n"
     ]
    }
   ],
   "source": [
    "ip_list = [\"192.168.20.1\", \"192.168.20.2\"]\n",
    "\n",
    "lbm = e2sar_py.ControlPlane.LBManager(uri_rt_obj.value(), False)\n",
    "\n",
    "try:\n",
    "    res = lbm.reserve_lb_in_seconds(\n",
    "        lb_id=\"mylb\", seconds=duration_seconds, senders=ip_list)\n",
    "    if res.has_error():\n",
    "        print(\"Reserve LB by seconds error: \", res.error())\n",
    "        free_lbmanager(lbm)\n",
    "    else:\n",
    "        print(\"Create LBManager obj succeeded: \", res.value())\n",
    "except RuntimeError as e:\n",
    "    print(\"Caught runtime error:\", e)\n",
    "\n",
    "assert lbm.get_uri().has_sync_addr() == True\n",
    "assert lbm.get_uri().has_data_addr() == True\n",
    "\n",
    "res = lbm.register_worker(\n",
    "    \"my_node\",  # node name\n",
    "    (e2sar_py.IPAddress.from_string(\"192.168.101.5\"), 10000),  # ip_addr\n",
    "    0.5,  # weight\n",
    "    10,  # source_count\n",
    "    1.0,  # min_factor\n",
    "    1.0   # max_factor\n",
    ")\n",
    "if res.has_error():\n",
    "    print(\"Register worker error: \", res.error())\n",
    "    free_lbmanager(lbm)\n",
    "assert(res.value() == 0)\n",
    "print(\"Register worker succeeded\")\n",
    "\n",
    "print()\n",
    "session_id = lbm.get_uri().session_id\n",
    "print(\"Session id is: \", session_id)\n",
    "session_token = lbm.get_uri().get_session_token().value()\n",
    "print(\"Session token is: \", session_token)\n",
    "\n",
    "print(\"\\nSend state for 25 times\")\n",
    "for i in range(25):\n",
    "    res = lbm.send_state(0.8, 1.0, True)\n",
    "    if res.has_error():\n",
    "        print(f\"{i + 1}-th send state error: \", res.error())\n",
    "    assert(res.value() == 0)\n",
    "    # print(f\"{i + 1}-th send state succeeded\")\n",
    "    time.sleep(0.1)  # sleep for 100ms\n",
    "\n",
    "# Get LB Status\n",
    "print()\n",
    "status_res = lbm.get_lb_status()\n",
    "if (status_res is None):\n",
    "    print(\"Get LB status error\")\n",
    "print(\"Get LB status succeeded: \", status_res)\n",
    "\n",
    "res = lbm.get_sender_addresses(status_res)\n",
    "assert(res == ip_list)\n",
    "workers = lbm.get_worker_statuses(status_res)\n",
    "assert len(workers) == 1\n",
    "\n",
    "# as_lb_status() usage\n",
    "res = lbm.as_lb_status(status_res)\n",
    "print(f\"Sender addresses: {res.senderAddressList}\")\n",
    "print(f\"Current Epoch: {res.currentEpoch}\")\n",
    "print(f\"Current predicted Event No: {res.currentPredictedEventNumber}\")\n",
    "print(f\"Workers: {res.workerStatusList}\")\n",
    "print(f\"Timestamp: {res.timestamp}\")\n",
    "print(f\"expiresAt: {res.expiresAt}\")\n",
    "assert(res.senderAddressList == ip_list)\n",
    "assert(len(res.workerStatusList) == 1)\n",
    "assert(res.workerStatusList[0].get_name() == \"my_node\")\n",
    "assert(abs(res.workerStatusList[0].get_fill_percent() - 0.8) < DELTA)\n",
    "assert(abs(res.workerStatusList[0].get_control_signal() - 1) < DELTA)\n",
    "\n",
    "print(\"\\nSend state for 25 times\")\n",
    "for i in range(25):\n",
    "    res = lbm.send_state(0.8, 1.0, True)\n",
    "    if res.has_error():\n",
    "        print(f\"{i + 1}-th send state error: \", res.error())\n",
    "    assert(res.value() == 0)\n",
    "    # print(f\"{i + 1}-th send state succeeded\")\n",
    "    time.sleep(0.1)  # sleep for 100ms\n",
    "\n",
    "# as_overview_message() usage\n",
    "print()\n",
    "overview_res = lbm.get_lb_overview()\n",
    "if (overview_res is None):\n",
    "    print(\"Get LB overview error\")\n",
    "    free_lbmanager(lbm)\n",
    "\n",
    "res = lbm.as_overview_message(overview_res)   # a list\n",
    "if not res:\n",
    "    print(\"Get overview message error\")\n",
    "    free_lbmanager(lbm)\n",
    "print(\"LB id: \", res[0].lb_id)\n",
    "assert(res[0].name == \"mylb\")\n",
    "assert(res[0].lb_status.senderAddressList == ip_list)\n",
    "print(len(res[0].lb_status.workerStatusList))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Free LB succeeded!\n"
     ]
    }
   ],
   "source": [
    "free_lbmanager(lbm)"
   ]
  }
 ],
 "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.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}