PD migrator: ignore 404s on delete + delete ONCALL_DEFAULT_CONTACT_METHOD (#1764)

This commit is contained in:
Vadim Stepanov 2023-04-17 13:12:21 +01:00 committed by GitHub
parent 0c42c2a86d
commit d99bb7b8bc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 14 additions and 25 deletions

View file

@ -79,30 +79,22 @@ docker run --rm \
-e PAGERDUTY_API_TOKEN="<PAGERDUTY_API_TOKEN>" \
-e ONCALL_API_URL="<ONCALL_API_URL>" \
-e ONCALL_API_TOKEN="<ONCALL_API_TOKEN>" \
-e ONCALL_DEFAULT_CONTACT_METHOD="sms" \
-e MODE="migrate" \
pd-oncall-migrator
```
### Migrate unsupported user notification rules
It's possible to specify a default contact method type for user notification rules that cannot be migrated as-is by
changing the `ONCALL_DEFAULT_CONTACT_METHOD` env variable.
Options are: `email`, `sms`, `phone_call`, `slack`, `telegram`, `mobile_app` (default is `email`).
### Migrate unsupported integration types
It's possible to migrate unsupported integration types to [Grafana OnCall incoming webhooks](https://grafana.com/docs/oncall/latest/integrations/available-integrations/configure-webhook/).
by changing UNSUPPORTED_INTEGRATION_TO_WEBHOOKS env variable:
To enable this feature, set env variable `UNSUPPORTED_INTEGRATION_TO_WEBHOOKS` to `true`:
```shell
docker run --rm \
-e PAGERDUTY_API_TOKEN="<PAGERDUTY_API_TOKEN>" \
-e ONCALL_API_URL="<ONCALL_API_URL>" \
-e ONCALL_API_TOKEN="<ONCALL_API_TOKEN>" \
-e ONCALL_DEFAULT_CONTACT_METHOD="sms" \
-e MODE="migrate" \
-e UNSUPPORTED_INTEGRATION_TO_WEBHOOKS="true" \
-e MODE="migrate" \
pd-oncall-migrator
```

View file

@ -98,7 +98,7 @@ def main() -> None:
rulesets = None
if EXPERIMENTAL_MIGRATE_EVENT_RULES:
print("▶ Fetching event rules (rulesets) ...")
print("▶ Fetching event rules (global rulesets)...")
rulesets = session.list_all("rulesets")
for ruleset in rulesets:
rules = session.list_all(f"rulesets/{ruleset['id']}/rules")
@ -173,7 +173,7 @@ def main() -> None:
print(TAB + format_integration(integration))
if rulesets is not None:
print("▶ Migrating event rules (rulesets) ...")
print("▶ Migrating event rules (global rulesets)...")
for ruleset in rulesets:
if not ruleset["flawed_escalation_policies"]:
migrate_ruleset(ruleset, escalation_policies, services)

View file

@ -14,9 +14,6 @@ ONCALL_API_URL = urljoin(
)
ONCALL_DELAY_OPTIONS = [1, 5, 15, 30, 60]
ONCALL_DEFAULT_CONTACT_METHOD = "notify_by_" + os.getenv(
"ONCALL_DEFAULT_CONTACT_METHOD", default="email"
)
PAGERDUTY_TO_ONCALL_CONTACT_METHOD_MAP = {
"sms_contact_method": "notify_by_sms",
"phone_contact_method": "notify_by_phone_call",

View file

@ -73,7 +73,12 @@ def create(path: str, payload: dict) -> dict:
def delete(path: str) -> None:
api_call("delete", path)
try:
api_call("delete", path)
except requests.exceptions.HTTPError as e:
# ignore 404s on delete so deleting resources manually while running the script doesn't break it
if e.response.status_code != 404:
raise
def update(path: str, payload: dict) -> dict:

View file

@ -76,7 +76,7 @@ def format_integration(integration: dict) -> str:
else:
# check if integration not supported, but UNSUPPORTED_INTEGRATION_TO_WEBHOOKS set
if integration.get("converted_to_webhook", False):
result = "{} {} Webhook integration will be created, Grafana OnCall not support this type directly ".format(
result = "{} {} cannot find appropriate Grafana OnCall integration type, integration will be migrated with type 'webhook'".format(
WARNING_SIGN, result
)
else:
@ -187,7 +187,7 @@ def format_ruleset(ruleset: dict) -> str:
def ruleset_report(rulesets: list[dict]) -> str:
result = "Event rules (rulesets) report:"
result = "Event rules (global rulesets) report:"
for ruleset in sorted(
rulesets,

View file

@ -1,10 +1,7 @@
import copy
from migrator import oncall_api_client
from migrator.config import (
ONCALL_DEFAULT_CONTACT_METHOD,
PAGERDUTY_TO_ONCALL_CONTACT_METHOD_MAP,
)
from migrator.config import PAGERDUTY_TO_ONCALL_CONTACT_METHOD_MAP
from migrator.utils import remove_duplicates, transform_wait_delay
@ -76,10 +73,8 @@ def transform_notification_rule(
notification_rule: dict, delay: int, user_id: str
) -> list[dict]:
contact_method_type = notification_rule["contact_method"]["type"]
oncall_type = PAGERDUTY_TO_ONCALL_CONTACT_METHOD_MAP[contact_method_type]
oncall_type = PAGERDUTY_TO_ONCALL_CONTACT_METHOD_MAP.get(
contact_method_type, ONCALL_DEFAULT_CONTACT_METHOD
)
notify_rule = {"user_id": user_id, "type": oncall_type, "important": False}
if not delay: