Skip to content

Add per-service mutex to prevent concurrent reconciliation #57

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mweibel wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add per-service mutex to prevent concurrent reconciliation #57

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions pkg/cloudscale_ccm/loadbalancer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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()
}
Comment on lines +316 to +325

@alakae alakae Aug 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I compared it with: https://github.com/cloudscale-ch/terraform-provider-cloudscale/blob/master/cloudscale/mutex_kv.go

  • any reason why you did not use the channel pattern you suggested in that repo?
  • I just now realized that both implementations do never actually delete map entries. In Terraform this should not be a problem, as the whole process is short-lived, in the CCM the process could live for weeks/months. Do you have an idea how we can address it? I don't, at least not a trivial one :)

@mweibel mweibel Sep 1, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

any reason why you did not use the channel pattern you suggested in that repo?

the mutex_kv.go is a slightly different case: it used a loop with timer for waiting until the lock is there. It also supports cancellation with context to support terraform-native timeouts (and ctrl+c). It used TryLock for this and we changed it to a channel-based approach to get rid of the loop and timer and instead use "native" Go polling.

This case here is different and much simpler: we just acquire the lock and nobody else waits on it. We don't really need context cancellation here since it's running as a service. While it may happen that the parent context could be cancelled for some reason, the lock/unlock should not be the point where this gets respected (and hasn't been so far).

I just now realized that both implementations do never actually delete map entries. In Terraform this should not be a problem, as the whole process is short-lived, in the CCM the process could live for weeks/months. Do you have an idea how we can address it? I don't, at least not a trivial one :)

That's true and I tried to address this in the commit message:

Locks are not cleaned up on service deletion to avoid issues with late-arriving goroutines.

How often this happens is questionable but it avoids potential issues without much cost: Even large clusters won't have more than a couple loadbalancers. Keeping them around would be a couple 100 bytes (one entry is ~150B). Of course if somebody creates and deletes services 1000s of times it eventually could crash the CCM, but that would rather speak for a misuse of the system than anything else TBH.

Still, We could remove the entry for a service in two ways:

  1. once EnsureLoadBalancerDeleted is successfully done
  2. Add a goroutine which periodically removes unused mutexes. This would need to track which services still exist.

point 1 is rather trivial, I just wonder if there are any edge cases. I don't think so, but I didn't add this change because the benefit is not 100% clear to me.

Let me know what you think - happy to add the deletion logic if you feel it's valuable.

@mweibel mweibel Sep 1, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As an aside: we could also use a plain map with an accompanying single mutex in this case. I think either approach would work well.

}

// GetLoadBalancer returns whether the specified load balancer exists, and
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
128 changes: 128 additions & 0 deletions pkg/cloudscale_ccm/loadbalancer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"
Expand Down Expand Up @@ -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()

Expand Down
10 changes: 10 additions & 0 deletions pkg/internal/testkit/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down