[DEV-5541] Record inconsistency and performance metrics in precomputer

Author: vmarkovtsev

server/athenian/api/align/spec

@@ -4,6 +4,7 @@
 from typing import Optional
 
 from aiohttp import web
+import aiohttp.web
 import aiomcache
 from asyncpg import IntegrityConstraintViolationError
 import sentry_sdk
@@ -502,3 +503,13 @@ async def set_account_features(request: AthenianWebRequest, id: int, body: dict)
                     ),
                 )
     return model_response(await get_account_features(id, request.sdb))
+
+
+async def get_account_health(
+    request: AthenianWebRequest,
+    id: Optional[int] = None,
+    since: Optional[datetime] = None,
+    until: Optional[datetime] = None,
+) -> aiohttp.web.Response:
+    """Return the account health metrics measured per hour."""
+    raise NotImplementedError

server/athenian/api/controllers/backoffice_controller.py

@@ -818,7 +818,7 @@ async def calculate_for_set_metrics(service, repos, for_sets):
 
 
 @expires_header(short_term_exptime)
-@weight(1)
+@weight(2)
 async def calc_metrics_code_checks(request: AthenianWebRequest, body: dict) -> web.Response:
     """Calculate metrics on continuous integration runs, such as GitHub Actions, Jenkins, Circle, \
     etc."""

server/athenian/api/controllers/metrics_controller.py

@@ -2056,11 +2056,11 @@ async def _call_cached(
     @staticmethod
     def _set_count_metrics(facts: pd.DataFrame, metrics: MinePullRequestMetrics) -> None:
         metrics.count = len(facts)
-        metrics.done_count = facts[PullRequestFacts.f.done].sum()
-        metrics.merged_count = (
-            facts[PullRequestFacts.f.merged].notnull() & ~facts[PullRequestFacts.f.done]
-        ).sum()
-        metrics.open_count = facts[PullRequestFacts.f.closed].isnull().sum()
+        metrics.done_count = int(facts[PullRequestFacts.f.done].sum())
+        metrics.merged_count = int(
+            (facts[PullRequestFacts.f.merged].notnull() & ~facts[PullRequestFacts.f.done]).sum(),
+        )
+        metrics.open_count = int(facts[PullRequestFacts.f.closed].isnull().sum())
 
 
 class ParticipantsMerge:

server/athenian/api/internal/features/entries.py

@@ -897,13 +897,15 @@ async def main_flow():
 
 
 def _set_count_metrics(releases: pd.DataFrame, metrics: MineReleaseMetrics) -> None:
-    metrics.count_by_tag = (releases[ReleaseFacts.f.matched_by].values == ReleaseMatch.tag).sum()
-    metrics.count_by_branch = (
-        releases[ReleaseFacts.f.matched_by].values == ReleaseMatch.branch
-    ).sum()
-    metrics.count_by_event = (
-        releases[ReleaseFacts.f.matched_by].values == ReleaseMatch.event
-    ).sum()
+    metrics.count_by_tag = int(
+        (releases[ReleaseFacts.f.matched_by].values == ReleaseMatch.tag).sum(),
+    )
+    metrics.count_by_branch = int(
+        (releases[ReleaseFacts.f.matched_by].values == ReleaseMatch.branch).sum(),
+    )
+    metrics.count_by_event = int(
+        (releases[ReleaseFacts.f.matched_by].values == ReleaseMatch.event).sum(),
+    )
 
 
 def _empty_mined_releases_df():

server/athenian/api/internal/miners/github/release_mine.py

@@ -18,6 +18,7 @@
     discover_accounts,
     notify_almost_expired_accounts,
     resolve_deployments,
+    store_external_health,
     sync_labels,
 )
 from athenian.api.precompute.context import PrecomputeContext
@@ -29,6 +30,7 @@
     "notify-almost-expired-accounts": notify_almost_expired_accounts.main,
     "discover-accounts": discover_accounts.main,
     "accounts": accounts.main,
+    "store-external-health": store_external_health.main,
 }
 
 
@@ -134,6 +136,14 @@ class Formatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawTextHelpForm
     accounts_parser.add_argument(
         "--timeout", type=int, default=20 * 60, help="Maximum processing time for one account",
     )
+
+    store_external_health_parser = subparsers.add_parser(
+        "store-external-health",
+        help="Persist external account data health metrics: metadata and performance",
+    )
+    store_external_health_parser.add_argument(
+        "--prometheus", type=str, help="Prometheus API endpoint",
+    )
     return parser.parse_args()
 
 

server/athenian/api/precompute/__main__.py

@@ -0,0 +1,113 @@
+import argparse
+from datetime import datetime, timezone
+import logging
+from typing import Any
+
+import aiohttp
+from sqlalchemy import insert, select
+
+from athenian.api import metadata
+from athenian.api.async_utils import gather
+from athenian.api.db import Database
+from athenian.api.models.persistentdata.models import HealthMetric
+from athenian.api.models.state.models import AccountGitHubAccount
+from athenian.api.precompute.context import PrecomputeContext
+
+
+async def _record_performance_metrics(
+    prometheus_endpoint: str,
+    rdb: Database,
+):
+    log = logging.getLogger(f"{metadata.__package__}._record_performance_metrics")
+    for pct in ("50", "95"):
+        inserted = []
+        for attempt in range(3):
+            async with aiohttp.ClientSession() as session:
+                now = datetime.now(timezone.utc)
+                async with session.get(
+                    f"{prometheus_endpoint}/api/v1/query",
+                    params={
+                        "query": (
+                            f"histogram_quantile(0.{pct}, "
+                            "sum(rate(request_latency_seconds_bucket[24h])) by "
+                            "(endpoint, account, le))"
+                        ),
+                    },
+                ) as response:
+                    if not response.ok:
+                        log.error(
+                            "[%d] failed to query Prometheus: %d: %s",
+                            attempt + 1,
+                            response.status,
+                            await response.text(),
+                        )
+                        continue
+                    for obj in (await response.json())["data"]["result"]:
+                        if (value := obj["value"][1]) != "NaN" and (
+                            account := obj["metric"]["account"]
+                        ) != "N/A":
+                            inserted.append(
+                                HealthMetric(
+                                    account_id=int(account),
+                                    name=f'p{pct}/{obj["metric"]["endpoint"]}',
+                                    created_at=now,
+                                    value=float(value),
+                                ).explode(with_primary_keys=True),
+                            )
+                    break
+        if inserted:
+            log.info("inserting %d p%s records", len(inserted), pct)
+            await rdb.execute_many(insert(HealthMetric), inserted)
+
+
+async def _record_inconsistency_metrics(
+    prometheus_endpoint: str,
+    sdb: Database,
+    rdb: Database,
+):
+    log = logging.getLogger(f"{metadata.__package__}._record_inconsistency_metrics")
+    acc_id_map = dict(
+        await sdb.fetch_all(select(AccountGitHubAccount.id, AccountGitHubAccount.account_id)),
+    )
+    inserted = []
+    for attempt in range(3):
+        async with aiohttp.ClientSession() as session:
+            now = datetime.now(timezone.utc)
+            async with session.get(
+                f"{prometheus_endpoint}/api/v1/query",
+                params={
+                    "query": "metadata_github_consistency_nodes_issues",
+                },
+            ) as response:
+                if not response.ok:
+                    log.error(
+                        "[%d] failed to query Prometheus: %d: %s",
+                        attempt + 1,
+                        response.status,
+                        await response.text(),
+                    )
+                    continue
+                for obj in (await response.json())["data"]["result"]:
+                    try:
+                        account = acc_id_map[int(obj["metric"]["acc_id"])]
+                    except KeyError:
+                        continue
+                    inserted.append(
+                        HealthMetric(
+                            account_id=account,
+                            name=f'inconsistency/{obj["metric"]["node_type"]}',
+                            created_at=now,
+                            value=int(obj["value"][1]),
+                        ).explode(with_primary_keys=True),
+                    )
+    if inserted:
+        log.info("inserting %d data inconsistency records", len(inserted))
+        await rdb.execute_many(insert(HealthMetric), inserted)
+
+
+async def main(context: PrecomputeContext, args: argparse.Namespace) -> Any:
+    """Fill missing commit references in the deployed components."""
+    await gather(
+        _record_performance_metrics(args.prometheus, context.rdb),
+        _record_inconsistency_metrics(args.prometheus, context.sdb, context.rdb),
+    )