Check Services with beacons - template.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46cbcc71-797f-49d5-866c-e8098af07755",
   "metadata": {},
   "outputs": [],
   "source": [
    "pip install sqlalchemy==2.0 psycopg psycopg[binary,pool]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "e265f046-306d-4c67-9e58-8d3cb0a50adc",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a1fd3f2705eb4fb79cf3d73c7b3d348e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox(children=(Text(value='', description='User:', placeholder='Введите имя пользователя'), Password(descripti…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7418b51933fd4bdfa9fa3c3499cb7cf9",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(Button(description='Beacons', icon='check', style=ButtonStyle(), tooltip='Check beacons visibil…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "b18215d4fe1442afbeb6262338bcb062",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(FloatProgress(value=0.0, max=1.0),))"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "15f7171a75534a188ec442cf3803a782",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "Output(layout=Layout(border_bottom='2px solid gray', border_left='2px solid gray', border_right='2px solid gra…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import time\n",
    "import asyncio\n",
    "import asyncssh\n",
    "import aiohttp\n",
    "from http.client import HTTPConnection\n",
    "from urllib.parse import urlparse\n",
    "import ipywidgets\n",
    "import threading\n",
    "from sqlalchemy import create_engine, text\n",
    "from sqlalchemy.orm import Session\n",
    "\n",
    "async def site_is_online_async(link, timeout,username,password):\n",
    "    \"\"\"Return True if the target URL is online.\n",
    "    Raise an exception otherwise.\n",
    "    \"\"\"\n",
    "    error = Exception(\"unknown scheme\")\n",
    "    url = link.url\n",
    "    parser = urlparse(url)\n",
    "    host = parser.netloc or parser.path.split(\"/\")[0]\n",
    "    \n",
    "    \n",
    "    if parser.scheme in (\"http\", \"https\"):\n",
    "        async with aiohttp.ClientSession() as session:\n",
    "            try:\n",
    "                await session.head(url, timeout=timeout,ssl=False)\n",
    "                return True\n",
    "            except asyncio.TimeoutError:\n",
    "                error = Exception(\"timed out\")\n",
    "            except Exception as e:\n",
    "                error = e\n",
    "                \n",
    "    if parser.scheme in (\"ssh\"):\n",
    "        if  parser.port == None : \n",
    "            port=22 \n",
    "        else: \n",
    "            port=parser.port\n",
    "            host=host.split(\":\")[0]\n",
    "\n",
    "        #print(host,port)\n",
    "        \n",
    "        try:\n",
    "            async with await asyncio.wait_for(asyncssh.connect(host=host,port=port,username=username, password=password, known_hosts=None), timeout=timeout) as conn:\n",
    "                print((await conn.run('tty', term_type='ansi')).stdout, end='')\n",
    "                return True\n",
    "        except asyncio.TimeoutError:\n",
    "                error = Exception(\"timed out\")\n",
    "        except Exception as e:\n",
    "                error = e\n",
    "                \n",
    "    if parser.scheme in (\"postgre\"):                \n",
    "        if  parser.port == None : \n",
    "            port=5432\n",
    "        else: \n",
    "            port=parser.port\n",
    "            host=host.split(\":\")[0]\n",
    "        \n",
    "        coro = asyncio.open_connection(host, port)\n",
    "        # execute the coroutine with a timeout\n",
    "        try:\n",
    "            # open the connection and wait for a moment\n",
    "            _,writer = await asyncio.wait_for(coro, timeout)\n",
    "            # close connection once opened\n",
    "            writer.close()\n",
    "            # indicate the connection can be opened\n",
    "            return True\n",
    "        except asyncio.TimeoutError:\n",
    "                error = Exception(\"timed out\")\n",
    "        except Exception as e:\n",
    "                error = e\n",
    "                \n",
    "    if parser.scheme in (\"kafka\"):                \n",
    "        if  parser.port == None : \n",
    "            port=9092\n",
    "        else: \n",
    "            port=parser.port\n",
    "            host=host.split(\":\")[0]\n",
    "        \n",
    "        coro = asyncio.open_connection(host, port)\n",
    "        # execute the coroutine with a timeout\n",
    "        try:\n",
    "            # open the connection and wait for a moment\n",
    "            _,writer = await asyncio.wait_for(coro, timeout)\n",
    "            # close connection once opened\n",
    "            writer.close()\n",
    "            # indicate the connection can be opened\n",
    "            return True\n",
    "        except asyncio.TimeoutError:\n",
    "                error = Exception(\"timed out\")\n",
    "        except Exception as e:\n",
    "                error = e\n",
    "                \n",
    "    if parser.scheme in (\"raw\"):                \n",
    "        if  parser.port == None : \n",
    "            port=0\n",
    "        else: \n",
    "            port=parser.port\n",
    "            host=host.split(\":\")[0]\n",
    "        \n",
    "        coro = asyncio.open_connection(host, port)\n",
    "        # execute the coroutine with a timeout\n",
    "        try:\n",
    "            # open the connection and wait for a moment\n",
    "            _,writer = await asyncio.wait_for(coro, timeout)\n",
    "            # close connection once opened\n",
    "            writer.close()\n",
    "            # indicate the connection can be opened\n",
    "            return True\n",
    "        except asyncio.TimeoutError:\n",
    "                error = Exception(\"timed out\")\n",
    "        except Exception as e:\n",
    "                error = e                \n",
    "                \n",
    "    raise error\n",
    "    \n",
    "#out_box = ipywidgets.Output(layout={'border': '2px solid black', 'overflow':'visible'},width='100px')   \n",
    "\n",
    "out_box = ipywidgets.Output(layout=ipywidgets.Layout(width='100%',\n",
    "                                           height='auto',\n",
    "                                           max_height='300px',\n",
    "                                           overflow='hidden scroll',border ='2px solid gray'))\n",
    "\n",
    "    \n",
    "def display_check_result(link, latency_ns, error, sink, agent_token):\n",
    "    \"\"\"Display the result of a connectivity check.\"\"\"\n",
    "    out_box.append_stdout(f'\\nService \"{link.service_name}\" [link: \"{link.url}\"] for role \"{link.role_name}\" is:')\n",
    "    \n",
    "    if (latency_ns):\n",
    "        sink.execute(text(f'''\n",
    "                           insert into dbo.probes \n",
    "                           (link_id,role_id,latency_ns,agent_token)\n",
    "                           values\n",
    "                           ({link.link_id},{link.role_id},{latency_ns},'{agent_token}');\n",
    "                           '''))\n",
    "        sink.commit()\n",
    "        out_box.append_stdout(f'\"Online!\" 👍, latency: {latency_ns*10e-9}s')\n",
    "    else:\n",
    "        sink.execute(text(f'''insert into dbo.probes \n",
    "                              (link_id,role_id,latency_ns,error_message,agent_token) \n",
    "                              values \n",
    "                              ({link.link_id},{link.role_id},null,'{error}','{agent_token}');\n",
    "                           '''))\n",
    "        sink.commit()\n",
    "        out_box.append_stdout(f'\"Offline?\" 👎 \\n  Error: \"{error}\"')\n",
    "\n",
    "progress = ipywidgets.FloatProgress(value=0.0, min=0.0, max=1.0)\n",
    "        \n",
    "\n",
    "async def _asynchronous_check(links,timeout,username,password,sink,agent_token):\n",
    "    async def _check(link,timeout,username,password,sink,agent_token):\n",
    "        error = \"\"\n",
    "        startTime = time.time_ns()\n",
    "        try:\n",
    "            await site_is_online_async(link,timeout,username,password)\n",
    "            latency = time.time_ns() - startTime\n",
    "        except Exception as e:\n",
    "            latency = None\n",
    "            #template = 'An exception of type \"{0}\" occurred. Arguments:\\n{1!r}'\n",
    "            error = str(e) #template.format(type(e).__name__, e.args)\n",
    "        display_check_result(link, latency,error,sink,agent_token)\n",
    "        return (error == \"\")\n",
    "\n",
    "    values_list = await asyncio.gather(*(_check(link,timeout,username,password,sink,agent_token) for link in links))\n",
    "    sink.commit()\n",
    "    sink.close()\n",
    "    return values_list\n",
    "   \n",
    "def check(timeout,links,username,password,sink,agent_token,callback_proc):\n",
    "    \n",
    "    try:\n",
    "        loop = asyncio.get_running_loop()\n",
    "    except RuntimeError:  # 'RuntimeError: There is no current event loop...'\n",
    "        loop = None\n",
    "\n",
    "    if loop and loop.is_running():\n",
    "        #print('Async event loop already running. Adding coroutine to the event loop.')\n",
    "        \n",
    "        tsk = loop.create_task(_asynchronous_check(links,timeout,username,password,sink,agent_token)) \n",
    "        # ^-- https://docs.python.org/3/library/asyncio-task.html#task-object\n",
    "        \n",
    "        # Optionally, a callback function can be executed when the coroutine completes\n",
    "        tsk.add_done_callback(\n",
    "            callback_proc\n",
    "        )\n",
    "    else:\n",
    "        print('Starting new event loop')\n",
    "        result = asyncio.run(_asynchronous_check(links,timeout,username,password,sink,agent_token))\n",
    "    \n",
    "username = ipywidgets.Text(description='User:', placeholder='Введите имя пользователя',value='')\n",
    "password = ipywidgets.Password(description='Password:', placeholder='Введите пароль')\n",
    "role     = ipywidgets.Select(\n",
    "                options=[\n",
    "                    'Developer',\n",
    "                    'QA Engineer',\n",
    "                    'Release Engineer',\n",
    "                    'Analyst',\n",
    "                    'Data Scientist',\n",
    "                    'DevOps Engineer',\n",
    "                    'Support Engineer'\n",
    "                ],\n",
    "                value='Developer',\n",
    "                rows=3,\n",
    "                description='Roles',\n",
    "                disabled=False,\n",
    "                tooltip='Check beacons visibility',\n",
    "            )\n",
    "\n",
    "\n",
    "token    = ipywidgets.Password(description='Token:', placeholder='Введите токен',value='')\n",
    "\n",
    "bvButton = ipywidgets.Button(\n",
    "            value=False,\n",
    "            description='Beacons',\n",
    "            disabled=False,\n",
    "            button_style='', \n",
    "            tooltip='Check beacons visibility',\n",
    "            icon='check' \n",
    ")\n",
    "\n",
    "sButton = ipywidgets.Button(description=\"Start!\",disabled=True, tooltip='Start tracing')\n",
    "cButton = ipywidgets.Button(description=\"Reset\", tooltip='Reset agent')\n",
    "\n",
    "\n",
    "def on_sButton_clicked(b):\n",
    "    progress.value = 0.0\n",
    "    out_box.append_stdout(f'\\n{time.time()} - Start tracing of services!\\n')\n",
    "    \n",
    "    engine=create_engine(f'postgresql+psycopg://{token.value}',\n",
    "                        echo=True, isolation_level=\"AUTOCOMMIT\")\n",
    "    \n",
    "    out_box.append_stdout(f'\\n{engine} - engine created!\\n') \n",
    "    \n",
    "    session = Session(engine)\n",
    "    \n",
    "    out_box.append_stdout(f'\\n{session} - session created!\\n') \n",
    "    \n",
    "    resultSet = session.execute(text(\n",
    "                                        f''' \n",
    "                                        SELECT url, role_name, service_name, link_id, role_id, is_beacon\n",
    "                                        FROM dbo.eligible_links where role_name in ('{role.value}') \n",
    "                                        ''')\n",
    "                                )\n",
    "    out_box.append_stdout(f'\\n{resultSet} - resultSet done!\\n') \n",
    "    \n",
    "    eligible_links = resultSet.fetchall()\n",
    "    \n",
    "    session.commit()\n",
    "    \n",
    "    out_box.append_stdout(f'\\nEngine.url:{engine.url}\\n') \n",
    "    \n",
    "    out_box.append_stdout(f'\\nAgent_token:{engine.url.username}\\n') \n",
    "    \n",
    "    check(timeout=15, links=eligible_links,\n",
    "          username=username.value,password=password.value, \n",
    "          sink=session, agent_token=engine.url.username,\n",
    "          callback_proc = lambda lmd: [out_box.append_stdout(f'\\nTracing of services: done with result={lmd.result()}  << return _asynchronous_check()\\n'),\n",
    "                                       check_callback(check_value_list=lmd.result())\n",
    "                                      ] \n",
    "          \n",
    "         )        \n",
    "                            \n",
    "    out_box.append_stdout(f'\\nAll tracing subtasks were submited!\\n') \n",
    "\n",
    "def on_cButton_clicked(b):\n",
    "    progress.value = 0\n",
    "    sButton.disabled=True\n",
    "    out_box.clear_output()\n",
    "\n",
    "    \n",
    "def check_callback(check_value_list):\n",
    "    progress.value = 1.0\n",
    "    if len(check_value_list) > 0: \n",
    "            if (False in check_value_list):\n",
    "                nrqty =0\n",
    "                for res in check_value_list:\n",
    "                    if not(res): nrqty +=1\n",
    "                \n",
    "                out_box.append_stdout(\n",
    "                   f'''\\n#######################################################################################\n",
    "                       \\n{nrqty} service(s) from {len(check_value_list)} are not reachable, incident reported to upstream \n",
    "                       \\n#######################################################################################\n",
    "                    '''\n",
    "                                  )\n",
    "    else:\n",
    "                out_box.append_stdout(\n",
    "                    '''\\n#######################################################################\n",
    "                       \\nServices are not available in ACL for current role (eligible_links view)\n",
    "                       \\n#######################################################################\n",
    "                    '''\n",
    "                                  )\n",
    "    pass\n",
    "    \n",
    "    \n",
    "def beacons_discovery_callback(discovery_value_list):\n",
    "    progress.value = 1.0\n",
    "    ## discovery_value_list = [False,True]\n",
    "    if len(discovery_value_list) > 0: \n",
    "            if (True in discovery_value_list):\n",
    "                   sButton.disabled=False\n",
    "            else:\n",
    "                nrqty =0\n",
    "                for res in discovery_value_list:\n",
    "                    if not(res): nrqty +=1\n",
    "                out_box.append_stdout(\n",
    "                   f'''\\n#####################################################\n",
    "                       \\nBeacons({nrqty} from {len(discovery_value_list)}) are not reachable, check VPN connectivity\n",
    "                       \\n#####################################################\n",
    "                    '''\n",
    "                                  )\n",
    "    else:\n",
    "                out_box.append_stdout(\n",
    "                    '''\\n#######################################################################\n",
    "                       \\nBeacons are not available in ACL for current role (eligible_links view)\n",
    "                       \\n#######################################################################\n",
    "                    '''\n",
    "                                  )\n",
    "    pass\n",
    "\n",
    "def on_bvButton_clicked(b):\n",
    "    out_box.append_stdout(f'\\n{time.time()} - Beacons discovery started!\\n')\n",
    "    \n",
    "    engine=create_engine(f'postgresql+psycopg://{token.value}',\n",
    "                        echo=True, isolation_level=\"AUTOCOMMIT\")\n",
    "    \n",
    "    out_box.append_stdout(f'\\n{engine} - engine created!\\n') \n",
    "    \n",
    "    session = Session(engine)\n",
    "    \n",
    "    out_box.append_stdout(f'\\n{session} - session created!\\n') \n",
    "    \n",
    "    resultSet = session.execute(text(\n",
    "                                        f''' \n",
    "                                        SELECT url, role_name, service_name, link_id, role_id, is_beacon\n",
    "                                        FROM dbo.eligible_links where role_name in ('{role.value}')\n",
    "                                        and is_beacon\n",
    "                                        ''')\n",
    "                                )\n",
    "    out_box.append_stdout(f'\\n{resultSet} - resultSet done!\\n') \n",
    "    \n",
    "    eligible_links = resultSet.fetchall()\n",
    "    \n",
    "    session.commit()\n",
    "    \n",
    "    out_box.append_stdout(f'\\nEngine.url:{engine.url}\\n') \n",
    "    \n",
    "    out_box.append_stdout(f'\\nAgent_token:{engine.url.username}\\n') \n",
    "    \n",
    "    #check_beacon_visibility(timeout=1, links=eligible_links,\n",
    "        \n",
    "    check(timeout=3, links=eligible_links,\n",
    "          username=username.value,password=password.value, \n",
    "          sink=session, agent_token=engine.url.username,\n",
    "          callback_proc = lambda lmd: [out_box.append_stdout(f'\\nBeacons discovery: done with result={lmd.result()}  << return _asynchronous_check()\\n'),\n",
    "                                         beacons_discovery_callback(discovery_value_list=lmd.result())\n",
    "                                        ] \n",
    "         )        \n",
    "                            \n",
    "bvButton.on_click(on_bvButton_clicked)\n",
    "sButton.on_click(on_sButton_clicked)\n",
    "cButton.on_click(on_cButton_clicked)\n",
    "\n",
    "display(ipywidgets.VBox([username,password,role,token]),\n",
    "        ipywidgets.HBox([bvButton,sButton,cButton]),\n",
    "        ipywidgets.HBox([progress]))\n",
    "display(out_box)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "41f69b45-8cf3-462a-aef8-a54dfd7a23de",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}