Merge pull request #3140 from athenianco/persist-health

[DEV-5541] Persist account data health metrics in rdb

Author: vmarkovtsev

server/athenian/api/internal/data_health_metrics.py

@@ -1,23 +1,49 @@
+import dataclasses
 from dataclasses import dataclass
+from datetime import datetime, timezone
 
+from sqlalchemy import insert
+
+from athenian.api.db import DatabaseLike
 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
+from athenian.api.models.persistentdata.models import HealthMetric
 
 
 @dataclass(frozen=True, slots=True)
 class DataHealthMetrics:
     """Collection of data error statistics to report."""
 
-    branches: BranchMinerMetrics
-    deployments: MineDeploymentsMetrics
-    prs: MinePullRequestMetrics
-    releases: MineReleaseMetrics
-    reposet: RepositorySetMetrics
+    branches: BranchMinerMetrics | None
+    deployments: MineDeploymentsMetrics | None
+    prs: MinePullRequestMetrics | None
+    releases: MineReleaseMetrics | None
+    reposet: RepositorySetMetrics | None
 
     @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()})
+        return DataHealthMetrics(
+            **{f.name: f.type.__args__[0].empty() for f in dataclasses.fields(cls)},
+        )
+
+    @classmethod
+    def skip(cls) -> "DataHealthMetrics":
+        """Initialize a new DataHealthMetrics instance filled with None-s, effectively disabling \
+        the feature."""
+        return DataHealthMetrics(**{f.name: None for f in dataclasses.fields(cls)})
+
+    async def persist(self, account: int, rdb: DatabaseLike) -> None:
+        """Insert all the measured metrics to the database."""
+        now = datetime.now(timezone.utc)
+        values = []
+        for f in dataclasses.fields(self):
+            v = getattr(self, f.name)
+            for m in v.as_db():
+                m.account_id = account
+                m.created_at = now
+                values.append(m.explode(with_primary_keys=True))
+        await rdb.execute_many(insert(HealthMetric), values)

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

@@ -129,6 +129,7 @@
     Release,
 )
 from athenian.api.models.metadata.jira import Issue
+from athenian.api.models.persistentdata.models import HealthMetric
 from athenian.api.models.web import NoSourceDataError
 from athenian.api.pandas_io import deserialize_args, serialize_args
 from athenian.api.response import ResponseError
@@ -1660,6 +1661,13 @@ def empty(cls) -> MinePullRequestMetrics:
         """Initialize a new MinePullRequestMetrics instance filled with zeros."""
         return MinePullRequestMetrics(0, 0, 0, 0)
 
+    def as_db(self) -> Iterator[HealthMetric]:
+        """Generate HealthMetric-s from this instance."""
+        yield HealthMetric(name="prs_count", value=self.count)
+        yield HealthMetric(name="prs_done_count", value=self.done_count)
+        yield HealthMetric(name="prs_merged_count", value=self.merged_count)
+        yield HealthMetric(name="prs_open_count", value=self.open_count)
+
 
 class PRFactsCalculator:
     """Calculator for Pull Requests facts."""
@@ -2047,12 +2055,12 @@ async def _call_cached(
 
     @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 = (
+        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.prs.open_count = facts[PullRequestFacts.f.closed].isnull().sum()
+        metrics.open_count = facts[PullRequestFacts.f.closed].isnull().sum()
 
 
 class ParticipantsMerge:

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

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 import logging
-from typing import Iterable, Optional
+from typing import Iterable, Iterator, Optional
 
 import aiomcache
 import numpy as np
@@ -16,6 +16,7 @@
 from athenian.api.internal.logical_repos import coerce_logical_repos
 from athenian.api.internal.prefixer import Prefixer
 from athenian.api.models.metadata.github import Branch, NodeCommit, NodeRepositoryRef, Repository
+from athenian.api.models.persistentdata.models import HealthMetric
 from athenian.api.pandas_io import deserialize_args, serialize_args
 from athenian.api.to_object_arrays import is_not_null
 from athenian.api.tracing import sentry_span
@@ -34,6 +35,12 @@ def empty(cls) -> "BranchMinerMetrics":
         """Initialize a new BranchMinerMetrics instance filled with zeros."""
         return BranchMinerMetrics(0, 0, 0)
 
+    def as_db(self) -> Iterator[HealthMetric]:
+        """Generate HealthMetric-s from this instance."""
+        yield HealthMetric(name="branches_count", value=self.count)
+        yield HealthMetric(name="branches_empty_count", value=self.empty_count)
+        yield HealthMetric(name="branches_no_default", value=self.no_default)
+
 
 @cached_methods
 class BranchMiner:

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

@@ -3,7 +3,7 @@
 from enum import Enum
 import logging
 import pickle
-from typing import Collection, Iterable, KeysView, Mapping, Optional, Sequence, Union
+from typing import Collection, Iterable, Iterator, KeysView, Mapping, Optional, Sequence, Union
 
 import aiomcache
 import numpy as np
@@ -48,6 +48,7 @@
     PushCommit,
     Release,
 )
+from athenian.api.models.persistentdata.models import HealthMetric
 from athenian.api.models.precomputed.models import GitHubCommitHistory
 from athenian.api.native.mi_heap_destroy_stl_allocator import make_mi_heap_allocator_capsule
 from athenian.api.pandas_io import deserialize_args, serialize_args
@@ -84,6 +85,12 @@ def empty(cls) -> "CommitDAGMetrics":
         """Initialize a new CommitDAGMetrics instance filled with zeros."""
         return CommitDAGMetrics(set(), set(), set())
 
+    def as_db(self) -> Iterator[HealthMetric]:
+        """Generate HealthMetric-s from this instance."""
+        yield HealthMetric(name="commits_pristine", value=len(self.pristine))
+        yield HealthMetric(name="commits_corrupted", value=len(self.corrupted))
+        yield HealthMetric(name="commits_orphaned", value=len(self.orphaned))
+
 
 def _postprocess_extract_commits(result, with_deployments=True, **_):
     if isinstance(result, tuple):

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

@@ -6,7 +6,7 @@
 import json
 import logging
 from operator import attrgetter
-from typing import Any, Collection, KeysView, Mapping, NamedTuple, Optional, Sequence
+from typing import Any, Collection, Iterator, KeysView, Mapping, NamedTuple, Optional, Sequence
 
 import aiomcache
 import numpy as np
@@ -124,6 +124,7 @@
     DeployedComponent,
     DeployedLabel,
     DeploymentNotification,
+    HealthMetric,
     ReleaseNotification,
 )
 from athenian.api.models.precomputed.models import (
@@ -153,6 +154,11 @@ def empty(cls) -> "MineDeploymentsMetrics":
         """Initialize a new MineDeploymentsMetrics instance filled with zeros."""
         return MineDeploymentsMetrics(0, 0)
 
+    def as_db(self) -> Iterator[HealthMetric]:
+        """Generate HealthMetric-s from this instance."""
+        yield HealthMetric(name="deployments_count", value=self.count)
+        yield HealthMetric(name="deployments_unresolved", value=self.unresolved)
+
 
 async def mine_deployments(
     repositories: Collection[str],

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

@@ -5,7 +5,7 @@
 import logging
 import pickle
 import re
-from typing import Iterable, KeysView, Mapping, Optional, Sequence
+from typing import Iterable, Iterator, KeysView, Mapping, Optional, Sequence
 
 import aiomcache
 import morcilla
@@ -57,7 +57,7 @@
     default_branch_alias,
 )
 from athenian.api.models.metadata.github import Branch, NodeCommit, PushCommit, Release, User
-from athenian.api.models.persistentdata.models import ReleaseNotification
+from athenian.api.models.persistentdata.models import HealthMetric, ReleaseNotification
 from athenian.api.models.precomputed.models import (
     GitHubRelease as PrecomputedRelease,
     GitHubReleaseMatchTimespan,
@@ -88,6 +88,15 @@ def empty(cls) -> "MineReleaseMetrics":
         """Initialize a new MineReleaseMetrics instance filled with zeros."""
         return MineReleaseMetrics(CommitDAGMetrics.empty(), 0, 0, 0, {}, 0)
 
+    def as_db(self) -> Iterator[HealthMetric]:
+        """Generate HealthMetric-s from this instance."""
+        yield from self.commits.as_db()
+        yield HealthMetric(name="releases_by_branch", value=self.count_by_branch)
+        yield HealthMetric(name="releases_by_tag", value=self.count_by_tag)
+        yield HealthMetric(name="releases_by_event", value=self.count_by_event)
+        yield HealthMetric(name="releases_empty", value=sum(self.empty_releases.values()))
+        yield HealthMetric(name="releases_unresolved", value=self.unresolved)
+
 
 class ReleaseLoader:
     """Loader for releases."""

server/athenian/api/internal/reposet.py

@@ -30,6 +30,7 @@
 from athenian.api.internal.miners.access_classes import access_classes
 from athenian.api.internal.prefixer import Prefixer, RepositoryName
 from athenian.api.models.metadata.github import AccountRepository, NodeUser
+from athenian.api.models.persistentdata.models import HealthMetric
 from athenian.api.models.state.models import LogicalRepository, RepositorySet, UserAccount
 from athenian.api.models.web import (
     ForbiddenError,
@@ -48,14 +49,19 @@
 class RepositorySetMetrics:
     """Reposet synchronization error statistics."""
 
-    count: int
+    length: int
     undead: int
 
     @classmethod
     def empty(cls) -> RepositorySetMetrics:
         """Initialize a new RepositorySetMetrics instance filled with zeros."""
         return RepositorySetMetrics(0, 0)
 
+    def as_db(self) -> Iterator[HealthMetric]:
+        """Generate HealthMetric-s from this instance."""
+        yield HealthMetric(name="reposet_length", value=self.length)
+        yield HealthMetric(name="reposet_undead", value=self.undead)
+
 
 def reposet_items_to_refs(items: list[tuple[int, str, str]]) -> Iterator[RepositoryReference]:
     """Convert the raw DB repository tuples to RepositoryReference."""

server/athenian/api/models/persistentdata/models.py

@@ -125,3 +125,18 @@ class DeployedComponent(create_time_mixin(created_at=True, updated_at=False), Ba
     # de-referenced commit node ID in metadata DB
     resolved_commit_node_id = Column(BigInteger(), info={"reset_nulls": True})
     resolved_at = Column(TIMESTAMP(timezone=True), nullable=True)
+
+
+class HealthMetric(Base):
+    """Account data health statistics."""
+
+    __tablename__ = "health_metrics"
+
+    name = Column(Text(), primary_key=True)
+    created_at = Column(
+        TIMESTAMP(timezone=True),
+        primary_key=True,
+        default=lambda: datetime.now(timezone.utc),
+        server_default=func.now(),
+    )
+    value = Column(JSONB().with_variant(JSON(), sqlite.dialect.name), nullable=False)

server/athenian/api/models/persistentdata/versions/41fcc01cccc0_add_health_metrics_table.py

@@ -0,0 +1,53 @@
+"""Add health_metrics table
+
+Revision ID: 41fcc01cccc0
+Revises: f0ae6bfc60b1
+Create Date: 2022-12-16 15:55:24.105902+00:00
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import sqlite
+from sqlalchemy.dialects.postgresql import JSONB
+
+# revision identifiers, used by Alembic.
+revision = "41fcc01cccc0"
+down_revision = "f0ae6bfc60b1"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    name = "health_metrics"
+    if op.get_bind().dialect.name == "postgresql":
+        schema_arg = {"schema": "athenian"}
+    else:
+        name = "athenian." + name
+        schema_arg = {}
+    op.create_table(
+        name,
+        sa.Column("account_id", sa.Integer(), primary_key=True),
+        sa.Column("name", sa.Text(), primary_key=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            primary_key=True,
+            server_default=sa.func.now(),
+        ),
+        sa.Column(
+            "value",
+            JSONB().with_variant(sa.JSON(), sqlite.dialect.name),
+            nullable=False,
+        ),
+        **schema_arg,
+    )
+
+
+def downgrade():
+    name = "health_metrics"
+    if op.get_bind().dialect.name == "postgresql":
+        schema_arg = {"schema": "athenian"}
+    else:
+        name = "athenian." + name
+        schema_arg = {}
+    op.drop_table(name, **schema_arg)

server/athenian/api/precompute/__main__.py

@@ -115,6 +115,11 @@ class Formatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawTextHelpForm
     accounts_parser.add_argument(
         "--skip-teams", action="store_true", help="Do not touch anything related to teams.",
     )
+    accounts_parser.add_argument(
+        "--skip-health-metrics",
+        action="store_true",
+        help="Neither measure nor persist account data health metrics.",
+    )
     accounts_parser.add_argument(
         "--time-from",
         type=deserialize_datetime,