c46dff09d9
# What this PR does Refactors the PagerDuty migration script to be a bit more generic + adds a migration script to migrate from Splunk OnCall (VictorOps) tldr; ```bash ❯ docker build -t oncall-migrator . [+] Building 0.4s (10/10) FINISHED ❯ docker run --rm \ -e MIGRATING_FROM="pagerduty" \ -e MODE="plan" \ -e ONCALL_API_URL="http://localhost:8080" \ -e ONCALL_API_TOKEN="<ONCALL_API_TOKEN>" \ -e PAGERDUTY_API_TOKEN="<PAGERDUTY_API_TOKEN>" \ oncall-migrator running pagerduty migration script... ❯ docker run --rm \ -e MIGRATING_FROM="splunk" \ -e MODE="plan" \ -e ONCALL_API_URL="http://localhost:8080" \ -e ONCALL_API_TOKEN="<ONCALL_API_TOKEN>" \ -e SPLUNK_API_ID="<SPLUNK_API_ID>" \ -e SPLUNK_API_KEY="<SPLUNK_API_KEY>" \ oncall-migrator migrating from splunk oncall... ``` https://www.loom.com/share/a855062d436a4ef79f030e22528d8c71 ## Checklist - [x] Unit, integration, and e2e (if applicable) tests updated - [x] Documentation added (or `pr:no public docs` PR label added if not required) - [x] Added the relevant release notes label (see labels prefixed w/ `release:`). These labels dictate how your PR will show up in the autogenerated release notes.
83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
import secrets
|
|
from urllib.parse import urljoin
|
|
|
|
import requests
|
|
|
|
|
|
class GrafanaAPIClient:
|
|
def __init__(self, base_url, username, password):
|
|
self.base_url = base_url
|
|
self.username = username
|
|
self.password = password
|
|
|
|
def _api_call(self, method: str, path: str, **kwargs):
|
|
return requests.request(
|
|
method,
|
|
urljoin(self.base_url, path),
|
|
auth=(self.username, self.password),
|
|
**kwargs,
|
|
)
|
|
|
|
def create_user_with_random_password(self, name: str, email: str):
|
|
return self._api_call(
|
|
"POST",
|
|
"/api/admin/users",
|
|
json={
|
|
"name": name,
|
|
"email": email,
|
|
"login": email.split("@")[0],
|
|
"password": secrets.token_urlsafe(15),
|
|
},
|
|
)
|
|
|
|
def get_all_users(self):
|
|
"""
|
|
https://grafana.com/docs/grafana/v10.3/developers/http_api/user/#search-users
|
|
"""
|
|
return self._api_call("GET", "/api/users").json()
|
|
|
|
def idemopotently_create_team_and_add_users(
|
|
self, team_name: str, user_emails: list[str]
|
|
) -> int:
|
|
"""
|
|
Get team by name
|
|
https://grafana.com/docs/grafana/v10.3/developers/http_api/team/#using-the-name-parameter
|
|
|
|
|
|
Create team
|
|
https://grafana.com/docs/grafana/v10.3/developers/http_api/team/#add-team
|
|
|
|
Add team members
|
|
https://grafana.com/docs/grafana/v10.3/developers/http_api/team/#add-team-member
|
|
"""
|
|
existing_team = self._api_call(
|
|
"GET", "/api/teams/search", params={"name": team_name}
|
|
).json()
|
|
|
|
if existing_team["teams"]:
|
|
# team already exists
|
|
team_id = existing_team["teams"][0]["id"]
|
|
else:
|
|
# team doesn't exist create it
|
|
response = self._api_call("POST", "/api/teams", json={"name": team_name})
|
|
|
|
if response.status_code == 200:
|
|
team_id = response.json()["teamId"]
|
|
else:
|
|
raise Exception(f"Failed to fetch/create Grafana team '{team_name}'")
|
|
|
|
grafana_users = self.get_all_users()
|
|
grafana_user_id_to_email_map = {}
|
|
|
|
for user_email in user_emails:
|
|
for grafana_user in grafana_users:
|
|
if grafana_user["email"] == user_email:
|
|
grafana_user_id_to_email_map[grafana_user["id"]] = user_email
|
|
break
|
|
|
|
for user_id in grafana_user_id_to_email_map.keys():
|
|
self._api_call(
|
|
"POST", f"/api/teams/{team_id}/members", json={"userId": user_id}
|
|
)
|
|
|
|
return team_id
|