diff --git a/README.md b/README.md index 29c0856..c1d139a 100644 --- a/README.md +++ b/README.md @@ -39,9 +39,10 @@ The provider controller can be configured using the following environment variab | `GIT_PROVIDER_LEADER_ELECTION` | bool | `false` | Use leader election for the controller manager | | `GIT_PROVIDER_MAX_ERROR_RETRY_INTERVAL` | duration | `1m` | The maximum interval between retries when an error occurs. Should be less than half of the poll interval | | `GIT_PROVIDER_MIN_ERROR_RETRY_INTERVAL` | duration | `1s` | The minimum interval between retries when an error occurs. Should be less than max-error-retry-interval | -| `GIT_PROVIDER_TIMEOUT` | duration | `4m` | The timeout time for each action. | +| `GIT_PROVIDER_TIMEOUT` | duration | `4m` | The time all the external calls of one reconcile share, clones and pushes included. A call still running at the deadline fails. Raise it for repositories that take longer to clone and push (with the chart: `--set-string env.GIT_PROVIDER_TIMEOUT=10m`) | | `GIT_PROVIDER_GIT_COMMIT_AUTHOR_NAME` | string | `krateo-git-provider` | The name to use for git commits. | | `GIT_PROVIDER_GIT_COMMIT_AUTHOR_EMAIL` | string | `contact@krateo.io` | The email to use for git commits. | +| `GIT_PROVIDER_RATE_LIMIT_GOVERNOR` | bool | `false` (the chart sets `true`) | Slow reconciles down per remote and credential when a git server signals a rate limit | ## CRD Reference To view the generated CR configuration schema, visit [this link](https://doc.crds.dev/github.com/krateoplatformops/git-provider). \ No newline at end of file diff --git a/go.mod b/go.mod index df9c55f..88e8c6a 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/go-git/go-git/v5 v5.19.2 github.com/go-logr/logr v1.4.4 github.com/krateo-platformops/plumbing v1.14.3 - github.com/krateo-platformops/provider-runtime v1.8.2 + github.com/krateo-platformops/provider-runtime v1.11.0 github.com/moby/moby/api v1.52.0 github.com/moby/moby/client v0.1.0 github.com/pkg/errors v0.9.1 diff --git a/go.sum b/go.sum index 0e9e551..4037336 100644 --- a/go.sum +++ b/go.sum @@ -149,8 +149,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/krateo-platformops/plumbing v1.14.3 h1:g60QTgbhIsZVTFq9JZveNx2QZauVyFeWHUAVxahxKRU= github.com/krateo-platformops/plumbing v1.14.3/go.mod h1:JMHwVbS2vdA/OSm3Gi3QCDTntFjPmrGtqZ6LQuhciZQ= -github.com/krateo-platformops/provider-runtime v1.8.2 h1:X3OxWfdsZfxOAR6IkdVD/cobCUo04mJKaxjHhSZmOQU= -github.com/krateo-platformops/provider-runtime v1.8.2/go.mod h1:3j80GRyb7mfg+FweV0Ev1Si/0n72lZEujb3JyYL3BMQ= +github.com/krateo-platformops/provider-runtime v1.11.0 h1:PLhAYZ2uzoUrw7S7HvQ3KXc2/htJ2mvboevJe+chqkE= +github.com/krateo-platformops/provider-runtime v1.11.0/go.mod h1:Ah+zDm8QQEI4zm8epsDQnWTGNzoi2jnwivKsxTX2IrY= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/mailru/easyjson v0.9.0 h1:PrnmzHw7262yW8sTBwxi1PdJA3Iw/EKBa8psRf7d9a4= diff --git a/helm/git-provider/values.schema.json b/helm/git-provider/values.schema.json index f3c42fc..46a8d3c 100644 --- a/helm/git-provider/values.schema.json +++ b/helm/git-provider/values.schema.json @@ -164,6 +164,10 @@ "type": "string", "title": "Poll interval (e.g. 3m)" }, + "GIT_PROVIDER_RATE_LIMIT_GOVERNOR": { + "type": "string", + "title": "Slow down when a git server signals a rate limit" + }, "GIT_PROVIDER_SERVICE_PORT_HACK": { "type": "string", "title": "Service port hack" diff --git a/helm/git-provider/values.yaml b/helm/git-provider/values.yaml index 7295b62..d029c1a 100644 --- a/helm/git-provider/values.yaml +++ b/helm/git-provider/values.yaml @@ -80,5 +80,6 @@ affinity: {} env: GIT_PROVIDER_DEBUG: "false" GIT_PROVIDER_POLL_INTERVAL: "3m" + GIT_PROVIDER_RATE_LIMIT_GOVERNOR: "true" GIT_PROVIDER_SERVICE_PORT_HACK: "false" HOME: "/tmp" diff --git a/internal/clients/git/git.go b/internal/clients/git/git.go index 76823ad..8b6a6c5 100644 --- a/internal/clients/git/git.go +++ b/internal/clients/git/git.go @@ -20,7 +20,6 @@ import ( "github.com/go-git/go-git/v5/plumbing/cache" gitclient "github.com/go-git/go-git/v5/plumbing/transport/client" - githttp "github.com/go-git/go-git/v5/plumbing/transport/http" "github.com/go-git/go-git/v5/storage" "github.com/go-git/go-billy/v5" @@ -90,11 +89,15 @@ func normalizeEmptyReasonError(err error) error { // krateo.io/external-create-pending annotation and writing the create result, // the managed resource is wedged forever on "cannot determine creation result". // -// Operations that need no custom client take the read lock and share go-git's -// default https client; only the cookie-jar and capability paths write, under -// the exclusive lock, and they restore the defaults before releasing it. +// Operations that need no custom client take the read lock and share +// defaultClient; only the cookie-jar and capability paths write, under the +// exclusive lock, and they restore the defaults before releasing it. var transportMu sync.RWMutex +// remoteListTimeout bounds listing a remote's refs. It is the timeout go-git's +// Remote.List applies itself; ListContext applies none. +const remoteListTimeout = 10 * time.Second + type Repo struct { rawURL string auth transport.AuthMethod @@ -132,12 +135,12 @@ type IndexOptions struct { ToPath string } -// cookieJarClient builds the HTTP client carrying the given git cookie. +// cookieJarClient builds the go-git client carrying the given git cookie. // // It returns (nil, nil) when there is no usable cookie, which means the caller -// must leave the global transport registry completely alone - go-git's default -// https client is already installed and is exactly what we want. -func cookieJarClient(cookie []byte) (*http.Client, error) { +// must leave the global transport registry completely alone - defaultClient is +// already installed and is exactly what we want. +func cookieJarClient(cookie []byte) (*signalClient, error) { cookie = bytes.Trim(cookie, "\n") if len(cookie) == 0 { return nil, nil @@ -173,7 +176,7 @@ func cookieJarClient(cookie []byte) (*http.Client, error) { }, ) - return &http.Client{Jar: jar}, nil + return newSignalClient(jar), nil } // azureDevOps reports whether the URL points at Azure DevOps, which needs the @@ -202,9 +205,9 @@ func lockTransport(cookie []byte, mutatesGlobals bool) (func(), error) { } if custom == nil && !mutatesGlobals { - // Nothing to install: go-git's default https client is already in the - // registry. Hold the read lock so that no writer can mutate the map - // out from under go-git while it reads it. + // Nothing to install: defaultClient is already in the registry. Hold + // the read lock so that no writer can mutate the map out from under + // go-git while it reads it. transportMu.RLock() return transportMu.RUnlock, nil } @@ -214,11 +217,12 @@ func lockTransport(cookie []byte, mutatesGlobals bool) (func(), error) { return transportMu.Unlock, nil } - gitclient.InstallProtocol("https", githttp.NewClient(custom)) + gitclient.InstallProtocol("https", custom) return func() { - // Restore the exact singleton go-git ships in its registry, rather than - // allocating an equivalent-but-different client on every release. - gitclient.InstallProtocol("https", githttp.DefaultClient) + // Restore the singleton rather than allocating an equivalent client on + // every release. It has to be ours, not go-git's: restoring go-git's + // would silently stop every later operation from reporting signals. + gitclient.InstallProtocol("https", defaultClient) transportMu.Unlock() }, nil } @@ -226,18 +230,18 @@ func lockTransport(cookie []byte, mutatesGlobals bool) (func(), error) { // GetLatestCommitRemote lists the remote refs and returns the tip of the // requested branch. It takes the transport lock; use getLatestCommitRemote when // the caller already holds it. -func GetLatestCommitRemote(opts ListOptions) (*string, error) { +func GetLatestCommitRemote(ctx context.Context, opts ListOptions) (*string, error) { release, err := lockTransport(opts.GitCookies, false) if err != nil { return nil, err } defer release() - return getLatestCommitRemote(opts) + return getLatestCommitRemote(ctx, opts) } // getLatestCommitRemote assumes the caller already holds the transport lock. -func getLatestCommitRemote(opts ListOptions) (*string, error) { +func getLatestCommitRemote(ctx context.Context, opts ListOptions) (*string, error) { tmpDir, err := os.MkdirTemp(opts.HomeDir, "git-provider-list-*") if err != nil { return nil, fmt.Errorf("failed to create temporary directory: %w", err) @@ -272,7 +276,10 @@ func getLatestCommitRemote(opts ListOptions) (*string, error) { return nil, err } - refs, err := remote.List(&git.ListOptions{ + ctx, cancel := context.WithTimeout(ctx, remoteListTimeout) + defer cancel() + + refs, err := remote.ListContext(ctx, &git.ListOptions{ Auth: opts.Auth, InsecureSkipTLS: opts.Insecure, }) @@ -293,7 +300,7 @@ func restoreUnsupportedCapabilities(oldUnsupportedCaps []capability.Capability) transport.UnsupportedCapabilities = oldUnsupportedCaps } -func isInGitCommitHistory(ctx context.Context, opts ListOptions, hash string) (bool, error) { +func IsInGitCommitHistory(ctx context.Context, opts ListOptions, hash string) (bool, error) { log := contexttools.LoggerFromCtx(ctx, logging.NewNopLogger()) tmpDir, err := os.MkdirTemp(opts.HomeDir, "git-provider-history-*") @@ -349,7 +356,7 @@ func isInGitCommitHistory(ctx context.Context, opts ListOptions, hash string) (b } } - res.repo, err = git.Clone(res.storer, res.fs, &cloneOpts) + res.repo, err = git.CloneContext(ctx, res.storer, res.fs, &cloneOpts) if err != nil { if strings.Contains(err.Error(), "couldn't find remote ref") { log.Warn("Branch not found in remote repository", "branch", opts.Branch, "url", opts.URL) @@ -439,7 +446,7 @@ func IsFuncInGitCommitHistory(ctx context.Context, opts ListOptions, f func(comm } } - res.repo, err = git.Clone(res.storer, res.fs, &cloneOpts) + res.repo, err = git.CloneContext(ctx, res.storer, res.fs, &cloneOpts) if err != nil { if strings.Contains(err.Error(), "couldn't find remote ref") { log.Warn("Branch not found in remote repository", "branch", opts.Branch, "url", opts.URL) @@ -523,11 +530,7 @@ func (s *Repo) UpdateIndex(idx *IndexOptions) error { } return nil } -func Clone(opts CloneOptions) (*Repo, error) { - return clone(context.Background(), opts) -} - -func clone(ctx context.Context, opts CloneOptions) (*Repo, error) { +func Clone(ctx context.Context, opts CloneOptions) (*Repo, error) { tmpDir, err := os.MkdirTemp(opts.HomeDir, "git-provider-clone-*") if err != nil { return nil, fmt.Errorf("failed to create temporary directory: %w", err) @@ -583,7 +586,7 @@ func clone(ctx context.Context, opts CloneOptions) (*Repo, error) { InsecureSkipTLS: opts.Insecure, } isOrphan := true - _, err = getLatestCommitRemote(ListOptions{ + _, err = getLatestCommitRemote(ctx, ListOptions{ URL: opts.URL, Auth: opts.Auth, Insecure: opts.Insecure, @@ -607,7 +610,7 @@ func clone(ctx context.Context, opts CloneOptions) (*Repo, error) { } res.isNewBranch = ptr.To(true) } - res.repo, err = git.Clone(res.storer, res.fs, &cloneOpts) + res.repo, err = git.CloneContext(ctx, res.storer, res.fs, &cloneOpts) if err != nil { return nil, fmt.Errorf("failed to clone repository: %w", normalizeEmptyReasonError(err)) } @@ -620,14 +623,6 @@ func clone(ctx context.Context, opts CloneOptions) (*Repo, error) { return res, err } -func IsInGitCommitHistory(opts ListOptions, hash string) (bool, error) { - return isInGitCommitHistory(context.Background(), opts, hash) -} - -func IsInGitCommitHistoryContext(ctx context.Context, opts ListOptions, hash string) (bool, error) { - return isInGitCommitHistory(ctx, opts, hash) -} - // Exists stats the local filesystem: it never opens a transport, so it must not // touch the global transport registry. func (s *Repo) Exists(path string) (bool, error) { @@ -763,7 +758,7 @@ func (s *Repo) Commit(path, msg string, opt *IndexOptions) (plumbing.Hash, error return hash, nil } -func (s *Repo) Push(downstream, branch string, insecure bool) error { +func (s *Repo) Push(ctx context.Context, downstream, branch string, insecure bool) error { release, err := lockTransport(s.cookie, false) if err != nil { return err @@ -772,7 +767,7 @@ func (s *Repo) Push(downstream, branch string, insecure bool) error { //Push the code to the remote if len(branch) == 0 { - return s.repo.Push(&git.PushOptions{ + return s.repo.PushContext(ctx, &git.PushOptions{ RemoteName: downstream, Auth: s.auth, InsecureSkipTLS: insecure, @@ -806,7 +801,7 @@ func (s *Repo) Push(downstream, branch string, insecure bool) error { } } - err = s.repo.Push(&git.PushOptions{ + err = s.repo.PushContext(ctx, &git.PushOptions{ RemoteName: downstream, Force: false, Auth: s.auth, @@ -821,7 +816,7 @@ func (s *Repo) Push(downstream, branch string, insecure bool) error { return nil } -func Pull(s *Repo, insecure bool) error { +func Pull(ctx context.Context, s *Repo, insecure bool) error { release, err := lockTransport(s.cookie, false) if err != nil { return err @@ -834,7 +829,7 @@ func Pull(s *Repo, insecure bool) error { return err } - err = wt.Pull(&git.PullOptions{ + err = wt.PullContext(ctx, &git.PullOptions{ RemoteName: "origin", Auth: s.auth, InsecureSkipTLS: insecure, diff --git a/internal/clients/git/git_test.go b/internal/clients/git/git_test.go index dccb46c..e5445f3 100644 --- a/internal/clients/git/git_test.go +++ b/internal/clients/git/git_test.go @@ -24,7 +24,7 @@ func TestIsInGitCommitHistory(t *testing.T) { hash := "6ecf0ef2c2dffb796033e5a02219af86ec6584e5" - exists, err := IsInGitCommitHistory(opts, hash) + exists, err := IsInGitCommitHistory(context.Background(), opts, hash) if err != nil { t.Errorf("Error checking commit history: %v", err) } @@ -41,7 +41,7 @@ func TestGetLatestCommitRemote(t *testing.T) { baseRepo := BaseSuite{} baseRepo.BuildBasicRepository() - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), }) require.NoError(t, err) @@ -49,7 +49,7 @@ func TestGetLatestCommitRemote(t *testing.T) { _, err = repo.FS().OpenFile("README.md", os.O_RDWR|os.O_CREATE, 0644) require.NoError(t, err) - commit, err := GetLatestCommitRemote(ListOptions{ + commit, err := GetLatestCommitRemote(context.Background(), ListOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -57,7 +57,7 @@ func TestGetLatestCommitRemote(t *testing.T) { expected := "6ecf0ef2c2dffb796033e5a02219af86ec6584e5" assert.Equal(t, expected, *commit) - commit, err = GetLatestCommitRemote(ListOptions{ + commit, err = GetLatestCommitRemote(context.Background(), ListOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "branch", }) @@ -70,7 +70,7 @@ func TestGetLatestCommitRemoteBranchNotFound(t *testing.T) { baseRepo := BaseSuite{} baseRepo.BuildBasicRepository() - _, err := GetLatestCommitRemote(ListOptions{ + _, err := GetLatestCommitRemote(context.Background(), ListOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "missing-branch", }) @@ -93,13 +93,13 @@ func TestPull(t *testing.T) { baseRepo := BaseSuite{} baseRepo.BuildBasicRepository() - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) require.NoError(t, err) - err = Pull(repo, false) + err = Pull(context.Background(), repo, false) require.NoError(t, err) } @@ -107,7 +107,7 @@ func TestBranch(t *testing.T) { baseRepo := BaseSuite{} baseRepo.BuildBasicRepository() - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -133,7 +133,7 @@ func TestCurrentBranch(t *testing.T) { baseRepo := BaseSuite{} baseRepo.BuildBasicRepository() - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -147,7 +147,7 @@ func TestGetLatestCommit(t *testing.T) { baseRepo := BaseSuite{} baseRepo.BuildBasicRepository() - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), }) require.NoError(t, err) @@ -162,7 +162,7 @@ func TestPush(t *testing.T) { baseRepo.BuildBasicRepository() t.Run("push without branch", func(t *testing.T) { - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -185,12 +185,12 @@ func TestPush(t *testing.T) { require.NoError(t, err) // Push to a different branch to avoid the "currently checked out" error - err = repo.Push("origin", "test-branch", false) + err = repo.Push(context.Background(), "origin", "test-branch", false) require.NoError(t, err) }) t.Run("push with specific branch", func(t *testing.T) { - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -219,12 +219,12 @@ func TestPush(t *testing.T) { }) require.NoError(t, err) - err = repo.Push("origin", "feature", false) + err = repo.Push(context.Background(), "origin", "feature", false) require.NoError(t, err) }) t.Run("push with insecure option", func(t *testing.T) { - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -247,12 +247,12 @@ func TestPush(t *testing.T) { require.NoError(t, err) // Push to a different branch to avoid the "currently checked out" error - err = repo.Push("origin", "insecure-test-branch", true) + err = repo.Push(context.Background(), "origin", "insecure-test-branch", true) require.NoError(t, err) }) t.Run("push non-existing local branch", func(t *testing.T) { - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -275,7 +275,7 @@ func TestPush(t *testing.T) { require.NoError(t, err) // Try to push to a branch that doesn't exist locally yet - err = repo.Push("origin", "new-branch", false) + err = repo.Push(context.Background(), "origin", "new-branch", false) require.NoError(t, err) }) } @@ -310,7 +310,7 @@ func TestIsFuncInGitCommitHistory_complex(t *testing.T) { baseRepo := BaseSuite{} baseRepo.BuildBasicRepository() - repo, err := Clone(CloneOptions{ + repo, err := Clone(context.Background(), CloneOptions{ URL: baseRepo.GetBasicLocalRepositoryURL(), Branch: "master", }) @@ -333,6 +333,6 @@ func TestIsFuncInGitCommitHistory_complex(t *testing.T) { require.False(t, hash.IsZero()) // Try to push to a branch that doesn't exist locally yet - err = repo.Push("origin", "new-branch", false) + err = repo.Push(context.Background(), "origin", "new-branch", false) require.NoError(t, err) } diff --git a/internal/clients/git/gittest/gittest.go b/internal/clients/git/gittest/gittest.go index 626ff46..cba23f0 100644 --- a/internal/clients/git/gittest/gittest.go +++ b/internal/clients/git/gittest/gittest.go @@ -3,13 +3,20 @@ // The controllers' Observe reads remote history through the git client, which accepts a local // path as a remote URL. A repository made here lets a test drive Observe down a chosen path with // no network and no git server. go-git serves a local path by running the git-upload-pack binary, -// so git must be installed where the tests run. +// so git must be installed where the tests run. Server puts one on HTTP for a test that needs the +// responses of a git server, such as rate limits. package gittest import ( + "net/http" + "net/http/cgi" + "net/http/httptest" "os" + "os/exec" "path/filepath" "strconv" + "strings" + "sync" "testing" "time" @@ -64,3 +71,192 @@ func (r *Repo) Commit(message string) string { } return hash.String() } + +// Step is one kind of request a git client makes to a smart HTTP server. +type Step string + +const ( + // AdvertiseUpload is the GET of the refs before a list, clone, fetch or pull. + AdvertiseUpload Step = "advertise-upload" + // UploadPack is the POST that downloads objects. + UploadPack Step = "upload-pack" + // AdvertiseReceive is the GET of the refs before a push. + AdvertiseReceive Step = "advertise-receive" + // ReceivePack is the POST that uploads objects. + ReceivePack Step = "receive-pack" +) + +// AllSteps refuses every kind of request. +var AllSteps = []Step{AdvertiseUpload, UploadPack, AdvertiseReceive, ReceivePack} + +// stepOf names the step r is, or "" for anything else. +func stepOf(r *http.Request) Step { + switch { + case strings.HasSuffix(r.URL.Path, "/info/refs") && r.URL.Query().Get("service") == "git-upload-pack": + return AdvertiseUpload + case strings.HasSuffix(r.URL.Path, "/info/refs") && r.URL.Query().Get("service") == "git-receive-pack": + return AdvertiseReceive + case strings.HasSuffix(r.URL.Path, "/git-upload-pack"): + return UploadPack + case strings.HasSuffix(r.URL.Path, "/git-receive-pack"): + return ReceivePack + } + return "" +} + +// refusal is the response a refused step gets. +type refusal struct { + status int + header map[string]string +} + +// Server serves one Repo over smart HTTP with the real `git http-backend`. A test refuses chosen +// steps with Refuse, the way a rate-limiting server does, or adds headers to every response with +// Annotate. So go-git runs for real against a real server, and every step can be refused on its +// own. +type Server struct { + // URL is the repository's remote URL on this server. + URL string + + mu sync.Mutex + refused map[Step]refusal + hung map[Step]bool + headers map[string]string + requests int + // release frees the handlers Hang is holding, so the server can shut down even when a client + // never gave up on them. + release chan struct{} +} + +// Serve starts a plain HTTP server for repo. +func Serve(t testing.TB, repo *Repo) *Server { + t.Helper() + s, h := newServer(t, repo) + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + t.Cleanup(func() { close(s.release) }) // cleanups run last first, so this goes before Close. + s.URL = srv.URL + "/" + filepath.Base(repo.URL) + return s +} + +// ServeTLS starts a server for repo with a certificate no client trusts, for testing the insecure +// option. +func ServeTLS(t testing.TB, repo *Repo) *Server { + t.Helper() + s, h := newServer(t, repo) + srv := httptest.NewTLSServer(h) + t.Cleanup(srv.Close) + t.Cleanup(func() { close(s.release) }) // cleanups run last first, so this goes before Close. + s.URL = srv.URL + "/" + filepath.Base(repo.URL) + return s +} + +func newServer(t testing.TB, repo *Repo) (*Server, http.Handler) { + t.Helper() + gitPath, err := exec.LookPath("git") + if err != nil { + t.Fatalf("finding git: %v", err) + } + // repo is checked out, and these servers exist to be pushed to. + if out, err := exec.Command(gitPath, "-C", repo.URL, "config", "receive.denyCurrentBranch", "ignore").CombinedOutput(); err != nil { + t.Fatalf("configuring test repository: %v: %s", err, out) + } + backend := &cgi.Handler{ + Path: gitPath, + Args: []string{"http-backend"}, + // REMOTE_USER makes http-backend accept pushes. + Env: []string{"GIT_PROJECT_ROOT=" + filepath.Dir(repo.URL), "GIT_HTTP_EXPORT_ALL=1", "REMOTE_USER=gittest"}, + InheritEnv: []string{"PATH", "HOME"}, + } + s := &Server{refused: map[Step]refusal{}, hung: map[Step]bool{}, headers: map[string]string{}, release: make(chan struct{})} + return s, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + s.requests++ + ref, refused := s.refused[stepOf(r)] + hung := s.hung[stepOf(r)] + for k, v := range s.headers { + w.Header().Set(k, v) + } + s.mu.Unlock() + if hung { + select { + case <-r.Context().Done(): + case <-s.release: + } + return + } + if refused { + for k, v := range ref.header { + w.Header().Set(k, v) + } + w.WriteHeader(ref.status) + return + } + backend.ServeHTTP(w, r) + }) +} + +// Refuse answers the given steps with status and header until Healthy is called. +func (s *Server) Refuse(status int, header map[string]string, steps ...Step) { + s.mu.Lock() + defer s.mu.Unlock() + for _, step := range steps { + s.refused[step] = refusal{status: status, header: header} + } +} + +// Hang accepts the given steps and never answers, the way a stuck server does. +func (s *Server) Hang(steps ...Step) { + s.mu.Lock() + defer s.mu.Unlock() + for _, step := range steps { + s.hung[step] = true + } +} + +// Annotate adds header to every response. +func (s *Server) Annotate(header map[string]string) { + s.mu.Lock() + defer s.mu.Unlock() + s.headers = header +} + +// Healthy clears Refuse and Annotate. +func (s *Server) Healthy() { + s.mu.Lock() + defer s.mu.Unlock() + s.refused = map[Step]refusal{} + s.headers = map[string]string{} +} + +// Requests is how many requests reached the server, refused or not. +func (s *Server) Requests() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.requests +} + +// Replay makes s answer the way the response at baseURL does: a refusal is applied to every step, +// and the rate-limit headers of a success are added to s's real git responses. It connects +// provider-runtime's reconcilertest helpers, which serve one recorded response to every request, +// to go-git, which needs a real server to get as far as a successful response. +func (s *Server) Replay(t testing.TB, baseURL string) { + t.Helper() + resp, err := http.Get(baseURL) //nolint:gosec // the test's own httptest server. + if err != nil { + t.Fatalf("reading the recorded response: %v", err) + } + defer func() { _ = resp.Body.Close() }() + + header := map[string]string{} + for k := range resp.Header { + if k == "Retry-After" || strings.HasPrefix(k, "X-Ratelimit-") { + header[k] = resp.Header.Get(k) + } + } + if resp.StatusCode >= http.StatusBadRequest { + s.Refuse(resp.StatusCode, header, AllSteps...) + return + } + s.Annotate(header) +} diff --git a/internal/clients/git/signal_test.go b/internal/clients/git/signal_test.go new file mode 100644 index 0000000..2a256ed --- /dev/null +++ b/internal/clients/git/signal_test.go @@ -0,0 +1,399 @@ +package git + +import ( + "context" + "net/http" + "strconv" + "testing" + "time" + + "github.com/go-git/go-git/v5/plumbing/object" + "github.com/go-git/go-git/v5/plumbing/transport" + gitclient "github.com/go-git/go-git/v5/plumbing/transport/client" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit/httpsignal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/krateoplatformops/git-provider/internal/clients/git/gittest" +) + +const ( + signalBranch = "main" + signalKey = "git.example.com|ns/creds/token" +) + +// throttles are the refusals the governor has to act on: GitHub's secondary limit and Azure +// DevOps's block (a 429 with Retry-After), and GitHub's primary limit (a 403 with no quota left). +var throttles = map[string]struct { + status int + header func() map[string]string +}{ + "429 with Retry-After": { + status: http.StatusTooManyRequests, + header: func() map[string]string { return map[string]string{"Retry-After": "120"} }, + }, + "403 with X-RateLimit-Remaining 0": { + status: http.StatusForbidden, + header: func() map[string]string { + return map[string]string{ + "X-RateLimit-Remaining": "0", + "X-RateLimit-Limit": "5000", + "X-RateLimit-Reset": strconv.FormatInt(time.Now().Add(2*time.Minute).Unix(), 10), + } + }, + }, +} + +func newGovernor(t *testing.T) *ratelimit.Governor { + t.Helper() + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 1}) + require.NoError(t, err) + return g +} + +// runtimeContext is what the runtime and the controllers hand the git client: the governor as the +// sink, and the key the controller gates this remote on. +func runtimeContext(g *ratelimit.Governor) context.Context { + return httpsignal.WithKey(ratelimit.WithSink(context.Background(), g), signalKey) +} + +// assertDeferred fails unless the governor now defers calls for signalKey until a server deadline. +func assertDeferred(t *testing.T, g *ratelimit.Governor) { + t.Helper() + _, d := g.Acquire(context.Background(), signalKey) + require.NotNil(t, d, "the throttled response never reached the governor: calls for %q are still let through", signalKey) + assert.Equal(t, ratelimit.DeferThrottled, d.Reason) + assert.True(t, d.Until.After(time.Now()), "the deferral carries no server deadline: %+v", d) +} + +// Each remote operation must reach the governor from every request it makes. Each case refuses one +// step of one operation, so a call that drops the context fails its own case: go-git's context-free +// forms run on context.Background(), and the signal is then lost with no error and no log. +func TestThrottledResponseReachesTheGovernor(t *testing.T) { + type env struct { + ctx context.Context + server *gittest.Server + opts ListOptions + home string + } + clone := func(t *testing.T, e env) *Repo { + t.Helper() + // A clone the test needs to get going. Its own requests go to the same server, before the + // refusal, on a context with no sink. + r, err := Clone(context.Background(), CloneOptions{URL: e.opts.URL, Branch: signalBranch, HomeDir: e.home}) + require.NoError(t, err) + t.Cleanup(func() { _ = r.Cleanup() }) + return r + } + commit := func(t *testing.T, r *Repo) { + t.Helper() + f, err := r.FS().Create("pushed.txt") + require.NoError(t, err) + require.NoError(t, f.Close()) + _, err = r.Commit(".", "pushed", &IndexOptions{}) + require.NoError(t, err) + } + + cases := map[string]struct { + step gittest.Step + // prepare runs before the refusal is switched on. + prepare func(t *testing.T, e env) *Repo + run func(e env, r *Repo) error + }{ + "GetLatestCommitRemote": { + step: gittest.AdvertiseUpload, + run: func(e env, _ *Repo) error { + _, err := GetLatestCommitRemote(e.ctx, e.opts) + return err + }, + }, + "Clone, listing the branch": { + step: gittest.AdvertiseUpload, + run: func(e env, _ *Repo) error { + _, err := Clone(e.ctx, CloneOptions{URL: e.opts.URL, Branch: signalBranch, HomeDir: e.home}) + return err + }, + }, + "Clone, downloading objects": { + step: gittest.UploadPack, + run: func(e env, _ *Repo) error { + _, err := Clone(e.ctx, CloneOptions{URL: e.opts.URL, Branch: signalBranch, HomeDir: e.home}) + return err + }, + }, + "IsInGitCommitHistory": { + step: gittest.UploadPack, + run: func(e env, _ *Repo) error { + _, err := IsInGitCommitHistory(e.ctx, e.opts, "0000000000000000000000000000000000000001") + return err + }, + }, + "IsFuncInGitCommitHistory": { + step: gittest.UploadPack, + run: func(e env, _ *Repo) error { + _, err := IsFuncInGitCommitHistory(e.ctx, e.opts, func(*object.Commit) bool { return false }) + return err + }, + }, + "Pull": { + step: gittest.AdvertiseUpload, + prepare: clone, + run: func(e env, r *Repo) error { return Pull(e.ctx, r, false) }, + }, + "Push, listing the remote": { + step: gittest.AdvertiseReceive, + prepare: func(t *testing.T, e env) *Repo { + r := clone(t, e) + commit(t, r) + return r + }, + run: func(e env, r *Repo) error { return r.Push(e.ctx, "origin", signalBranch, false) }, + }, + "Push, uploading objects": { + step: gittest.ReceivePack, + prepare: func(t *testing.T, e env) *Repo { + r := clone(t, e) + commit(t, r) + return r + }, + run: func(e env, r *Repo) error { return r.Push(e.ctx, "origin", signalBranch, false) }, + }, + "Push to the default refspec": { + step: gittest.AdvertiseReceive, + prepare: func(t *testing.T, e env) *Repo { + r := clone(t, e) + commit(t, r) + return r + }, + run: func(e env, r *Repo) error { return r.Push(e.ctx, "origin", "", false) }, + }, + } + + for throttleName, th := range throttles { + for name, tc := range cases { + t.Run(throttleName+"/"+name, func(t *testing.T) { + origin := gittest.NewRepo(t, signalBranch) + origin.Commit("first commit") + srv := gittest.Serve(t, origin) + g := newGovernor(t) + e := env{ + ctx: runtimeContext(g), + server: srv, + opts: ListOptions{URL: srv.URL, Branch: signalBranch, HomeDir: t.TempDir()}, + home: t.TempDir(), + } + + var r *Repo + if tc.prepare != nil { + r = tc.prepare(t, e) + } + srv.Refuse(th.status, th.header(), tc.step) + + require.Error(t, tc.run(e, r), "the refused request must fail the operation") + assertDeferred(t, g) + }) + } + } +} + +// The cookie path installs its own client for the length of the operation, so it needs the signal +// transport too, and has to put defaultClient back: go-git's own client would drop every signal +// from then on, in the whole process. +func TestThrottledResponseReachesTheGovernorWithACookie(t *testing.T) { + origin := gittest.NewRepo(t, signalBranch) + origin.Commit("first commit") + srv := gittest.ServeTLS(t, origin) + srv.Refuse(http.StatusTooManyRequests, map[string]string{"Retry-After": "120"}, gittest.AdvertiseUpload) + g := newGovernor(t) + + _, err := GetLatestCommitRemote(runtimeContext(g), ListOptions{ + URL: srv.URL, Branch: signalBranch, HomeDir: t.TempDir(), Insecure: true, GitCookies: testCookie, + }) + require.Error(t, err) + assertDeferred(t, g) + + ep, err := transport.NewEndpoint("https://example.com/o/r.git") + require.NoError(t, err) + got, err := gitclient.NewClient(ep) + require.NoError(t, err) + assert.Same(t, defaultClient, got) +} + +// spec.insecure is an API field, and go-git refuses it for a wrapped transport: the session fails +// with "expected underlying client transport to be of type: *http.Transport". The first half is +// that regression. The second proves the option is what lets the client through: without it the +// same server is rejected on its certificate, so insecure really does skip verification and +// nothing else does. +func TestInsecureRemotesKeepWorkingAndReportSignals(t *testing.T) { + origin := gittest.NewRepo(t, signalBranch) + want := origin.Commit("first commit") + srv := gittest.ServeTLS(t, origin) + opts := ListOptions{URL: srv.URL, Branch: signalBranch, HomeDir: t.TempDir(), Insecure: true} + + got, err := GetLatestCommitRemote(context.Background(), opts) + require.NoError(t, err) + assert.Equal(t, want, *got) + + r, err := Clone(context.Background(), CloneOptions{URL: srv.URL, Branch: signalBranch, HomeDir: t.TempDir(), Insecure: true}) + require.NoError(t, err) + t.Cleanup(func() { _ = r.Cleanup() }) + + f, err := r.FS().Create("pushed.txt") + require.NoError(t, err) + require.NoError(t, f.Close()) + _, err = r.Commit(".", "pushed", &IndexOptions{}) + require.NoError(t, err) + require.NoError(t, r.Push(context.Background(), "origin", signalBranch, true)) + + srv.Refuse(http.StatusTooManyRequests, map[string]string{"Retry-After": "120"}, gittest.AdvertiseUpload) + g := newGovernor(t) + _, err = GetLatestCommitRemote(runtimeContext(g), opts) + require.Error(t, err) + assertDeferred(t, g) + + srv.Healthy() + opts.Insecure = false + _, err = GetLatestCommitRemote(context.Background(), opts) + require.Error(t, err) + assert.Contains(t, err.Error(), "certificate", "without insecure the self-signed server must be rejected") +} + +// The transport must not throttle a healthy server or change what go-git sees: a whole clone and +// push succeed and leave the key at its ceiling. +func TestHealthyResponsesAreNotThrottled(t *testing.T) { + origin := gittest.NewRepo(t, signalBranch) + origin.Commit("first commit") + srv := gittest.Serve(t, origin) + // GitHub sends these on every response. + srv.Annotate(map[string]string{ + "X-RateLimit-Limit": "5000", + "X-RateLimit-Remaining": "4999", + "X-RateLimit-Reset": strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10), + }) + g := newGovernor(t) + ctx := runtimeContext(g) + + r, err := Clone(ctx, CloneOptions{URL: srv.URL, Branch: signalBranch, HomeDir: t.TempDir()}) + require.NoError(t, err) + t.Cleanup(func() { _ = r.Cleanup() }) + f, err := r.FS().Create("pushed.txt") + require.NoError(t, err) + require.NoError(t, f.Close()) + _, err = r.Commit(".", "pushed", &IndexOptions{}) + require.NoError(t, err) + require.NoError(t, r.Push(ctx, "origin", signalBranch, false)) + + release, d := g.Acquire(context.Background(), signalKey) + require.Nil(t, d) + release() + assert.Equal(t, g.Ceiling(), g.Capacity(signalKey)) +} + +// An advisory on a successful response must arrive too: Azure DevOps announces its delay on a 200, +// before it starts refusing. +func TestAdvisoryOnASuccessfulResponseReachesTheGovernor(t *testing.T) { + origin := gittest.NewRepo(t, signalBranch) + origin.Commit("first commit") + srv := gittest.Serve(t, origin) + srv.Annotate(map[string]string{"Retry-After": "5", "X-RateLimit-Delay": "1.5"}) + + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 8}) + require.NoError(t, err) + + _, err = GetLatestCommitRemote(runtimeContext(g), ListOptions{URL: srv.URL, Branch: signalBranch, HomeDir: t.TempDir()}) + require.NoError(t, err) + assert.Less(t, g.Capacity(signalKey), g.Ceiling(), "the advisory never reached the governor") +} + +// The caller's deadline has to stop an operation the server never answers. A remote call used to +// run with no deadline at all, so a hung server held its worker, and now its governor slot, for +// good. Each case hangs one step of one operation. +func TestHungRemoteStopsAtTheCallersDeadline(t *testing.T) { + const deadline = 300 * time.Millisecond + + cases := map[string]struct { + step gittest.Step + prepare bool + run func(ctx context.Context, url, home string, r *Repo) error + }{ + "listing refs": { + step: gittest.AdvertiseUpload, + run: func(ctx context.Context, url, home string, _ *Repo) error { + _, err := GetLatestCommitRemote(ctx, ListOptions{URL: url, Branch: signalBranch, HomeDir: home}) + return err + }, + }, + "cloning": { + step: gittest.UploadPack, + run: func(ctx context.Context, url, home string, _ *Repo) error { + _, err := Clone(ctx, CloneOptions{URL: url, Branch: signalBranch, HomeDir: home}) + return err + }, + }, + "reading the history": { + step: gittest.UploadPack, + run: func(ctx context.Context, url, home string, _ *Repo) error { + _, err := IsInGitCommitHistory(ctx, ListOptions{URL: url, Branch: signalBranch, HomeDir: home}, "0") + return err + }, + }, + "pushing": { + step: gittest.ReceivePack, + prepare: true, + run: func(ctx context.Context, _, _ string, r *Repo) error { + return r.Push(ctx, "origin", signalBranch, false) + }, + }, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + origin := gittest.NewRepo(t, signalBranch) + origin.Commit("first commit") + srv := gittest.Serve(t, origin) + home := t.TempDir() + + var r *Repo + if tc.prepare { + var err error + r, err = Clone(context.Background(), CloneOptions{URL: srv.URL, Branch: signalBranch, HomeDir: home}) + require.NoError(t, err) + t.Cleanup(func() { _ = r.Cleanup() }) + f, err := r.FS().Create("pushed.txt") + require.NoError(t, err) + require.NoError(t, f.Close()) + _, err = r.Commit(".", "pushed", &IndexOptions{}) + require.NoError(t, err) + } + srv.Hang(tc.step) + + ctx, cancel := context.WithTimeout(context.Background(), deadline) + defer cancel() + + done := make(chan error, 1) + go func() { done <- tc.run(ctx, srv.URL, home, r) }() + + select { + case err := <-done: + require.Error(t, err) + assert.Contains(t, err.Error(), context.DeadlineExceeded.Error()) + case <-time.After(30 * time.Second): + t.Fatal("the operation outlived its deadline: a hung server would hold this worker and its slot for good") + } + }) + } +} + +// Installing the client is process-wide, so it must be exactly as wide as it needs to be: ssh, git +// and file keep go-git's own clients, and both HTTP schemes get ours. A remote on plain http is +// usual for a self-hosted git server. +func TestOnlyHTTPSchemesUseTheSignalClient(t *testing.T) { + for _, scheme := range []string{"http", "https"} { + assert.Same(t, defaultClient, gitclient.Protocols[scheme], "scheme %s", scheme) + } + for _, scheme := range []string{"ssh", "git", "file"} { + _, isSignal := gitclient.Protocols[scheme].(*signalClient) + assert.False(t, isSignal, "scheme %s must keep go-git's own client", scheme) + assert.NotNil(t, gitclient.Protocols[scheme], "scheme %s", scheme) + } +} diff --git a/internal/clients/git/transport.go b/internal/clients/git/transport.go new file mode 100644 index 0000000..2295be6 --- /dev/null +++ b/internal/clients/git/transport.go @@ -0,0 +1,82 @@ +package git + +import ( + "crypto/tls" + "net/http" + + "github.com/go-git/go-git/v5/plumbing/transport" + gitclient "github.com/go-git/go-git/v5/plumbing/transport/client" + githttp "github.com/go-git/go-git/v5/plumbing/transport/http" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit/httpsignal" +) + +// defaultClient is the go-git client for every http and https remote. It is the only way the +// rate-limit governor hears from a git server: go-git does not expose response headers, so the +// signal has to be read in the HTTP transport underneath it. +var defaultClient = newSignalClient(nil) + +// insecureTLS is the base transport for remotes with spec.insecure set. It is shared rather than +// built per session, so those remotes reuse connections like every other remote. +var insecureTLS = newInsecureTLS() + +// init installs defaultClient before main runs, so no goroutine can be inside a go-git operation +// yet and transportMu is not needed. Only the two HTTP schemes change: ssh, git and file keep +// go-git's own clients, and they have no HTTP response to read a signal from. +func init() { + gitclient.InstallProtocol("http", defaultClient) + gitclient.InstallProtocol("https", defaultClient) +} + +func newInsecureTLS() *http.Transport { + t := http.DefaultTransport.(*http.Transport).Clone() + if t.TLSClientConfig == nil { + t.TLSClientConfig = &tls.Config{} + } + t.TLSClientConfig.InsecureSkipVerify = true // #nosec G402 -- only for a remote whose spec.insecure asks for it. + return t +} + +// signalClient is a go-git transport whose HTTP requests go through httpsignal.Transport. A signal +// reaches the governor only when the request carries the context the runtime handed to Connect or +// the ExternalClient method, so every remote operation here must use go-git's Context form. +// +// It exists because githttp.NewClient alone can't do this. When an endpoint sets InsecureSkipTLS, +// go-git asserts that the client's Transport is an *http.Transport so it can clone and reconfigure +// it, and fails the session for any wrapper. So signalClient holds one client per TLS mode and +// clears the option from the endpoint, and go-git never needs to reconfigure anything. +// +// git-provider sets no other per-endpoint transport option (CABundle, ClientCert, ProxyOptions). +// Setting one means adding a mode here: go-git fails the session with the same assertion. +type signalClient struct { + verified transport.Transport + insecure transport.Transport +} + +// newSignalClient returns a client using jar, which may be nil. +func newSignalClient(jar http.CookieJar) *signalClient { + return &signalClient{ + verified: githttp.NewClient(&http.Client{Jar: jar, Transport: httpsignal.NewTransport(nil)}), + insecure: githttp.NewClient(&http.Client{Jar: jar, Transport: httpsignal.NewTransport(insecureTLS)}), + } +} + +// pick returns the client for ep's TLS mode and the endpoint to give it. The endpoint is a copy +// when it changes, because the caller's belongs to go-git. +func (c *signalClient) pick(ep *transport.Endpoint) (transport.Transport, *transport.Endpoint) { + if !ep.InsecureSkipTLS { + return c.verified, ep + } + cleared := *ep + cleared.InsecureSkipTLS = false + return c.insecure, &cleared +} + +func (c *signalClient) NewUploadPackSession(ep *transport.Endpoint, auth transport.AuthMethod) (transport.UploadPackSession, error) { + client, ep := c.pick(ep) + return client.NewUploadPackSession(ep, auth) +} + +func (c *signalClient) NewReceivePackSession(ep *transport.Endpoint, auth transport.AuthMethod) (transport.ReceivePackSession, error) { + client, ep := c.pick(ep) + return client.NewReceivePackSession(ep, auth) +} diff --git a/internal/clients/git/transport_test.go b/internal/clients/git/transport_test.go index 0b6c0b9..05bcc71 100644 --- a/internal/clients/git/transport_test.go +++ b/internal/clients/git/transport_test.go @@ -6,7 +6,6 @@ import ( "github.com/go-git/go-git/v5/plumbing/transport" gitclient "github.com/go-git/go-git/v5/plumbing/transport/client" - githttp "github.com/go-git/go-git/v5/plumbing/transport/http" ) // domain \t httponly \t path \t secure \t expires \t name \t value @@ -117,13 +116,13 @@ func TestTransportRegistryIsRaceFree(t *testing.T) { wg.Wait() - // Every writer must leave go-git's default https client installed, so that + // Every writer must leave defaultClient installed, so that // the readers above are entitled to skip installing anything at all. got, err := gitclient.NewClient(ep) if err != nil { t.Fatal(err) } - if want := githttp.DefaultClient; got != want { - t.Fatalf("https transport not restored: got %#v, want the go-git default", got) + if want := transport.Transport(defaultClient); got != want { + t.Fatalf("https transport not restored: got %#v, want defaultClient", got) } } diff --git a/internal/controllers/common/ratelimitkey/ratelimitkey.go b/internal/controllers/common/ratelimitkey/ratelimitkey.go new file mode 100644 index 0000000..c731304 --- /dev/null +++ b/internal/controllers/common/ratelimitkey/ratelimitkey.go @@ -0,0 +1,57 @@ +// Package ratelimitkey names the rate-limited scope of a git remote, for provider-runtime's +// rate-limit governor. It is shared by the repo and localresource controllers so the key a call is +// gated on and the key its signals are reported under can't drift apart: a signal under any other +// string throttles a key no call is gated on. +package ratelimitkey + +import ( + "context" + "fmt" + "net/url" + "strings" + + commonv1 "github.com/krateo-platformops/provider-runtime/apis/common/v1" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit/httpsignal" +) + +// For returns the scope of calls to rawURL made with the credential ref points at: +// host|namespace/name/key, or the host alone when there is no credential, which is the anonymous +// quota. Git servers count quota per identity, so two credentials on one host are two scopes. The +// host is httpsignal.HostKey of the URL. The credential is named by its reference, never its value, +// because a key that changes when the token is rotated splits one quota into several. +// +// It returns "" for a remote that can't report a signal, which the runtime reads as not gated per +// endpoint: anything but http or https (ssh, git, a local path) has no response headers to read, +// and a URL that doesn't parse never produces a request. +// +// For parses strings and does no I/O, so a RateLimitKeyer can call it. +func For(rawURL string, ref *commonv1.SecretKeySelector) string { + if !isHTTP(rawURL) { + return "" + } + u, err := url.Parse(rawURL) + if err != nil { + return "" + } + host := httpsignal.HostKey(u) + if host == "" { + return "" + } + if ref == nil { + return host + } + return fmt.Sprintf("%s|%s/%s/%s", host, ref.Namespace, ref.Name, ref.Key) +} + +// WithRemote returns the context for calls to one remote: their signals are reported under +// For(rawURL, ref) rather than the bare request host. +func WithRemote(ctx context.Context, rawURL string, ref *commonv1.SecretKeySelector) context.Context { + return httpsignal.WithKey(ctx, For(rawURL, ref)) +} + +// isHTTP compares the scheme by hand because an scp-like remote, git@host:org/repo.git, is not a +// URL and url.Parse refuses it. +func isHTTP(rawURL string) bool { + lower := strings.ToLower(rawURL) + return strings.HasPrefix(lower, "http://") || strings.HasPrefix(lower, "https://") +} diff --git a/internal/controllers/common/ratelimitkey/ratelimitkey_test.go b/internal/controllers/common/ratelimitkey/ratelimitkey_test.go new file mode 100644 index 0000000..cb594e1 --- /dev/null +++ b/internal/controllers/common/ratelimitkey/ratelimitkey_test.go @@ -0,0 +1,88 @@ +package ratelimitkey + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "testing" + + commonv1 "github.com/krateo-platformops/provider-runtime/apis/common/v1" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit/httpsignal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var creds = &commonv1.SecretKeySelector{ + Reference: commonv1.Reference{Name: "git-creds", Namespace: "team-a"}, + Key: "token", +} + +func TestFor(t *testing.T) { + cases := map[string]struct { + url string + ref *commonv1.SecretKeySelector + want string + }{ + "https remote with a credential": {"https://github.com/org/repo.git", creds, "github.com|team-a/git-creds/token"}, + "http remote": {"http://gitea.local/org/repo", creds, "gitea.local|team-a/git-creds/token"}, + "a port is part of the host": {"https://gitea.local:3000/org/repo", creds, "gitea.local:3000|team-a/git-creds/token"}, + "the host is lower-cased": {"https://GitHub.com/org/repo", creds, "github.com|team-a/git-creds/token"}, + "the scheme's case is ignored": {"HTTPS://github.com/org/repo", creds, "github.com|team-a/git-creds/token"}, + "no credential is the anonymous quota": {"https://github.com/org/repo", nil, "github.com"}, + "a password in the URL stays out of the key": { + "https://someone:s3cret@github.com/org/repo", creds, "github.com|team-a/git-creds/token", + }, + "another key of the same secret is another token": { + "https://github.com/org/repo", + &commonv1.SecretKeySelector{Reference: creds.Reference, Key: "other"}, + "github.com|team-a/git-creds/other", + }, + "ssh has no response headers": {"ssh://git@github.com/org/repo.git", creds, ""}, + "scp-like has no response headers": {"git@github.com:org/repo.git", creds, ""}, + "a local path has no server": {"/var/repos/repo.git", creds, ""}, + "file has no server": {"file:///var/repos/repo.git", creds, ""}, + "no URL": {"", creds, ""}, + "a URL with no host": {"https:///org/repo", creds, ""}, + "a URL that does not parse": {"https://bad host/org", creds, ""}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + assert.Equal(t, tc.want, For(tc.url, tc.ref)) + }) + } +} + +// The host part must be the same string httpsignal keys on: docs/usage.md requires a keyer to +// return HostKey of the base URL. +func TestForAgreesWithHostKey(t *testing.T) { + u, err := url.Parse("https://Gitea.Local:3000/org/repo") + require.NoError(t, err) + assert.Equal(t, httpsignal.HostKey(u), For(u.String(), nil)) +} + +type recorder struct{ signals []ratelimit.Signal } + +func (r *recorder) Observe(s ratelimit.Signal) { r.signals = append(r.signals, s) } + +// A signal from the context WithRemote returns carries For's key. Without it, it would carry the +// bare host, and no call would be gated on that. +func TestWithRemoteReportsUnderTheKeyCallsAreGatedOn(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "60") + w.WriteHeader(http.StatusTooManyRequests) + })) + defer srv.Close() + + rec := &recorder{} + ctx := WithRemote(ratelimit.WithSink(context.Background(), rec), srv.URL, creds) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil) + require.NoError(t, err) + resp, err := (&http.Client{Transport: httpsignal.NewTransport(nil)}).Do(req) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + require.Len(t, rec.signals, 1) + assert.Equal(t, For(srv.URL, creds), rec.signals[0].Key) +} diff --git a/internal/controllers/controllers_test.go b/internal/controllers/controllers_test.go new file mode 100644 index 0000000..cd42faf --- /dev/null +++ b/internal/controllers/controllers_test.go @@ -0,0 +1,42 @@ +package controllers + +import ( + "testing" + + "github.com/krateo-platformops/provider-runtime/pkg/controller" + "github.com/krateo-platformops/provider-runtime/pkg/logging" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "k8s.io/client-go/rest" + ctrl "sigs.k8s.io/controller-runtime" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" + + "github.com/krateoplatformops/git-provider/apis" + "github.com/krateoplatformops/git-provider/internal/controllers/common/option" +) + +// Each reconciler that takes the governor registers with it, which raises its ceiling by +// MaxConcurrentReconciles. So the ceiling shows that every controller Setup builds is governed: one +// that forgets the option is silently ungated, and nothing else would fail. +func TestSetupGovernsEveryReconciler(t *testing.T) { + // The manager is never started, so nothing connects to this address. + mgr, err := ctrl.NewManager(&rest.Config{Host: "http://127.0.0.1:1"}, ctrl.Options{ + Metrics: metricsserver.Options{BindAddress: "0"}, + }) + require.NoError(t, err) + require.NoError(t, apis.AddToScheme(mgr.GetScheme())) + + o := controller.Options{ + Logger: logging.NewNopLogger(), + MaxConcurrentReconciles: 5, + RateLimit: controller.RateLimitOptions{ + Enabled: true, + }, + } + require.NoError(t, o.SetupRateLimitGovernor(mgr)) + require.NotNil(t, o.RateLimitGovernor) + + require.NoError(t, Setup(mgr, option.SetupOptions{Controller: option.ControllerOptions{Options: o}})) + + assert.Equal(t, 2*o.MaxConcurrentReconciles, o.RateLimitGovernor.Ceiling()) +} diff --git a/internal/controllers/localresource/localresource.go b/internal/controllers/localresource/localresource.go index cc75cc8..227c5ca 100644 --- a/internal/controllers/localresource/localresource.go +++ b/internal/controllers/localresource/localresource.go @@ -37,6 +37,7 @@ import ( credentialhelper "github.com/krateoplatformops/git-provider/internal/controllers/common/credentialHelper" "github.com/krateoplatformops/git-provider/internal/controllers/common/footer" "github.com/krateoplatformops/git-provider/internal/controllers/common/option" + "github.com/krateoplatformops/git-provider/internal/controllers/common/ratelimitkey" "github.com/krateoplatformops/git-provider/internal/controllers/common/templating" "github.com/krateoplatformops/git-provider/internal/tools/copier" "github.com/krateoplatformops/git-provider/internal/tools/localfs" @@ -73,6 +74,7 @@ func Setup(mgr ctrl.Manager, o option.SetupOptions) error { reconciler.WithLogger(log), reconciler.WithRecorder(plumbingevent.NewAPIRecorder(recorder)), reconciler.WithTimeout(o.Controller.Timeout), + reconciler.WithRateLimitGovernor(o.Controller.RateLimitGovernor), ) git.CommitAuthorEmail = o.Git.CommitAuthorEmail @@ -163,6 +165,19 @@ type external struct { rec record.EventRecorder } +var _ reconciler.RateLimitKeyer = (*external)(nil) + +// RateLimitKey implements reconciler.RateLimitKeyer. Every call of a LocalResource goes to one +// remote with one credential, so one key represents them all. It is on the client and not the +// connector because Connect makes no git call: it reads Secrets from the cluster. +func (e *external) RateLimitKey(mg resource.Managed) (string, error) { + cr, ok := mg.(*localResourcev1alpha1.LocalResource) + if !ok { + return "", errors.New(errNotLocalResource) + } + return ratelimitkey.For(cr.Spec.ToRepo.Url, cr.Spec.ToRepo.Credentials.SecretRef), nil +} + // Observe follows the provider-runtime Ready contract (v1.8.0): every path that returns a nil // error sets Ready. No error here follows a Ready write, so none is wrapped with // reconciler.ReadyAssessed: each one means "could not tell", and the runtime answers it with @@ -209,7 +224,7 @@ func (e *external) Observe(ctx context.Context, mg resource.Managed) (reconciler } var hasGitProviderPreviuslyCommitted bool - if hash, err := git.IsFuncInGitCommitHistory(ctx, git.ListOptions{ + if hash, err := git.IsFuncInGitCommitHistory(ratelimitkey.WithRemote(ctx, cr.Spec.ToRepo.Url, cr.Spec.ToRepo.Credentials.SecretRef), git.ListOptions{ URL: cr.Spec.ToRepo.Url, Auth: e.cfg.ToRepoCreds.Transport, Insecure: e.cfg.Insecure, @@ -403,7 +418,9 @@ func validateFromRefNamespace(refNamespace, crNamespace string) error { func (e *external) SyncLocalResources(ctx context.Context, cr *localResourcev1alpha1.LocalResource, commitMessage string) error { spec := cr.Spec.DeepCopy() - toRepo, err := git.Clone(git.CloneOptions{ + toCtx := ratelimitkey.WithRemote(ctx, spec.ToRepo.Url, spec.ToRepo.Credentials.SecretRef) + + toRepo, err := git.Clone(toCtx, git.CloneOptions{ URL: spec.ToRepo.Url, Auth: e.cfg.ToRepoCreds.Transport, Insecure: e.cfg.Insecure, @@ -588,7 +605,7 @@ func (e *external) SyncLocalResources(ctx context.Context, cr *localResourcev1al } log.Debug("Target LocalResource committed", "branch", toRepo.CurrentBranch(), "commitId", toLocalResourceCommitId) - err = toRepo.Push("origin", toRepo.CurrentBranch(), e.cfg.Insecure) + err = toRepo.Push(toCtx, "origin", toRepo.CurrentBranch(), e.cfg.Insecure) if err != nil { return fmt.Errorf("unable to push target LocalResource: %w", err) } diff --git a/internal/controllers/localresource/ratelimit_test.go b/internal/controllers/localresource/ratelimit_test.go new file mode 100644 index 0000000..97a2ffe --- /dev/null +++ b/internal/controllers/localresource/ratelimit_test.go @@ -0,0 +1,198 @@ +package localresource + +import ( + "context" + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + commonv1 "github.com/krateo-platformops/provider-runtime/apis/common/v1" + "github.com/krateo-platformops/provider-runtime/pkg/logging" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit" + "github.com/krateo-platformops/provider-runtime/pkg/reconciler" + "github.com/krateo-platformops/provider-runtime/pkg/reconciler/reconcilertest" + "github.com/krateo-platformops/provider-runtime/pkg/resource" + resourcefake "github.com/krateo-platformops/provider-runtime/pkg/resource/fake" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/events" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "github.com/krateoplatformops/git-provider/apis" + localresourcev1alpha1 "github.com/krateoplatformops/git-provider/apis/localresource/v1alpha1" + "github.com/krateoplatformops/git-provider/internal/clients/git/gittest" + credentialhelper "github.com/krateoplatformops/git-provider/internal/controllers/common/credentialHelper" +) + +const governorBranch = "main" + +// credentials is an anonymous git client: the fake server takes any request. +var credentials = credentialhelper.Credentials{} + +func resetIn(d time.Duration) string { + return strconv.FormatInt(time.Now().Add(d).Unix(), 10) +} + +func secretRef(key string) *commonv1.SecretKeySelector { + return &commonv1.SecretKeySelector{ + Reference: commonv1.Reference{Name: "git-creds", Namespace: "team-a"}, + Key: key, + } +} + +func localResourceFor(url string, creds localresourcev1alpha1.Credentials) *localresourcev1alpha1.LocalResource { + return &localresourcev1alpha1.LocalResource{ + ObjectMeta: metav1.ObjectMeta{Name: "publish-000", Namespace: "team-a"}, + Spec: localresourcev1alpha1.LocalResourceSpec{ + ToRepo: localresourcev1alpha1.LocalResourceOpts{Url: url, Branch: governorBranch, Credentials: creds}, + }, + } +} + +func TestRateLimitKey(t *testing.T) { + const url = "https://github.com/org/repo.git" + const key = "github.com|team-a/git-creds/token" + cases := map[string]struct { + url string + creds localresourcev1alpha1.Credentials + want string + }{ + "basic: the username does not change the identity's key": {url, localresourcev1alpha1.Credentials{ + AuthMethod: "basic", SecretRef: secretRef("token"), UsernameRef: secretRef("username"), + }, key}, + "bearer": {url, localresourcev1alpha1.Credentials{AuthMethod: "bearer", SecretRef: secretRef("token")}, key}, + "cookiefile": {url, localresourcev1alpha1.Credentials{AuthMethod: "cookiefile", SecretRef: secretRef("token")}, key}, + "no credential": {url, localresourcev1alpha1.Credentials{}, "github.com"}, + "ssh remote": {"git@github.com:org/repo.git", localresourcev1alpha1.Credentials{SecretRef: secretRef("token")}, ""}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + // A client with no clients of its own: reading the Secret, or anything else in the + // cluster, panics on the nil client. + got, err := (&external{}).RateLimitKey(localResourceFor(tc.url, tc.creds)) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +// The runtime calls RateLimitKey before every gated call, also while the endpoint is throttled. So +// it can't touch the cluster or the git server, and the value of the Secret can't be in the key. +func TestRateLimitKeyDoesNoIO(t *testing.T) { + var requests int + srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { requests++ })) + defer srv.Close() + + cr := localResourceFor(srv.URL, localresourcev1alpha1.Credentials{AuthMethod: "bearer", SecretRef: secretRef("token")}) + + // The external has no clients, so a read of the Secret panics on the nil client. + got, err := (&external{}).RateLimitKey(cr) + require.NoError(t, err) + assert.NotEmpty(t, got) + assert.NotContains(t, got, "a-token") + assert.Zero(t, requests, "RateLimitKey reached the git server") +} + +func TestRateLimitKeyRejectsAnotherKind(t *testing.T) { + _, err := (&external{}).RateLimitKey(&resourcefake.Managed{}) + require.Error(t, err) +} + +func governedReconciler(t *testing.T, cr *localresourcev1alpha1.LocalResource, g *ratelimit.Governor) (*reconciler.Reconciler, client.Client) { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, apis.AddToScheme(scheme)) + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "git-creds", Namespace: "team-a"}, + Data: map[string][]byte{"token": []byte("a-token")}, + } + kube := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cr, secret).WithStatusSubresource(cr).Build() + + r := reconciler.NewReconciler(&resourcefake.Manager{Client: kube, Scheme: scheme}, + resource.ManagedKind(localresourcev1alpha1.LocalResourceGroupVersionKind), + reconciler.WithExternalConnecter(&connector{ + kube: kube, + log: logging.NewNopLogger(), + recorder: events.NewFakeRecorder(100), + homeDir: t.TempDir(), + }), + reconciler.WithTimeout(30*time.Second), + reconciler.WithRateLimitGovernor(g), + ) + return r, kube +} + +// The chain the governor depends on, end to end: a real server refuses a real go-git operation run +// by the real connector and reconciler, and the next pass is deferred without touching the server. +// It fails if the context stops reaching the request, if the transport isn't installed, or if the +// key the calls are gated on is not the key the signal is reported under. +func TestThrottledRemoteDefersTheNextPass(t *testing.T) { + for name, tc := range map[string]struct { + status int + header map[string]string + }{ + "429 with Retry-After": {http.StatusTooManyRequests, map[string]string{"Retry-After": "120"}}, + "403 with X-RateLimit-Remaining 0": {http.StatusForbidden, map[string]string{ + "X-RateLimit-Remaining": "0", + "X-RateLimit-Reset": resetIn(2 * time.Minute), + }}, + } { + t.Run(name, func(t *testing.T) { + repo := gittest.NewRepo(t, governorBranch) + repo.Commit("first commit") + srv := gittest.Serve(t, repo) + srv.Refuse(tc.status, tc.header, gittest.AllSteps...) + + cr := localResourceFor(srv.URL, localresourcev1alpha1.Credentials{AuthMethod: "bearer", SecretRef: secretRef("token")}) + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 1}) + require.NoError(t, err) + r, kube := governedReconciler(t, cr, g) + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: cr.Name, Namespace: cr.Namespace}} + + // The pass that gets the refusal is still an error. + _, err = r.Reconcile(context.Background(), req) + require.Error(t, err) + reached := srv.Requests() + require.NotZero(t, reached) + + res, err := r.Reconcile(context.Background(), req) + require.NoError(t, err, "a deferral is not a failure") + assert.Greater(t, res.RequeueAfter, time.Minute, "the requeue must honour the server's deadline") + assert.LessOrEqual(t, res.RequeueAfter, 2*time.Minute+time.Second) + assert.Equal(t, reached, srv.Requests(), "a deferred pass reached the git server") + + got := &localresourcev1alpha1.LocalResource{} + require.NoError(t, kube.Get(context.Background(), req.NamespacedName, got)) + synced := got.GetCondition(commonv1.TypeSynced) + assert.Equal(t, commonv1.ReasonReconcileDeferred, synced.Reason) + assert.Contains(t, synced.Message, "|team-a/git-creds/token is rate-limited by the server") + }) + } +} + +// reconcilertest's conformance check, through Observe. It is the one provider-runtime asks every +// provider to run: a provider that never reports looks exactly like an endpoint never limited. +func TestReportsRateLimits(t *testing.T) { + reconcilertest.AssertReportsRateLimits(t, func(ctx context.Context, baseURL string) error { + repo := gittest.NewRepo(t, governorBranch) + repo.Commit("first commit") + srv := gittest.Serve(t, repo) + srv.Replay(t, baseURL) + + e := &external{log: logging.NewNopLogger(), cfg: &gitClientOpts{ + ToRepoCreds: &credentials, + HomeDir: t.TempDir(), + }} + _, err := e.Observe(ctx, localResourceFor(srv.URL, localresourcev1alpha1.Credentials{SecretRef: secretRef("token")})) + return err + }) +} diff --git a/internal/controllers/repo/ratelimit_test.go b/internal/controllers/repo/ratelimit_test.go new file mode 100644 index 0000000..968a5fe --- /dev/null +++ b/internal/controllers/repo/ratelimit_test.go @@ -0,0 +1,430 @@ +package repo + +import ( + "context" + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + plumbingevent "github.com/krateo-platformops/plumbing/kubeutil/event" + commonv1 "github.com/krateo-platformops/provider-runtime/apis/common/v1" + "github.com/krateo-platformops/provider-runtime/pkg/logging" + "github.com/krateo-platformops/provider-runtime/pkg/ratelimit" + "github.com/krateo-platformops/provider-runtime/pkg/reconciler" + "github.com/krateo-platformops/provider-runtime/pkg/reconciler/reconcilertest" + "github.com/krateo-platformops/provider-runtime/pkg/resource" + resourcefake "github.com/krateo-platformops/provider-runtime/pkg/resource/fake" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/events" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "github.com/krateoplatformops/git-provider/apis" + repov1alpha1 "github.com/krateoplatformops/git-provider/apis/repo/v1alpha1" + "github.com/krateoplatformops/git-provider/internal/clients/git/gittest" + "github.com/krateoplatformops/git-provider/internal/controllers/common/ratelimitkey" +) + +func secretRef(name string) *commonv1.SecretKeySelector { + return &commonv1.SecretKeySelector{ + Reference: commonv1.Reference{Name: name, Namespace: "team-a"}, + Key: "token", + } +} + +func repoBetween(from, to repov1alpha1.RepoOpts) *repov1alpha1.Repo { + return &repov1alpha1.Repo{ + ObjectMeta: metav1.ObjectMeta{Name: "sample", Namespace: "team-a"}, + Spec: repov1alpha1.RepoSpec{FromRepo: repov1alpha1.FromRepoOpts{RepoOpts: from}, ToRepo: to}, + } +} + +func remote(url, secret string) repov1alpha1.RepoOpts { + return repov1alpha1.RepoOpts{Url: url, Branch: branch, SecretRef: secretRef(secret)} +} + +func TestRateLimitKeys(t *testing.T) { + const github = "https://github.com/org/repo" + cases := map[string]struct { + from, to repov1alpha1.RepoOpts + want []string + }{ + "one host and one credential is one scope": { + remote(github+"-a", "gh-token"), remote(github+"-b", "gh-token"), + []string{"github.com|team-a/gh-token/token"}, + }, + "the auth method and username do not change the scope": { + repov1alpha1.RepoOpts{Url: github, SecretRef: secretRef("gh-token"), AuthMethod: "bearer"}, + repov1alpha1.RepoOpts{Url: github, SecretRef: secretRef("gh-token"), UsernameRef: secretRef("gh-user")}, + []string{"github.com|team-a/gh-token/token"}, + }, + "two credentials on one host are two scopes": { + remote(github, "read-token"), remote(github, "write-token"), + []string{"github.com|team-a/read-token/token", "github.com|team-a/write-token/token"}, + }, + "two hosts are two scopes": { + remote("https://source.developers.google.com/p/r", "gcp-cookie"), remote(github, "git-provider"), + []string{"source.developers.google.com|team-a/gcp-cookie/token", "github.com|team-a/git-provider/token"}, + }, + "anonymous on both ends of one host": { + repov1alpha1.RepoOpts{Url: github}, repov1alpha1.RepoOpts{Url: github}, []string{"github.com"}, + }, + "ssh has no signals to gate on": { + remote("git@github.com:org/repo.git", "gh-token"), remote("git@github.com:org/repo.git", "gh-token"), []string{""}, + }, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + // A client with no clients of its own: reading a Secret panics on the nil client. + got, err := (&external{}).RateLimitKeys(repoBetween(tc.from, tc.to)) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +// Each key must be the exact string that remote's signals carry, or a refusal throttles a key no +// call is gated on. +func TestRateLimitKeysAreTheKeysSignalsReport(t *testing.T) { + cr := repoBetween(remote("https://a.example.com/r", "read-token"), remote("https://b.example.com/r", "write-token")) + got, err := (&external{}).RateLimitKeys(cr) + require.NoError(t, err) + assert.Equal(t, []string{keyOf(cr, true), keyOf(cr, false)}, got) +} + +// The runtime calls RateLimitKeys before every gated call, also while the endpoint is throttled. +func TestRateLimitKeysDoesNoIO(t *testing.T) { + var requests int + srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { requests++ })) + defer srv.Close() + + // The external has no clients, so a read of the Secret panics on the nil client. + got, err := (&external{}).RateLimitKeys(repoBetween(remote(srv.URL, "gh-token"), remote(srv.URL, "other-token"))) + require.NoError(t, err) + assert.Len(t, got, 2) + assert.Zero(t, requests, "RateLimitKeys reached the git server") +} + +func TestRateLimitKeysRejectsAnotherKind(t *testing.T) { + _, err := (&external{}).RateLimitKeys(&resourcefake.Managed{}) + require.Error(t, err) +} + +func governedReconciler(t *testing.T, cr *repov1alpha1.Repo, g *ratelimit.Governor, extra ...reconciler.ReconcilerOption) (*reconciler.Reconciler, client.Client) { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, apis.AddToScheme(scheme)) + objects := []client.Object{cr} + for _, name := range []string{"gh-token", "read-token", "write-token"} { + objects = append(objects, &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "team-a"}, + Data: map[string][]byte{"token": []byte("a-token")}, + }) + } + kube := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).WithStatusSubresource(cr).Build() + + r := reconciler.NewReconciler(&resourcefake.Manager{Client: kube, Scheme: scheme}, + resource.ManagedKind(repov1alpha1.RepoGroupVersionKind), + append([]reconciler.ReconcilerOption{ + reconciler.WithExternalConnecter(&connector{ + kube: kube, + log: logging.NewNopLogger(), + recorder: events.NewFakeRecorder(1000), + }), + reconciler.WithTimeout(30 * time.Second), + reconciler.WithRateLimitGovernor(g), + }, extra...)..., + ) + return r, kube +} + +// The chain the governor depends on, end to end, for each kind of call a Repo makes to its remotes: +// a real server refuses a real go-git operation run by the real connector and reconciler, and the +// next pass is deferred without touching the server. Each case fails on its own if that operation +// drops the context, because the signal is then lost with no error. +func TestThrottledRemoteDefersTheNextPass(t *testing.T) { + throttles := map[string]struct { + status int + header map[string]string + }{ + "429 with Retry-After": {http.StatusTooManyRequests, map[string]string{"Retry-After": "120"}}, + "403 with X-RateLimit-Remaining 0": {http.StatusForbidden, map[string]string{ + "X-RateLimit-Remaining": "0", + "X-RateLimit-Reset": strconv.FormatInt(time.Now().Add(2*time.Minute).Unix(), 10), + }}, + } + calls := map[string]struct { + step gittest.Step + // synced is a Repo that already pushed, so its pass observes the remotes; otherwise the + // pass creates, which syncs them. + synced bool + }{ + "Observe listing fromRepo": {step: gittest.AdvertiseUpload, synced: true}, + "Observe cloning toRepo": {step: gittest.UploadPack, synced: true}, + "Create cloning the remotes": {step: gittest.UploadPack}, + "Create pushing to toRepo": {step: gittest.AdvertiseReceive}, + "Create uploading the push": {step: gittest.ReceivePack}, + } + for throttleName, th := range throttles { + for callName, call := range calls { + t.Run(throttleName+"/"+callName, func(t *testing.T) { + git := gittest.NewRepo(t, branch) + head := git.Commit("first commit") + srv := gittest.Serve(t, git) + srv.Refuse(th.status, th.header, call.step) + + // One host and one credential, so the Repo is keyed. The toRepo branch has + // nothing of fromRepo's, so a sync has something to push. + cr := repoBetween(remote(srv.URL, "gh-token"), remote(srv.URL, "gh-token")) + cr.Spec.FromRepo.Path = "/" + cr.Spec.ToRepo.Path = "/copied" + if call.synced { + cr.Status.TargetCommitId = head + cr.Status.OriginCommitId = head + cr.Status.TargetBranch = branch + cr.Status.OriginBranch = branch + } + + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 1}) + require.NoError(t, err) + r, kube := governedReconciler(t, cr, g) + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: cr.Name, Namespace: cr.Namespace}} + + // The pass that gets the refusal is still an error. + _, err = r.Reconcile(context.Background(), req) + require.Error(t, err) + reached := srv.Requests() + + res, err := r.Reconcile(context.Background(), req) + require.NoError(t, err, "a deferral is not a failure") + assert.Greater(t, res.RequeueAfter, time.Minute, "the requeue must honour the server's deadline") + assert.LessOrEqual(t, res.RequeueAfter, 2*time.Minute+time.Second) + assert.Equal(t, reached, srv.Requests(), "a deferred pass reached the git server") + + got := &repov1alpha1.Repo{} + require.NoError(t, kube.Get(context.Background(), req.NamespacedName, got)) + synced := got.GetCondition(commonv1.TypeSynced) + assert.Equal(t, commonv1.ReasonReconcileDeferred, synced.Reason) + assert.Contains(t, synced.Message, "|team-a/gh-token/token is rate-limited by the server") + }) + } + } +} + +// A Repo between two remotes is the case one key could not describe. Whichever remote refuses, the +// next pass is deferred before it touches either of them, and the object names the remote that +// refused. Each case runs the real reconciler and go-git against two real servers, so it fails if +// the Repo is gated on only one of its remotes. +func TestMixedRepoIsDeferredByEitherRemote(t *testing.T) { + calls := map[string]struct { + fromRefuses bool + step gittest.Step + synced bool + }{ + "Observe, fromRepo refuses the list": {fromRefuses: true, step: gittest.AdvertiseUpload, synced: true}, + "Observe, toRepo refuses the clone": {step: gittest.AdvertiseUpload, synced: true}, + "Create, toRepo refuses the clone": {step: gittest.AdvertiseUpload}, + "Create, fromRepo refuses the clone": {fromRefuses: true, step: gittest.UploadPack}, + "Create, toRepo refuses the push": {step: gittest.ReceivePack}, + } + for name, tc := range calls { + t.Run(name, func(t *testing.T) { + f := newObserveFixture(t) + f.origin.Commit("a file the target does not have yet, so a sync has something to push") + fromSrv, toSrv := gittest.Serve(t, f.origin), gittest.Serve(t, f.target) + + cr := f.syncedRepo() + cr.Spec.FromRepo.Url, cr.Spec.FromRepo.SecretRef = fromSrv.URL, secretRef("read-token") + cr.Spec.ToRepo.Url, cr.Spec.ToRepo.SecretRef = toSrv.URL, secretRef("write-token") + if !tc.synced { + cr.Status = repov1alpha1.RepoStatus{} + } + + refused := toSrv + if tc.fromRefuses { + refused = fromSrv + } + refused.Refuse(http.StatusTooManyRequests, map[string]string{"Retry-After": "120"}, tc.step) + refusedKey, healthyKey := keyOf(cr, tc.fromRefuses), keyOf(cr, !tc.fromRefuses) + require.NotEqual(t, refusedKey, healthyKey) + + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 1}) + require.NoError(t, err) + r, kube := governedReconciler(t, cr, g) + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: cr.Name, Namespace: cr.Namespace}} + + // The pass that gets the refusal is still an error. + _, err = r.Reconcile(context.Background(), req) + require.Error(t, err) + fromRequests, toRequests := fromSrv.Requests(), toSrv.Requests() + + res, err := r.Reconcile(context.Background(), req) + require.NoError(t, err, "a deferral is not a failure") + assert.Greater(t, res.RequeueAfter, time.Minute, "the requeue must honour the server's deadline") + assert.Equal(t, fromRequests, fromSrv.Requests(), "a deferred pass reached fromRepo") + assert.Equal(t, toRequests, toSrv.Requests(), "a deferred pass reached toRepo") + + got := &repov1alpha1.Repo{} + require.NoError(t, kube.Get(context.Background(), req.NamespacedName, got)) + synced := got.GetCondition(commonv1.TypeSynced) + assert.Equal(t, commonv1.ReasonReconcileDeferred, synced.Reason) + assert.Contains(t, synced.Message, refusedKey+" is rate-limited by the server") + assert.NotContains(t, synced.Message, healthyKey) + + // The healthy remote stays open to whatever else is gated on it. + release, d := g.Acquire(context.Background(), healthyKey) + require.Nil(t, d, "the refusal throttled the healthy remote") + release() + }) + } +} + +// A server that never answers must not keep a governor slot past the reconcile timeout. The +// governor here allows one call, so the Acquire below succeeds only if the pass gave its slot back. +func TestHungRemoteGivesItsSlotBackAtTheTimeout(t *testing.T) { + git := gittest.NewRepo(t, branch) + head := git.Commit("first commit") + srv := gittest.Serve(t, git) + srv.Hang(gittest.UploadPack) + + cr := repoBetween(remote(srv.URL, "gh-token"), remote(srv.URL, "gh-token")) + cr.Status.TargetCommitId = head + cr.Status.OriginCommitId = head + keys, err := (&external{}).RateLimitKeys(cr) + require.NoError(t, err) + require.Len(t, keys, 1) + + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 1}) + require.NoError(t, err) + r, _ := governedReconciler(t, cr, g, reconciler.WithTimeout(300*time.Millisecond)) + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: cr.Name, Namespace: cr.Namespace}} + + done := make(chan error, 1) + go func() { + _, err := r.Reconcile(context.Background(), req) + done <- err + }() + select { + case err := <-done: + require.Error(t, err) + case <-time.After(30 * time.Second): + t.Fatal("the pass outlived the reconcile timeout, so it still holds its governor slot") + } + + release, d := g.AcquireAll(context.Background(), keys) + require.Nil(t, d, "the timed-out pass kept its slot") + release() +} + +// Whatever the object is gated on, a refusal has to land on the remote that refused: a Repo's two +// remotes are often two quotas, and other objects are gated on each of them. +func TestSignalsAreReportedPerRemote(t *testing.T) { + for name, fromRefuses := range map[string]bool{"fromRepo refuses": true, "toRepo refuses": false} { + t.Run(name, func(t *testing.T) { + f := newObserveFixture(t) + fromSrv, toSrv := gittest.Serve(t, f.origin), gittest.Serve(t, f.target) + cr := f.syncedRepo() + cr.Spec.FromRepo.Url, cr.Spec.FromRepo.SecretRef = fromSrv.URL, secretRef("read-token") + cr.Spec.ToRepo.Url, cr.Spec.ToRepo.SecretRef = toSrv.URL, secretRef("write-token") + + refused := toSrv + if fromRefuses { + refused = fromSrv + } + refused.Refuse(http.StatusTooManyRequests, map[string]string{"Retry-After": "120"}, gittest.AllSteps...) + refusedKey, healthyKey := keyOf(cr, fromRefuses), keyOf(cr, !fromRefuses) + + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 1}) + require.NoError(t, err) + _, err = newObserveExternal().Observe(ratelimit.WithSink(context.Background(), g), cr) + require.Error(t, err) + + _, d := g.Acquire(context.Background(), refusedKey) + require.NotNil(t, d, "the refusal never reached the key of the remote that refused") + assert.Equal(t, ratelimit.DeferThrottled, d.Reason) + + release, d := g.Acquire(context.Background(), healthyKey) + require.Nil(t, d, "the healthy remote was throttled by the other one's refusal") + release() + }) + } +} + +// The same for a sync, which makes the most calls of any pass and touches each remote more than +// once. Each remote has its own server here, so a refusal reported under the other remote's key +// fails, which a Repo with one scope could never show. +func TestSyncReportsSignalsPerRemote(t *testing.T) { + cases := map[string]struct { + fromRefuses bool + step gittest.Step + }{ + "cloning toRepo": {false, gittest.AdvertiseUpload}, + "cloning fromRepo": {true, gittest.AdvertiseUpload}, + "listing toRepo's refs": {false, gittest.AdvertiseReceive}, + "pushing to toRepo": {false, gittest.ReceivePack}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + f := newObserveFixture(t) + f.origin.Commit("a file the target does not have yet, so the sync has something to push") + fromSrv, toSrv := gittest.Serve(t, f.origin), gittest.Serve(t, f.target) + cr := f.syncedRepo() + cr.Spec.FromRepo.Url, cr.Spec.FromRepo.SecretRef = fromSrv.URL, secretRef("read-token") + cr.Spec.ToRepo.Url, cr.Spec.ToRepo.SecretRef = toSrv.URL, secretRef("write-token") + + refused := toSrv + if tc.fromRefuses { + refused = fromSrv + } + refused.Refuse(http.StatusTooManyRequests, map[string]string{"Retry-After": "120"}, tc.step) + + g, err := ratelimit.New(ratelimit.Options{MaxConcurrentReconciles: 1}) + require.NoError(t, err) + e := newObserveExternal() + e.rec = *plumbingevent.NewAPIRecorder(events.NewFakeRecorder(100)) + + err = e.SyncRepos(ratelimit.WithSink(context.Background(), g), cr, "first commit") + require.Error(t, err) + + _, d := g.Acquire(context.Background(), keyOf(cr, tc.fromRefuses)) + require.NotNil(t, d, "the refusal never reached the key of the remote that refused") + assert.Equal(t, ratelimit.DeferThrottled, d.Reason) + + release, d := g.Acquire(context.Background(), keyOf(cr, !tc.fromRefuses)) + require.Nil(t, d, "the healthy remote was throttled by the other one's refusal") + release() + }) + } +} + +func keyOf(cr *repov1alpha1.Repo, from bool) string { + if from { + return ratelimitkey.For(cr.Spec.FromRepo.Url, cr.Spec.FromRepo.SecretRef) + } + return ratelimitkey.For(cr.Spec.ToRepo.Url, cr.Spec.ToRepo.SecretRef) +} + +// reconcilertest's conformance check, through Observe. It is the one provider-runtime asks every +// provider to run: a provider that never reports looks exactly like an endpoint never limited. +func TestReportsRateLimits(t *testing.T) { + reconcilertest.AssertReportsRateLimits(t, func(ctx context.Context, baseURL string) error { + f := newObserveFixture(t) + fromSrv, toSrv := gittest.Serve(t, f.origin), gittest.Serve(t, f.target) + fromSrv.Replay(t, baseURL) + toSrv.Replay(t, baseURL) + + cr := f.syncedRepo() + cr.Spec.FromRepo.Url, cr.Spec.ToRepo.Url = fromSrv.URL, toSrv.URL + _, err := newObserveExternal().Observe(ctx, cr) + return err + }) +} diff --git a/internal/controllers/repo/repo.go b/internal/controllers/repo/repo.go index f7d3575..22ac362 100644 --- a/internal/controllers/repo/repo.go +++ b/internal/controllers/repo/repo.go @@ -23,6 +23,7 @@ import ( repov1alpha1 "github.com/krateoplatformops/git-provider/apis/repo/v1alpha1" "github.com/krateoplatformops/git-provider/internal/clients/git" "github.com/krateoplatformops/git-provider/internal/controllers/common/option" + "github.com/krateoplatformops/git-provider/internal/controllers/common/ratelimitkey" "github.com/krateoplatformops/git-provider/internal/controllers/common/templating" "github.com/krateoplatformops/git-provider/internal/tools/copier" "github.com/krateoplatformops/git-provider/internal/tools/template" @@ -66,6 +67,7 @@ func Setup(mgr ctrl.Manager, o option.SetupOptions) error { reconciler.WithLogger(log), reconciler.WithRecorder(plumbingevent.NewAPIRecorder(recorder)), reconciler.WithTimeout(o.Controller.Timeout), + reconciler.WithRateLimitGovernor(o.Controller.RateLimitGovernor), ) return ctrl.NewControllerManagedBy(mgr). @@ -121,6 +123,29 @@ type external struct { rec plumbingevent.APIRecorder } +var _ reconciler.RateLimitKeysProvider = (*external)(nil) + +// RateLimitKeys implements reconciler.RateLimitKeysProvider, because every external call of a Repo +// talks to both of its remotes: Observe lists fromRepo and clones toRepo, and a sync clones both +// and pushes to toRepo. Each key is the one that remote's signals are reported under, so a call is +// deferred when either remote is refusing, and takes a slot in both. Remotes on one host with one +// credential are one scope, so they return one key. +// +// It is on the client and not the connector because Connect makes no git call: it reads Secrets +// from the cluster. Gating Connect on these keys would count calls that never reach those remotes. +func (e *external) RateLimitKeys(mg resource.Managed) ([]string, error) { + cr, ok := mg.(*repov1alpha1.Repo) + if !ok { + return nil, errNotRepo + } + from := ratelimitkey.For(cr.Spec.FromRepo.Url, cr.Spec.FromRepo.SecretRef) + to := ratelimitkey.For(cr.Spec.ToRepo.Url, cr.Spec.ToRepo.SecretRef) + if from == to { + return []string{to}, nil + } + return []string{from, to}, nil +} + // Observe follows the provider-runtime Ready contract (v1.8.0): every path that returns a nil // error sets Ready, and an error returned after Ready was set is wrapped with // reconciler.ReadyAssessed. A plain error means "could not tell", and the runtime answers it with @@ -175,7 +200,7 @@ func (e *external) Observe(ctx context.Context, mg resource.Managed) (reconciler ResourceUpToDate: true, }, nil } - latestCommit, err := git.GetLatestCommitRemote(git.ListOptions{ + latestCommit, err := git.GetLatestCommitRemote(ratelimitkey.WithRemote(ctx, cr.Spec.FromRepo.Url, cr.Spec.FromRepo.SecretRef), git.ListOptions{ URL: cr.Spec.FromRepo.Url, Auth: e.cfg.FromRepoCreds, Insecure: e.cfg.Insecure, @@ -189,7 +214,7 @@ func (e *external) Observe(ctx context.Context, mg resource.Managed) (reconciler return reconciler.ExternalObservation{}, err } - isTargetRepoSynced, err := git.IsInGitCommitHistoryContext(ctx, git.ListOptions{ + isTargetRepoSynced, err := git.IsInGitCommitHistory(ratelimitkey.WithRemote(ctx, cr.Spec.ToRepo.Url, cr.Spec.ToRepo.SecretRef), git.ListOptions{ URL: cr.Spec.ToRepo.Url, Auth: e.cfg.ToRepoCreds, Insecure: e.cfg.Insecure, @@ -359,7 +384,10 @@ func (e *external) SyncRepos(ctx context.Context, cr *repov1alpha1.Repo, commitM altBranch = ptr.To(spec.ToRepo.CloneFromBranch) } - toRepo, err := git.Clone(git.CloneOptions{ + toCtx := ratelimitkey.WithRemote(ctx, spec.ToRepo.Url, spec.ToRepo.SecretRef) + fromCtx := ratelimitkey.WithRemote(ctx, spec.FromRepo.Url, spec.FromRepo.SecretRef) + + toRepo, err := git.Clone(toCtx, git.CloneOptions{ URL: spec.ToRepo.Url, Auth: e.cfg.ToRepoCreds, Insecure: e.cfg.Insecure, @@ -378,7 +406,7 @@ func (e *external) SyncRepos(ctx context.Context, cr *repov1alpha1.Repo, commitM e.rec.Event(cr, plumbingevent.Normal("TargetRepoCloned", "Reconciling", fmt.Sprintf("Successfully cloned target repo: %s", spec.ToRepo.Url))) e.log.Debug(fmt.Sprintf("Target repo on branch %s", toRepo.CurrentBranch())) - fromRepo, err := git.Clone(git.CloneOptions{ + fromRepo, err := git.Clone(fromCtx, git.CloneOptions{ URL: spec.FromRepo.Url, Auth: e.cfg.FromRepoCreds, Insecure: e.cfg.Insecure, @@ -506,7 +534,7 @@ func (e *external) SyncRepos(ctx context.Context, cr *repov1alpha1.Repo, commitM e.log.Info("Target repo committed", "branch", toRepo.CurrentBranch(), "commitId", toRepoCommitId) e.rec.Event(cr, plumbingevent.Normal("RepoCommitSuccess", "Reconciling", fmt.Sprintf("Target repo committed on branch %s", toRepo.CurrentBranch()))) - err = toRepo.Push("origin", toRepo.CurrentBranch(), e.cfg.Insecure) + err = toRepo.Push(toCtx, "origin", toRepo.CurrentBranch(), e.cfg.Insecure) if err != nil { return e.failSync(ctx, cr, fmt.Errorf("unable to push target repo: %w", err)) } diff --git a/internal/controllers/repo/repo_test.go b/internal/controllers/repo/repo_test.go index 632639d..16231af 100644 --- a/internal/controllers/repo/repo_test.go +++ b/internal/controllers/repo/repo_test.go @@ -344,7 +344,7 @@ func deleteGiteaRepo(t *testing.T, name string) { func cloneGiteaRepo(t *testing.T, repoName, branch string) *gitclient.Repo { t.Helper() - repo, err := gitclient.Clone(gitclient.CloneOptions{ + repo, err := gitclient.Clone(context.Background(), gitclient.CloneOptions{ URL: giteaRepoURL(repoName), Auth: gitHTTPAuth(), Insecure: true, @@ -412,7 +412,7 @@ func commitFilesToRepo(t *testing.T, repoName, branch, message string, files map ToPath: "/", }) require.NoError(t, err) - require.NoError(t, repo.Push("origin", branch, true)) + require.NoError(t, repo.Push(context.Background(), "origin", branch, true)) commitID, err := repo.GetLatestCommit(repo.CurrentBranch()) require.NoError(t, err) @@ -442,7 +442,7 @@ func assertRemoteFileAbsent(t *testing.T, repoName, branch, path string) { func latestRemoteCommit(t *testing.T, repoName, branch string) string { t.Helper() - commitID, err := gitclient.GetLatestCommitRemote(gitclient.ListOptions{ + commitID, err := gitclient.GetLatestCommitRemote(context.Background(), gitclient.ListOptions{ URL: giteaRepoURL(repoName), Auth: gitHTTPAuth(), Insecure: true, diff --git a/main.go b/main.go index 9eef14b..c696a87 100644 --- a/main.go +++ b/main.go @@ -40,6 +40,7 @@ func main() { maxErrorRetryInterval := flag.Duration("max-error-retry-interval", env.Duration(fmt.Sprintf("%s_MAX_ERROR_RETRY_INTERVAL", envVarPrefix), 1*time.Minute), "The maximum interval between retries when an error occurs. This should be less than the half of the poll interval.") minErrorRetryInterval := flag.Duration("min-error-retry-interval", env.Duration(fmt.Sprintf("%s_MIN_ERROR_RETRY_INTERVAL", envVarPrefix), 1*time.Second), "The minimum interval between retries when an error occurs. This should be less than max-error-retry-interval.") timeout := flag.Duration("timeout", env.Duration(fmt.Sprintf("%s_TIMEOUT", envVarPrefix), 4*time.Minute), "The timeout for each reconcile.") + rateLimitGovernor := flag.Bool("rate-limit-governor", env.Bool(fmt.Sprintf("%s_RATE_LIMIT_GOVERNOR", envVarPrefix), false), "Slow reconciles down per remote and credential when a git server signals a rate limit.") gitCommitAuthorName := flag.String("git-commit-author-name", env.String(fmt.Sprintf("%s_GIT_COMMIT_AUTHOR_NAME", envVarPrefix), "krateo-git-provider"), "The name to use for git commits.") gitCommitAuthorEmail := flag.String("git-commit-author-email", env.String(fmt.Sprintf("%s_GIT_COMMIT_AUTHOR_EMAIL", envVarPrefix), "contact@krateo.io"), "The email to use for git commits.") @@ -73,6 +74,7 @@ func main() { WithValues("git-commit-author-name", *gitCommitAuthorName). WithValues("git-commit-author-email", *gitCommitAuthorEmail). WithValues("timeout", timeout.String()). + WithValues("rate-limit-governor", *rateLimitGovernor). Info("Starting Git Provider") cfg, err := ctrl.GetConfig() @@ -103,6 +105,13 @@ func main() { MaxConcurrentReconciles: *maxReconcileRate, PollInterval: *pollInterval, GlobalRateLimiter: ratelimiter.NewGlobalExponential(*minErrorRetryInterval, *maxErrorRetryInterval), + RateLimit: controller.RateLimitOptions{ + Enabled: *rateLimitGovernor, + }, + } + if err := o.SetupRateLimitGovernor(mgr); err != nil { + log.Error(err, "Cannot set up the rate-limit governor") + os.Exit(1) } if err := apis.AddToScheme(mgr.GetScheme()); err != nil {