feat: add endpoints webhook for node autonomy

charts/yurt-manager/templates/yurt-manager-auto-generated.yaml

@@ -1419,6 +1419,25 @@ webhooks:
     resources:
     - deployments
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: yurt-manager-webhook-service
+      namespace: {{ .Release.Namespace }}
+      path: /mutate-core-openyurt-io-v1-endpoints
+  failurePolicy: Ignore
+  name: mutate.core.v1.endpoints.openyurt.io
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - UPDATE
+    resources:
+    - endpoints
+  sideEffects: None
 - admissionReviewVersions:
   - v1
   - v1beta1

pkg/yurtmanager/controller/util/pod/pod_util.go

@@ -58,6 +58,16 @@ func IsPodReadyConditionTrue(status v1.PodStatus) bool {
 	return condition != nil && condition.Status == v1.ConditionTrue
 }
 
+// IsPodCrashLoopBackOff returns true if a pod is in CrashLoopBackOff state; false otherwise.
+func IsPodCrashLoopBackOff(status v1.PodStatus) bool {
+	for _, c := range status.ContainerStatuses {
+		if c.State.Waiting != nil && c.State.Waiting.Reason == "CrashLoopBackOff" {
+			return true
+		}
+	}
+	return false
+}
+
 // GetPodReadyCondition extracts the pod ready condition from the given status and returns that.
 // Returns nil if the condition is not present.
 func GetPodReadyCondition(status v1.PodStatus) *v1.PodCondition {

pkg/yurtmanager/controller/util/pod/pod_util_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/rand"
@@ -168,3 +169,49 @@ func TestUpdatePodCondition(t *testing.T) {
 		})
 	}
 }
+
+func TestIsPodCrashLoopBackoff(t *testing.T) {
+	testCases := []struct {
+		name   string
+		status v1.PodStatus
+		expect bool
+	}{
+		{
+			name: "yes",
+			status: v1.PodStatus{
+				ContainerStatuses: []v1.ContainerStatus{
+					{
+						State: v1.ContainerState{
+							Waiting: &v1.ContainerStateWaiting{
+								Reason: "CrashLoopBackOff",
+							},
+						},
+					},
+				},
+			},
+			expect: true,
+		},
+		{
+			name: "no",
+			status: v1.PodStatus{
+				ContainerStatuses: []v1.ContainerStatus{
+					{
+						State: v1.ContainerState{},
+					},
+				},
+			},
+			expect: false,
+		},
+		{
+			name:   "empty",
+			status: v1.PodStatus{},
+			expect: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.expect, IsPodCrashLoopBackOff(tc.status))
+		})
+	}
+}

pkg/yurtmanager/webhook/endpoints/v1/endpoints_default.go

@@ -0,0 +1,130 @@
+/*
+Copyright 2024 The OpenYurt Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	nodeutil "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/util/node"
+	podutil "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/util/pod"
+)
+
+// Default satisfies the defaulting webhook interface.
+func (webhook *EndpointsHandler) Default(ctx context.Context, obj runtime.Object) error {
+	endpoints, ok := obj.(*corev1.Endpoints)
+	if !ok {
+		return fmt.Errorf("expected an Endpoints object but got %T", obj)
+	}
+
+	return remapAutonomyEndpoints(ctx, webhook.Client, endpoints)
+}
+
+// isNodeAutonomous checks if the node has autonomy annotations
+// and returns true if it does, false otherwise.
+func isNodeAutonomous(ctx context.Context, c client.Client, nodeName string) (bool, error) {
+	node := &corev1.Node{}
+	err := c.Get(ctx, client.ObjectKey{Name: nodeName}, node)
+	if err != nil {
+		// If node doesn't exist, it doesn't have autonomy
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	return nodeutil.IsPodBoundenToNode(node), nil
+}
+
+// isPodCrashLoopBackOff checks if the pod is crashloopbackoff
+// and returns true if it is, false otherwise.
+func isPodCrashLoopBackOff(ctx context.Context, c client.Client, podName, namespace string) (bool, error) {
+	pod := &corev1.Pod{}
+	err := c.Get(ctx, client.ObjectKey{Name: podName, Namespace: namespace}, pod)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	return podutil.IsPodCrashLoopBackOff(pod.Status), nil
+}
+
+// remapAutonomyEndpoints remaps the notReadyAddresses to the readyAddresses
+// for the subsets scheduled to nodes that have autonomy annotations.
+// The function checks the pod status and if the pod is not in crashloopbackoff,
+// it remaps the address to readyAddresses.
+func remapAutonomyEndpoints(ctx context.Context, client client.Client, endpoints *corev1.Endpoints) error {
+	// Track nodes with autonomy to avoid repeated checks
+	nodesWithAutonomy := make(map[string]bool)
+
+	// Get all the notReadyAddresses for subsets
+	for i, s := range endpoints.Subsets {
+		// Create a zero-length slice with the same underlying array
+		newNotReadyAddresses := s.NotReadyAddresses[:0]
+
+		for _, a := range s.NotReadyAddresses {
+			if a.NodeName == nil || a.TargetRef == nil {
+				newNotReadyAddresses = append(newNotReadyAddresses, a)
+				continue
+			}
+
+			// Get the node and check autonomy annotations
+			hasAutonomy, ok := nodesWithAutonomy[*a.NodeName]
+			if !ok {
+				isAutonomous, err := isNodeAutonomous(ctx, client, *a.NodeName)
+				if err != nil {
+					return err
+				}
+				// Store autonomy status for future checks
+				nodesWithAutonomy[*a.NodeName] = isAutonomous
+				hasAutonomy = isAutonomous
+			}
+
+			// If the node doesn't have autonomy, skip
+			if !hasAutonomy {
+				newNotReadyAddresses = append(newNotReadyAddresses, a)
+				continue
+			}
+
+			// Get the pod
+			isPodCrashLoopBackOff, err := isPodCrashLoopBackOff(ctx, client, a.TargetRef.Name, a.TargetRef.Namespace)
+			if err != nil {
+				return err
+			}
+
+			if isPodCrashLoopBackOff {
+				newNotReadyAddresses = append(newNotReadyAddresses, a)
+				continue
+			}
+
+			// Move the address to the ready addresses in the subset
+			endpoints.Subsets[i].Addresses = append(endpoints.Subsets[i].Addresses, *a.DeepCopy())
+		}
+
+		// Update the subset with the new notReadyAddresses
+		endpoints.Subsets[i].NotReadyAddresses = newNotReadyAddresses
+	}
+
+	return nil
+}