[DEV-5541] Measure the internal data health metrics

Author: vmarkovtsev

server/athenian/api/internal/data_health_metrics.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+
+from athenian.api.internal.features.entries import MinePullRequestMetrics
+from athenian.api.internal.miners.github.branches import BranchMinerMetrics
+from athenian.api.internal.miners.github.deployment import MineDeploymentsMetrics
+from athenian.api.internal.miners.github.release_load import MineReleaseMetrics
+from athenian.api.internal.reposet import RepositorySetMetrics
+
+
+@dataclass(frozen=True, slots=True)
+class DataHealthMetrics:
+    """Collection of data error statistics to report."""
+
+    branches: BranchMinerMetrics
+    deployments: MineDeploymentsMetrics
+    prs: MinePullRequestMetrics
+    releases: MineReleaseMetrics
+    reposet: RepositorySetMetrics
+
+    @classmethod
+    def empty(cls) -> "DataHealthMetrics":
+        """Initialize a new DataHealthMetrics instance filled with zeros."""
+        return DataHealthMetrics(**{k: v.empty() for k, v in cls.__annotations__.items()})

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

@@ -3,6 +3,7 @@
 import asyncio
 from collections.abc import Iterator
 import dataclasses
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from functools import partial, reduce
 from itertools import chain
@@ -1596,7 +1597,7 @@ def make_calculator(
     )
 
 
-@dataclasses.dataclass(frozen=True, slots=True)
+@dataclass(frozen=True, slots=True)
 class MetricsLineRequest:
     """Common base for multiple metrics request classes."""
 
@@ -1622,7 +1623,7 @@ def all_jira_filters(self) -> Iterator[JIRAFilter]:
         return (td.jira_filter for td in self.teams)
 
 
-@dataclasses.dataclass(frozen=True, slots=True)
+@dataclass(frozen=True, slots=True)
 class TeamSpecificFilters:
     """Filters that are different for each team."""
 
@@ -1645,6 +1646,21 @@ def __sentry_repr__(self):
         return str(dikt)
 
 
+@dataclass(slots=True)
+class MinePullRequestMetrics:
+    """Various statistics about mined pull requests."""
+
+    count: int
+    done_count: int
+    merged_count: int
+    open_count: int
+
+    @classmethod
+    def empty(cls) -> MinePullRequestMetrics:
+        """Initialize a new MinePullRequestMetrics instance filled with zeros."""
+        return MinePullRequestMetrics(0, 0, 0, 0)
+
+
 class PRFactsCalculator:
     """Calculator for Pull Requests facts."""
 
@@ -1696,6 +1712,7 @@ async def __call__(
         with_jira: JIRAEntityToFetch | int,
         branches: Optional[pd.DataFrame] = None,
         default_branches: Optional[dict[str, str]] = None,
+        metrics: Optional[MinePullRequestMetrics] = None,
     ) -> pd.DataFrame:
         """
         Calculate facts about pull request on GitHub.
@@ -1725,6 +1742,8 @@ async def __call__(
         )
         if df.empty:
             df = pd.DataFrame(columns=PullRequestFacts.f)
+        if metrics is not None:
+            self._set_count_metrics(df, metrics)
         return df
 
     @sentry_span
@@ -2026,6 +2045,15 @@ async def _call_cached(
         )
         return all_facts_df, with_jira
 
+    @staticmethod
+    def _set_count_metrics(facts: pd.DataFrame, metrics: MinePullRequestMetrics) -> None:
+        metrics.prs.count = len(facts)
+        metrics.prs.done_count = facts[PullRequestFacts.f.done].sum()
+        metrics.prs.merged_count = (
+            facts[PullRequestFacts.f.merged].notnull() & ~facts[PullRequestFacts.f.done]
+        ).sum()
+        metrics.prs.open_count = facts[PullRequestFacts.f.closed].isnull().sum()
+
 
 class ParticipantsMerge:
     """Utilities to merge multiple collections of participants."""

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

@@ -25,9 +25,15 @@
 class BranchMinerMetrics:
     """Branch source data error statistics."""
 
-    empty: int
+    count: int
+    empty_count: int
     no_default: int
 
+    @classmethod
+    def empty(cls) -> "BranchMinerMetrics":
+        """Initialize a new BranchMinerMetrics instance filled with zeros."""
+        return BranchMinerMetrics(0, 0, 0)
+
 
 @cached_methods
 class BranchMiner:
@@ -176,7 +182,9 @@ async def extract_branches(
                     if items:
                         report("the following repositories have 0 branches: %s", items)
                         if metrics is not None:
-                            metrics.empty += len(items)
+                            metrics.empty_count += len(items)
+        if metrics is not None:
+            metrics.count = len(branches)
         return branches, default_branches
 
     @classmethod

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

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 import logging
@@ -67,6 +68,23 @@ class FilterCommitsProperty(Enum):
 DAG = tuple[np.ndarray, np.ndarray, np.ndarray]
 
 
+@dataclass(slots=True)
+class CommitDAGMetrics:
+    """Commit DAG error statistics, per repository.
+
+    Intersections are possible. For example, `bool(pristine & corrupted)` can be `True`.
+    """
+
+    pristine: set[str]
+    corrupted: set[str]
+    orphaned: set[str]
+
+    @classmethod
+    def empty(cls) -> "CommitDAGMetrics":
+        """Initialize a new CommitDAGMetrics instance filled with zeros."""
+        return CommitDAGMetrics(set(), set(), set())
+
+
 def _postprocess_extract_commits(result, with_deployments=True, **_):
     if isinstance(result, tuple):
         if with_deployments:
@@ -367,6 +385,7 @@ async def fetch_repository_commits(
     mdb: Database,
     pdb: Database,
     cache: Optional[aiomcache.Client],
+    metrics: Optional[CommitDAGMetrics] = None,
 ) -> dict[str, tuple[bool, DAG]]:
     """
     Load full commit DAGs for the given repositories.
@@ -379,6 +398,7 @@ async def fetch_repository_commits(
                     3. Commit timestamp. \
                     4. Commit repository name.
     :param prune: Remove any commits that are not accessible from `branches`.
+    :param metrics: Mutable error statistics, will be written on new fetches.
     :return: Map from repository names to their DAG consistency indicators and bodies.
     """
     if branches.empty:
@@ -495,6 +515,8 @@ async def execute():
     for repo, pdag in repos.items():
         if repo not in result:
             result[repo] = (True, _empty_dag()) if prune else pdag
+        if metrics is not None and len(result[repo][1][0]) == 0:
+            metrics.pristine.add(repo)
     return result
 
 
@@ -636,10 +658,12 @@ async def _fetch_commit_history_dag(
     meta_ids: tuple[int, ...],
     mdb: Database,
     alloc=None,
+    metrics: Optional[CommitDAGMetrics] = None,
 ) -> tuple[bool, str, np.ndarray, np.ndarray, np.ndarray]:
+    # these are some approximately sensible defaults, found by experiment
     max_stop_heads = 25
     max_inner_partitions = 25
-    log = logging.getLogger("%s._fetch_commit_history_dag" % metadata.__package__)
+    log = logging.getLogger(f"{metadata.__package__}._fetch_commit_history_dag")
     # there can be duplicates, remove them
     head_hashes = np.asarray(head_hashes, dtype="S40")
     head_ids = np.asarray(head_ids, dtype=int)
@@ -688,13 +712,16 @@ async def _fetch_commit_history_dag(
         bads, bad_seeds, bad_hashes = verify_edges_integrity(new_edges, alloc)
         if bads:
             log.warning(
-                "%d @ %d new DAG edges are not consistent (%d commits / %d existing): %s",
+                "%s: %d @ %d new DAG edges are not consistent (%d commits / %d existing): %s",
+                repo,
                 len(bads),
                 len(bad_seeds),
                 len(bad_hashes),
                 len(hashes),
                 [new_edges[i] for i in bad_seeds[:10]],
             )
+            if metrics is not None:
+                metrics.corrupted.add(repo)
             consistent = False
             for i in bads[::-1]:
                 new_edges.pop(i)
@@ -733,6 +760,8 @@ async def _fetch_commit_history_dag(
                     "skipping orphans which are suspiciously young: %s",
                     ", ".join(removed_orphans_hashes),
                 )
+                if metrics is not None:
+                    metrics.orphaned.add(repo)
                 consistent = False
                 for i in sorted(removed_orphans_indexes, reverse=True):
                     new_edges.pop(i)

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

@@ -141,6 +141,19 @@
 from athenian.api.unordered_unique import in1d_str, map_array_values, unordered_unique
 
 
+@dataclass(slots=True)
+class MineDeploymentsMetrics:
+    """Deployment mining error statistics."""
+
+    count: int
+    unresolved: int
+
+    @classmethod
+    def empty(cls) -> "MineDeploymentsMetrics":
+        """Initialize a new MineDeploymentsMetrics instance filled with zeros."""
+        return MineDeploymentsMetrics(0, 0)
+
+
 async def mine_deployments(
     repositories: Collection[str],
     participants: ReleaseParticipants,
@@ -166,6 +179,7 @@ async def mine_deployments(
     cache: Optional[aiomcache.Client],
     with_extended_prs: bool = False,
     with_jira: bool = False,
+    metrics: Optional[MineDeploymentsMetrics] = None,
 ) -> pd.DataFrame:
     """Gather facts about deployments that satisfy the specified filters.
 
@@ -209,6 +223,7 @@ async def mine_deployments(
         cache,
         with_extended_prs,
         with_jira,
+        metrics,
     )
     if notifications.empty:
         return pd.DataFrame()
@@ -240,6 +255,9 @@ async def mine_deployments(
     subst.fill(pd.DataFrame())
     joined["labels"].values[no_labels] = subst
 
+    if metrics is not None:
+        metrics.count = len(joined)
+
     return joined
 
 
@@ -327,6 +345,7 @@ async def _mine_deployments(
     cache: Optional[aiomcache.Client],
     with_extended_prs: bool,
     with_jira: bool,
+    metrics: Optional[MineDeploymentsMetrics],
 ) -> tuple[
     pd.DataFrame,
     pd.DataFrame,
@@ -338,6 +357,7 @@ async def _mine_deployments(
     bool,
     bool,
 ]:
+    log = logging.getLogger(f"{metadata.__package__}.mine_deployments")
     if not isinstance(repositories, (set, frozenset, KeysView)):
         repositories = set(repositories)
     if repositories:
@@ -361,10 +381,19 @@ async def _mine_deployments(
         rdb,
         cache,
     )
+    notifications_count_before_pruning = len(notifications)
     (notifications, components), labels = await gather(
         fetch_components_and_prune_unresolved(notifications, prefixer, account, rdb),
         fetch_labels(notifications.index.values, account, rdb),
     )
+    if unresolved := notifications_count_before_pruning - len(notifications):
+        log.warning(
+            "removed %d / %d unresolved notifications",
+            unresolved,
+            notifications_count_before_pruning,
+        )
+        if metrics is not None:
+            metrics.unresolved = unresolved
     if notifications.empty:
         return (*repeat(pd.DataFrame(), 7), with_extended_prs, with_jira)
 
@@ -409,7 +438,7 @@ async def _mine_deployments(
             rdb,
             cache,
         ),
-        name="_fetch_precomputed_deployed_releases(%d)" % len(notifications),
+        name=f"_fetch_precomputed_deployed_releases({len(notifications)})",
     )
     facts = await _fetch_precomputed_deployment_facts(
         notifications.index.values, default_branches, release_settings, account, pdb,
@@ -579,7 +608,6 @@ async def _invalidate_precomputed_on_out_of_order_notifications(
     not yet precomputed.
 
     A unicode string array with the names of affected precomputed deployments is returned.
-
     """
     if len(to_invalidate := _find_invalid_precomputed_deploys(notifications, missed_mask)):
         log = logging.getLogger(f"{metadata.__package__}.mine_deployments")

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

@@ -244,10 +244,8 @@ async def fetch_components_and_prune_unresolved(
     """Remove deployment notifications with unresolved components. Fetch the components."""
     components = await read_sql_query(
         select(DeployedComponent).where(
-            and_(
-                DeployedComponent.account_id == account,
-                DeployedComponent.deployment_name.in_any_values(notifications.index.values),
-            ),
+            DeployedComponent.account_id == account,
+            DeployedComponent.deployment_name.in_any_values(notifications.index.values),
         ),
         rdb,
         DeployedComponent,