notebooks/zeph_topology_execution.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Internet topology analysis\n",
    "\n",
    "*Hypothesis*: Zeph will work at scale with Diamond-Miner. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "from pathlib import Path\n",
    "\n",
    "logger = logging.getLogger(__name__)\n",
    "logger.setLevel(logging.DEBUG)\n",
    "script_formatter = logging.Formatter(\n",
    "    \"%(asctime)s :: SCRIPT :: %(levelname)s :: %(message)s\"\n",
    ")\n",
    "stream_handler = logging.StreamHandler()\n",
    "stream_handler.setLevel(logging.DEBUG)\n",
    "stream_handler.setFormatter(script_formatter)\n",
    "logger.addHandler(stream_handler)\n",
    "\n",
    "# Directory of the experiment\n",
    "exp_dir = Path(\"../resources/data/measurements/exp3/\")\n",
    "exp_dir.mkdir(parents=True, exist_ok=True)\n",
    "\n",
    "# Directory of the total prefixes and exploitation prefixes pickle files\n",
    "prefixes_dir = exp_dir / \"prefixes\"\n",
    "prefixes_dir.mkdir(parents=True, exist_ok=True)\n",
    "exploitation_dir = exp_dir / \"exploitation\"\n",
    "exploitation_dir.mkdir(parents=True, exist_ok=True)\n",
    "\n",
    "file_handler = logging.FileHandler(exp_dir / \"log.txt\")\n",
    "file_handler.setLevel(logging.INFO)\n",
    "file_handler.setFormatter(script_formatter)\n",
    "logger.addHandler(file_handler)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Configuration\n",
    "\n",
    "In this section:\n",
    "\n",
    "* we get the configuration of the Iris API and database\n",
    "* we get the configuration of the experiment itself"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get Iris API / database credentials\n",
    "import json\n",
    "import requests\n",
    "from zeph.drivers import create_auth_header\n",
    "\n",
    "def get_database_url(url, headers):\n",
    "    req = requests.get(url + \"/users/me/services\", headers=headers)\n",
    "    if req.status_code != 200:\n",
    "        raise ValueError(\"Unable to get the database URL\")\n",
    "    creds = req.json()[\"clickhouse\"]\n",
    "    return f\"{creds['base_url']}?database={creds['database']}&user={creds['username']}&password={creds['password']}\"\n",
    "\n",
    "config = json.load(open(\"../config.json\"))\n",
    "headers = create_auth_header(config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"])\n",
    "database_url = get_database_url(config[\"iris_url\"], headers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Experiment parameters\n",
    "tool = \"diamond-miner\"\n",
    "epsilon = 0.1\n",
    "protocol = \"icmp\"\n",
    "min_ttl = 8\n",
    "max_ttl = 32\n",
    "\n",
    "measurement_tags = [\"!public\", \"zeph-evaluation\", \"exp3\"]\n",
    "\n",
    "\n",
    "# This can be overrided by the pilot configuration (see below)\n",
    "n_cycles = 6\n",
    "global_budget = 11_994_366\n",
    "\n",
    "# You can generate this file by following these instructions: https://github.com/dioptra-io/zeph#-generate-the-bgp-prefix-file\n",
    "bgp_prefixes_path = Path(\"../resources/data/bgp_prefixes.pickle\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Pilot definition (optional)\n",
    "\n",
    "If you don't want to run the experiment on the entire universe of BGP prefixes, you can define a pilot."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Enable/disable pilot experiment\n",
    "enable_pilot = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pickle\n",
    "\n",
    "if not enable_pilot:\n",
    "    with bgp_prefixes_path.open(\"rb\") as fd:\n",
    "        bgp_prefixes = pickle.load(fd)\n",
    "        logger.info(f\"Number of BGP prefixes {len(bgp_prefixes)}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "\n",
    "def pilot_bgp_prefixes(bgp_prefixes, n_prefixes):\n",
    "    current_n_prefixes = 0\n",
    "    subset_bgp_prefixes = []\n",
    "\n",
    "    random.shuffle(bgp_prefixes)\n",
    "\n",
    "    for bgp_prefix in bgp_prefixes:\n",
    "        if current_n_prefixes > n_prefixes:\n",
    "            break\n",
    "\n",
    "        subset_bgp_prefixes.append(bgp_prefix)\n",
    "        current_n_prefixes += len(bgp_prefix)\n",
    "\n",
    "    logger.info(f\"Number of /24 prefixes: {current_n_prefixes}\")\n",
    "    return subset_bgp_prefixes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Optionally override experiment parameters\n",
    "if enable_pilot:\n",
    "    n_cycles = 10\n",
    "    global_budget = 100_000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### BGP prefixes subset creation\n",
    "\n",
    "Here you can define the subset of BGP prefixes you want to run the pilot experiment on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "if enable_pilot:\n",
    "    # Enable/diable bgp prefix subset creation\n",
    "    create_bgp_prefixes_subset = False    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-01-21 15:07:14,436 :: SCRIPT :: INFO :: Number of BGP prefixes 6111\n"
     ]
    }
   ],
   "source": [
    "import pickle\n",
    "\n",
    "if enable_pilot and not create_bgp_prefixes_subset:\n",
    "    # Override prefix path\n",
    "    bgp_prefixes_path = Path(\"../resources/data/bgp_prefixes_pilot.pickle\")\n",
    "\n",
    "    if not create_bgp_prefixes_subset:\n",
    "        with bgp_prefixes_path.open(\"rb\") as fd:\n",
    "            bgp_prefixes = pickle.load(fd)\n",
    "        logger.info(f\"Number of BGP prefixes {len(bgp_prefixes)}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "if enable_pilot and create_bgp_prefixes_subset:\n",
    "        with bgp_prefixes_path.open(\"rb\") as fd:\n",
    "            bgp_prefixes = pickle.load(fd)\n",
    "        bgp_prefixes = pilot_bgp_prefixes(bgp_prefixes, global_budget)\n",
    "        logger.info(f\"Number of BGP prefixes {len(bgp_prefixes)}\")\n",
    "\n",
    "        # Override prefix path\n",
    "        bgp_prefixes_path = Path(\"../resources/data/bgp_prefixes_pilot.pickle\")\n",
    "        with bgp_prefixes_path.open(\"wb\") as fd:\n",
    "            pickle.dump(bgp_prefixes, fd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Instance definition\n",
    "\n",
    "In this section we define the instance(s) of the experiment.\n",
    "An instance is one workflow run with a set of parameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from zeph.main import create_selector\n",
    "from zeph.drivers import iris_driver, get_previous_measurement_agents\n",
    "\n",
    "\n",
    "def adaptive_instance(\n",
    "    name,\n",
    "    n_cycles,\n",
    "    compute_budget,\n",
    "    bgp_prefixes=None,\n",
    "    bgp_awareness=True,\n",
    "    exploitation_only=False,\n",
    "    previous_measurement_uuid=None,\n",
    "    dry_run=False,\n",
    "):\n",
    "    \"\"\"Instance of the experiment.\"\"\"\n",
    "    agents_uuid = None\n",
    "    \n",
    "    for _ in range(n_cycles):\n",
    "\n",
    "        if previous_measurement_uuid:\n",
    "            logger.debug(\"Get previous measurement agents\")\n",
    "            headers = create_auth_header(config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"])\n",
    "            agents_uuid = get_previous_measurement_agents(\n",
    "                config[\"iris_url\"], previous_measurement_uuid, headers\n",
    "            )\n",
    "\n",
    "        logger.debug(f\"Create selector with previous measurement {previous_measurement_uuid}\")\n",
    "        selector = create_selector(\n",
    "            database_url, \n",
    "            epsilon, \n",
    "            bgp_prefixes, \n",
    "            previous_measurement_uuid=previous_measurement_uuid,\n",
    "            previous_agents_uuid=agents_uuid,\n",
    "            bgp_awareness=bgp_awareness,\n",
    "        )\n",
    "\n",
    "        logger.debug(\"Execute Iris driver\")\n",
    "        measurement_uuid, exploitation_per_agent, prefixes_per_agent = iris_driver(\n",
    "            config[\"iris_url\"],\n",
    "            config[\"iris_username\"],\n",
    "            config[\"iris_password\"],\n",
    "            name,\n",
    "            tool,\n",
    "            protocol,\n",
    "            min_ttl,\n",
    "            max_ttl,\n",
    "            selector,\n",
    "            compute_budget,\n",
    "            logger,\n",
    "            measurement_tags=measurement_tags,\n",
    "            exploitation_only=exploitation_only,\n",
    "            dry_run=dry_run,\n",
    "        )\n",
    "\n",
    "        previous_measurement_uuid = measurement_uuid\n",
    "\n",
    "        recap = {k: len(v) for k, v in prefixes_per_agent.items()}\n",
    "        logger.info(f\"{name} - {measurement_uuid}: {recap}\")\n",
    "\n",
    "        with (exploitation_dir / (\"exploitation_\" + measurement_uuid + \".pickle\")).open(\n",
    "            \"wb\"\n",
    "        ) as fd:\n",
    "            pickle.dump(exploitation_per_agent, fd)\n",
    "        with (prefixes_dir / (\"prefixes_\" + measurement_uuid + \".pickle\")).open(\n",
    "            \"wb\"\n",
    "        ) as fd:\n",
    "            pickle.dump(prefixes_per_agent, fd)\n",
    "        yield measurement_uuid"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Experiment definition\n",
    "\n",
    "In this section we define the experiment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dry run, skip the execution\n",
    "dry_run = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from math import floor\n",
    "\n",
    "zeph_uuids_dm = []\n",
    "zeph_dm = adaptive_instance(\n",
    "    \"edgenet-1\",\n",
    "    n_cycles,\n",
    "    lambda _: floor(0.20 * global_budget),\n",
    "    bgp_prefixes=bgp_prefixes,\n",
    "    bgp_awareness=False,\n",
    "    exploitation_only=False,\n",
    "    previous_measurement_uuid=\"d9ed821d-850a-4fb7-b7c1-8766344803d1\",\n",
    "    dry_run=dry_run,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Experiment execution\n",
    "\n",
    "We execute the experiment by running the workflow on the instance(s)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "from zeph.drivers import create_auth_header\n",
    "\n",
    "def check_measurement_finished(url, username, password, measurement_uuid):\n",
    "    headers = create_auth_header(url, username, password)\n",
    "    req = requests.get(url + f\"/measurements/{measurement_uuid}\", headers=headers)\n",
    "    return req.json()[\"state\"] == \"finished\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "\n",
    "for zeph_uuid_dm in zeph_dm:\n",
    "\n",
    "    zeph_uuids_dm.append(zeph_uuid_dm)\n",
    "\n",
    "    while True:\n",
    "        check_zeph_dm = check_measurement_finished(\n",
    "            config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"], zeph_uuid_dm\n",
    "        )\n",
    "\n",
    "        if check_zeph_dm:\n",
    "            break\n",
    "        time.sleep(10)\n",
    "\n",
    "\n",
    "with (exp_dir / \"production.txt\").open(\"w\") as fd:\n",
    "    for uuid in zeph_uuids_dm:\n",
    "        fd.write(uuid + \"\\n\")"
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "4a7c0178b9cdc3269eb00ccea0263d7c21cd35e4d7be0ec2e63761aafbb806b4"
  },
  "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.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}