feat: hardware denormalization for hardware listing (#1614)

* feat(hardware): add denormalized hardware status schema and models

- Add HardwareStatus model to store aggregated test/build/boot status by platform
- Add LatestCheckout model to track latest checkout per origin/tree/repository
- Add PendingTest model to queue tests for aggregation processing
- Add ProcessedHardwareStatus model to track processed entities
- Add SimplifiedStatusChoices enum for aggregated status values
- Create database migration for new tables with appropriate indexes

* feat(hardware): implement real-time hardware status aggregation

- Add aggregation_helpers module with checkout and test aggregation logic
- Integrate aggregation into kcidbng_ingester db_worker
- Populate LatestCheckout table with checkout tracking
- Populate PendingTest table with test records for processing
- Run aggregations before buffer flush to ensure data consistency

* feat(hardware): add backlog processing and query support

- Add process_pending_aggregations management command for backlog processing
- Implement batch processing with configurable batch size and loop mode
- Add hardware status aggregation from PendingTest queue
- Add get_hardware_listing_data_from_status_table query function
- Query denormalized hardware_status table for improved performance

* test: fix db_worker tests mocking aggregation helper

* feat: add cronjob to delete unused hardware status rows every week

* feat: add command to backfill data into PendingTests and LatestCheckout

Author: gustavobtflores

backend/kernelCI/settings.py

@@ -179,6 +179,13 @@ def get_json_env_var(name, default):
                 "--yes",
             ],
         ),
+        (
+            "0 0 * * 1",
+            "django.core.management.call_command",
+            [
+                "delete_unused_hardware_status",
+            ],
+        ),
     ]
 
 # Email settings for SMTP backend

backend/kernelCI_app/management/commands/backfill_hardware_aggregations.py

@@ -0,0 +1,133 @@
+import time
+from datetime import timedelta
+from typing import Any, Callable, Sequence
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+from django.db.models.query import QuerySet
+from django.utils import timezone
+
+from kernelCI_app.management.commands.helpers.aggregation_helpers import (
+    aggregate_checkouts,
+    aggregate_tests,
+)
+from kernelCI_app.models import (
+    Checkouts,
+    LatestCheckout,
+    PendingTest,
+    Tests,
+)
+from kernelCI_app.helpers.logger import out
+
+
+class Command(BaseCommand):
+    help = """
+        Backfill hardware aggregations (LatestCheckout and PendingTest)
+        which can later be collected and processed by the process_pending_aggregations command.
+    """
+
+    def add_arguments(self, parser: Any) -> None:
+        parser.add_argument(
+            "--days",
+            type=int,
+            default=30,
+            help="Number of days to look back (default: 30)",
+        )
+        parser.add_argument(
+            "--truncate",
+            action="store_true",
+            help="""Truncate destination tables before backfilling.
+                This is a destructive operation and should be used with caution.""",
+        )
+        parser.add_argument(
+            "--batch-size",
+            type=int,
+            default=2000,
+            help="Batch size for processing (default: 2000)",
+        )
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        days = options["days"]
+        truncate = options["truncate"]
+        batch_size = options["batch_size"]
+
+        cutoff_date = timezone.now() - timedelta(days=days)
+        out(f"Backfilling hardware aggregations since {cutoff_date}...")
+
+        if truncate:
+            out("Truncating LatestCheckout and PendingTest tables...")
+            LatestCheckout.objects.all().delete()
+            PendingTest.objects.all().delete()
+            out("Truncation complete.")
+
+        out("Backfilling LatestCheckout...")
+        checkouts_qs = Checkouts.objects.filter(start_time__gte=cutoff_date).order_by(
+            "-start_time"
+        )
+
+        # Process checkouts to insert into latest_checkout table
+        self.process_queryset(
+            checkouts_qs, aggregate_checkouts, batch_size, "Checkouts"
+        )
+
+        out("Backfilling PendingTest...")
+        tests_qs = (
+            Tests.objects.filter(
+                start_time__gte=cutoff_date,
+                environment_misc__platform__isnull=False,
+                build__checkout__id__in=LatestCheckout.objects.values_list(
+                    "checkout_id", flat=True
+                ),
+            )
+            .select_related("build")
+            .order_by("start_time")
+        )
+
+        # Process tests to insert into pending_tests table
+        self.process_queryset(tests_qs, aggregate_tests, batch_size, "Tests")
+
+        out("Backfill complete.")
+
+    def process_queryset(
+        self,
+        queryset: QuerySet[Any],
+        aggregator_func: Callable[[Sequence[Any]], None],
+        batch_size: int,
+        label: str,
+    ) -> None:
+        """
+        Process a queryset in batches using the provided aggregator function
+
+        Iterates through the queryset in chunks, collects items into batches,
+        and processes each batch using the aggregator function within a transaction
+        """
+        count = 0
+        total_processed = 0
+        batch = []
+
+        t0 = time.time()
+        for item in queryset.iterator(chunk_size=batch_size):
+            try:
+                batch.append(item)
+                count += 1
+                if count >= batch_size:
+                    with transaction.atomic():
+                        aggregator_func(batch)
+                    total_processed += count
+                    out(
+                        f"Processed {total_processed} {label} (elapsed time: {time.time() - t0:.2f}s)"
+                    )
+                    batch = []
+                    count = 0
+            except Exception as e:
+                out(f"Error processing {label}: {e}")
+                batch = []
+                count = 0
+                continue
+
+        if batch:
+            aggregator_func(batch)
+            total_processed += count
+            out(
+                f"Processed {total_processed} {label} (elapsed time: {time.time() - t0:.2f}s)"
+            )

backend/kernelCI_app/management/commands/delete_unused_hardware_status.py

@@ -0,0 +1,122 @@
+"""
+Management command to delete unused entries from hardware_status table.
+
+Removes HardwareStatus entries that have no corresponding checkout_id in the LatestCheckout table.
+"""
+
+import logging
+from django.core.management.base import BaseCommand
+from django.db import transaction
+from kernelCI_app.models import HardwareStatus, LatestCheckout, ProcessedHardwareStatus
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    help = (
+        "Delete HardwareStatus entries with no corresponding checkout_id "
+        "in the LatestCheckout table"
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Show what would be deleted without actually deleting",
+        )
+        parser.add_argument(
+            "--batch-size",
+            type=int,
+            default=10000,
+            help="Number of records to delete per batch (default: 10000)",
+        )
+
+    def handle(self, *args, **options):
+        dry_run = options["dry_run"]
+        batch_size = options["batch_size"]
+
+        with transaction.atomic():
+            valid_checkout_ids = set(
+                LatestCheckout.objects.values_list("checkout_id", flat=True)
+            )
+
+            orphaned_hardware_entries = HardwareStatus.objects.exclude(
+                checkout_id__in=valid_checkout_ids
+            ).values_list("checkout_id", flat=True)
+            orphaned_hardware_count = orphaned_hardware_entries.count()
+
+            orphaned_processed_hardware_entries = (
+                ProcessedHardwareStatus.objects.exclude(
+                    checkout_id__in=valid_checkout_ids
+                )
+            ).values_list("hardware_key", flat=True)
+
+            orphaned_processed_hardware_count = (
+                orphaned_processed_hardware_entries.count()
+            )
+
+            if orphaned_hardware_count == 0 and orphaned_processed_hardware_count == 0:
+                self.stdout.write(
+                    self.style.SUCCESS(
+                        "No orphaned HardwareStatus/ProcessedHardwareStatus entries found."
+                    )
+                )
+                return
+
+            if dry_run:
+                self.stdout.write(
+                    self.style.WARNING(
+                        f"[DRY RUN] Would delete {orphaned_hardware_count} HardwareStatus entries and "
+                        f"{orphaned_processed_hardware_count} ProcessedHardwareStatus 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 "
+                "with no corresponding LatestCheckout."
+            )
+
+            total_hardware_deleted = 0
+            total_processed_hardware_deleted = 0
+            while True:
+                hardware_batch_ids = list(orphaned_hardware_entries[:batch_size])
+                processed_hardware_batch_ids = list(
+                    orphaned_processed_hardware_entries[:batch_size]
+                )
+
+                if not hardware_batch_ids and not processed_hardware_batch_ids:
+                    break
+
+                if hardware_batch_ids:
+                    hardware_delete_count = HardwareStatus.objects.filter(
+                        checkout_id__in=hardware_batch_ids
+                    ).delete()[0]
+                    self.stdout.write(
+                        f"Deleted hardware_status(n={hardware_delete_count}) entries "
+                        f"(total: {total_hardware_deleted}/{orphaned_hardware_count})"
+                    )
+                    total_hardware_deleted += hardware_delete_count
+
+                if processed_hardware_batch_ids:
+                    processed_hardware_delete_count = (
+                        ProcessedHardwareStatus.objects.filter(
+                            hardware_key__in=processed_hardware_batch_ids
+                        ).delete()[0]
+                    )
+
+                    total_processed_hardware_deleted += processed_hardware_delete_count
+
+                    self.stdout.write(
+                        f"Deleted processed_hardware_status(n={processed_hardware_delete_count}) entries "
+                        f"(total: {total_processed_hardware_deleted}/{orphaned_processed_hardware_count})"
+                    )
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"Successfully deleted hardware_status(n={total_hardware_deleted}) "
+                f"and processed_hardware_status(n={total_processed_hardware_deleted})."
+            )
+        )

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

@@ -0,0 +1,103 @@
+import time
+from typing import Sequence
+
+
+from django.db import connection
+from kernelCI_app.helpers.logger import out
+from kernelCI_app.models import (
+    Checkouts,
+    PendingTest,
+    StatusChoices,
+    SimplifiedStatusChoices,
+    Tests,
+)
+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:
+    return PendingTest(
+        test_id=t.id,
+        origin=t.origin,
+        platform=t.environment_misc.get("platform"),
+        compatible=t.environment_compatible,
+        build_id=t.build_id,
+        status=simplify_status(t.status),
+        is_boot=is_boot(t.path) if t.path else False,
+    )
+
+
+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.id,
+            checkout.origin,
+            checkout.tree_name,
+            checkout.git_repository_url,
+            checkout.git_repository_branch,
+            checkout.start_time,
+        )
+        for checkout in checkouts_instances
+    ]
+
+    with connection.cursor() as cursor:
+        cursor.executemany(
+            """
+            INSERT INTO latest_checkout (
+                checkout_id, origin, tree_name,
+                git_repository_url, git_repository_branch, start_time
+            )
+            VALUES (%s, %s, %s, %s, %s, %s)
+            ON CONFLICT (origin, tree_name, git_repository_url, git_repository_branch)
+            DO UPDATE SET
+                start_time = EXCLUDED.start_time,
+                checkout_id = EXCLUDED.checkout_id
+            WHERE latest_checkout.start_time < EXCLUDED.start_time
+            """,
+            values,
+        )
+    out(f"inserted {len(checkouts_instances)} checkouts in {time.time() - t0:.3f}s")
+
+
+def aggregate_tests(
+    tests_instances: Sequence[Tests],
+) -> None:
+    """Insert tests data on pending_tests table to be processed later"""
+    t0 = time.time()
+    pending_tests = (
+        convert_test(test)
+        for test in tests_instances
+        if test.environment_misc and test.environment_misc.get("platform") is not None
+    )
+
+    if pending_tests:
+        pending_tests_inserted = PendingTest.objects.bulk_create(
+            pending_tests,
+            ignore_conflicts=True,
+        )
+        out(
+            f"bulk_create pending_tests: n={len(pending_tests_inserted)} in {time.time() - t0:.3f}s"
+        )
+
+
+def aggregate_checkouts_and_tests(
+    checkouts_instances: Sequence[Checkouts],
+    tests_instances: Sequence[Tests],
+) -> None:
+    aggregate_checkouts(checkouts_instances)
+    aggregate_tests(tests_instances)