DFP proposal

Author: yuval-k

api/v1alpha1/upstream_policy_types.go

@@ -29,11 +29,16 @@ type UpstreamList struct {
 	Items           []Upstream `json:"items"`
 }
 
-// +kubebuilder:validation:XValidation:message="There must one and only one upstream type set",rule="1 == (self.aws != null?1:0) + (self.static != null?1:0)"
+// +kubebuilder:validation:XValidation:message="There must one and only one upstream type set",rule="1 == (self.aws != null?1:0) + (self.static != null?1:0) + (self.dynamicForwardProxy != null?1:0)"
 type UpstreamSpec struct {
-	Aws    *AwsUpstream    `json:"aws,omitempty"`
-	Static *StaticUpstream `json:"static,omitempty"`
+	Aws                 *AwsUpstream         `json:"aws,omitempty"`
+	Static              *StaticUpstream      `json:"static,omitempty"`
+	DynamicForwardProxy *DynamicForwardProxy `json:"dynamicForwardProxy,omitempty"`
 }
+
+type DynamicForwardProxy struct {
+}
+
 type AwsUpstream struct {
 	Region    string                      `json:"region,omitempty"`
 	SecretRef corev1.LocalObjectReference `json:"secretRef,omitempty"`

internal/gateway2/extensions2/plugins/upstream/aws.go

@@ -110,10 +110,10 @@ func (p *plugin2) processBackendAws(
 ) error {
 
 	functionName := dest.FunctionName
-	if p.needFilter == nil {
-		p.needFilter = make(map[string]bool)
+	if p.needAwsFilter == nil {
+		p.needAwsFilter = make(map[string]bool)
 	}
-	p.needFilter[pCtx.FilterChainName] = true
+	p.needAwsFilter[pCtx.FilterChainName] = true
 
 	lambdaRouteFunc := &awspb.AWSLambdaPerRoute{
 		Async: false,

internal/gateway2/extensions2/plugins/upstream/dfp.go

@@ -0,0 +1,74 @@
+package upstream
+
+import (
+	"context"
+
+	envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	envoy_config_route_v3 "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	envoyclusters "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/dynamic_forward_proxy/v3"
+	envoydfp "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/dynamic_forward_proxy/v3"
+
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/kgateway-dev/kgateway/api/v1alpha1"
+	"github.com/kgateway-dev/kgateway/internal/gateway2/ir"
+)
+
+func (p *plugin2) processDFPRoute(ctx context.Context, pCtx *ir.RouteContext, outputRoute *envoy_config_route_v3.Route) {
+
+	for _, b := range pCtx.In.Backends {
+		if u := b.Backend.Upstream; u != nil {
+			if ir, ok := u.ObjIr.(*UpstreamIr); ok {
+				if ir.dfpFilterConfig != nil {
+					filters := p.neededDfpFilter[pCtx.FilterChainName]
+					if _, ok := filters[u.ClusterName()]; !ok {
+						filters[u.ClusterName()] = ir.dfpFilterConfig
+						p.neededDfpFilter[pCtx.FilterChainName] = filters
+					}
+				}
+			}
+		}
+	}
+
+	dfpFilterConfig := &envoydfp.FilterConfig{
+		ImplementationSpecifier: &envoydfp.FilterConfig_SubClusterConfig{
+			SubClusterConfig: &envoydfp.SubClusterConfig{},
+		},
+	}
+	var anyFilter anypb.Any
+	err := anypb.MarshalFrom(&anyFilter, dfpFilterConfig, proto.MarshalOptions{Deterministic: true})
+	// error should never happen here. panic?
+	if err != nil {
+		panic(err)
+	}
+	// TODO: make sure anyFilter makes it to the perfilter config of the route.
+}
+
+func processDFPCluster(ctx context.Context, in *v1alpha1.DynamicForwardProxy, out *envoy_config_cluster_v3.Cluster) {
+
+	out.LbPolicy = envoy_config_cluster_v3.Cluster_CLUSTER_PROVIDED
+	c := &envoyclusters.ClusterConfig{
+		ClusterImplementationSpecifier: &envoyclusters.ClusterConfig_SubClustersConfig{
+			SubClustersConfig: &envoyclusters.SubClustersConfig{
+				LbPolicy: envoy_config_cluster_v3.Cluster_LEAST_REQUEST,
+			},
+		},
+	}
+	var anyCluster anypb.Any
+	err := anypb.MarshalFrom(&anyCluster, c, proto.MarshalOptions{Deterministic: true})
+	// error should never happen here. panic?
+	if err != nil {
+		panic(err)
+	}
+
+	// the upstream has a DNS name. We need Envoy to resolve the DNS name
+	// set the type to strict dns
+	out.ClusterDiscoveryType = &envoy_config_cluster_v3.Cluster_ClusterType{
+		ClusterType: &envoy_config_cluster_v3.Cluster_CustomClusterType{
+			Name:        "envoy.clusters.dynamic_forward_proxy",
+			TypedConfig: &anyCluster,
+		},
+	}
+
+}

internal/gateway2/extensions2/plugins/upstream/plugin.go

@@ -6,13 +6,15 @@ import (
 	"maps"
 	"time"
 
+	"google.golang.org/protobuf/proto"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
 
 	envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	envoy_config_listener_v3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	envoy_config_route_v3 "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	envoydfp "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/dynamic_forward_proxy/v3"
 	awspb "github.com/solo-io/envoy-gloo/go/config/filter/http/aws_lambda/v2"
 	skubeclient "istio.io/istio/pkg/config/schema/kubeclient"
 	"istio.io/istio/pkg/kube/kclient"
@@ -62,7 +64,8 @@ func (d *upstreamDestination) Equals(in any) bool {
 }
 
 type UpstreamIr struct {
-	AwsSecret *ir.Secret
+	AwsSecret       *ir.Secret
+	dfpFilterConfig *envoydfp.FilterConfig
 }
 
 func (u *UpstreamIr) data() map[string][]byte {
@@ -77,13 +80,17 @@ func (u *UpstreamIr) Equals(other any) bool {
 	if !ok {
 		return false
 	}
-	return maps.EqualFunc(u.data(), otherUpstream.data(), func(a, b []byte) bool {
+	if !maps.EqualFunc(u.data(), otherUpstream.data(), func(a, b []byte) bool {
 		return bytes.Equal(a, b)
-	})
+	}) {
+		return false
+	}
+	return proto.Equal(otherUpstream.dfpFilterConfig, u.dfpFilterConfig)
 }
 
 type plugin2 struct {
-	needFilter map[string]bool
+	needAwsFilter   map[string]bool
+	neededDfpFilter map[string]map[string]*envoydfp.FilterConfig
 }
 
 func registerTypes(ourCli versioned.Interface) {
@@ -171,6 +178,14 @@ func buildTranslateFunc(secrets *krtcollections.SecretIndex) func(krtctx krt.Han
 				// return error
 			}
 		}
+		if i.Spec.DynamicForwardProxy != nil {
+			ir.dfpFilterConfig = &envoydfp.FilterConfig{
+				ImplementationSpecifier: &envoydfp.FilterConfig_SubClusterConfig{
+					SubClusterConfig: &envoydfp.SubClusterConfig{},
+				},
+			}
+		}
+
 		return &ir
 	}
 }
@@ -195,6 +210,8 @@ func processUpstream(ctx context.Context, in ir.Upstream, out *envoy_config_clus
 		processStatic(ctx, spec.Static, out)
 	case spec.Aws != nil:
 		processAws(ctx, spec.Aws, ir, out)
+	case spec.DynamicForwardProxy != nil:
+		processDFPCluster(ctx, spec.DynamicForwardProxy, out)
 	}
 }
 
@@ -221,7 +238,10 @@ func processEndpoints(up *v1alpha1.Upstream) *ir.EndpointsForUpstream {
 }
 
 func newPlug(ctx context.Context, tctx ir.GwTranslationCtx) ir.ProxyTranslationPass {
-	return &plugin2{}
+	return &plugin2{
+		needAwsFilter:   map[string]bool{},
+		neededDfpFilter: map[string]map[string]*envoydfp.FilterConfig{},
+	}
 }
 
 func (p *plugin2) Name() string {
@@ -237,7 +257,7 @@ func (p *plugin2) ApplyVhostPlugin(ctx context.Context, pCtx *ir.VirtualHostCont
 
 // called 0 or more times
 func (p *plugin2) ApplyForRoute(ctx context.Context, pCtx *ir.RouteContext, outputRoute *envoy_config_route_v3.Route) error {
-
+	p.processDFPRoute(ctx, pCtx, outputRoute)
 	return nil
 }
 
@@ -257,16 +277,20 @@ func (p *plugin2) ApplyForRouteBackend(
 // if a plugin emits new filters, they must be with a plugin unique name.
 // any filter returned from route config must be disabled, so it doesnt impact other routes.
 func (p *plugin2) HttpFilters(ctx context.Context, fc ir.FilterChainCommon) ([]plugins.StagedHttpFilter, error) {
-	if !p.needFilter[fc.FilterChainName] {
-		return nil, nil
+	var ret []plugins.StagedHttpFilter
+	if p.needAwsFilter[fc.FilterChainName] {
+		filterConfig := &awspb.AWSLambdaConfig{}
+		pluginStage := plugins.DuringStage(plugins.OutAuthStage)
+		f, _ := plugins.NewStagedFilter(FilterName, filterConfig, pluginStage)
+		ret = append(ret, f)
+	}
+	for clstrName, filterConfig := range p.neededDfpFilter[fc.FilterChainName] {
+		pluginStage := plugins.DuringStage(plugins.OutAuthStage)
+		f, _ := plugins.NewStagedFilter(clstrName, filterConfig, pluginStage)
+		ret = append(ret, f)
 	}
-	filterConfig := &awspb.AWSLambdaConfig{}
-	pluginStage := plugins.DuringStage(plugins.OutAuthStage)
-	f, _ := plugins.NewStagedFilter(FilterName, filterConfig, pluginStage)
 
-	return []plugins.StagedHttpFilter{
-		f,
-	}, nil
+	return ret, nil
 }
 
 func (p *plugin2) UpstreamHttpFilters(ctx context.Context) ([]plugins.StagedUpstreamHttpFilter, error) {

internal/gateway2/ir/iface.go

@@ -35,8 +35,9 @@ func (r *RouteBackendContext) AddTypedConfig(key string, v *anypb.Any) {
 }
 
 type RouteContext struct {
-	Policy PolicyIR
-	In     HttpRouteRuleMatchIR
+	FilterChainName string
+	Policy          PolicyIR
+	In              HttpRouteRuleMatchIR
 }
 
 type ProxyTranslationPass interface {

internal/gateway2/translator/irtranslator/route.go

@@ -215,8 +215,9 @@ func (h *httpRouteConfigurationTranslator) runRoutePlugins(ctx context.Context,
 			}
 			for _, pol := range pols {
 				pctx := &ir.RouteContext{
-					Policy: pol.PolicyIr,
-					In:     in,
+					FilterChainName: h.fc.FilterChainName,
+					Policy:          pol.PolicyIr,
+					In:              in,
 				}
 				err := pass.ApplyForRoute(ctx, pctx, out)
 				if err != nil {