From e34475041aec08ec4ae66940db785ac3d2303785 Mon Sep 17 00:00:00 2001 From: Daniele Stefano Ferru Date: Sun, 28 Sep 2025 03:14:08 +0200 Subject: [PATCH] Provisioning: move job operator to enterprise and enable PR worker (#111663) --- .gitignore | 3 + pkg/operators/provisioning/config.go | 24 +- pkg/operators/provisioning/jobs_operator.go | 243 ------------------ pkg/operators/register.go | 6 - .../webhooks/pullrequest/changes.go | 6 +- .../webhooks/pullrequest/changes_test.go | 2 +- .../webhooks/pullrequest/worker.go | 2 +- .../apis/provisioning/webhooks/register.go | 14 +- 8 files changed, 29 insertions(+), 271 deletions(-) delete mode 100644 pkg/operators/provisioning/jobs_operator.go diff --git a/.gitignore b/.gitignore index d6235ae4498..fb7d5d30a9a 100644 --- a/.gitignore +++ b/.gitignore @@ -130,6 +130,9 @@ profile.cov /public/app/extensions !/public/app/extensions/.keep +# Enterprise operators +/pkg/operators/enterprise_* +/pkg/operators/**/enterprise_* debug.test /examples/*/dist diff --git a/pkg/operators/provisioning/config.go b/pkg/operators/provisioning/config.go index 273070ac156..0cf3465823b 100644 --- a/pkg/operators/provisioning/config.go +++ b/pkg/operators/provisioning/config.go @@ -33,11 +33,13 @@ import ( // provisioningControllerConfig contains the configuration that overlaps for the jobs and repo controllers type provisioningControllerConfig struct { - provisioningClient *client.Clientset - resyncInterval time.Duration - repoFactory repository.Factory - unified resources.ResourceStore - clients resources.ClientFactory + provisioningClient *client.Clientset + resyncInterval time.Duration + repoFactory repository.Factory + unified resources.ResourceStore + clients resources.ClientFactory + tokenExchangeClient *authn.TokenExchangeClient + tlsConfig rest.TLSClientConfig } // expects: @@ -179,11 +181,13 @@ func setupFromConfig(cfg *setting.Cfg, registry prometheus.Registerer) (controll clients := resources.NewClientFactoryForMultipleAPIServers(configProviders) return &provisioningControllerConfig{ - provisioningClient: provisioningClient, - repoFactory: repoFactory, - unified: unified, - clients: clients, - resyncInterval: operatorSec.Key("resync_interval").MustDuration(60 * time.Second), + provisioningClient: provisioningClient, + repoFactory: repoFactory, + unified: unified, + clients: clients, + resyncInterval: operatorSec.Key("resync_interval").MustDuration(60 * time.Second), + tokenExchangeClient: tokenExchangeClient, + tlsConfig: tlsConfig, }, nil } diff --git a/pkg/operators/provisioning/jobs_operator.go b/pkg/operators/provisioning/jobs_operator.go deleted file mode 100644 index f901988e0e0..00000000000 --- a/pkg/operators/provisioning/jobs_operator.go +++ /dev/null @@ -1,243 +0,0 @@ -package provisioning - -import ( - "context" - "fmt" - "log/slog" - "os" - "os/signal" - "syscall" - "time" - - "github.com/grafana/grafana-app-sdk/logging" - "github.com/prometheus/client_golang/prometheus" - "k8s.io/client-go/tools/cache" - - "github.com/grafana/grafana/pkg/infra/tracing" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/export" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/migrate" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/move" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/sync" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" - "github.com/grafana/grafana/pkg/server" - "github.com/grafana/grafana/pkg/setting" - - "github.com/grafana/grafana/apps/provisioning/pkg/controller" - informer "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions" - "github.com/grafana/grafana/apps/provisioning/pkg/repository" - deletepkg "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/delete" -) - -func RunJobController(deps server.OperatorDependencies) error { - logger := logging.NewSLogLogger(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ - Level: slog.LevelDebug, - })).With("logger", "provisioning-job-controller") - logger.Info("Starting provisioning job controller") - - tracingConfig, err := tracing.ProvideTracingConfig(deps.Config) - if err != nil { - return fmt.Errorf("failed to provide tracing config: %w", err) - } - - tracer, err := tracing.ProvideService(tracingConfig) - if err != nil { - return fmt.Errorf("failed to provide tracing service: %w", err) - } - - controllerCfg, err := setupJobsControllerFromConfig(deps.Config, deps.Registerer) - if err != nil { - return fmt.Errorf("failed to setup operator: %w", err) - } - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - sigChan := make(chan os.Signal, 1) - signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) - go func() { - <-sigChan - fmt.Println("Received shutdown signal, stopping controllers") - cancel() - }() - - // Jobs informer and controller (resync ~60s like in register.go) - jobInformerFactory := informer.NewSharedInformerFactoryWithOptions( - controllerCfg.provisioningClient, - controllerCfg.resyncInterval, - ) - jobInformer := jobInformerFactory.Provisioning().V0alpha1().Jobs() - jobController, err := controller.NewJobController(jobInformer) - if err != nil { - return fmt.Errorf("failed to create job controller: %w", err) - } - - logger.Info("jobs controller started") - - var startHistoryInformers func() - if controllerCfg.historyExpiration > 0 { - // History jobs informer and controller (separate factory with resync == expiration) - historyInformerFactory := informer.NewSharedInformerFactoryWithOptions( - controllerCfg.provisioningClient, - controllerCfg.historyExpiration, - ) - historyJobInformer := historyInformerFactory.Provisioning().V0alpha1().HistoricJobs() - _, err = controller.NewHistoryJobController( - controllerCfg.provisioningClient.ProvisioningV0alpha1(), - historyJobInformer, - controllerCfg.historyExpiration, - ) - if err != nil { - return fmt.Errorf("failed to create history job controller: %w", err) - } - logger.Info("history cleanup enabled", "expiration", controllerCfg.historyExpiration.String()) - startHistoryInformers = func() { historyInformerFactory.Start(ctx.Done()) } - } else { - startHistoryInformers = func() {} - } - // HistoryWriter can be either Loki or the API server - // TODO: Loki configuration and setup in the same way we do for the API server - // https://github.com/grafana/git-ui-sync-project/issues/508 - // var jobHistoryWriter jobs.HistoryWriter - // if b.jobHistoryLoki != nil { - // jobHistoryWriter = b.jobHistoryLoki - // } else { - // jobHistoryWriter = jobs.NewAPIClientHistoryWriter(provisioningClient.ProvisioningV0alpha1()) - // } - - jobHistoryWriter := jobs.NewAPIClientHistoryWriter(controllerCfg.provisioningClient.ProvisioningV0alpha1()) - jobStore, err := jobs.NewJobStore(controllerCfg.provisioningClient.ProvisioningV0alpha1(), 30*time.Second, deps.Registerer) - if err != nil { - return fmt.Errorf("create API client job store: %w", err) - } - - workers, err := setupWorkers(controllerCfg, deps.Registerer, tracer) - if err != nil { - return fmt.Errorf("setup workers: %w", err) - } - - repoGetter := resources.NewRepositoryGetter( - controllerCfg.repoFactory, - controllerCfg.provisioningClient.ProvisioningV0alpha1(), - ) - - // This is basically our own JobQueue system - driver, err := jobs.NewConcurrentJobDriver( - controllerCfg.concurrentDrivers, - controllerCfg.maxJobTimeout, - controllerCfg.cleanupInterval, - controllerCfg.jobInterval, - controllerCfg.leaseRenewalInterval, - jobStore, - repoGetter, - jobHistoryWriter, - jobController.InsertNotifications(), - deps.Registerer, - workers..., - ) - if err != nil { - return fmt.Errorf("create concurrent job driver: %w", err) - } - - go func() { - logger.Info("jobs controller started") - if err := driver.Run(ctx); err != nil { - logger.Error("job driver failed", "error", err) - } - }() - - // Start informers - go jobInformerFactory.Start(ctx.Done()) - go startHistoryInformers() - - // Optionally wait for job cache sync; history cleanup can rely on resync events - if !cache.WaitForCacheSync(ctx.Done(), jobInformer.Informer().HasSynced) { - return fmt.Errorf("failed to sync job informer cache") - } - - <-ctx.Done() - return nil -} - -type jobsControllerConfig struct { - provisioningControllerConfig - historyExpiration time.Duration - maxJobTimeout time.Duration - cleanupInterval time.Duration - jobInterval time.Duration - leaseRenewalInterval time.Duration - concurrentDrivers int -} - -func setupJobsControllerFromConfig(cfg *setting.Cfg, registry prometheus.Registerer) (*jobsControllerConfig, error) { - controllerCfg, err := setupFromConfig(cfg, registry) - if err != nil { - return nil, err - } - - return &jobsControllerConfig{ - provisioningControllerConfig: *controllerCfg, - historyExpiration: cfg.SectionWithEnvOverrides("operator").Key("history_expiration").MustDuration(0), - concurrentDrivers: cfg.SectionWithEnvOverrides("operator").Key("concurrent_drivers").MustInt(3), - maxJobTimeout: cfg.SectionWithEnvOverrides("operator").Key("max_job_timeout").MustDuration(20 * time.Minute), - cleanupInterval: cfg.SectionWithEnvOverrides("operator").Key("cleanup_interval").MustDuration(time.Minute), - jobInterval: cfg.SectionWithEnvOverrides("operator").Key("job_interval").MustDuration(30 * time.Second), - leaseRenewalInterval: cfg.SectionWithEnvOverrides("operator").Key("lease_renewal_interval").MustDuration(30 * time.Second), - }, nil -} - -func setupWorkers(controllerCfg *jobsControllerConfig, registry prometheus.Registerer, tracer tracing.Tracer) ([]jobs.Worker, error) { - clients := controllerCfg.clients - parsers := resources.NewParserFactory(clients) - resourceLister := resources.NewResourceLister(controllerCfg.unified) - repositoryResources := resources.NewRepositoryResourcesFactory(parsers, clients, resourceLister) - statusPatcher := controller.NewRepositoryStatusPatcher(controllerCfg.provisioningClient.ProvisioningV0alpha1()) - - workers := make([]jobs.Worker, 0) - - metrics := jobs.RegisterJobMetrics(registry) - - // Sync - syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync, tracer) - syncWorker := sync.NewSyncWorker( - clients, - repositoryResources, - nil, // HACK: we have updated the worker to check for nil - statusPatcher.Patch, - syncer, - metrics, - tracer, - ) - workers = append(workers, syncWorker) - - // Export - stageIfPossible := repository.WrapWithStageAndPushIfPossible - exportWorker := export.NewExportWorker( - clients, - repositoryResources, - export.ExportAll, - stageIfPossible, - metrics, - ) - workers = append(workers, exportWorker) - - // Migrate - cleaner := migrate.NewNamespaceCleaner(clients) - unifiedStorageMigrator := migrate.NewUnifiedStorageMigrator( - cleaner, - exportWorker, - syncWorker, - ) - migrationWorker := migrate.NewMigrationWorkerFromUnified(unifiedStorageMigrator) - workers = append(workers, migrationWorker) - - // Delete - deleteWorker := deletepkg.NewWorker(syncWorker, stageIfPossible, repositoryResources, metrics) - workers = append(workers, deleteWorker) - - // Move - moveWorker := move.NewWorker(syncWorker, stageIfPossible, repositoryResources, metrics) - workers = append(workers, moveWorker) - - return workers, nil -} diff --git a/pkg/operators/register.go b/pkg/operators/register.go index 0eaa337eae8..b31b9837fb0 100644 --- a/pkg/operators/register.go +++ b/pkg/operators/register.go @@ -7,12 +7,6 @@ import ( ) func init() { - server.RegisterOperator(server.Operator{ - Name: "provisioning-jobs", - Description: "Watch provisioning jobs and manage job history cleanup", - RunFunc: provisioning.RunJobController, - }) - server.RegisterOperator(server.Operator{ Name: "provisioning-repo", Description: "Watch provisioning repositories", diff --git a/pkg/registry/apis/provisioning/webhooks/pullrequest/changes.go b/pkg/registry/apis/provisioning/webhooks/pullrequest/changes.go index 4c9a49a7db4..f8e485e8fe5 100644 --- a/pkg/registry/apis/provisioning/webhooks/pullrequest/changes.go +++ b/pkg/registry/apis/provisioning/webhooks/pullrequest/changes.go @@ -55,11 +55,11 @@ type fileChangeInfo struct { type evaluator struct { render ScreenshotRenderer parsers resources.ParserFactory - urlProvider func(namespace string) string + urlProvider func(ctx context.Context, namespace string) string metrics screenshotMetrics } -func NewEvaluator(render ScreenshotRenderer, parsers resources.ParserFactory, urlProvider func(namespace string) string, registry prometheus.Registerer) Evaluator { +func NewEvaluator(render ScreenshotRenderer, parsers resources.ParserFactory, urlProvider func(ctx context.Context, namespace string) string, registry prometheus.Registerer) Evaluator { metrics := registerScreenshotMetrics(registry) return &evaluator{ render: render, @@ -80,7 +80,7 @@ func (e *evaluator) Evaluate(ctx context.Context, repo repository.Reader, opts p rendererAvailable := e.render.IsAvailable(ctx) shouldRender := rendererAvailable && len(changes) == 1 && cfg.Spec.GitHub.GenerateDashboardPreviews info := changeInfo{ - GrafanaBaseURL: e.urlProvider(cfg.Namespace), + GrafanaBaseURL: e.urlProvider(ctx, cfg.Namespace), MissingImageRenderer: !rendererAvailable, } diff --git a/pkg/registry/apis/provisioning/webhooks/pullrequest/changes_test.go b/pkg/registry/apis/provisioning/webhooks/pullrequest/changes_test.go index 1d647256a2c..c8f0c33e92a 100644 --- a/pkg/registry/apis/provisioning/webhooks/pullrequest/changes_test.go +++ b/pkg/registry/apis/provisioning/webhooks/pullrequest/changes_test.go @@ -754,7 +754,7 @@ func TestCalculateChanges(t *testing.T) { tt.setupMocks(parser, reader, progress, renderer, parserFactory) - evaluator := NewEvaluator(renderer, parserFactory, func(_ string) string { + evaluator := NewEvaluator(renderer, parserFactory, func(_ context.Context, _ string) string { if tt.grafanaBaseURL != "" { return tt.grafanaBaseURL } diff --git a/pkg/registry/apis/provisioning/webhooks/pullrequest/worker.go b/pkg/registry/apis/provisioning/webhooks/pullrequest/worker.go index 299febe9786..9e15e63ab12 100644 --- a/pkg/registry/apis/provisioning/webhooks/pullrequest/worker.go +++ b/pkg/registry/apis/provisioning/webhooks/pullrequest/worker.go @@ -28,7 +28,7 @@ func ProvidePullRequestWorker( configProvider apiserver.RestConfigProvider, registry prometheus.Registerer, ) *PullRequestWorker { - urlProvider := func(_ string) string { + urlProvider := func(_ context.Context, _ string) string { return cfg.AppURL } diff --git a/pkg/registry/apis/provisioning/webhooks/register.go b/pkg/registry/apis/provisioning/webhooks/register.go index 4fc02db9041..e19eaff18de 100644 --- a/pkg/registry/apis/provisioning/webhooks/register.go +++ b/pkg/registry/apis/provisioning/webhooks/register.go @@ -26,7 +26,7 @@ import ( type WebhookExtraBuilder struct { provisioningapis.ExtraBuilder isPublic bool - urlProvider func(namespace string) string + urlProvider func(ctx context.Context, namespace string) string } // FIXME: separate the URL provider from connector to simplify operators @@ -38,7 +38,7 @@ func (b *WebhookExtraBuilder) WebhookURL(ctx context.Context, r *provisioning.Re gvr := provisioning.RepositoryResourceInfo.GroupVersionResource() webhookURL := fmt.Sprintf( "%sapis/%s/%s/namespaces/%s/%s/%s/webhook", - b.urlProvider(r.GetNamespace()), + b.urlProvider(ctx, r.GetNamespace()), gvr.Group, gvr.Version, r.GetNamespace(), @@ -66,10 +66,10 @@ func ProvideWebhooksWithImages( configProvider apiserver.RestConfigProvider, registry prometheus.Registerer, ) *WebhookExtraBuilder { - urlProvider := func(_ string) string { + urlProvider := func(_ context.Context, _ string) string { return cfg.AppURL } - isPublic := isPublicURL(urlProvider("")) + isPublic := isPublicURL(urlProvider(context.Background(), "")) return &WebhookExtraBuilder{ isPublic: isPublic, @@ -102,11 +102,11 @@ func ProvideWebhooksWithImages( } func ProvideWebhooks(provisioningURL string, registry prometheus.Registerer) *WebhookExtraBuilder { - urlProvider := func(_ string) string { + urlProvider := func(_ context.Context, _ string) string { return provisioningURL } - isPublic := isPublicURL(urlProvider("")) + isPublic := isPublicURL(urlProvider(context.Background(), "")) return &WebhookExtraBuilder{ isPublic: isPublic, @@ -131,7 +131,7 @@ type WebhookExtraWithImages struct { func NewWebhookExtraWithImages( render *renderConnector, webhook *webhookConnector, - urlProvider func(namespace string) string, + urlProvider func(ctx context.Context, namespace string) string, workers []jobs.Worker, ) *WebhookExtraWithImages { return &WebhookExtraWithImages{