Add Upstream Service Targeting to Property Override Extension

agent/envoyextensions/builtin/aws-lambda/aws_lambda.go

@@ -72,16 +72,19 @@ func (a *awsLambda) CanApply(config *extensioncommon.RuntimeConfig) bool {
 
 // PatchRoute modifies the routing configuration for a service of kind TerminatingGateway. If the kind is
 // not TerminatingGateway, then it can not be modified.
-func (a *awsLambda) PatchRoute(r *extensioncommon.RuntimeConfig, route *envoy_route_v3.RouteConfiguration) (*envoy_route_v3.RouteConfiguration, bool, error) {
-	if r.Kind != api.ServiceKindTerminatingGateway {
-		return route, false, nil
+func (a *awsLambda) PatchRoute(p extensioncommon.RoutePayload) (*envoy_route_v3.RouteConfiguration, bool, error) {
+	cfg := p.RuntimeConfig
+	if cfg.Kind != api.ServiceKindTerminatingGateway {
+		return p.Message, false, nil
 	}
 
 	// Only patch outbound routes.
-	if extensioncommon.IsRouteToLocalAppCluster(route) {
-		return route, false, nil
+	if p.IsInbound() {
+		return p.Message, false, nil
 	}
 
+	route := p.Message
+
 	for _, virtualHost := range route.VirtualHosts {
 		for _, route := range virtualHost.Routes {
 			action, ok := route.Action.(*envoy_route_v3.Route_Route)
@@ -101,16 +104,18 @@ func (a *awsLambda) PatchRoute(r *extensioncommon.RuntimeConfig, route *envoy_ro
 }
 
 // PatchCluster patches the provided envoy cluster with data required to support an AWS lambda function
-func (a *awsLambda) PatchCluster(_ *extensioncommon.RuntimeConfig, c *envoy_cluster_v3.Cluster) (*envoy_cluster_v3.Cluster, bool, error) {
+func (a *awsLambda) PatchCluster(p extensioncommon.ClusterPayload) (*envoy_cluster_v3.Cluster, bool, error) {
 	// Only patch outbound clusters.
-	if extensioncommon.IsLocalAppCluster(c) {
-		return c, false, nil
+	if p.IsInbound() {
+		return p.Message, false, nil
 	}
 
 	transportSocket, err := extensioncommon.MakeUpstreamTLSTransportSocket(&envoy_tls_v3.UpstreamTlsContext{
 		Sni: "*.amazonaws.com",
 	})
 
+	c := p.Message
+
 	if err != nil {
 		return c, false, fmt.Errorf("failed to make transport socket: %w", err)
 	}
@@ -168,9 +173,11 @@ func (a *awsLambda) PatchCluster(_ *extensioncommon.RuntimeConfig, c *envoy_clus
 
 // PatchFilter patches the provided envoy filter with an inserted lambda filter being careful not to
 // overwrite the http filters.
-func (a *awsLambda) PatchFilter(_ *extensioncommon.RuntimeConfig, filter *envoy_listener_v3.Filter, isInboundListener bool) (*envoy_listener_v3.Filter, bool, error) {
+func (a *awsLambda) PatchFilter(p extensioncommon.FilterPayload) (*envoy_listener_v3.Filter, bool, error) {
+	filter := p.Message
+
 	// Only patch outbound filters.
-	if isInboundListener {
+	if p.IsInbound() {
 		return filter, false, nil
 	}
 

agent/envoyextensions/builtin/aws-lambda/aws_lambda_test.go

@@ -202,7 +202,10 @@ func TestPatchCluster(t *testing.T) {
 
 			// Test patching the cluster
 			rc := extensioncommon.RuntimeConfig{}
-			patchedCluster, patchSuccess, err := tc.lambda.PatchCluster(&rc, tc.input)
+			patchedCluster, patchSuccess, err := tc.lambda.PatchCluster(extensioncommon.ClusterPayload{
+				RuntimeConfig: &rc,
+				Message:       tc.input,
+			})
 			if tc.isErrExpected {
 				assert.Error(t, err)
 				assert.False(t, patchSuccess)
@@ -307,7 +310,10 @@ func TestPatchRoute(t *testing.T) {
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
 			l := awsLambda{}
-			r, ok, err := l.PatchRoute(tc.conf, tc.route)
+			r, ok, err := l.PatchRoute(extensioncommon.RoutePayload{
+				RuntimeConfig: tc.conf,
+				Message:       tc.route,
+			})
 			require.NoError(t, err)
 			require.Equal(t, tc.expectRoute, r)
 			require.Equal(t, tc.expectBool, ok)
@@ -456,7 +462,14 @@ func TestPatchFilter(t *testing.T) {
 				PayloadPassthrough: true,
 				InvocationMode:     "asynchronous",
 			}
-			f, ok, err := l.PatchFilter(nil, tc.filter, tc.isInboundFilter)
+			d := extensioncommon.TrafficDirectionOutbound
+			if tc.isInboundFilter {
+				d = extensioncommon.TrafficDirectionInbound
+			}
+			f, ok, err := l.PatchFilter(extensioncommon.FilterPayload{
+				Message:          tc.filter,
+				TrafficDirection: d,
+			})
 			require.Equal(t, tc.expectBool, ok)
 			if tc.expectErr == "" {
 				require.NoError(t, err)

agent/envoyextensions/builtin/ext-authz/structs.go

@@ -601,7 +601,7 @@ func (t Target) clusterName(cfg *cmn.RuntimeConfig) (string, error) {
 
 	for service, upstream := range cfg.Upstreams {
 		if service == t.Service {
-			for sni := range upstream.SNI {
+			for sni := range upstream.SNIs {
 				return sni, nil
 			}
 		}

agent/envoyextensions/builtin/http/localratelimit/ratelimit.go

@@ -102,10 +102,11 @@ func (p *ratelimit) CanApply(config *extensioncommon.RuntimeConfig) bool {
 
 // PatchFilter inserts a http local rate_limit filter at the head of
 // envoy.filters.network.http_connection_manager filters
-func (p ratelimit) PatchFilter(_ *extensioncommon.RuntimeConfig, filter *envoy_listener_v3.Filter, isInboundListener bool) (*envoy_listener_v3.Filter, bool, error) {
+func (r ratelimit) PatchFilter(p extensioncommon.FilterPayload) (*envoy_listener_v3.Filter, bool, error) {
+	filter := p.Message
 	// rate limit is only applied to the inbound listener of the service itself
 	// since the limit is aggregated from all downstream connections.
-	if !isInboundListener {
+	if !p.IsInbound() {
 		return filter, false, nil
 	}
 
@@ -123,34 +124,34 @@ func (p ratelimit) PatchFilter(_ *extensioncommon.RuntimeConfig, filter *envoy_l
 
 	tokenBucket := envoy_type_v3.TokenBucket{}
 
-	if p.TokensPerFill != nil {
+	if r.TokensPerFill != nil {
 		tokenBucket.TokensPerFill = &wrappers.UInt32Value{
-			Value: uint32(*p.TokensPerFill),
+			Value: uint32(*r.TokensPerFill),
 		}
 	}
-	if p.MaxTokens != nil {
-		tokenBucket.MaxTokens = uint32(*p.MaxTokens)
+	if r.MaxTokens != nil {
+		tokenBucket.MaxTokens = uint32(*r.MaxTokens)
 	}
 
-	if p.FillInterval != nil {
-		tokenBucket.FillInterval = durationpb.New(time.Duration(*p.FillInterval) * time.Second)
+	if r.FillInterval != nil {
+		tokenBucket.FillInterval = durationpb.New(time.Duration(*r.FillInterval) * time.Second)
 	}
 
 	var FilterEnabledDefault *envoy_core_v3.RuntimeFractionalPercent
-	if p.FilterEnabled != nil {
+	if r.FilterEnabled != nil {
 		FilterEnabledDefault = &envoy_core_v3.RuntimeFractionalPercent{
 			DefaultValue: &envoy_type_v3.FractionalPercent{
-				Numerator:   *p.FilterEnabled,
+				Numerator:   *r.FilterEnabled,
 				Denominator: envoy_type_v3.FractionalPercent_HUNDRED,
 			},
 		}
 	}
 
 	var FilterEnforcedDefault *envoy_core_v3.RuntimeFractionalPercent
-	if p.FilterEnforced != nil {
+	if r.FilterEnforced != nil {
 		FilterEnforcedDefault = &envoy_core_v3.RuntimeFractionalPercent{
 			DefaultValue: &envoy_type_v3.FractionalPercent{
-				Numerator:   *p.FilterEnforced,
+				Numerator:   *r.FilterEnforced,
 				Denominator: envoy_type_v3.FractionalPercent_HUNDRED,
 			},
 		}

agent/envoyextensions/builtin/lua/lua.go

@@ -67,14 +67,16 @@ func (l *lua) CanApply(config *extensioncommon.RuntimeConfig) bool {
 	return string(config.Kind) == l.ProxyType
 }
 
-func (l *lua) matchesListenerDirection(isInboundListener bool) bool {
+func (l *lua) matchesListenerDirection(p extensioncommon.FilterPayload) bool {
+	isInboundListener := p.IsInbound()
 	return (!isInboundListener && l.Listener == "outbound") || (isInboundListener && l.Listener == "inbound")
 }
 
 // PatchFilter inserts a lua filter directly prior to envoy.filters.http.router.
-func (l *lua) PatchFilter(_ *extensioncommon.RuntimeConfig, filter *envoy_listener_v3.Filter, isInboundListener bool) (*envoy_listener_v3.Filter, bool, error) {
+func (l *lua) PatchFilter(p extensioncommon.FilterPayload) (*envoy_listener_v3.Filter, bool, error) {
+	filter := p.Message
 	// Make sure filter matches extension config.
-	if !l.matchesListenerDirection(isInboundListener) {
+	if !l.matchesListenerDirection(p) {
 		return filter, false, nil
 	}