diff --git a/pkg/cloudscale_ccm/loadbalancer.go b/pkg/cloudscale_ccm/loadbalancer.go index b13d405..961833c 100644 --- a/pkg/cloudscale_ccm/loadbalancer.go +++ b/pkg/cloudscale_ccm/loadbalancer.go @@ -5,11 +5,13 @@ import ( "fmt" "slices" "strings" + "sync" "github.com/cloudscale-ch/cloudscale-go-sdk/v6" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/record" "k8s.io/klog/v2" @@ -308,6 +310,19 @@ type loadbalancer struct { srv serverMapper k8s kubernetes.Interface recorder record.EventRecorder + muMap sync.Map +} + +func (l *loadbalancer) lockForService(uid types.UID) func() { + rawMu, _ := l.muMap.LoadOrStore(string(uid), new(sync.Mutex)) + mu := rawMu.(*sync.Mutex) + klog.V(4).InfoS("acquiring service lock", "uid", uid) + mu.Lock() + + return func() { + klog.V(4).InfoS("releasing service lock", "uid", uid) + mu.Unlock() + } } // GetLoadBalancer returns whether the specified load balancer exists, and @@ -391,6 +406,8 @@ func (l *loadbalancer) EnsureLoadBalancer( service *v1.Service, nodes []*v1.Node, ) (*v1.LoadBalancerStatus, error) { + unlock := l.lockForService(service.UID) + defer unlock() // Detect configuration issues and abort if they are found serviceInfo := newServiceInfo(service, clusterName) @@ -497,6 +514,8 @@ func (l *loadbalancer) UpdateLoadBalancer( service *v1.Service, nodes []*v1.Node, ) error { + unlock := l.lockForService(service.UID) + defer unlock() // Detect configuration issues and abort if they are found serviceInfo := newServiceInfo(service, clusterName) @@ -556,6 +575,8 @@ func (l *loadbalancer) EnsureLoadBalancerDeleted( clusterName string, service *v1.Service, ) error { + unlock := l.lockForService(service.UID) + defer unlock() // Detect configuration issues and abort if they are found serviceInfo := newServiceInfo(service, clusterName) diff --git a/pkg/cloudscale_ccm/loadbalancer_test.go b/pkg/cloudscale_ccm/loadbalancer_test.go index 74043b0..f7aee3f 100644 --- a/pkg/cloudscale_ccm/loadbalancer_test.go +++ b/pkg/cloudscale_ccm/loadbalancer_test.go @@ -1,7 +1,11 @@ package cloudscale_ccm import ( + "encoding/json" + "net/http" + "sync" "testing" + "time" "github.com/cloudscale-ch/cloudscale-go-sdk/v6" "github.com/stretchr/testify/assert" @@ -179,6 +183,130 @@ func TestLoadBalancer_EnsureLoadBalancer(t *testing.T) { } } +func TestLoadBalancer_ConcurrentCreate(t *testing.T) { + t.Parallel() + + apiServer := testkit.NewMockAPIServer() + + createCount := 0 + var lbs []cloudscale.LoadBalancer + var mu sync.Mutex + + // Custom handler for /v1/load-balancers to track creates. + // The sleep before appending to lbs increases the race window so that + // both goroutines can see an empty list before either creates. + apiServer.HandleFunc("/v1/load-balancers", func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodPost: + time.Sleep(200 * time.Millisecond) + + mu.Lock() + createCount++ + lb := cloudscale.LoadBalancer{ + HREF: "/v1/load-balancers/lb-uuid-1", + UUID: "lb-uuid-1", + Name: "k8s-service-test-uid", + Status: "running", + ZonalResource: cloudscale.ZonalResource{ + Zone: cloudscale.Zone{Slug: "rma1"}, + }, + Flavor: cloudscale.LoadBalancerFlavorStub{Slug: "lb-standard"}, + } + lbs = append(lbs, lb) + mu.Unlock() + + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(lb) + case http.MethodGet: + mu.Lock() + defer mu.Unlock() + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(lbs) + } + }) + + // Mock server endpoint for node mapping. + serverUUID := "08d56bfe-40d0-4c68-a915-54f846c28c9e" + apiServer.WithServers([]cloudscale.Server{{ + UUID: serverUUID, + Name: "node-1", + ZonalResource: cloudscale.ZonalResource{ + Zone: cloudscale.Zone{Slug: "rma1"}, + }, + Interfaces: []cloudscale.Interface{{ + Type: "private", + Addresses: []cloudscale.Address{{ + Address: "10.0.0.1", + Subnet: cloudscale.SubnetStub{UUID: "subnet-uuid-1"}, + }}, + }}, + }}) + + // Mock the remaining LB endpoints so reconciliation can proceed. + apiServer.On("/v1/load-balancers/pools", 200, []cloudscale.LoadBalancerPool{}) + apiServer.On("/v1/load-balancers/listeners", 200, []cloudscale.LoadBalancerListener{}) + apiServer.On("/v1/load-balancers/health-monitors", 200, []cloudscale.LoadBalancerHealthMonitor{}) + apiServer.On("/v1/floating-ips", 200, []cloudscale.FloatingIP{}) + + apiServer.Start() + defer apiServer.Close() + + client := fake.NewSimpleClientset() + fakeDiscovery, ok := client.Discovery().(*fakediscovery.FakeDiscovery) + require.True(t, ok, "couldn't convert Discovery() to *FakeDiscovery") + fakeDiscovery.FakedServerVersion = &version.Info{ + Major: "1", + Minor: "34", + } + + l := &loadbalancer{ + lbs: lbMapper{client: apiServer.Client()}, + srv: serverMapper{client: apiServer.Client()}, + k8s: client, + recorder: record.NewFakeRecorder(10), + } + + service := &v1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-service", + Namespace: "default", + UID: "test-uid", + Annotations: map[string]string{ + LoadBalancerName: "k8s-service-test-uid", + LoadBalancerFlavor: "lb-standard", + LoadBalancerZone: "rma1", + }, + }, + Spec: v1.ServiceSpec{ + Type: v1.ServiceTypeLoadBalancer, + Ports: []v1.ServicePort{ + {Protocol: v1.ProtocolTCP, Port: 80, NodePort: 80}, + }, + }, + } + + _, _ = l.k8s.CoreV1().Services("default").Create(t.Context(), service, metav1.CreateOptions{}) + + nodes := []*v1.Node{{ + ObjectMeta: metav1.ObjectMeta{Name: "node-1"}, + Spec: v1.NodeSpec{ProviderID: "cloudscale://" + serverUUID}, + }} + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _, _ = l.EnsureLoadBalancer(t.Context(), "test-cluster", service, nodes) + }() + go func() { + defer wg.Done() + _, _ = l.EnsureLoadBalancer(t.Context(), "test-cluster", service, nodes) + }() + wg.Wait() + + assert.Equal(t, 1, createCount, "expected exactly one LB creation") +} + func TestFilterNodesBySelector(t *testing.T) { t.Parallel() diff --git a/pkg/internal/testkit/api.go b/pkg/internal/testkit/api.go index 9b37821..635715e 100644 --- a/pkg/internal/testkit/api.go +++ b/pkg/internal/testkit/api.go @@ -138,6 +138,16 @@ func (m *MockAPIServer) Start() { m.server = httptest.NewServer(m.mux) } +// HandleFunc registers a custom handler for the given pattern. +// This allows intercepting requests dynamically, e.g. to track invocations. +func (m *MockAPIServer) HandleFunc(pattern string, handler http.HandlerFunc) { + if m.mux == nil { + m.mux = http.NewServeMux() + m.On("/", 404, "{}") + } + m.mux.HandleFunc(pattern, handler) +} + // Close stops/closes the server and resets it. func (m *MockAPIServer) Close() { if m.server != nil {