From 7609eb645e8cc186e919f6f228b40828881ecde2 Mon Sep 17 00:00:00 2001 From: ravilock Date: Thu, 5 Feb 2026 11:57:34 -0300 Subject: [PATCH 1/8] test: add test for concurrency evaluation --- Makefile | 2 +- pkg/build/buildkit/autodiscovery/k8s_test.go | 242 +++++++++++++++++++ pkg/build/buildkit/autodiscovery/leaser.go | 20 +- 3 files changed, 257 insertions(+), 7 deletions(-) diff --git a/Makefile b/Makefile index a4fd8bd..f830228 100644 --- a/Makefile +++ b/Makefile @@ -50,4 +50,4 @@ generate: .PHONY: build/container-image build/container-image: - $(DOCKER) build -t tsuru/deploy-agent-local:latest ./ + $(DOCKER) build -t 100.64.100.100:5000/tsuru/deploy-agent:dev ./ diff --git a/pkg/build/buildkit/autodiscovery/k8s_test.go b/pkg/build/buildkit/autodiscovery/k8s_test.go index 6e18af6..21e5d3f 100644 --- a/pkg/build/buildkit/autodiscovery/k8s_test.go +++ b/pkg/build/buildkit/autodiscovery/k8s_test.go @@ -8,15 +8,21 @@ import ( "bytes" "context" "encoding/json" + "errors" "os" + "sync" "testing" "time" + k8sErrors "k8s.io/apimachinery/pkg/api/errors" + + "github.com/moby/buildkit/client" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/tsuru/deploy-agent/pkg/build/grpc_build_v1" "github.com/tsuru/deploy-agent/pkg/build/metadata" appsv1 "k8s.io/api/apps/v1" + coordinationv1 "k8s.io/api/coordination/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -25,10 +31,106 @@ import ( "k8s.io/apimachinery/pkg/watch" fakeDynamic "k8s.io/client-go/dynamic/fake" "k8s.io/client-go/kubernetes/fake" + clientTesting "k8s.io/client-go/testing" kuberntesTesting "k8s.io/client-go/testing" "k8s.io/utils/ptr" ) +type leaseReactor struct { + leases []*coordinationv1.Lease + lock sync.Mutex +} + +func newLeaseReactor() *leaseReactor { + return &leaseReactor{ + leases: make([]*coordinationv1.Lease, 0), + } +} + +func (l *leaseReactor) Handles(action clientTesting.Action) bool { + return action.GetResource().Resource == "leases" +} + +func (l *leaseReactor) React(action clientTesting.Action) (handled bool, ret runtime.Object, err error) { + l.lock.Lock() + defer l.lock.Unlock() + switch action.GetVerb() { + case "list": + if _, ok := action.(clientTesting.ListAction); ok { + leaseList := &coordinationv1.LeaseList{ + Items: []coordinationv1.Lease{}, + } + for _, lease := range l.leases { + leaseList.Items = append(leaseList.Items, *lease.DeepCopy()) + } + return true, leaseList, nil + } + return false, nil, nil + case "create": + if createAction, ok := action.(clientTesting.CreateAction); ok { + lease, err := l.createLease(createAction) + return true, lease, err + } + return false, nil, nil + case "get": + if getAction, ok := action.(clientTesting.GetAction); ok { + lease, err := l.getLease(getAction) + return true, lease, err + } + return false, nil, nil + case "update": + if updateAction, ok := action.(clientTesting.UpdateAction); ok { + lease, err := l.updateLease(updateAction) + return true, lease, err + } + return false, nil, nil + case "delete": + panic("should not be called") + } + return false, nil, nil +} + +func (l *leaseReactor) getLease(getAction clientTesting.GetAction) (ret runtime.Object, err error) { + for _, lease := range l.leases { + if lease.Name == getAction.GetName() && lease.Namespace == getAction.GetNamespace() { + return lease.DeepCopy(), nil + } + } + return nil, k8sErrors.NewNotFound(coordinationv1.Resource("leases"), getAction.GetName()) +} + +func (l *leaseReactor) createLease(createAction clientTesting.CreateAction) (ret runtime.Object, err error) { + leaseObject, ok := createAction.GetObject().(*coordinationv1.Lease) + if !ok { + return nil, errors.New("not a lease object") + } + for _, lease := range l.leases { + if lease.Name == leaseObject.Name && lease.Namespace == leaseObject.Namespace { + return nil, k8sErrors.NewAlreadyExists(coordinationv1.Resource("leases"), leaseObject.Name) + } + } + lease := leaseObject.DeepCopy() + lease.CreationTimestamp = metav1.Now() + l.leases = append(l.leases, lease) + return leaseObject, nil +} + +func (l *leaseReactor) updateLease(updateAction clientTesting.UpdateAction) (ret runtime.Object, err error) { + updateLeaseObject, ok := updateAction.GetObject().(*coordinationv1.Lease) + if !ok { + return nil, errors.New("not a lease object") + } + for i, currentLease := range l.leases { + if currentLease.Name == updateLeaseObject.Name && currentLease.Namespace == updateLeaseObject.Namespace { + lease := updateLeaseObject.DeepCopy() + lease.CreationTimestamp = currentLease.CreationTimestamp + l.leases[i] = lease + return updateLeaseObject, nil + } + } + return nil, k8sErrors.NewNotFound(coordinationv1.Resource("leases"), updateLeaseObject.Name) +} + func TestK8sDiscoverer_Discover(t *testing.T) { buildKitPod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ @@ -250,6 +352,146 @@ func TestK8sDiscoverer_DiscoverWithStatefulsetInitialUpscale(t *testing.T) { }) } +func TestK8sDiscoverer_ConcurrencySafetyWithLeases(t *testing.T) { + // We only keep one buildkit pod to ensure both discoverers are trying to acquire the same lease + buildkitPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "buildkit-0", + Namespace: "tsuru", + Labels: map[string]string{ + "app": "buildkit", + }, + }, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + PodIP: "127.0.0.1", + Conditions: []corev1.PodCondition{ + { + Type: corev1.PodReady, + Status: corev1.ConditionTrue, + }, + }, + }, + } + statefulset := &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "buildkit", + Namespace: "tsuru", + }, + Spec: appsv1.StatefulSetSpec{ + Replicas: ptr.To(int32(1)), + }, + } + + kubeClient := fake.NewSimpleClientset(buildkitPod, statefulset) + kubeClient.PrependWatchReactor("*", func(action kuberntesTesting.Action) (handled bool, ret watch.Interface, err error) { + watcher := watch.NewFake() + go func() { + time.Sleep(time.Millisecond * 100) + watcher.Add(buildkitPod) + }() + return true, watcher, nil + }) + leaseReactor := newLeaseReactor() + kubeClient.ReactionChain = append([]clientTesting.Reactor{leaseReactor}, kubeClient.ReactionChain...) + + dynamicClient := fakeDynamic.NewSimpleDynamicClient(runtime.NewScheme()) + + var client1 *client.Client + var cleanups1 func() + var buf1 bytes.Buffer + doneCh1 := make(chan struct{}) + discoverer1 := K8sDiscoverer{ + KubernetesInterface: kubeClient, + DynamicInterface: dynamicClient, + } + req1 := &grpc_build_v1.BuildRequest{ + App: &grpc_build_v1.TsuruApp{ + Name: "test-app-1", + }, + } + var client2 *client.Client + var cleanups2 func() + var buf2 bytes.Buffer + doneCh2 := make(chan struct{}) + discoverer2 := K8sDiscoverer{ + KubernetesInterface: kubeClient, + DynamicInterface: dynamicClient, + } + req2 := &grpc_build_v1.BuildRequest{ + App: &grpc_build_v1.TsuruApp{ + Name: "test-app-2", + }, + } + discoveryOptions := KubernertesDiscoveryOptions{ + PodSelector: "app=buildkit", + Namespace: "tsuru", + Timeout: time.Minute * 2, + Statefulset: "buildkit", + SetTsuruAppLabel: false, + } + + go func() { + client, cleanups, ns, err := discoverer1.Discover(context.Background(), discoveryOptions, req1, &buf1) + require.Equal(t, buildkitPod.Namespace, ns) + require.NoError(t, err) + require.NotNil(t, cleanups) + client1 = client + cleanups1 = cleanups + doneCh1 <- struct{}{} + }() + go func() { + client, cleanups, ns, err := discoverer2.Discover(context.Background(), discoveryOptions, req2, &buf2) + require.Equal(t, buildkitPod.Namespace, ns) + require.NoError(t, err) + require.NotNil(t, cleanups) + client2 = client + cleanups2 = cleanups + doneCh2 <- struct{}{} + }() + + var firstAquiredClient *client.Client + var firstAquiredCleanups func() + select { + case <-doneCh1: + firstAquiredClient = client1 + firstAquiredCleanups = cleanups1 + case <-doneCh2: + firstAquiredClient = client2 + firstAquiredCleanups = cleanups2 + } + require.NotNil(t, firstAquiredClient) + time.Sleep(leaseDuration) // wait a bit to increase chances of the second discoverer trying to acquire the lease while the first one has it + + leaseList, err := kubeClient.CoordinationV1().Leases("tsuru").List(context.TODO(), metav1.ListOptions{}) + require.NoError(t, err) + require.Len(t, leaseList.Items, 1, "there should be only one lease created") + firstAquiredLease := leaseList.Items[0].DeepCopy() + firstAquiredCleanups() + require.NotZero(t, *firstAquiredLease.Spec.HolderIdentity, "the first discoverer should acquire the lease and set the holder identity") + + time.Sleep(leaseDuration) // wait a bit to ensure the first discoverer has released the lease and the second one has a chance to acquire it + + leaseList, err = kubeClient.CoordinationV1().Leases("tsuru").List(context.TODO(), metav1.ListOptions{}) + require.NoError(t, err) + require.Len(t, leaseList.Items, 1, "there should be only one lease created") + secondAquiredLease := leaseList.Items[0].DeepCopy() + require.NotZero(t, *secondAquiredLease.Spec.HolderIdentity, "the second discoverer should acquire the lease and set the holder identity") + require.NotEqual(t, *firstAquiredLease.Spec.HolderIdentity, *secondAquiredLease.Spec.HolderIdentity, "the second discoverer should acquire a different lease after the first one releases it") + if firstAquiredClient == client1 { + cleanups2() + } else { + cleanups1() + } + + time.Sleep(leaseDuration) // wait a bit to ensure the cleanup has released the lease + leaseList, err = kubeClient.CoordinationV1().Leases("tsuru").List(context.TODO(), metav1.ListOptions{}) + require.NoError(t, err) + require.Len(t, leaseList.Items, 1) + notHeldLease := leaseList.Items[0] + require.Zero(t, *notHeldLease.Spec.HolderIdentity) +} + func TestK8sDiscoverer_BuildkitPodNamespace(t *testing.T) { t.Run("use provided namespace when UseSameNamespaceAsApp is false", func(t *testing.T) { opts := KubernertesDiscoveryOptions{ diff --git a/pkg/build/buildkit/autodiscovery/leaser.go b/pkg/build/buildkit/autodiscovery/leaser.go index 217c51d..2223c55 100644 --- a/pkg/build/buildkit/autodiscovery/leaser.go +++ b/pkg/build/buildkit/autodiscovery/leaser.go @@ -20,6 +20,12 @@ import ( "k8s.io/klog" ) +var ( + leaseDuration = 5 * time.Second + renewDeadline = 2 * time.Second + retryPeriod = 500 * time.Millisecond +) + type leaser struct { kubernetesInterface kubernetes.Interface leasablePodsCh <-chan *corev1.Pod @@ -38,6 +44,7 @@ func newLeaser(kubernetesInterface kubernetes.Interface, leasablePodsCh <-chan * } holderName = hostname } + holderName = fmt.Sprintf("%s-%d", holderName, time.Now().UnixNano()) leasedPodsCh := make(chan *corev1.Pod, 1) return &leaser{ kubernetesInterface: kubernetesInterface, @@ -98,8 +105,8 @@ func (l *leaser) acquireLeaseForAllPods(ctx context.Context, opts KubernertesDis // it is a blocking call and only returns after the lease is lost or the given context is canceled. // it should always be used in a separate goroutine. func (l *leaser) acquireLeaseForPod(ctx context.Context, pod *corev1.Pod, opts KubernertesDiscoveryOptions) { - uniqueHolderName := fmt.Sprintf("%s-%d", l.holderName, time.Now().Unix()) - klog.V(4).Infof("Attempting to acquire the lease for pod %s/%s under holder name %q...", pod.Namespace, pod.Name, uniqueHolderName) + fmt.Printf("Attempting to acquire the lease for pod %s/%s under holder name %q...\n", pod.Namespace, pod.Name, l.holderName) + klog.V(4).Infof("Attempting to acquire the lease for pod %s/%s under holder name %q...", pod.Namespace, pod.Name, l.holderName) leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{ Lock: &resourcelock.LeaseLock{ LeaseMeta: metav1.ObjectMeta{ @@ -108,17 +115,18 @@ func (l *leaser) acquireLeaseForPod(ctx context.Context, pod *corev1.Pod, opts K }, Client: l.kubernetesInterface.CoordinationV1(), LockConfig: resourcelock.ResourceLockConfig{ - Identity: uniqueHolderName, + Identity: l.holderName, }, }, ReleaseOnCancel: true, - LeaseDuration: 5 * time.Second, - RenewDeadline: 2 * time.Second, - RetryPeriod: 500 * time.Millisecond, + LeaseDuration: leaseDuration, + RenewDeadline: renewDeadline, + RetryPeriod: retryPeriod, Callbacks: leaderelection.LeaderCallbacks{ OnStartedLeading: func(_ context.Context) { select { case l.leasedPodsCh <- pod: + fmt.Printf("Selected BuildKit pod: %s/%s under holder name %q\n", pod.Namespace, pod.Name, l.holderName) klog.V(4).Infof("Selected BuildKit pod: %s/%s", pod.Namespace, pod.Name) case <-ctx.Done(): From aeb05daac7e6b7afef28f24c0c30e07c33f92c8d Mon Sep 17 00:00:00 2001 From: ravilock Date: Thu, 5 Feb 2026 12:03:56 -0300 Subject: [PATCH 2/8] fix: remove deferred early cancellation context --- pkg/build/buildkit/autodiscovery/k8s.go | 7 ++----- pkg/build/buildkit/autodiscovery/k8s_test.go | 2 +- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/pkg/build/buildkit/autodiscovery/k8s.go b/pkg/build/buildkit/autodiscovery/k8s.go index 220b8d8..47c4265 100644 --- a/pkg/build/buildkit/autodiscovery/k8s.go +++ b/pkg/build/buildkit/autodiscovery/k8s.go @@ -121,9 +121,6 @@ func (d *K8sDiscoverer) discoverBuildKitClientFromApp(ctx context.Context, opts } func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts KubernertesDiscoveryOptions, namespace string, w io.Writer) (*corev1.Pod, error) { - deadlineCtx, deadlineCancel := context.WithCancel(ctx) - defer deadlineCancel() - metrics.BuildsWaitingForLease.WithLabelValues(namespace).Inc() defer metrics.BuildsWaitingForLease.WithLabelValues(namespace).Dec() @@ -134,7 +131,7 @@ func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts Kubernerte } } - watchCtx, watchCancel := context.WithCancel(deadlineCtx) + watchCtx, watchCancel := context.WithCancel(ctx) defer watchCancel() podWatcher, err := d.KubernetesInterface.CoreV1().Pods(namespace).Watch(watchCtx, metav1.ListOptions{ @@ -152,7 +149,7 @@ func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts Kubernerte if err != nil { return nil, fmt.Errorf("failed to create pod leaser: %w", err) } - go leaser.acquireLeaseForAllPods(deadlineCtx, opts) + go leaser.acquireLeaseForAllPods(ctx, opts) for { select { diff --git a/pkg/build/buildkit/autodiscovery/k8s_test.go b/pkg/build/buildkit/autodiscovery/k8s_test.go index 21e5d3f..f2c9d8c 100644 --- a/pkg/build/buildkit/autodiscovery/k8s_test.go +++ b/pkg/build/buildkit/autodiscovery/k8s_test.go @@ -488,7 +488,7 @@ func TestK8sDiscoverer_ConcurrencySafetyWithLeases(t *testing.T) { leaseList, err = kubeClient.CoordinationV1().Leases("tsuru").List(context.TODO(), metav1.ListOptions{}) require.NoError(t, err) require.Len(t, leaseList.Items, 1) - notHeldLease := leaseList.Items[0] + notHeldLease := leaseList.Items[0].DeepCopy() require.Zero(t, *notHeldLease.Spec.HolderIdentity) } From fc4ea30bb33ccfba7bcc5e09bd381e1694804a9a Mon Sep 17 00:00:00 2001 From: ravilock Date: Thu, 5 Feb 2026 12:16:46 -0300 Subject: [PATCH 3/8] test: fix race condition --- pkg/build/buildkit/autodiscovery/k8s_test.go | 30 +++++++++----------- 1 file changed, 14 insertions(+), 16 deletions(-) diff --git a/pkg/build/buildkit/autodiscovery/k8s_test.go b/pkg/build/buildkit/autodiscovery/k8s_test.go index f2c9d8c..990ddf8 100644 --- a/pkg/build/buildkit/autodiscovery/k8s_test.go +++ b/pkg/build/buildkit/autodiscovery/k8s_test.go @@ -16,7 +16,6 @@ import ( k8sErrors "k8s.io/apimachinery/pkg/api/errors" - "github.com/moby/buildkit/client" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/tsuru/deploy-agent/pkg/build/grpc_build_v1" @@ -397,7 +396,6 @@ func TestK8sDiscoverer_ConcurrencySafetyWithLeases(t *testing.T) { dynamicClient := fakeDynamic.NewSimpleDynamicClient(runtime.NewScheme()) - var client1 *client.Client var cleanups1 func() var buf1 bytes.Buffer doneCh1 := make(chan struct{}) @@ -410,7 +408,6 @@ func TestK8sDiscoverer_ConcurrencySafetyWithLeases(t *testing.T) { Name: "test-app-1", }, } - var client2 *client.Client var cleanups2 func() var buf2 bytes.Buffer doneCh2 := make(chan struct{}) @@ -432,35 +429,33 @@ func TestK8sDiscoverer_ConcurrencySafetyWithLeases(t *testing.T) { } go func() { - client, cleanups, ns, err := discoverer1.Discover(context.Background(), discoveryOptions, req1, &buf1) + _, cleanups, ns, err := discoverer1.Discover(context.Background(), discoveryOptions, req1, &buf1) require.Equal(t, buildkitPod.Namespace, ns) require.NoError(t, err) require.NotNil(t, cleanups) - client1 = client cleanups1 = cleanups doneCh1 <- struct{}{} }() go func() { - client, cleanups, ns, err := discoverer2.Discover(context.Background(), discoveryOptions, req2, &buf2) + _, cleanups, ns, err := discoverer2.Discover(context.Background(), discoveryOptions, req2, &buf2) require.Equal(t, buildkitPod.Namespace, ns) require.NoError(t, err) require.NotNil(t, cleanups) - client2 = client cleanups2 = cleanups doneCh2 <- struct{}{} }() - var firstAquiredClient *client.Client + var firstAquiredClient string var firstAquiredCleanups func() + var secondAquiredCleanups func() select { case <-doneCh1: - firstAquiredClient = client1 + firstAquiredClient = "client-1" firstAquiredCleanups = cleanups1 case <-doneCh2: - firstAquiredClient = client2 + firstAquiredClient = "client-2" firstAquiredCleanups = cleanups2 } - require.NotNil(t, firstAquiredClient) time.Sleep(leaseDuration) // wait a bit to increase chances of the second discoverer trying to acquire the lease while the first one has it leaseList, err := kubeClient.CoordinationV1().Leases("tsuru").List(context.TODO(), metav1.ListOptions{}) @@ -471,6 +466,13 @@ func TestK8sDiscoverer_ConcurrencySafetyWithLeases(t *testing.T) { require.NotZero(t, *firstAquiredLease.Spec.HolderIdentity, "the first discoverer should acquire the lease and set the holder identity") time.Sleep(leaseDuration) // wait a bit to ensure the first discoverer has released the lease and the second one has a chance to acquire it + if firstAquiredClient == "client-1" { + <-doneCh2 + secondAquiredCleanups = cleanups2 + } else { + <-doneCh1 + secondAquiredCleanups = cleanups1 + } leaseList, err = kubeClient.CoordinationV1().Leases("tsuru").List(context.TODO(), metav1.ListOptions{}) require.NoError(t, err) @@ -478,11 +480,7 @@ func TestK8sDiscoverer_ConcurrencySafetyWithLeases(t *testing.T) { secondAquiredLease := leaseList.Items[0].DeepCopy() require.NotZero(t, *secondAquiredLease.Spec.HolderIdentity, "the second discoverer should acquire the lease and set the holder identity") require.NotEqual(t, *firstAquiredLease.Spec.HolderIdentity, *secondAquiredLease.Spec.HolderIdentity, "the second discoverer should acquire a different lease after the first one releases it") - if firstAquiredClient == client1 { - cleanups2() - } else { - cleanups1() - } + secondAquiredCleanups() time.Sleep(leaseDuration) // wait a bit to ensure the cleanup has released the lease leaseList, err = kubeClient.CoordinationV1().Leases("tsuru").List(context.TODO(), metav1.ListOptions{}) From 95d013292e6dd02456c770261c95875fede4d4b7 Mon Sep 17 00:00:00 2001 From: ravilock Date: Thu, 5 Feb 2026 16:56:02 -0300 Subject: [PATCH 4/8] refactor: set holdername on notifier and leaser for better debug --- Makefile | 2 +- pkg/build/buildkit/autodiscovery/k8s.go | 71 ++++++++++++------- pkg/build/buildkit/autodiscovery/leaser.go | 14 +--- .../buildkit/autodiscovery/podNotifier.go | 7 +- 4 files changed, 51 insertions(+), 43 deletions(-) diff --git a/Makefile b/Makefile index f830228..a4fd8bd 100644 --- a/Makefile +++ b/Makefile @@ -50,4 +50,4 @@ generate: .PHONY: build/container-image build/container-image: - $(DOCKER) build -t 100.64.100.100:5000/tsuru/deploy-agent:dev ./ + $(DOCKER) build -t tsuru/deploy-agent-local:latest ./ diff --git a/pkg/build/buildkit/autodiscovery/k8s.go b/pkg/build/buildkit/autodiscovery/k8s.go index 47c4265..05ab206 100644 --- a/pkg/build/buildkit/autodiscovery/k8s.go +++ b/pkg/build/buildkit/autodiscovery/k8s.go @@ -9,6 +9,7 @@ import ( "encoding/json" "fmt" "io" + "os" "strconv" "strings" "time" @@ -120,6 +121,33 @@ func (d *K8sDiscoverer) discoverBuildKitClientFromApp(ctx context.Context, opts return c, cleanUps(cfns...), nil } +func (d *K8sDiscoverer) buildkitPodNamespace(ctx context.Context, opts KubernertesDiscoveryOptions, app string) (string, error) { + if !opts.UseSameNamespaceAsApp { + return opts.Namespace, nil + } + + klog.V(4).Infof("Discovering the namespace where app %s is running on...", app) + + tsuruApp, err := d.DynamicInterface.Resource(tsuruAppGVR).Namespace(metadata.TsuruAppNamespace).Get(ctx, app, metav1.GetOptions{}) + if err != nil { + return "", err + } + + // See more about App resource at: https://github.com/tsuru/tsuru/blob/main/provision/kubernetes/pkg/apis/tsuru/v1/types.go#L24 + ns, found, err := unstructured.NestedString(tsuruApp.Object, "spec", "namespaceName") + if err != nil { + return "", err + } + + if !found { + return "", fmt.Errorf("failed to fetch namespace in the App resource") + } + + klog.V(4).Infof("App %s is running on namespace %s...", app, ns) + + return ns, nil +} + func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts KubernertesDiscoveryOptions, namespace string, w io.Writer) (*corev1.Pod, error) { metrics.BuildsWaitingForLease.WithLabelValues(namespace).Inc() defer metrics.BuildsWaitingForLease.WithLabelValues(namespace).Dec() @@ -142,10 +170,15 @@ func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts Kubernerte return nil, fmt.Errorf("failed to create pod watcher: %w", err) } - notifier, leasablePodsCh := newPodNotifier(podWatcher) + holderName, err := getHolderName() + if err != nil { + return nil, fmt.Errorf("failed to get lease holder name: %w", err) + } + + notifier, leasablePodsCh := newPodNotifier(podWatcher, holderName) go notifier.notify(watchCtx, isPodReady) - leaser, leasedPodsCh, err := newLeaser(d.KubernetesInterface, leasablePodsCh) + leaser, leasedPodsCh, err := newLeaser(d.KubernetesInterface, leasablePodsCh, holderName) if err != nil { return nil, fmt.Errorf("failed to create pod leaser: %w", err) } @@ -167,31 +200,17 @@ func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts Kubernerte } } -func (d *K8sDiscoverer) buildkitPodNamespace(ctx context.Context, opts KubernertesDiscoveryOptions, app string) (string, error) { - if !opts.UseSameNamespaceAsApp { - return opts.Namespace, nil - } - - klog.V(4).Infof("Discovering the namespace where app %s is running on...", app) - - tsuruApp, err := d.DynamicInterface.Resource(tsuruAppGVR).Namespace(metadata.TsuruAppNamespace).Get(ctx, app, metav1.GetOptions{}) - if err != nil { - return "", err - } - - // See more about App resource at: https://github.com/tsuru/tsuru/blob/main/provision/kubernetes/pkg/apis/tsuru/v1/types.go#L24 - ns, found, err := unstructured.NestedString(tsuruApp.Object, "spec", "namespaceName") - if err != nil { - return "", err - } - - if !found { - return "", fmt.Errorf("failed to fetch namespace in the App resource") +func getHolderName() (string, error) { + holderName := os.Getenv("POD_NAME") + if holderName == "" { + hostname, err := os.Hostname() + if err != nil { + return "", err + } + holderName = hostname } - - klog.V(4).Infof("App %s is running on namespace %s...", app, ns) - - return ns, nil + holderName = fmt.Sprintf("%s-%d", holderName, time.Now().UnixNano()) + return holderName, nil } func isPodReady(pod *corev1.Pod) bool { diff --git a/pkg/build/buildkit/autodiscovery/leaser.go b/pkg/build/buildkit/autodiscovery/leaser.go index 2223c55..5f16f75 100644 --- a/pkg/build/buildkit/autodiscovery/leaser.go +++ b/pkg/build/buildkit/autodiscovery/leaser.go @@ -7,7 +7,6 @@ package autodiscovery import ( "context" "fmt" - "os" "strings" "sync" "time" @@ -35,16 +34,7 @@ type leaser struct { holderName string } -func newLeaser(kubernetesInterface kubernetes.Interface, leasablePodsCh <-chan *corev1.Pod) (*leaser, <-chan *corev1.Pod, error) { - holderName := os.Getenv("POD_NAME") - if holderName == "" { - hostname, err := os.Hostname() - if err != nil { - return nil, nil, err - } - holderName = hostname - } - holderName = fmt.Sprintf("%s-%d", holderName, time.Now().UnixNano()) +func newLeaser(kubernetesInterface kubernetes.Interface, leasablePodsCh <-chan *corev1.Pod, holderName string) (*leaser, <-chan *corev1.Pod, error) { leasedPodsCh := make(chan *corev1.Pod, 1) return &leaser{ kubernetesInterface: kubernetesInterface, @@ -105,7 +95,6 @@ func (l *leaser) acquireLeaseForAllPods(ctx context.Context, opts KubernertesDis // it is a blocking call and only returns after the lease is lost or the given context is canceled. // it should always be used in a separate goroutine. func (l *leaser) acquireLeaseForPod(ctx context.Context, pod *corev1.Pod, opts KubernertesDiscoveryOptions) { - fmt.Printf("Attempting to acquire the lease for pod %s/%s under holder name %q...\n", pod.Namespace, pod.Name, l.holderName) klog.V(4).Infof("Attempting to acquire the lease for pod %s/%s under holder name %q...", pod.Namespace, pod.Name, l.holderName) leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{ Lock: &resourcelock.LeaseLock{ @@ -126,7 +115,6 @@ func (l *leaser) acquireLeaseForPod(ctx context.Context, pod *corev1.Pod, opts K OnStartedLeading: func(_ context.Context) { select { case l.leasedPodsCh <- pod: - fmt.Printf("Selected BuildKit pod: %s/%s under holder name %q\n", pod.Namespace, pod.Name, l.holderName) klog.V(4).Infof("Selected BuildKit pod: %s/%s", pod.Namespace, pod.Name) case <-ctx.Done(): diff --git a/pkg/build/buildkit/autodiscovery/podNotifier.go b/pkg/build/buildkit/autodiscovery/podNotifier.go index d08823e..6da865e 100644 --- a/pkg/build/buildkit/autodiscovery/podNotifier.go +++ b/pkg/build/buildkit/autodiscovery/podNotifier.go @@ -15,11 +15,12 @@ import ( type podNotifier struct { podWatcher watch.Interface pods chan<- *corev1.Pod + holderName string } -func newPodNotifier(podWatcher watch.Interface) (*podNotifier, <-chan *corev1.Pod) { +func newPodNotifier(podWatcher watch.Interface, holderName string) (*podNotifier, <-chan *corev1.Pod) { pods := make(chan *corev1.Pod) - return &podNotifier{podWatcher: podWatcher, pods: pods}, pods + return &podNotifier{podWatcher: podWatcher, pods: pods, holderName: holderName}, pods } type filterCondition func(pod *corev1.Pod) bool @@ -43,7 +44,7 @@ func (n *podNotifier) notify(ctx context.Context, conditions ...filterCondition) if applyConditions(pod, conditions...) { n.pods <- pod } else { - klog.V(4).Infof("Pod %s/%s is not ready yet", pod.Namespace, pod.Name) + klog.V(4).Infof("Pod %s/%s is not ready yet - holderName %q", pod.Namespace, pod.Name, n.holderName) } case <-ctx.Done(): return From ec5c7ae55753c05b1665114919b1f0e5d488839d Mon Sep 17 00:00:00 2001 From: ravilock Date: Fri, 6 Feb 2026 18:47:40 -0300 Subject: [PATCH 5/8] fix: releasing leases should take effect immediately --- pkg/build/buildkit/autodiscovery/k8s.go | 6 +++--- pkg/build/buildkit/autodiscovery/leaser.go | 1 - 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/pkg/build/buildkit/autodiscovery/k8s.go b/pkg/build/buildkit/autodiscovery/k8s.go index 05ab206..decf489 100644 --- a/pkg/build/buildkit/autodiscovery/k8s.go +++ b/pkg/build/buildkit/autodiscovery/k8s.go @@ -187,14 +187,14 @@ func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts Kubernerte for { select { case <-time.After(opts.Timeout): - go leaser.releaseAll() + leaser.releaseAll() return nil, fmt.Errorf("max deadline of %s exceeded to discover BuildKit pod", opts.Timeout) case leasedPod, ok := <-leasedPodsCh: if !ok { - go leaser.releaseAll() + leaser.releaseAll() return nil, fmt.Errorf("leased pods channel was closed before acquiring any lease") } - go leaser.releaseAll(releaseOptions{except: leasedPod.Name}) + leaser.releaseAll(releaseOptions{except: leasedPod.Name}) return leasedPod, nil } } diff --git a/pkg/build/buildkit/autodiscovery/leaser.go b/pkg/build/buildkit/autodiscovery/leaser.go index 5f16f75..0c0fad5 100644 --- a/pkg/build/buildkit/autodiscovery/leaser.go +++ b/pkg/build/buildkit/autodiscovery/leaser.go @@ -57,7 +57,6 @@ func (l *leaser) releaseAll(opts ...releaseOptions) { } else { opt = opts[0] } - l.leaseAcquiringWg.Wait() for name, leaseCancel := range l.leaseCancelByPod { if opt.except == name { continue From ca679e5f2440f67787b679057554aeae842d2bff Mon Sep 17 00:00:00 2001 From: ravilock Date: Mon, 9 Feb 2026 09:10:12 -0300 Subject: [PATCH 6/8] refactor: add holder name on logs for better debuggging --- pkg/build/buildkit/autodiscovery/leaser.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/build/buildkit/autodiscovery/leaser.go b/pkg/build/buildkit/autodiscovery/leaser.go index 0c0fad5..8b7421d 100644 --- a/pkg/build/buildkit/autodiscovery/leaser.go +++ b/pkg/build/buildkit/autodiscovery/leaser.go @@ -94,7 +94,7 @@ func (l *leaser) acquireLeaseForAllPods(ctx context.Context, opts KubernertesDis // it is a blocking call and only returns after the lease is lost or the given context is canceled. // it should always be used in a separate goroutine. func (l *leaser) acquireLeaseForPod(ctx context.Context, pod *corev1.Pod, opts KubernertesDiscoveryOptions) { - klog.V(4).Infof("Attempting to acquire the lease for pod %s/%s under holder name %q...", pod.Namespace, pod.Name, l.holderName) + klog.V(4).Infof("Attempting to acquire the lease for pod %s/%s under holder name %s", pod.Namespace, pod.Name, l.holderName) leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{ Lock: &resourcelock.LeaseLock{ LeaseMeta: metav1.ObjectMeta{ @@ -114,7 +114,7 @@ func (l *leaser) acquireLeaseForPod(ctx context.Context, pod *corev1.Pod, opts K OnStartedLeading: func(_ context.Context) { select { case l.leasedPodsCh <- pod: - klog.V(4).Infof("Selected BuildKit pod: %s/%s", pod.Namespace, pod.Name) + klog.V(4).Infof("Selected BuildKit pod: %s/%s under holder name %s", pod.Namespace, pod.Name, l.holderName) case <-ctx.Done(): klog.V(4).Infof("Received context cancellation: %s/%s", pod.Namespace, pod.Name) @@ -123,5 +123,5 @@ func (l *leaser) acquireLeaseForPod(ctx context.Context, pod *corev1.Pod, opts K OnStoppedLeading: func() {}, }, }) - klog.V(4).Infof("Shutting off the lease for %s/%s pod", pod.Namespace, pod.Name) + klog.V(4).Infof("Shutting off the lease acquirer for %s/%s pod under holder name %s", pod.Namespace, pod.Name, l.holderName) } From 5b641e9be9c36ad95aa74908bb8f27ed8bc14ce5 Mon Sep 17 00:00:00 2001 From: ravilock Date: Mon, 9 Feb 2026 10:00:10 -0300 Subject: [PATCH 7/8] test: add test for race condition on leaser --- .../buildkit/autodiscovery/leaser_test.go | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 pkg/build/buildkit/autodiscovery/leaser_test.go diff --git a/pkg/build/buildkit/autodiscovery/leaser_test.go b/pkg/build/buildkit/autodiscovery/leaser_test.go new file mode 100644 index 0000000..74de8f1 --- /dev/null +++ b/pkg/build/buildkit/autodiscovery/leaser_test.go @@ -0,0 +1,72 @@ +// Copyright 2026 tsuru authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package autodiscovery + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +// TestLeaser_ConcurrentMapAccess tests for concurrent map access race conditions. +// This test will fail with -race flag if the leaseCancelByPod map is accessed +// without proper synchronization (mutex). +// +// The race condition being tested: +// - acquireLeaseForAllPods() writes to leaseCancelByPod map (goroutine) +// - releaseAll() reads from leaseCancelByPod map (multiple goroutines) +// - Without mutex protection, concurrent write/read causes a data race +func TestLeaser_ConcurrentMapAccess(t *testing.T) { + kubeClient := fake.NewSimpleClientset() + leasablePodsCh := make(chan *corev1.Pod, 20) + holderName := "test-holder" + + leaser, _, err := newLeaser(kubeClient, leasablePodsCh, holderName) + require.NoError(t, err, "failed to create leaser") + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + // Start the goroutine that writes to the map + go leaser.acquireLeaseForAllPods(ctx, KubernertesDiscoveryOptions{ + LeasePrefix: "test-", + }) + + // Send multiple pods to trigger concurrent map writes + for i := range 20 { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "buildkit-" + string(rune('a'+i)), + Namespace: "tsuru", + }, + } + leasablePodsCh <- pod + } + + // Launch multiple goroutines that read from the map concurrently + // This creates write/read races if no mutex protection exists + concurrency := 5 + done := make(chan struct{}, concurrency) + + for g := range concurrency { + go func(id int) { + for range 20 { + leaser.releaseAll() + time.Sleep(time.Microsecond * 50) + } + done <- struct{}{} + }(g) + } + + // Wait for all reader goroutines to complete + for range concurrency { + <-done + } +} From b4051b8d24449adce1a6d62341c2719d57e4daaa Mon Sep 17 00:00:00 2001 From: ravilock Date: Mon, 9 Feb 2026 10:00:44 -0300 Subject: [PATCH 8/8] fix: race condition on leaser and add package level comment --- pkg/build/buildkit/autodiscovery/k8s.go | 5 +++++ pkg/build/buildkit/autodiscovery/leaser.go | 7 +++++++ 2 files changed, 12 insertions(+) diff --git a/pkg/build/buildkit/autodiscovery/k8s.go b/pkg/build/buildkit/autodiscovery/k8s.go index decf489..e5b392c 100644 --- a/pkg/build/buildkit/autodiscovery/k8s.go +++ b/pkg/build/buildkit/autodiscovery/k8s.go @@ -2,6 +2,11 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. +// Package autodiscovery is responsible for discovering BuildKit instances running in Kubernetes clusters, +// by watching for pods with specific labels and acquiring a lease on them to ensure exclusive access. +// It also handles setting and unsetting Tsuru app labels on the discovered BuildKit pods, +// allowing for better integration with Tsuru's app management. +// The discovery process includes a timeout mechanism to prevent indefinite waiting for a BuildKit pod to become available. package autodiscovery import ( diff --git a/pkg/build/buildkit/autodiscovery/leaser.go b/pkg/build/buildkit/autodiscovery/leaser.go index 8b7421d..7356c24 100644 --- a/pkg/build/buildkit/autodiscovery/leaser.go +++ b/pkg/build/buildkit/autodiscovery/leaser.go @@ -31,6 +31,7 @@ type leaser struct { leasedPodsCh chan<- *corev1.Pod leaseAcquiringWg *sync.WaitGroup leaseCancelByPod map[string]context.CancelFunc + leaseCancelMutex *sync.Mutex holderName string } @@ -42,6 +43,7 @@ func newLeaser(kubernetesInterface kubernetes.Interface, leasablePodsCh <-chan * leasedPodsCh: leasedPodsCh, leaseAcquiringWg: &sync.WaitGroup{}, leaseCancelByPod: make(map[string]context.CancelFunc), + leaseCancelMutex: &sync.Mutex{}, holderName: holderName, }, leasedPodsCh, nil } @@ -57,6 +59,7 @@ func (l *leaser) releaseAll(opts ...releaseOptions) { } else { opt = opts[0] } + l.leaseCancelMutex.Lock() for name, leaseCancel := range l.leaseCancelByPod { if opt.except == name { continue @@ -64,6 +67,7 @@ func (l *leaser) releaseAll(opts ...releaseOptions) { klog.V(4).Infof("Releasing lock for %s pod", name) leaseCancel() } + l.leaseCancelMutex.Unlock() } // acquireLeaseForAllPods tries to acquire leases for all pods received on leasablePodsCh. @@ -73,12 +77,15 @@ func (l *leaser) acquireLeaseForAllPods(ctx context.Context, opts KubernertesDis // NOTE:(ravilock) the usage of WaitGroup here is to ensure that we only close the leasedPodsCh // after all goroutines that might write to it are done. i.e. The goroutines that acquire leases for a buildkit pod. for leasablePod := range l.leasablePodsCh { + l.leaseCancelMutex.Lock() if _, found := l.leaseCancelByPod[leasablePod.Name]; found { + l.leaseCancelMutex.Unlock() continue } leaseCtx, leaseCancel := context.WithCancel(ctx) l.leaseCancelByPod[leasablePod.Name] = leaseCancel + l.leaseCancelMutex.Unlock() l.leaseAcquiringWg.Add(1) go func() {