policy: gather policy calculation in the repository

squeed · aanm · commit 52d61cb11424 · 2024-11-15T13:22:42.000Z
This simplifies the policy calculation flow by centralizing calculations
entirely within the policy package. Endpoints no longer need to lock the
repository manually; policy calculation manages this itself.

Some care is needed to preserve the logic to skip policy revisisions;
endpoints are informed when they can skip policy calculations and thus
must pass this back down to the repository when calculating policy.

Signed-off-by: Casey Callendrello &lt;cdc@isovalent.com&gt;
diff --git a/pkg/endpoint/metrics.go b/pkg/endpoint/metrics.go
@@ -90,6 +90,16 @@ func (s *regenerationStatistics) GetMap() map[string]*spanstat.SpanStat {
 	return result
 }
 
+// used by PolicyRepository.GetSelectorPolicy
+func (s *regenerationStatistics) WaitingForPolicyRepository() *spanstat.SpanStat {
+	return &s.waitingForPolicyRepository
+}
+
+// used by PolicyRepository.GetSelectorPolicy
+func (s *regenerationStatistics) PolicyCalculation() *spanstat.SpanStat {
+	return &s.policyCalculation
+}
+
 // endpointPolicyStatusMap is a map to store the endpoint id and the policy
 // enforcement status. It is used only to send metrics to prometheus.
 type endpointPolicyStatusMap struct {
diff --git a/pkg/endpoint/policy.go b/pkg/endpoint/policy.go
@@ -190,48 +190,28 @@ func (e *Endpoint) regeneratePolicy(stats *regenerationStatistics, datapathRegen
 	e.unlock()
 
 	e.getLogger().Debug("Starting policy recalculation...")
-
-	stats.waitingForPolicyRepository.Start()
-	repo := e.policyGetter.GetPolicyRepository()
-	repo.RLock() // Be sure to release this lock!
-	stats.waitingForPolicyRepository.End(true)
-
-	result.policyRevision = repo.GetRevision()
-
-	// Recompute policy for this endpoint only if not already done for this revision
-	// and identity.
-	if e.nextPolicyRevision >= result.policyRevision && e.desiredPolicy != nil {
-
-		if !forcePolicyCompute {
-			if logger := e.getLogger(); logging.CanLogAt(logger.Logger, logrus.DebugLevel) {
-				e.getLogger().WithFields(logrus.Fields{
-					"policyRevision.next": e.nextPolicyRevision,
-					"policyRevision.repo": result.policyRevision,
-					"policyChanged":       e.nextPolicyRevision > e.policyRevision,
-				}).Debug("Skipping unnecessary endpoint policy recalculation")
-			}
-			repo.RUnlock()
-			return result, nil
-		} else {
-			e.getLogger().Debug("Forced policy recalculation")
-		}
+	skipPolicyRevision := e.nextPolicyRevision
+	if forcePolicyCompute || e.desiredPolicy == nil {
+		e.getLogger().Debug("Forced policy recalculation")
+		skipPolicyRevision = 0
 	}
 
-	stats.policyCalculation.Start()
-	defer func() { stats.policyCalculation.End(err == nil) }()
-
-	// UpdatePolicy ensures the SelectorPolicy is fully resolved.
-	// Endpoint lock must not be held!
-	// TODO: GH-7515: Consider ways to compute policy outside of the
-	// endpoint regeneration process, ideally as part of the policy change
-	// handler.
-	selectorPolicy, err := repo.GetPolicyCache().UpdatePolicy(securityIdentity)
+	var selectorPolicy policy.SelectorPolicy
+	selectorPolicy, result.policyRevision, err = e.policyGetter.GetPolicyRepository().GetSelectorPolicy(securityIdentity, skipPolicyRevision, stats)
 	if err != nil {
-		e.getLogger().WithError(err).Warning("Failed to update policy")
-		repo.RUnlock()
+		e.getLogger().WithError(err).Warning("Failed to calculate SelectorPolicy")
 		return nil, err
 	}
-	repo.RUnlock() // Done with policy repository; release this now as Consume() can be slow
+
+	// selectorPolicy is nil if skipRevision was matched.
+	if selectorPolicy == nil {
+		e.getLogger().WithFields(logrus.Fields{
+			"policyRevision.next": e.nextPolicyRevision,
+			"policyRevision.repo": result.policyRevision,
+			"policyChanged":       e.nextPolicyRevision > e.policyRevision,
+		}).Debug("Skipping unnecessary endpoint policy recalculation")
+		return result, err
+	}
 
 	// Add new redirects before Consume() so that all required proxy ports are available for it.
 	var desiredRedirects map[string]uint16
diff --git a/pkg/policy/distillery.go b/pkg/policy/distillery.go
@@ -112,16 +112,6 @@ func (cache *PolicyCache) LocalEndpointIdentityRemoved(identity *identityPkg.Ide
 	cache.delete(identity)
 }
 
-// UpdatePolicy resolves the policy for the security identity of the specified
-// endpoint and caches it for future use.
-//
-// The caller must provide threadsafety for iteration over the policy
-// repository.
-func (cache *PolicyCache) UpdatePolicy(identity *identityPkg.Identity) (SelectorPolicy, error) {
-	sp, _, err := cache.updateSelectorPolicy(identity)
-	return sp, err
-}
-
 // GetAuthTypes returns the AuthTypes required by the policy between the localID and remoteID, if
 // any, otherwise returns nil.
 func (cache *PolicyCache) GetAuthTypes(localID, remoteID identityPkg.NumericIdentity) AuthTypes {
diff --git a/pkg/policy/repository.go b/pkg/policy/repository.go
@@ -28,6 +28,7 @@ import (
 	"github.com/cilium/cilium/pkg/metrics"
 	"github.com/cilium/cilium/pkg/option"
 	"github.com/cilium/cilium/pkg/policy/api"
+	"github.com/cilium/cilium/pkg/spanstat"
 )
 
 // PolicyContext is an interface policy resolution functions use to access the Repository.
@@ -125,6 +126,15 @@ type PolicyRepository interface {
 	GetAuthTypes(localID identity.NumericIdentity, remoteID identity.NumericIdentity) AuthTypes
 	GetEnvoyHTTPRules(l7Rules *api.L7Rules, ns string) (*cilium.HttpNetworkPolicyRules, bool)
 	GetPolicyCache() *PolicyCache
+
+	// GetSelectorPolicy computes the SelectorPolicy for a given identity.
+	//
+	// It returns nil if skipRevision is >= than the already calculated version.
+	// This is used to skip policy calculation when a certain revision delta is
+	// known to not affect the given identity. Pass a skipRevision of 0 to force
+	// calculation.
+	GetSelectorPolicy(id *identity.Identity, skipRevision uint64, stats GetPolicyStatistics) (SelectorPolicy, uint64, error)
+
 	GetRevision() uint64
 	GetRulesList() *models.Policy
 	GetSelectorCache() *SelectorCache
@@ -138,6 +148,11 @@ type PolicyRepository interface {
 	Start()
 }
 
+type GetPolicyStatistics interface {
+	WaitingForPolicyRepository() *spanstat.SpanStat
+	PolicyCalculation() *spanstat.SpanStat
+}
+
 // Repository is a list of policy rules which in combination form the security
 // policy. A policy repository can be
 type Repository struct {
@@ -930,3 +945,37 @@ func wildcardRule(lbls labels.LabelArray, ingress bool) *rule {
 
 	return r
 }
+
+// GetSelectorPolicy computes the SelectorPolicy for a given identity.
+//
+// It returns nil if skipRevision is >= than the already calculated version.
+// This is used to skip policy calculation when a certain revision delta is
+// known to not affect the given identity. Pass a skipRevision of 0 to force
+// calculation.
+func (r *Repository) GetSelectorPolicy(id *identity.Identity, skipRevision uint64, stats GetPolicyStatistics) (SelectorPolicy, uint64, error) {
+	stats.WaitingForPolicyRepository().Start()
+	r.RLock()
+	defer r.RUnlock()
+	stats.WaitingForPolicyRepository().End(true)
+
+	rev := r.GetRevision()
+
+	// Do we already have a given revision?
+	// If so, skip calculation.
+	if skipRevision >= rev {
+		return nil, rev, nil
+	}
+
+	stats.PolicyCalculation().Start()
+	// This may call back in to the (locked) repository to generate the
+	// selector policy
+	sp, updated, err := r.policyCache.updateSelectorPolicy(id)
+	stats.PolicyCalculation().EndError(err)
+
+	// If we hit cache, reset the statistics.
+	if !updated {
+		stats.PolicyCalculation().Reset()
+	}
+
+	return sp, rev, nil
+}
