opsgenie-cleanup/OpsGenieCleanUp.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## OpsGenie Clean Up Utils\n",
    "\n",
    "Removing users from OpsGenie application involves many step and can be painfully slow. Recently I had to remove few hundreds of them. Let me share my solution.\n",
    "\n",
    "GenieClient class and the notes **(scroll below)** can help to clean up Users, Teams, Integrations from OpsGenie.\n",
    "\n",
    "### Scenarios and features\n",
    " - Dump user list\n",
    " - Scenario 1: Clean up users who is in CSV file\n",
    " - Scenario 2: Clean up users who is NOT in CSV file\n",
    " - Scenario 3: Clean up user one by one\n",
    " - Clean up disabled integrations\n",
    " - Clean up vacant teams\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "from urllib.parse import urlencode\n",
    "import csv\n",
    "\n",
    "\n",
    "class GenieClient():\n",
    "    def __init__(self, key):\n",
    "        self.url = 'https://api.opsgenie.com/v2/%s'\n",
    "        self.headers = {\n",
    "            'Content-Type': 'application/json',\n",
    "            'Authorization': 'GenieKey ' + key,\n",
    "            'Cache-control': 'no-cache'\n",
    "        }\n",
    "\n",
    "    def request(self, path, **kwargs):\n",
    "        \"\"\"\n",
    "        Sends an HTTP GET request.\n",
    "        Returns a yielded iteration. \n",
    "        example: next(request('users', sort='fullName', query='role:admin AND locale:en_US'))\n",
    "        \"\"\"\n",
    "        url = self.url % path + ('?' + urlencode(kwargs) if len(kwargs) else '')\n",
    "        last = False\n",
    "        while not last:\n",
    "            r = requests.get(url, headers=self.headers).json()\n",
    "            yield r\n",
    "            if 'paging' in r and 'next' in r['paging']:\n",
    "                url = r['paging']['next']\n",
    "            else:\n",
    "                last = True\n",
    "\n",
    "    def delete(self, path, **kwargs):\n",
    "        \"\"\"\n",
    "        Sends an HTTP DELETE request.\n",
    "        example: delete('users/' + username)\n",
    "        \"\"\"\n",
    "        url = self.url % path + ('?' + urlencode(kwargs) if len(kwargs) else '')\n",
    "        return requests.delete(url, headers=self.headers).json()\n",
    "\n",
    "    def patch(self, path, payload, **kwargs):\n",
    "        \"\"\"\n",
    "        Sends an HTTP PATCH request.\n",
    "        example: patch('escalations/' + username, {'rules': rules})\n",
    "        \"\"\"\n",
    "        url = self.url % path + ('?' + urlencode(kwargs) if len(kwargs) else '')\n",
    "        return requests.patch(url, json=payload, headers=self.headers).json()\n",
    "\n",
    "    def dump_users(self, filename, query=''):\n",
    "        \"\"\"\n",
    "        Dumps user list to csv file\n",
    "        example:\n",
    "          filename='users.csv'\n",
    "          query='role:user AND verified:1'\n",
    "        \"\"\"\n",
    "        with open(filename, 'w', newline='') as csvfile:\n",
    "            csvw = csv.writer(csvfile, delimiter=',', quotechar='|', quoting=csv.QUOTE_MINIMAL)\n",
    "            csvw.writerow(('name', 'email', 'role', 'verified'))\n",
    "            for r in self.request('users', query=query):\n",
    "                print('.')\n",
    "                if 'data' in r:\n",
    "                    for u in r['data']:\n",
    "                        csvw.writerow((u['fullName'], u['username'], u['role']['name'], u['verified']))\n",
    "\n",
    "    def for_existing_user(func):\n",
    "        \"\"\"\n",
    "        Decorator method for exiting user\n",
    "        \"\"\"\n",
    "        def inner(self, username):\n",
    "            r = requests.get(self.url % 'users/' + username, headers=self.headers)\n",
    "            if r.status_code == requests.codes.ok and 'data' in r.json():\n",
    "                return func(self, username)\n",
    "            else:\n",
    "                print('[Error] User account {} not found'.format(username))\n",
    "        return inner\n",
    "\n",
    "    def remove_user_from_escalations(self, username):\n",
    "        \"\"\"\n",
    "        Removes given user from its escalations. \n",
    "        Newly vacant escalations and routing-rules will be deleted\n",
    "        \"\"\"\n",
    "        for user_esc in next(self.request(f'users/{username}/escalations'))['data']:\n",
    "            escalation = next(self.request('escalations/' + user_esc['id']))['data']\n",
    "            esc_rules = list(filter(lambda r: not(r['recipient']['type'] == 'user' and username == r['recipient']['username']), escalation['rules']))\n",
    "            if len(esc_rules) == 0:\n",
    "                if 'ownerTeam' in escalation:\n",
    "                    for rule in next(self.request('/teams/' + escalation['ownerTeam']['id'] + '/routing-rules'))['data']:\n",
    "                        if ('notify' in rule and rule['notify']['type'] == 'escalation' and rule['notify']['id'] == user_esc['id']):\n",
    "                            try:\n",
    "                                r = self.delete(\n",
    "                                    'teams/' + escalation['ownerTeam']['id'] + '/routing-rules/' + rule['id'])\n",
    "                                if 'result' not in r or r['result'] != 'Deleted':\n",
    "                                    raise Exception(r['message'])\n",
    "                                print('[ OK  ] Remove vacant routing-rules {} for {}'.format(rule['notify']['name'], username))\n",
    "                            except Exception as e:\n",
    "                                print('[Error] Remove vacant routing-rules {} for {}'.format(rule['notify']['name'], username))\n",
    "                                print(\"\\tException \" + str(e))\n",
    "                try:\n",
    "                    r = self.delete('escalations/' + user_esc['id'])\n",
    "                    if 'result' not in r or r['result'] != 'Deleted':\n",
    "                        raise Exception(r['message'])\n",
    "                    print('[ OK  ] Remove vacant escalation {} for {}'.format(user_esc['name'], username))\n",
    "                except Exception as e:\n",
    "                    print('[Error] Remove vacant escalation {} for {}'.format(user_esc['name'], username))\n",
    "                    print(\"\\tException \" + str(e))\n",
    "            else:\n",
    "                try:\n",
    "                    r = self.patch('escalations/' + user_esc['id'], {'rules': esc_rules})\n",
    "                    if 'result' not in r or r['result'] != 'Updated':\n",
    "                        raise Exception(r['message'])\n",
    "                    print('[ OK  ] Remove {} from escalation {}'.format(username, user_esc['name']))\n",
    "                except Exception as e:\n",
    "                    print('[Error] Remove {} from escalation {}'.format(username, user_esc['name']))\n",
    "                    print(\"\\tException \" + str(e))\n",
    "\n",
    "    def remove_user_from_schedules(self, username):\n",
    "        \"\"\"\n",
    "        Removes given user from its schedules. \n",
    "        Newly vacant rotations will be deleted.\n",
    "        Vacant schedules can be removed by delete_empty_teams() method.\n",
    "        \"\"\"\n",
    "        for schedule in next(self.request(f'users/{username}/schedules'))['data']:\n",
    "            rotations = next(self.request('schedules/' + schedule['id'] + '/rotations'))['data']\n",
    "            for kr, rotation in enumerate(rotations):\n",
    "                participants = list(filter(lambda p: not('username' in p and p['username'] == username), rotation['participants']))\n",
    "                if len(participants) == 0:\n",
    "                    try:\n",
    "                        r = self.delete('schedules/{}/rotations/{}'.format(schedule['id'], rotations[kr]['id']))\n",
    "                        if 'result' not in r or r['result'] != 'Deleted':\n",
    "                            raise Exception(r['message'])\n",
    "                        print('[ OK  ] Remove {} and the vacant rotation {}/{}'.format(username, schedule['name'], rotation['name']))\n",
    "                    except Exception as e:\n",
    "                        print('[Error] Remove {} and the vacant rotation {}/{}'.format(username, schedule['name'], rotation['name']))\n",
    "                        print(\"\\tException \" + str(e))\n",
    "                else:\n",
    "                    try:\n",
    "                        r = self.patch('schedules/{}/rotations/{}'.format(schedule['id'], rotations[kr]['id']),\n",
    "                                       {'participants': participants})\n",
    "                        if 'result' not in r or r['result'] != 'Updated':\n",
    "                            raise Exception(r['message'])\n",
    "                        print('[ OK  ] Remove {} from rotation {}/{}'.format(username, schedule['name'], rotation['name']))\n",
    "                    except Exception as e:\n",
    "                        print('[Error] Remove {} from rotation {}/{}'.format(username, schedule['name'], rotation['name']))\n",
    "                        print(\"\\tException \" + str(e))\n",
    "\n",
    "    def remove_user_from_teams(self, username):\n",
    "        \"\"\"\n",
    "        Removes given user from its schedules. \n",
    "        Newly vacant rotations will be deleted.\n",
    "        Vacant schedules can be removed by delete_empty_teams() method.\n",
    "        \"\"\"\n",
    "        for team in next(self.request(f'users/{username}/teams'))['data']:\n",
    "            try:\n",
    "                r = self.delete('teams/{}/members/{}'.format(team['id'], username))\n",
    "                if 'result' not in r or r['result'] != 'Removed':\n",
    "                    raise Exception(r['message'])\n",
    "                print('[ OK  ] Remove {} from team {}'.format(username, team['name']))\n",
    "            except Exception as e:\n",
    "                print('[Error] Remove {} from team {}'.format(username, team['name']))\n",
    "                print(\"\\tException \" + str(e))\n",
    "\n",
    "    def delete_user(self, username):\n",
    "        \"\"\"\n",
    "        Deletes user from the app. \n",
    "        \"\"\"\n",
    "        try:\n",
    "            r = self.delete('users/' + username)\n",
    "            if 'result' not in r or r['result'] != 'Deleted':\n",
    "                raise Exception(r['message'])\n",
    "            print('[ OK  ] Remove user account {}'.format(username))\n",
    "        except Exception as e:\n",
    "            print('[Error] Remove user account {}'.format(username))\n",
    "            print(\"\\tException \" + str(e))\n",
    "\n",
    "    @for_existing_user\n",
    "    def deep_user_remove(self, username):\n",
    "        \"\"\"\n",
    "        All in one user clean up. Removes given user from escalations, schedules, teams and from the application. \n",
    "        \"\"\"        \n",
    "        self.remove_user_from_escalations(username)\n",
    "        self.remove_user_from_schedules(username)\n",
    "        self.remove_user_from_teams(username)\n",
    "        self.delete_user(username)\n",
    "        \n",
    "    def delete_empty_teams(self):\n",
    "        \"\"\"\n",
    "        Deletes all vacant teams (teams without members).\n",
    "        \"\"\"        \n",
    "        for teams in g.request('teams'):\n",
    "            for team in teams['data']:\n",
    "                info = next(g.request('teams/{}'.format(team['id'])))['data']\n",
    "                if not 'members' in info:\n",
    "                    try:\n",
    "                        r = self.delete('teams/{}'.format(team['id']))\n",
    "                        if 'result' not in r or r['result'] != 'Deleted':\n",
    "                            raise Exception(r['message'])\n",
    "                        print('[ OK  ] Deleted empty team {}'.format(team['name']))\n",
    "                    except Exception as e:\n",
    "                        print('[Error] Deleted empty team {}'.format(team['name']))\n",
    "                        print(\"\\tException \" + str(e))\n",
    "                else:\n",
    "                    print('[Skipp] Team {} with {} member(s)'.format(team['name'], len(info['members'])))\n",
    "        \n",
    "    def delete_disabled_integrations(self):\n",
    "        \"\"\"\n",
    "        Removes all disabled integrations. \n",
    "        \"\"\"        \n",
    "        for integrations in g.request('integrations'):\n",
    "            for integration in integrations['data']:\n",
    "                if not integration['enabled']:\n",
    "                    try:\n",
    "                        r = self.delete('integrations/{}'.format(integration['id']))\n",
    "                        if 'result' not in r or r['result'] != 'Deleted':\n",
    "                            raise Exception(r['message'])\n",
    "                        print('[ OK  ] Deleted disabled integration {}'.format(integration['name']))\n",
    "                    except Exception as e:\n",
    "                        print('[Error] Deleted disabled integration {}'.format(integration['name']))\n",
    "                        print(\"\\tException \" + str(e))\n",
    "                else:\n",
    "                    print('[Skipp] Enabled integration {}'.format(integration['name']))\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Instantiate the GenieClient class\n",
    "As an imput the OpsGenie API Key need to be provided. You can find your API Key at https://app.opsgenie.com/integration page. All account has a *Default API (API)* out of the box. In case you would not like to mess up with the Default API, you can create a new API (Rest API HTTPS over JSON) and make sure it is not limited to read only access.  \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = GenieClient(input('OpsGenie API Key: '))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Backup, backup, backup\n",
    "\n",
    "I highly recommend to backup all data, before the operation. Either you may remove members unintentionally, either they will complain later, when they realise can't login anymore.\n",
    "\n",
    "The most comprehensive available backup solution is here: https://github.com/opsgenie/opsgenie-configuration-backup\n",
    "\n",
    "\n",
    "## Dump user list\n",
    "Although you surly made a backup in the mentioned way you also can dump the current user-list into a CSV file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read-only operation to dump users into a CSV file\n",
    "g.dump_users('users-before-clanup.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Scenario 1: Clean up users who are in CSV file\n",
    "In this scenario we only clean up users who are listed in CSV file.\n",
    "The content of CSV is coma `','` delimitered file with at least one column called `'email'`. Example:\n",
    "```\n",
    "email,name\n",
    "john,doe@example.com,John Doe\n",
    "oliver,nadj@example.com,Oliver Nadj\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# I used panda library for read CSV and manipulating data.\n",
    "import pandas as pd\n",
    "to_remove = pd.read_csv('users-to-remove.csv')['email'].tolist()\n",
    "# double check first 20 users\n",
    "to_remove[0:20]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Take a deep breath and remove them\n",
    "for u in to_remove:\n",
    "    g.deep_user_remove(u)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Scenario 2: Clean up users who are NOT in CSV file\n",
    "In this scenario is a bit more clomplicated than the previous one. \n",
    "we only lean up users who are NOT listed in CSV file and protect who are listed.\n",
    "The content of CSV is coma `','` delimitered file with at least one column called `'email'`. Example:\n",
    "```\n",
    "email,name\n",
    "john,doe@example.com,John Doe\n",
    "oliver,nadj@example.com,Oliver Nadj\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# I used panda library for read CSV and manipulating data.\n",
    "import pandas as pd\n",
    "\n",
    "# Let's dump all users 1st\n",
    "g.dump_users('all-users.csv')\n",
    "allusers = pd.read_csv('all-users.csv')['email'].tolist()\n",
    "allusers[0:6]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Load the list of protected users (users who will remain, but anyone else will be removed)\n",
    "to_protect = pd.read_csv('./protected-users.csv')['email'].tolist()\n",
    "to_protect[0:3]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Let's create a list of the users who will be removed\n",
    "with open('users-to-remove.csv', 'w', newline='') as csvfile:\n",
    "    csvw = csv.writer(csvfile, delimiter=',', quotechar='|', quoting=csv.QUOTE_MINIMAL)\n",
    "    csvw.writerow(['email'])\n",
    "    for u in allusers:\n",
    "        if not(u in to_protect):\n",
    "            csvw.writerow([u])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Now we have a list of the users we wanted to clean up\n",
    "to_remove = pd.read_csv('users-to-remove.csv')['email'].tolist()\n",
    "to_remove[0:20]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Take a deep breath and remove them\n",
    "for u in to_remove:\n",
    "    g.deep_user_remove(u)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Scenario 3: Clean up user one by one\n",
    "In case you would like to warm up, you can try with simple requests and clean user one by one"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get user info\n",
    "email = 'john.doe@example.com'\n",
    "next(g.request('users/' + email)) # next() is needed to trigger yielded iteration."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get scheduels of the user\n",
    "next(g.request('users/' + email + '/schedules'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get temas of the user\n",
    "next(g.request('users/' + email + '/teams'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Removes given user from escalations, schedules, teams and deletes the user from application\n",
    "g.remove_user_from_escalations(email)\n",
    "g.remove_user_from_schedules(email)\n",
    "g.remove_user_from_teams(email)\n",
    "g.delete_user(email)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Clean up disabled integrations\n",
    "This featue may not as handy as a previous one, but can be usefull if you have auto generated integrations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Deletes all disabled integrations\n",
    "g.delete_disabled_integrations()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Clean up vacant teams\n",
    "It removes all vacant teams (teams with no members in it) and it's escalations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Removes vacant teams.\n",
    "g.delete_empty_teams()"
   ]
  }
 ],
 "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.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}