allow multiple endpoint plugins to apply on a single backend (#11144)

Author: stevenctl

internal/kgateway/endpoints/prioritize.go

@@ -14,14 +14,26 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
-func PrioritizeEndpoints(logger *zap.Logger, priorityInfo *PriorityInfo, ep ir.EndpointsForBackend, ucc ir.UniqlyConnectedClient) *envoy_config_endpoint_v3.ClusterLoadAssignment {
+// EndpointsInputs is the collective IR that can be modified
+// by endpoint plugins to influence the ClusterLoadAssignment.
+type EndpointsInputs struct {
+	EndpointsForBackend ir.EndpointsForBackend
+	PriorityInfo        *PriorityInfo
+}
+
+// PrioritizeEndpoints converts EndpointsInputs into a ClusterLoadAssignment.
+func PrioritizeEndpoints(
+	logger *zap.Logger,
+	ucc ir.UniqlyConnectedClient,
+	inputs EndpointsInputs,
+) *envoy_config_endpoint_v3.ClusterLoadAssignment {
 	lbInfo := LoadBalancingInfo{
 		PodLabels:    ucc.Labels,
 		PodLocality:  ucc.Locality,
-		PriorityInfo: priorityInfo,
+		PriorityInfo: inputs.PriorityInfo,
 	}
 
-	return prioritizeWithLbInfo(logger, ep, lbInfo)
+	return prioritizeWithLbInfo(logger, inputs.EndpointsForBackend, lbInfo)
 }
 
 type LoadBalancingInfo struct {

internal/kgateway/extensions2/plugin/plugin.go

@@ -5,9 +5,9 @@ import (
 	"encoding/json"
 
 	envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
-	envoy_config_endpoint_v3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	gwv1 "sigs.k8s.io/gateway-api/apis/v1"
 
+	"github.com/kgateway-dev/kgateway/v2/internal/kgateway/endpoints"
 	"github.com/kgateway-dev/kgateway/v2/internal/kgateway/ir"
 
 	"istio.io/istio/pkg/kube/krt"
@@ -35,8 +35,8 @@ type (
 		kctx krt.HandlerContext,
 		ctx context.Context,
 		ucc ir.UniqlyConnectedClient,
-		in ir.EndpointsForBackend,
-	) (*envoy_config_endpoint_v3.ClusterLoadAssignment, uint64)
+		out *endpoints.EndpointsInputs,
+	) uint64
 )
 
 // TODO: consider changing PerClientProcessBackend to look like this:

internal/kgateway/extensions2/plugins/destrule/destrule_plugin.go

@@ -9,9 +9,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
-	envoy_config_endpoint_v3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	envoy_type_v3 "github.com/envoyproxy/go-control-plane/envoy/type/v3"
-	"github.com/solo-io/go-utils/contextutils"
 	"istio.io/api/networking/v1alpha3"
 	"istio.io/istio/pkg/config/schema/gvr"
 	"istio.io/istio/pkg/kube/krt"
@@ -55,25 +53,30 @@ type destrulePlugin struct {
 	destinationRulesIndex DestinationRuleIndex
 }
 
-func (d *destrulePlugin) processEndpoints(kctx krt.HandlerContext, ctx context.Context, ucc ir.UniqlyConnectedClient, in ir.EndpointsForBackend) (*envoy_config_endpoint_v3.ClusterLoadAssignment, uint64) {
-	destrule := d.destinationRulesIndex.FetchDestRulesFor(kctx, ucc.Namespace, in.Hostname, ucc.Labels)
+// processEndpoints tries to find a destination rule
+// for the backend and if it does, it updates the PriorityInfo on `out`.
+func (d *destrulePlugin) processEndpoints(
+	kctx krt.HandlerContext,
+	ctx context.Context,
+	ucc ir.UniqlyConnectedClient,
+	out *endpoints.EndpointsInputs,
+) uint64 {
+	destrule := d.destinationRulesIndex.FetchDestRulesFor(kctx, ucc.Namespace, out.EndpointsForBackend.Hostname, ucc.Labels)
 	if destrule == nil {
-		return nil, 0
+		return 0
 	}
 
-	logger := contextutils.LoggerFrom(ctx).Desugar()
-	trafficPolicy := getTrafficPolicy(destrule, in.Port)
+	trafficPolicy := getTrafficPolicy(destrule, out.EndpointsForBackend.Port)
 	localityLb := getLocalityLbSetting(trafficPolicy)
-	var priorityInfo *endpoints.PriorityInfo
-	var additionalHash uint64
-	if localityLb != nil {
-		priorityInfo = getPriorityInfoFromDestrule(localityLb)
-		hasher := fnv.New64()
-		hasher.Write([]byte(destrule.UID))
-		hasher.Write([]byte(fmt.Sprintf("%v", destrule.Generation)))
-		additionalHash = hasher.Sum64()
+	if localityLb == nil {
+		return 0
 	}
-	return endpoints.PrioritizeEndpoints(logger, priorityInfo, in, ucc), additionalHash
+
+	out.PriorityInfo = getPriorityInfoFromDestrule(localityLb)
+	hasher := fnv.New64()
+	hasher.Write([]byte(destrule.UID))
+	hasher.Write([]byte(fmt.Sprintf("%v", destrule.Generation)))
+	return hasher.Sum64()
 }
 
 func (d *destrulePlugin) processBackend(kctx krt.HandlerContext, ctx context.Context, ucc ir.UniqlyConnectedClient, in ir.BackendObjectIR, outCluster *envoy_config_cluster_v3.Cluster) {

internal/kgateway/proxy_syncer/cla_test.go

@@ -61,7 +61,11 @@ func TestTranslatesDestrulesFailoverPriority(t *testing.T) {
 		}),
 	}
 
-	cla := endpoints.PrioritizeEndpoints(nil, priorityInfo, *efu, ucc)
+	epInputs := endpoints.EndpointsInputs{
+		EndpointsForBackend: *efu,
+		PriorityInfo:        priorityInfo,
+	}
+	cla := endpoints.PrioritizeEndpoints(nil, ucc, epInputs)
 	g.Expect(cla.Endpoints).To(gomega.HaveLen(2))
 
 	remoteLocality := cla.Endpoints[0]
@@ -123,7 +127,11 @@ func TestTranslatesDestrulesFailover(t *testing.T) {
 
 	priorityInfo := &endpoints.PriorityInfo{}
 
-	cla := endpoints.PrioritizeEndpoints(nil, priorityInfo, *efu, ucc)
+	epInputs := endpoints.EndpointsInputs{
+		EndpointsForBackend: *efu,
+		PriorityInfo:        priorityInfo,
+	}
+	cla := endpoints.PrioritizeEndpoints(nil, ucc, epInputs)
 	g.Expect(cla.Endpoints).To(gomega.HaveLen(2))
 
 	remoteLocality := cla.Endpoints[0]

internal/kgateway/translator/irtranslator/backend.go

@@ -79,12 +79,16 @@ func (t *BackendTranslator) runPolicies(
 	inlineEps *ir.EndpointsForBackend,
 	out *clusterv3.Cluster,
 ) {
-	for gk, policyPlugin := range t.ContributedPolicies {
-		if inlineEps != nil && policyPlugin.PerClientProcessEndpoints != nil {
-			if cla, _ := policyPlugin.PerClientProcessEndpoints(kctx, ctx, ucc, *inlineEps); cla != nil {
-				out.LoadAssignment = cla
-			}
+	// if the backend was initialized with inlineEps then we
+	// need an EndpointsInputs to run plugins against
+	var endpointInputs *endpoints.EndpointsInputs
+	if inlineEps != nil {
+		endpointInputs = &endpoints.EndpointsInputs{
+			EndpointsForBackend: *inlineEps,
 		}
+	}
+
+	for gk, policyPlugin := range t.ContributedPolicies {
 		// TODO: in theory it would be nice to do `ProcessBackend` once, and only do
 		// the the per-client processing for each client.
 		// that would require refactoring and thinking about the proper IR, so we'll punt on that for
@@ -94,6 +98,11 @@ func (t *BackendTranslator) runPolicies(
 			policyPlugin.PerClientProcessBackend(kctx, ctx, ucc, backend, out)
 		}
 
+		// run endpoint plugins if we have endpoints to process
+		if endpointInputs != nil && policyPlugin.PerClientProcessEndpoints != nil {
+			policyPlugin.PerClientProcessEndpoints(kctx, ctx, ucc, endpointInputs)
+		}
+
 		if policyPlugin.ProcessBackend == nil {
 			continue
 		}
@@ -102,13 +111,13 @@ func (t *BackendTranslator) runPolicies(
 		}
 	}
 
-	// if no plugin initialized the inline CLA, and the cluster type needs one, do it now
-	if out.GetLoadAssignment() == nil && inlineEps != nil && clusterSupportsInlineCLA(out) {
+	// for clusters that want a CLA _and_ initialized with inlineEps, build the CLA.
+	// never overwrite the CLA that was already initialized (potentially within a plugin).
+	if out.GetLoadAssignment() == nil && endpointInputs != nil && clusterSupportsInlineCLA(out) {
 		out.LoadAssignment = endpoints.PrioritizeEndpoints(
-			contextutils.LoggerFrom(ctx).Desugar(), // TODO BackendTranslator's logger
-			nil,
-			*inlineEps,
+			contextutils.LoggerFrom(ctx).Desugar(),
 			ucc,
+			*endpointInputs,
 		)
 	}
 }

internal/kgateway/translator/translator.go

@@ -141,20 +141,13 @@ func (s *CombinedTranslator) TranslateGateway(kctx krt.HandlerContext, ctx conte
 }
 
 func (s *CombinedTranslator) TranslateEndpoints(kctx krt.HandlerContext, ucc ir.UniqlyConnectedClient, ep ir.EndpointsForBackend) (*envoy_config_endpoint_v3.ClusterLoadAssignment, uint64) {
-	// check if we have a plugin to do it
-	cla, additionalHash := proccessWithPlugins(s.endpointPlugins, kctx, context.TODO(), ucc, ep)
-	if cla != nil {
-		return cla, additionalHash
+	epInputs := endpoints.EndpointsInputs{
+		EndpointsForBackend: ep,
 	}
-	return endpoints.PrioritizeEndpoints(s.logger, nil, ep, ucc), 0
-}
-
-func proccessWithPlugins(plugins []extensionsplug.EndpointPlugin, kctx krt.HandlerContext, ctx context.Context, ucc ir.UniqlyConnectedClient, in ir.EndpointsForBackend) (*envoy_config_endpoint_v3.ClusterLoadAssignment, uint64) {
-	for _, processEnddpoints := range plugins {
-		cla, additionalHash := processEnddpoints(kctx, context.TODO(), ucc, in)
-		if cla != nil {
-			return cla, additionalHash
-		}
+	var hash uint64
+	for _, processEndpoints := range s.endpointPlugins {
+		additionalHash := processEndpoints(kctx, context.TODO(), ucc, &epInputs)
+		hash ^= additionalHash
 	}
-	return nil, 0
+	return endpoints.PrioritizeEndpoints(s.logger, ucc, epInputs), hash
 }