Repository: kubernetes/kubernetes (language: go)

Issue:
Support skip watch Headless Service in server-side.
### What would you like to be added?

Support watch service with field-selector spec.clusterIP!=None to skip headless service in kube-apiserver. And kubelet/kube-proxy can skip watch headless-service in the future.
![image](https://github.com/kubernetes/kubernetes/assets/42128207/55a3df12-e70d-4b09-b5a4-a5a6ebff890e)

Now, endpoints will be added labels [service.kubernetes.io/headless](https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/endpoint/endpoints_controller.go#L518) for headless service  by endpoints-controller,  and kube-proxy will ignore  event for it by label-selector. https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-proxy/app/server.go#L913. But headless service haven't be added label and kube-proxy  still watch for it.

For kubelet, it watch service to add service-env for pods , but it ignores headless service when  build service-env-map(https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/kubelet_pods.go#L549). So kubelet don't need to watch headless service either.

### Why is this needed?

In a large-cluster with more then 5000 nodes,  when we run kubeflow task( tfjob:pod:headlessService=1:10:10) with qps 10, it can save huge network bandwidth and cpu for kube-apiserver.



Discussion / hints:
/sig network api-machinery
> In a large-cluster with more then 5000 nodes, when we run kubeflow task( tfjob:pod:headlessService=1:10:10) with qps 1

how many headless Services does that kubeflow task creates?
/sig scalability
this falls into that area too
> > In a large-cluster with more then 5000 nodes, when we run kubeflow task( tfjob:pod:headlessService=1:10:10) with qps 1
> 
> how many headless Services does that kubeflow task creates?

Under steady-state conditions, the cluster maintains 30,000 headless services. During daily new tasks, an additional 10,000 headless services will be created. The number of nodes in the cluster is around 7,000


interesting, I can see lately a lot of consumption of headless Services specially in AI/ML workloads,  instead of labelling Services we can add a field selector on the ClusterIP field, WDYT @wojtek-t @thockin ?

/cc @alculquicondor @ahg-g
Submitted https://github.com/kubernetes/kubernetes/pull/122541 , we are already having a selector for PodIP https://github.com/kubernetes/kubernetes/pull/50091/ so I think is ok having a field selector for the clusterIP field, that will bring the benefit of being able to build informers that filter headless services by using ` spec.clusterIP!="None"`
/assign
> Submitted #122541 , we are already having a selector for PodIP #50091 so I think is ok having a field selector for the clusterIP field, that will bring the benefit of being able to build informers that filter headless services by using ` spec.clusterIP!="None"` /assign

@aojea 
We firstly consider support the function  by field-selector for clusterIP,  compared to add label to headless service as we discussed in [#122502](https://github.com/kubernetes/kubernetes/pull/122502) , the main disadvantages is:

kube-apiserver needs to support field-selector with clusterIP  firstly, and then kubelet/kube-proxy can be updated. Otherwise, clients performing List-watch operations will receive errors like this,  which is not suitable for updating on existing clusters. 
![image](https://github.com/kubernetes/kubernetes/assets/42128207/c25d9026-d5bd-4f1f-a31a-b70630039c0b)
When using  the way for adding labels for headless-service,  kube-proxy  does not need to be updated (As it always list-watch with label-selector service.kubernetes.io/headless not exists) .
@aojea  If adding a field-selector  for clusterIP is considered to be ok. I think we can add IndexFields for service spec.clusterIP  as well (such as node metadata.name)，it can reduce events-count converted from kube-apiserver common channel to every cacheWatcher，reduce cpu-usage for kube-apiserver. 
> . I think we can add IndexFields for service spec.clusterIP as well (such as node metadata.name)，it can reduce events-count converted from kube-apiserver common channel to every cacheWatcher，reduce cpu-usage for kube-apiserver.

That is on sig-scalability area and better @wojtek-t to decide in that, I can see how Headless services are more used lately in Kubernetes , special fo AI/ML workloads, so I think is a fair request to implement this field selector and allow to filter server side headless service, but I do not know the tradeoffs of using IndexFields 
> instead of labelling Services we can add a field selector on the ClusterIP field, WDYT @wojtek-t @thockin ?

+1

I would be ok with adding IndexFields for headless services too [re tradeoff, it adds a bit of memory usage, but it is fairly low - there was a similar thing happening recently in https://github.com/kubernetes/kubernetes/issues/120778 that has a lot of good analysis done]
> I would be ok with adding IndexFields for headless services too [re tradeoff, it adds a bit of memory usage, but it is fairly low - there was a similar thing happening recently in https://github.com/[/issues/120778](https://github.com/kubernetes/kubernetes/issues/120778) that has a lot of good analysis done]

I can see that IndexFields implementation is feature gated, I think we can iterate here and go with the fieldselector first and evaluate later the IndexFields addition
The documentation currently says that headless services are labeled. Is this not accurate?

https://kubernetes.io/docs/reference/labels-annotations-taints/#servicekubernetesioheadless

> The documentation currently says that headless services are labeled. Is this not accurate?
> 
> https://kubernetes.io/docs/reference/labels-annotations-taints/#servicekubernetesioheadless

The control plane adds this label to an **Endpoints object** when the **owning Service** is headless.


/triage accepted
/reopen

We had to revert the fix https://github.com/kubernetes/kubernetes/pull/123896/files
@aojea: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes/kubernetes/issues/122394#issuecomment-1991532163):

>/reopen
>
>We had to revert the fix https://github.com/kubernetes/kubernetes/pull/123896/files


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/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

Fix patch:
diff --git a/cmd/kube-proxy/app/server.go b/cmd/kube-proxy/app/server.go
index 5706e307f441f..91d4d6e93b61f 100644
--- a/cmd/kube-proxy/app/server.go
+++ b/cmd/kube-proxy/app/server.go
@@ -537,7 +537,13 @@ func (s *ProxyServer) Run(ctx context.Context) error {
 	// Note: RegisterHandler() calls need to happen before creation of Sources because sources
 	// only notify on changes, and the initial update (on process start) may be lost if no handlers
 	// are registered yet.
-	serviceConfig := config.NewServiceConfig(ctx, informerFactory.Core().V1().Services(), s.Config.ConfigSyncPeriod.Duration)
+	// don't watch headless services for kube-proxy, they are proxied by DNS.
+	serviceInformerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.Config.ConfigSyncPeriod.Duration,
+		informers.WithTweakListOptions(func(options *metav1.ListOptions) {
+			options.LabelSelector = labelSelector.String()
+			options.FieldSelector = fields.OneTermNotEqualSelector("spec.clusterIP", v1.ClusterIPNone).String()
+		}))
+	serviceConfig := config.NewServiceConfig(ctx, serviceInformerFactory.Core().V1().Services(), s.Config.ConfigSyncPeriod.Duration)
 	serviceConfig.RegisterEventHandler(s.Proxier)
 	go serviceConfig.Run(ctx.Done())
 
@@ -553,6 +559,7 @@ func (s *ProxyServer) Run(ctx context.Context) error {
 	// This has to start after the calls to NewServiceConfig because that
 	// function must configure its shared informer event handlers first.
 	informerFactory.Start(wait.NeverStop)
+	serviceInformerFactory.Start(wait.NeverStop)
 
 	// Make an informer that selects for our nodename.
 	currentNodeInformerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.Config.ConfigSyncPeriod.Duration,


Test patch:
diff --git a/pkg/registry/core/service/strategy_test.go b/pkg/registry/core/service/strategy_test.go
index d5697d61d8243..f57c8b875cbdd 100644
--- a/pkg/registry/core/service/strategy_test.go
+++ b/pkg/registry/core/service/strategy_test.go
@@ -21,10 +21,12 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
@@ -789,10 +791,16 @@ func TestDropTypeDependentFields(t *testing.T) {
 }
 
 func TestMatchService(t *testing.T) {
+	noHeadlessServiceRequirement, err := labels.NewRequirement(v1.IsHeadlessService, selection.DoesNotExist, nil)
+	if err != nil {
+		t.Fatalf("Error creating no headless service requirement: %v", err)
+	}
+	noHeadlessServiceLabelSelector := labels.NewSelector().Add(*noHeadlessServiceRequirement)
 	testCases := []struct {
 		name          string
 		in            *api.Service
 		fieldSelector fields.Selector
+		labelSelector labels.Selector
 		expectMatch   bool
 	}{
 		{
@@ -805,6 +813,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("metadata.name=test"),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -817,6 +826,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("metadata.namespace=testns"),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -829,6 +839,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("metadata.name=nomatch"),
+			labelSelector: labels.Everything(),
 			expectMatch:   false,
 		},
 		{
@@ -841,6 +852,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("metadata.namespace=nomatch"),
+			labelSelector: labels.Everything(),
 			expectMatch:   false,
 		},
 		{
@@ -849,6 +861,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{Type: api.ServiceTypeLoadBalancer},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.type=LoadBalancer"),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -857,6 +870,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{Type: api.ServiceTypeNodePort},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.type=LoadBalancer"),
+			labelSelector: labels.Everything(),
 			expectMatch:   false,
 		},
 		{
@@ -865,6 +879,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP=None"),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -873,6 +888,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: "192.168.1.1"},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP=None"),
+			labelSelector: labels.Everything(),
 			expectMatch:   false,
 		},
 		{
@@ -881,6 +897,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: "192.168.1.1"},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP=192.168.1.1"),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -889,6 +906,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: "192.168.1.1"},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP!=None"),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -897,6 +915,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: "192.168.1.1"},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP!=\"\""),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -905,6 +924,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: "2001:db2::1"},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP=2001:db2::1"),
+			labelSelector: labels.Everything(),
 			expectMatch:   true,
 		},
 		{
@@ -913,6 +933,7 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP=192.168.1.1"),
+			labelSelector: labels.Everything(),
 			expectMatch:   false,
 		},
 		{
@@ -921,18 +942,49 @@ func TestMatchService(t *testing.T) {
 				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
 			},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP=2001:db2::1"),
+			labelSelector: labels.Everything(),
 			expectMatch:   false,
 		},
 		{
 			name:          "no match on empty service",
 			in:            &api.Service{},
 			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP=None"),
+			labelSelector: labels.Everything(),
+			expectMatch:   false,
+		},
+		{
+			name: "no match on headless service",
+			in: &api.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						v1.IsHeadlessService: "",
+					},
+				},
+			},
+			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP!=None"),
+			labelSelector: noHeadlessServiceLabelSelector,
+			expectMatch:   false,
+		},
+		{
+			name: "no match on headless service",
+			in: &api.Service{
+				Spec: api.ServiceSpec{ClusterIP: api.ClusterIPNone},
+			},
+			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP!=None"),
+			labelSelector: noHeadlessServiceLabelSelector,
 			expectMatch:   false,
 		},
+		{
+			name:          "match on empty service",
+			in:            &api.Service{},
+			fieldSelector: fields.ParseSelectorOrDie("spec.clusterIP!=None"),
+			labelSelector: noHeadlessServiceLabelSelector,
+			expectMatch:   true,
+		},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			m := Matcher(labels.Everything(), testCase.fieldSelector)
+			m := Matcher(testCase.labelSelector, testCase.fieldSelector)
 			result, err := m.Matches(testCase.in)
 			if err != nil {
 				t.Errorf("Unexpected error %v", err)