Implement LRU cache for storing SVIDs in SPIRE Agent

cmd/spire-agent/cli/api/common.go

@@ -13,6 +13,8 @@ import (
 	"google.golang.org/grpc/metadata"
 )
 
+const commandTimeout = 5 * time.Second
+
 type workloadClient struct {
 	workload.SpiffeWorkloadAPIClient
 	timeout time.Duration
@@ -71,7 +73,7 @@ func adaptCommand(env *cli.Env, clientsMaker workloadClientMaker, cmd command) *
 		clientsMaker: clientsMaker,
 		cmd:          cmd,
 		env:          env,
-		timeout:      cli.DurationFlag(time.Second),
+		timeout:      cli.DurationFlag(commandTimeout),
 	}
 
 	fs := flag.NewFlagSet(cmd.name(), flag.ContinueOnError)

cmd/spire-agent/cli/run/run.go

@@ -106,7 +106,8 @@ type experimentalConfig struct {
 
 	Flags fflag.RawConfig `hcl:"feature_flags"`
 
-	UnusedKeys []string `hcl:",unusedKeys"`
+	UnusedKeys           []string `hcl:",unusedKeys"`
+	X509SVIDCacheMaxSize int      `hcl:"x509_svid_cache_max_size"`
 }
 
 type Command struct {
@@ -394,6 +395,11 @@ func NewAgentConfig(c *Config, logOptions []log.Option, allowUnknownConfig bool)
 		}
 	}
 
+	if c.Agent.Experimental.X509SVIDCacheMaxSize < 0 {
+		return nil, errors.New("x509_svid_cache_max_size should not be negative")
+	}
+	ac.X509SVIDCacheMaxSize = c.Agent.Experimental.X509SVIDCacheMaxSize
+
 	serverHostPort := net.JoinHostPort(c.Agent.ServerAddress, strconv.Itoa(c.Agent.ServerPort))
 	ac.ServerAddress = fmt.Sprintf("dns:///%s", serverHostPort)
 

cmd/spire-agent/cli/run/run_test.go

@@ -727,6 +727,42 @@ func TestNewAgentConfig(t *testing.T) {
 				require.Nil(t, c)
 			},
 		},
+		{
+			msg: "x509_svid_cache_max_size is set",
+			input: func(c *Config) {
+				c.Agent.Experimental.X509SVIDCacheMaxSize = 100
+			},
+			test: func(t *testing.T, c *agent.Config) {
+				require.EqualValues(t, 100, c.X509SVIDCacheMaxSize)
+			},
+		},
+		{
+			msg: "x509_svid_cache_max_size is not set",
+			input: func(c *Config) {
+			},
+			test: func(t *testing.T, c *agent.Config) {
+				require.EqualValues(t, 0, c.X509SVIDCacheMaxSize)
+			},
+		},
+		{
+			msg: "x509_svid_cache_max_size is zero",
+			input: func(c *Config) {
+				c.Agent.Experimental.X509SVIDCacheMaxSize = 0
+			},
+			test: func(t *testing.T, c *agent.Config) {
+				require.EqualValues(t, 0, c.X509SVIDCacheMaxSize)
+			},
+		},
+		{
+			msg:         "x509_svid_cache_max_size is negative",
+			expectError: true,
+			input: func(c *Config) {
+				c.Agent.Experimental.X509SVIDCacheMaxSize = -10
+			},
+			test: func(t *testing.T, c *agent.Config) {
+				require.Nil(t, c)
+			},
+		},
 		{
 			msg: "allowed_foreign_jwt_claims provided",
 			input: func(c *Config) {

pkg/agent/agent.go

@@ -203,18 +203,19 @@ func (a *Agent) attest(ctx context.Context, sto storage.Storage, cat catalog.Cat
 
 func (a *Agent) newManager(ctx context.Context, sto storage.Storage, cat catalog.Catalog, metrics telemetry.Metrics, as *node_attestor.AttestationResult, cache *storecache.Cache) (manager.Manager, error) {
 	config := &manager.Config{
-		SVID:            as.SVID,
-		SVIDKey:         as.Key,
-		Bundle:          as.Bundle,
-		Catalog:         cat,
-		TrustDomain:     a.c.TrustDomain,
-		ServerAddr:      a.c.ServerAddress,
-		Log:             a.c.Log.WithField(telemetry.SubsystemName, telemetry.Manager),
-		Metrics:         metrics,
-		WorkloadKeyType: a.c.WorkloadKeyType,
-		Storage:         sto,
-		SyncInterval:    a.c.SyncInterval,
-		SVIDStoreCache:  cache,
+		SVID:             as.SVID,
+		SVIDKey:          as.Key,
+		Bundle:           as.Bundle,
+		Catalog:          cat,
+		TrustDomain:      a.c.TrustDomain,
+		ServerAddr:       a.c.ServerAddress,
+		Log:              a.c.Log.WithField(telemetry.SubsystemName, telemetry.Manager),
+		Metrics:          metrics,
+		WorkloadKeyType:  a.c.WorkloadKeyType,
+		Storage:          sto,
+		SyncInterval:     a.c.SyncInterval,
+		SVIDCacheMaxSize: a.c.X509SVIDCacheMaxSize,
+		SVIDStoreCache:   cache,
 	}
 
 	mgr := manager.New(config)

pkg/agent/api/delegatedidentity/v1/service.go

@@ -82,24 +82,24 @@ func (s *Service) isCallerAuthorized(ctx context.Context, log logrus.FieldLogger
 		}
 	}
 
-	identities := s.manager.MatchingIdentities(callerSelectors)
-	numRegisteredIDs := len(identities)
+	entries := s.manager.MatchingRegistrationEntries(callerSelectors)
+	numRegisteredEntries := len(entries)
 
-	if numRegisteredIDs == 0 {
+	if numRegisteredEntries == 0 {
 		log.Error("no identity issued")
 		return nil, status.Error(codes.PermissionDenied, "no identity issued")
 	}
 
-	for _, identity := range identities {
-		if _, ok := s.authorizedDelegates[identity.Entry.SpiffeId]; ok {
+	for _, entry := range entries {
+		if _, ok := s.authorizedDelegates[entry.SpiffeId]; ok {
 			return callerSelectors, nil
 		}
 	}
 
 	// caller has identity associeted with but none is authorized
 	log.WithFields(logrus.Fields{
-		"num_registered_ids": numRegisteredIDs,
-		"default_id":         identities[0].Entry.SpiffeId,
+		"num_registered_entries": numRegisteredEntries,
+		"default_id":             entries[0].SpiffeId,
 	}).Error("Permission denied; caller not configured as an authorized delegate.")
 
 	return nil, status.Error(codes.PermissionDenied, "caller not configured as an authorized delegate")
@@ -120,7 +120,11 @@ func (s *Service) SubscribeToX509SVIDs(req *delegatedidentityv1.SubscribeToX509S
 		return status.Error(codes.InvalidArgument, "could not parse provided selectors")
 	}
 
-	subscriber := s.manager.SubscribeToCacheChanges(selectors)
+	subscriber, err := s.manager.SubscribeToCacheChanges(ctx, selectors)
+	if err != nil {
+		log.WithError(err).Error("Subscribe to cache changes failed")
+		return err
+	}
 	defer subscriber.Finish()
 
 	for {
@@ -268,11 +272,11 @@ func (s *Service) FetchJWTSVIDs(ctx context.Context, req *delegatedidentityv1.Fe
 	}
 	var spiffeIDs []spiffeid.ID
 
-	identities := s.manager.MatchingIdentities(selectors)
-	for _, identity := range identities {
-		spiffeID, err := spiffeid.FromString(identity.Entry.SpiffeId)
+	entries := s.manager.MatchingRegistrationEntries(selectors)
+	for _, entry := range entries {
+		spiffeID, err := spiffeid.FromString(entry.SpiffeId)
 		if err != nil {
-			log.WithField(telemetry.SPIFFEID, identity.Entry.SpiffeId).WithError(err).Error("Invalid requested SPIFFE ID")
+			log.WithField(telemetry.SPIFFEID, entry.SpiffeId).WithError(err).Error("Invalid requested SPIFFE ID")
 			return nil, status.Errorf(codes.InvalidArgument, "invalid requested SPIFFE ID: %v", err)
 		}
 

pkg/agent/api/delegatedidentity/v1/service_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/andres-erbsen/clock"
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/spiffe/go-spiffe/v2/bundle/spiffebundle"
@@ -653,10 +654,6 @@ func (fa FakeAttestor) Attest(ctx context.Context) ([]*common.Selector, error) {
 	return fa.selectors, fa.err
 }
 
-func (m *FakeManager) MatchingIdentities(selectors []*common.Selector) []cache.Identity {
-	return m.identities
-}
-
 type FakeManager struct {
 	manager.Manager
 
@@ -677,9 +674,9 @@ func (m *FakeManager) subscriberDone() {
 	atomic.AddInt32(&m.subscribers, -1)
 }
 
-func (m *FakeManager) SubscribeToCacheChanges(selectors cache.Selectors) cache.Subscriber {
+func (m *FakeManager) SubscribeToCacheChanges(ctx context.Context, selectors cache.Selectors) (cache.Subscriber, error) {
 	atomic.AddInt32(&m.subscribers, 1)
-	return newFakeSubscriber(m, m.updates)
+	return newFakeSubscriber(m, m.updates), nil
 }
 
 func (m *FakeManager) FetchJWTSVID(ctx context.Context, spiffeID spiffeid.ID, audience []string) (*client.JWTSVID, error) {
@@ -692,6 +689,14 @@ func (m *FakeManager) FetchJWTSVID(ctx context.Context, spiffeID spiffeid.ID, au
 	}, nil
 }
 
+func (m *FakeManager) MatchingRegistrationEntries(selectors []*common.Selector) []*common.RegistrationEntry {
+	out := make([]*common.RegistrationEntry, 0, len(m.identities))
+	for _, identity := range m.identities {
+		out = append(out, identity.Entry)
+	}
+	return out
+}
+
 type fakeSubscriber struct {
 	m      *FakeManager
 	ch     chan *cache.WorkloadUpdate
@@ -794,5 +799,5 @@ func (m *FakeManager) SubscribeToBundleChanges() *cache.BundleStream {
 
 func newTestCache() *cache.Cache {
 	log, _ := test.NewNullLogger()
-	return cache.New(log, trustDomain1, bundle1, telemetry.Blackhole{})
+	return cache.New(log, trustDomain1, bundle1, telemetry.Blackhole{}, 0, clock.NewMock())
 }

pkg/agent/config.go

@@ -59,6 +59,9 @@ type Config struct {
 	// SyncInterval controls how often the agent sync synchronizer waits
 	SyncInterval time.Duration
 
+	// X509SVIDCacheMaxSize is a soft limit of max number of SVIDs that would be stored in cache
+	X509SVIDCacheMaxSize int
+
 	// Trust domain and associated CA bundle
 	TrustDomain spiffeid.TrustDomain
 	TrustBundle []*x509.Certificate

pkg/agent/endpoints/sdsv2/handler.go

@@ -31,7 +31,7 @@ type Attestor interface {
 }
 
 type Manager interface {
-	SubscribeToCacheChanges(key cache.Selectors) cache.Subscriber
+	SubscribeToCacheChanges(ctx context.Context, key cache.Selectors) (cache.Subscriber, error)
 	FetchWorkloadUpdate(selectors []*common.Selector) *cache.WorkloadUpdate
 }
 
@@ -64,7 +64,11 @@ func (h *Handler) StreamSecrets(stream discovery_v2.SecretDiscoveryService_Strea
 		return err
 	}
 
-	sub := h.c.Manager.SubscribeToCacheChanges(selectors)
+	sub, err := h.c.Manager.SubscribeToCacheChanges(stream.Context(), selectors)
+	if err != nil {
+		log.WithError(err).Error("Subscribe to cache changes failed")
+		return err
+	}
 	defer sub.Finish()
 
 	updch := sub.Updates()

pkg/agent/endpoints/sdsv2/handler_test.go

@@ -552,7 +552,7 @@ func NewFakeManager(t *testing.T) *FakeManager {
 	}
 }
 
-func (m *FakeManager) SubscribeToCacheChanges(selectors cache.Selectors) cache.Subscriber {
+func (m *FakeManager) SubscribeToCacheChanges(ctx context.Context, selectors cache.Selectors) (cache.Subscriber, error) {
 	require.Equal(m.t, workloadSelectors, selectors)
 
 	updch := make(chan *cache.WorkloadUpdate, 1)
@@ -568,7 +568,7 @@ func (m *FakeManager) SubscribeToCacheChanges(selectors cache.Selectors) cache.S
 	return NewFakeSubscriber(updch, func() {
 		delete(m.subs, key)
 		close(updch)
-	})
+	}), nil
 }
 
 func (m *FakeManager) FetchWorkloadUpdate(selectors []*common.Selector) *cache.WorkloadUpdate {

pkg/agent/endpoints/sdsv3/handler.go

@@ -39,7 +39,7 @@ type Attestor interface {
 }
 
 type Manager interface {
-	SubscribeToCacheChanges(key cache.Selectors) cache.Subscriber
+	SubscribeToCacheChanges(ctx context.Context, key cache.Selectors) (cache.Subscriber, error)
 	FetchWorkloadUpdate(selectors []*common.Selector) *cache.WorkloadUpdate
 }
 
@@ -74,7 +74,11 @@ func (h *Handler) StreamSecrets(stream secret_v3.SecretDiscoveryService_StreamSe
 		return err
 	}
 
-	sub := h.c.Manager.SubscribeToCacheChanges(selectors)
+	sub, err := h.c.Manager.SubscribeToCacheChanges(stream.Context(), selectors)
+	if err != nil {
+		log.WithError(err).Error("Subscribe to cache changes failed")
+		return err
+	}
 	defer sub.Finish()
 
 	updch := sub.Updates()

pkg/agent/endpoints/sdsv3/handler_test.go

@@ -1288,7 +1288,7 @@ func NewFakeManager(t *testing.T) *FakeManager {
 	}
 }
 
-func (m *FakeManager) SubscribeToCacheChanges(selectors cache.Selectors) cache.Subscriber {
+func (m *FakeManager) SubscribeToCacheChanges(ctx context.Context, selectors cache.Selectors) (cache.Subscriber, error) {
 	require.Equal(m.t, workloadSelectors, selectors)
 
 	updch := make(chan *cache.WorkloadUpdate, 1)
@@ -1304,7 +1304,7 @@ func (m *FakeManager) SubscribeToCacheChanges(selectors cache.Selectors) cache.S
 	return NewFakeSubscriber(updch, func() {
 		delete(m.subs, key)
 		close(updch)
-	})
+	}), nil
 }
 
 func (m *FakeManager) FetchWorkloadUpdate(selectors []*common.Selector) *cache.WorkloadUpdate {

pkg/agent/endpoints/workload/handler.go

@@ -30,8 +30,8 @@ import (
 )
 
 type Manager interface {
-	SubscribeToCacheChanges(cache.Selectors) cache.Subscriber
-	MatchingIdentities([]*common.Selector) []cache.Identity
+	SubscribeToCacheChanges(ctx context.Context, key cache.Selectors) (cache.Subscriber, error)
+	MatchingRegistrationEntries(selectors []*common.Selector) []*common.RegistrationEntry
 	FetchJWTSVID(ctx context.Context, spiffeID spiffeid.ID, audience []string) (*client.JWTSVID, error)
 	FetchWorkloadUpdate([]*common.Selector) *cache.WorkloadUpdate
 }
@@ -85,15 +85,15 @@ func (h *Handler) FetchJWTSVID(ctx context.Context, req *workload.JWTSVIDRequest
 
 	log = log.WithField(telemetry.Registered, true)
 
-	identities := h.c.Manager.MatchingIdentities(selectors)
-	for _, identity := range identities {
-		if req.SpiffeId != "" && identity.Entry.SpiffeId != req.SpiffeId {
+	entries := h.c.Manager.MatchingRegistrationEntries(selectors)
+	for _, entry := range entries {
+		if req.SpiffeId != "" && entry.SpiffeId != req.SpiffeId {
 			continue
 		}
 
-		spiffeID, err := spiffeid.FromString(identity.Entry.SpiffeId)
+		spiffeID, err := spiffeid.FromString(entry.SpiffeId)
 		if err != nil {
-			log.WithField(telemetry.SPIFFEID, identity.Entry.SpiffeId).WithError(err).Error("Invalid requested SPIFFE ID")
+			log.WithField(telemetry.SPIFFEID, entry.SpiffeId).WithError(err).Error("Invalid requested SPIFFE ID")
 			return nil, status.Errorf(codes.InvalidArgument, "invalid requested SPIFFE ID: %v", err)
 		}
 
@@ -138,7 +138,11 @@ func (h *Handler) FetchJWTBundles(req *workload.JWTBundlesRequest, stream worklo
 		return err
 	}
 
-	subscriber := h.c.Manager.SubscribeToCacheChanges(selectors)
+	subscriber, err := h.c.Manager.SubscribeToCacheChanges(ctx, selectors)
+	if err != nil {
+		log.WithError(err).Error("Subscribe to cache changes failed")
+		return err
+	}
 	defer subscriber.Finish()
 
 	var previousResp *workload.JWTBundlesResponse
@@ -224,7 +228,11 @@ func (h *Handler) FetchX509SVID(_ *workload.X509SVIDRequest, stream workload.Spi
 		return err
 	}
 
-	subscriber := h.c.Manager.SubscribeToCacheChanges(selectors)
+	subscriber, err := h.c.Manager.SubscribeToCacheChanges(ctx, selectors)
+	if err != nil {
+		log.WithError(err).Error("Subscribe to cache changes failed")
+		return err
+	}
 	defer subscriber.Finish()
 
 	for {
@@ -250,7 +258,11 @@ func (h *Handler) FetchX509Bundles(_ *workload.X509BundlesRequest, stream worklo
 		return err
 	}
 
-	subscriber := h.c.Manager.SubscribeToCacheChanges(selectors)
+	subscriber, err := h.c.Manager.SubscribeToCacheChanges(ctx, selectors)
+	if err != nil {
+		log.WithError(err).Error("Subscribe to cache changes failed")
+		return err
+	}
 	defer subscriber.Finish()
 
 	var previousResp *workload.X509BundlesResponse