[DEV-5540] Implement /account/health logic

Author: vmarkovtsev

server/athenian/api/application.py

@@ -894,6 +894,7 @@ def raise_graceful_exit():
         loop = asyncio.get_event_loop()
         loop.remove_signal_handler(signal.SIGINT)
         loop.remove_signal_handler(signal.SIGTERM)
+        await self._shutdown()
         loop.add_signal_handler(signal.SIGTERM, raise_graceful_exit)
         os.kill(os.getpid(), signal.SIGTERM)
 

server/athenian/api/controllers/backoffice_controller.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 import logging
 from sqlite3 import IntegrityError, OperationalError
 from typing import Optional
@@ -7,6 +7,7 @@
 import aiohttp.web
 import aiomcache
 from asyncpg import IntegrityConstraintViolationError
+from dateutil.rrule import HOURLY, rrule
 import sentry_sdk
 from sqlalchemy import delete, insert, select, union_all, update
 
@@ -19,6 +20,7 @@
     get_user_account_status_from_request,
 )
 from athenian.api.internal.account_feature import get_account_features
+from athenian.api.internal.data_health_metrics import measure_accounts_health
 from athenian.api.internal.jira import fetch_jira_installation_progress, get_jira_id
 from athenian.api.internal.logical_repos import coerce_logical_repos
 from athenian.api.internal.miners.github.branches import BranchMiner
@@ -37,6 +39,7 @@
     UserAccount,
 )
 from athenian.api.models.web import (
+    AccountsHealth,
     DatabaseConflict,
     ForbiddenError,
     InvalidRequestError,
@@ -508,8 +511,40 @@ async def set_account_features(request: AthenianWebRequest, id: int, body: dict)
 async def get_account_health(
     request: AthenianWebRequest,
     id: Optional[int] = None,
-    since: Optional[datetime] = None,
-    until: Optional[datetime] = None,
+    since: Optional[str] = None,
+    until: Optional[str] = None,
 ) -> aiohttp.web.Response:
     """Return the account health metrics measured per hour."""
-    raise NotImplementedError
+    try:
+        if since is not None:
+            since = deserialize_datetime(since)
+        else:
+            since = datetime.now(timezone.utc) - timedelta(hours=1)
+        if until is not None:
+            until = deserialize_datetime(until)
+        else:
+            until = datetime.now(timezone.utc)
+    except ValueError as e:
+        e.path = "since" if isinstance(since, str) else "until"
+        raise ResponseError(InvalidRequestError.from_validation_error(e))
+    if id is not None:
+        ids = [id]
+    else:
+        ids = (
+            await read_sql_query(
+                select(Account.id)
+                .where(Account.expires_at > datetime.now(timezone.utc))
+                .order_by(Account.id),
+                request.sdb,
+                [Account.id],
+            )
+        )[Account.id.name].values
+    since = (since + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
+    until = until.replace(minute=0, second=0, microsecond=0)
+    datetimes = list(rrule(HOURLY, dtstart=since, until=until))
+    return model_response(
+        AccountsHealth(
+            datetimes=datetimes,
+            accounts=await measure_accounts_health(ids, datetimes, request.rdb),
+        ),
+    )

server/athenian/api/internal/data_health_metrics.py

@@ -1,8 +1,9 @@
 import dataclasses
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
+from typing import Collection
 
-from sqlalchemy import insert
+from sqlalchemy import insert, select
 
 from athenian.api.db import DatabaseLike
 from athenian.api.internal.features.entries import MinePullRequestMetrics
@@ -11,6 +12,7 @@
 from athenian.api.internal.miners.github.release_load import MineReleaseMetrics
 from athenian.api.internal.reposet import RepositorySetMetrics
 from athenian.api.models.persistentdata.models import HealthMetric
+from athenian.api.models.web import AccountHealth
 
 
 @dataclass(frozen=True, slots=True)
@@ -47,3 +49,110 @@ async def persist(self, account: int, rdb: DatabaseLike) -> None:
                 m.created_at = now
                 values.append(m.explode(with_primary_keys=True))
         await rdb.execute_many(insert(HealthMetric), values)
+
+
+async def measure_accounts_health(
+    ids: Collection[int],
+    time_points: list[datetime],
+    rdb: DatabaseLike,
+):
+    """Collect all the supported health metrics for each account in `ids`."""
+    rows = await rdb.fetch_all(
+        select(HealthMetric)
+        .where(
+            HealthMetric.account_id.in_(ids),
+            HealthMetric.created_at >= time_points[0],
+            HealthMetric.created_at < time_points[-1] + timedelta(hours=1),
+        )
+        .order_by(HealthMetric.created_at),
+    )
+    ltp = len(time_points)
+    result = {
+        acc: AccountHealth(
+            broken_branches=[0] * ltp,
+            broken_dags=[0] * ltp,
+            deployments=[0] * ltp,
+            empty_releases=[0] * ltp,
+            endpoint_p50={},
+            endpoint_p95={},
+            event_releases=[0] * ltp,
+            inconsistent_nodes={},
+            pending_fetch_branches=[0] * ltp,
+            pending_fetch_prs=[0] * ltp,
+            prs_count=[0] * ltp,
+            released_prs_ratio=[0] * ltp,
+            reposet_problems=[0] * ltp,
+            repositories_count=[0] * ltp,
+            unresolved_deployments=[0] * ltp,
+            unresolved_releases=[0] * ltp,
+        )
+        for acc in ids
+    }
+    start_time = time_points[0]
+    one_hour = timedelta(hours=1)
+    acc_col = HealthMetric.account_id.name
+    created_col = HealthMetric.created_at.name
+    name_col = HealthMetric.name.name
+    value_col = HealthMetric.value.name
+    for row in rows:
+        model = result[row[acc_col]]
+        dt = row[created_col]
+        pos = (dt - start_time) // one_hour
+        val = row[value_col]
+        match row[name_col]:
+            case "branches_count":
+                continue
+            case ["branches_empty_count", "branches_no_default"]:
+                model.broken_branches[pos] += val
+            case ["commits_pristine", "commits_corrupted", "commits_orphaned"]:
+                model.broken_dags[pos] += val
+            case "deployments_count":
+                model.deployments[pos] = val
+            case "deployments_unresolved":
+                model.unresolved_deployments[pos] = val
+            case "releases_unresolved":
+                model.unresolved_releases[pos] = val
+            case "releases_by_event":
+                model.event_releases[pos] = val
+            case "releases_empty":
+                model.empty_releases[pos] = val
+            case "reposet_problems":
+                model.empty_releases[pos] = val
+            case "reposet_length":
+                model.repositories_count[pos] = val
+            case "prs_count":
+                model.prs_count[pos] = val
+            case "prs_done_count":
+                model.released_prs_ratio[pos] = val
+            case p50 if p50.startswith("p50/"):
+                endpoint = p50[4:]
+                try:
+                    model.endpoint_p50[endpoint][pos] = val
+                except KeyError:
+                    vals = [None] * ltp
+                    vals[pos] = val
+                    model.endpoint_p50[endpoint] = vals
+            case p95 if p95.startswith("p95/"):
+                endpoint = p95[4:]
+                try:
+                    model.endpoint_p95[endpoint][pos] = val
+                except KeyError:
+                    vals = [None] * ltp
+                    vals[pos] = val
+                    model.endpoint_p95[endpoint] = vals
+            case node if node.startswith("inconsistency/"):
+                node = node[14:]
+                try:
+                    model.inconsistent_nodes[node][pos] = val
+                except KeyError:
+                    vals = [0] * ltp
+                    vals[pos] = val
+                    model.inconsistent_nodes[node] = vals
+    for model in result.values():
+        released_prs_ratio = model.released_prs_ratio
+        for i, (done, count) in enumerate(zip(released_prs_ratio, model.prs_count)):
+            if count == 0:
+                released_prs_ratio[i] = 0
+            else:
+                released_prs_ratio[i] = done / count
+    return result

server/athenian/api/internal/datetime_utils.py

@@ -21,7 +21,7 @@ def coarsen_time_interval(time_from: datetime, time_to: datetime) -> Tuple[date,
 def split_to_time_intervals(
     date_from: date,
     date_to: date,
-    granularities: Union[str, List[str]],
+    granularities: str | list[str],
     tzoffset: Optional[int],
 ) -> Tuple[Union[List[datetime], List[List[datetime]]], timedelta]:
     """Produce time interval boundaries from the min and the max dates and the interval lengths \

server/athenian/api/models/web/__init__.py

@@ -1,10 +1,12 @@
 from athenian.api.models.web.accepted_invitation import AcceptedInvitation
 from athenian.api.models.web.account import Account
+from athenian.api.models.web.account_health import AccountHealth
 from athenian.api.models.web.account_status import AccountStatus
 from athenian.api.models.web.account_user_change_request import (
     AccountUserChangeRequest,
     UserChangeStatus,
 )
+from athenian.api.models.web.accounts_health import AccountsHealth
 from athenian.api.models.web.base_model_ import AllOf, Enum, Model, Slots
 from athenian.api.models.web.calculated_code_check_histogram import CalculatedCodeCheckHistogram
 from athenian.api.models.web.calculated_code_check_metrics import CalculatedCodeCheckMetrics

server/athenian/api/models/web/account_health.py

@@ -0,0 +1,22 @@
+from athenian.api.models.web.base_model_ import Model
+
+
+class AccountHealth(Model):
+    """Map from health metric names to metric values."""
+
+    broken_branches: list[int]
+    broken_dags: list[int]
+    deployments: list[int]
+    empty_releases: list[int]
+    endpoint_p50: dict[str, list[float]]
+    endpoint_p95: dict[str, list[float]]
+    event_releases: list[int]
+    inconsistent_nodes: dict[str, list[int]]
+    pending_fetch_branches: list[int]
+    pending_fetch_prs: list[int]
+    prs_count: list[int]
+    released_prs_ratio: list[float]
+    reposet_problems: list[int]
+    repositories_count: list[int]
+    unresolved_deployments: list[int]
+    unresolved_releases: list[int]

server/athenian/api/models/web/accounts_health.py

@@ -0,0 +1,12 @@
+from datetime import datetime
+
+from athenian.api.models.web.account_health import AccountHealth
+from athenian.api.models.web.base_model_ import Model
+
+
+class AccountsHealth(Model):
+    """The metric measurement timestamps and the measured metric values for each requested \
+    account."""
+
+    accounts: dict[int, AccountHealth]
+    datetimes: list[datetime]

server/athenian/api/models/web/calculated_code_check_metrics_item.py

@@ -1,5 +1,3 @@
-from typing import List
-
 from athenian.api.models.web.base_model_ import AllOf, Model
 from athenian.api.models.web.calculated_linear_metric_values import CalculatedLinearMetricValues
 from athenian.api.models.web.for_set_code_checks import _CalculatedCodeCheckCommon
@@ -10,7 +8,7 @@ class _CalculatedCodeCheckMetricsItem(Model, GranularityMixin, sealed=False):
     """Series of calculated metrics for a specific set of repositories and commit authors."""
 
     granularity: str
-    values: List[CalculatedLinearMetricValues]
+    values: list[CalculatedLinearMetricValues]
 
 
 CalculatedCodeCheckMetricsItem = AllOf(

server/athenian/api/models/web/calculated_linear_metric_values.py

@@ -1,4 +1,4 @@
-import datetime
+from datetime import date
 from typing import Optional
 
 from athenian.api.models.web.base_model_ import Model
@@ -7,7 +7,7 @@
 class CalculatedLinearMetricValues(Model):
     """Calculated metrics: date, values, confidences."""
 
-    date: datetime.date
+    date: date
     values: list[object]
     confidence_scores: Optional[list[int]]
     confidence_mins: Optional[list[object]]

server/athenian/api/models/web/calculated_pull_request_metrics_item.py

@@ -1,5 +1,3 @@
-from typing import List
-
 from athenian.api.models.web.base_model_ import Model
 from athenian.api.models.web.calculated_linear_metric_values import CalculatedLinearMetricValues
 from athenian.api.models.web.for_set_pull_requests import ForSetPullRequests
@@ -11,4 +9,4 @@ class CalculatedPullRequestMetricsItem(Model, GranularityMixin):
 
     for_: (ForSetPullRequests, "for")
     granularity: str
-    values: List[CalculatedLinearMetricValues]
+    values: list[CalculatedLinearMetricValues]