feat: continue using and drain serving endpointslices during termination

Signed-off-by: Daniele Fognini <[email protected]>

internal/gatewayapi/listener.go

@@ -541,7 +541,7 @@ func destinationSettingFromHostAndPort(host string, port uint32) []*ir.Destinati
 		{
 			Weight:    ptr.To[uint32](1),
 			Protocol:  ir.GRPC,
-			Endpoints: []*ir.DestinationEndpoint{ir.NewDestEndpoint(host, port)},
+			Endpoints: []*ir.DestinationEndpoint{ir.NewDestEndpoint(host, port, false)},
 		},
 	}
 }

internal/gatewayapi/route.go

@@ -1280,9 +1280,7 @@ func (t *Translator) processDestination(backendRefContext BackendRefContext,
 		} else {
 			backendIps := resources.GetServiceImport(backendNamespace, string(backendRef.Name)).Spec.IPs
 			for _, ip := range backendIps {
-				ep := ir.NewDestEndpoint(
-					ip,
-					uint32(*backendRef.Port))
+				ep := ir.NewDestEndpoint(ip, uint32(*backendRef.Port), false)
 				endpoints = append(endpoints, ep)
 			}
 		}
@@ -1412,9 +1410,7 @@ func (t *Translator) processServiceDestinationSetting(
 		endpoints, addrType = getIREndpointsFromEndpointSlices(endpointSlices, servicePort.Name, servicePort.Protocol)
 	} else {
 		// Fall back to Service ClusterIP routing
-		ep := ir.NewDestEndpoint(
-			service.Spec.ClusterIP,
-			uint32(*backendRef.Port))
+		ep := ir.NewDestEndpoint(service.Spec.ClusterIP, uint32(*backendRef.Port), false)
 		endpoints = append(endpoints, ep)
 	}
 
@@ -1625,15 +1621,25 @@ func getIREndpointsFromEndpointSlice(endpointSlice *discoveryv1.EndpointSlice, p
 	for _, endpoint := range endpointSlice.Endpoints {
 		for _, endpointPort := range endpointSlice.Ports {
 			// Check if the endpoint port matches the service port
-			// and if endpoint is Ready
-			if *endpointPort.Name == portName &&
-				*endpointPort.Protocol == portProtocol &&
-				// Unknown state (nil) should be interpreted as Ready, see https://pkg.go.dev/k8s.io/api/discovery/v1#EndpointConditions
-				(endpoint.Conditions.Ready == nil || *endpoint.Conditions.Ready) {
+			if *endpointPort.Name != portName || *endpointPort.Protocol != portProtocol {
+				continue
+			}
+			conditions := endpoint.Conditions
+			// Unknown Serving/Terminating (nil) should fall-back to Ready, see https://pkg.go.dev/k8s.io/api/discovery/v1#EndpointConditions
+			if conditions.Serving != nil && conditions.Terminating != nil {
+				// Check if the endpoint is serving
+				if !*conditions.Serving {
+					continue
+				}
+				// Drain the endpoint if it is being terminated
+				draining := *conditions.Terminating
+				for _, address := range endpoint.Addresses {
+					ep := ir.NewDestEndpoint(address, uint32(*endpointPort.Port), draining)
+					endpoints = append(endpoints, ep)
+				}
+			} else if conditions.Ready == nil || *conditions.Ready {
 				for _, address := range endpoint.Addresses {
-					ep := ir.NewDestEndpoint(
-						address,
-						uint32(*endpointPort.Port))
+					ep := ir.NewDestEndpoint(address, uint32(*endpointPort.Port), false)
 					endpoints = append(endpoints, ep)
 				}
 			}

internal/gatewayapi/route_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -23,7 +24,7 @@ func TestGetIREndpointsFromEndpointSlices(t *testing.T) {
 		endpointSlices    []*discoveryv1.EndpointSlice
 		portName          string
 		portProtocol      corev1.Protocol
-		expectedEndpoints int
+		expectedEndpoints []*ir.DestinationEndpoint
 		expectedAddrType  ir.DestinationAddressType
 	}{
 		{
@@ -51,10 +52,14 @@ func TestGetIREndpointsFromEndpointSlices(t *testing.T) {
 					},
 				},
 			},
-			portName:          "http",
-			portProtocol:      corev1.ProtocolTCP,
-			expectedEndpoints: 3,
-			expectedAddrType:  ir.IP,
+			portName:     "http",
+			portProtocol: corev1.ProtocolTCP,
+			expectedEndpoints: []*ir.DestinationEndpoint{
+				{Host: "192.0.2.1", Port: 80, Draining: false},
+				{Host: "192.0.2.2", Port: 80, Draining: false},
+				{Host: "2001:db8::1", Port: 80, Draining: false},
+			},
+			expectedAddrType: ir.IP,
 		},
 		{
 			name: "Mixed IP and FQDN endpoints",
@@ -80,10 +85,13 @@ func TestGetIREndpointsFromEndpointSlices(t *testing.T) {
 					},
 				},
 			},
-			portName:          "http",
-			portProtocol:      corev1.ProtocolTCP,
-			expectedEndpoints: 2,
-			expectedAddrType:  ir.MIXED,
+			portName:     "http",
+			portProtocol: corev1.ProtocolTCP,
+			expectedEndpoints: []*ir.DestinationEndpoint{
+				{Host: "192.0.2.1", Port: 80, Draining: false},
+				{Host: "example.com", Port: 80, Draining: false},
+			},
+			expectedAddrType: ir.MIXED,
 		},
 		{
 			name: "Dual-stack IP endpoints",
@@ -111,10 +119,15 @@ func TestGetIREndpointsFromEndpointSlices(t *testing.T) {
 					},
 				},
 			},
-			portName:          "http",
-			portProtocol:      corev1.ProtocolTCP,
-			expectedEndpoints: 4,
-			expectedAddrType:  ir.IP,
+			portName:     "http",
+			portProtocol: corev1.ProtocolTCP,
+			expectedEndpoints: []*ir.DestinationEndpoint{
+				{Host: "192.0.2.1", Port: 80, Draining: false},
+				{Host: "192.0.2.2", Port: 80, Draining: false},
+				{Host: "2001:db8::1", Port: 80, Draining: false},
+				{Host: "2001:db8::2", Port: 80, Draining: false},
+			},
+			expectedAddrType: ir.IP,
 		},
 		{
 			name: "Dual-stack with FQDN",
@@ -150,10 +163,43 @@ func TestGetIREndpointsFromEndpointSlices(t *testing.T) {
 					},
 				},
 			},
-			portName:          "http",
-			portProtocol:      corev1.ProtocolTCP,
-			expectedEndpoints: 3,
-			expectedAddrType:  ir.MIXED,
+			portName:     "http",
+			portProtocol: corev1.ProtocolTCP,
+			expectedEndpoints: []*ir.DestinationEndpoint{
+				{Host: "192.0.2.1", Port: 80, Draining: false},
+				{Host: "2001:db8::1", Port: 80, Draining: false},
+				{Host: "example.com", Port: 80, Draining: false},
+			},
+			expectedAddrType: ir.MIXED,
+		},
+		{
+			name: "Keep serving and terminating as draining",
+			endpointSlices: []*discoveryv1.EndpointSlice{
+				{
+					ObjectMeta:  metav1.ObjectMeta{Name: "slice1"},
+					AddressType: discoveryv1.AddressTypeIPv4,
+					Endpoints: []discoveryv1.Endpoint{
+						{Addresses: []string{"192.0.2.1"}, Conditions: discoveryv1.EndpointConditions{
+							Ready: ptr.To(false), Serving: ptr.To(true), Terminating: ptr.To(true),
+						}},
+						{Addresses: []string{"192.0.2.2"}, Conditions: discoveryv1.EndpointConditions{
+							Ready: ptr.To(false), Serving: ptr.To(false), Terminating: ptr.To(true),
+						}},
+						{Addresses: []string{"192.0.2.3"}, Conditions: discoveryv1.EndpointConditions{
+							Ready: ptr.To(false),
+						}},
+					},
+					Ports: []discoveryv1.EndpointPort{
+						{Name: ptr.To("http"), Port: ptr.To(int32(80)), Protocol: ptr.To(corev1.ProtocolTCP)},
+					},
+				},
+			},
+			portName:     "http",
+			portProtocol: corev1.ProtocolTCP,
+			expectedEndpoints: []*ir.DestinationEndpoint{
+				{Host: "192.0.2.1", Port: 80, Draining: true},
+			},
+			expectedAddrType: ir.IP,
 		},
 	}
 
@@ -170,10 +216,13 @@ func TestGetIREndpointsFromEndpointSlices(t *testing.T) {
 				fmt.Printf("  Endpoint %d:\n", i+1)
 				fmt.Printf("    Address: %s\n", endpoint.Host)
 				fmt.Printf("    Port: %d\n", endpoint.Port)
+				fmt.Printf("    Draining: %t\n", endpoint.Draining)
 
 			}
 
 			fmt.Println()
+			require.Equal(t, tt.expectedEndpoints, endpoints)
+			require.Equal(t, tt.expectedAddrType, *addrType)
 		})
 	}
 }

internal/ir/xds.go

@@ -1414,6 +1414,8 @@ type DestinationEndpoint struct {
 	Port uint32 `json:"port" yaml:"port"`
 	// Path refers to the Unix Domain Socket
 	Path *string `json:"path,omitempty" yaml:"path,omitempty"`
+	// Draining is true if this endpoint should be drained
+	Draining bool `json:"draining,omitempty" yaml:"draining,omitempty"`
 }
 
 // Validate the fields within the DestinationEndpoint structure
@@ -1445,10 +1447,11 @@ func (d DestinationEndpoint) Validate() error {
 }
 
 // NewDestEndpoint creates a new DestinationEndpoint.
-func NewDestEndpoint(host string, port uint32) *DestinationEndpoint {
+func NewDestEndpoint(host string, port uint32, draining bool) *DestinationEndpoint {
 	return &DestinationEndpoint{
-		Host: host,
-		Port: port,
+		Host:     host,
+		Port:     port,
+		Draining: draining,
 	}
 }
 

internal/xds/translator/cluster.go

@@ -435,13 +435,18 @@ func buildXdsClusterLoadAssignment(clusterName string, destSettings []*ir.Destin
 		}
 
 		for _, irEp := range ds.Endpoints {
+			healthStatus := corev3.HealthStatus_UNKNOWN
+			if irEp.Draining {
+				healthStatus = corev3.HealthStatus_DRAINING
+			}
 			lbEndpoint := &endpointv3.LbEndpoint{
 				Metadata: metadata,
 				HostIdentifier: &endpointv3.LbEndpoint_Endpoint{
 					Endpoint: &endpointv3.Endpoint{
 						Address: buildAddress(irEp),
 					},
 				},
+				HealthStatus: healthStatus,
 			}
 			// Set default weight of 1 for all endpoints.
 			lbEndpoint.LoadBalancingWeight = &wrapperspb.UInt32Value{Value: 1}

internal/xds/translator/oidc.go

@@ -352,9 +352,7 @@ func createOAuth2TokenEndpointCluster(tCtx *types.ResourceVersionTable,
 	ds = &ir.DestinationSetting{
 		Weight: ptr.To[uint32](1),
 		Endpoints: []*ir.DestinationEndpoint{
-			ir.NewDestEndpoint(
-				cluster.hostname,
-				cluster.port),
+			ir.NewDestEndpoint(cluster.hostname, cluster.port, false),
 		},
 	}
 

internal/xds/translator/ratelimit.go

@@ -483,7 +483,7 @@ func (t *Translator) createRateLimitServiceCluster(tCtx *types.ResourceVersionTa
 	ds := &ir.DestinationSetting{
 		Weight:    ptr.To[uint32](1),
 		Protocol:  ir.GRPC,
-		Endpoints: []*ir.DestinationEndpoint{ir.NewDestEndpoint(host, port)},
+		Endpoints: []*ir.DestinationEndpoint{ir.NewDestEndpoint(host, port, false)},
 	}
 
 	tSocket, err := buildRateLimitTLSocket()

internal/xds/translator/testdata/in/xds-ir/http-route.yaml

@@ -26,3 +26,6 @@ http:
       - endpoints:
         - host: "1.2.3.4"
           port: 50000
+        - host: "1.2.3.5"
+          port: 50000
+          draining: true

internal/xds/translator/testdata/out/xds-ir/http-route.endpoints.yaml

@@ -7,6 +7,13 @@
             address: 1.2.3.4
             portValue: 50000
       loadBalancingWeight: 1
+    - endpoint:
+        address:
+          socketAddress:
+            address: 1.2.3.5
+            portValue: 50000
+      healthStatus: DRAINING
+      loadBalancingWeight: 1
     loadBalancingWeight: 1
     locality:
       region: first-route-dest/backend/0

internal/xds/translator/utils.go

@@ -179,7 +179,7 @@ func addClusterFromURL(url string, tCtx *types.ResourceVersionTable) error {
 
 	ds = &ir.DestinationSetting{
 		Weight:    ptr.To[uint32](1),
-		Endpoints: []*ir.DestinationEndpoint{ir.NewDestEndpoint(uc.hostname, uc.port)},
+		Endpoints: []*ir.DestinationEndpoint{ir.NewDestEndpoint(uc.hostname, uc.port, false)},
 	}
 
 	clusterArgs := &xdsClusterArgs{

release-notes/current.yaml

@@ -22,6 +22,7 @@ new features: |
   Added support for preserving the user defined HTTPRoute match order in EnvoyProxy API
   Added support for specifying dynamic metadata namespaces that External Processing services can access read from and write to in EnvoyExtensionPolicy API
   Added support for API Key Authentication in the SecurityPolicy API
+  Continue using and drain endpoints during their graceful termination, as indicated by their respective EndpointConditions
 
 bug fixes: |
   Fixed a nil pointer error that occurs when a SecurityPolicy refers to a UDS backend