Skip to content

Commit d7bec9d

Browse files
jangraevell-pdrickbrouwerCopilot
authored
GitHub Runner Scaler: Respect paging for jobs (#7973)
* Add failing test Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Fetch multiple pages if available Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Add Changgelog-entry Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Update CHANGELOG.md Signed-off-by: Rick Brouwer <rickbrouwer@gmail.com> * Cache job-pages instead of Jobs, so we can make sure, that the caching can be used for all pages instead of only the first one. Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Merge-fixes Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Add a maximum for pages to fetch Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Update CHANGELOG.md Signed-off-by: Rick Brouwer <rickbrouwer@gmail.com> * GitHub Runner Scaler: Add optional maxPages trigger parameter and raise pagination log level Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * GitHub Runner Scaler: Fix test build, validate maxPages, add changelog entry - Add missing githubDefaultMaxPages constant referenced by tests (test package failed to compile without it). - Validate that maxPages is >= 1 in parseGitHubRunnerMetadata so a misconfigured value fails fast instead of silently limiting pagination to a single page. - Add a CHANGELOG.md entry for #7969 under Fixes. - Add a regression test for the maxPages validation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Remove max-pages truncation for repositories Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> * Update CHANGELOG.md Signed-off-by: Rick Brouwer <rickbrouwer@gmail.com> --------- Signed-off-by: Jan Grävell <jan.graevell@prodyna.com> Signed-off-by: Jan Grävell <141423264+jangraevell-pd@users.noreply.github.com> Signed-off-by: Rick Brouwer <rickbrouwer@gmail.com> Co-authored-by: Rick Brouwer <rickbrouwer@gmail.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 94fc9f3 commit d7bec9d

4 files changed

Lines changed: 487 additions & 85 deletions

File tree

‎pkg/scalers/github_runner_scaler.go‎

Lines changed: 102 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -27,27 +27,31 @@ import (
2727
)
2828

2929
const (
30-
ORG = "org"
31-
ENT = "ent"
32-
REPO = "repo"
33-
githubDefaultPerPage = 30
34-
// githubScalerMaxCacheEntries caps the etags, previousJobs, and
35-
// previousWfrs maps. Without it the etags map grows once per workflow run
36-
// for the lifetime of the operator pod (the URL contains the run ID), and
37-
// previousJobs and previousWfrs grow once per distinct repository name
38-
// (previousJobs once per distinct (repository, run ID) pair) returned by
39-
// the API.
30+
ORG = "org"
31+
ENT = "ent"
32+
REPO = "repo"
33+
githubDefaultPerPage = 30
34+
githubJobsPerPage = 100
35+
githubDefaultMaxPages = 50
36+
// githubScalerMaxCacheEntries caps the etags, previousJobPages, and
37+
// previousWfrs maps. Without it the etags and previousJobPages maps grow
38+
// once per workflow run page for the lifetime of the operator pod (the URL
39+
// contains the run ID and page number), and previousWfrs grows once per
40+
// distinct repository name returned by the API.
4041
githubScalerMaxCacheEntries = 5000
4142
)
4243

4344
var reservedLabels = []string{"self-hosted", "linux", "x64"}
4445

45-
// jobCacheKey identifies a single workflow run's job list within
46-
// previousJobs. A repository can have several runs queued/in_progress at
47-
// once, so the run ID is required in addition to the repository name.
46+
// jobCacheKey identifies a single page of a single workflow run's job list
47+
// within previousJobPages. A repository can have several runs
48+
// queued/in_progress at once, and a run can have more jobs than fit on a
49+
// single page, so the run ID and page number are required in addition to the
50+
// repository name.
4851
type jobCacheKey struct {
4952
repo string
5053
runID int64
54+
page int
5155
}
5256

5357
type githubRunnerScaler struct {
@@ -60,7 +64,7 @@ type githubRunnerScaler struct {
6064
etags map[string]string
6165
previousRepos []string
6266
previousWfrs map[string]map[string]*WorkflowRuns
63-
previousJobs map[jobCacheKey][]Job
67+
previousJobPages map[jobCacheKey][]Job
6468
rateLimit RateLimit
6569
previousQueueLength int64
6670
previousQueueLengthTime time.Time
@@ -77,6 +81,7 @@ type githubRunnerMetadata struct {
7781
EnableEtags bool `keda:"name=enableEtags, order=triggerMetadata;resolvedEnv, default=false"`
7882
MatchUnlabeledJobsWithUnlabeledRunners bool `keda:"name=matchUnlabeledJobsWithUnlabeledRunners, order=triggerMetadata;resolvedEnv, default=false"`
7983
TargetWorkflowQueueLength int64 `keda:"name=targetWorkflowQueueLength, order=triggerMetadata;resolvedEnv, default=1"`
84+
MaxPages int `keda:"name=maxPages, order=triggerMetadata;resolvedEnv, default=50"`
8085
TriggerIndex int
8186
ApplicationID int64 `keda:"name=applicationID, order=triggerMetadata;resolvedEnv, optional"`
8287
InstallationID int64 `keda:"name=installationID, order=triggerMetadata;resolvedEnv, optional"`
@@ -393,7 +398,7 @@ func NewGitHubRunnerScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
393398

394399
etags := make(map[string]string)
395400
previousRepos := []string{}
396-
previousJobs := make(map[jobCacheKey][]Job)
401+
previousJobPages := make(map[jobCacheKey][]Job)
397402
previousWfrs := make(map[string]map[string]*WorkflowRuns)
398403
rateLimit := RateLimit{}
399404
previousQueueLength := int64(0)
@@ -408,7 +413,7 @@ func NewGitHubRunnerScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
408413
scaledObject: config.ScaledObject,
409414
etags: etags,
410415
previousRepos: previousRepos,
411-
previousJobs: previousJobs,
416+
previousJobPages: previousJobPages,
412417
previousWfrs: previousWfrs,
413418
rateLimit: rateLimit,
414419
previousQueueLength: previousQueueLength,
@@ -477,6 +482,10 @@ func parseGitHubRunnerMetadata(config *scalersconfig.ScalerConfig) (*githubRunne
477482

478483
meta.TriggerIndex = config.TriggerIndex
479484

485+
if meta.MaxPages < 1 {
486+
return nil, fmt.Errorf("maxPages must be greater than or equal to 1, got %d", meta.MaxPages)
487+
}
488+
480489
return meta, nil
481490
}
482491

@@ -676,29 +685,38 @@ func stripDeadRuns(allWfrs []WorkflowRuns) []WorkflowRun {
676685

677686
// jobsAPIURL returns the GitHub API URL for a workflow run's jobs, used both
678687
// to fetch the jobs and as the etags cache key for that same request.
679-
func (s *githubRunnerScaler) jobsAPIURL(repoName string, workflowRunID int64) string {
680-
return fmt.Sprintf("%s/repos/%s/%s/actions/runs/%d/jobs?per_page=100",
688+
func (s *githubRunnerScaler) jobsAPIURL(repoName string, workflowRunID int64, page int) string {
689+
apiURL := fmt.Sprintf("%s/repos/%s/%s/actions/runs/%d/jobs?per_page=%d",
681690
s.metadata.GithubAPIURL,
682691
url.PathEscape(s.metadata.Owner),
683692
url.PathEscape(repoName),
684-
workflowRunID)
693+
workflowRunID,
694+
githubJobsPerPage)
695+
if page > 1 {
696+
apiURL = fmt.Sprintf("%s&page=%d", apiURL, page)
697+
}
698+
return apiURL
685699
}
686700

687-
// getWorkflowRunJobs returns a list of jobs for a given workflow run
688-
func (s *githubRunnerScaler) getWorkflowRunJobs(ctx context.Context, workflowRunID int64, repoName string) ([]Job, error) {
689-
apiURL := s.jobsAPIURL(repoName, workflowRunID)
701+
// fetchWorkflowRunJobsPage fetches a single page of jobs for a workflow run.
702+
// Each page is cached and validated against GitHub independently, keyed by
703+
// its own URL (which encodes both the workflow run and the page number). A
704+
// 304 on one page never assumes that other pages are also unchanged.
705+
func (s *githubRunnerScaler) fetchWorkflowRunJobsPage(ctx context.Context, workflowRunID int64, repoName string, page int) ([]Job, error) {
706+
apiURL := s.jobsAPIURL(repoName, workflowRunID, page)
690707

691708
body, statusCode, err := s.getGithubRequest(ctx, apiURL, s.metadata, s.httpClient)
692709
if err != nil {
693710
return nil, err
694711
}
695-
key := jobCacheKey{repo: repoName, runID: workflowRunID}
712+
key := jobCacheKey{repo: repoName, runID: workflowRunID, page: page}
696713
if statusCode == 304 && s.metadata.EnableEtags {
697-
if jobs, ok := s.previousJobs[key]; ok {
698-
return jobs, nil
714+
if cachedJobs, ok := s.previousJobPages[key]; ok {
715+
return cachedJobs, nil
699716
}
700-
// Stale etag without a paired previousJobs entry, e.g. after pruneCaches
701-
// evicted the previous entry. Drop the etag and retry as a cache miss.
717+
// Stale etag without a paired previousJobPages entry, e.g. after
718+
// pruneCaches evicted the previous entry. Drop the etag and retry as
719+
// a cache miss.
702720
delete(s.etags, apiURL)
703721
body, statusCode, err = s.getGithubRequest(ctx, apiURL, s.metadata, s.httpClient)
704722
if err != nil {
@@ -709,17 +727,50 @@ func (s *githubRunnerScaler) getWorkflowRunJobs(ctx context.Context, workflowRun
709727
}
710728
}
711729

712-
var jobs Jobs
713-
err = json.Unmarshal(body, &jobs)
714-
if err != nil {
730+
var parsed Jobs
731+
if err := json.Unmarshal(body, &parsed); err != nil {
715732
return nil, err
716733
}
717734

718735
if s.metadata.EnableEtags {
719-
s.previousJobs[key] = jobs.Jobs
736+
s.previousJobPages[key] = parsed.Jobs
720737
}
721738

722-
return jobs.Jobs, nil
739+
return parsed.Jobs, nil
740+
}
741+
742+
// getWorkflowRunJobs returns a list of jobs for a given workflow run. A workflow
743+
// run can have more jobs than fit on a single page, so all pages are fetched
744+
// and combined until GitHub returns a page with fewer than githubJobsPerPage jobs,
745+
// or until the maxPages parameter is reached.
746+
// When ETags are enabled, every page is independently fetched and validated,
747+
// so a 304 on an earlier page never short-circuits the fetch of later pages.
748+
func (s *githubRunnerScaler) getWorkflowRunJobs(ctx context.Context, workflowRunID int64, repoName string) ([]Job, error) {
749+
var allJobs []Job
750+
page := 1
751+
752+
for {
753+
jobs, err := s.fetchWorkflowRunJobsPage(ctx, workflowRunID, repoName, page)
754+
if err != nil {
755+
return nil, err
756+
}
757+
758+
allJobs = append(allJobs, jobs...)
759+
760+
// GitHub returned fewer than a full page, so there are no more jobs left to fetch
761+
if len(jobs) < githubJobsPerPage {
762+
break
763+
}
764+
765+
if page >= s.metadata.MaxPages {
766+
s.logger.Info(fmt.Sprintf("Reached max page limit (%d) while fetching jobs for workflow run %d in repo %s, results may be incomplete", s.metadata.MaxPages, workflowRunID, repoName))
767+
break
768+
}
769+
770+
page++
771+
}
772+
773+
return allJobs, nil
723774
}
724775

725776
// getWorkflowRuns returns a list of workflow runs for a given repository
@@ -849,22 +900,29 @@ func (s *githubRunnerScaler) pruneCaches(currentRepos []string) {
849900
}
850901
}
851902
evictExcess(s.previousWfrs, githubScalerMaxCacheEntries)
852-
evictExcess(s.previousJobs, githubScalerMaxCacheEntries)
903+
evictExcess(s.previousJobPages, githubScalerMaxCacheEntries)
853904
evictExcess(s.etags, githubScalerMaxCacheEntries)
854905
}
855906

856-
// pruneCompletedJobs removes previousJobs entries for runs that are no
857-
// longer queued/in_progress, so a completed run's cached job list is not
858-
// held indefinitely waiting for size-based eviction in pruneCaches.
907+
// pruneCompletedJobs removes previousJobPages and etags entries for job
908+
// pages belonging to runs that are no longer queued/in_progress, so a
909+
// completed run's cached job pages are not held indefinitely waiting for
910+
// size-based eviction in pruneCaches. previousJobPages is keyed per page
911+
// while activeWfrs has no notion of pages, so each cached (repo, runID,
912+
// page) entry is matched against activeWfrs by (repo, runID) alone.
859913
func (s *githubRunnerScaler) pruneCompletedJobs(activeWfrs []WorkflowRun) {
860-
active := make(map[jobCacheKey]struct{}, len(activeWfrs))
861-
for _, wfr := range activeWfrs {
862-
active[jobCacheKey{repo: wfr.Repository.Name, runID: wfr.ID}] = struct{}{}
863-
}
864-
for key := range s.previousJobs {
865-
if _, ok := active[key]; !ok {
866-
delete(s.previousJobs, key)
867-
delete(s.etags, s.jobsAPIURL(key.repo, key.runID))
914+
isActive := func(repo string, runID int64) bool {
915+
for _, wfr := range activeWfrs {
916+
if wfr.Repository.Name == repo && wfr.ID == runID {
917+
return true
918+
}
919+
}
920+
return false
921+
}
922+
for key := range s.previousJobPages {
923+
if !isActive(key.repo, key.runID) {
924+
delete(s.previousJobPages, key)
925+
delete(s.etags, s.jobsAPIURL(key.repo, key.runID, key.page))
868926
}
869927
}
870928
}

0 commit comments

Comments
 (0)