feat: filter entry lists by managed hints

.github/workflows/pr_build.yaml

@@ -85,7 +85,7 @@ jobs:
           make load-images
       - name: Test image
         run: |
-          docker tag ghcr.io/spiffe/spire-controller-manager:devel ghcr.io/spiffe/spire-controller-manager:nightly
+          docker tag "ghcr.io/${GITHUB_REPOSITORY}:devel" ghcr.io/spiffe/spire-controller-manager:devel
           (cd demo; ./test.sh)
 
   success:

api/v1alpha1/controllermanagerconfig_types.go

@@ -193,6 +193,13 @@ type ControllerManagerConfigurationSpec struct {
 	// used when enableEntryListCache is true. Defaults to 1h if unset or zero.
 	// +optional
 	EntryListCacheReloadInterval time.Duration `json:"entryListCacheReloadInterval,omitempty"`
+
+	// EnableEntryListHintFilter limits SPIRE ListEntries calls to hints used by
+	// managed ClusterSPIFFEID and ClusterStaticEntry resources. If any managed
+	// resource has an empty hint, the controller falls back to an unfiltered list.
+	// Defaults to false.
+	// +optional
+	EnableEntryListHintFilter bool `json:"enableEntryListHintFilter,omitempty"`
 }
 
 // ReconcileConfig configuration used to enable/disable syncing various types

cmd/main.go

@@ -259,7 +259,8 @@ func parseConfig() (Config, error) {
 		"entryIDPrefix", retval.ctrlConfig.EntryIDPrefix,
 		"entryIDPrefixCleanup", printCleanup,
 		"enableEntryListCache", retval.ctrlConfig.EnableEntryListCache,
-		"entryListCacheReloadInterval", retval.ctrlConfig.EntryListCacheReloadInterval)
+		"entryListCacheReloadInterval", retval.ctrlConfig.EntryListCacheReloadInterval,
+		"enableEntryListHintFilter", retval.ctrlConfig.EnableEntryListHintFilter)
 
 	switch {
 	case retval.ctrlConfig.TrustDomain == "":
@@ -271,6 +272,8 @@ func parseConfig() (Config, error) {
 		return retval, errors.New("validating webhook configuration name is required configuration")
 	case retval.ctrlConfig.EnableEntryListCache && retval.ctrlConfig.EntryIDPrefix == "":
 		return retval, errors.New("enableEntryListCache requires entryIDPrefix to be set")
+	case retval.ctrlConfig.EnableEntryListHintFilter && retval.ctrlConfig.EntryIDPrefix == "":
+		return retval, errors.New("enableEntryListHintFilter requires entryIDPrefix to be set")
 	case retval.ctrlConfig.ControllerManagerConfigurationSpec.Webhook.CertDir != "":
 		setupLog.Info("certDir configuration is ignored", "certDir", retval.ctrlConfig.ControllerManagerConfigurationSpec.Webhook.CertDir)
 	}
@@ -393,6 +396,7 @@ func run(mainConfig Config) (err error) {
 
 			EnableEntryListCache:         mainConfig.ctrlConfig.EnableEntryListCache,
 			EntryListCacheReloadInterval: mainConfig.ctrlConfig.EntryListCacheReloadInterval,
+			EnableEntryListHintFilter:    mainConfig.ctrlConfig.EnableEntryListHintFilter,
 		})
 	}
 
@@ -556,6 +560,7 @@ func staticRun(mainConfig Config) (err error) {
 
 			EnableEntryListCache:         mainConfig.ctrlConfig.EnableEntryListCache,
 			EntryListCacheReloadInterval: mainConfig.ctrlConfig.EntryListCacheReloadInterval,
+			EnableEntryListHintFilter:    mainConfig.ctrlConfig.EnableEntryListHintFilter,
 		})
 		go func() {
 			err := entryReconciler.Run(ctx)

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/onsi/ginkgo/v2 v2.32.0
 	github.com/onsi/gomega v1.42.1
 	github.com/prometheus/client_golang v1.23.2
+	github.com/samber/lo v1.53.0
 	github.com/spiffe/go-spiffe/v2 v2.8.1
 	github.com/spiffe/spire-api-sdk v1.15.1
 	github.com/stretchr/testify v1.11.1

go.sum

@@ -128,6 +128,8 @@ github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4
 github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
+github.com/samber/lo v1.53.0 h1:t975lj2py4kJPQ6haz1QMgtId2gtmfktACxIXArw3HM=
+github.com/samber/lo v1.53.0/go.mod h1:4+MXEGsJzbKGaUEQFKBq2xtfuznW9oz/WrgyzMzRoM0=
 github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
 github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spiffe/go-spiffe/v2 v2.8.1 h1:eXZMLsu+3MLEPJyGJkolqtVrteZfQdUpOWj6LTiDl/E=

pkg/spireapi/entryapi.go

@@ -20,10 +20,13 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/samber/lo"
+	"github.com/samber/lo/parallel"
 	entryv1 "github.com/spiffe/spire-api-sdk/proto/spire/api/server/entry/v1"
 	"github.com/spiffe/spire-api-sdk/proto/spire/api/types"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
@@ -41,7 +44,7 @@ const (
 type Field string
 
 type EntryClient interface {
-	ListEntries(ctx context.Context) ([]Entry, error)
+	ListEntries(ctx context.Context, hints ...string) ([]Entry, error)
 	CreateEntries(ctx context.Context, entries []Entry) ([]Status, error)
 	UpdateEntries(ctx context.Context, entries []Entry) ([]Status, error)
 	DeleteEntries(ctx context.Context, entryIDs []string) ([]Status, error)
@@ -56,11 +59,51 @@ type entryClient struct {
 	api entryv1.EntryClient
 }
 
-func (c entryClient) ListEntries(ctx context.Context) ([]Entry, error) {
+func (c entryClient) ListEntries(ctx context.Context, hints ...string) ([]Entry, error) {
+	filterHints := lo.Uniq(lo.Filter(hints, func(hint string, _ int) bool {
+		return hint != ""
+	}))
+	if len(filterHints) == 0 {
+		return c.listEntries(ctx, nil)
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	type listEntriesResult struct {
+		entries []Entry
+		err     error
+	}
+
+	results := parallel.Map(filterHints, func(hint string, _ int) listEntriesResult {
+		entries, err := c.listEntries(ctx, &entryv1.ListEntriesRequest_Filter{
+			ByHint: wrapperspb.String(hint),
+		})
+		if err != nil {
+			cancel()
+		}
+		return listEntriesResult{entries: entries, err: err}
+	})
+	if result, ok := lo.Find(results, func(result listEntriesResult) bool {
+		return result.err != nil
+	}); ok {
+		return nil, result.err
+	}
+
+	entries := lo.FlatMap(results, func(result listEntriesResult, _ int) []Entry {
+		return result.entries
+	})
+	return lo.UniqBy(entries, func(entry Entry) string {
+		return entry.ID
+	}), nil
+}
+
+func (c entryClient) listEntries(ctx context.Context, filter *entryv1.ListEntriesRequest_Filter) ([]Entry, error) {
 	var entries []*types.Entry
 	var pageToken string
 	for {
 		resp, err := c.api.ListEntries(ctx, &entryv1.ListEntriesRequest{
+			Filter:    filter,
 			PageToken: pageToken,
 			PageSize:  entryListPageSize,
 		})

pkg/spireapi/entryapi_test.go

@@ -92,6 +92,40 @@ func TestEntryAPIListEntries(t *testing.T) {
 	}
 }
 
+func TestEntryAPIListEntriesByHint(t *testing.T) {
+	server, client := startEntryAPIServer(t)
+	withHint := func(entry Entry, hint string) Entry {
+		entry.Hint = hint
+		return entry
+	}
+	server.setEntries(t,
+		withHint(entry1, "cluster-a"),
+		withHint(entry2, "cluster-b"),
+		withHint(entry3, "cluster-a"),
+	)
+
+	actualEntries, err := client.ListEntries(ctx, "cluster-a", "cluster-b", "cluster-a", "")
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []Entry{
+		withHint(entry1, "cluster-a"),
+		withHint(entry2, "cluster-b"),
+		withHint(entry3, "cluster-a"),
+	}, actualEntries)
+	assert.ElementsMatch(t, []string{"cluster-a", "cluster-b"}, server.getListEntryHints())
+}
+
+func TestEntryAPIListEntriesByHintDeduplicatesEntries(t *testing.T) {
+	server, client := startEntryAPIServer(t)
+	entry := entry1
+	entry.Hint = "cluster-a"
+	server.setEntries(t, entry)
+	server.duplicateListEntries = true
+
+	actualEntries, err := client.ListEntries(ctx, "cluster-a")
+	require.NoError(t, err)
+	assert.Equal(t, []Entry{entry}, actualEntries)
+}
+
 func TestCreateEntries(t *testing.T) {
 	server, client := startEntryAPIServer(t)
 
@@ -393,8 +427,10 @@ type entryServer struct {
 
 	mtx     sync.RWMutex
 	entries []*apitypes.Entry
+	hints   []string
 
 	clearUnsupportedFields bool
+	duplicateListEntries   bool
 
 	listEntriesErr        error
 	batchCreateEntriesErr error
@@ -405,11 +441,25 @@ type entryServer struct {
 func (s *entryServer) ListEntries(_ context.Context, req *entryv1.ListEntriesRequest) (*entryv1.ListEntriesResponse, error) {
 	resp := new(entryv1.ListEntriesResponse)
 
-	s.mtx.RLock()
-	defer s.mtx.RUnlock()
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
 
-	start, end, more := listBounds(req.PageToken, int(req.PageSize), len(s.entries), func(i int) string { return s.entries[i].Id })
-	for _, entry := range s.entries[start:end] {
+	entries := s.entries
+	if hint := req.GetFilter().GetByHint().GetValue(); hint != "" {
+		s.hints = append(s.hints, hint)
+		entries = nil
+		for _, entry := range s.entries {
+			if entry.Hint == hint {
+				entries = append(entries, entry)
+			}
+		}
+	}
+	if s.duplicateListEntries && len(entries) == 1 {
+		entries = append(entries, entries[0])
+	}
+
+	start, end, more := listBounds(req.PageToken, int(req.PageSize), len(entries), func(i int) string { return entries[i].Id })
+	for _, entry := range entries[start:end] {
 		resp.Entries = append(resp.Entries, entry)
 		if more {
 			resp.NextPageToken = entry.Id
@@ -496,6 +546,12 @@ func (s *entryServer) getEntries(t *testing.T) []Entry {
 	return entries
 }
 
+func (s *entryServer) getListEntryHints() []string {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	return append([]string(nil), s.hints...)
+}
+
 func (s *entryServer) setEntries(t *testing.T, entries ...Entry) {
 	s.clearEntries()
 	for _, entry := range entries {

pkg/spireentry/entry_list_cache.go

@@ -35,11 +35,12 @@ type entryListCache struct {
 	reloadAfter time.Duration
 	entries     map[string]spireapi.Entry
 	nextReload  time.Time
+	filterKey   string
 }
 
 // fresh reports whether the cache can be served without listing from the server.
-func (c *entryListCache) fresh() bool {
-	return c.entries != nil && time.Now().Before(c.nextReload)
+func (c *entryListCache) fresh(filterKey string) bool {
+	return c.entries != nil && c.filterKey == filterKey && time.Now().Before(c.nextReload)
 }
 
 // snapshot returns the cached entries as a slice.
@@ -52,11 +53,12 @@ func (c *entryListCache) snapshot() []spireapi.Entry {
 }
 
 // replace rebuilds the cache from a fresh server list and arms the next reload.
-func (c *entryListCache) replace(entries []spireapi.Entry) {
+func (c *entryListCache) replace(filterKey string, entries []spireapi.Entry) {
 	c.entries = make(map[string]spireapi.Entry, len(entries))
 	for _, entry := range entries {
 		c.entries[entry.ID] = entry
 	}
+	c.filterKey = filterKey
 	c.nextReload = time.Now().Add(c.reloadAfter)
 }