Compile dikastes/collector policies once instead of interpreting per flow (rebase of #13267)

app-policy/checker/bench_egress_test.go

@@ -97,16 +97,30 @@ var egressPortsPerRule = []struct {
 func BenchmarkEvaluateEgressAllowList(b *testing.B) {
 	// A flow on a port few rules share: rejected on the port comparison nearly everywhere.
 	b.Run("TailPort", func(b *testing.B) {
-		benchEvaluateEgressAllowList(b, egressTailPortFlow)
+		benchEvaluateEgressAllowList(b, egressTailPortFlow, false)
 	})
 	// A flow on the most popular port: ~18% of rules share it and go on to the address check.
 	b.Run("PopularPort", func(b *testing.B) {
-		benchEvaluateEgressAllowList(b, egressPopularPortFlow)
+		benchEvaluateEgressAllowList(b, egressPopularPortFlow, false)
 	})
 	// No rule matches, so the walk covers the whole tier and ends in the tier default deny.
 	// Ordering cannot reduce the scan depth here, only the cost of each rejected rule.
 	b.Run("Denied", func(b *testing.B) {
-		benchEvaluateEgressAllowList(b, egressDeniedFlow)
+		benchEvaluateEgressAllowList(b, egressDeniedFlow, false)
+	})
+}
+
+// BenchmarkEvaluateEgressAllowListCompiled is BenchmarkEvaluateEgressAllowList with the store's
+// policies compiled, as when a PolicyCompiler is configured.
+func BenchmarkEvaluateEgressAllowListCompiled(b *testing.B) {
+	b.Run("TailPort", func(b *testing.B) {
+		benchEvaluateEgressAllowList(b, egressTailPortFlow, true)
+	})
+	b.Run("PopularPort", func(b *testing.B) {
+		benchEvaluateEgressAllowList(b, egressPopularPortFlow, true)
+	})
+	b.Run("Denied", func(b *testing.B) {
+		benchEvaluateEgressAllowList(b, egressDeniedFlow, true)
 	})
 }
 
@@ -154,16 +168,19 @@ func egressFlow(destIP string, destPort int32) *MockFlow {
 	}
 }
 
-func benchEvaluateEgressAllowList(b *testing.B, caseFor egressCaseFunc) {
+func benchEvaluateEgressAllowList(b *testing.B, caseFor egressCaseFunc, compiled bool) {
 	_, restoreLogging := withBenchLogging(log.WarnLevel)
 	defer restoreLogging()
 
 	store, ep, target := buildEgressAllowListStore()
+	if compiled {
+		compileStoreForTest(store)
+	}
 	c := caseFor(target)
 
 	// Pre-flight outside the timed loop: prove the walk is the one the case intends, so that
 	// a fixture change cannot silently turn a full walk into an early exit.
-	trace, err := Evaluate(EnforcedOnly, rules.RuleDirEgress, store, ep, c.flow)
+	trace, err := Evaluate(EnforcedOnly, rules.RuleDirEgress, store, ep, c.flow, nil)
 	if err != nil {
 		b.Fatalf("evaluation failed: %v", err)
 	}
@@ -178,7 +195,7 @@ func benchEvaluateEgressAllowList(b *testing.B, caseFor egressCaseFunc) {
 	b.ReportAllocs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		benchTraceSink, _ = Evaluate(EnforcedOnly, rules.RuleDirEgress, store, ep, c.flow)
+		benchTraceSink, _ = Evaluate(EnforcedOnly, rules.RuleDirEgress, store, ep, c.flow, benchTraceSink[:0])
 	}
 	b.StopTimer()
 	b.ReportMetric(float64(c.rulesWalked), "rules/op")
@@ -251,7 +268,10 @@ func buildEgressAllowListStore() (*policystore.PolicyStore, *proto.WorkloadEndpo
 		tier.EgressPolicies = append(tier.EgressPolicies, policyID)
 	}
 
+	// The endpoint goes into the store, as dikastes' per-pod store holds it: evaluation resolves
+	// an endpoint's compiled form by identity.
 	ep := &proto.WorkloadEndpoint{Tiers: []*proto.TierInfo{tier}}
+	store.Endpoint = ep
 	return store, ep, target
 }
 

app-policy/checker/bench_test.go

@@ -78,6 +78,16 @@ func defaultBaselinePolicyScaleParams() baselinePolicyScaleParams {
 	}
 }
 
+// oneRulePerPolicyScaleParams is the same total rule count spread over many
+// single-rule policies, a common shape where per-policy work (resolving the
+// policy from the endpoint's tier) costs as much as evaluating its one rule.
+func oneRulePerPolicyScaleParams() baselinePolicyScaleParams {
+	p := defaultBaselinePolicyScaleParams()
+	p.numPolicies = 2000
+	p.rulesPerPolicy = 1
+	return p
+}
+
 // ipSetSizeHistogram is the per-node IP set size distribution measured in the same
 // deployment: 3,708 sets, ~256k members in total, dominated by tiny sets with a long
 // tail of large ones.
@@ -127,9 +137,43 @@ func BenchmarkEvaluateBaselinePolicyScale(b *testing.B) {
 	b.Run("MatchEarly", func(b *testing.B) {
 		benchEvaluateBaselinePolicyScale(b, defaultBaselinePolicyScaleParams(), log.WarnLevel, true)
 	})
+	// Same rule count spread over single-rule policies, so per-policy work is
+	// not amortized over 68 rules.
+	b.Run("OneRulePerPolicy", func(b *testing.B) {
+		benchEvaluateBaselinePolicyScale(b, oneRulePerPolicyScaleParams(), log.WarnLevel, false)
+	})
+}
+
+// BenchmarkEvaluateBaselinePolicyScaleCompiled is BenchmarkEvaluateBaselinePolicyScale
+// with the store's policies compiled, as when a PolicyCompiler is configured. Missing
+// IP sets warn at compile time (outside the timed loop) rather than per flow, so there
+// is no MissingSetsLogsOff variant to distinguish.
+func BenchmarkEvaluateBaselinePolicyScaleCompiled(b *testing.B) {
+	b.Run("AllSetsPresent", func(b *testing.B) {
+		benchEvaluateBaselinePolicyScaleCompiled(b, defaultBaselinePolicyScaleParams(), log.WarnLevel, false)
+	})
+	b.Run("MissingSets", func(b *testing.B) {
+		p := defaultBaselinePolicyScaleParams()
+		p.numMissingIPSets = 8
+		benchEvaluateBaselinePolicyScaleCompiled(b, p, log.WarnLevel, false)
+	})
+	b.Run("MatchEarly", func(b *testing.B) {
+		benchEvaluateBaselinePolicyScaleCompiled(b, defaultBaselinePolicyScaleParams(), log.WarnLevel, true)
+	})
+	b.Run("OneRulePerPolicy", func(b *testing.B) {
+		benchEvaluateBaselinePolicyScaleCompiled(b, oneRulePerPolicyScaleParams(), log.WarnLevel, false)
+	})
 }
 
 func benchEvaluateBaselinePolicyScale(b *testing.B, p baselinePolicyScaleParams, level log.Level, matchEarly bool) {
+	benchEvaluateBaselinePolicyScaleImpl(b, p, level, matchEarly, false)
+}
+
+func benchEvaluateBaselinePolicyScaleCompiled(b *testing.B, p baselinePolicyScaleParams, level log.Level, matchEarly bool) {
+	benchEvaluateBaselinePolicyScaleImpl(b, p, level, matchEarly, true)
+}
+
+func benchEvaluateBaselinePolicyScaleImpl(b *testing.B, p baselinePolicyScaleParams, level log.Level, matchEarly, compiled bool) {
 	logger := log.StandardLogger()
 	counter, restoreLogging := withBenchLogging(level)
 	defer restoreLogging()
@@ -138,6 +182,17 @@ func benchEvaluateBaselinePolicyScale(b *testing.B, p baselinePolicyScaleParams,
 	if matchEarly {
 		addMatchEarlyPolicy(store, ep)
 	}
+	if compiled {
+		// Compiling moves the missing-set warnings to compile time (once per
+		// missing reference), off the per-flow path entirely.
+		compileStoreForTest(store)
+		if logger.IsLevelEnabled(log.WarnLevel) && counter.count.Load() != int64(expectedWarns) {
+			b.Fatalf("expected %d 'IPSet not found' warnings at compile time, got %d",
+				expectedWarns, counter.count.Load())
+		}
+		counter.count.Store(0)
+		expectedWarns = 0
+	}
 	flow := &MockFlow{
 		SourceIP:   net.ParseIP(benchSourceIP),
 		DestIP:     net.ParseIP(benchDestIP),
@@ -148,7 +203,7 @@ func benchEvaluateBaselinePolicyScale(b *testing.B, p baselinePolicyScaleParams,
 
 	// Pre-flight outside the timed loop: prove the walk is the intended one and that
 	// the warning count matches the analytic count, so that warnings/op is exact.
-	trace, _ := Evaluate(EnforcedOnly, rules.RuleDirIngress, store, ep, flow)
+	trace, _ := Evaluate(EnforcedOnly, rules.RuleDirIngress, store, ep, flow, nil)
 	if matchEarly {
 		if len(trace) != 1 || trace[0].Action != rules.RuleActionAllow || trace[0].Index != 0 {
 			b.Fatalf("expected an immediate allow from the match-early policy, got %v", trace)
@@ -167,7 +222,8 @@ func benchEvaluateBaselinePolicyScale(b *testing.B, p baselinePolicyScaleParams,
 	b.ReportAllocs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		benchTraceSink, _ = Evaluate(EnforcedOnly, rules.RuleDirIngress, store, ep, flow)
+		// Reuse the trace buffer, as the felix collector does.
+		benchTraceSink, _ = Evaluate(EnforcedOnly, rules.RuleDirIngress, store, ep, flow, benchTraceSink[:0])
 	}
 	b.StopTimer()
 	b.ReportMetric(float64(counter.count.Load())/float64(b.N), "warnings/op")
@@ -237,7 +293,10 @@ func buildBaselinePolicyStore(p baselinePolicyScaleParams) (*policystore.PolicyS
 		expectedWarns += refCount[id]
 	}
 
+	// The endpoint goes into the store, as dikastes' per-pod store holds it:
+	// evaluation resolves an endpoint's compiled form by identity.
 	ep := &proto.WorkloadEndpoint{Tiers: []*proto.TierInfo{tier}}
+	store.Endpoint = ep
 	return store, ep, expectedWarns
 }