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

Provisioning: introduce jobs controller (#109252)

Browse Source 
* Add basic job controller
* Replace the existing in-memory channel
This commit is contained in:
Roberto Jiménez Sánchez
2025-08-07 12:51:29 +03:00
committed by GitHub
parent 81531dcd7b
commit c82d2af867
7 changed files with 166 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/provisioning/controller/job.go
+58
View File
@@ -0,0 +1,58 @@
package controller
import (
"k8s.io/client-go/tools/cache"
"github.com/grafana/grafana-app-sdk/logging"
informer "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions/provisioning/v0alpha1"
)
const (
jobControllerLoggerName = "provisioning-job-controller"
)
// JobController manages job create notifications.
type JobController struct {
jobSynced cache.InformerSynced
logger logging.Logger
// notification channel for job create events (replaces InsertNotifications)
notifications chan struct{}
}
// NewJobController creates a new JobController.
func NewJobController(
jobInformer informer.JobInformer,
) (*JobController, error) {
jc := &JobController{
jobSynced: jobInformer.Informer().HasSynced,
logger: logging.DefaultLogger.With("logger", jobControllerLoggerName),
notifications: make(chan struct{}, 1),
}
_, err := jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
// Send notification for job create events (replaces InsertNotifications)
jc.sendNotification()
},
})
if err != nil {
return nil, err
}
return jc, nil
}
// InsertNotifications returns a channel that receives notifications when jobs are created.
// This replaces the InsertNotifications method from persistentstore.go.
func (jc *JobController) InsertNotifications() chan struct{} {
return jc.notifications
}
func (jc *JobController) sendNotification() {
select {
case jc.notifications <- struct{}{}:
default:
// Don't block if there's already a notification waiting
}
}
pkg/registry/apis/provisioning/controller/job_test.go
+85
View File
@@ -0,0 +1,85 @@
package controller
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
provisioningfake "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/fake"
provisioninginformers "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions"
)
func TestJobController_New(t *testing.T) {
client := provisioningfake.NewSimpleClientset()
informerFactory := provisioninginformers.NewSharedInformerFactory(client, time.Second)
jobInformer := informerFactory.Provisioning().V0alpha1().Jobs()
controller, err := NewJobController(jobInformer)
require.NoError(t, err)
assert.NotNil(t, controller)
assert.NotNil(t, controller.notifications)
}
func TestJobController_InsertNotifications(t *testing.T) {
client := provisioningfake.NewSimpleClientset()
informerFactory := provisioninginformers.NewSharedInformerFactory(client, time.Second)
jobInformer := informerFactory.Provisioning().V0alpha1().Jobs()
controller, err := NewJobController(jobInformer)
require.NoError(t, err)
notifications := controller.InsertNotifications()
assert.NotNil(t, notifications)
// Test that notification is sent
controller.sendNotification()
select {
case <-notifications:
// Success - notification received
case <-time.After(time.Second):
t.Fatal("Expected notification but didn't receive one")
}
}
func TestJobController_NotificationOnJobCreate(t *testing.T) {
client := provisioningfake.NewSimpleClientset()
informerFactory := provisioninginformers.NewSharedInformerFactory(client, time.Second)
jobInformer := informerFactory.Provisioning().V0alpha1().Jobs()
controller, err := NewJobController(jobInformer)
require.NoError(t, err)
// Start informer and wait for cache sync
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
defer cancel()
informerFactory.Start(ctx.Done())
informerFactory.WaitForCacheSync(ctx.Done())
// Get notifications channel
notifications := controller.InsertNotifications()
// Create a job - this should trigger a notification
_, err = client.ProvisioningV0alpha1().Jobs("default").Create(ctx, &provisioning.Job{
ObjectMeta: metav1.ObjectMeta{
Name: "test-job",
Namespace: "default",
},
}, metav1.CreateOptions{})
require.NoError(t, err)
// Wait for notification
select {
case <-notifications:
// Success - notification received
case <-time.After(time.Second * 2):
t.Fatal("Expected notification but didn't receive one")
}
}
pkg/registry/apis/provisioning/jobs/concurrent_driver.go
+4
View File
@@ -20,6 +20,7 @@ type ConcurrentJobDriver struct {
repoGetter RepoGetter
historicJobs History
workers []Worker
notifications chan struct{}
}
// NewConcurrentJobDriver creates a new concurrent job driver that spawns multiple job drivers.
@@ -29,6 +30,7 @@ func NewConcurrentJobDriver(
store Store,
repoGetter RepoGetter,
historicJobs History,
notifications chan struct{},
workers ...Worker,
) (*ConcurrentJobDriver, error) {
if numDrivers <= 0 {
@@ -63,6 +65,7 @@ func NewConcurrentJobDriver(
repoGetter: repoGetter,
historicJobs: historicJobs,
workers: workers,
notifications: notifications,
}, nil
}
@@ -117,6 +120,7 @@ func (c *ConcurrentJobDriver) Run(ctx context.Context) error {
c.store,
c.repoGetter,
c.historicJobs,
c.notifications,
c.workers...,
)
if err != nil {
pkg/registry/apis/provisioning/jobs/driver.go
+6 -5
View File
@@ -36,10 +36,6 @@ type Store interface {
// An abandoned job is one that has been claimed by a worker, but the worker has not updated the job in a while.
Cleanup(ctx context.Context) error
// InsertNotifications returns a channel that will have a value sent to it when a new job is inserted.
// This is used to wake up the job driver when a new job is inserted.
InsertNotifications() chan struct{}
// Update saves the job back to the store.
Update(ctx context.Context, job *provisioning.Job) (*provisioning.Job, error)
@@ -77,6 +73,9 @@ type jobDriver struct {
// Workers process the job.
// Only the first worker who supports the job will process it; the rest are ignored.
workers []Worker
// notifications channel for job create events
notifications chan struct{}
}
func NewJobDriver(
@@ -84,6 +83,7 @@ func NewJobDriver(
store Store,
repoGetter RepoGetter,
historicJobs History,
notifications chan struct{},
workers ...Worker,
) (*jobDriver, error) {
return &jobDriver{
@@ -94,6 +94,7 @@ func NewJobDriver(
repoGetter: repoGetter,
historicJobs: historicJobs,
workers: workers,
notifications: notifications,
}, nil
}
@@ -120,7 +121,7 @@ func (d *jobDriver) Run(ctx context.Context) error {
return ctx.Err()
case <-jobTicker.C:
d.processJobsUntilDoneOrError(ctx)
case <-d.store.InsertNotifications():
case <-d.notifications:
d.processJobsUntilDoneOrError(ctx)
}
}
pkg/registry/apis/provisioning/jobs/persistentstore.go
+3 -24
View File
@@ -87,28 +87,17 @@ type persistentStore struct {
// expiry is the time after which a job is considered abandoned.
// If a job is abandoned, it will have its claim cleaned up periodically.
expiry time.Duration
// notifications has a signal sent to it when a new job is inserted. If a value already exists, nothing is sent.
//
// This is very similar to the concept of a Waker in Rust: <https://doc.rust-lang.org/std/task/struct.Waker.html>
notifications chan struct{}
}
func NewJobStore(
jobStore jobStorage,
expiry time.Duration,
) (*persistentStore, error) {
func NewJobStore(jobStore jobStorage, expiry time.Duration) (*persistentStore, error) {
if expiry <= 0 {
expiry = time.Second * 30
}
return &persistentStore{
jobStore: jobStore,
clock: time.Now,
expiry: expiry,
notifications: make(chan struct{}, 1),
clock: time.Now,
expiry: expiry,
}, nil
}
@@ -446,19 +435,9 @@ func (s *persistentStore) Insert(ctx context.Context, namespace string, spec pro
return nil, apifmt.Errorf("unexpected object type %T", obj)
}
select {
case s.notifications <- struct{}{}:
default:
// We don't want to block if there is already a notification waiting.
}
return created, nil
}
func (s *persistentStore) InsertNotifications() chan struct{} {
return s.notifications
}
// generateJobName creates and updates the job's name to one that fits it.
func (s *persistentStore) generateJobName(job *provisioning.Job) {
switch job.Spec.Action {
pkg/registry/apis/provisioning/jobs/store_mock.go
+1 -48
View File
@@ -1,4 +1,4 @@
// Code generated by mockery v2.53.4. DO NOT EDIT.
// Code generated by mockery v2.52.4. DO NOT EDIT.
package jobs
@@ -182,53 +182,6 @@ func (_c *MockStore_Complete_Call) RunAndReturn(run func(context.Context, *v0alp
return _c
}
// InsertNotifications provides a mock function with no fields
func (_m *MockStore) InsertNotifications() chan struct{} {
ret := _m.Called()
if len(ret) == 0 {
panic("no return value specified for InsertNotifications")
}
var r0 chan struct{}
if rf, ok := ret.Get(0).(func() chan struct{}); ok {
r0 = rf()
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(chan struct{})
}
}
return r0
}
// MockStore_InsertNotifications_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'InsertNotifications'
type MockStore_InsertNotifications_Call struct {
*mock.Call
}
// InsertNotifications is a helper method to define mock.On call
func (_e *MockStore_Expecter) InsertNotifications() *MockStore_InsertNotifications_Call {
return &MockStore_InsertNotifications_Call{Call: _e.mock.On("InsertNotifications")}
}
func (_c *MockStore_InsertNotifications_Call) Run(run func()) *MockStore_InsertNotifications_Call {
_c.Call.Run(func(args mock.Arguments) {
run()
})
return _c
}
func (_c *MockStore_InsertNotifications_Call) Return(_a0 chan struct{}) *MockStore_InsertNotifications_Call {
_c.Call.Return(_a0)
return _c
}
func (_c *MockStore_InsertNotifications_Call) RunAndReturn(run func() chan struct{}) *MockStore_InsertNotifications_Call {
_c.Call.Return(run)
return _c
}
// Update provides a mock function with given fields: ctx, job
func (_m *MockStore) Update(ctx context.Context, job *v0alpha1.Job) (*v0alpha1.Job, error) {
ret := _m.Called(ctx, job)
pkg/registry/apis/provisioning/register.go
+9
View File
@@ -603,7 +603,9 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
// Informer with resync interval used for health check and reconciliation
sharedInformerFactory := informers.NewSharedInformerFactory(c, 60*time.Second)
repoInformer := sharedInformerFactory.Provisioning().V0alpha1().Repositories()
jobInformer := sharedInformerFactory.Provisioning().V0alpha1().Jobs()
go repoInformer.Informer().Run(postStartHookCtx.Done())
go jobInformer.Informer().Run(postStartHookCtx.Done())
b.client = c.ProvisioningV0alpha1()
@@ -675,6 +677,12 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
syncWorker,
}
// Create JobController to handle job create notifications
jobController, err := controller.NewJobController(jobInformer)
if err != nil {
return err
}
// Add any extra workers
for _, extra := range b.extras {
workers = append(workers, extra.GetJobWorkers()...)
@@ -688,6 +696,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
30*time.Second, // Periodically look for new jobs
30*time.Second, // Lease renewal interval
b.jobs, b, b.jobHistory,
jobController.InsertNotifications(),
workers...,
)
if err != nil {