Changeset View
Changeset View
Standalone View
Standalone View
swh/icinga_plugins/save_code_now.py
# Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021-2022 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import random | |||||
import time | import time | ||||
from typing import Dict, List | from typing import Dict, List, Union | ||||
import requests | import requests | ||||
from .base_check import BaseCheck | from .base_check import BaseCheck | ||||
REPORT_MSG = "Save code now request for origin" | REPORT_MSG = "Save code now request for origin" | ||||
WAITING_STATUSES = ("not yet scheduled", "running", "scheduled") | WAITING_STATUSES = ("not yet scheduled", "running", "scheduled") | ||||
class SaveCodeNowCheck(BaseCheck): | class SaveCodeNowCheck(BaseCheck): | ||||
TYPE = "SAVECODENOW" | TYPE = "SAVECODENOW" | ||||
DEFAULT_WARNING_THRESHOLD = 60 | DEFAULT_WARNING_THRESHOLD = 60 | ||||
DEFAULT_CRITICAL_THRESHOLD = 120 | DEFAULT_CRITICAL_THRESHOLD = 120 | ||||
def __init__(self, obj: Dict, origin: str, visit_type: str) -> None: | def __init__( | ||||
self, obj: Dict, origin: Union[str, List[str]], visit_type: str | |||||
) -> None: | |||||
super().__init__(obj, application="scn") | super().__init__(obj, application="scn") | ||||
if isinstance(origin, list): | |||||
origin = random.choice(origin) | |||||
self.api_url = obj["swh_web_url"].rstrip("/") | self.api_url = obj["swh_web_url"].rstrip("/") | ||||
self.poll_interval = obj["poll_interval"] | self.poll_interval = obj["poll_interval"] | ||||
self.origin = origin | self.origin = origin | ||||
self.visit_type = visit_type | self.visit_type = visit_type | ||||
self.register_prometheus_gauge("duration", "seconds", ["status"]) | self.register_prometheus_gauge("duration", "seconds", ["status"]) | ||||
self.register_prometheus_gauge("status", "") | self.register_prometheus_gauge("status", "") | ||||
@staticmethod | @staticmethod | ||||
def api_url_scn(root_api_url: str, origin: str, visit_type: str) -> str: | def api_url_scn(root_api_url: str, origin: str, visit_type: str) -> str: | ||||
"""Compute the save code now api url for a given origin""" | """Compute the save code now api url for a given origin""" | ||||
return f"{root_api_url}/api/1/origin/save/{visit_type}/url/{origin}/" | return f"{root_api_url}/api/1/origin/save/{visit_type}/url/{origin}/" | ||||
def main(self) -> int: | def main(self) -> int: | ||||
"""Scenario description: | """Scenario description: | ||||
1. Requests a save code now request via the api for origin self.origin with type | 1. Requests a save code now request via the api for the given origin (or | ||||
self.visit_type. | an origin picked at random in the list) with type self.visit_type. | ||||
2. Polling regularly at self.poll_interval seconds the completion status. | 2. Polling regularly at self.poll_interval seconds the completion status. | ||||
3. When either succeeded, failed or threshold exceeded, report approximate time | 3. When either succeeded, failed or threshold exceeded, report approximate time | ||||
of completion. This will warn if thresholds are exceeded. | of completion. This will warn if thresholds are exceeded. | ||||
""" | """ | ||||
start_time: float = time.time() | start_time: float = time.time() | ||||
▲ Show 20 Lines • Show All 74 Lines • Show Last 20 Lines |