chore: add detailed debug logging for lease manager and reconciler

Enhanced observability across the concurrency lease manager and pipeline
reconciler by adding debug-level logs. These logs tracked key operations
such as lease acquisition, queue claims, promotion attempts, and
reconciliation states to simplify troubleshooting of queue behavior.

Signed-off-by: Chmouel Boudjnah <chmouel@redhat.com>

Author: chmouel

pkg/queue/lease_manager.go

@@ -53,8 +53,10 @@ func NewManagerForBackend(
 	leaseNamespace, backend string,
 ) ManagerInterface {
 	if backend == settings.ConcurrencyBackendLease {
+		logger.Debugf("initializing lease-backed concurrency manager in namespace %s", leaseNamespace)
 		return NewLeaseManager(logger, kube, tekton, leaseNamespace)
 	}
+	logger.Debugf("initializing in-memory concurrency manager for backend %q", backend)
 	return NewManager(logger)
 }
 
@@ -93,6 +95,7 @@ func (m *LeaseManager) RemoveRepository(repo *v1alpha1.Repository) {
 	releaseCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
+	m.logger.Debugf("deleting concurrency lease for repository %s", RepoKey(repo))
 	if err := m.kube.CoordinationV1().Leases(m.leaseNamespace).Delete(releaseCtx, repoLeaseName(RepoKey(repo)), metav1.DeleteOptions{}); err != nil &&
 		!apierrors.IsNotFound(err) {
 		m.logger.Warnf("failed to delete queue lease for repository %s: %v", RepoKey(repo), err)
@@ -138,8 +141,13 @@ func (m *LeaseManager) RunningPipelineRuns(repo *v1alpha1.Repository) []string {
 
 func (m *LeaseManager) AddListToRunningQueue(ctx context.Context, repo *v1alpha1.Repository, orderedList []string) ([]string, error) {
 	if repo.Spec.ConcurrencyLimit == nil || *repo.Spec.ConcurrencyLimit == 0 {
+		m.logger.Debugf("skipping lease queue claim for repository %s because concurrency limit is disabled", RepoKey(repo))
 		return []string{}, nil
 	}
+	m.logger.Debugf(
+		"attempting to claim queued pipelineruns for repository %s with concurrency limit %d and preferred order %v",
+		RepoKey(repo), *repo.Spec.ConcurrencyLimit, orderedList,
+	)
 	return m.claimNextQueued(ctx, repo, orderedList, "")
 }
 
@@ -148,6 +156,7 @@ func (m *LeaseManager) AddToPendingQueue(*v1alpha1.Repository, []string) error {
 }
 
 func (m *LeaseManager) RemoveFromQueue(ctx context.Context, repo *v1alpha1.Repository, prKey string) bool {
+	m.logger.Debugf("clearing queue claim for pipelinerun %s in repository %s", prKey, RepoKey(repo))
 	if err := m.clearClaim(ctx, repo, prKey); err != nil {
 		m.logger.Warnf("failed to clear queue claim for %s: %v", prKey, err)
 		return false
@@ -157,18 +166,25 @@ func (m *LeaseManager) RemoveFromQueue(ctx context.Context, repo *v1alpha1.Repos
 
 func (m *LeaseManager) RemoveAndTakeItemFromQueue(ctx context.Context, repo *v1alpha1.Repository, run *tektonv1.PipelineRun) string {
 	if repo.Spec.ConcurrencyLimit == nil || *repo.Spec.ConcurrencyLimit == 0 {
+		m.logger.Debugf("not attempting queue handoff for repository %s because concurrency limit is disabled", RepoKey(repo))
 		return ""
 	}
 
 	orderedList := executionOrderList(run)
+	m.logger.Debugf(
+		"removing pipelinerun %s from repository %s and attempting to claim next queued item from order %v",
+		PrKey(run), RepoKey(repo), orderedList,
+	)
 	claimed, err := m.claimNextQueued(ctx, repo, orderedList, PrKey(run))
 	if err != nil {
 		m.logger.Warnf("failed to claim next queued pipelinerun for repository %s: %v", RepoKey(repo), err)
 		return ""
 	}
 	if len(claimed) == 0 {
+		m.logger.Debugf("no queued pipelinerun available to claim for repository %s after removing %s", RepoKey(repo), PrKey(run))
 		return ""
 	}
+	m.logger.Debugf("claimed next queued pipelinerun %s for repository %s", claimed[0], RepoKey(repo))
 	return claimed[0]
 }
 
@@ -194,27 +210,42 @@ func (m *LeaseManager) claimNextQueued(ctx context.Context, repo *v1alpha1.Repos
 
 		occupied := len(state.running) + len(state.claimed)
 		available := *repo.Spec.ConcurrencyLimit - occupied
+		m.logger.Debugf(
+			"lease queue state for repository %s: running=%d claimed=%d queued=%d occupied=%d available=%d exclude=%q preferred=%v",
+			RepoKey(repo), len(state.running), len(state.claimed), len(state.queued), occupied, available, excludeKey, preferredOrder,
+		)
 		if available <= 0 {
+			m.logger.Debugf("repository %s has no available concurrency slots", RepoKey(repo))
 			return nil
 		}
 
 		for _, pr := range state.queued {
 			if available == 0 {
 				break
 			}
+			m.logger.Debugf("attempting to claim queued pipelinerun %s for repository %s", PrKey(&pr), RepoKey(repo))
 			ok, err := m.claimPipelineRun(lockCtx, &pr)
 			if err != nil {
 				return err
 			}
 			if ok {
 				claimed = append(claimed, PrKey(&pr))
 				available--
+				m.logger.Debugf(
+					"claimed queued pipelinerun %s for repository %s; remaining available slots %d",
+					PrKey(&pr), RepoKey(repo), available,
+				)
+				continue
 			}
+			m.logger.Debugf("pipelinerun %s could not be claimed for repository %s", PrKey(&pr), RepoKey(repo))
 		}
 
 		return nil
 	})
 
+	if err == nil {
+		m.logger.Debugf("finished lease queue claim for repository %s; claimed=%v", RepoKey(repo), claimed)
+	}
 	return claimed, err
 }
 
@@ -264,6 +295,10 @@ func (m *LeaseManager) getRepoQueueState(ctx context.Context, repo *v1alpha1.Rep
 			}
 		case kubeinteraction.StateQueued:
 			if !IsRecoverableQueuedPipelineRun(&pr) {
+				m.logger.Debugf(
+					"skipping queued pipelinerun %s for repository %s because it is not recoverable",
+					PrKey(&pr), RepoKey(repo),
+				)
 				continue
 			}
 			position, ok := executionOrderIndex(&pr)
@@ -276,8 +311,10 @@ func (m *LeaseManager) getRepoQueueState(ctx context.Context, repo *v1alpha1.Rep
 				position:  position,
 			}
 			if m.hasActiveClaim(&pr, now) {
+				m.logger.Debugf("queued pipelinerun %s for repository %s already has an active claim", PrKey(&pr), RepoKey(repo))
 				state.claimed = append(state.claimed, pr)
 			} else {
+				m.logger.Debugf("queued pipelinerun %s for repository %s is available for claiming", PrKey(&pr), RepoKey(repo))
 				state.queued = append(state.queued, pr)
 			}
 		}
@@ -293,6 +330,10 @@ func (m *LeaseManager) getRepoQueueState(ctx context.Context, repo *v1alpha1.Rep
 		return compareQueueCandidates(&state.queued[i], &state.queued[j], preferredIndex, orderMeta)
 	})
 
+	m.logger.Debugf(
+		"computed lease queue state for repository %s with running=%v claimed=%v queued=%v",
+		RepoKey(repo), pipelineRunKeys(state.running), pipelineRunKeys(state.claimed), pipelineRunKeys(state.queued),
+	)
 	return state, nil
 }
 
@@ -347,28 +388,39 @@ func (m *LeaseManager) hasActiveClaim(pr *tektonv1.PipelineRun, now time.Time) b
 
 	claimedTime, err := time.Parse(time.RFC3339Nano, claimedAt)
 	if err != nil {
+		m.logger.Debugf("ignoring invalid queue claim timestamp %q on pipelinerun %s: %v", claimedAt, PrKey(pr), err)
 		return false
 	}
 
-	return now.Sub(claimedTime) <= m.claimTTL
+	age := now.Sub(claimedTime)
+	active := age <= m.claimTTL
+	m.logger.Debugf(
+		"evaluated queue claim for pipelinerun %s: claimedBy=%s age=%s ttl=%s active=%t",
+		PrKey(pr), claimedBy, age, m.claimTTL, active,
+	)
+	return active
 }
 
 func (m *LeaseManager) claimPipelineRun(ctx context.Context, pr *tektonv1.PipelineRun) (bool, error) {
+	claimedAt := m.now().UTC().Format(time.RFC3339Nano)
+	m.logger.Debugf("patching pipelinerun %s with queue claim owned by %s at %s", PrKey(pr), m.identity, claimedAt)
 	mergePatch := map[string]any{
 		"metadata": map[string]any{
 			"annotations": map[string]any{
 				keys.QueueClaimedBy: m.identity,
-				keys.QueueClaimedAt: m.now().UTC().Format(time.RFC3339Nano),
+				keys.QueueClaimedAt: claimedAt,
 			},
 		},
 	}
 
 	if _, err := action.PatchPipelineRun(ctx, m.logger, "queue claim", m.tekton, pr, mergePatch); err != nil {
 		if apierrors.IsNotFound(err) {
+			m.logger.Debugf("pipelinerun %s disappeared before queue claim could be recorded", PrKey(pr))
 			return false, nil
 		}
 		return false, err
 	}
+	m.logger.Debugf("successfully recorded queue claim for pipelinerun %s", PrKey(pr))
 	return true, nil
 }
 
@@ -381,15 +433,21 @@ func (m *LeaseManager) clearClaim(ctx context.Context, repo *v1alpha1.Repository
 	pr, err := m.tekton.TektonV1().PipelineRuns(nameParts[0]).Get(ctx, nameParts[1], metav1.GetOptions{})
 	if err != nil {
 		if apierrors.IsNotFound(err) {
+			m.logger.Debugf("pipelinerun %s was already deleted before queue claim cleanup", prKey)
 			return nil
 		}
 		return err
 	}
 
 	if pr.GetAnnotations()[keys.Repository] != repo.Name {
+		m.logger.Debugf(
+			"skipping queue claim cleanup for pipelinerun %s because it belongs to repository %s instead of %s",
+			prKey, pr.GetAnnotations()[keys.Repository], repo.Name,
+		)
 		return nil
 	}
 
+	m.logger.Debugf("removing queue claim annotations from pipelinerun %s", prKey)
 	mergePatch := map[string]any{
 		"metadata": map[string]any{
 			"annotations": map[string]any{
@@ -403,21 +461,28 @@ func (m *LeaseManager) clearClaim(ctx context.Context, repo *v1alpha1.Repository
 	if err != nil && !apierrors.IsNotFound(err) {
 		return err
 	}
+	m.logger.Debugf("queue claim cleanup completed for pipelinerun %s", prKey)
 	return nil
 }
 
 func (m *LeaseManager) withRepoLease(ctx context.Context, repo *v1alpha1.Repository, fn func(context.Context) error) error {
 	leaseName := repoLeaseName(RepoKey(repo))
 
 	for attempt := 0; attempt < defaultLeaseAcquireRetries; attempt++ {
+		m.logger.Debugf(
+			"attempting to acquire concurrency lease %s for repository %s (attempt %d/%d)",
+			leaseName, RepoKey(repo), attempt+1, defaultLeaseAcquireRetries,
+		)
 		acquired, err := m.tryAcquireLease(ctx, leaseName)
 		if err != nil {
 			return err
 		}
 		if acquired {
+			m.logger.Debugf("acquired concurrency lease %s for repository %s", leaseName, RepoKey(repo))
 			defer m.releaseLease(leaseName)
 			return fn(ctx)
 		}
+		m.logger.Debugf("concurrency lease %s for repository %s is currently held by another watcher", leaseName, RepoKey(repo))
 
 		select {
 		case <-ctx.Done():
@@ -435,6 +500,7 @@ func (m *LeaseManager) tryAcquireLease(ctx context.Context, leaseName string) (b
 
 	lease, err := leases.Get(ctx, leaseName, metav1.GetOptions{})
 	if apierrors.IsNotFound(err) {
+		m.logger.Debugf("creating new concurrency lease %s for identity %s", leaseName, m.identity)
 		_, err = leases.Create(ctx, &coordinationv1.Lease{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      leaseName,
@@ -448,9 +514,11 @@ func (m *LeaseManager) tryAcquireLease(ctx context.Context, leaseName string) (b
 			},
 		}, metav1.CreateOptions{})
 		if err == nil {
+			m.logger.Debugf("created and acquired concurrency lease %s", leaseName)
 			return true, nil
 		}
 		if apierrors.IsAlreadyExists(err) {
+			m.logger.Debugf("concurrency lease %s was created by another watcher before acquisition completed", leaseName)
 			return false, nil
 		}
 		return false, err
@@ -460,6 +528,11 @@ func (m *LeaseManager) tryAcquireLease(ctx context.Context, leaseName string) (b
 	}
 
 	if !m.canTakeLease(lease, now.Time) {
+		holder := ""
+		if lease.Spec.HolderIdentity != nil {
+			holder = *lease.Spec.HolderIdentity
+		}
+		m.logger.Debugf("cannot acquire concurrency lease %s because it is still held by %s", leaseName, holder)
 		return false, nil
 	}
 
@@ -473,11 +546,13 @@ func (m *LeaseManager) tryAcquireLease(ctx context.Context, leaseName string) (b
 
 	if _, err := leases.Update(ctx, updated, metav1.UpdateOptions{}); err != nil {
 		if apierrors.IsConflict(err) {
+			m.logger.Debugf("conflict while updating concurrency lease %s; another watcher won the race", leaseName)
 			return false, nil
 		}
 		return false, err
 	}
 
+	m.logger.Debugf("updated concurrency lease %s to holder %s", leaseName, m.identity)
 	return true, nil
 }
 
@@ -549,3 +624,11 @@ func repoLeaseName(repoKey string) string {
 	sum := sha256.Sum256([]byte(repoKey))
 	return fmt.Sprintf("pac-concurrency-%s", hex.EncodeToString(sum[:8]))
 }
+
+func pipelineRunKeys(prs []tektonv1.PipelineRun) []string {
+	keys := make([]string, 0, len(prs))
+	for i := range prs {
+		keys = append(keys, PrKey(&prs[i]))
+	}
+	return keys
+}

pkg/reconciler/controller.go

@@ -124,8 +124,10 @@ func runLeaseQueueRecovery(logger *zap.SugaredLogger, impl *controller.Impl, lis
 		logger.Warnf("failed to list queued PipelineRuns for lease recovery: %v", err)
 		return
 	}
+	logger.Debugf("lease queue recovery selected %d repository candidate(s): %v", len(recoveryKeys), recoveryKeys)
 
 	for _, key := range recoveryKeys {
+		logger.Debugf("enqueuing queued pipelinerun %s for lease recovery", key.String())
 		impl.EnqueueKey(key)
 	}
 }

pkg/reconciler/queue_pipelineruns.go

@@ -57,6 +57,7 @@ func (r *Reconciler) queuePipelineRun(ctx context.Context, logger *zap.SugaredLo
 	// if concurrency was set and later removed or changed to zero
 	// then remove pipelineRun from Queue and update pending state to running
 	if repo.Spec.ConcurrencyLimit != nil && *repo.Spec.ConcurrencyLimit == 0 {
+		logger.Debugf("concurrency disabled for repository %s; promoting queued pipelinerun %s immediately", repo.GetName(), pr.GetName())
 		_ = r.qm.RemoveAndTakeItemFromQueue(ctx, repo, pr)
 		if err := r.updatePipelineRunToInProgress(ctx, logger, repo, pr); err != nil {
 			return fmt.Errorf("failed to update PipelineRun to in_progress: %w", err)
@@ -69,21 +70,29 @@ func (r *Reconciler) queuePipelineRun(ctx context.Context, logger *zap.SugaredLo
 	maxIterations := 5
 
 	orderedList := queuepkg.FilterPipelineRunByState(ctx, r.run.Clients.Tekton, strings.Split(order, ","), tektonv1.PipelineRunSpecStatusPending, kubeinteraction.StateQueued)
+	logger.Debugf(
+		"processing queued pipelinerun %s/%s for repository %s with concurrency limit %v and ordered candidates %v",
+		pr.Namespace, pr.Name, repo.GetName(), repo.Spec.ConcurrencyLimit, orderedList,
+	)
 	for {
+		logger.Debugf("attempting queue acquisition loop %d for repository %s and pipelinerun %s/%s", itered+1, repo.GetName(), pr.Namespace, pr.Name)
 		acquired, err := r.qm.AddListToRunningQueue(ctx, repo, orderedList)
 		if err != nil {
 			return fmt.Errorf("failed to add to queue: %s: %w", pr.GetName(), err)
 		}
+		logger.Debugf("queue acquisition for repository %s returned candidates %v", repo.GetName(), acquired)
 		if len(acquired) == 0 {
 			logger.Infof("no new PipelineRun acquired for repo %s", repo.GetName())
 			break
 		}
 
 		for _, prKeys := range acquired {
+			logger.Debugf("attempting to promote queued pipelinerun %s for repository %s", prKeys, repo.GetName())
 			nsName := strings.Split(prKeys, "/")
 			pr, err = r.run.Clients.Tekton.TektonV1().PipelineRuns(nsName[0]).Get(ctx, nsName[1], metav1.GetOptions{})
 			if err != nil {
 				logger.Info("failed to get pr with namespace and name: ", nsName[0], nsName[1])
+				logger.Debugf("clearing queue claim for missing pipelinerun %s after failed fetch", prKeys)
 				_ = r.qm.RemoveFromQueue(ctx, repo, prKeys)
 			} else {
 				if err := r.updatePipelineRunToInProgress(ctx, logger, repo, pr); err != nil {
@@ -97,13 +106,16 @@ func (r *Reconciler) queuePipelineRun(ctx context.Context, logger *zap.SugaredLo
 						continue
 					}
 					logger.Errorf("failed to update pipelineRun to in_progress: %w", err)
+					logger.Debugf("recording queue promotion failure for pipelinerun %s after promotion error", queuepkg.PrKey(pr))
 					retryErr := r.recordQueuePromotionFailure(ctx, logger, pr, err)
 					if retryErr != nil {
 						return fmt.Errorf("failed to record queue promotion failure for %s after promotion error: %w", pr.GetName(), retryErr)
 					}
+					logger.Debugf("removing queue claim for pipelinerun %s after promotion failure", prKeys)
 					_ = r.qm.RemoveFromQueue(ctx, repo, prKeys)
 					return fmt.Errorf("failed to update pipelineRun to in_progress: %w", err)
 				}
+				logger.Debugf("successfully promoted queued pipelinerun %s for repository %s", queuepkg.PrKey(pr), repo.GetName())
 				processed = true
 			}
 		}
@@ -123,7 +135,10 @@ func (r *Reconciler) pipelineRunReachedStartedState(ctx context.Context, pr *tek
 	if err != nil {
 		return false, err
 	}
-	return latest.GetAnnotations()[keys.State] == kubeinteraction.StateStarted, nil
+	started := latest.GetAnnotations()[keys.State] == kubeinteraction.StateStarted
+	r.run.Clients.Log.Debugf("checked latest state for pipelinerun %s/%s while handling queue promotion: state=%s started=%t",
+		pr.Namespace, pr.Name, latest.GetAnnotations()[keys.State], started)
+	return started, nil
 }
 
 func (r *Reconciler) recordQueuePromotionFailure(ctx context.Context, logger *zap.SugaredLogger, pr *tektonv1.PipelineRun, cause error) error {
@@ -135,6 +150,10 @@ func (r *Reconciler) recordQueuePromotionFailure(ctx context.Context, logger *za
 		}
 	}
 	retries++
+	logger.Debugf(
+		"recording queue promotion failure for pipelinerun %s/%s with retry=%d cause=%v",
+		pr.Namespace, pr.Name, retries, cause,
+	)
 
 	annotations := map[string]any{
 		keys.QueuePromotionRetries: strconv.Itoa(retries),
@@ -149,5 +168,6 @@ func (r *Reconciler) recordQueuePromotionFailure(ctx context.Context, logger *za
 	if err != nil {
 		return err
 	}
+	logger.Debugf("recorded queue promotion failure annotations for pipelinerun %s/%s", pr.Namespace, pr.Name)
 	return nil
 }