homelab/portainer
1
0
Fork 0
mirror of https://github.com/portainer/portainer.git synced 2026-08-07 09:54:49 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor(gitops): remove read cache from sources and workflows handlers (#3283)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Chaim Lev-Ari
2026-07-27 08:17:59 +03:00
committed by GitHub
parent 0852cf7f8f
commit 4c872de6fc
13 changed files with 6 additions and 406 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/http/handler/gitops/sources/cache_test.go
-172
View File
@@ -1,172 +0,0 @@
package sources
import (
"net/http"
"net/http/httptest"
"testing"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
"github.com/portainer/portainer/api/datastore"
gittypes "github.com/portainer/portainer/api/git/types"
"github.com/segmentio/encoding/json"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// Each test below primes the source cache with a read before mutating, then
// reads again and asserts the change is reflected. The handler is built with a
// non-expiring cache (newTestHandlerNoCacheExpiry), so a fresh result can only
// come from invalidation on the write — not from the TTL elapsing.
func TestGitSourceCreate_InvalidatesCache(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
user := &portainer.User{ID: 1, Role: portainer.AdministratorRole}
require.NoError(t, store.UpdateTx(func(tx dataservices.DataStoreTx) error {
return tx.User().Create(user)
}))
h := newTestHandlerNoCacheExpiry(t, store)
// prime the cache with the (empty) list
rr := httptest.NewRecorder()
h.ServeHTTP(rr, buildListReq(t, user.ID, ""))
require.Empty(t, decodeSources(t, rr))
body, err := json.Marshal(GitSourceCreatePayload{
URL: "https://github.com/org/repo.git",
Name: "my-source",
})
require.NoError(t, err)
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildCreateReq(t, user.ID, body))
require.Equal(t, http.StatusCreated, rr.Code)
// the new source must appear despite the primed empty cache
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildListReq(t, user.ID, ""))
got := decodeSources(t, rr)
require.Len(t, got, 1)
require.Equal(t, "my-source", got[0].Name)
}
func TestSourceDelete_InvalidatesCache(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
var srcID portainer.SourceID
user := &portainer.User{ID: 1, Role: portainer.AdministratorRole}
require.NoError(t, store.UpdateTx(func(tx dataservices.DataStoreTx) error {
src := &portainer.Source{Name: "to-delete", Type: portainer.SourceTypeGit, Git: &gittypes.GitSource{URL: "https://github.com/org/repo"}}
require.NoError(t, tx.Source().Create(adminUserContext, src))
srcID = src.ID
return tx.User().Create(user)
}))
h := newTestHandlerNoCacheExpiry(t, store)
// prime the cache with the source present
rr := httptest.NewRecorder()
h.ServeHTTP(rr, buildListReq(t, user.ID, ""))
require.Len(t, decodeSources(t, rr), 1)
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildDeleteReq(t, user.ID, int(srcID)))
require.Equal(t, http.StatusNoContent, rr.Code)
// the deleted source must be gone despite the primed cache
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildListReq(t, user.ID, ""))
require.Empty(t, decodeSources(t, rr))
}
func TestGitSourceUpdate_InvalidatesCache(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
var srcID portainer.SourceID
user := &portainer.User{ID: 1, Role: portainer.AdministratorRole}
require.NoError(t, store.UpdateTx(func(tx dataservices.DataStoreTx) error {
src := &portainer.Source{Name: "old-name", Type: portainer.SourceTypeGit, Git: &gittypes.GitSource{URL: "https://github.com/org/repo"}}
require.NoError(t, tx.Source().Create(adminUserContext, src))
srcID = src.ID
return tx.User().Create(user)
}))
h := newTestHandlerNoCacheExpiry(t, store)
// prime the cache with the original name
rr := httptest.NewRecorder()
h.ServeHTTP(rr, buildListReq(t, user.ID, ""))
primed := decodeSources(t, rr)
require.Len(t, primed, 1)
require.Equal(t, "old-name", primed[0].Name)
body, err := json.Marshal(GitSourceUpdatePayload{Name: new("new-name")})
require.NoError(t, err)
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildUpdateReq(t, user.ID, int(srcID), body))
require.Equal(t, http.StatusOK, rr.Code)
// the updated name must be reflected despite the primed cache
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildListReq(t, user.ID, ""))
got := decodeSources(t, rr)
require.Len(t, got, 1)
require.Equal(t, "new-name", got[0].Name)
}
// TestSourceMutation_InvalidatesSummaryCache verifies the summary endpoint, which
// shares the cache entry with the list via getSources, also sees fresh data.
func TestSourceMutation_InvalidatesSummaryCache(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
user := &portainer.User{ID: 1, Role: portainer.AdministratorRole}
require.NoError(t, store.UpdateTx(func(tx dataservices.DataStoreTx) error {
return tx.User().Create(user)
}))
h := newTestHandlerNoCacheExpiry(t, store)
// prime the cache with the empty summary
rr := httptest.NewRecorder()
h.ServeHTTP(rr, buildSummaryReq(t, user.ID))
require.Equal(t, 0, summaryTotal(t, rr))
body, err := json.Marshal(GitSourceCreatePayload{
URL: "https://github.com/org/repo.git",
Name: "my-source",
})
require.NoError(t, err)
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildCreateReq(t, user.ID, body))
require.Equal(t, http.StatusCreated, rr.Code)
// the summary count must reflect the new source despite the primed cache
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildSummaryReq(t, user.ID))
assert.Equal(t, 1, summaryTotal(t, rr))
}
func summaryTotal(t *testing.T, rr *httptest.ResponseRecorder) int {
t.Helper()
require.Equal(t, http.StatusOK, rr.Code, "unexpected status: %s", rr.Body.String())
var summary map[string]int
require.NoError(t, json.NewDecoder(rr.Body).Decode(&summary))
total := 0
for _, count := range summary {
total += count
}
return total
}
api/http/handler/gitops/sources/create_git.go
-2
View File
@@ -97,8 +97,6 @@ func (h *Handler) gitSourceCreate(w http.ResponseWriter, r *http.Request) *httpe
return httperror.InternalServerError("Unable to persist source status", err)
}
h.invalidateCache()
if err := h.sourceScheduler.Reconcile(src.ID); err != nil {
log.Warn().Err(err).Int("source_id", int(src.ID)).Msg("source scheduler reconcile failed after source creation")
}
api/http/handler/gitops/sources/delete.go
-2
View File
@@ -93,8 +93,6 @@ func (h *Handler) sourceDelete(w http.ResponseWriter, r *http.Request) *httperro
return httperror.InternalServerError("Unable to delete source", err)
}
h.invalidateCache()
if err := h.sourceScheduler.Reconcile(portainer.SourceID(sourceID)); err != nil {
log.Warn().Err(err).Int("source_id", sourceID).Msg("source scheduler reconcile failed after source deletion")
}
api/http/handler/gitops/sources/handler.go
-15
View File
@@ -2,9 +2,7 @@ package sources
import (
"net/http"
"time"
gocache "github.com/patrickmn/go-cache"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
"github.com/portainer/portainer/api/gitops/scheduling"
@@ -15,17 +13,11 @@ import (
"github.com/gorilla/mux"
)
const (
cacheTTL = 30 * time.Second
cacheCleanupInterval = 10 * time.Minute
)
// Handler is the HTTP handler for the GitOps sources API.
type Handler struct {
*mux.Router
dataStore dataservices.DataStore
gitService portainer.GitService
cache *gocache.Cache
k8sFactory *cli.ClientFactory
sourceScheduler *scheduling.SourceScheduler
}
@@ -35,7 +27,6 @@ func NewHandler(bouncer security.BouncerService, dataStore dataservices.DataStor
Router: mux.NewRouter(),
dataStore: dataStore,
gitService: gitService,
cache: gocache.New(cacheTTL, cacheCleanupInterval),
k8sFactory: k8sFactory,
sourceScheduler: sourceScheduler,
}
@@ -58,9 +49,3 @@ func NewHandler(bouncer security.BouncerService, dataStore dataservices.DataStor
return h
}
// invalidateCache clears the cached source lists so the next read reflects
// the latest datastore state. Called after any mutating operation.
func (h *Handler) invalidateCache() {
h.cache.Flush()
}
api/http/handler/gitops/sources/helpers_test.go
-11
View File
@@ -14,7 +14,6 @@ import (
"github.com/portainer/portainer/api/http/security"
"github.com/portainer/portainer/api/internal/testhelpers"
gocache "github.com/patrickmn/go-cache"
"github.com/segmentio/encoding/json"
"github.com/stretchr/testify/require"
)
@@ -51,16 +50,6 @@ func newTestHandler(t *testing.T, store dataservices.DataStore) *Handler {
return NewHandler(testhelpers.NewTestRequestBouncer(), store, nil, nil, nil)
}
// newTestHandlerNoCacheExpiry returns a handler whose source cache never expires,
// so cache-invalidation tests can prove a write clears the cache rather than the
// TTL simply elapsing between requests.
func newTestHandlerNoCacheExpiry(t *testing.T, store dataservices.DataStore) *Handler {
t.Helper()
h := newTestHandler(t, store)
h.cache = gocache.New(gocache.NoExpiration, cacheCleanupInterval)
return h
}
func adminRestrictedContext(userID portainer.UserID) *security.RestrictedRequestContext {
return &security.RestrictedRequestContext{
UserID: userID,
api/http/handler/gitops/sources/list.go
+1 -30
View File
@@ -3,8 +3,6 @@ package sources
import (
"context"
"net/http"
"slices"
"strconv"
"strings"
portainer "github.com/portainer/portainer/api"
@@ -17,8 +15,6 @@ import (
httperror "github.com/portainer/portainer/pkg/libhttp/error"
"github.com/portainer/portainer/pkg/libhttp/request"
"github.com/portainer/portainer/pkg/libhttp/response"
gocache "github.com/patrickmn/go-cache"
)
// Source is the list item response for a GitOps source
@@ -56,9 +52,7 @@ func (h *Handler) list(w http.ResponseWriter, r *http.Request) *httperror.Handle
return httperror.InternalServerError("Unable to retrieve info from request context", err)
}
key := cacheKey(securityContext)
sources, err := h.getSources(r.Context(), key, securityContext)
sources, err := h.fetchSources(r.Context(), securityContext)
if err != nil {
return httperror.InternalServerError("Unable to retrieve sources", err)
}
@@ -94,29 +88,6 @@ func (h *Handler) list(w http.ResponseWriter, r *http.Request) *httperror.Handle
return response.JSON(w, results.Items)
}
func (h *Handler) getSources(ctx context.Context, key string, sc *security.RestrictedRequestContext) ([]Source, error) {
if cached, ok := h.cache.Get(key); ok {
return slices.Clone(cached.([]Source)), nil
}
result, err := h.fetchSources(ctx, sc)
if err != nil {
return nil, err
}
h.cache.Set(key, result, gocache.DefaultExpiration)
return slices.Clone(result), nil
}
func cacheKey(sc *security.RestrictedRequestContext) string {
teamIDs := make([]string, len(sc.UserMemberships))
for i, membership := range sc.UserMemberships {
teamIDs[i] = strconv.Itoa(int(membership.TeamID))
}
slices.Sort(teamIDs)
return strconv.Itoa(int(sc.UserID)) + ":" + strconv.FormatBool(sc.IsAdmin) + ":" + strings.Join(teamIDs, ",")
}
func (h *Handler) fetchSources(ctx context.Context, sc *security.RestrictedRequestContext) ([]Source, error) {
var allSrcs []portainer.Source
var stats map[portainer.SourceID]workflows.SourceStats
api/http/handler/gitops/sources/summary.go
+1 -3
View File
@@ -27,9 +27,7 @@ func (h *Handler) summary(w http.ResponseWriter, r *http.Request) *httperror.Han
return httperror.InternalServerError("Unable to retrieve info from request context", err)
}
key := cacheKey(securityContext)
sources, err := h.getSources(r.Context(), key, securityContext)
sources, err := h.fetchSources(r.Context(), securityContext)
if err != nil {
return httperror.InternalServerError("Unable to retrieve sources", err)
}
api/http/handler/gitops/sources/update_access.go
-2
View File
@@ -73,8 +73,6 @@ func (h *Handler) gitSourceUpdateAccess(w http.ResponseWriter, r *http.Request)
return httperror.InternalServerError("Unable to update source access", err)
}
h.invalidateCache()
return response.JSON(w, src)
}
api/http/handler/gitops/sources/update_git.go
-2
View File
@@ -120,8 +120,6 @@ func (h *Handler) gitSourceUpdate(w http.ResponseWriter, r *http.Request) *httpe
return httperror.InternalServerError("Unable to persist source status", err)
}
h.invalidateCache()
if err := h.sourceScheduler.Reconcile(src.ID); err != nil {
log.Warn().Err(err).Int("source_id", int(src.ID)).Msg("source scheduler reconcile failed after source update")
}
api/http/handler/gitops/workflows/handler.go
-9
View File
@@ -2,9 +2,7 @@ package workflows
import (
"net/http"
"time"
gocache "github.com/patrickmn/go-cache"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
"github.com/portainer/portainer/api/kubernetes/cli"
@@ -13,16 +11,10 @@ import (
"github.com/gorilla/mux"
)
const (
cacheTTL = 30 * time.Second
cacheCleanupInterval = 10 * time.Minute
)
type Handler struct {
*mux.Router
dataStore dataservices.DataStore
gitService portainer.GitService
cache *gocache.Cache
k8sFactory *cli.ClientFactory
}
@@ -31,7 +23,6 @@ func NewHandler(dataStore dataservices.DataStore, gitService portainer.GitServic
Router: mux.NewRouter(),
dataStore: dataStore,
gitService: gitService,
cache: gocache.New(cacheTTL, cacheCleanupInterval),
k8sFactory: k8sFactory,
}
api/http/handler/gitops/workflows/list.go
+3 -29
View File
@@ -3,11 +3,8 @@ package workflows
import (
"cmp"
"net/http"
"slices"
"strconv"
"strings"
gocache "github.com/patrickmn/go-cache"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
svc "github.com/portainer/portainer/api/gitops/workflows"
@@ -54,9 +51,7 @@ func (h *Handler) list(w http.ResponseWriter, r *http.Request) *httperror.Handle
return httperror.InternalServerError("Unable to retrieve info from request context", err)
}
key := cacheKey(securityContext, endpointIDs)
items, err := h.getWorkflows(key, securityContext, endpointIDs)
items, err := h.getWorkflows(securityContext, endpointIDs)
if err != nil {
return httperror.InternalServerError("Unable to retrieve workflows", err)
}
@@ -111,11 +106,7 @@ func hasArtifactMatching(w svc.Workflow, pred func(svc.ArtifactDetail) bool) boo
return slicesx.Some(w.Artifacts, pred)
}
func (h *Handler) getWorkflows(key string, sc *security.RestrictedRequestContext, endpointIDs []portainer.EndpointID) ([]svc.Workflow, error) {
if cached, ok := h.cache.Get(key); ok {
return slices.Clone(cached.([]svc.Workflow)), nil
}
func (h *Handler) getWorkflows(sc *security.RestrictedRequestContext, endpointIDs []portainer.EndpointID) ([]svc.Workflow, error) {
var result []svc.Workflow
err := h.dataStore.ViewTx(func(tx dataservices.DataStoreTx) error {
var err error
@@ -125,23 +116,6 @@ func (h *Handler) getWorkflows(key string, sc *security.RestrictedRequestContext
if err != nil {
return nil, err
}
h.cache.Set(key, result, gocache.DefaultExpiration)
return slices.Clone(result), nil
}
func cacheKey(sc *security.RestrictedRequestContext, endpointIDs []portainer.EndpointID) string {
ids := make([]string, len(endpointIDs))
for i, id := range endpointIDs {
ids[i] = strconv.Itoa(int(id))
}
slices.Sort(ids)
teamIDs := make([]string, len(sc.UserMemberships))
for i, membership := range sc.UserMemberships {
teamIDs[i] = strconv.Itoa(int(membership.TeamID))
}
slices.Sort(teamIDs)
return strconv.Itoa(int(sc.UserID)) + ":" + strconv.FormatBool(sc.IsAdmin) + ":" + strings.Join(ids, ",") + ":" + strings.Join(teamIDs, ",")
return result, nil
}
api/http/handler/gitops/workflows/list_test.go
-128
View File
@@ -5,8 +5,6 @@ import (
"net/http"
"net/http/httptest"
"testing"
"testing/synctest"
"time"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
@@ -184,132 +182,6 @@ func TestWorkflowsList_Sort(t *testing.T) {
assert.Equal(t, "alpha", items[2].Name)
}
// Uses testing/synctest to control time.Now() without real sleeps.
// The Handler is created outside the bubble so its go-cache cleanup goroutine
// does not join the bubble. Inside the bubble all time.Now() calls return
// fake time, so cache.Set stores a fake expiry and cache.Get compares
// against the same fake clock — consistent without touching real time.
func TestWorkflowsList_Cache(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
require.NoError(t, store.UpdateTx(func(tx dataservices.DataStoreTx) error {
createGitStack(t, tx, &portainer.Stack{
ID: 1, Name: "initial-stack",
GitConfig: gitConfig("https://github.com/x/initial"),
})
require.NoError(t, tx.User().Create(&portainer.User{ID: 1, Role: portainer.AdministratorRole}))
return nil
}))
// Create the handler outside the bubble so the go-cache cleanup goroutine
// is not part of the bubble and does not block synctest.Test from returning.
h := NewHandler(store, nil, nil)
synctest.Test(t, func(t *testing.T) {
// First request at fake T=0: populates cache.
rr := httptest.NewRecorder()
h.ServeHTTP(rr, buildWorkflowsReq(t, 1, portainer.AdministratorRole, ""))
require.Len(t, decodeWorkflows(t, rr), 1)
// Mutate the store while cache is still warm.
createGitStack(t, store, &portainer.Stack{
ID: 2, Name: "new-stack",
GitConfig: gitConfig("https://github.com/x/new"),
})
// Second request — same cache key, should return stale cached result.
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildWorkflowsReq(t, 1, portainer.AdministratorRole, ""))
assert.Len(t, decodeWorkflows(t, rr), 1, "cache hit: new stack should not appear yet")
// Advance fake clock past the cache TTL. synctest unblocks immediately
// since no other goroutines are in the bubble.
time.Sleep(cacheTTL + time.Second)
// Third request — cache expired, should now fetch fresh data.
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildWorkflowsReq(t, 1, portainer.AdministratorRole, ""))
assert.Len(t, decodeWorkflows(t, rr), 2, "after TTL expiry: both stacks should appear")
})
}
func TestWorkflowsList_CacheImmutableAfterSort(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
for i, name := range []string{"alpha", "beta", "gamma"} {
require.NoError(t, store.UpdateTx(func(tx dataservices.DataStoreTx) error {
createGitStack(t, tx, &portainer.Stack{
ID: portainer.StackID(i + 1),
Name: name,
GitConfig: gitConfig("https://github.com/x/" + name),
})
require.NoError(t, tx.User().Create(&portainer.User{ID: 1, Role: portainer.AdministratorRole}))
return nil
}))
}
h := NewHandler(store, nil, nil)
// First request: no sort — cache miss, populates cache as [alpha, beta, gamma].
rr := httptest.NewRecorder()
h.ServeHTTP(rr, buildWorkflowsReq(t, 1, portainer.AdministratorRole, ""))
items := decodeWorkflows(t, rr)
require.Len(t, items, 3)
require.Equal(t, "alpha", items[0].Name)
// Second request: sort desc — cache hit, sorts the shared slice in-place to [gamma, beta, alpha].
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildWorkflowsReq(t, 1, portainer.AdministratorRole, "sort=name&order=desc"))
items = decodeWorkflows(t, rr)
require.Len(t, items, 3)
require.Equal(t, "gamma", items[0].Name)
// Third request: no sort — should still return insertion order [alpha, beta, gamma],
// but without a defensive clone the mutated cache returns [gamma, beta, alpha].
rr = httptest.NewRecorder()
h.ServeHTTP(rr, buildWorkflowsReq(t, 1, portainer.AdministratorRole, ""))
items = decodeWorkflows(t, rr)
require.Len(t, items, 3)
assert.Equal(t, "alpha", items[0].Name, "sort must not mutate the cached slice")
}
func TestWorkflowsList_CacheSeparateKeys(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
require.NoError(t, store.UpdateTx(func(tx dataservices.DataStoreTx) error {
createGitStack(t, tx, &portainer.Stack{
ID: 1, Name: "env1-stack", EndpointID: 1,
GitConfig: gitConfig("https://github.com/x/1"),
})
createGitStack(t, tx, &portainer.Stack{
ID: 2, Name: "env2-stack", EndpointID: 2,
GitConfig: gitConfig("https://github.com/x/2"),
})
require.NoError(t, tx.User().Create(&portainer.User{ID: 1, Role: portainer.AdministratorRole}))
return nil
}))
h := NewHandler(store, nil, nil)
rr1 := httptest.NewRecorder()
h.ServeHTTP(rr1, buildWorkflowsReq(t, 1, portainer.AdministratorRole, "endpointIds[]=1"))
items1 := decodeWorkflows(t, rr1)
require.Len(t, items1, 1)
assert.Equal(t, "env1-stack", items1[0].Name)
rr2 := httptest.NewRecorder()
h.ServeHTTP(rr2, buildWorkflowsReq(t, 1, portainer.AdministratorRole, "endpointIds[]=2"))
items2 := decodeWorkflows(t, rr2)
require.Len(t, items2, 1)
assert.Equal(t, "env2-stack", items2[0].Name)
}
func TestWorkflowsList_StatusFilter(t *testing.T) {
t.Parallel()
_, store := datastore.MustNewTestStore(t, false, true)
api/http/handler/gitops/workflows/summary.go
+1 -1
View File
@@ -26,7 +26,7 @@ func (h *Handler) summary(w http.ResponseWriter, r *http.Request) *httperror.Han
return httperror.InternalServerError("Unable to retrieve info from request context", err)
}
items, err := h.getWorkflows(cacheKey(securityContext, nil), securityContext, nil)
items, err := h.getWorkflows(securityContext, nil)
if err != nil {
return httperror.InternalServerError("Unable to retrieve workflows", err)
}