public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add concurrent job processing

Browse Source
This commit is contained in:
Roberto Jimenez Sanchez
2025-08-05 12:01:51 +02:00
parent ff10f6da67
commit 30742df676
3 changed files with 162 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/provisioning/jobs/concurrent_driver.go
+136
View File
@@ -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()
}
pkg/registry/apis/provisioning/jobs/driver.go
+14 -35
View File
@@ -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
}
}
pkg/registry/apis/provisioning/register.go
+12 -5
View File
@@ -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(),