test: add test for concurrency evaluation

pkg/build/buildkit/autodiscovery/k8s.go

@@ -2,13 +2,19 @@
 // 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 (
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"
 	"strconv"
 	"strings"
 	"time"
@@ -120,10 +126,34 @@ func (d *K8sDiscoverer) discoverBuildKitClientFromApp(ctx context.Context, opts
 	return c, cleanUps(cfns...), nil
 }
 
-func (d *K8sDiscoverer) discoverBuildKitPod(ctx context.Context, opts KubernertesDiscoveryOptions, namespace string, w io.Writer) (*corev1.Pod, error) {
-	deadlineCtx, deadlineCancel := context.WithCancel(ctx)
-	defer deadlineCancel()
+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()
 
@@ -134,7 +164,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{
@@ -145,56 +175,47 @@ 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)
 	}
-	go leaser.acquireLeaseForAllPods(deadlineCtx, opts)
+	go leaser.acquireLeaseForAllPods(ctx, opts)
 
 	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
 		}
 	}
 }
 
-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 {