From cdc6a6114cfcc4ba773bca0810125d8581981ce7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roberto=20Jim=C3=A9nez=20S=C3=A1nchez?= Date: Wed, 12 Nov 2025 14:59:27 +0100 Subject: [PATCH] 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> --- .../provisioning/pkg/controller/historyjob.go | 52 +-- .../provisioning/controller/repository.go | 18 +- .../provisioning/jobs/concurrent_driver.go | 23 +- pkg/registry/apis/provisioning/jobs/driver.go | 98 +++++- .../apis/provisioning/jobs/persistentstore.go | 323 +++++++++++++++--- .../apis/provisioning/webhooks/webhook.go | 24 +- 6 files changed, 444 insertions(+), 94 deletions(-) diff --git a/apps/provisioning/pkg/controller/historyjob.go b/apps/provisioning/pkg/controller/historyjob.go index 2e4e13f359d..bdfefb0b613 100644 --- a/apps/provisioning/pkg/controller/historyjob.go +++ b/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") } diff --git a/pkg/registry/apis/provisioning/controller/repository.go b/pkg/registry/apis/provisioning/controller/repository.go index b0ba13385b6..ffeb3f1fb9f 100644 --- a/pkg/registry/apis/provisioning/controller/repository.go +++ b/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 } diff --git a/pkg/registry/apis/provisioning/jobs/concurrent_driver.go b/pkg/registry/apis/provisioning/jobs/concurrent_driver.go index 1fdc32bea60..dce8d65a97e 100644 --- a/pkg/registry/apis/provisioning/jobs/concurrent_driver.go +++ b/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") } diff --git a/pkg/registry/apis/provisioning/jobs/driver.go b/pkg/registry/apis/provisioning/jobs/driver.go index 7902aad12fc..b1cb0ee6da1 100644 --- a/pkg/registry/apis/provisioning/jobs/driver.go +++ b/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 } } diff --git a/pkg/registry/apis/provisioning/jobs/persistentstore.go b/pkg/registry/apis/provisioning/jobs/persistentstore.go index 78494aa82cc..b7c7e47bd35 100644 --- a/pkg/registry/apis/provisioning/jobs/persistentstore.go +++ b/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 } diff --git a/pkg/registry/apis/provisioning/webhooks/webhook.go b/pkg/registry/apis/provisioning/webhooks/webhook.go index 4e5bfffb28a..859437f1dda 100644 --- a/pkg/registry/apis/provisioning/webhooks/webhook.go +++ b/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 }