diff --git a/pkg/build/buildkit/autodiscovery/k8s.go b/pkg/build/buildkit/autodiscovery/k8s.go index 220b8d8..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 ( @@ -9,6 +14,7 @@ import ( "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 { diff --git a/pkg/build/buildkit/autodiscovery/k8s_test.go b/pkg/build/buildkit/autodiscovery/k8s_test.go index 6e18af6..990ddf8 100644 --- a/pkg/build/buildkit/autodiscovery/k8s_test.go +++ b/pkg/build/buildkit/autodiscovery/k8s_test.go @@ -8,15 +8,20 @@ import ( "bytes" "context" "encoding/json" + "errors" "os" + "sync" "testing" "time" + k8sErrors "k8s.io/apimachinery/pkg/api/errors" + "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 +30,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 +351,145 @@ 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 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 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() { + _, cleanups, ns, err := discoverer1.Discover(context.Background(), discoveryOptions, req1, &buf1) + require.Equal(t, buildkitPod.Namespace, ns) + require.NoError(t, err) + require.NotNil(t, cleanups) + cleanups1 = cleanups + doneCh1 <- struct{}{} + }() + go func() { + _, cleanups, ns, err := discoverer2.Discover(context.Background(), discoveryOptions, req2, &buf2) + require.Equal(t, buildkitPod.Namespace, ns) + require.NoError(t, err) + require.NotNil(t, cleanups) + cleanups2 = cleanups + doneCh2 <- struct{}{} + }() + + var firstAquiredClient string + var firstAquiredCleanups func() + var secondAquiredCleanups func() + select { + case <-doneCh1: + firstAquiredClient = "client-1" + firstAquiredCleanups = cleanups1 + case <-doneCh2: + firstAquiredClient = "client-2" + firstAquiredCleanups = cleanups2 + } + 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 + 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) + 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") + 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{}) + require.NoError(t, err) + require.Len(t, leaseList.Items, 1) + notHeldLease := leaseList.Items[0].DeepCopy() + 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..7356c24 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" @@ -20,24 +19,23 @@ 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 leasedPodsCh chan<- *corev1.Pod leaseAcquiringWg *sync.WaitGroup leaseCancelByPod map[string]context.CancelFunc + leaseCancelMutex *sync.Mutex 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 - } +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, @@ -45,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 } @@ -60,7 +59,7 @@ func (l *leaser) releaseAll(opts ...releaseOptions) { } else { opt = opts[0] } - l.leaseAcquiringWg.Wait() + l.leaseCancelMutex.Lock() for name, leaseCancel := range l.leaseCancelByPod { if opt.except == name { continue @@ -68,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. @@ -77,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() { @@ -98,8 +101,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) { - 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) + 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{ @@ -108,18 +110,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: - 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) @@ -128,5 +130,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) } 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 + } +} 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