branch_worker: introduce can_do_sync() check

Github has a limit on the number of tokens used by an app, which is
5k/1h for a free-tier org [1].

In BPF CI due to number of pending PRs, we regularly get very close to
hitting the limit (remaining < 500) and have relevant alerts set up.

Introduce can_do_sync() method to BranchWorker that will check if
there is enough remaining tokens to run through the
sync_patches. *Enough* is defined by MIN_REMAINING_GITHUB_TOKENS,
which is currently a constant for simplicity.

In case the number of remaining tokens is too low, the sync_patches
will simply be skipped for the worker. Effectively KPD will wait until
github will refresh the limit without crashing.

[1] https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api?apiVersion=2022-11-28#primary-rate-limit-for-authenticated-users

Signed-off-by: Ihor Solodrai <[email protected]>

Author: theihor

kernel_patches_daemon/branch_worker.py

@@ -39,11 +39,10 @@
 
 from kernel_patches_daemon.config import (
     EmailConfig,
-    SERIES_ID_SEPARATOR,
     SERIES_TARGET_SEPARATOR,
 )
 from kernel_patches_daemon.github_connector import GithubConnector
-from kernel_patches_daemon.github_logs import GithubFailedJobLog, GithubLogExtractor
+from kernel_patches_daemon.github_logs import GithubLogExtractor
 from kernel_patches_daemon.patchwork import Patchwork, Series, Subject
 from kernel_patches_daemon.stats import HistogramMetricTimer
 from kernel_patches_daemon.status import (
@@ -94,6 +93,12 @@
 MERGE_CONFLICT_LABEL = "merge-conflict"
 UPSTREAM_REMOTE_NAME = "upstream"
 
+# We get 5k tokens per hour. When this value is checked, we should be
+# able to do at least one more sync loop. Depending on number of PRs,
+# one iteration can use more or less tokens, but generally BPF CI uses
+# 4k in 10-15 iterations, so 1k should be enough almost always.
+MIN_REMAINING_GITHUB_TOKENS = 1000
+
 # fmt: off
 EMAIL_TEMPLATE_BASE: Final[str] = """\
 Dear patch submitter,
@@ -649,6 +654,12 @@ def update_e2e_test_branch_and_update_pr(
 
         self._update_e2e_pr(title, base_branch, branch_name, pushed)
 
+    def can_do_sync(self) -> bool:
+        github_ratelimit = self.git.get_rate_limit()
+        if github_ratelimit.core.remaining < MIN_REMAINING_GITHUB_TOKENS:
+            return False
+        return True
+
     def do_sync(self) -> None:
         # fetch most recent upstream
         if UPSTREAM_REMOTE_NAME in [x.name for x in self.repo_local.remotes]:

kernel_patches_daemon/github_sync.py

@@ -25,7 +25,6 @@
 from kernel_patches_daemon.config import (
     BranchConfig,
     KPDConfig,
-    SERIES_TARGET_SEPARATOR,
 )
 from kernel_patches_daemon.github_logs import (
     BpfGithubLogExtractor,
@@ -280,13 +279,22 @@ async def sync_patches(self) -> None:
         as separate commit
         """
 
+        sync_workers = [
+            (branch, worker)
+            for (branch, worker) in self.workers.items()
+            if worker.can_do_sync()
+        ]
+        if not sync_workers:
+            logger.warn("No branch workers that can_do_sync(), skipping sync_patches()")
+            return
+
         # sync mirror and fetch current states of PRs
         loop = asyncio.get_event_loop()
 
         self.drop_counters()
         sync_start = time.time()
 
-        for branch, worker in self.workers.items():
+        for branch, worker in sync_workers:
             logging.info(f"Refreshing repo info for {branch}.")
             await loop.run_in_executor(None, worker.fetch_repo_branch)
             await loop.run_in_executor(None, worker.get_pulls)
@@ -298,7 +306,7 @@ async def sync_patches(self) -> None:
         mirror_done = time.time()
 
         with HistogramMetricTimer(patchwork_fetch_duration):
-            for branch, worker in self.workers.items():
+            for branch, worker in sync_workers:
                 await loop.run_in_executor(
                     None, worker.update_e2e_test_branch_and_update_pr, branch
                 )
@@ -313,7 +321,7 @@ async def sync_patches(self) -> None:
 
         # sync old subjects
         subject_names = {x.subject for x in self.subjects}
-        for worker in self.workers.values():
+        for _, worker in sync_workers:
             for subject_name, pr in worker.prs.items():
                 if subject_name in subject_names:
                     continue
@@ -359,7 +367,7 @@ async def sync_patches(self) -> None:
         self.set_counter("mirror_duration", mirror_done - sync_start)
         self.set_counter("pw_fetch_duration", pw_done - mirror_done)
         self.set_counter("patch_and_update_duration", patches_done - pw_done)
-        for worker in self.workers.values():
+        for _, worker in sync_workers:
             for pr in worker.prs.values():
                 if worker._is_relevant_pr(pr):
                     self.increment_counter("prs_total")

tests/test_github_sync.py

@@ -85,6 +85,10 @@ def setUp(self) -> None:
         self.addCleanup(patcher.stop)
 
         self._gh = GithubSyncMock()
+        for worker in self._gh.workers.values():
+            rate_limit = MagicMock()
+            rate_limit.core.remaining = 5000
+            worker.git.get_rate_limit = MagicMock(return_value=rate_limit)
 
     def test_init_with_base_directory(self) -> None:
         @dataclass