murdock: implement numeric priorities

murdock/config.py

@@ -138,6 +138,11 @@ class TaskSettings(BaseSettings):
     env: dict = Field(default=dict())
 
 
+class PrioritySettings(BaseSettings):
+    labels: dict[str, int] = Field(default={})
+    branches: dict[str, int] = Field(default={})
+
+
 class MurdockSettings(BaseSettings):
     push: PushSettings = Field(default=PushSettings())
     pr: PRSettings = Field(default=PRSettings())
@@ -146,6 +151,7 @@ class MurdockSettings(BaseSettings):
     failfast: bool = Field(default=False)
     artifacts: List[str] = Field(default=[])
     tasks: List[TaskSettings] = Field(default=[TaskSettings()])
+    priorities: PrioritySettings = Field(default=PrioritySettings())
 
 
 _ENV_FILE = os.getenv("ENV_FILE", os.path.join(os.path.dirname(__file__), "..", ".env"))

murdock/job.py

@@ -16,6 +16,7 @@
 
 
 UNSAFE_ENVS = ["CI_JOB_TOKEN", "CI_SCRIPTS_DIR"]
+JOB_FASTTRACK_BONUS = 100
 
 
 class MurdockJob:
@@ -51,6 +52,9 @@ def __init__(
             if self.pr is not None
             else False
         )  # type: ignore[union-attr]
+        self.priority: int = MurdockJob.calculate_priority(
+            config, pr, ref, self.fasttracked
+        )
         self.artifacts: Optional[List[str]] = None
         self.token: str = secrets.token_urlsafe(32)
         self.scripts_dir: str = GLOBAL_CONFIG.scripts_dir
@@ -70,6 +74,27 @@ def __init__(
             self._logger_context["pr"] = str(self.pr.number)
         self._logger = LOGGER.bind(**self._logger_context)
 
+    @staticmethod
+    def calculate_priority(
+        config: MurdockSettings = MurdockSettings(),
+        pr: Optional[PullRequestInfo] = None,
+        ref: Optional[str] = None,
+        fasttracked: bool = False,
+    ):
+        priority = 0
+
+        if fasttracked:
+            priority += JOB_FASTTRACK_BONUS
+
+        if config.priorities:
+            if pr and config.priorities.labels:
+                for label in pr.labels:
+                    priority += config.priorities.labels.get(label, 0)
+            elif ref and config.priorities.branches:
+                priority += config.priorities.branches.get(ref, 0)
+
+        return priority
+
     @staticmethod
     def create_dir(work_dir: str) -> None:
         logger = LOGGER.bind(dir=str(work_dir))
@@ -137,6 +162,7 @@ def model(self) -> JobModel:
             prinfo=self.pr,
             ref=self.ref,
             fasttracked=self.fasttracked,
+            priority=self.priority,
             trigger=self.trigger,
             triggered_by=self.triggered_by,
             env=self.safe_env,
@@ -228,6 +254,15 @@ def __eq__(self, other) -> bool:
     def __hash__(self) -> int:
         return hash(self.uid)
 
+    def __lt__(self, other):
+        """
+        Helper to enable putting MurdockJob into PriorityQueue.
+
+        The logic is turned around here, as PriorityQueue returns *smallest*
+        value, but we prefer "higher priority value == higher priority"
+        """
+        return self.priority >= other.priority
+
     def set_start_time(self, start_time: datetime):
         if start_time.tzinfo is None:
             raise ValueError("Incomplete time object, no time zone defined")

murdock/models.py

@@ -104,6 +104,10 @@ class JobModel(BaseModel):
         None,
         title="Whether the job can be fasttracked",
     )
+    priority: int = Field(
+        0,
+        title="Job priority",
+    )
     status: Optional[dict] = Field(
         None,
         title="Status of the job",
@@ -155,6 +159,7 @@ class ManualJobParamModel(BaseModel):
     fasttrack: Optional[bool] = Field(
         default=False, title="Run the job in fasttrack mode."
     )
+    priority: Optional[int] = Field(default=0, title="Change job priority.")
 
 
 class ManualJobBranchParamModel(ManualJobParamModel):

murdock/murdock.py

@@ -75,7 +75,6 @@ def __init__(
         self.queued: MurdockJobList = MurdockJobList()
         self.running: MurdockJobPool = MurdockJobPool(num_workers)
         self.queue: asyncio.Queue = asyncio.Queue()
-        self.fasttrack_queue: asyncio.Queue = asyncio.Queue()
         self.notifier = Notifier()
         self.instrumentator = Instrumentator()
         self.db = database(database_type)
@@ -204,18 +203,13 @@ async def job_processing_task(self):
         structlog.contextvars.bind_contextvars(worker=current_task)
         LOGGER.debug("Starting worker")
         while True:
-            if self.fasttrack_queue.qsize():
-                job = self.fasttrack_queue.get_nowait()
+            try:
+                job = await self.queue.get()
                 await self._process_job(job)
-                self.fasttrack_queue.task_done()
-            else:
-                try:
-                    job = await self.queue.get()
-                    await self._process_job(job)
-                    self.queue.task_done()
-                except RuntimeError as exc:
-                    LOGGER.warning("Exiting worker", exception=str(exc))
-                    break
+                self.queue.task_done()
+            except RuntimeError as exc:
+                LOGGER.warning(f"Exiting worker {current_task}: {exc}")
+                break
 
     async def job_prepare(self, job: MurdockJob):
         logger = LOGGER.bind(**job.logging_context)
@@ -291,14 +285,13 @@ async def add_job_to_queue(self, job: MurdockJob):
                 "target_url": self.base_url,
             },
         )
-        all_busy = all(running is not None for running in self.running.jobs)
         self.queued.add(job)
         job.state = "queued"
-        if all_busy and job.fasttracked:
-            self.fasttrack_queue.put_nowait(job)
-        else:
-            self.queue.put_nowait(job)
-        logger.info("Job added to queued jobs")
+
+        self.queue.put_nowait(job)
+
+        LOGGER.info(f"{job} added to queued jobs")
+
         self.job_queue_counter.inc()
         await self.reload_jobs()
 
@@ -538,23 +531,36 @@ async def handle_pull_request_event(self, event: dict):
             await set_commit_status(job.commit.sha, status)
             return
 
-        if action in ["unlabeled", "edited"]:
+        if action in ["edited"]:
             return
 
+        if action in ["unlabeled"]:
+            label_name = event["label"]["name"]
+            if label_name not in config.priorities.labels.keys():
+                return
+            elif not self.queued.search_by_pr_number(pull_request.number):
+                # skip re-queing if there's no queued job
+                return
+
         if action == "opened" and CI_CONFIG.ready_label in pull_request.labels:
             # A PR opened with "Ready label" already set will be scheduled via
             # the "labeled" action
             return
 
         if action == "labeled":
-            if event["label"]["name"] != CI_CONFIG.ready_label:
+            # return if the ready label was set for an already queued job,
+            # or if the label wouldn't change the job's priority (and thus
+            # queue position).
+            # otherwise, fall through to re-scheduling the job.
+            label_name = event["label"]["name"]
+            if label_name == CI_CONFIG.ready_label:
+                # Skip already queued jobs for ready label
+                if self.queued.search_by_pr_number(pull_request.number):
+                    return
+            elif label_name not in config.priorities.labels.keys():
                 return
-            # Skip already queued jobs
-            if event["label"][
-                "name"
-            ] == CI_CONFIG.ready_label and self.queued.search_by_pr_number(
-                pull_request.number
-            ):
+            elif not self.queued.search_by_pr_number(pull_request.number):
+                # skip re-queing if there's no queued job
                 return
 
         await self.schedule_job(job)
@@ -640,13 +646,17 @@ async def reload_jobs(self):
         await self.notify_message_to_clients(json.dumps({"cmd": "reload"}))
 
     def get_queued_jobs(self, query: JobQueryModel = JobQueryModel()) -> List[JobModel]:
-        return sorted(
-            [
-                job.model()
-                for job in self.queued.search_with_query(query)
-                if query.states is None or "queued" in query.states
-            ],
-            key=lambda job: job.fasttracked,  # type: ignore[return-value,arg-type]
+        return list(
+            map(
+                lambda job: job.model(),
+                sorted(
+                    [
+                        job
+                        for job in self.queued.search_with_query(query)
+                        if query.states is None or "queued" in query.states
+                    ]
+                ),
+            )
         )
 
     def get_running_jobs(