feat: continue using and drain serving endpointslices during termination

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

@@ -1320,9 +1320,7 @@
 		} 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)
 			}
 		}
@@ -1452,9 +1450,7 @@
 		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)
 	}
 
@@ -1665,15 +1661,25 @@
 	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

@@ -1472,6 +1472,8 @@
 	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
@@ -1503,10 +1505,11 @@
 }
 
 // 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

@@ -547,7 +547,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

@@ -26,6 +26,7 @@ new features: |
   Added support for cost specifier in the rate limit 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
   Added support for GEP-1731 (HTTPRoute Retries)
   Added support for routing to Backend resources in the GRPCRoute, TCPRoute and UDPRoute APIs