Provisioning: Improve logging and tracing in job processing (#113454)

* Provisioning: Improve logging and tracing in job processing

- Add comprehensive tracing with OpenTelemetry spans across all job operations
- Enhance logging with consistent style: lowercase, concise messages, appropriate log levels
- Use past tense for completed lifecycle events (e.g., 'stopped' vs 'stop')
- Add structured logging with contextual attributes for better searchability
- Handle graceful shutdowns without throwing errors on context cancellation
- Refactor Cleanup method into listExpiredJobs and cleanUpExpiredJob for better code quality
- Avoid double logging by only logging errors when handled locally
- Add tracing and logging to historyjob controller cleanup operations

Files modified:
- pkg/registry/apis/provisioning/jobs/driver.go: Add tracing spans and improve error handling for graceful shutdown
- pkg/registry/apis/provisioning/jobs/concurrent_driver.go: Add tracing and consistent logging
- pkg/registry/apis/provisioning/jobs/persistentstore.go: Add comprehensive tracing and logging to all public methods, refactor cleanup
- apps/provisioning/pkg/controller/historyjob.go: Add tracing and improve logging consistency

* Update pkg/registry/apis/provisioning/jobs/persistentstore.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Refactor logging in persistentstore.go

- Remove debug log statements at the start of job operations for cleaner output
- Maintain structured logging with contextual attributes for improved traceability

Files modified:
- pkg/registry/apis/provisioning/jobs/persistentstore.go: Clean up logging for job operations

* Enhance logging and tracing in provisioning job operations

- Introduce OpenTelemetry spans for better observability in job processing and webhook handling
- Improve structured logging with contextual attributes for key operations
- Remove unnecessary tracing spans in long-running functions to streamline performance
- Update error handling to record errors in spans for better traceability

Files modified:
- pkg/registry/apis/provisioning/controller/repository.go: Add tracing and structured logging to sync job operations
- pkg/registry/apis/provisioning/jobs/concurrent_driver.go: Remove tracing span from long-running function
- pkg/registry/apis/provisioning/jobs/driver.go: Enhance logging and tracing in job processing
- pkg/registry/apis/provisioning/webhooks/webhook.go: Implement tracing and structured logging for webhook connections

* Update pkg/registry/apis/provisioning/jobs/driver.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Improve error handling in ConcurrentJobDriver to differentiate between graceful shutdown and unexpected stops

* Remove unused import in driver.go to clean up code

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

apps/provisioning/pkg/controller/historyjob.go

@@ -58,32 +58,42 @@ func NewHistoryJobController(
 func (c *HistoryJobController) cleanupJob(obj interface{}) {
 	job, ok := obj.(*provisioning.HistoricJob)
 	if !ok {
-		c.logger.Error("Expected HistoricJob but got", "type", obj)
+		c.logger.Error("unexpected object type - expected HistoricJob", "type", obj)
 		return
 	}
 
 	age := time.Since(job.CreationTimestamp.Time)
-	if age > c.expirationTime {
-		namespace := job.Namespace
-		ctx, _, err := identity.WithProvisioningIdentity(context.Background(), namespace)
-		if err != nil {
-			c.logger.Error("Failed to set provisioning identity for cleanup", "error", err)
+
+	// Only cleanup jobs older than expiration time
+	if age <= c.expirationTime {
+		return
+	}
+
+	logger := c.logger.With(
+		"job", job.Name,
+		"namespace", job.Namespace,
+		"age", age,
+	)
+
+	logger.Debug("start cleanup expired historic job")
+
+	namespace := job.Namespace
+	ctx, _, err := identity.WithProvisioningIdentity(context.Background(), namespace)
+	if err != nil {
+		logger.Error("failed to set provisioning identity", "error", err)
+		return
+	}
+
+	ctx = request.WithNamespace(ctx, namespace)
+	err = c.client.HistoricJobs(job.Namespace).Delete(ctx, job.Name, metav1.DeleteOptions{})
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			logger.Debug("historic job already deleted")
 			return
 		}
-
-		ctx = request.WithNamespace(ctx, namespace)
-		err = c.client.HistoricJobs(job.Namespace).Delete(ctx, job.Name, metav1.DeleteOptions{})
-		if err != nil && !apierrors.IsNotFound(err) {
-			c.logger.Error("Failed to delete expired HistoryJob",
-				"namespace", job.Namespace,
-				"name", job.Name,
-				"age", age,
-				"error", err)
-		} else {
-			c.logger.Info("Deleted expired HistoryJob",
-				"namespace", job.Namespace,
-				"name", job.Name,
-				"age", age)
-		}
+		logger.Error("failed to delete expired historic job", "error", err)
+		return
 	}
+
+	logger.Info("deleted expired historic job")
 }

pkg/registry/apis/provisioning/controller/repository.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -140,6 +141,9 @@ func repoKeyFunc(obj any) (string, error) {
 }
 
 // Run starts the RepositoryController.
+//
+// Note: This function intentionally does NOT create a tracing span because it runs indefinitely
+// until shutdown. Individual processing operations already have their own spans.
 func (rc *RepositoryController) Run(ctx context.Context, workerCount int) {
 	defer utilruntime.HandleCrash()
 	defer rc.queue.ShutDown()
@@ -386,21 +390,31 @@ func shouldUseIncrementalSync(ctx context.Context, versioned repository.Versione
 }
 
 func (rc *RepositoryController) addSyncJob(ctx context.Context, obj *provisioning.Repository, syncOptions *provisioning.SyncJobOptions) error {
+	ctx, span := rc.tracer.Start(ctx, "provisioning.controller.add_sync_job")
+	defer span.End()
+
+	span.SetAttributes(
+		attribute.String("repository", obj.GetName()),
+		attribute.String("namespace", obj.Namespace),
+		attribute.Bool("incremental", syncOptions != nil && syncOptions.Incremental),
+	)
+
 	job, err := rc.jobs.Insert(ctx, obj.Namespace, provisioning.JobSpec{
 		Repository: obj.GetName(),
 		Action:     provisioning.JobActionPull,
 		Pull:       syncOptions,
 	})
 	if apierrors.IsAlreadyExists(err) {
-		logging.FromContext(ctx).Info("sync job already exists, nothing triggered")
+		logging.FromContext(ctx).Info("sync job already exists")
 		return nil
 	}
 	if err != nil {
+		span.RecordError(err)
 		// FIXME: should we update the status of the repository if we fail to add the job?
 		return fmt.Errorf("error adding sync job: %w", err)
 	}
 
-	logging.FromContext(ctx).Info("sync job triggered", "job", job.Name)
+	span.SetAttributes(attribute.String("job.name", job.Name))
 	return nil
 }
 

pkg/registry/apis/provisioning/jobs/concurrent_driver.go

@@ -75,9 +75,12 @@ func NewConcurrentJobDriver(
 
 // Run starts multiple job drivers concurrently and handles cleanup coordination.
 // This is a blocking function that will run until the context is canceled or an error occurs.
+//
+// Note: This function intentionally does NOT create a tracing span because it runs indefinitely
+// until shutdown. Individual job processing and cleanup operations already have their own spans.
 func (c *ConcurrentJobDriver) Run(ctx context.Context) error {
 	logger := logging.FromContext(ctx).With("logger", "concurrent-job-driver", "num_drivers", c.numDrivers)
-	logger.Info("starting concurrent job driver with lease-based cleanup", "cleanup_interval", c.cleanupInterval)
+	logger.Info("start concurrent job driver", "num_drivers", c.numDrivers, "cleanup_interval", c.cleanupInterval)
 
 	// Set up cleanup ticker - runs more frequently with lease-based approach
 	cleanupTicker := time.NewTicker(c.cleanupInterval)
@@ -85,7 +88,7 @@ func (c *ConcurrentJobDriver) Run(ctx context.Context) error {
 
 	// Initial cleanup
 	if err := c.store.Cleanup(ctx); err != nil {
-		logger.Error("failed to clean up old jobs at start", "error", err)
+		logger.Error("failed initial cleanup", "error", err)
 	}
 
 	var wg sync.WaitGroup
@@ -99,10 +102,10 @@ func (c *ConcurrentJobDriver) Run(ctx context.Context) error {
 			select {
 			case <-cleanupTicker.C:
 				if err := c.store.Cleanup(ctx); err != nil {
-					logger.Error("failed to cleanup jobs", "error", err)
+					logger.Error("failed cleanup", "error", err)
 				}
 			case <-ctx.Done():
-				logger.Debug("cleanup goroutine stopping")
+				logger.Debug("cleanup routine stopped")
 				return
 			}
 		}
@@ -133,13 +136,13 @@ func (c *ConcurrentJobDriver) Run(ctx context.Context) error {
 				return
 			}
 
-			driverLogger.Debug("starting job driver")
+			driverLogger.Info("start job driver")
 			if err := driver.Run(driverCtx); err != nil {
 				driverLogger.Error("job driver failed", "error", err)
 				errChan <- err
 				return
 			}
-			driverLogger.Debug("job driver stopped")
+			driverLogger.Info("job driver stopped")
 		}(i)
 	}
 
@@ -157,6 +160,10 @@ func (c *ConcurrentJobDriver) Run(ctx context.Context) error {
 		}
 	}
 
-	logger.Info("all job driver workers stopped")
-	return ctx.Err()
+	if ctx.Err() != nil {
+		logger.Info("all job drivers gracefully stopped")
+		return nil
+	}
+
+	return fmt.Errorf("concurrent job driver stopped unexpectedly")
 }

pkg/registry/apis/provisioning/jobs/driver.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apiserver/pkg/endpoints/request"
 
@@ -14,6 +15,7 @@ import (
 	"github.com/grafana/grafana/apps/provisioning/pkg/apifmt"
 	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/infra/tracing"
 )
 
 // Store is an abstraction for the storage API.
@@ -105,6 +107,9 @@ func NewJobDriver(
 // Run drives jobs to completion. This is a blocking function.
 // It will run until the context is canceled or an error occurs.
 // This is a thread-safe function; it may be called from multiple goroutines.
+//
+// Note: This function intentionally does NOT create a tracing span because it runs indefinitely
+// until shutdown. Individual job processing operations already have their own spans.
 func (d *jobDriver) Run(ctx context.Context) error {
 	jobTicker := time.NewTicker(d.jobInterval)
 	defer jobTicker.Stop()
@@ -122,7 +127,8 @@ func (d *jobDriver) Run(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			logger.Info("job driver stopped")
+			return nil // Context cancellation is expected during shutdown
 		case <-jobTicker.C:
 			d.processJobsUntilDoneOrError(ctx)
 		case <-d.notifications:
@@ -134,6 +140,11 @@ func (d *jobDriver) Run(ctx context.Context) error {
 // This will keep processing jobs until there are none left (or we hit an error)
 func (d *jobDriver) processJobsUntilDoneOrError(ctx context.Context) {
 	for {
+		// Check if context is cancelled before attempting to claim jobs
+		if ctx.Err() != nil {
+			return
+		}
+
 		err := d.claimAndProcessOneJob(ctx)
 		if err != nil {
 			if !errors.Is(err, ErrNoJobs) {
@@ -145,11 +156,17 @@ func (d *jobDriver) processJobsUntilDoneOrError(ctx context.Context) {
 }
 
 func (d *jobDriver) claimAndProcessOneJob(ctx context.Context) error {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.claim_and_process_one_job")
+	defer span.End()
+
 	logger := logging.FromContext(ctx)
 
 	// Claim a job to work on.
 	claimedJob, rollback, err := d.store.Claim(ctx)
 	if err != nil {
+		if !errors.Is(err, ErrNoJobs) {
+			span.RecordError(err)
+		}
 		return apifmt.Errorf("failed to claim job: %w", err)
 	}
 	// Ensure that the job is cleaned up if we fail to complete it.
@@ -159,9 +176,15 @@ func (d *jobDriver) claimAndProcessOneJob(ctx context.Context) error {
 	namespace := claimedJob.GetNamespace()
 	logger = logger.With("job", claimedJob.GetName(), "namespace", namespace)
 	ctx = logging.Context(ctx, logger)
-	logger.Debug("claimed a job")
 	d.currentJob = claimedJob
 
+	span.SetAttributes(
+		attribute.String("job.name", claimedJob.GetName()),
+		attribute.String("job.namespace", namespace),
+		attribute.String("job.repository", claimedJob.Spec.Repository),
+		attribute.String("job.action", string(claimedJob.Spec.Action)),
+	)
+
 	// Now that we have a job, we need to augment our namespace to grant ourselves permission to work on it.
 	// Incidentally, this also limits our permissions to only the namespace of the job.
 	ctx = request.WithNamespace(ctx, namespace)
@@ -188,11 +211,26 @@ func (d *jobDriver) claimAndProcessOneJob(ctx context.Context) error {
 	end := time.Now()
 	logger.Debug("job processed", "duration", end.Sub(recorder.Started()), "error", err)
 
-	// Capture job timeout
-	if jobctx.Err() != nil && err == nil {
+	// Check if parent context was cancelled (graceful shutdown)
+	if ctx.Err() != nil {
+		logger.Debug("context cancel - job will retry")
+		// Don't complete the job - let it be retried by another worker
+		d.mu.Lock()
+		d.currentJob = nil
+		d.mu.Unlock()
+		return nil
+	}
+
+	// Capture job timeout (but not parent context cancellation)
+	if jobctx.Err() != nil && err == nil && ctx.Err() == nil {
 		err = jobctx.Err()
 	}
 
+	// Record job processing error on span
+	if err != nil {
+		span.RecordError(err)
+	}
+
 	// Complete the job
 	d.mu.Lock()
 	d.currentJob.Status = recorder.Complete(ctx, err)
@@ -205,27 +243,29 @@ func (d *jobDriver) claimAndProcessOneJob(ctx context.Context) error {
 	err = d.historicJobs.WriteJob(ctx, d.currentJob.DeepCopy())
 	if err != nil {
 		// We're not going to return this as it is not critical. Not ideal, but not critical.
-		logger.Warn("failed to create historic job", "historic_job", *d.currentJob, "error", err)
-	} else {
-		logger.Debug("created historic job", "historic_job", *d.currentJob)
+		logger.Warn("failed to write historic job", "error", err)
 	}
 
 	// Mark the job as completed.
 	if err := d.store.Complete(ctx, d.currentJob); err != nil {
+		span.RecordError(err)
 		return apifmt.Errorf("failed to complete job '%s' in '%s': %w", d.currentJob.GetName(), d.currentJob.GetNamespace(), err)
 	}
-	logger.Debug("job completed")
+	logger.Info("job complete")
 
 	return nil
 }
 
 // leaseRenewalLoop continuously renews the lease for a job until the context is cancelled.
 // If lease renewal fails persistently, it signals via the leaseExpired channel.
+//
+// Note: This function intentionally does NOT create a tracing span because it runs indefinitely
+// for the lifetime of a job. Individual RenewLease calls already have their own spans.
 func (d *jobDriver) leaseRenewalLoop(ctx context.Context, logger logging.Logger, leaseExpired chan struct{}) {
 	ticker := time.NewTicker(d.leaseRenewalInterval)
 	defer ticker.Stop()
 
-	logger.Debug("starting lease renewal loop", "renewal_interval", d.leaseRenewalInterval)
+	logger.Debug("start lease renewal loop", "renewal_interval", d.leaseRenewalInterval)
 
 	consecutiveFailures := 0
 	maxFailures := 3 // Allow a few failures before giving up
@@ -233,7 +273,7 @@ func (d *jobDriver) leaseRenewalLoop(ctx context.Context, logger logging.Logger,
 	for {
 		select {
 		case <-ctx.Done():
-			logger.Debug("lease renewal loop stopping")
+			logger.Debug("lease renewal loop stopped")
 			return
 		case <-ticker.C:
 			d.mu.Lock()
@@ -267,13 +307,12 @@ func (d *jobDriver) leaseRenewalLoop(ctx context.Context, logger logging.Logger,
 					logger.Debug("lease renewal recovered", "previous_failures", consecutiveFailures)
 				}
 				consecutiveFailures = 0
-				logger.Debug("lease renewed successfully")
 			}
 		}
 	}
 }
 
-// processJobWithLeaseCheck processes a job but aborts if the lease expires.
+// processJobWithLeaseCheck processes a job but aborts if the lease expires or context is cancelled.
 func (d *jobDriver) processJobWithLeaseCheck(ctx context.Context, recorder JobProgressRecorder, leaseExpired <-chan struct{}) error {
 	// Run the job processing in a goroutine so we can monitor lease expiry
 	resultChan := make(chan error, 1)
@@ -287,11 +326,16 @@ func (d *jobDriver) processJobWithLeaseCheck(ctx context.Context, recorder JobPr
 	case <-leaseExpired:
 		return apifmt.Errorf("job aborted due to lease expiry")
 	case <-ctx.Done():
+		// Return context error directly - caller will determine if this is due to graceful shutdown
+		// or job timeout based on which context was cancelled
 		return ctx.Err()
 	}
 }
 
 func (d *jobDriver) processJob(ctx context.Context, recorder JobProgressRecorder) error {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.process_job")
+	defer span.End()
+
 	logger := logging.FromContext(ctx)
 	d.mu.Lock()
 	if d.currentJob == nil {
@@ -305,6 +349,11 @@ func (d *jobDriver) processJob(ctx context.Context, recorder JobProgressRecorder
 	namespace := d.currentJob.Namespace
 	d.mu.Unlock()
 
+	span.SetAttributes(
+		attribute.String("job.repository", repoName),
+		attribute.String("job.action", string(job.Spec.Action)),
+	)
+
 	for _, worker := range d.workers {
 		if !worker.IsSupported(ctx, *job) {
 			continue
@@ -312,12 +361,13 @@ func (d *jobDriver) processJob(ctx context.Context, recorder JobProgressRecorder
 
 		repo, err := d.repoGetter.GetRepository(ctx, namespace, repoName)
 		if err != nil {
+			span.RecordError(err)
 			return apifmt.Errorf("failed to get repository '%s': %w", repoName, err)
 		}
 
 		r := repo.Config()
 		if r.DeletionTimestamp != nil && !r.DeletionTimestamp.IsZero() {
-			logger.Info("repository is marked for deletion, skipping processing job",
+			logger.Info("repository marked for deletion - skip job",
 				"name", r.Name,
 				"namespace", r.Namespace,
 				"deletionTimestamp", r.DeletionTimestamp,
@@ -325,14 +375,23 @@ func (d *jobDriver) processJob(ctx context.Context, recorder JobProgressRecorder
 			return nil
 		}
 
-		return worker.Process(ctx, repo, *job, recorder)
+		err = worker.Process(ctx, repo, *job, recorder)
+		if err != nil {
+			span.RecordError(err)
+		}
+		return err
 	}
 
-	return apifmt.Errorf("no workers were registered to handle the job")
+	err := apifmt.Errorf("no workers were registered to handle the job")
+	span.RecordError(err)
+	return err
 }
 
 func (d *jobDriver) onProgress() ProgressFn {
 	return func(ctx context.Context, status provisioning.JobStatus) error {
+		ctx, span := tracing.Start(ctx, "provisioning.jobs.update_progress")
+		defer span.End()
+
 		logging.FromContext(ctx).Debug("job progress", "status", status)
 
 		const maxRetries = 3
@@ -376,9 +435,16 @@ func (d *jobDriver) onProgress() ProgressFn {
 			// Update succeeded, update our local copy
 			*d.currentJob = *updated
 			d.mu.Unlock()
+
+			span.SetAttributes(
+				attribute.String("job.state", string(status.State)),
+				attribute.Int("attempt", attempt+1),
+			)
 			return nil
 		}
 
-		return apifmt.Errorf("failed to update job progress after %d attempts", maxRetries)
+		err := apifmt.Errorf("failed to update job progress after %d attempts", maxRetries)
+		span.RecordError(err)
+		return err
 	}
 }

pkg/registry/apis/provisioning/jobs/persistentstore.go

@@ -8,6 +8,7 @@ import (
 	"strconv"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -18,6 +19,7 @@ import (
 	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
 	client "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/infra/tracing"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -100,6 +102,16 @@ func NewJobStore(provisioningClient client.ProvisioningV0alpha1Interface, expiry
 // If err is not nil, the job and rollback values are always nil.
 // The err may be ErrNoJobs if there are no jobs to claim.
 func (s *persistentStore) Claim(ctx context.Context) (job *provisioning.Job, rollback func(), err error) {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.claim")
+	defer func() {
+		if err != nil && !errors.Is(err, ErrNoJobs) {
+			span.RecordError(err)
+		}
+		span.End()
+	}()
+
+	logger := logging.FromContext(ctx).With("operation", "claim")
+
 	requirement, err := labels.NewRequirement(LabelJobClaim, selection.DoesNotExist, nil)
 	if err != nil {
 		return nil, nil, apifmt.Errorf("could not create requirement: %w", err)
@@ -114,9 +126,12 @@ func (s *persistentStore) Claim(ctx context.Context) (job *provisioning.Job, rol
 	}
 
 	if len(jobs.Items) == 0 {
+		logger.Debug("no jobs available to claim")
 		return nil, nil, ErrNoJobs
 	}
 
+	logger.Debug("found jobs available", "count", len(jobs.Items))
+
 	for _, job := range jobs.Items {
 		if job.Labels == nil {
 			job.Labels = make(map[string]string)
@@ -145,6 +160,20 @@ func (s *persistentStore) Claim(ctx context.Context) (job *provisioning.Job, rol
 			return nil, nil, apifmt.Errorf("failed to claim job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
 		}
 
+		logger.Info("job claim complete",
+			"job", updatedJob.GetName(),
+			"namespace", updatedJob.GetNamespace(),
+			"repository", updatedJob.Spec.Repository,
+			"action", updatedJob.Spec.Action,
+		)
+
+		span.SetAttributes(
+			attribute.String("job.name", updatedJob.GetName()),
+			attribute.String("job.namespace", updatedJob.GetNamespace()),
+			attribute.String("job.repository", updatedJob.Spec.Repository),
+			attribute.String("job.action", string(updatedJob.Spec.Action)),
+		)
+
 		return updatedJob.DeepCopy(), func() {
 			// Rolling back does not need to care about the parent's cancellation state.
 			// This will also use the parent context (i.e. from the for loop!), ensuring we have permissions to do this.
@@ -181,50 +210,99 @@ func (s *persistentStore) Claim(ctx context.Context) (job *provisioning.Job, rol
 	}
 
 	// We failed to claim any jobs.
+	logger.Debug("no jobs claimed - all already claimed by others")
 	return nil, nil, ErrNoJobs
 }
 
 // Update saves the job back to the store.
 func (s *persistentStore) Update(ctx context.Context, job *provisioning.Job) (*provisioning.Job, error) {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.update")
+	defer span.End()
+
+	logger := logging.FromContext(ctx).With(
+		"operation", "update",
+		"job", job.GetName(),
+		"namespace", job.GetNamespace(),
+	)
+
+	span.SetAttributes(
+		attribute.String("job.name", job.GetName()),
+		attribute.String("job.namespace", job.GetNamespace()),
+	)
+
 	// Set up the provisioning identity for this namespace
 	ctx, _, err := identity.WithProvisioningIdentity(ctx, job.GetNamespace())
 	if err != nil {
+		span.RecordError(err)
 		return nil, apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err)
 	}
 
 	updatedJob, err := s.client.Jobs(job.GetNamespace()).Update(ctx, job, metav1.UpdateOptions{})
 	if err != nil {
+		span.RecordError(err)
 		return nil, apifmt.Errorf("failed to update job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
 	}
 
+	logger.Debug("update job complete")
 	return updatedJob, nil
 }
 
 // Get retrieves a job by name for conflict resolution.
 func (s *persistentStore) Get(ctx context.Context, namespace, name string) (*provisioning.Job, error) {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.get")
+	defer span.End()
+
+	logger := logging.FromContext(ctx).With(
+		"operation", "get",
+		"job", name,
+		"namespace", namespace,
+	)
+
+	span.SetAttributes(
+		attribute.String("job.name", name),
+		attribute.String("job.namespace", namespace),
+	)
+
 	// Set up provisioning identity to access jobs across all namespaces
 	ctx, _, err := identity.WithProvisioningIdentity(ctx, namespace)
 	if err != nil {
+		span.RecordError(err)
 		return nil, apifmt.Errorf("failed to grant provisioning identity for job lookup: %w", err)
 	}
 
 	// Use Get to directly fetch the job by name
 	job, err := s.client.Jobs(namespace).Get(ctx, name, metav1.GetOptions{})
 	if err != nil {
+		span.RecordError(err)
 		return nil, apifmt.Errorf("failed to get job by name '%s': %w", name, err)
 	}
 
+	logger.Debug("get job complete")
 	return job, nil
 }
 
 // Complete marks a job as completed and moves it to the historic job store.
 // When in the historic store, there is no more claim on the job.
 func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) error {
-	logger := logging.FromContext(ctx).With("namespace", job.GetNamespace(), "job", job.GetName())
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.complete")
+	defer span.End()
+
+	logger := logging.FromContext(ctx).With(
+		"operation", "complete",
+		"namespace", job.GetNamespace(),
+		"job", job.GetName(),
+	)
+
+	span.SetAttributes(
+		attribute.String("job.name", job.GetName()),
+		attribute.String("job.namespace", job.GetNamespace()),
+		attribute.String("job.action", string(job.Spec.Action)),
+	)
 
 	// Set up the provisioning identity for this namespace
 	ctx, _, err := identity.WithProvisioningIdentity(ctx, job.GetNamespace())
 	if err != nil {
+		span.RecordError(err)
 		return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err)
 	}
 
@@ -235,6 +313,7 @@ func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) e
 	// This is a best-effort operation; if the job is not in the claimed state, we will still attempt to move it to the historic job store.
 	err = s.client.Jobs(job.GetNamespace()).Delete(ctx, job.GetName(), metav1.DeleteOptions{})
 	if err != nil {
+		span.RecordError(err)
 		return apifmt.Errorf("failed to delete job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
 	}
 	logger.Debug("deleted job from job store")
@@ -246,26 +325,44 @@ func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) e
 	delete(job.Labels, LabelJobClaim)
 	s.queueMetrics.DecreaseQueueSize(string(job.Spec.Action))
 
-	logger.Debug("job completion done")
+	logger.Debug("complete job complete")
 	return nil
 }
 
 // RenewLease renews the lease for a claimed job, extending its expiry time.
 // Returns an error if the lease cannot be renewed (e.g., job was completed or lease expired).
 func (s *persistentStore) RenewLease(ctx context.Context, job *provisioning.Job) error {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.renew_lease")
+	defer span.End()
+
+	logger := logging.FromContext(ctx).With(
+		"operation", "renew_lease",
+		"job", job.GetName(),
+		"namespace", job.GetNamespace(),
+	)
+
+	span.SetAttributes(
+		attribute.String("job.name", job.GetName()),
+		attribute.String("job.namespace", job.GetNamespace()),
+	)
+
 	if job.Labels == nil || job.Labels[LabelJobClaim] == "" {
-		return apifmt.Errorf("job '%s' in '%s' is not claimed", job.GetName(), job.GetNamespace())
+		err := apifmt.Errorf("job '%s' in '%s' is not claimed", job.GetName(), job.GetNamespace())
+		span.RecordError(err)
+		return err
 	}
 
 	// Set up the provisioning identity for this namespace
 	ctx, _, err := identity.WithProvisioningIdentity(ctx, job.GetNamespace())
 	if err != nil {
+		span.RecordError(err)
 		return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err)
 	}
 
 	// Fetch the latest version to avoid conflicts
 	latestJob, err := s.client.Jobs(job.GetNamespace()).Get(ctx, job.GetName(), metav1.GetOptions{})
 	if err != nil {
+		span.RecordError(err)
 		if apierrors.IsNotFound(err) {
 			return apifmt.Errorf("failed to renew lease for job '%s' in '%s': job no longer exists", job.GetName(), job.GetNamespace())
 		}
@@ -274,7 +371,9 @@ func (s *persistentStore) RenewLease(ctx context.Context, job *provisioning.Job)
 
 	// Verify we still own the lease
 	if latestJob.Labels == nil || latestJob.Labels[LabelJobClaim] == "" {
-		return apifmt.Errorf("lease lost for job '%s' in '%s': no longer claimed", job.GetName(), job.GetNamespace())
+		err := apifmt.Errorf("lease lost for job '%s' in '%s': no longer claimed", job.GetName(), job.GetNamespace())
+		span.RecordError(err)
+		return err
 	}
 
 	// Update the claim timestamp to current time
@@ -284,85 +383,213 @@ func (s *persistentStore) RenewLease(ctx context.Context, job *provisioning.Job)
 	// Update the job in storage with the latest resource version
 	_, err = s.client.Jobs(job.GetNamespace()).Update(ctx, updatedJob, metav1.UpdateOptions{})
 	if apierrors.IsConflict(err) {
-		return apifmt.Errorf("failed to renew lease for job '%s' in '%s': lease conflict", job.GetName(), job.GetNamespace())
+		err := apifmt.Errorf("failed to renew lease for job '%s' in '%s': lease conflict", job.GetName(), job.GetNamespace())
+		span.RecordError(err)
+		return err
 	}
 	if apierrors.IsNotFound(err) {
-		return apifmt.Errorf("failed to renew lease for job '%s' in '%s': job no longer exists", job.GetName(), job.GetNamespace())
+		err := apifmt.Errorf("failed to renew lease for job '%s' in '%s': job no longer exists", job.GetName(), job.GetNamespace())
+		span.RecordError(err)
+		return err
 	}
 	if err != nil {
+		span.RecordError(err)
 		return apifmt.Errorf("failed to renew lease for job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
 	}
 
 	// Update the job's claim timestamp and resource version in memory
 	job.Labels[LabelJobClaim] = updatedJob.Labels[LabelJobClaim]
 	job.ResourceVersion = updatedJob.ResourceVersion
+
+	logger.Debug("renew lease complete")
 	return nil
 }
 
 // Cleanup finds jobs with expired leases and marks them as failed.
 // This replaces the old cleanup mechanism and should be called more frequently.
 func (s *persistentStore) Cleanup(ctx context.Context) error {
-	// Set up provisioning identity to access jobs across all namespaces
-	ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") // "*" grants access to all namespaces
-	if err != nil {
-		return apifmt.Errorf("failed to grant provisioning identity for cleanup: %w", err)
-	}
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.cleanup")
+	defer span.End()
 
-	// Find jobs with expired leases (older than expiry time)
-	expiry := s.clock().Add(-s.expiry).UnixMilli()
-	requirement, err := labels.NewRequirement(LabelJobClaim, selection.LessThan, []string{strconv.FormatInt(expiry, 10)})
-	if err != nil {
-		return apifmt.Errorf("could not create requirement: %w", err)
-	}
+	startTime := s.clock()
+	logger := logging.FromContext(ctx).With("operation", "cleanup")
 
-	timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	jobs, err := s.client.Jobs("").List(timeoutCtx, metav1.ListOptions{
-		LabelSelector: labels.NewSelector().Add(*requirement).String(),
-		Limit:         100, // Process in batches
-	})
-	cancel()
+	// List expired jobs
+	jobs, err := s.listExpiredJobs(ctx)
 	if err != nil {
-		return apifmt.Errorf("failed to list jobs with expired leases: %w", err)
+		span.RecordError(err)
+		return err
 	}
 
 	// If no jobs found, cleanup is complete
-	if len(jobs.Items) == 0 {
+	if len(jobs) == 0 {
+		duration := s.clock().Sub(startTime)
+		logger.Info("cleanup complete - no expired jobs found", "duration", duration)
+		span.SetAttributes(
+			attribute.Int("count", 0),
+			attribute.Int64("duration_ms", duration.Milliseconds()),
+		)
 		return nil
 	}
 
-	for _, job := range jobs.Items {
-		// Mark job as failed due to lease expiry and archive it
-		job := job.DeepCopy()
-		job.Status.State = provisioning.JobStateError
-		job.Status.Message = "Job failed due to lease expiry - worker may have crashed or lost connection"
+	logger.Info("found expired jobs", "count", len(jobs))
 
-		// Set namespace context for the completion
-		ctx, _, err = identity.WithProvisioningIdentity(ctx, job.GetNamespace())
-		if err != nil {
-			return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err)
-		}
-
-		// Use Complete to properly archive the failed job
-		if err := s.Complete(ctx, job); err != nil {
-			if apierrors.IsNotFound(err) {
-				// Job was already completed/deleted by another process
-				continue
-			}
-			return apifmt.Errorf("failed to complete expired job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
+	// Clean up each expired job
+	for _, job := range jobs {
+		if err := s.cleanUpExpiredJob(ctx, job); err != nil {
+			span.RecordError(err)
+			return err
 		}
 	}
 
+	duration := s.clock().Sub(startTime)
+	logger.Info("cleanup complete",
+		"duration", duration,
+		"count", len(jobs),
+	)
+
+	span.SetAttributes(
+		attribute.Int("count", len(jobs)),
+		attribute.Int64("duration_ms", duration.Milliseconds()),
+	)
+
+	return nil
+}
+
+// listExpiredJobs returns jobs with expired leases.
+func (s *persistentStore) listExpiredJobs(ctx context.Context) ([]provisioning.Job, error) {
+	logger := logging.FromContext(ctx)
+
+	// Set up provisioning identity to access jobs across all namespaces
+	ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") // "*" grants access to all namespaces
+	if err != nil {
+		return nil, apifmt.Errorf("failed to grant provisioning identity for cleanup: %w", err)
+	}
+
+	// Find jobs with expired leases (older than expiry time)
+	expiry := s.clock().Add(-s.expiry).UnixMilli()
+	expiryTime := time.UnixMilli(expiry)
+	logger.Debug("search for expired jobs", "expiry_threshold", expiryTime.Format(time.RFC3339))
+
+	requirement, err := labels.NewRequirement(LabelJobClaim, selection.LessThan, []string{strconv.FormatInt(expiry, 10)})
+	if err != nil {
+		return nil, apifmt.Errorf("could not create requirement: %w", err)
+	}
+
+	listCtx, listSpan := tracing.Start(ctx, "provisioning.jobs.cleanup.list_expired_jobs")
+	defer listSpan.End()
+
+	listSpan.SetAttributes(
+		attribute.String("expiry_threshold", expiryTime.Format(time.RFC3339)),
+		attribute.Int64("expiry_duration_seconds", int64(s.expiry.Seconds())),
+	)
+
+	timeoutCtx, cancel := context.WithTimeout(listCtx, 5*time.Second)
+	defer cancel()
+
+	jobList, err := s.client.Jobs("").List(timeoutCtx, metav1.ListOptions{
+		LabelSelector: labels.NewSelector().Add(*requirement).String(),
+		Limit:         100, // Process in batches
+	})
+	if err != nil {
+		listSpan.RecordError(err)
+		return nil, apifmt.Errorf("failed to list jobs with expired leases: %w", err)
+	}
+
+	listSpan.SetAttributes(attribute.Int("jobs_found", len(jobList.Items)))
+	return jobList.Items, nil
+}
+
+// cleanUpExpiredJob marks a single expired job as failed and archives it.
+func (s *persistentStore) cleanUpExpiredJob(ctx context.Context, job provisioning.Job) error {
+	// Calculate how long the job has been expired
+	var expiredFor time.Duration
+	var claimTimestamp time.Time
+	if claimTime, exists := job.Labels[LabelJobClaim]; exists {
+		claimMillis, parseErr := strconv.ParseInt(claimTime, 10, 64)
+		if parseErr == nil {
+			claimTimestamp = time.UnixMilli(claimMillis)
+			expiredFor = s.clock().Sub(claimTimestamp)
+		}
+	}
+
+	logger := logging.FromContext(ctx).With(
+		"job", job.GetName(),
+		"namespace", job.GetNamespace(),
+		"repository", job.Spec.Repository,
+		"action", job.Spec.Action,
+		"expired_for", expiredFor,
+	)
+
+	if !claimTimestamp.IsZero() {
+		logger = logger.With("claim_time", claimTimestamp.Format(time.RFC3339))
+	}
+
+	jobCtx, jobSpan := tracing.Start(ctx, "provisioning.jobs.cleanup.complete_expired_job")
+	defer jobSpan.End()
+
+	jobSpan.SetAttributes(
+		attribute.String("job.name", job.GetName()),
+		attribute.String("job.namespace", job.GetNamespace()),
+		attribute.String("job.repository", job.Spec.Repository),
+		attribute.String("job.action", string(job.Spec.Action)),
+		attribute.String("job.expired_for", expiredFor.String()),
+	)
+
+	// Mark job as failed due to lease expiry and archive it
+	jobCopy := job.DeepCopy()
+	jobCopy.Status.State = provisioning.JobStateError
+	jobCopy.Status.Message = "Job failed due to lease expiry - worker may have crashed or lost connection"
+
+	// Set namespace context for the completion
+	jobCtx, _, err := identity.WithProvisioningIdentity(jobCtx, job.GetNamespace())
+	if err != nil {
+		jobSpan.RecordError(err)
+		return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err)
+	}
+
+	// Use Complete to properly archive the failed job
+	if err := s.Complete(jobCtx, jobCopy); err != nil {
+		if apierrors.IsNotFound(err) {
+			// Job was already completed/deleted by another process - this is expected
+			logger.Warn("job already completed or deleted by another process")
+			return nil
+		}
+		jobSpan.RecordError(err)
+		return apifmt.Errorf("failed to complete expired job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
+	}
+
+	logger.Info("clean up expired job complete")
 	return nil
 }
 
 func (s *persistentStore) Insert(ctx context.Context, namespace string, spec provisioning.JobSpec) (*provisioning.Job, error) {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.insert")
+	defer span.End()
+
+	logger := logging.FromContext(ctx).With(
+		"operation", "insert",
+		"namespace", namespace,
+		"repository", spec.Repository,
+		"action", spec.Action,
+	)
+
+	span.SetAttributes(
+		attribute.String("job.namespace", namespace),
+		attribute.String("job.repository", spec.Repository),
+		attribute.String("job.action", string(spec.Action)),
+	)
+
 	if spec.Repository == "" {
-		return nil, errors.New("missing repository in job")
+		err := errors.New("missing repository in job")
+		span.RecordError(err)
+		return nil, err
 	}
 
 	// Set up the provisioning identity for this namespace
 	ctx, _, err := identity.WithProvisioningIdentity(ctx, namespace)
 	if err != nil {
+		span.RecordError(err)
 		return nil, apifmt.Errorf("failed to get provisioning identity for '%s': %w", namespace, err)
 	}
 
@@ -376,19 +603,27 @@ func (s *persistentStore) Insert(ctx context.Context, namespace string, spec pro
 		Spec: spec,
 	}
 	if err := mutateJobAction(job); err != nil {
+		span.RecordError(err)
 		return nil, err
 	}
 	generateJobName(job) // Side-effect: updates the job's name.
+
+	logger = logger.With("job", job.GetName())
+	span.SetAttributes(attribute.String("job.name", job.GetName()))
+
 	created, err := s.client.Jobs(namespace).Create(ctx, job, metav1.CreateOptions{})
 	if apierrors.IsAlreadyExists(err) {
+		span.RecordError(err)
 		return nil, apifmt.Errorf("job '%s' in '%s' already exists: %w", job.GetName(), job.GetNamespace(), err)
 	}
 	if err != nil {
+		span.RecordError(err)
 		return nil, apifmt.Errorf("failed to create job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
 	}
 
 	s.queueMetrics.IncreaseQueueSize(string(job.Spec.Action))
 
+	logger.Info("insert job complete")
 	return created, nil
 }
 

pkg/registry/apis/provisioning/webhooks/webhook.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
@@ -17,6 +18,7 @@ import (
 	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
 	"github.com/grafana/grafana/apps/provisioning/pkg/repository"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/infra/tracing"
 	provisioningapis "github.com/grafana/grafana/pkg/registry/apis/provisioning"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/webhooks/pullrequest"
 	"github.com/prometheus/client_golang/prometheus"
@@ -122,8 +124,16 @@ func (s *webhookConnector) Connect(ctx context.Context, name string, opts runtim
 	}
 
 	return provisioningapis.WithTimeout(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		logger := logging.FromContext(r.Context()).With("logger", "webhook-connector", "repo", name)
-		ctx := logging.Context(r.Context(), logger)
+		ctx, span := tracing.Start(r.Context(), "provisioning.webhook.handle")
+		defer span.End()
+
+		span.SetAttributes(
+			attribute.String("repository", name),
+			attribute.String("namespace", namespace),
+		)
+
+		logger := logging.FromContext(ctx).With("logger", "webhook-connector", "repo", name)
+		ctx = logging.Context(ctx, logger)
 		if !s.webhooksEnabled {
 			responder.Error(errors.NewBadRequest("webhooks are not enabled"))
 			return
@@ -140,12 +150,15 @@ func (s *webhookConnector) Connect(ctx context.Context, name string, opts runtim
 
 		rsp, err := hooks.Webhook(ctx, r)
 		if err != nil {
+			span.RecordError(err)
 			responder.Error(err)
 			return
 		}
 
 		if rsp == nil {
-			responder.Error(fmt.Errorf("expecting a response"))
+			err := fmt.Errorf("expecting a response")
+			span.RecordError(err)
+			responder.Error(err)
 			return
 		}
 
@@ -162,12 +175,17 @@ func (s *webhookConnector) Connect(ctx context.Context, name string, opts runtim
 		if rsp.Job != nil {
 			rsp.Job.Repository = name
 			actionTaken = string(rsp.Job.Action)
+			span.SetAttributes(attribute.String("job.action", actionTaken))
+
 			job, err := s.core.GetJobQueue().Insert(ctx, namespace, *rsp.Job)
 			if err != nil {
+				span.RecordError(err)
 				logger.Error("failed to insert job", "error", err)
 				responder.Error(err)
 				return
 			}
+			span.SetAttributes(attribute.String("job.name", job.Name))
+			logger.Info("webhook job created", "job", job.Name, "action", actionTaken)
 			responder.Object(rsp.Code, job)
 			return
 		}