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

Provisioning: Refactor tweaks to support MT controllers (#110581)

Browse Source 
* Refactor common code to support MT controllers

* Delete original status files
This commit is contained in:
Roberto Jiménez Sánchez
2025-09-04 10:06:50 +00:00
committed by GitHub
parent 55b638ea98
commit 7d630ec3b1
11 changed files with 82 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/provisioning/controller/status.go → apps/provisioning/pkg/controller/status.go
View File
pkg/registry/apis/provisioning/controller/status_test.go → apps/provisioning/pkg/controller/status_test.go
View File
pkg/registry/apis/provisioning/jobs/driver.go
+1 -1
View File
@@ -276,7 +276,7 @@ func (d *jobDriver) processJob(ctx context.Context, job *provisioning.Job, recor
continue
}
repo, err := d.repoGetter.GetRepository(ctx, job.Spec.Repository)
repo, err := d.repoGetter.GetRepository(ctx, job.Namespace, job.Spec.Repository)
if err != nil {
return apifmt.Errorf("failed to get repository '%s': %w", job.Spec.Repository, err)
}
pkg/registry/apis/provisioning/jobs/migrate/worker.go
+11 -2
View File
@@ -21,6 +21,13 @@ type MigrationWorker struct {
unifiedMigrator Migrator
}
func NewMigrationWorkerFromUnified(unifiedMigrator Migrator) *MigrationWorker {
return &MigrationWorker{
unifiedMigrator: unifiedMigrator,
}
}
// HACK: we should decouple the implementation of these two
func NewMigrationWorker(
legacyMigrator Migrator,
unifiedMigrator Migrator,
@@ -57,12 +64,14 @@ func (w *MigrationWorker) Process(ctx context.Context, repo repository.Repositor
// Block migrate for legacy resources if repository type is folder
if repo.Config().Spec.Sync.Target == provisioning.SyncTargetTypeFolder {
if dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, w.storageStatus) {
// HACK: we should not have to check for storage existence here
if w.storageStatus != nil && dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, w.storageStatus) {
return errors.New("migration of legacy resources is not supported for folder-type repositories")
}
}
if dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, w.storageStatus) {
// HACK: we should not have to check for storage existence here
if w.storageStatus != nil && dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, w.storageStatus) {
return w.legacyMigrator.Migrate(ctx, rw, *options, progress)
}
pkg/registry/apis/provisioning/jobs/queue.go
+1 -1
View File
@@ -11,7 +11,7 @@ import (
//
//go:generate mockery --name RepoGetter --structname MockRepoGetter --inpackage --filename repo_getter_mock.go --with-expecter
type RepoGetter interface {
GetRepository(ctx context.Context, name string) (repository.Repository, error)
GetRepository(ctx context.Context, namespace, name string) (repository.Repository, error)
}
// JobProgressRecorder is a function that can be called to record the progress of a job
pkg/registry/apis/provisioning/jobs/repo_getter_mock.go
+16 -15
View File
@@ -1,4 +1,4 @@
// Code generated by mockery v2.52.4. DO NOT EDIT.
// Code generated by mockery v2.53.4. DO NOT EDIT.
package jobs
@@ -22,9 +22,9 @@ func (_m *MockRepoGetter) EXPECT() *MockRepoGetter_Expecter {
return &MockRepoGetter_Expecter{mock: &_m.Mock}
}
// GetRepository provides a mock function with given fields: ctx, name
func (_m *MockRepoGetter) GetRepository(ctx context.Context, name string) (repository.Repository, error) {
ret := _m.Called(ctx, name)
// GetRepository provides a mock function with given fields: ctx, namespace, name
func (_m *MockRepoGetter) GetRepository(ctx context.Context, namespace string, name string) (repository.Repository, error) {
ret := _m.Called(ctx, namespace, name)
if len(ret) == 0 {
panic("no return value specified for GetRepository")
@@ -32,19 +32,19 @@ func (_m *MockRepoGetter) GetRepository(ctx context.Context, name string) (repos
var r0 repository.Repository
var r1 error
if rf, ok := ret.Get(0).(func(context.Context, string) (repository.Repository, error)); ok {
return rf(ctx, name)
if rf, ok := ret.Get(0).(func(context.Context, string, string) (repository.Repository, error)); ok {
return rf(ctx, namespace, name)
}
if rf, ok := ret.Get(0).(func(context.Context, string) repository.Repository); ok {
r0 = rf(ctx, name)
if rf, ok := ret.Get(0).(func(context.Context, string, string) repository.Repository); ok {
r0 = rf(ctx, namespace, name)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(repository.Repository)
}
}
if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
r1 = rf(ctx, name)
if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
r1 = rf(ctx, namespace, name)
} else {
r1 = ret.Error(1)
}
@@ -59,14 +59,15 @@ type MockRepoGetter_GetRepository_Call struct {
// GetRepository is a helper method to define mock.On call
// - ctx context.Context
// - namespace string
// - name string
func (_e *MockRepoGetter_Expecter) GetRepository(ctx interface{}, name interface{}) *MockRepoGetter_GetRepository_Call {
return &MockRepoGetter_GetRepository_Call{Call: _e.mock.On("GetRepository", ctx, name)}
func (_e *MockRepoGetter_Expecter) GetRepository(ctx interface{}, namespace interface{}, name interface{}) *MockRepoGetter_GetRepository_Call {
return &MockRepoGetter_GetRepository_Call{Call: _e.mock.On("GetRepository", ctx, namespace, name)}
}
func (_c *MockRepoGetter_GetRepository_Call) Run(run func(ctx context.Context, name string)) *MockRepoGetter_GetRepository_Call {
func (_c *MockRepoGetter_GetRepository_Call) Run(run func(ctx context.Context, namespace string, name string)) *MockRepoGetter_GetRepository_Call {
_c.Call.Run(func(args mock.Arguments) {
run(args[0].(context.Context), args[1].(string))
run(args[0].(context.Context), args[1].(string), args[2].(string))
})
return _c
}
@@ -76,7 +77,7 @@ func (_c *MockRepoGetter_GetRepository_Call) Return(_a0 repository.Repository, _
return _c
}
func (_c *MockRepoGetter_GetRepository_Call) RunAndReturn(run func(context.Context, string) (repository.Repository, error)) *MockRepoGetter_GetRepository_Call {
func (_c *MockRepoGetter_GetRepository_Call) RunAndReturn(run func(context.Context, string, string) (repository.Repository, error)) *MockRepoGetter_GetRepository_Call {
_c.Call.Return(run)
return _c
}
pkg/registry/apis/provisioning/jobs/sync/worker.go
+2 -1
View File
@@ -58,7 +58,8 @@ func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, jo
cfg := repo.Config()
logger := logging.FromContext(ctx).With("job", job.GetName(), "namespace", job.GetNamespace())
// Check if we are onboarding from legacy storage
if dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, r.storageStatus) {
// HACK -- this should be handled outside of this worker
if r.storageStatus != nil && dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, r.storageStatus) {
return fmt.Errorf("sync not supported until storage has migrated")
}
pkg/registry/apis/provisioning/register.go
+6 -5
View File
@@ -105,7 +105,7 @@ type APIBuilder struct {
repoFactory repository.Factory
client client.ProvisioningV0alpha1Interface
access authlib.AccessChecker
statusPatcher *controller.RepositoryStatusPatcher
statusPatcher *appcontroller.RepositoryStatusPatcher
healthChecker *controller.HealthChecker
// Extras provides additional functionality to the API.
extras []Extra
@@ -130,7 +130,7 @@ func NewAPIBuilder(
) *APIBuilder {
clients := resources.NewClientFactory(configProvider)
parsers := resources.NewParserFactory(clients)
resourceLister := resources.NewResourceLister(unified, unified, legacyMigrator, storageStatus)
resourceLister := resources.NewResourceListerForMigrations(unified, legacyMigrator, storageStatus)
b := &APIBuilder{
onlyApiServer: onlyApiServer,
@@ -363,7 +363,7 @@ func (b *APIBuilder) GetJobQueue() jobs.Queue {
return b.jobs
}
func (b *APIBuilder) GetStatusPatcher() *controller.RepositoryStatusPatcher {
func (b *APIBuilder) GetStatusPatcher() *appcontroller.RepositoryStatusPatcher {
return b.statusPatcher
}
@@ -636,7 +636,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
return fmt.Errorf("create API client job store: %w", err)
}
b.statusPatcher = controller.NewRepositoryStatusPatcher(b.GetClient())
b.statusPatcher = appcontroller.NewRepositoryStatusPatcher(b.GetClient())
b.healthChecker = controller.NewHealthChecker(&repository.Tester{}, b.statusPatcher)
// if running solely CRUD, skip the rest of the setup
@@ -731,6 +731,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
jobHistoryWriter = jobs.NewAPIClientHistoryWriter(b.GetClient())
}
repoGetter := resources.NewRepositoryGetter(b.repoFactory, b.client)
// This is basically our own JobQueue system
driver, err := jobs.NewConcurrentJobDriver(
3, // 3 drivers for now
@@ -738,7 +739,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
time.Minute, // Cleanup jobs
30*time.Second, // Periodically look for new jobs
30*time.Second, // Lease renewal interval
b.jobs, b, jobHistoryWriter,
b.jobs, repoGetter, jobHistoryWriter,
jobController.InsertNotifications(),
workers...,
)
pkg/registry/apis/provisioning/resources/object.go
+18 -11
View File
@@ -24,22 +24,29 @@ type ResourceLister interface {
Stats(ctx context.Context, namespace, repository string) (*provisioning.ResourceStats, error)
}
type ResourceStore interface {
resourcepb.ManagedObjectIndexClient
resourcepb.ResourceIndexClient
}
type ResourceListerFromSearch struct {
managed resourcepb.ManagedObjectIndexClient
index resourcepb.ResourceIndexClient
store ResourceStore
legacyMigrator legacy.LegacyMigrator
storageStatus dualwrite.Service
}
func NewResourceLister(
managed resourcepb.ManagedObjectIndexClient,
index resourcepb.ResourceIndexClient,
func NewResourceLister(store ResourceStore) ResourceLister {
return &ResourceListerFromSearch{store: store}
}
// FIXME: the logic about migration and storage should probably be separated from this
func NewResourceListerForMigrations(
store ResourceStore,
legacyMigrator legacy.LegacyMigrator,
storageStatus dualwrite.Service,
) ResourceLister {
return &ResourceListerFromSearch{
index: index,
managed: managed,
store: store,
legacyMigrator: legacyMigrator,
storageStatus: storageStatus,
}
@@ -47,7 +54,7 @@ func NewResourceLister(
// List implements ResourceLister.
func (o *ResourceListerFromSearch) List(ctx context.Context, namespace, repository string) (*provisioning.ResourceList, error) {
objects, err := o.managed.ListManagedObjects(ctx, &resourcepb.ListManagedObjectsRequest{
objects, err := o.store.ListManagedObjects(ctx, &resourcepb.ListManagedObjectsRequest{
Namespace: namespace,
Kind: string(utils.ManagerKindRepo),
Id: repository,
@@ -85,7 +92,7 @@ func (o *ResourceListerFromSearch) Stats(ctx context.Context, namespace, reposit
req.Id = repository
}
counts, err := o.managed.CountManagedObjects(ctx, req)
counts, err := o.store.CountManagedObjects(ctx, req)
if err != nil {
return nil, err
}
@@ -126,7 +133,7 @@ func (o *ResourceListerFromSearch) Stats(ctx context.Context, namespace, reposit
}
// Get the stats based on what a migration could support
if dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, o.storageStatus) {
if o.storageStatus != nil && o.legacyMigrator != nil && dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, o.storageStatus) {
rsp, err := o.legacyMigrator.Migrate(ctx, legacy.MigrateOptions{
Namespace: namespace,
Resources: []schema.GroupResource{{
@@ -157,7 +164,7 @@ func (o *ResourceListerFromSearch) Stats(ctx context.Context, namespace, reposit
}
// Get full instance stats
info, err := o.index.GetStats(ctx, &resourcepb.ResourceStatsRequest{
info, err := o.store.GetStats(ctx, &resourcepb.ResourceStatsRequest{
Namespace: namespace,
})
if err != nil {
pkg/registry/apis/provisioning/resources/repository.go
+25
View File
@@ -10,6 +10,7 @@ import (
"k8s.io/apimachinery/pkg/runtime/schema"
provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
provisioningv0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1"
"github.com/grafana/grafana/apps/provisioning/pkg/repository"
"github.com/grafana/grafana/apps/provisioning/pkg/safepath"
"github.com/grafana/grafana/pkg/apimachinery/utils"
@@ -126,3 +127,27 @@ func (r *repositoryResourcesFactory) Client(ctx context.Context, repo repository
repoName: repo.Config().Name,
}, nil
}
type RepositoryGetter struct {
factory repository.Factory
client provisioningv0alpha1.ProvisioningV0alpha1Interface
}
func NewRepositoryGetter(
factory repository.Factory,
client provisioningv0alpha1.ProvisioningV0alpha1Interface,
) *RepositoryGetter {
return &RepositoryGetter{
factory: factory,
client: client,
}
}
func (r *RepositoryGetter) GetRepository(ctx context.Context, namespace, repoName string) (repository.Repository, error) {
repo, err := r.client.Repositories(namespace).Get(ctx, repoName, metav1.GetOptions{})
if err != nil {
return nil, fmt.Errorf("get repository %q: %w", repoName, err)
}
return r.factory.Build(ctx, repo)
}
pkg/registry/apis/provisioning/test.go
+2 -1
View File
@@ -15,12 +15,13 @@ import (
"k8s.io/apiserver/pkg/registry/rest"
provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
appcontroller "github.com/grafana/grafana/apps/provisioning/pkg/controller"
"github.com/grafana/grafana/apps/provisioning/pkg/repository"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/controller"
)
type StatusPatcherProvider interface {
GetStatusPatcher() *controller.RepositoryStatusPatcher
GetStatusPatcher() *appcontroller.RepositoryStatusPatcher
}
type HealthCheckerProvider interface {