From 61b9bc248fb005cea4cee85bcc00670b8cbe8343 Mon Sep 17 00:00:00 2001 From: andres-portainer <91705312+andres-portainer@users.noreply.github.com> Date: Thu, 26 Mar 2026 07:47:54 -0300 Subject: [PATCH] fix(schedule): abstract simple loops with RunOnInterval() BE-12765 (#2163) --- api/chisel/service.go | 21 ++-- api/git/git_integration_test.go | 13 --- api/git/service.go | 85 +++++++--------- .../customtemplate_git_fetch_test.go | 2 +- api/http/handler/edgestacks/edgestack_test.go | 2 +- .../endpointedge_status_inspect_test.go | 2 +- api/http/handler/system/version_test.go | 2 +- api/http/handler/teams/team_list_test.go | 2 +- .../users/user_create_access_token_test.go | 2 +- api/http/handler/users/user_delete_test.go | 2 +- .../users/user_get_access_tokens_test.go | 2 +- api/http/handler/users/user_list_test.go | 2 +- .../users/user_remove_access_token_test.go | 2 +- api/http/handler/users/user_update_test.go | 2 +- api/http/security/bouncer.go | 14 +-- api/http/security/bouncer_test.go | 10 +- api/http/server.go | 2 +- pkg/schedule/ticker.go | 32 +++++++ pkg/schedule/ticker_test.go | 96 +++++++++++++++++++ 19 files changed, 187 insertions(+), 108 deletions(-) create mode 100644 pkg/schedule/ticker.go create mode 100644 pkg/schedule/ticker_test.go diff --git a/api/chisel/service.go b/api/chisel/service.go index dcd992997..7833f9ab0 100644 --- a/api/chisel/service.go +++ b/api/chisel/service.go @@ -11,6 +11,7 @@ import ( portainer "github.com/portainer/portainer/api" "github.com/portainer/portainer/api/dataservices" "github.com/portainer/portainer/api/http/proxy" + "github.com/portainer/portainer/pkg/schedule" chserver "github.com/jpillora/chisel/server" "github.com/jpillora/chisel/share/ccrypto" @@ -233,23 +234,13 @@ func (service *Service) startTunnelVerificationLoop() { Float64("check_interval_seconds", tunnelCleanupInterval.Seconds()). Msg("starting tunnel management process") - ticker := time.NewTicker(tunnelCleanupInterval) + schedule.RunOnInterval(service.shutdownCtx, tunnelCleanupInterval, service.checkTunnels, func() { + log.Debug().Msg("shutting down tunnel service") - for { - select { - case <-ticker.C: - service.checkTunnels() - case <-service.shutdownCtx.Done(): - log.Debug().Msg("shutting down tunnel service") - - if err := service.StopTunnelServer(); err != nil { - log.Debug().Err(err).Msg("stopped tunnel service") - } - - ticker.Stop() - return + if err := service.StopTunnelServer(); err != nil { + log.Debug().Err(err).Msg("stopped tunnel service") } - } + }) } // checkTunnels finds the first tunnel that has not had any activity recently diff --git a/api/git/git_integration_test.go b/api/git/git_integration_test.go index f9d6a94a0..09680e5fb 100644 --- a/api/git/git_integration_test.go +++ b/api/git/git_integration_test.go @@ -340,19 +340,6 @@ func TestService_purgeCacheByTTL_Github(t *testing.T) { assert.Equal(t, 0, service.repoFileCache.Len()) } -func TestService_canStopCacheCleanTimer_whenContextDone(t *testing.T) { - timeout := 10 * time.Millisecond - deadlineCtx, cancel := context.WithDeadline(context.TODO(), time.Now().Add(10*timeout)) - defer cancel() - - service := NewService(deadlineCtx) - assert.False(t, service.timerHasStopped(), "timer should not be stopped") - - <-time.After(20 * timeout) - - assert.True(t, service.timerHasStopped(), "timer should be stopped") -} - func TestService_HardRefresh_ListRefs_GitHub(t *testing.T) { ensureIntegrationTest(t) diff --git a/api/git/service.go b/api/git/service.go index f44a8c4fa..44da2375b 100644 --- a/api/git/service.go +++ b/api/git/service.go @@ -4,9 +4,10 @@ import ( "context" "strconv" "strings" - "sync" "time" + "github.com/portainer/portainer/pkg/schedule" + "github.com/go-git/go-git/v5" "github.com/go-git/go-git/v5/plumbing" githttp "github.com/go-git/go-git/v5/plumbing/transport/http" @@ -29,11 +30,9 @@ type RepoManager interface { // Service represents a service for managing Git. type Service struct { - shutdownCtx context.Context - azure RepoManager - git RepoManager - timerStopped bool - mut sync.Mutex + shutdownCtx context.Context + azure RepoManager + git RepoManager cacheEnabled bool // Cache the result of repository refs, key is repository URL @@ -52,25 +51,26 @@ func newService(ctx context.Context, cacheSize int, cacheTTL time.Duration) *Ser shutdownCtx: ctx, azure: NewAzureClient(), git: NewGitClient(false), - timerStopped: false, cacheEnabled: cacheSize > 0, } - if service.cacheEnabled { - var err error - service.repoRefCache, err = lru.New(cacheSize) - if err != nil { - log.Debug().Err(err).Msg("failed to create ref cache") - } + if !service.cacheEnabled { + return service + } - service.repoFileCache, err = lru.New(cacheSize) - if err != nil { - log.Debug().Err(err).Msg("failed to create file cache") - } + var err error + service.repoRefCache, err = lru.New(cacheSize) + if err != nil { + log.Debug().Err(err).Msg("failed to create ref cache") + } - if cacheTTL > 0 { - go service.startCacheCleanTimer(cacheTTL) - } + service.repoFileCache, err = lru.New(cacheSize) + if err != nil { + log.Debug().Err(err).Msg("failed to create file cache") + } + + if cacheTTL > 0 { + go service.startCacheCleanTimer(cacheTTL) } return service @@ -78,29 +78,7 @@ func newService(ctx context.Context, cacheSize int, cacheTTL time.Duration) *Ser // startCacheCleanTimer starts a timer to purge caches periodically func (service *Service) startCacheCleanTimer(d time.Duration) { - ticker := time.NewTicker(d) - - for { - select { - case <-ticker.C: - service.purgeCache() - - case <-service.shutdownCtx.Done(): - ticker.Stop() - service.mut.Lock() - service.timerStopped = true - service.mut.Unlock() - return - } - } -} - -// timerHasStopped shows the CacheClean timer state with thread-safe way -func (service *Service) timerHasStopped() bool { - service.mut.Lock() - defer service.mut.Unlock() - ret := service.timerStopped - return ret + schedule.RunOnInterval(service.shutdownCtx, d, service.purgeCache, nil) } // CloneRepository clones a git repository using the specified URL in the specified @@ -338,15 +316,16 @@ func filterFiles(paths []string, includedExts []string) []string { } func GetBasicAuth(username, password string) *githttp.BasicAuth { - if password != "" { - if username == "" { - username = "token" - } - - return &githttp.BasicAuth{ - Username: username, - Password: password, - } + if password == "" { + return nil + } + + if username == "" { + username = "token" + } + + return &githttp.BasicAuth{ + Username: username, + Password: password, } - return nil } diff --git a/api/http/handler/customtemplates/customtemplate_git_fetch_test.go b/api/http/handler/customtemplates/customtemplate_git_fetch_test.go index 0b7769862..94de7d63b 100644 --- a/api/http/handler/customtemplates/customtemplate_git_fetch_test.go +++ b/api/http/handler/customtemplates/customtemplate_git_fetch_test.go @@ -183,7 +183,7 @@ func Test_customTemplateGitFetch(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") - requestBouncer := security.NewRequestBouncer(store, jwtService, nil) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, nil) gitService := &TestGitService{ targetFilePath: filepath.Join(template1.ProjectPath, template1.GitConfig.ConfigFilePath), diff --git a/api/http/handler/edgestacks/edgestack_test.go b/api/http/handler/edgestacks/edgestack_test.go index 324704216..caae7c9d2 100644 --- a/api/http/handler/edgestacks/edgestack_test.go +++ b/api/http/handler/edgestacks/edgestack_test.go @@ -48,7 +48,7 @@ func setupHandler(t *testing.T) (*Handler, string) { } handler := NewHandler( - security.NewRequestBouncer(store, jwtService, apiKeyService), + security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService), store, edgestacks.NewService(store), ) diff --git a/api/http/handler/endpointedge/endpointedge_status_inspect_test.go b/api/http/handler/endpointedge/endpointedge_status_inspect_test.go index 526fc58de..18d0d3c76 100644 --- a/api/http/handler/endpointedge/endpointedge_status_inspect_test.go +++ b/api/http/handler/endpointedge/endpointedge_status_inspect_test.go @@ -107,7 +107,7 @@ func mustSetupHandler(t *testing.T) *Handler { } handler := NewHandler( - security.NewRequestBouncer(store, jwtService, apiKeyService), + security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService), store, fs, chisel.NewService(store, shutdownCtx, nil), diff --git a/api/http/handler/system/version_test.go b/api/http/handler/system/version_test.go index 71d26ffa3..d3027d12f 100644 --- a/api/http/handler/system/version_test.go +++ b/api/http/handler/system/version_test.go @@ -39,7 +39,7 @@ func Test_getSystemVersion(t *testing.T) { require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) h := NewHandler(requestBouncer, &portainer.Status{}, store, nil, nil) diff --git a/api/http/handler/teams/team_list_test.go b/api/http/handler/teams/team_list_test.go index 5d7d412cf..b77187b8d 100644 --- a/api/http/handler/teams/team_list_test.go +++ b/api/http/handler/teams/team_list_test.go @@ -35,7 +35,7 @@ func Test_teamList(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) h := NewHandler(requestBouncer) h.DataStore = store diff --git a/api/http/handler/users/user_create_access_token_test.go b/api/http/handler/users/user_create_access_token_test.go index 996f00cd6..a6aa0e2c3 100644 --- a/api/http/handler/users/user_create_access_token_test.go +++ b/api/http/handler/users/user_create_access_token_test.go @@ -38,7 +38,7 @@ func Test_userCreateAccessToken(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) rateLimiter := security.NewRateLimiter(10, 1*time.Second, 1*time.Hour) passwordChecker := security.NewPasswordStrengthChecker(store.SettingsService) diff --git a/api/http/handler/users/user_delete_test.go b/api/http/handler/users/user_delete_test.go index be2626411..731365c17 100644 --- a/api/http/handler/users/user_delete_test.go +++ b/api/http/handler/users/user_delete_test.go @@ -30,7 +30,7 @@ func Test_deleteUserRemovesAccessTokens(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) rateLimiter := security.NewRateLimiter(10, 1*time.Second, 1*time.Hour) passwordChecker := security.NewPasswordStrengthChecker(store.SettingsService) diff --git a/api/http/handler/users/user_get_access_tokens_test.go b/api/http/handler/users/user_get_access_tokens_test.go index 8d4b6fee9..82a215650 100644 --- a/api/http/handler/users/user_get_access_tokens_test.go +++ b/api/http/handler/users/user_get_access_tokens_test.go @@ -37,7 +37,7 @@ func Test_userGetAccessTokens(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) rateLimiter := security.NewRateLimiter(10, 1*time.Second, 1*time.Hour) passwordChecker := security.NewPasswordStrengthChecker(store.SettingsService) diff --git a/api/http/handler/users/user_list_test.go b/api/http/handler/users/user_list_test.go index 64b99f442..6c017e88b 100644 --- a/api/http/handler/users/user_list_test.go +++ b/api/http/handler/users/user_list_test.go @@ -36,7 +36,7 @@ func Test_userList(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) rateLimiter := security.NewRateLimiter(10, 1*time.Second, 1*time.Hour) passwordChecker := security.NewPasswordStrengthChecker(store.SettingsService) diff --git a/api/http/handler/users/user_remove_access_token_test.go b/api/http/handler/users/user_remove_access_token_test.go index f0b2441d8..94e8d9b46 100644 --- a/api/http/handler/users/user_remove_access_token_test.go +++ b/api/http/handler/users/user_remove_access_token_test.go @@ -36,7 +36,7 @@ func Test_userRemoveAccessToken(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) rateLimiter := security.NewRateLimiter(10, 1*time.Second, 1*time.Hour) passwordChecker := security.NewPasswordStrengthChecker(store.SettingsService) diff --git a/api/http/handler/users/user_update_test.go b/api/http/handler/users/user_update_test.go index ae0a1b4cf..4d976df9e 100644 --- a/api/http/handler/users/user_update_test.go +++ b/api/http/handler/users/user_update_test.go @@ -30,7 +30,7 @@ func Test_updateUserRemovesAccessTokens(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - requestBouncer := security.NewRequestBouncer(store, jwtService, apiKeyService) + requestBouncer := security.NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) rateLimiter := security.NewRateLimiter(10, 1*time.Second, 1*time.Hour) passwordChecker := security.NewPasswordStrengthChecker(store.SettingsService) diff --git a/api/http/security/bouncer.go b/api/http/security/bouncer.go index fdd92c37c..35bb13a93 100644 --- a/api/http/security/bouncer.go +++ b/api/http/security/bouncer.go @@ -1,6 +1,7 @@ package security import ( + "context" "net/http" "slices" "strings" @@ -13,6 +14,7 @@ import ( httperrors "github.com/portainer/portainer/api/http/errors" "github.com/portainer/portainer/pkg/featureflags" httperror "github.com/portainer/portainer/pkg/libhttp/error" + "github.com/portainer/portainer/pkg/schedule" "github.com/pkg/errors" "github.com/rs/zerolog/log" @@ -68,7 +70,7 @@ var ( ) // NewRequestBouncer initializes a new RequestBouncer -func NewRequestBouncer(dataStore dataservices.DataStore, jwtService portainer.JWTService, apiKeyService apikey.APIKeyService) *RequestBouncer { +func NewRequestBouncer(ctx context.Context, dataStore dataservices.DataStore, jwtService portainer.JWTService, apiKeyService apikey.APIKeyService) *RequestBouncer { b := &RequestBouncer{ dataStore: dataStore, jwtService: jwtService, @@ -77,7 +79,7 @@ func NewRequestBouncer(dataStore dataservices.DataStore, jwtService portainer.JW csp: true, } - go b.cleanUpExpiredJWT() + go schedule.RunOnInterval(ctx, time.Hour, b.cleanUpExpiredJWTPass, nil) return b } @@ -400,14 +402,6 @@ func (bouncer *RequestBouncer) cleanUpExpiredJWTPass() { }) } -func (bouncer *RequestBouncer) cleanUpExpiredJWT() { - ticker := time.NewTicker(time.Hour) - - for range ticker.C { - bouncer.cleanUpExpiredJWTPass() - } -} - // apiKeyLookup looks up an verifies an api-key by: // - computing the digest of the raw api-key // - verifying it exists in cache/database diff --git a/api/http/security/bouncer_test.go b/api/http/security/bouncer_test.go index 0a0f3bd7f..bd0a3948d 100644 --- a/api/http/security/bouncer_test.go +++ b/api/http/security/bouncer_test.go @@ -48,7 +48,7 @@ func Test_mwAuthenticateFirst(t *testing.T) { apiKeyService := apikey.NewAPIKeyService(nil, nil) - bouncer := NewRequestBouncer(store, jwtService, apiKeyService) + bouncer := NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) tests := []struct { name string @@ -319,7 +319,7 @@ func Test_apiKeyLookup(t *testing.T) { jwtService, err := jwt.NewService("1h", store) require.NoError(t, err, "Error initiating jwt service") apiKeyService := apikey.NewAPIKeyService(store.APIKeyRepository(), store.User()) - bouncer := NewRequestBouncer(store, jwtService, apiKeyService) + bouncer := NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) t.Run("missing x-api-key header fails api-key lookup", func(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "/", nil) @@ -496,7 +496,7 @@ func TestJWTRevocation(t *testing.T) { apiKeyService := apikey.NewAPIKeyService(nil, nil) - bouncer := NewRequestBouncer(store, jwtService, apiKeyService) + bouncer := NewRequestBouncer(t.Context(), store, jwtService, apiKeyService) r, err := http.NewRequest(http.MethodGet, "url", nil) require.NoError(t, err) @@ -539,7 +539,7 @@ func TestJWTRevocation(t *testing.T) { } func TestCSPHeaderDefault(t *testing.T) { - b := NewRequestBouncer(nil, nil, nil) + b := NewRequestBouncer(t.Context(), nil, nil, nil) srv := httptest.NewServer( b.PublicAccess(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})), @@ -557,7 +557,7 @@ func TestCSPHeaderDefault(t *testing.T) { } func TestCSPHeaderDisabled(t *testing.T) { - b := NewRequestBouncer(nil, nil, nil) + b := NewRequestBouncer(t.Context(), nil, nil, nil) b.DisableCSP() srv := httptest.NewServer( diff --git a/api/http/server.go b/api/http/server.go index 54089a254..38711a409 100644 --- a/api/http/server.go +++ b/api/http/server.go @@ -121,7 +121,7 @@ type Server struct { func (server *Server) Start() error { kubernetesTokenCacheManager := server.KubernetesTokenCacheManager - requestBouncer := security.NewRequestBouncer(server.DataStore, server.JWTService, server.APIKeyService) + requestBouncer := security.NewRequestBouncer(server.ShutdownCtx, server.DataStore, server.JWTService, server.APIKeyService) if !server.CSP { requestBouncer.DisableCSP() } diff --git a/pkg/schedule/ticker.go b/pkg/schedule/ticker.go new file mode 100644 index 000000000..e5b2ffaeb --- /dev/null +++ b/pkg/schedule/ticker.go @@ -0,0 +1,32 @@ +package schedule + +import ( + "context" + "time" +) + +// RunOnInterval calls fn on every tick of a ticker with the given interval, +// stopping when ctx is done. If cleanup is non-nil it is called once after +// the context is cancelled and before the function returns. The ticker is +// always stopped before returning. +func RunOnInterval(ctx context.Context, interval time.Duration, fn func(), cleanup func()) { + if cleanup != nil { + defer cleanup() + } + + if fn == nil { + return + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + fn() + case <-ctx.Done(): + return + } + } +} diff --git a/pkg/schedule/ticker_test.go b/pkg/schedule/ticker_test.go new file mode 100644 index 000000000..e9f16798a --- /dev/null +++ b/pkg/schedule/ticker_test.go @@ -0,0 +1,96 @@ +package schedule + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// run starts RunOnInterval in a goroutine and returns a channel that closes +// when RunOnInterval returns. +func run(ctx context.Context, interval time.Duration, fn func(), cleanup func()) <-chan struct{} { + done := make(chan struct{}) + + go func() { + RunOnInterval(ctx, interval, fn, cleanup) + close(done) + }() + + return done +} + +func TestRunOnInterval_CallsFnOnTick(t *testing.T) { + var calls atomic.Int32 + + run(t.Context(), time.Millisecond, func() { calls.Add(1) }, nil) + + assert.Eventually(t, func() bool { + return calls.Load() >= 3 + }, time.Second, time.Millisecond) +} + +func TestRunOnInterval_StopsWhenContextCancelled(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + done := run(ctx, time.Hour, func() {}, nil) + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("RunOnInterval did not return after context cancellation") + } +} + +func TestRunOnInterval_CallsCleanupOnCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + + var cleanupCalled atomic.Bool + done := run(ctx, time.Hour, func() {}, func() { cleanupCalled.Store(true) }) + + cancel() + <-done + + assert.True(t, cleanupCalled.Load()) +} + +func TestRunOnInterval_CleanupCalledExactlyOnce(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + + var cleanupCount atomic.Int32 + done := run(ctx, time.Millisecond, func() {}, func() { cleanupCount.Add(1) }) + + // Let fn tick a few times before cancelling. + assert.Eventually(t, func() bool { return cleanupCount.Load() == 0 }, time.Second, time.Millisecond) + cancel() + <-done + + assert.Equal(t, int32(1), cleanupCount.Load()) +} + +func TestRunOnInterval_NilCleanupDoesNotPanic(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + done := run(ctx, time.Hour, func() {}, nil) + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("RunOnInterval did not return after context cancellation") + } +} + +func TestRunOnInterval_FnNotCalledAfterCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + var calls atomic.Int32 + done := run(ctx, time.Millisecond, func() { calls.Add(1) }, nil) + <-done + + assert.Equal(t, int32(0), calls.Load()) +}