Custom event based build reports

Signed-off-by: magic_rb <[email protected]>

Author: MagicRB

buildbot_nix/__init__.py

@@ -9,7 +9,7 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Iterable
 import typing
 import copy
 
@@ -86,10 +86,11 @@ class BuildTrigger(buildstep.ShellMixin, steps.BuildStep):
     running: bool
     wait_for_finish_deferred: defer.Deferred[tuple[list[int], int]] | None
     brids: list[int]
+    consumers: dict[int, Any]
 
     @dataclass
     class ScheduledJob:
-        job: NixEvalJobSuccess
+        job: NixEvalJob
         builder_ids: dict[int, int]
         results: defer.Deferred[list[int]]
 
@@ -135,6 +136,7 @@ def __init__(
             None
         )
         self.brids = []
+        self.consumers = {}
         super().__init__(**kwargs)
 
     def interrupt(self, reason: str | Failure) -> None:
@@ -267,6 +269,9 @@ def schedule(
             yield self.addURL(f"{scheduler.name} #{brid}", url)
             self._add_results(brid)
 
+        if not self.combine_builds:
+            self.produceEventForBuildRequestsById(brids.values(), "started-nix-build", None)
+
         return brids, results_deferred
 
     @defer.inlineCallbacks
@@ -351,31 +356,72 @@ def get_failed_dependents(
         return removed
 
     @defer.inlineCallbacks
-    def produceEventForBuilds(
+
+    @defer.inlineCallbacks
+    def produceEventForBuildRequestsById(
             self,
-            build_ids: list[int],
+            buildrequest_ids: Iterable[int],
             event: str,
             result: None | int,
     ) -> Generator[Any, object, None]:
-        for build_id in build_ids:
-          build = yield self.master.data.get(('builds', str(build_id)))
-          buildT = typing.cast(dict[str, Any], build)
-          if result is not None:
-              buildT["results"] = result
-          self.master.mq.produce(("builds", str(build_id), event), copy.deepcopy(buildT))
+        buildrequest_ids = list(buildrequest_ids)
+
+        def gotBuild(key: int, build: Any) -> None:
+            if build['buildrequestid'] in buildrequest_ids:
+              log.info("got build {key} : {build}", key = key, build = build)
+              # we only care about the first started build
+              buildrequest_ids.remove(int(build['buildrequestid']))
+              if not buildrequest_ids:
+                  del self.consumers[build['buildrequestid']]
+              self.produceEventForBuild(event, build, result)
+
+        for buildrequest_id in buildrequest_ids:
+          builds: Any = yield self.master.data.get(('buildrequests', str(buildrequest_id), 'builds'))
+
+          if not builds:
+              # only subscribe to new builds, if there are no existing builds
+              self.consumers[buildrequest_id] = yield self.master.mq.startConsuming(
+                  gotBuild, ('builds', None, None)
+              )
+          # send starts for any exsting builds
+          for build in builds:
+              self.produceEventForBuild(event, build, result)
 
     @defer.inlineCallbacks
-    def produceEvent(
+    def produceEventForBuildById(
             self,
             event: str,
+            build_id: int,
             result: None | int,
     ) -> Generator[Any, object, None]:
-        yield self.produceEventForBuilds([self.build.buildid], event, result)
+        build: Any = yield self.master.data.get(('builds', str(build_id)))
+        self.produceEventForBuild(event, build, result)
+
+    def produceEventForBuild(
+            self,
+            event: str,
+            build: Any,
+            result: None | int,
+    ) -> None:
+        if result is not None:
+            build["results"] = result
+        self.master.mq.produce(("builds", str(build['buildid']), event), copy.deepcopy(build))
 
     @defer.inlineCallbacks
     def run(self) -> Generator[Any, Any, None]:
+        """
+        This function implements a relatively simple scheduling algorithm. At the start we compute the
+        interdependencies between each of the jobs we want to run and at every iteration we schedule those
+        who's dependencies have completed successfully. If a job fails, we recursively fail every jobs which
+        depends on it.
+        We also run fake builds for failed evaluations so that they nicely show up in the UI and also Forge
+        reports.
+        """
         if self.combine_builds:
-            self.produceEvent("started-nix-build", None)
+            self.produceEventForBuildById("started-nix-build", self.build.buildid, None)
+
+        done: list[BuildTrigger.DoneJob] = []
+        scheduled: list[BuildTrigger.ScheduledJob] = []
 
         self.running = True
         build_props = self.build.getProperties()
@@ -384,15 +430,19 @@ def run(self) -> Generator[Any, Any, None]:
 
         # inject failed buildsteps for any failed eval jobs we got
         overall_result = SUCCESS if not self.failed_jobs else util.FAILURE
+        # inject failed buildsteps for any failed eval jobs we got
         if self.failed_jobs:
             scheduler_log.addStdout("The following jobs failed to evaluate:\n")
-        for failed_job in self.failed_jobs:
-            scheduler_log.addStdout(f"\t- {failed_job.attr} failed eval\n")
-            brids, _ = yield self.schedule(
-                ss_for_trigger,
-                *self.schedule_eval_failure(failed_job),
-            )
-            self.brids.extend(brids)
+            for failed_job in self.failed_jobs:
+                scheduler_log.addStdout(f"\t- {failed_job.attr} failed eval\n")
+                brids, results_deferred = yield self.schedule(
+                    ss_for_trigger,
+                    *self.schedule_eval_failure(failed_job),
+                )
+                scheduled.append(
+                    BuildTrigger.ScheduledJob(failed_job, brids, results_deferred)
+                )
+                self.brids.extend(brids)
 
         # get all job derivations
         job_set = {job.drvPath for job in self.successful_jobs}
@@ -411,8 +461,6 @@ def run(self) -> Generator[Any, Any, None]:
             self.successful_jobs, job_closures
         )
 
-        done: list[BuildTrigger.DoneJob] = []
-        scheduled: list[BuildTrigger.ScheduledJob] = []
         while build_schedule_order or scheduled:
             scheduler_log.addStdout("Scheduling...\n")
 
@@ -437,8 +485,6 @@ def run(self) -> Generator[Any, Any, None]:
                     scheduled.append(
                         BuildTrigger.ScheduledJob(build, brids, results_deferred)
                     )
-                    if not self.combine_builds:
-                        self.produceEventForBuilds(brids.values(), "started-nix-build", None)
                     self.brids.extend(brids.values())
                 elif failed_build is not None and self.build.reason == "rebuild":
                     failed_builds.remove_build(build.drvPath)
@@ -467,8 +513,6 @@ def run(self) -> Generator[Any, Any, None]:
                     BuildTrigger.ScheduledJob(job, brids, results_deferred)
                 )
 
-                if not self.combine_builds:
-                    self.produceEventForBuilds(brids.values(), "started-nix-build", None)
                 self.brids.extend(brids.values())
 
             scheduler_log.addStdout("Waiting...\n")
@@ -491,46 +535,49 @@ def run(self) -> Generator[Any, Any, None]:
             scheduler_log.addStdout(
                 f"Found finished build {job.attr}, result {util.Results[result].upper()}\n"
             )
+            log.info(
+                f"Found finished build {job.attr}, result {util.Results[result].upper()}, brids: {brids}"
+            )
             if not self.combine_builds:
-                self.produceEventForBuilds(brids.values(), "finished-nix-build", result)
+                self.produceEventForBuildRequestsById(brids.values(), "finished-nix-build", result)
 
             # if it failed, remove all dependent jobs, schedule placeholders and add them to the list of scheduled jobs
-            if result != SUCCESS:
-                failed_builds.add_build(job.drvPath, datetime.now(tz=UTC))
+            if isinstance(job, NixEvalJobSuccess):
+                if result != SUCCESS:
+                    failed_builds.add_build(job.drvPath, datetime.now(tz=UTC))
 
-                removed = self.get_failed_dependents(
-                    job, build_schedule_order, job_closures
-                )
-                for removed_job in removed:
-                    scheduler, props = self.schedule_dependency_failed(
-                        removed_job, job
-                    )
-                    brids, results_deferred = yield self.schedule(
-                        ss_for_trigger, scheduler, props
+                    removed = self.get_failed_dependents(
+                        job, build_schedule_order, job_closures
                     )
-                    build_schedule_order.remove(removed_job)
-                    scheduled.append(
-                        BuildTrigger.ScheduledJob(removed_job, brids, results_deferred)
+                    for removed_job in removed:
+                        scheduler, props = self.schedule_dependency_failed(
+                            removed_job, job
+                        )
+                        brids, results_deferred = yield self.schedule(
+                            ss_for_trigger, scheduler, props
+                        )
+                        build_schedule_order.remove(removed_job)
+                        scheduled.append(
+                            BuildTrigger.ScheduledJob(removed_job, brids, results_deferred)
+                        )
+                        self.brids.extend(brids.values())
+                    scheduler_log.addStdout(
+                        "\t- removed jobs: "
+                        + ", ".join([job.drvPath for job in removed])
+                        + "\n"
                     )
-                    if not self.combine_builds:
-                        self.produceEventForBuilds(brids.values(), "started-nix-build", None)
-                    self.brids.extend(brids.values())
-                scheduler_log.addStdout(
-                    "\t- removed jobs: "
-                    + ", ".join([job.drvPath for job in removed])
-                    + "\n"
-                )
+
+                for dep in job_closures:
+                    if job.drvPath in job_closures[dep]:
+                        job_closures[dep].remove(job.drvPath)
 
             overall_result = worst_status(result, overall_result)
             scheduler_log.addStdout(
                 f"\t- new result: {util.Results[overall_result].upper()} \n"
             )
 
-            for dep in job_closures:
-                if job.drvPath in job_closures[dep]:
-                    job_closures[dep].remove(job.drvPath)
         if self.combine_builds:
-            self.produceEvent("finished-nix-build", overall_result)
+            self.produceEventForBuildById("finished-nix-build", self.build.buildid, overall_result)
         scheduler_log.addStdout("Done!\n")
         return overall_result
 

buildbot_nix/gitea_projects.py

@@ -5,6 +5,7 @@
 from typing import Any
 from urllib.parse import urlparse
 
+from buildbot.interfaces import IReportGenerator
 from buildbot.config.builder import BuilderConfig
 from buildbot.plugins import util
 from buildbot.process.properties import Interpolate
@@ -29,6 +30,7 @@
     paginated_github_request,
     slugify_project_name,
 )
+from .nix_status_generator import BuildNixEvalStatusGenerator
 from .models import GiteaConfig
 from .projects import GitBackend, GitProject
 
@@ -107,36 +109,6 @@ def belongs_to_org(self) -> bool:
         # TODO Gitea doesn't include this information
         return False  # self.data["owner"]["type"] == "Organization"
 
-
-class ModifyingGiteaStatusPush(GiteaStatusPush):
-    def checkConfig(
-        self,
-        modifyingFilter: Callable[[Any], Any | None] = lambda x: x,  # noqa: N803
-        **kwargs: Any,
-    ) -> Any:
-        self.modifyingFilter = modifyingFilter
-
-        return super().checkConfig(**kwargs)
-
-    def reconfigService(
-        self,
-        modifyingFilter: Callable[[Any], Any | None] = lambda x: x,  # noqa: N803
-        **kwargs: Any,
-    ) -> Any:
-        self.modifyingFilter = modifyingFilter
-
-        return super().reconfigService(**kwargs)
-
-    @defer.inlineCallbacks
-    def sendMessage(self, reports: Any) -> Any:
-        reports = self.modifyingFilter(reports)
-        if reports is None:
-            return
-
-        result = yield super().sendMessage(reports)
-        return result
-
-
 class GiteaBackend(GitBackend):
     config: GiteaConfig
     webhook_secret: str
@@ -165,12 +137,14 @@ def create_reload_builder(self, worker_names: list[str]) -> BuilderConfig:
         )
 
     def create_reporter(self) -> ReporterBase:
-        return ModifyingGiteaStatusPush(
+        return GiteaStatusPush(
             baseURL=self.config.instance_url,
             token=Interpolate(self.config.token),
             context=Interpolate("buildbot/%(prop:status_name)s"),
             context_pr=Interpolate("buildbot/%(prop:status_name)s"),
-            modifyingFilter=filter_for_combined_builds,
+            generators = [
+                BuildNixEvalStatusGenerator(),
+            ],
         )
 
     def create_change_hook(self) -> dict[str, Any]: