Repository: kubernetes/kubernetes (language: go)

Issue:
Still seeing the issue for endpoints staying out of sync
### What happened?

This issue https://github.com/kubernetes/kubernetes/issues/125638 was supposed to have fixed the issue where endpoint stay out of sync 
```
I0807 14:01:51.613700       2 endpoints_controller.go:348] "Error syncing endpoints, retrying" service="test1/test-qa" err="endpoints informer cache is out of date, resource version 10168236546 already processed for endpoints test1/test-qa"
I0807 14:01:51.624576       2 endpoints_controller.go:348] "Error syncing endpoints, retrying" service="test1/test-qa" err="endpoints informer cache is out of date, resource version 10168236546 already processed for endpoints test1/test-qa"
I0807 14:01:51.645704       2 endpoints_controller.go:348] "Error syncing endpoints, retrying" service="test1/test-qa" err="endpoints informer cache is out of date, resource version 10168236546 already processed for endpoints test1/test-qa"
I0807 14:01:51.686942       2 endpoints_controller.go:348] "Error syncing endpoints, retrying" service="test1/test-qa" err="endpoints informer cache is out of date, resource version 10168236546 already processed for endpoints test1/test-qa"
I0807 14:01:51.768648       2 endpoints_controller.go:348] "Error syncing endpoints, retrying" service="test1/test-qa" err="endpoints informer cache is out of date, resource version 10168236546 already processed for endpoints test1/test-qa"
I0807 14:01:51.808043       2 endpoints_controller.go:348] "Error syncing endpoints, retrying" service="test1/test2-qa" err="endpoints informer cache is out of date, resource version 10168250766 already processed for endpoints test1/test2-qa"
I0807 14:01:51.930345       2 endpoints_controller.go:348] "Error syncing endpoints, retrying" service="test1/test-qa" err="endpoints informer cache is out of date, resource version 10168236546 already processed for endpoints test1/test-qa"
```
I also wrote a small script which would get me the out of sync endpoints compared to the endpointslices 
```
from kubernetes.client import CoreV1Api, DiscoveryV1Api
from hubspot_kube_utils.client import build_kube_client
import json
import os
from datetime import datetime

def extract_ips_from_endpoint(endpoint):
    ips = set()
    if endpoint.subsets:
        for subset in endpoint.subsets:
            if subset.addresses:
                ips.update(addr.ip for addr in subset.addresses)
            if subset.not_ready_addresses:
                ips.update(addr.ip for addr in subset.not_ready_addresses)
    return ips

def extract_ips_from_endpoint_slice(slice):
    if not slice.endpoints:
        return set()
    return set(address for endpoint in slice.endpoints
               for address in (endpoint.addresses or []))

def compare_endpoints_and_slices(core_client, discovery_client):
    all_mismatches = []

    try:
        namespaces = core_client.list_namespace()
    except Exception as e:
        print(f"Error listing namespaces: {e}")
        return all_mismatches

    for ns in namespaces.items:
        namespace = ns.metadata.name
        print(f"Processing namespace: {namespace}")

        try:
            endpoints = core_client.list_namespaced_endpoints(namespace)
        except Exception as e:
            print(f"Error listing endpoints in namespace {namespace}: {e}")
            continue

        for endpoint in endpoints.items:
            name = endpoint.metadata.name

            try:
                slices = discovery_client.list_namespaced_endpoint_slice(namespace, label_selector=f"kubernetes.io/service-name={name}")
            except Exception as e:
                print(f"Error listing endpoint slices for service {name} in namespace {namespace}: {e}")
                continue

            endpoint_ips = extract_ips_from_endpoint(endpoint)
            slice_ips = set()

            for slice in slices.items:
                slice_ips.update(extract_ips_from_endpoint_slice(slice))

            if endpoint_ips != slice_ips:
                mismatch = {
                    "namespace": namespace,
                    "service_name": name,
                    "endpoint_ips": list(endpoint_ips),
                    "slice_ips": list(slice_ips),
                    "missing_in_endpoint": list(slice_ips - endpoint_ips),
                    "missing_in_slice": list(endpoint_ips - slice_ips)
                }
                all_mismatches.append(mismatch)

        print(f"Completed processing namespace: {namespace}")
        print("---")

    return all_mismatches

def save_to_json(data, cluster_name):
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"{cluster_name}_mismatches_{timestamp}.json"

    with open(filename, 'w') as f:
        json.dump(data, f, indent=2)

    print(f"Mismatch data for cluster {cluster_name} saved to {filename}")

def main():
    clusters = ["test"]
    all_cluster_mismatches = {}

    for cluster_name in clusters:
        print(f"Processing cluster: {cluster_name}")

        try:
            kube_client = build_kube_client(host="TEST",
                              token="TOKEN")

            core_client = CoreV1Api(kube_client)
            discovery_client = DiscoveryV1Api(kube_client)

            mismatches = compare_endpoints_and_slices(core_client, discovery_client)

            all_cluster_mismatches[cluster_name] = mismatches

            save_to_json(mismatches, cluster_name)

            print(f"Completed processing cluster: {cluster_name}")
            print(f"Total mismatches found in this cluster: {len(mismatches)}")
        except Exception as e:
            print(f"Error processing cluster {cluster_name}: {e}")


if __name__ == "__main__":
    main()
```

### What did you expect to happen?

I expect the endpoints to eventually sync and reflect the most upto date information. 

### How can we reproduce it (as minimally and precisely as possible)?

I have just deployed the newer patch to our cluster and that has resulted in endpoints never ending up being updated if the status goes out of sync. 

### Anything else we need to know?

_No response_

### Kubernetes version

Client Version: v1.29.7
Kustomize Version: v5.0.4-0.20230601165947-6ce0bf390ce3
Server Version: v1.29.7


### Cloud provider

<details>

</details>


### OS version

almalinux-9

### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

cri-o

### Related plugins (CNI, CSI, ...) and versions (if applicable)

_No response_


Discussion / hints:
This issue is currently awaiting triage.

If a SIG or subproject determines this is a relevant issue, they will accept it by applying the `triage/accepted` label and provide further guidance.

The `triage/accepted` label can be added by org members by writing `/triage accepted` in a comment.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>
/sig network
@kedar700 from the logs, endpoint controller was retrying test1/test-qa in the expected way. It has retried 5 times and the next retry should be after 320ms, which should long enough for the informer cache to get the updated endpoints. If there is no more logs related to the failure, I think it may have synced it successfully.

The python script used to detect the failure seems capturing diff between endpoints and endpointslice at a moment, but that could happen if there is any Pod change in the cluster, and technically the script is comparing endpoints of one moment with endpointslice of another moment.

So did you see the endpoints not being synced **after** endpoint controller stopped retrying it? 
Yeah i still see it and see similar messages continuously. Let me post some additional log messages after the one i posted above.
Observed the same issue in 1.29 and 1.30.
I keep seeing the same error then new pods were added behind a service. And the endpoint never catch up after all pods are settled.
`I0809 03:28:48.648274      11 endpoints_controller.go:355] "Error syncing endpoints, retrying" logger="endpoints-controller" service="kube-system/kube-proxy" err="endpoints informer cache is out of date, resource version 1340462 already processed for endpoints kube-system/kube-proxy"`

@kedar700 @mengqiy could you provide complete log of kube-controller-manager and the yamls of the related endpoints and endpointslice?
/assign @tnqn 

Are you able to triage this since you are already looking at this? 
Yes i can get that over by EOD today
> Are you able to triage this since you are already looking at this?

Sure, I'm waiting for more logs and the content of the endpoints to understand how it happened. Currently there are only retry logs which is expected in some cases.
@tnqn 
We found a customer that triggered an edge case in your change of #125675, which will cause Endpoints never updated.

They have a mutatingWebhook on Endpoints object that automatically injects a label, let's say `dummy-k: value`.
1. During syncServices, the endpoints controller [compares the Service's labels against Endpoints' labels](https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/endpoint/endpoints_controller.go#L499C34-L499C47), and they won't match since Endpoints have the extra "dummy-k: value".
2. the endpoints controller will [update the Endpoint's labels to be same as Service's labels](https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/endpoint/endpoints_controller.go#L540) by removing the `dummy-k: value` label. 
3. their mutatingWebhook will automatically adds back the `dummy-k: value` label as part of the update request. From the APIServer's POV, the Endpoints object don't have any changes, thus **the update request will succeed but returns the same resourceVersion as before.**
4. the change in #125675 will mark that resourceVersion as stale. 
5. Any future pods events won't be handled correctly due to the step#3 didn't result in a new resourceVersion.

I think we should fix this by check the resulting resourceVersion from the update request, and only mark the old resourceVersion as stale when a new resourceVersion is generated.

Not sure whether this is the same case others in this thread encountered :D
@M00nF1sh thanks for explaining the use case and the analysis. I did analyze the code after the issue was opened and suspected the only possibility is the returned resource version somehow equals to the one in the request but didn't know how it could happen after it has checked the desired endpoints and the current endpoints are different: https://github.com/kubernetes/kubernetes/blob/95b3fe9f15cdcaf98098be398478e70365b12dd7/pkg/controller/endpoint/endpoints_controller.go#L495-L503

Now I understand this could happen when the endpoints appear different on client side but are mutated to be same on server side. Do you plan to create a PR since you have proposed a fix?

@kedar700 @mengqiy could you confirm if this is also your case, i.e. there is a mutatingWebhook for endpoints objects that could make the endpoints different from what endpoints controller expects?
@tnqn 
Yeah, i can do a fix for this
@tnqn @M00nF1sh - can we assign to you?

@thockin Sure, I'm already assigned. 

@M00nF1sh Given that we've had three user reports on this issue, I think it's important to address it as soon as possible. Could you please let me know when you plan to create the PR? If you're currently occupied, I'm happy to handle it and submit the PR.

Fix patch:
diff --git a/pkg/controller/endpoint/endpoints_controller.go b/pkg/controller/endpoint/endpoints_controller.go
index 19596223e520d..8a61211f6082b 100644
--- a/pkg/controller/endpoint/endpoints_controller.go
+++ b/pkg/controller/endpoint/endpoints_controller.go
@@ -532,12 +532,13 @@ func (e *Controller) syncService(ctx context.Context, key string) error {
 	}
 
 	logger.V(4).Info("Update endpoints", "service", klog.KObj(service), "readyEndpoints", totalReadyEps, "notreadyEndpoints", totalNotReadyEps)
+	var updatedEndpoints *v1.Endpoints
 	if createEndpoints {
 		// No previous endpoints, create them
 		_, err = e.client.CoreV1().Endpoints(service.Namespace).Create(ctx, newEndpoints, metav1.CreateOptions{})
 	} else {
 		// Pre-existing
-		_, err = e.client.CoreV1().Endpoints(service.Namespace).Update(ctx, newEndpoints, metav1.UpdateOptions{})
+		updatedEndpoints, err = e.client.CoreV1().Endpoints(service.Namespace).Update(ctx, newEndpoints, metav1.UpdateOptions{})
 	}
 	if err != nil {
 		if createEndpoints && errors.IsForbidden(err) {
@@ -564,7 +565,10 @@ func (e *Controller) syncService(ctx context.Context, key string) error {
 	// If the current endpoints is updated we track the old resource version, so
 	// if we obtain this resource version again from the lister we know is outdated
 	// and we need to retry later to wait for the informer cache to be up-to-date.
-	if !createEndpoints {
+	// there are some operations (webhooks, truncated endpoints, ...) that can potentially cause endpoints updates became noop
+	// and return the same resourceVersion.
+	// Ref: https://issues.k8s.io/127370 , https://issues.k8s.io/126578
+	if updatedEndpoints != nil && updatedEndpoints.ResourceVersion != currentEndpoints.ResourceVersion {
 		e.staleEndpointsTracker.Stale(currentEndpoints)
 	}
 	return nil


Test patch:
diff --git a/test/integration/endpoints/endpoints_test.go b/test/integration/endpoints/endpoints_test.go
index 2dc71791d772c..516482273c9d5 100644
--- a/test/integration/endpoints/endpoints_test.go
+++ b/test/integration/endpoints/endpoints_test.go
@@ -28,6 +28,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
@@ -35,6 +36,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/endpoint"
 	"k8s.io/kubernetes/test/integration/framework"
 	"k8s.io/kubernetes/test/utils/ktesting"
+	netutils "k8s.io/utils/net"
 )
 
 func TestEndpointUpdates(t *testing.T) {
@@ -605,3 +607,199 @@ func newExternalNameService(namespace, name string) *v1.Service {
 	svc.Spec.ExternalName = "google.com"
 	return svc
 }
+
+func TestEndpointTruncate(t *testing.T) {
+	// Disable ServiceAccount admission plugin as we don't have serviceaccount controller running.
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, framework.DefaultTestServerFlags(), framework.SharedEtcd())
+	defer server.TearDownFn()
+
+	client, err := clientset.NewForConfig(server.ClientConfig)
+	if err != nil {
+		t.Fatalf("Error creating clientset: %v", err)
+	}
+
+	informers := informers.NewSharedInformerFactory(client, 0)
+
+	tCtx := ktesting.Init(t)
+	epController := endpoint.NewEndpointController(
+		tCtx,
+		informers.Core().V1().Pods(),
+		informers.Core().V1().Services(),
+		informers.Core().V1().Endpoints(),
+		client,
+		0)
+
+	// Start informer and controllers
+	informers.Start(tCtx.Done())
+	go epController.Run(tCtx, 1)
+
+	// Create namespace
+	ns := framework.CreateNamespaceOrDie(client, "test-endpoints-truncate", t)
+	defer framework.DeleteNamespaceOrDie(client, ns, t)
+
+	// Create a pod with labels
+	basePod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "test-pod",
+			Labels: labelMap(),
+		},
+		Spec: v1.PodSpec{
+			NodeName: "fake-node",
+			Containers: []v1.Container{
+				{
+					Name:  "fakename",
+					Image: "fakeimage",
+					Ports: []v1.ContainerPort{
+						{
+							Name:          "port-443",
+							ContainerPort: 443,
+						},
+					},
+				},
+			},
+		},
+		Status: v1.PodStatus{
+			Phase: v1.PodRunning,
+			Conditions: []v1.PodCondition{
+				{
+					Type:   v1.PodReady,
+					Status: v1.ConditionTrue,
+				},
+			},
+			PodIP: "10.0.0.1",
+			PodIPs: []v1.PodIP{
+				{
+					IP: "10.0.0.1",
+				},
+			},
+		},
+	}
+
+	// create 1001 Pods to reach endpoint max capacity that is set to 1000
+	allPodNames := sets.New[string]()
+	baseIP := netutils.BigForIP(netutils.ParseIPSloppy("10.0.0.1"))
+	for i := 0; i < 1001; i++ {
+		pod := basePod.DeepCopy()
+		pod.Name = fmt.Sprintf("%s-%d", basePod.Name, i)
+		allPodNames.Insert(pod.Name)
+		podIP := netutils.AddIPOffset(baseIP, i).String()
+		pod.Status.PodIP = podIP
+		pod.Status.PodIPs[0] = v1.PodIP{IP: podIP}
+		createdPod, err := client.CoreV1().Pods(ns.Name).Create(tCtx, pod, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatalf("Failed to create pod %s: %v", pod.Name, err)
+		}
+
+		createdPod.Status = pod.Status
+		_, err = client.CoreV1().Pods(ns.Name).UpdateStatus(tCtx, createdPod, metav1.UpdateOptions{})
+		if err != nil {
+			t.Fatalf("Failed to update status of pod %s: %v", pod.Name, err)
+		}
+	}
+
+	// Create a service associated to the pod
+	svc := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-service",
+			Namespace: ns.Name,
+			Labels: map[string]string{
+				"foo": "bar",
+			},
+		},
+		Spec: v1.ServiceSpec{
+			Selector: map[string]string{
+				"foo": "bar",
+			},
+			Ports: []v1.ServicePort{
+				{Name: "port-443", Port: 443, Protocol: "TCP", TargetPort: intstr.FromInt32(443)},
+			},
+		},
+	}
+	_, err = client.CoreV1().Services(ns.Name).Create(tCtx, svc, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("Failed to create service %s: %v", svc.Name, err)
+	}
+
+	var truncatedPodName string
+	// poll until associated Endpoints to the previously created Service exists
+	if err := wait.PollUntilContextTimeout(tCtx, 1*time.Second, 10*time.Second, true, func(context.Context) (bool, error) {
+		podNames := sets.New[string]()
+		endpoints, err := client.CoreV1().Endpoints(ns.Name).Get(tCtx, svc.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, nil
+		}
+
+		for _, subset := range endpoints.Subsets {
+			for _, address := range subset.Addresses {
+				podNames.Insert(address.TargetRef.Name)
+			}
+		}
+
+		if podNames.Len() != 1000 {
+			return false, nil
+		}
+
+		truncated, ok := endpoints.Annotations[v1.EndpointsOverCapacity]
+		if !ok || truncated != "truncated" {
+			return false, nil
+		}
+		// There is only 1 truncated Pod.
+		truncatedPodName, _ = allPodNames.Difference(podNames).PopAny()
+		return true, nil
+	}); err != nil {
+		t.Fatalf("endpoints not found: %v", err)
+	}
+
+	// Update the truncated Pod several times to make endpoints controller resync the service.
+	truncatedPod, err := client.CoreV1().Pods(ns.Name).Get(tCtx, truncatedPodName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to get pod %s: %v", truncatedPodName, err)
+	}
+	for i := 0; i < 10; i++ {
+		truncatedPod.Status.Conditions[0].Status = v1.ConditionFalse
+		truncatedPod, err = client.CoreV1().Pods(ns.Name).UpdateStatus(tCtx, truncatedPod, metav1.UpdateOptions{})
+		if err != nil {
+			t.Fatalf("Failed to update status of pod %s: %v", truncatedPod.Name, err)
+		}
+		truncatedPod.Status.Conditions[0].Status = v1.ConditionTrue
+		truncatedPod, err = client.CoreV1().Pods(ns.Name).UpdateStatus(tCtx, truncatedPod, metav1.UpdateOptions{})
+		if err != nil {
+			t.Fatalf("Failed to update status of pod %s: %v", truncatedPod.Name, err)
+		}
+	}
+
+	// delete 501 Pods
+	for i := 500; i < 1001; i++ {
+		podName := fmt.Sprintf("%s-%d", basePod.Name, i)
+		err = client.CoreV1().Pods(ns.Name).Delete(tCtx, podName, metav1.DeleteOptions{})
+		if err != nil {
+			t.Fatalf("error deleting test pod: %v", err)
+		}
+	}
+
+	// poll until endpoints for deleted Pod are no longer in Endpoints.
+	if err := wait.PollUntilContextTimeout(tCtx, 1*time.Second, 10*time.Second, true, func(context.Context) (bool, error) {
+		endpoints, err := client.CoreV1().Endpoints(ns.Name).Get(tCtx, svc.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, nil
+		}
+
+		numEndpoints := 0
+		for _, subset := range endpoints.Subsets {
+			numEndpoints += len(subset.Addresses)
+		}
+
+		if numEndpoints != 500 {
+			return false, nil
+		}
+
+		truncated, ok := endpoints.Annotations[v1.EndpointsOverCapacity]
+		if ok || truncated == "truncated" {
+			return false, nil
+		}
+
+		return true, nil
+	}); err != nil {
+		t.Fatalf("error checking for no endpoints with terminating pods: %v", err)
+	}
+}