fix(egress): prevent stale active-endpoint overwrite (#1701)

Make the endpoint watcher the sole active-endpoint annotation writer so a Service snapshot captured during startup cannot replace a newer endpoint. Preserve intentionally empty snapshots across endpoint-to-zero-to-same transitions, select the cleared annotation from the configured egress family, reject malformed or wrong-family endpoints, and preserve the other family in dual-stack state.

Fixes #1673.

Signed-off-by: Marcel Fest <marcel.fest@telekom.de>

pkg/endpoints/endpoints.go

@@ -200,28 +200,41 @@ func (p *Processor) updateAnnotations(service *v1.Service, lastKnownGoodEndpoint
 	egressUpdateFunc func(context.Context, *v1.Service) error) {
 	// Set the service accordingly
 	if service.Annotations[kubevip.Egress] == "true" {
-		ip := net.ParseIP(*lastKnownGoodEndpoint)
+		if *lastKnownGoodEndpoint != "" {
+			ip := net.ParseIP(*lastKnownGoodEndpoint)
+			expectIPv6 := service.Annotations[kubevip.EgressIPv6] == "true"
+			if ip == nil || (ip.To4() == nil) != expectIPv6 {
+				log.Warn("ignoring active endpoint with unexpected address family",
+					"service", service.Name,
+					"namespace", service.Namespace,
+					"endpoint", *lastKnownGoodEndpoint,
+					"expected_ipv6", expectIPv6)
+				return
+			}
+		}
 
 		// Store old values from ServiceSnapshot to detect if annotation actually changed
 		// We use the ServiceSnapshot instead of the service parameter because the service parameter
 		// may have stale annotations if the last update failed
 		var oldEndpoint, oldEndpointIPv6 string
+		snapshotFound := false
 		if p.instances != nil {
 			serviceInstance := instance.FindServiceInstance(service, *p.instances)
-			if serviceInstance != nil {
+			if serviceInstance != nil && serviceInstance.ServiceSnapshot != nil {
+				snapshotFound = true
 				oldEndpoint = serviceInstance.ServiceSnapshot.Annotations[kubevip.ActiveEndpoint]
 				oldEndpointIPv6 = serviceInstance.ServiceSnapshot.Annotations[kubevip.ActiveEndpointIPv6]
 			}
 		}
-		// Fall back to service annotations if we couldn't find the instance
-		if oldEndpoint == "" && oldEndpointIPv6 == "" {
+		// Empty annotations in an existing snapshot are meaningful after a zero-endpoint transition.
+		if !snapshotFound {
 			oldEndpoint = service.Annotations[kubevip.ActiveEndpoint]
 			oldEndpointIPv6 = service.Annotations[kubevip.ActiveEndpointIPv6]
 		}
 
 		// Determine which annotation to update based on IP version
 		var endpoint, endpointIPv6 string
-		if ip.To4() == nil && !p.config.EnableEndpoints {
+		if service.Annotations[kubevip.EgressIPv6] == "true" && !p.config.EnableEndpoints {
 			// IPv6
 			endpointIPv6 = *lastKnownGoodEndpoint
 			endpoint = oldEndpoint // Preserve existing IPv4 if any

pkg/endpoints/endpoints_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/kube-vip/kube-vip/pkg/endpoints/providers"
+	"github.com/kube-vip/kube-vip/pkg/instance"
 	"github.com/kube-vip/kube-vip/pkg/kubevip"
 	"github.com/kube-vip/kube-vip/pkg/lease"
 	"github.com/kube-vip/kube-vip/pkg/metrics"
@@ -17,6 +18,7 @@ import (
 	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
 )
 
 func TestShouldAllowReconcileWithoutEndpoints(t *testing.T) {
@@ -47,6 +49,191 @@ type fakeWorker struct {
 	processCalled bool
 }
 
+type annotationUpdate struct {
+	endpoint     string
+	endpointIPv6 string
+}
+
+type recordingProvider struct {
+	providers.Provider
+	updates []annotationUpdate
+}
+
+func (p *recordingProvider) UpdateServiceAnnotation(_ context.Context, endpoint, endpointIPv6 string,
+	_ *v1.Service, _ *kubernetes.Clientset) error {
+	p.updates = append(p.updates, annotationUpdate{endpoint: endpoint, endpointIPv6: endpointIPv6})
+	return nil
+}
+
+func TestUpdateAnnotationsZeroEndpointsThenSameEndpoint(t *testing.T) {
+	for _, enableEndpoints := range []bool{true, false} {
+		providerName := "EndpointSlices"
+		provider := providers.NewEndpointslices()
+		if enableEndpoints {
+			providerName = "Endpoints"
+			provider = providers.NewEndpoints()
+		}
+
+		for _, family := range []struct {
+			name       string
+			endpoint   string
+			other      string
+			egressIPv6 bool
+		}{
+			{name: "IPv4", endpoint: "10.0.0.1", other: "fd00::1"},
+			{name: "IPv6", endpoint: "fd00::1", other: "10.0.0.1", egressIPv6: true},
+		} {
+			t.Run(providerName+"/"+family.name, func(t *testing.T) {
+				annotations := map[string]string{kubevip.Egress: "true"}
+				if family.egressIPv6 {
+					annotations[kubevip.EgressIPv6] = "true"
+				}
+				if !enableEndpoints {
+					if family.egressIPv6 {
+						annotations[kubevip.ActiveEndpoint] = family.other
+						annotations[kubevip.ActiveEndpointIPv6] = family.endpoint
+					} else {
+						annotations[kubevip.ActiveEndpoint] = family.endpoint
+						annotations[kubevip.ActiveEndpointIPv6] = family.other
+					}
+				} else {
+					annotations[kubevip.ActiveEndpoint] = family.endpoint
+				}
+				service := &v1.Service{ObjectMeta: metav1.ObjectMeta{
+					Name: "test-service", Namespace: "default", UID: "test-uid", Annotations: annotations,
+				}}
+				serviceInstance := &instance.Instance{ServiceSnapshot: service.DeepCopy()}
+				instances := []*instance.Instance{serviceInstance}
+				recorder := &recordingProvider{Provider: provider}
+				processor := &Processor{
+					config:    &kubevip.Config{EnableEndpoints: enableEndpoints},
+					provider:  recorder,
+					instances: &instances,
+				}
+
+				updateSnapshot := func(_ context.Context, updated *v1.Service) error {
+					serviceInstance.ServiceSnapshot = updated
+					return nil
+				}
+
+				noEndpoint := ""
+				processor.updateAnnotations(service, &noEndpoint, nil, updateSnapshot)
+				repopulatedEndpoint := family.endpoint
+				processor.updateAnnotations(service, &repopulatedEndpoint, nil, updateSnapshot)
+
+				cleared := annotationUpdate{}
+				repopulated := annotationUpdate{endpoint: family.endpoint}
+				if !enableEndpoints {
+					if family.egressIPv6 {
+						cleared = annotationUpdate{endpoint: family.other}
+						repopulated = annotationUpdate{endpoint: family.other, endpointIPv6: family.endpoint}
+					} else {
+						cleared = annotationUpdate{endpointIPv6: family.other}
+						repopulated = annotationUpdate{endpoint: family.endpoint, endpointIPv6: family.other}
+					}
+				}
+				want := []annotationUpdate{cleared, repopulated}
+				if len(recorder.updates) != len(want) {
+					t.Fatalf("annotation updates = %+v, want %+v", recorder.updates, want)
+				}
+				for index := range want {
+					if recorder.updates[index] != want[index] {
+						t.Errorf("annotation update %d = %+v, want %+v", index, recorder.updates[index], want[index])
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestUpdateAnnotationsEndpointSlicesClearsConfiguredFamily(t *testing.T) {
+	for _, test := range []struct {
+		name       string
+		egressIPv6 bool
+		want       annotationUpdate
+	}{
+		{name: "IPv4", want: annotationUpdate{endpointIPv6: "fd00::1"}},
+		{name: "IPv6", egressIPv6: true, want: annotationUpdate{endpoint: "10.0.0.1"}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			annotations := map[string]string{
+				kubevip.Egress:             "true",
+				kubevip.ActiveEndpoint:     "10.0.0.1",
+				kubevip.ActiveEndpointIPv6: "fd00::1",
+			}
+			if test.egressIPv6 {
+				annotations[kubevip.EgressIPv6] = "true"
+			}
+			service := &v1.Service{ObjectMeta: metav1.ObjectMeta{
+				Name: "test-service", Namespace: "default", UID: "test-uid", Annotations: annotations,
+			}}
+			instances := []*instance.Instance{{ServiceSnapshot: service.DeepCopy()}}
+			recorder := &recordingProvider{Provider: providers.NewEndpointslices()}
+			processor := &Processor{
+				config:    &kubevip.Config{EnableEndpoints: false},
+				provider:  recorder,
+				instances: &instances,
+			}
+
+			noEndpoint := ""
+			processor.updateAnnotations(service, &noEndpoint, nil, func(context.Context, *v1.Service) error { return nil })
+
+			if len(recorder.updates) != 1 || recorder.updates[0] != test.want {
+				t.Fatalf("annotation updates = %+v, want [%+v]", recorder.updates, test.want)
+			}
+		})
+	}
+}
+
+func TestUpdateAnnotationsValidatesEndpointFamily(t *testing.T) {
+	tests := []struct {
+		name       string
+		endpoint   string
+		egressIPv6 bool
+		want       annotationUpdate
+		wantUpdate bool
+	}{
+		{name: "invalid address", endpoint: "not-an-ip"},
+		{name: "IPv6 endpoint for IPv4 egress", endpoint: "fd00::1"},
+		{name: "IPv4 endpoint for IPv6 egress", endpoint: "10.0.0.1", egressIPv6: true},
+		{name: "IPv4 endpoint", endpoint: "10.0.0.2", want: annotationUpdate{endpoint: "10.0.0.2", endpointIPv6: "fd00::1"}, wantUpdate: true},
+		{name: "IPv6 endpoint", endpoint: "fd00::2", egressIPv6: true, want: annotationUpdate{endpoint: "10.0.0.1", endpointIPv6: "fd00::2"}, wantUpdate: true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			annotations := map[string]string{
+				kubevip.Egress:             "true",
+				kubevip.ActiveEndpoint:     "10.0.0.1",
+				kubevip.ActiveEndpointIPv6: "fd00::1",
+			}
+			if test.egressIPv6 {
+				annotations[kubevip.EgressIPv6] = "true"
+			}
+			service := &v1.Service{ObjectMeta: metav1.ObjectMeta{
+				Name: "test-service", Namespace: "default", Annotations: annotations,
+			}}
+			recorder := &recordingProvider{Provider: providers.NewEndpointslices()}
+			processor := &Processor{
+				config:   &kubevip.Config{EnableEndpoints: false},
+				provider: recorder,
+			}
+
+			processor.updateAnnotations(service, &test.endpoint, nil, nil)
+
+			if !test.wantUpdate {
+				if len(recorder.updates) != 0 {
+					t.Fatalf("annotation updates = %+v, want none", recorder.updates)
+				}
+				return
+			}
+			if len(recorder.updates) != 1 || recorder.updates[0] != test.want {
+				t.Fatalf("annotation updates = %+v, want [%+v]", recorder.updates, test.want)
+			}
+		})
+	}
+}
+
 func (f *fakeWorker) processInstance(_ *servicecontext.Context, _ *v1.Service) error {
 	f.processCalled = true
 	return nil

pkg/services/services.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/kube-vip/kube-vip/pkg/egress"
 	"github.com/kube-vip/kube-vip/pkg/endpoints"
-	"github.com/kube-vip/kube-vip/pkg/endpoints/providers"
 	"github.com/kube-vip/kube-vip/pkg/instance"
 	"github.com/kube-vip/kube-vip/pkg/kubevip"
 	"github.com/kube-vip/kube-vip/pkg/lease"
@@ -376,17 +375,6 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc
 					return err
 				}
 			}
-
-			var provider providers.Provider
-			if p.config.EnableEndpoints {
-				provider = providers.NewEndpoints()
-			} else {
-				provider = providers.NewEndpointslices()
-			}
-			err := provider.UpdateServiceAnnotation(ctx, svc.Annotations[kubevip.ActiveEndpoint], svc.Annotations[kubevip.ActiveEndpointIPv6], svc, p.clientSet)
-			if err != nil {
-				log.Warn("[service] configuring egress", "service", svc.Name, "namespace", svc.Namespace, "err", err)
-			}
 		}
 	}
 

pkg/services/services_test.go

@@ -1,14 +1,90 @@
 package services
 
 import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"sync"
 	"testing"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 
 	"github.com/kube-vip/kube-vip/pkg/instance"
+	"github.com/kube-vip/kube-vip/pkg/kubevip"
 )
 
+func TestConfigureServiceDoesNotOverwriteActiveEndpoint(t *testing.T) {
+	const selectedEndpoint = "172.30.2.40"
+
+	staleService := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-service", Namespace: "default", UID: "test-uid",
+			Annotations: map[string]string{kubevip.Egress: "true", kubevip.ActiveEndpoint: ""},
+		},
+		Spec: v1.ServiceSpec{LoadBalancerIP: "10.114.44.149"},
+	}
+	currentService := staleService.DeepCopy()
+	currentService.Annotations[kubevip.ActiveEndpoint] = selectedEndpoint
+	currentService.ResourceVersion = "2"
+
+	var mutex sync.Mutex
+	updateRequests := 0
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		mutex.Lock()
+		defer mutex.Unlock()
+		writer.Header().Set("Content-Type", "application/json")
+		switch request.Method {
+		case http.MethodGet:
+			if err := json.NewEncoder(writer).Encode(currentService); err != nil {
+				t.Errorf("encode Service response: %v", err)
+			}
+		case http.MethodPut:
+			updatedService := &v1.Service{}
+			if err := json.NewDecoder(request.Body).Decode(updatedService); err != nil {
+				http.Error(writer, err.Error(), http.StatusBadRequest)
+				return
+			}
+			updateRequests++
+			currentService = updatedService
+			if err := json.NewEncoder(writer).Encode(currentService); err != nil {
+				t.Errorf("encode updated Service response: %v", err)
+			}
+		default:
+			http.Error(writer, "unexpected request", http.StatusMethodNotAllowed)
+		}
+	}))
+	defer server.Close()
+
+	clientSet, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL})
+	if err != nil {
+		t.Fatalf("create Kubernetes client: %v", err)
+	}
+	processor := &Processor{
+		config: &kubevip.Config{
+			DisableServiceUpdates:  true,
+			EnableServicesElection: true,
+		},
+		clientSet: clientSet,
+	}
+	serviceInstance := &instance.Instance{ServiceSnapshot: staleService}
+	if err := processor.configureService(context.Background(), serviceInstance, staleService, &sync.WaitGroup{}); err != nil {
+		t.Fatalf("configureService returned error: %v", err)
+	}
+
+	mutex.Lock()
+	defer mutex.Unlock()
+	if updateRequests != 0 {
+		t.Fatalf("configureService sent %d stale Service updates, want none", updateRequests)
+	}
+	if got := currentService.Annotations[kubevip.ActiveEndpoint]; got != selectedEndpoint {
+		t.Fatalf("active endpoint = %q, want %q", got, selectedEndpoint)
+	}
+}
+
 // Test_upnpLeaseDurationForService tests whether the default lease duration is used, and whether the annotation
 // overrides it correctly.
 //