notebooks/zeph_allocation_execution.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Prefix allocation analysis\n",
    "\n",
    "*Hypothesis*: for the same per-agent probing budget, allocating prefixes to agents based on an adaptive approach\n",
    "(with the possibility of a prefix to be probed from any number from 0 to n agents)\n",
    "will allow more to be discovered than allocating prefixes to agents randomly (with each prefix being probed from precisely one agent)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "from pathlib import Path\n",
    "\n",
    "logger = logging.getLogger(__name__)\n",
    "logger.setLevel(logging.INFO)\n",
    "script_formatter = logging.Formatter(\n",
    "    \"%(asctime)s :: SCRIPT :: %(levelname)s :: %(message)s\"\n",
    ")\n",
    "stream_handler = logging.StreamHandler()\n",
    "stream_handler.setLevel(logging.INFO)\n",
    "stream_handler.setFormatter(script_formatter)\n",
    "logger.addHandler(stream_handler)\n",
    "\n",
    "# Directory of the experiment\n",
    "exp_dir = Path(\"../resources/data/measurements/exp2-pilot/\")\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": 3,
   "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 = \"yarrp\"\n",
    "protocol = \"icmp\"\n",
    "min_ttl = 8\n",
    "max_ttl = 32\n",
    "\n",
    "measurement_tags = [\"!public\", \"zeph-evaluation\", \"exp2\"]\n",
    "\n",
    "# This can be overrided by the pilot configuration (see below)\n",
    "n_agents = 5\n",
    "n_cycles = 10\n",
    "global_budget = 11_881_416\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_agents = 1\n",
    "    n_cycles = 3\n",
    "    global_budget = 10_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-20 21:23:01,822 :: SCRIPT :: INFO :: Number of BGP prefixes 701\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": "markdown",
   "metadata": {},
   "source": [
    "### Adaptive instance"
   ]
  },
  {
   "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",
    "    epsilon,\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",
    "\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",
    "        measurement_uuid, exploitation_per_agent, prefixes_per_agent = iris_driver(\n",
    "            config[\"iris_url\"],\n",
    "            config[\"iris_username\"],\n",
    "            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": [
    "### Constrained instance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "from zeph.drivers import create_auth_header, get_agents\n",
    "\n",
    "def get_agents_budget(iris_url, iris_username, iris_password, agents_tag, compute_budget):\n",
    "    \"\"\"Get the agents budget.\"\"\"\n",
    "    agents_budget = {}\n",
    "    headers = create_auth_header(iris_url, iris_username, iris_password)\n",
    "    agents = get_agents(iris_url, agents_tag, headers)\n",
    "    for agent in agents:\n",
    "        agents_budget[agent[\"uuid\"]] = compute_budget(agent[\"parameters\"][\"max_probing_rate\"])\n",
    "    return agents_budget"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from zeph.selectors.constrained import ConstrainedRandomSelector, ConstrainedEpsilonDFGSelector\n",
    "\n",
    "def create_constrained_selector(\n",
    "    name, compute_budget, bgp_prefixes, rl=None, measurement_uuid=None\n",
    "):\n",
    "    agents_budget = get_agents_budget(\n",
    "        config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"], name, compute_budget\n",
    "    )\n",
    "\n",
    "    if rl:\n",
    "        selector = ConstrainedEpsilonDFGSelector(\n",
    "            database_url,\n",
    "            epsilon=0.1,\n",
    "            agents_budget=agents_budget,\n",
    "            authorized_prefixes=bgp_prefixes,\n",
    "        )\n",
    "        logger.debug(\"Get discoveries\")\n",
    "        discoveries = selector.compute_discoveries_links(measurement_uuid, agents_budget.keys())\n",
    "\n",
    "        logger.debug(\"Compute rank\")\n",
    "        selector.rank_per_agent = selector.compute_rank(discoveries)\n",
    "\n",
    "        logger.debug(\"Compute dispatch\")\n",
    "        selector.dispatch_per_agent = selector.compute_dispatch()\n",
    "    else:\n",
    "        selector = ConstrainedRandomSelector(\n",
    "            agents_budget=agents_budget, authorized_prefixes=bgp_prefixes\n",
    "        )\n",
    "\n",
    "    return selector"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "def constrained_instance(\n",
    "    name,\n",
    "    n_cycles,\n",
    "    compute_budget,\n",
    "    rl=False,\n",
    "    bgp_prefixes=None,\n",
    "    dry_run=False,\n",
    "):\n",
    "    \"\"\"Instance of the experiment.\"\"\"\n",
    "    measurement_uuid = None\n",
    "    for _ in range(n_cycles):\n",
    "        selector = create_constrained_selector(\n",
    "            name,\n",
    "            compute_budget,\n",
    "            bgp_prefixes=bgp_prefixes,\n",
    "            rl=rl,\n",
    "            measurement_uuid=measurement_uuid,\n",
    "        )\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",
    "            dry_run=dry_run,\n",
    "        )\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.\n",
    "Here we have:\n",
    "\n",
    "* zeph\n",
    "* exploration only \n",
    "* constrained (ark-like)\n",
    "* constrained zeph\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dry run, skip the execution\n",
    "dry_run = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "adaptive_no_bgp_uuids = []\n",
    "adaptive_no_bgp = adaptive_instance(\n",
    "    \"edgenet-1\",\n",
    "    n_cycles,\n",
    "    0.1,\n",
    "    lambda _: global_budget // n_agents,\n",
    "    bgp_prefixes=bgp_prefixes,\n",
    "    bgp_awareness=False,\n",
    "    exploitation_only=False,\n",
    "    dry_run=dry_run,\n",
    ")\n",
    "\n",
    "exploration_no_bgp_uuids = []\n",
    "exploration_no_bgp = adaptive_instance(\n",
    "    \"edgenet-2\",\n",
    "    n_cycles,\n",
    "    1,\n",
    "    lambda _: global_budget // n_agents,\n",
    "    bgp_prefixes=bgp_prefixes,\n",
    "    bgp_awareness=False,\n",
    "    exploitation_only=False,\n",
    "    dry_run=dry_run,\n",
    ")\n",
    "\n",
    "# Ark-like\n",
    "constrained_uuids = []\n",
    "constrained = constrained_instance(\n",
    "    \"edgenet-3\",\n",
    "    n_cycles,\n",
    "    lambda _: global_budget // n_agents,\n",
    "    bgp_prefixes=bgp_prefixes,\n",
    "    dry_run=dry_run,\n",
    ")\n",
    "\n",
    "# Ark-like\n",
    "constrained_adaptive_uuids = []\n",
    "constrained_adaptive = constrained_instance(\n",
    "    \"edgenet-4\",\n",
    "    n_cycles,\n",
    "    lambda _: global_budget // n_agents,\n",
    "    rl=True,\n",
    "    bgp_prefixes=bgp_prefixes,\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": 17,
   "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 (\n",
    "    adaptive_no_bgp_uuid,\n",
    "    exploration_no_bgp_uuid,\n",
    "    constrained_uuid,\n",
    "    constrained_adaptive_uuid,\n",
    ") in zip(\n",
    "    adaptive_no_bgp,\n",
    "    exploration_no_bgp,\n",
    "    constrained,\n",
    "    constrained_adaptive,\n",
    "):\n",
    "\n",
    "    adaptive_no_bgp_uuids.append(adaptive_no_bgp_uuid)\n",
    "    exploration_no_bgp_uuids.append(exploration_no_bgp_uuid)\n",
    "    constrained_uuids.append(constrained_uuid)\n",
    "    constrained_adaptive_uuids.append(constrained_adaptive_uuid)\n",
    "\n",
    "    while True:\n",
    "        check_adaptive_no_bgp = check_measurement_finished(\n",
    "            config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"], adaptive_no_bgp_uuid\n",
    "        )\n",
    "\n",
    "        check_exploration_no_bgp = check_measurement_finished(\n",
    "            config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"], exploration_no_bgp_uuid\n",
    "        )\n",
    "\n",
    "        check_constrained = check_measurement_finished(\n",
    "            config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"], constrained_uuid\n",
    "        )\n",
    "\n",
    "        check_constrained_adaptive = check_measurement_finished(\n",
    "            config[\"iris_url\"], config[\"iris_username\"], config[\"iris_password\"], constrained_adaptive_uuid\n",
    "        )\n",
    "\n",
    "        if (\n",
    "            check_adaptive_no_bgp\n",
    "            and check_exploration_no_bgp\n",
    "            and check_constrained\n",
    "            and check_constrained_adaptive\n",
    "        ):\n",
    "            break\n",
    "        time.sleep(10)\n",
    "\n",
    "\n",
    "with (exp_dir / \"zeph.txt\").open(\"w\") as fd:\n",
    "    for uuid in adaptive_no_bgp_uuids:\n",
    "        fd.write(uuid + \"\\n\")\n",
    "with (exp_dir / \"exploration.txt\").open(\"w\") as fd:\n",
    "    for uuid in exploration_no_bgp_uuids:\n",
    "        fd.write(uuid + \"\\n\")\n",
    "with (exp_dir / \"constrained_exploration.txt\").open(\"w\") as fd:\n",
    "    for uuid in constrained_uuids:\n",
    "        fd.write(uuid + \"\\n\")\n",
    "with (exp_dir / \"constrained_zeph.txt\").open(\"w\") as fd:\n",
    "    for uuid in constrained_adaptive_uuids:\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
}