wip

Author: MarceloRobert

backend/kernelCI_app/helpers/trees.py

@@ -57,6 +57,7 @@ def sanitize_tree(
     """Sanitizes a checkout that was returned by a 'treelisting-like' query
 
     Returns a Checkout object"""
+
     build_status = StatusCount(
         PASS=checkout["pass_builds"],
         FAIL=checkout["fail_builds"],
@@ -87,13 +88,23 @@ def sanitize_tree(
         "skip": checkout["skip_boots"],
     }
 
-    if isinstance(checkout.get("git_commit_tags"), str):
+    # Has to check if it's a string because sqlite doesn't support ArrayFields.
+    # So if the query came from sqlite, it will be a string.
+    git_commit_tags = checkout.get("git_commit_tags")
+    if isinstance(git_commit_tags, str):
         try:
             checkout["git_commit_tags"] = json.loads(checkout["git_commit_tags"])
             if not isinstance(checkout["git_commit_tags"], list):
                 checkout["git_commit_tags"] = []
         except json.JSONDecodeError:
             checkout["git_commit_tags"] = []
+    elif git_commit_tags and isinstance(git_commit_tags, list):
+        first_tag = git_commit_tags[0]
+        if isinstance(first_tag, str):
+            # The git_commit_tags comes as list[str] on a normal query, but `Checkout`
+            # expects list[list[str]]. This is a workaround, the queries should *always*
+            # return a simples list[str].
+            checkout["git_commit_tags"] = [git_commit_tags]
 
     return Checkout(
         **checkout,

backend/kernelCI_app/management/commands/helpers/aggregation_helpers.py

@@ -2,29 +2,20 @@
 from typing import Sequence
 
 
-from django.db import connection
+from django.db import connections
 from kernelCI_app.helpers.logger import out
 from kernelCI_app.models import (
     Checkouts,
     PendingTest,
-    StatusChoices,
-    SimplifiedStatusChoices,
+    PendingBuilds,
     Tests,
+    Builds,
+    simplify_status,
 )
 from kernelCI_app.utils import is_boot
-from typing import Optional
 
 
-def simplify_status(status: Optional[StatusChoices]) -> SimplifiedStatusChoices:
-    if status == StatusChoices.PASS:
-        return SimplifiedStatusChoices.PASS
-    elif status == StatusChoices.FAIL:
-        return SimplifiedStatusChoices.FAIL
-    else:
-        return SimplifiedStatusChoices.INCONCLUSIVE
-
-
-def convert_test(t: Tests) -> PendingTest:
+def _convert_test(t: Tests) -> PendingTest:
     return PendingTest(
         test_id=t.id,
         origin=t.origin,
@@ -36,14 +27,78 @@ def convert_test(t: Tests) -> PendingTest:
     )
 
 
+def _update_tree_listing(*, checkouts_instances: Sequence[Checkouts]):
+    """
+    Whenever a checkout updates the latest_checkout table,
+    we update the tree_listing table, zeroing the counts"""
+
+    t0 = time.time()
+    checkout_values = [
+        (
+            checkout.id,
+            checkout.origin,
+            checkout.tree_name,
+            checkout.git_repository_url,
+            checkout.git_repository_branch,
+            checkout.git_commit_hash,
+            checkout.git_commit_name,
+            checkout.git_commit_tags,
+            checkout.start_time,
+            checkout.origin_builds_finish_time,
+            checkout.origin_tests_finish_time,
+        )
+        for checkout in checkouts_instances
+    ]
+
+    with connections["default"].cursor() as cursor:
+        # Set values as 0 when inserting a new tree and update as 0 when tree already exists
+        cursor.executemany(
+            """
+            INSERT INTO tree_listing (
+                checkout_id, origin, tree_name,
+                git_repository_url, git_repository_branch, git_commit_hash,
+                git_commit_name, git_commit_tags, start_time,
+                origin_builds_finish_time, origin_tests_finish_time,
+                pass_builds, fail_builds, inconclusive_builds,
+                pass_boots, fail_boots, inconclusive_boots,
+                pass_tests, fail_tests, inconclusive_tests
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 0, 0, 0, 0, 0, 0, 0, 0, 0)
+            ON CONFLICT (origin, tree_name, git_repository_url, git_repository_branch)
+            DO UPDATE SET
+                checkout_id = EXCLUDED.checkout_id,
+                git_commit_hash = EXCLUDED.git_commit_hash,
+                git_commit_name = EXCLUDED.git_commit_name,
+                git_commit_tags = EXCLUDED.git_commit_tags,
+                start_time = EXCLUDED.start_time,
+                origin_builds_finish_time = EXCLUDED.origin_builds_finish_time,
+                origin_tests_finish_time = EXCLUDED.origin_tests_finish_time,
+                pass_builds = 0,
+                fail_builds = 0,
+                inconclusive_builds = 0,
+                pass_boots = 0,
+                fail_boots = 0,
+                inconclusive_boots = 0,
+                pass_tests = 0,
+                fail_tests = 0,
+                inconclusive_tests = 0
+            WHERE tree_listing.start_time < EXCLUDED.start_time
+            """,
+            checkout_values,
+        )
+    out(
+        f"upserted {len(checkouts_instances)} tree_listing trees in {time.time() - t0:.3f}s"
+    )
+
+
 def aggregate_checkouts(checkouts_instances: Sequence[Checkouts]) -> None:
     """
     Insert checkouts on latest_checkouts table,
     maintaining only the latest ones for each
     (origin, tree_name, git_repository_url, git_repository_branch) combination
     """
     t0 = time.time()
-    values = [
+    checkout_values = [
         (
             checkout.id,
             checkout.origin,
@@ -55,7 +110,7 @@ def aggregate_checkouts(checkouts_instances: Sequence[Checkouts]) -> None:
         for checkout in checkouts_instances
     ]
 
-    with connection.cursor() as cursor:
+    with connections["default"].cursor() as cursor:
         cursor.executemany(
             """
             INSERT INTO latest_checkout (
@@ -69,7 +124,7 @@ def aggregate_checkouts(checkouts_instances: Sequence[Checkouts]) -> None:
                 checkout_id = EXCLUDED.checkout_id
             WHERE latest_checkout.start_time < EXCLUDED.start_time
             """,
-            values,
+            checkout_values,
         )
     out(f"inserted {len(checkouts_instances)} checkouts in {time.time() - t0:.3f}s")
 
@@ -80,7 +135,7 @@ def aggregate_tests(
     """Insert tests data on pending_tests table to be processed later"""
     t0 = time.time()
     pending_tests = (
-        convert_test(test)
+        _convert_test(test)
         for test in tests_instances
         if test.environment_misc and test.environment_misc.get("platform") is not None
     )
@@ -95,9 +150,29 @@ def aggregate_tests(
         )
 
 
-def aggregate_checkouts_and_tests(
+def aggregate_builds(
+    build_instances: Sequence[Builds],
+) -> None:
+    """Insert builds data on pending_builds table to be processed later"""
+    t0 = time.time()
+    pending_builds = (PendingBuilds.from_builds(build) for build in build_instances)
+
+    if pending_builds:
+        pending_builds_inserted = PendingBuilds.objects.bulk_create(
+            pending_builds,
+            ignore_conflicts=True,
+        )
+        out(
+            f"bulk_create pending_builds: n={len(pending_builds_inserted)} in {time.time() - t0:.3f}s"
+        )
+
+
+def aggregate_checkouts_and_pendings(
     checkouts_instances: Sequence[Checkouts],
     tests_instances: Sequence[Tests],
+    build_instances: Sequence[Builds],
 ) -> None:
     aggregate_checkouts(checkouts_instances)
+    _update_tree_listing(checkouts_instances=checkouts_instances)
     aggregate_tests(tests_instances)
+    aggregate_builds(build_instances)

backend/kernelCI_app/management/commands/helpers/denormal.py

@@ -0,0 +1,55 @@
+from kernelCI_app.models import TreeListing
+
+
+def _update_tree_listing(*, tree_listing_objects: list[TreeListing]) -> None:
+    """Helper function to centralize and reutilize updates to TreeListing"""
+
+    if not tree_listing_objects:
+        return
+
+    TreeListing.objects.bulk_create(
+        tree_listing_objects,
+        update_conflicts=True,
+        unique_fields=[
+            "origin",
+            "tree_name",
+            "git_repository_branch",
+            "git_repository_url",
+        ],
+        update_fields=[
+            "field_timestamp",
+            "checkout_id",
+            "origin",
+            "tree_name",
+            "git_repository_url",
+            "git_repository_branch",
+            "git_commit_hash",
+            "git_commit_name",
+            "git_commit_tags",
+            "start_time",
+            "origin_builds_finish_time",
+            "origin_tests_finish_time",
+            "pass_builds",
+            "fail_builds",
+            "done_builds",
+            "miss_builds",
+            "skip_builds",
+            "error_builds",
+            "null_builds",
+            "pass_boots",
+            "fail_boots",
+            "done_boots",
+            "miss_boots",
+            "skip_boots",
+            "error_boots",
+            "null_boots",
+            "pass_tests",
+            "fail_tests",
+            "done_tests",
+            "miss_tests",
+            "skip_tests",
+            "error_tests",
+            "null_tests",
+        ],
+    )
+    print(f"Updated {len(tree_listing_objects)} trees in TreeListing", flush=True)

backend/kernelCI_app/management/commands/helpers/kcidbng_ingester.py

@@ -26,7 +26,7 @@
 )
 import kcidb_io
 from kernelCI_app.management.commands.helpers.aggregation_helpers import (
-    aggregate_checkouts_and_tests,
+    aggregate_checkouts_and_pendings,
 )
 from django.db import connections, transaction
 from kernelCI_app.models import Issues, Checkouts, Builds, Tests, Incidents
@@ -200,9 +200,10 @@ def flush_buffers(
             consume_buffer(builds_buf, "builds")
             consume_buffer(tests_buf, "tests")
             consume_buffer(incidents_buf, "incidents")
-            aggregate_checkouts_and_tests(
+            aggregate_checkouts_and_pendings(
                 checkouts_instances=checkouts_buf,
                 tests_instances=tests_buf,
+                build_instances=builds_buf,
             )
         for filename, filepath in buffer_files:
             os.rename(filepath, os.path.join(archive_dir, filename))