don't watch headless services on kube-proxy

Author: Sakuralbj

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,

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)