From 0d3c5cb4dbde9102533dfccd6f68ee6fcebcfac1 Mon Sep 17 00:00:00 2001 From: Adam Harwell Date: Tue, 11 Aug 2026 05:06:33 -0400 Subject: [PATCH] fix(scaleset): adopt existing GitHub scale sets Recover scale sets before API creation and after duplicate-create races. Match exact names in group-scoped results even when GitHub omits the runner-group ID. --- runner/scalesets.go | 24 +- runner/scalesets_test.go | 184 +++++++++++++++ util/github/scalesets/client.go | 52 ++++- util/github/scalesets/client_test.go | 52 +++++ util/github/scalesets/scalesets.go | 61 ++++- workers/scaleset/scaleset.go | 64 +++--- workers/scaleset/scaleset_adoption_test.go | 256 +++++++++++++++++++++ 7 files changed, 652 insertions(+), 41 deletions(-) create mode 100644 runner/scalesets_test.go create mode 100644 util/github/scalesets/client_test.go create mode 100644 workers/scaleset/scaleset_adoption_test.go diff --git a/runner/scalesets.go b/runner/scalesets.go index 3704f43f6..8911e84d5 100644 --- a/runner/scalesets.go +++ b/runner/scalesets.go @@ -261,13 +261,29 @@ func (r *Runner) CreateEntityScaleSet(ctx context.Context, entityType params.For Enabled: ¶m.Enabled, } - runnerScaleSet, err := scalesetCli.CreateRunnerScaleSet(ctx, createParam) - if err != nil { - return params.ScaleSet{}, fmt.Errorf("error creating runner scale set: %w", err) + runnerScaleSet, lookupErr := scalesetCli.GetRunnerScaleSetByNameAndRunnerGroup(ctx, int(runnerGroupID), param.Name) + created := false + if lookupErr != nil { + if !errors.Is(lookupErr, runnerErrors.ErrNotFound) { + return params.ScaleSet{}, fmt.Errorf("error finding runner scale set: %w", lookupErr) + } + + runnerScaleSet, err = scalesetCli.CreateRunnerScaleSet(ctx, createParam) + if err != nil { + if !errors.Is(err, scalesets.ErrRunnerScaleSetExists) { + return params.ScaleSet{}, fmt.Errorf("error creating runner scale set: %w", err) + } + runnerScaleSet, err = scalesetCli.GetRunnerScaleSetByNameAndRunnerGroup(ctx, int(runnerGroupID), param.Name) + if err != nil { + return params.ScaleSet{}, fmt.Errorf("error finding existing runner scale set: %w", err) + } + } else { + created = true + } } defer func() { - if err != nil { + if err != nil && created { if innerErr := scalesetCli.DeleteRunnerScaleSet(ctx, runnerScaleSet.ID); innerErr != nil { slog.With(slog.Any("error", innerErr)).ErrorContext(ctx, "failed to cleanup scale set") } diff --git a/runner/scalesets_test.go b/runner/scalesets_test.go new file mode 100644 index 000000000..7fbc17763 --- /dev/null +++ b/runner/scalesets_test.go @@ -0,0 +1,184 @@ +// Copyright 2026 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +//go:build testing + +package runner + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + commonParams "github.com/cloudbase/garm-provider-common/params" + "github.com/cloudbase/garm/auth" + storeMocks "github.com/cloudbase/garm/database/common/mocks" + "github.com/cloudbase/garm/params" +) + +const testScaleSetActionsToken = "eyJhbGciOiJub25lIn0.eyJleHAiOjQxNDk5MzYwMDB9." + +type scaleSetAPI struct { + server *httptest.Server + existing bool + createConflict bool + createRequests int + deleteRequests int +} + +func newScaleSetAPI(t *testing.T) *scaleSetAPI { + t.Helper() + + api := new(scaleSetAPI) + api.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/rate_limit": + _, _ = w.Write([]byte(`{"resources":{"core":{"limit":5000,"remaining":5000}}}`)) + case strings.HasSuffix(r.URL.Path, "/repos/owner/repo/actions/runners/registration-token"): + _, _ = fmt.Fprintf(w, `{"token":"registration-token","expires_at":%q}`, time.Now().Add(time.Hour).Format(time.RFC3339)) + case r.URL.Path == "/actions/runner-registration": + _, _ = fmt.Fprintf(w, `{"url":%q,"token":%q}`, api.server.URL, testScaleSetActionsToken) + case strings.HasPrefix(r.URL.Path, "/_apis/runtime/runnerscalesets"): + api.handleScaleSets(t, w, r) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(api.server.Close) + return api +} + +func (a *scaleSetAPI) handleScaleSets(t *testing.T, w http.ResponseWriter, r *http.Request) { + t.Helper() + + switch r.Method { + case http.MethodGet: + if a.existing { + _, _ = w.Write([]byte(`{"count":1,"value":[{"id":42,"name":"existing","runnerGroupId":1}]}`)) + return + } + _, _ = w.Write([]byte(`{"count":0,"value":[]}`)) + case http.MethodPost: + a.createRequests++ + if a.createConflict { + a.existing = true + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"Bad Request","details":"failed: \"{\\\"typeName\\\":\\\"GitHub.Actions.Runtime.WebApi.RunnerScaleSetExistsException, GitHub.Actions.Runtime.WebApi\\\"}\""}`)) + return + } + a.existing = true + _, _ = w.Write([]byte(`{"id":42,"name":"existing","runnerGroupId":1}`)) + case http.MethodDelete: + a.deleteRequests++ + w.WriteHeader(http.StatusNoContent) + default: + t.Errorf("unexpected scale-set request: %s", r.Method) + } +} + +func newScaleSetRunner(t *testing.T, api *scaleSetAPI, createErr error) (*Runner, context.Context) { + t.Helper() + + ctx := auth.GetAdminContext(context.Background()) + store := storeMocks.NewStore(t) + templateID := uint(1) + credentials, err := json.Marshal(params.GithubPAT{OAuth2Token: "token"}) + require.NoError(t, err) + entity := params.ForgeEntity{ + ID: "entity-id", + Owner: "owner", + Name: "repo", + EntityType: params.ForgeEntityTypeRepository, + Credentials: params.ForgeCredentials{ + APIBaseURL: api.server.URL + "/", + UploadBaseURL: api.server.URL + "/", + BaseURL: api.server.URL, + AuthType: params.ForgeAuthTypePAT, + ForgeType: params.GithubEndpointType, + CredentialsPayload: credentials, + }, + } + store.EXPECT().GetForgeEntity(ctx, params.ForgeEntityTypeRepository, entity.ID).Return(entity, nil).Once() + store.EXPECT().GetTemplate(ctx, templateID).Return(params.Template{ + ID: templateID, + OSType: commonParams.Linux, + ForgeType: params.GithubEndpointType, + }, nil).Once() + store.EXPECT().CreateEntityScaleSet(ctx, entity, mock.MatchedBy(func(param params.CreateScaleSetParams) bool { + return param.ScaleSetID == 42 + })).Return(params.ScaleSet{ScaleSetID: 42}, createErr).Once() + + return &Runner{store: store}, ctx +} + +func createExistingScaleSet(t *testing.T, runner *Runner, ctx context.Context) (params.ScaleSet, error) { + t.Helper() + templateID := uint(1) + return runner.CreateEntityScaleSet(ctx, params.ForgeEntityTypeRepository, "entity-id", params.CreateScaleSetParams{ + Name: "existing", + OSType: commonParams.Linux, + TemplateID: &templateID, + }) +} + +func TestCreateEntityScaleSetAdoptsExistingScaleSet(t *testing.T) { + api := newScaleSetAPI(t) + api.existing = true + runner, ctx := newScaleSetRunner(t, api, nil) + + scaleSet, err := createExistingScaleSet(t, runner, ctx) + require.NoError(t, err) + require.Equal(t, 42, scaleSet.ScaleSetID) + require.Zero(t, api.createRequests) +} + +func TestCreateEntityScaleSetRecoversCreateConflict(t *testing.T) { + api := newScaleSetAPI(t) + api.createConflict = true + runner, ctx := newScaleSetRunner(t, api, nil) + + scaleSet, err := createExistingScaleSet(t, runner, ctx) + require.NoError(t, err) + require.Equal(t, 42, scaleSet.ScaleSetID) + require.Equal(t, 1, api.createRequests) +} + +func TestCreateEntityScaleSetDoesNotDeleteAdoptedScaleSet(t *testing.T) { + api := newScaleSetAPI(t) + api.existing = true + runner, ctx := newScaleSetRunner(t, api, errors.New("database unavailable")) + + _, err := createExistingScaleSet(t, runner, ctx) + require.Error(t, err) + require.Zero(t, api.deleteRequests) +} + +func TestCreateEntityScaleSetDeletesCreatedScaleSetOnDatabaseFailure(t *testing.T) { + api := newScaleSetAPI(t) + runner, ctx := newScaleSetRunner(t, api, errors.New("database unavailable")) + + _, err := createExistingScaleSet(t, runner, ctx) + require.Error(t, err) + require.Equal(t, 1, api.createRequests) + require.Equal(t, 1, api.deleteRequests) +} diff --git a/util/github/scalesets/client.go b/util/github/scalesets/client.go index 9583d41a4..0824d82e1 100644 --- a/util/github/scalesets/client.go +++ b/util/github/scalesets/client.go @@ -15,9 +15,13 @@ package scalesets import ( + "encoding/json" + "errors" "fmt" "io" "net/http" + "strconv" + "strings" "sync" "github.com/google/go-github/v84/github" @@ -28,6 +32,48 @@ import ( "github.com/cloudbase/garm/runner/common" ) +// ErrRunnerScaleSetExists is returned when creating an existing runner scale set. +var ErrRunnerScaleSetExists = errors.New("runner scale set already exists") + +type actionsErrorResponse struct { + TypeName string `json:"typeName"` + Details string `json:"details"` +} + +func isRunnerScaleSetExistsType(typeName string) bool { + typeName, _, _ = strings.Cut(typeName, ",") + typeName = strings.TrimSpace(typeName) + if index := strings.LastIndexByte(typeName, '.'); index >= 0 { + typeName = typeName[index+1:] + } + return typeName == "RunnerScaleSetExistsException" +} + +func isRunnerScaleSetExists(body []byte) bool { + var response actionsErrorResponse + if err := json.Unmarshal(body, &response); err != nil { + return false + } + if isRunnerScaleSetExistsType(response.TypeName) { + return true + } + + start := strings.IndexByte(response.Details, '{') + end := strings.LastIndexByte(response.Details, '}') + if start < 0 || end < start { + return false + } + details := response.Details[start : end+1] + var nestedResponse actionsErrorResponse + if err := json.Unmarshal([]byte(details), &nestedResponse); err != nil { + details, err = strconv.Unquote(`"` + details + `"`) + if err != nil || json.Unmarshal([]byte(details), &nestedResponse) != nil { + return false + } + } + return isRunnerScaleSetExistsType(nestedResponse.TypeName) +} + func NewClient(cli common.GithubClient) (*ScaleSetClient, error) { return &ScaleSetClient{ ghCli: cli, @@ -108,7 +154,11 @@ func (s *ScaleSetClient) Do(req *http.Request) (*http.Response, error) { case 404: return nil, runnerErrors.NewNotFoundError("resource %s not found: %q", req.URL.String(), string(body)) case 400: - return nil, runnerErrors.NewBadRequestError("bad request while calling %s: %q", req.URL.String(), string(body)) + badRequest := runnerErrors.NewBadRequestError("bad request while calling %s: %q", req.URL.String(), string(body)) + if isRunnerScaleSetExists(body) { + return nil, fmt.Errorf("%w: %w", ErrRunnerScaleSetExists, badRequest) + } + return nil, badRequest case 409: return nil, runnerErrors.NewConflictError("conflict while calling %s: %q", req.URL.String(), string(body)) case 401, 403: diff --git a/util/github/scalesets/client_test.go b/util/github/scalesets/client_test.go new file mode 100644 index 000000000..9d2204c66 --- /dev/null +++ b/util/github/scalesets/client_test.go @@ -0,0 +1,52 @@ +// Copyright 2026 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package scalesets + +import ( + "errors" + "net/http" + "net/http/httptest" + "testing" + + runnerErrors "github.com/cloudbase/garm-provider-common/errors" +) + +func TestDoRecognizesNestedRunnerScaleSetExistsError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"Bad Request","details":"error creating runner scale set: failed: \"{\\\"typeName\\\":\\\"GitHub.Actions.Runtime.WebApi.RunnerScaleSetExistsException, GitHub.Actions.Runtime.WebApi\\\"}\""}`)) + })) + t.Cleanup(server.Close) + + client := &ScaleSetClient{httpClient: server.Client()} + req, err := http.NewRequest(http.MethodPost, server.URL, nil) + if err != nil { + t.Fatal(err) + } + + _, err = client.Do(req) + if !errors.Is(err, ErrRunnerScaleSetExists) { + t.Fatalf("expected ErrRunnerScaleSetExists, got %v", err) + } + if !errors.Is(err, runnerErrors.ErrBadRequest) { + t.Fatalf("expected ErrBadRequest, got %v", err) + } +} + +func TestRunnerScaleSetExistsTypeRequiresExactName(t *testing.T) { + if isRunnerScaleSetExistsType("NotRunnerScaleSetExistsException") { + t.Fatal("unexpected duplicate classification") + } +} diff --git a/util/github/scalesets/scalesets.go b/util/github/scalesets/scalesets.go index 896129995..3da42b8e6 100644 --- a/util/github/scalesets/scalesets.go +++ b/util/github/scalesets/scalesets.go @@ -20,6 +20,8 @@ import ( "encoding/json" "fmt" "net/http" + "net/url" + "strconv" runnerErrors "github.com/cloudbase/garm-provider-common/errors" "github.com/cloudbase/garm/params" @@ -30,6 +32,10 @@ const ( scaleSetEndpoint = "_apis/runtime/runnerscalesets" ) +func matchesRunnerScaleSet(scaleSet params.RunnerScaleSet, runnerGroupID int, name string) bool { + return scaleSet.Name == name && (scaleSet.RunnerGroupID == 0 || scaleSet.RunnerGroupID == int64(runnerGroupID)) +} + const ( HeaderActionsActivityID = "ActivityId" HeaderGitHubRequestID = "X-GitHub-Request-Id" @@ -43,7 +49,11 @@ func (s *ScaleSetClient) GetRunnerScaleSetByNameAndRunnerGroup(ctx context.Conte } }() - path := fmt.Sprintf("%s?runnerGroupId=%d&name=%s", scaleSetEndpoint, runnerGroupID, name) + query := url.Values{ + "runnerGroupId": []string{strconv.Itoa(runnerGroupID)}, + "name": []string{name}, + } + path := scaleSetEndpoint + "?" + query.Encode() req, err := s.newActionsRequest(ctx, http.MethodGet, path, nil) if err != nil { return params.RunnerScaleSet{}, err @@ -55,17 +65,44 @@ func (s *ScaleSetClient) GetRunnerScaleSetByNameAndRunnerGroup(ctx context.Conte } defer resp.Body.Close() - var runnerScaleSetList *params.RunnerScaleSetsResponse + var runnerScaleSetList params.RunnerScaleSetsResponse if err := json.NewDecoder(resp.Body).Decode(&runnerScaleSetList); err != nil { return params.RunnerScaleSet{}, fmt.Errorf("failed to decode response: %w", err) } - if runnerScaleSetList.Count == 0 { + + switch runnerScaleSetList.Count { + case 0: + allScaleSets, err := s.ListRunnerScaleSets(ctx, runnerGroupID) + if err != nil { + return params.RunnerScaleSet{}, fmt.Errorf("failed to list runner scale sets: %w", err) + } + var matchingScaleSet params.RunnerScaleSet + matches := 0 + for _, scaleSet := range allScaleSets.RunnerScaleSets { + if matchesRunnerScaleSet(scaleSet, runnerGroupID, name) { + matchingScaleSet = scaleSet + matches++ + } + } + if matches == 1 { + return matchingScaleSet, nil + } + if matches > 1 { + return params.RunnerScaleSet{}, runnerErrors.NewConflictError("multiple runner scale sets exist with name %s and runner group ID %d", name, runnerGroupID) + } return params.RunnerScaleSet{}, runnerErrors.NewNotFoundError("runner scale set with name %s and runner group ID %d was not found", name, runnerGroupID) + case 1: + if len(runnerScaleSetList.RunnerScaleSets) != 1 { + return params.RunnerScaleSet{}, fmt.Errorf("runner scale set response count does not match its values") + } + scaleSet := runnerScaleSetList.RunnerScaleSets[0] + if !matchesRunnerScaleSet(scaleSet, runnerGroupID, name) { + return params.RunnerScaleSet{}, fmt.Errorf("runner scale set response does not match the requested name and runner group") + } + return scaleSet, nil + default: + return params.RunnerScaleSet{}, runnerErrors.NewConflictError("multiple runner scale sets exist with name %s and runner group ID %d", name, runnerGroupID) } - - // Runner scale sets must have a uniqe name. Attempting to create a runner scale set with the same name as - // an existing scale set will result in a Bad Request (400) error. - return runnerScaleSetList.RunnerScaleSets[0], nil } func (s *ScaleSetClient) GetRunnerScaleSetByID(ctx context.Context, runnerScaleSetID int) (_ params.RunnerScaleSet, err error) { @@ -95,8 +132,8 @@ func (s *ScaleSetClient) GetRunnerScaleSetByID(ctx context.Context, runnerScaleS return runnerScaleSet, nil } -// ListRunnerScaleSets lists all runner scale sets in a github entity. -func (s *ScaleSetClient) ListRunnerScaleSets(ctx context.Context) (_ *params.RunnerScaleSetsResponse, err error) { +// ListRunnerScaleSets lists all runner scale sets in a runner group. +func (s *ScaleSetClient) ListRunnerScaleSets(ctx context.Context, runnerGroupID int) (_ *params.RunnerScaleSetsResponse, err error) { s.recordOperation("ListRunnerScaleSets") defer func() { if err != nil { @@ -104,7 +141,11 @@ func (s *ScaleSetClient) ListRunnerScaleSets(ctx context.Context) (_ *params.Run } }() - req, err := s.newActionsRequest(ctx, http.MethodGet, scaleSetEndpoint, nil) + query := url.Values{ + "runnerGroupId": []string{strconv.Itoa(runnerGroupID)}, + } + path := scaleSetEndpoint + "?" + query.Encode() + req, err := s.newActionsRequest(ctx, http.MethodGet, path, nil) if err != nil { return nil, err } diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go index 418756c0d..fc4d66cbb 100644 --- a/workers/scaleset/scaleset.go +++ b/workers/scaleset/scaleset.go @@ -33,6 +33,7 @@ import ( "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner/common" garmUtil "github.com/cloudbase/garm/util" + "github.com/cloudbase/garm/util/github/scalesets" ) func NewWorker(ctx context.Context, store dbCommon.Store, scaleSet params.ScaleSet, provider common.Provider) (*Worker, error) { @@ -85,6 +86,35 @@ type Worker struct { quit chan struct{} } +func (w *Worker) recordScaleSetID(entity params.ForgeEntity, scaleSetID int) error { + updateParams := params.UpdateScaleSetParams{ + ScaleSetID: scaleSetID, + } + if _, err := w.store.UpdateEntityScaleSet(w.ctx, entity, w.scaleSet.ID, updateParams, nil); err != nil { + return fmt.Errorf("failed to update scale set: %w", err) + } + + if err := w.SetLastMessageID(0); err != nil { + return fmt.Errorf("failed to reset last message id: %w", err) + } + w.scaleSet.ScaleSetID = scaleSetID + return nil +} + +func (w *Worker) adoptScaleSet(entity params.ForgeEntity, scaleSet params.RunnerScaleSet) error { + if w.scaleSet.ScaleSetID == scaleSet.ID { + return nil + } + if w.scaleSet.ScaleSetID != 0 { + return runnerErrors.NewConflictError( + "scale set already exists in github and it differs from the ID we know (github: %d vs local: %d)", + scaleSet.ID, + w.scaleSet.ScaleSetID, + ) + } + return w.recordScaleSetID(entity, scaleSet.ID) +} + func (w *Worker) ensureScaleSetInGitHub() error { entity, err := w.scaleSet.GetEntity() if err != nil { @@ -106,15 +136,7 @@ func (w *Worker) ensureScaleSetInGitHub() error { } scaleSet, err := cli.GetRunnerScaleSetByNameAndRunnerGroup(w.ctx, int(rgID), w.scaleSet.Name) if err == nil { - // The scale set exists - if scaleSet.ID != w.scaleSet.ScaleSetID { - // The scale set exists in github, but the ID differs from what we know to be true. - // It is possible that the scale set is being managed by some other auto scaler. - // We error here, as there is no way to listen on a scale set that already has a listener - // or is being managed by something else. - return fmt.Errorf("scale set already exists in github and it differs from the ID we know (github: %d vs local: %d)", scaleSet.ID, w.scaleSet.ScaleSetID) - } - return nil + return w.adoptScaleSet(entity, scaleSet) } if !errors.Is(err, runnerErrors.ErrNotFound) { return fmt.Errorf("failed to get scale set: %w", err) @@ -134,25 +156,15 @@ func (w *Worker) ensureScaleSetInGitHub() error { } runnerScaleSet, err := cli.CreateRunnerScaleSet(w.ctx, createScaleSetParams) if err != nil { + if errors.Is(err, scalesets.ErrRunnerScaleSetExists) { + existingScaleSet, lookupErr := cli.GetRunnerScaleSetByNameAndRunnerGroup(w.ctx, int(rgID), w.scaleSet.Name) + if lookupErr == nil { + return w.adoptScaleSet(entity, existingScaleSet) + } + } return fmt.Errorf("error creating runner scale set: %w", err) } - - // update the DB scale set - updateParams := params.UpdateScaleSetParams{ - ScaleSetID: runnerScaleSet.ID, - } - _, err = w.store.UpdateEntityScaleSet(w.ctx, entity, w.scaleSet.ID, updateParams, nil) - if err != nil { - return fmt.Errorf("failed to update scale set: %w", err) - } - - // The scale set was recreated. We need to reset the last message ID we recorded previously. - if err := w.SetLastMessageID(0); err != nil { - return fmt.Errorf("failed to reset last message id: %w", err) - } - w.scaleSet.ScaleSetID = runnerScaleSet.ID - - return nil + return w.recordScaleSetID(entity, runnerScaleSet.ID) } func (w *Worker) Stop() error { diff --git a/workers/scaleset/scaleset_adoption_test.go b/workers/scaleset/scaleset_adoption_test.go new file mode 100644 index 000000000..118956109 --- /dev/null +++ b/workers/scaleset/scaleset_adoption_test.go @@ -0,0 +1,256 @@ +// Copyright 2026 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +//go:build testing + +package scaleset + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/google/go-github/v84/github" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + runnerErrors "github.com/cloudbase/garm-provider-common/errors" + "github.com/cloudbase/garm/cache" + storeMocks "github.com/cloudbase/garm/database/common/mocks" + "github.com/cloudbase/garm/params" + runnerMocks "github.com/cloudbase/garm/runner/common/mocks" +) + +const testActionsToken = "eyJhbGciOiJub25lIn0.eyJleHAiOjQxNDk5MzYwMDB9." + +func newScaleSetWorkerForTest( + t *testing.T, + store *storeMocks.Store, + scaleSet params.ScaleSet, + actionsHandler http.HandlerFunc, +) *Worker { + t.Helper() + + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/actions/runner-registration" { + assert.Equal(t, http.MethodPost, r.Method) + _, _ = fmt.Fprintf(w, `{"url":%q,"token":%q}`, server.URL, testActionsToken) + return + } + actionsHandler(w, r) + })) + t.Cleanup(server.Close) + + baseURL, err := url.Parse(server.URL) + require.NoError(t, err) + entity := params.ForgeEntity{ + ID: scaleSet.RepoID, + EntityType: params.ForgeEntityTypeRepository, + Owner: "owner", + Name: "repo", + Credentials: params.ForgeCredentials{ + BaseURL: server.URL, + }, + } + githubClient := runnerMocks.NewGithubClient(t) + githubClient.EXPECT(). + CreateEntityRegistrationToken(mock.Anything). + Return(&github.RegistrationToken{ + Token: github.Ptr("registration-token"), + ExpiresAt: &github.Timestamp{Time: time.Now().Add(time.Hour)}, + }, nil, nil). + Once() + githubClient.EXPECT().GithubBaseURL().Return(baseURL).Once() + githubClient.EXPECT().GetEntityRunnerGroupIDByName(mock.Anything, scaleSet.GitHubRunnerGroup).Return(int64(1), nil).Once() + githubClient.EXPECT().GetEntity().Return(entity).Maybe() + cache.SetGithubClient(scaleSet.RepoID, githubClient) + t.Cleanup(func() { cache.DeleteGithubClient(scaleSet.RepoID) }) + + return &Worker{ + ctx: context.Background(), + store: store, + scaleSet: scaleSet, + } +} + +func testScaleSet(t *testing.T) params.ScaleSet { + t.Helper() + return params.ScaleSet{ + ID: 4, + Name: "example-garm-123", + RepoID: t.Name(), + GitHubRunnerGroup: "default", + } +} + +func expectScaleSetIDUpdate(t *testing.T, store *storeMocks.Store, scaleSet params.ScaleSet, githubID int) { + t.Helper() + + entity := params.ForgeEntity{ID: scaleSet.RepoID, EntityType: params.ForgeEntityTypeRepository} + store.EXPECT(). + UpdateEntityScaleSet( + mock.Anything, + entity, + scaleSet.ID, + mock.MatchedBy(func(update params.UpdateScaleSetParams) bool { + return update.ScaleSetID == githubID + }), + mock.Anything, + ). + Return(params.ScaleSet{}, nil). + Once() + store.EXPECT().SetScaleSetLastMessageID(mock.Anything, scaleSet.ID, int64(0)).Return(nil).Once() +} + +func TestEnsureScaleSetInGitHubAdoptsExistingScaleSet(t *testing.T) { + scaleSet := testScaleSet(t) + store := storeMocks.NewStore(t) + expectScaleSetIDUpdate(t, store, scaleSet, 42) + + w := newScaleSetWorkerForTest(t, store, scaleSet, func(rw http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodGet, r.Method) + assert.Equal(t, "1", r.URL.Query().Get("runnerGroupId")) + assert.Equal(t, scaleSet.Name, r.URL.Query().Get("name")) + _, _ = fmt.Fprintf(rw, `{"count":1,"value":[{"id":42,"name":%q,"runnerGroupId":1}]}`, scaleSet.Name) + }) + + require.NoError(t, w.ensureScaleSetInGitHub()) + assert.Equal(t, 42, w.scaleSet.ScaleSetID) +} + +func TestEnsureScaleSetInGitHubEscapesLookupName(t *testing.T) { + scaleSet := testScaleSet(t) + scaleSet.Name = "example+garm&123" + store := storeMocks.NewStore(t) + expectScaleSetIDUpdate(t, store, scaleSet, 42) + + w := newScaleSetWorkerForTest(t, store, scaleSet, func(rw http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodGet, r.Method) + assert.Equal(t, scaleSet.Name, r.URL.Query().Get("name")) + _, _ = fmt.Fprintf(rw, `{"count":1,"value":[{"id":42,"name":%q,"runnerGroupId":1}]}`, scaleSet.Name) + }) + + require.NoError(t, w.ensureScaleSetInGitHub()) +} + +func TestEnsureScaleSetInGitHubFallsBackToRunnerGroupListWithoutRunnerGroupID(t *testing.T) { + scaleSet := testScaleSet(t) + store := storeMocks.NewStore(t) + expectScaleSetIDUpdate(t, store, scaleSet, 42) + + w := newScaleSetWorkerForTest(t, store, scaleSet, func(rw http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodGet, r.Method) + assert.Equal(t, "1", r.URL.Query().Get("runnerGroupId")) + if r.URL.Query().Get("name") != "" { + _, _ = rw.Write([]byte(`{"count":0,"value":[]}`)) + return + } + _, _ = fmt.Fprintf(rw, `{"count":1,"value":[{"id":42,"name":%q}]}`, scaleSet.Name) + }) + + require.NoError(t, w.ensureScaleSetInGitHub()) + assert.Equal(t, 42, w.scaleSet.ScaleSetID) +} + +func TestEnsureScaleSetInGitHubRecoversCreateConflict(t *testing.T) { + scaleSet := testScaleSet(t) + store := storeMocks.NewStore(t) + expectScaleSetIDUpdate(t, store, scaleSet, 42) + + createAttempted := false + w := newScaleSetWorkerForTest(t, store, scaleSet, func(rw http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + if createAttempted && r.URL.Query().Get("name") == "" { + _, _ = fmt.Fprintf(rw, `{"count":1,"value":[{"id":42,"name":%q,"runnerGroupId":1}]}`, scaleSet.Name) + return + } + _, _ = rw.Write([]byte(`{"count":0,"value":[]}`)) + case http.MethodPost: + createAttempted = true + rw.WriteHeader(http.StatusBadRequest) + _, _ = rw.Write([]byte(`{"typeName":"RunnerScaleSetExistsException"}`)) + default: + t.Errorf("unexpected Actions request: %s %s", r.Method, r.URL.String()) + } + }) + + require.NoError(t, w.ensureScaleSetInGitHub()) + assert.True(t, createAttempted) + assert.Equal(t, 42, w.scaleSet.ScaleSetID) +} + +func TestEnsureScaleSetInGitHubPreservesUnrelatedBadRequest(t *testing.T) { + scaleSet := testScaleSet(t) + store := storeMocks.NewStore(t) + createAttempted := false + w := newScaleSetWorkerForTest(t, store, scaleSet, func(rw http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + createAttempted = true + rw.WriteHeader(http.StatusBadRequest) + _, _ = rw.Write([]byte(`{"typeName":"InvalidRunnerScaleSetException"}`)) + return + } + if createAttempted { + _, _ = fmt.Fprintf(rw, `{"count":1,"value":[{"id":42,"name":%q,"runnerGroupId":1}]}`, scaleSet.Name) + return + } + _, _ = rw.Write([]byte(`{"count":0,"value":[]}`)) + }) + + err := w.ensureScaleSetInGitHub() + require.Error(t, err) + assert.ErrorIs(t, err, runnerErrors.ErrBadRequest) + assert.Zero(t, w.scaleSet.ScaleSetID) +} + +func TestEnsureScaleSetInGitHubCreatesMissingScaleSet(t *testing.T) { + scaleSet := testScaleSet(t) + store := storeMocks.NewStore(t) + expectScaleSetIDUpdate(t, store, scaleSet, 42) + + createCount := 0 + w := newScaleSetWorkerForTest(t, store, scaleSet, func(rw http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + createCount++ + _, _ = fmt.Fprintf(rw, `{"id":42,"name":%q,"runnerGroupId":1}`, scaleSet.Name) + return + } + _, _ = rw.Write([]byte(`{"count":0,"value":[]}`)) + }) + + require.NoError(t, w.ensureScaleSetInGitHub()) + assert.Equal(t, 1, createCount) + assert.Equal(t, 42, w.scaleSet.ScaleSetID) +} + +func TestEnsureScaleSetInGitHubPreservesExistingIDMismatch(t *testing.T) { + scaleSet := testScaleSet(t) + scaleSet.ScaleSetID = 7 + store := storeMocks.NewStore(t) + w := newScaleSetWorkerForTest(t, store, scaleSet, func(rw http.ResponseWriter, _ *http.Request) { + _, _ = fmt.Fprintf(rw, `{"count":1,"value":[{"id":42,"name":%q,"runnerGroupId":1}]}`, scaleSet.Name) + }) + + err := w.ensureScaleSetInGitHub() + require.Error(t, err) + var conflict *runnerErrors.ConflictError + assert.ErrorAs(t, err, &conflict) +}