Feat: tree listing denormalization

backend/kernelCI_app/management/commands/delete_unused_hardware_status.py

@@ -7,7 +7,7 @@
 import logging
 from django.core.management.base import BaseCommand
 from django.db import transaction
-from kernelCI_app.models import HardwareStatus, LatestCheckout, ProcessedHardwareStatus
+from kernelCI_app.models import HardwareStatus, LatestCheckout, ProcessedListingItems
 
 logger = logging.getLogger(__name__)
 
@@ -46,10 +46,10 @@ def handle(self, *args, **options):
             orphaned_hardware_count = orphaned_hardware_entries.count()
 
             orphaned_processed_hardware_entries = (
-                ProcessedHardwareStatus.objects.exclude(
+                ProcessedListingItems.objects.exclude(
                     checkout_id__in=valid_checkout_ids
                 )
-            ).values_list("hardware_key", flat=True)
+            ).values_list("listing_item_key", flat=True)
 
             orphaned_processed_hardware_count = (
                 orphaned_processed_hardware_entries.count()
@@ -58,7 +58,7 @@ def handle(self, *args, **options):
             if orphaned_hardware_count == 0 and orphaned_processed_hardware_count == 0:
                 self.stdout.write(
                     self.style.SUCCESS(
-                        "No orphaned HardwareStatus/ProcessedHardwareStatus entries found."
+                        "No orphaned HardwareStatus/ProcessedListingItems entries found."
                     )
                 )
                 return
@@ -67,15 +67,15 @@ def handle(self, *args, **options):
                 self.stdout.write(
                     self.style.WARNING(
                         f"[DRY RUN] Would delete {orphaned_hardware_count} HardwareStatus entries and "
-                        f"{orphaned_processed_hardware_count} ProcessedHardwareStatus entries "
+                        f"{orphaned_processed_hardware_count} ProcessedListingItems entries "
                         "Run without --dry-run to execute deletion."
                     )
                 )
                 return
 
             self.stdout.write(
                 f"Found {orphaned_hardware_count} HardwareStatus entries "
-                f"and {orphaned_processed_hardware_count} ProcessedHardwareStatus entries "
+                f"and {orphaned_processed_hardware_count} ProcessedListingItems entries "
                 "with no corresponding LatestCheckout."
             )
 
@@ -102,8 +102,8 @@ def handle(self, *args, **options):
 
                 if processed_hardware_batch_ids:
                     processed_hardware_delete_count = (
-                        ProcessedHardwareStatus.objects.filter(
-                            hardware_key__in=processed_hardware_batch_ids
+                        ProcessedListingItems.objects.filter(
+                            listing_item_key__in=processed_hardware_batch_ids
                         ).delete()[0]
                     )
 

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

@@ -2,14 +2,16 @@
 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,
+    PendingBuilds,
     StatusChoices,
     SimplifiedStatusChoices,
     Tests,
+    Builds,
 )
 from kernelCI_app.utils import is_boot
 from typing import Optional
@@ -24,6 +26,15 @@ def simplify_status(status: Optional[StatusChoices]) -> SimplifiedStatusChoices:
         return SimplifiedStatusChoices.INCONCLUSIVE
 
 
+def convert_build(b: Builds) -> PendingBuilds:
+    return PendingBuilds(
+        build_id=b.id,
+        origin=b.origin,
+        checkout_id=b.checkout_id,
+        status=simplify_status(b.status),
+    )
+
+
 def convert_test(t: Tests) -> PendingTest:
     return PendingTest(
         test_id=t.id,
@@ -36,6 +47,68 @@ 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.
+
+    We should also remove the old tree row from the listing
+    """
+
+    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,
+        )
+        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,
+                build_pass, build_failed, build_inc,
+                boot_pass, boot_failed, boot_inc,
+                test_pass, test_failed, test_inc
+            )
+            VALUES (%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,
+                build_pass = 0,
+                build_failed = 0,
+                build_inc = 0,
+                boot_pass = 0,
+                boot_failed = 0,
+                boot_inc = 0,
+                test_pass = 0,
+                test_failed = 0,
+                test_inc = 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,
@@ -55,7 +128,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 (
@@ -95,9 +168,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 = (convert_build(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)
     aggregate_tests(tests_instances)
+    aggregate_builds(build_instances)

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))