From 30742df6762417effe747a9c97e159512835ee06 Mon Sep 17 00:00:00 2001 From: Roberto Jimenez Sanchez Date: Tue, 5 Aug 2025 12:01:51 +0200 Subject: [PATCH] Add concurrent job processing --- .../provisioning/jobs/concurrent_driver.go | 136 ++++++++++++++++++ pkg/registry/apis/provisioning/jobs/driver.go | 49 ++----- pkg/registry/apis/provisioning/register.go | 17 ++- 3 files changed, 162 insertions(+), 40 deletions(-) create mode 100644 pkg/registry/apis/provisioning/jobs/concurrent_driver.go diff --git a/pkg/registry/apis/provisioning/jobs/concurrent_driver.go b/pkg/registry/apis/provisioning/jobs/concurrent_driver.go new file mode 100644 index 00000000000..23868d8c439 --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/concurrent_driver.go @@ -0,0 +1,136 @@ +package jobs + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/grafana/grafana-app-sdk/logging" +) + +// ConcurrentJobDriver manages multiple jobDriver instances for concurrent job processing. +type ConcurrentJobDriver struct { + numDrivers int + jobTimeout time.Duration + cleanupInterval time.Duration + jobInterval time.Duration + store Store + repoGetter RepoGetter + historicJobs History + workers []Worker +} + +// NewConcurrentJobDriver creates a new concurrent job driver that spawns multiple job drivers. +func NewConcurrentJobDriver( + numDrivers int, + jobTimeout, cleanupInterval, jobInterval time.Duration, + store Store, + repoGetter RepoGetter, + historicJobs History, + workers ...Worker, +) (*ConcurrentJobDriver, error) { + if numDrivers <= 0 { + return nil, fmt.Errorf("numWorkers must be greater than 0, got %d", numDrivers) + } + if cleanupInterval < jobTimeout { + return nil, fmt.Errorf("the cleanup interval must be larger than the jobTimeout (cleanup:%s < job:%s)", + cleanupInterval.String(), jobTimeout.String()) + } + return &ConcurrentJobDriver{ + numDrivers: numDrivers, + jobTimeout: jobTimeout, + cleanupInterval: cleanupInterval, + jobInterval: jobInterval, + store: store, + repoGetter: repoGetter, + historicJobs: historicJobs, + workers: workers, + }, nil +} + +// 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. +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") + + // Set up cleanup ticker - only one cleanup process for all workers + cleanupTicker := time.NewTicker(c.cleanupInterval) + defer cleanupTicker.Stop() + + // Initial cleanup + if err := c.store.Cleanup(ctx); err != nil { + logger.Error("failed to clean up old jobs at start", "error", err) + } + + var wg sync.WaitGroup + errChan := make(chan error, c.numDrivers+1) // +1 for cleanup goroutine + + // Start cleanup goroutine + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-cleanupTicker.C: + if err := c.store.Cleanup(ctx); err != nil { + logger.Error("failed to cleanup jobs", "error", err) + } + case <-ctx.Done(): + logger.Debug("cleanup goroutine stopping") + return + } + } + }() + + // Start driver goroutines + for i := 0; i < c.numDrivers; i++ { + wg.Add(1) + go func(driverID int) { + defer wg.Done() + + driverLogger := logger.With("driver_id", driverID) + driverCtx := logging.Context(ctx, driverLogger) + + driver, err := NewJobDriver( + c.jobTimeout, + c.jobInterval, + c.store, + c.repoGetter, + c.historicJobs, + c.workers..., + ) + if err != nil { + driverLogger.Error("failed to create job driver", "error", err) + errChan <- err + return + } + + driverLogger.Debug("starting job driver") + if err := driver.Run(driverCtx); err != nil { + driverLogger.Error("job driver failed", "error", err) + errChan <- err + return + } + driverLogger.Debug("job driver stopped") + }(i) + } + + // Wait for all drivers to finish + go func() { + wg.Wait() + close(errChan) + }() + + // Return the first error encountered, if any + for err := range errChan { + if err != nil { + logger.Error("concurrent job driver error", "error", err) + return err + } + } + + logger.Info("all job driver workers stopped") + return ctx.Err() +} diff --git a/pkg/registry/apis/provisioning/jobs/driver.go b/pkg/registry/apis/provisioning/jobs/driver.go index 988422aacff..39c1e7b083c 100644 --- a/pkg/registry/apis/provisioning/jobs/driver.go +++ b/pkg/registry/apis/provisioning/jobs/driver.go @@ -3,7 +3,6 @@ package jobs import ( "context" "errors" - "fmt" "time" "k8s.io/apiserver/pkg/endpoints/request" @@ -47,14 +46,11 @@ var _ Store = (*persistentStore)(nil) // jobDriver drives jobs to completion and manages the job queue. // There may be multiple jobDrivers running in parallel. -// The jobDriver deals with cleaning up upon death and ensuring that jobs remain claimable. +// The jobDriver processes jobs but does not handle cleanup - that's handled by ConcurrentJobDriver. type jobDriver struct { // Timeout for processing a job. This must be less than a claim expiry. jobTimeout time.Duration - // CleanupInterval is the time between cleanup runs. - cleanupInterval time.Duration - // JobInterval is the time between job ticks. This should be relatively low. jobInterval time.Duration @@ -72,34 +68,26 @@ type jobDriver struct { } func NewJobDriver( - jobTimeout, cleanupInterval, jobInterval time.Duration, + jobTimeout, jobInterval time.Duration, store Store, repoGetter RepoGetter, historicJobs History, workers ...Worker, ) (*jobDriver, error) { - if cleanupInterval < jobTimeout { - return nil, fmt.Errorf("the cleanup interval must be larger than the jobTimeout (cleanup:%s < job:%s)", - cleanupInterval.String(), jobTimeout.String()) - } return &jobDriver{ - jobTimeout: jobTimeout, - cleanupInterval: cleanupInterval, - jobInterval: jobInterval, - store: store, - repoGetter: repoGetter, - historicJobs: historicJobs, - workers: workers, + jobTimeout: jobTimeout, + jobInterval: jobInterval, + store: store, + repoGetter: repoGetter, + historicJobs: historicJobs, + workers: workers, }, nil } // Run drives jobs to completion. This is a blocking function. -// It will run until the context is canceled. +// 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. -func (d *jobDriver) Run(ctx context.Context) { - cleanupTicker := time.NewTicker(d.cleanupInterval) - defer cleanupTicker.Stop() - +func (d *jobDriver) Run(ctx context.Context) error { jobTicker := time.NewTicker(d.jobInterval) defer jobTicker.Stop() @@ -107,13 +95,7 @@ func (d *jobDriver) Run(ctx context.Context) { ctx = logging.Context(ctx, logger) ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") // "*" grants us access to all namespaces. if err != nil { - logger.Error("failed to grant provisioning identity; this will panic!", "error", err) - panic("unreachable?: failed to grant provisioning identity: " + err.Error()) - } - - // Remove old jobs - if err = d.store.Cleanup(ctx); err != nil { - logger.Error("failed to clean up old jobs at start", "error", err) + return apifmt.Errorf("failed to grant provisioning identity: %w", err) } // Drive without waiting on startup. @@ -121,12 +103,8 @@ func (d *jobDriver) Run(ctx context.Context) { for { select { - case <-cleanupTicker.C: - if err := d.store.Cleanup(ctx); err != nil { - logger.Error("failed to cleanup jobs", "error", err) - } - - // These events do not queue if the worker is already running + case <-ctx.Done(): + return ctx.Err() case <-jobTicker.C: d.processJobsUntilDoneOrError(ctx) case <-d.store.InsertNotifications(): @@ -240,3 +218,4 @@ func (d *jobDriver) onProgress(job *provisioning.Job) ProgressFn { return nil } } + diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 48656eba416..f2b00ef4085 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -633,17 +633,24 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH workers = append(workers, extra.GetJobWorkers()...) } - driver, err := jobs.NewJobDriver( - time.Minute*20, // Max time for each job - time.Minute*22, // Cleanup any checked out jobs. FIXME: this is slow if things crash/fail! - time.Second*30, // Periodically look for new jobs + // This is basically our own JobQueue system + driver, err := jobs.NewConcurrentJobDriver( + 3, // 3 drivers for now + 20*time.Minute, // Max time for each job + 22*time.Minute, // Cleanup any checked out jobs. FIXME: this is slow if things crash/fail! + 30*time.Second, // Periodically look for new jobs b.jobs, b, b.jobHistory, workers..., ) if err != nil { return err } - go driver.Run(postStartHookCtx.Context) + + go func() { + if err := driver.Run(postStartHookCtx.Context); err != nil { + logging.FromContext(postStartHookCtx.Context).Error("job driver failed", "error", err) + } + }() repoController, err := controller.NewRepositoryController( b.GetClient(),