diff --git a/pkg/registry/apis/provisioning/jobs/export/dummyworker.go b/pkg/registry/apis/provisioning/jobs/export/dummy.go similarity index 83% rename from pkg/registry/apis/provisioning/jobs/export/dummyworker.go rename to pkg/registry/apis/provisioning/jobs/export/dummy.go index 92961ae0900..de455ce9c34 100644 --- a/pkg/registry/apis/provisioning/jobs/export/dummyworker.go +++ b/pkg/registry/apis/provisioning/jobs/export/dummy.go @@ -8,21 +8,18 @@ import ( "github.com/grafana/grafana-app-sdk/logging" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/util" ) -type dummyExportWorker struct{} - -func NewDummyExportWorker() ExportWorker { - return &dummyExportWorker{} -} - -func (s *dummyExportWorker) Export(ctx context.Context, +func dummyExport( + ctx context.Context, repo repository.Repository, - options provisioning.ExportJobOptions, - progress func(provisioning.JobStatus) error, + job provisioning.Job, + progress jobs.ProgressFn, ) (*provisioning.JobStatus, error) { + options := job.Spec.Export logger := logging.FromContext(ctx).With("logger", "exporter", "repository", repo.Config().Name, "namespace", repo.Config().Name) logger.Info("start export", "folder", options.Folder) @@ -38,10 +35,10 @@ func (s *dummyExportWorker) Export(ctx context.Context, Progress: 0, } - err := progress(status) - if err != nil { + if err := progress(ctx, status); err != nil { return nil, err } + time.Sleep(200 * time.Millisecond) for i := 0; i < size; i++ { status.Progress = (float64(i) / float64(size)) * 100 @@ -65,8 +62,7 @@ func (s *dummyExportWorker) Export(ctx context.Context, dashboards, } - err = progress(status) - if err != nil { + if err := progress(ctx, status); err != nil { return nil, err } } diff --git a/pkg/registry/apis/provisioning/jobs/export/worker.go b/pkg/registry/apis/provisioning/jobs/export/worker.go index 98796d44315..28d1a2ec9ef 100644 --- a/pkg/registry/apis/provisioning/jobs/export/worker.go +++ b/pkg/registry/apis/provisioning/jobs/export/worker.go @@ -10,83 +10,89 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/client-go/dynamic" "github.com/grafana/grafana-app-sdk/logging" apiutils "github.com/grafana/grafana/pkg/apimachinery/utils" folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/pkg/infra/slugify" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath" ) -// Export reads from grafana and writes to a a repository -type ExportWorker interface { - Export(ctx context.Context, - repo repository.Repository, - options provisioning.ExportJobOptions, - progress func(provisioning.JobStatus) error, - ) (*provisioning.JobStatus, error) +type ExportWorker struct { + parsers *resources.ParserFactory } -type exportWorker struct { - client *resources.DynamicClient // namespaced! - dashboards dynamic.ResourceInterface - folders dynamic.ResourceInterface - repository repository.Repository +func NewExportWorker(parsers *resources.ParserFactory) *ExportWorker { + return &ExportWorker{parsers: parsers} } -func NewExportWorker( - repo repository.Repository, - dynamicClient *resources.DynamicClient, -) (ExportWorker, error) { - if dynamicClient.GetNamespace() != repo.Config().Namespace { - return nil, fmt.Errorf("bad setup, exporter needs a namespaced client matching the repository") +func (r *ExportWorker) IsSupported(ctx context.Context, job provisioning.Job) bool { + return job.Spec.Action == provisioning.JobActionExport +} + +//nolint:gocyclo +func (r *ExportWorker) Process(ctx context.Context, repo repository.Repository, job provisioning.Job, progress jobs.ProgressFn) (*provisioning.JobStatus, error) { + if repo.Config().Spec.ReadOnly { + return &provisioning.JobStatus{ + State: provisioning.JobStateError, + Errors: []string{"Exporting to a read only repository is not supported"}, + }, nil } - folders := dynamicClient.Resource(schema.GroupVersionResource{ + + options := job.Spec.Export + if options == nil { + return &provisioning.JobStatus{ + State: provisioning.JobStateError, + Errors: []string{"Export job missing export settings"}, + }, nil + } + + // TODO: remove this dummy export + if job.Spec.Export.Branch == "*dummy*" { + return dummyExport(ctx, repo, job, progress) + } + + parser, err := r.parsers.GetParser(ctx, repo) + if err != nil { + return nil, fmt.Errorf("failed to get parser for %s: %w", repo.Config().GetName(), err) + } + dynamicClient := parser.Client() + if repo.Config().Namespace != dynamicClient.GetNamespace() { + return nil, fmt.Errorf("namespace mismatch") + } + + foldersClient := dynamicClient.Resource(schema.GroupVersionResource{ Group: folders.GROUP, Version: folders.VERSION, Resource: folders.RESOURCE, }) - dashboards := dynamicClient.Resource(schema.GroupVersionResource{ + dashboardsClient := dynamicClient.Resource(schema.GroupVersionResource{ Group: "dashboard.grafana.app", Version: "v1alpha1", Resource: "dashboards", }) - return &exportWorker{ - client: dynamicClient, - folders: folders, - dashboards: dashboards, - repository: repo, - }, nil -} - -//nolint:gocyclo -func (r *exportWorker) Export(ctx context.Context, - repo repository.Repository, - options provisioning.ExportJobOptions, - progress func(provisioning.JobStatus) error, -) (*provisioning.JobStatus, error) { logger := logging.FromContext(ctx) status := provisioning.JobStatus{ State: provisioning.JobStateWorking, Message: "reading folder tree...", } - err := progress(status) - if err != nil { + if err := progress(ctx, status); err != nil { return nil, err } + ref := options.Branch // only valid for git (defaults to the configured repo) if options.Prefix != "" { options.Prefix = safepath.Clean(options.Prefix) } // TODO: handle pagination - rawList, err := r.folders.List(ctx, metav1.ListOptions{Limit: 10000}) + rawList, err := foldersClient.List(ctx, metav1.ListOptions{Limit: 10000}) if err != nil { return nil, fmt.Errorf("failed to list folders: %w", err) } @@ -106,8 +112,7 @@ func (r *exportWorker) Export(ctx context.Context, } status.Message = fmt.Sprintf("exporting folders (%d)...", len(rawFolders)) - err = progress(status) - if err != nil { + if err := progress(ctx, status); err != nil { return nil, err } @@ -128,7 +133,7 @@ func (r *exportWorker) Export(ctx context.Context, } logger := logger.With("path", p) - _, err = r.repository.Read(ctx, p, ref) + _, err = repo.Read(ctx, p, ref) if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) { logger.Error("failed to check if folder exists before writing", "error", err) return fmt.Errorf("failed to check if folder exists before writing: %w", err) @@ -139,7 +144,7 @@ func (r *exportWorker) Export(ctx context.Context, } // Create with an empty body will make a folder (or .keep file if unsupported) - if err := r.repository.Create(ctx, p, ref, nil, "export folder `"+p+"`"); err != nil { + if err := repo.Create(ctx, p, ref, nil, "export folder `"+p+"`"); err != nil { logger.Error("failed to write a folder in repository", "error", err) return fmt.Errorf("failed to write folder in repo: %w", err) } @@ -153,13 +158,12 @@ func (r *exportWorker) Export(ctx context.Context, status.Summary = append(status.Summary, summary) status.Message = "writing dashboards..." - err = progress(status) - if err != nil { + if err := progress(ctx, status); err != nil { return nil, err } // TODO: handle pagination - dashboardList, err := r.dashboards.List(ctx, metav1.ListOptions{Limit: 1000}) + dashboardList, err := dashboardsClient.List(ctx, metav1.ListOptions{Limit: 1000}) if err != nil { return nil, fmt.Errorf("failed to list dashboards: %w", err) } @@ -220,7 +224,7 @@ func (r *exportWorker) Export(ctx context.Context, } // Write the file - err = r.repository.Write(ctx, fileName, ref, body, commitMessage) + err = repo.Write(ctx, fileName, ref, body, commitMessage) if err != nil { summary.Error++ logger.Error("failed to write a file in repository", "error", err) diff --git a/pkg/registry/apis/provisioning/jobs/pullrequest/render.go b/pkg/registry/apis/provisioning/jobs/pullrequest/render.go index 6e62fc3fbae..798ccbde1ff 100644 --- a/pkg/registry/apis/provisioning/jobs/pullrequest/render.go +++ b/pkg/registry/apis/provisioning/jobs/pullrequest/render.go @@ -8,25 +8,20 @@ import ( "time" "github.com/grafana/grafana/pkg/apimachinery/identity" - provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/pkg/models" "github.com/grafana/grafana/pkg/services/rendering" "github.com/grafana/grafana/pkg/storage/unified/blob" ) type Renderer struct { - cfg *provisioning.Repository render rendering.Service blobstore blob.PublicBlobStore - id identity.Requester } -func NewRenderer(cfg *provisioning.Repository, render rendering.Service, blobstore blob.PublicBlobStore, id identity.Requester) *Renderer { +func NewRenderer(render rendering.Service, blobstore blob.PublicBlobStore) *Renderer { return &Renderer{ - cfg: cfg, render: render, blobstore: blobstore, - id: id, } } @@ -34,12 +29,13 @@ func (r *Renderer) IsAvailable(ctx context.Context) bool { return r.render != nil && r.render.IsAvailable(ctx) && r.blobstore.IsAvailable() } -func (r *Renderer) RenderDashboardPreview(ctx context.Context, path string, ref string) (string, error) { - url := fmt.Sprintf("admin/provisioning/%s/dashboard/preview/%s?kiosk&ref=%s", r.cfg.Name, path, ref) +func (r *Renderer) RenderDashboardPreview(ctx context.Context, namespace, repoName, path, ref string) (string, error) { + url := fmt.Sprintf("admin/provisioning/%s/dashboard/preview/%s?kiosk&ref=%s", repoName, path, ref) // fmt.Printf("RENDER: http://localhost:3000/render/%s\n", url) - renderContext := identity.WithRequester(context.Background(), r.id) - result, err := r.render.Render(renderContext, rendering.RenderPNG, rendering.Opts{ + // TODO: why were we using a different context? + // renderContext := identity.WithRequester(context.Background(), r.id) + result, err := r.render.Render(ctx, rendering.RenderPNG, rendering.Opts{ CommonOpts: rendering.CommonOpts{ Path: url, AuthOpts: rendering.AuthOpts{ @@ -65,8 +61,8 @@ func (r *Renderer) RenderDashboardPreview(ctx context.Context, path string, ref return "", err } - return r.blobstore.SaveBlob(ctx, r.cfg.Namespace, ext, body, map[string]string{ - "repo": r.cfg.Name, + return r.blobstore.SaveBlob(ctx, namespace, ext, body, map[string]string{ + "repo": repoName, "path": path, // only used when saving in GCS/S3++ "ref": ref, }) diff --git a/pkg/registry/apis/provisioning/jobs/pullrequest/worker.go b/pkg/registry/apis/provisioning/jobs/pullrequest/worker.go index 2debd9e3a88..27741d656ea 100644 --- a/pkg/registry/apis/provisioning/jobs/pullrequest/worker.go +++ b/pkg/registry/apis/provisioning/jobs/pullrequest/worker.go @@ -10,20 +10,15 @@ import ( "os" "path" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "github.com/grafana/grafana-app-sdk/logging" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" ) -type PullRequestWorker interface { - Process(ctx context.Context, - repo repository.Repository, - options provisioning.PullRequestJobOptions, - progress func(provisioning.JobStatus) error, - ) (*provisioning.JobStatus, error) -} - // resourcePreview represents a resource that has changed in a pull request. type resourcePreview struct { Filename string @@ -71,24 +66,22 @@ type PullRequestRepo interface { // PreviewRenderer is an interface for rendering a preview of a file type PreviewRenderer interface { IsAvailable(ctx context.Context) bool - RenderDashboardPreview(ctx context.Context, path string, ref string) (string, error) + RenderDashboardPreview(ctx context.Context, namespace, repoName, path, ref string) (string, error) } -type pullRequestCommenter struct { - repo PullRequestRepo - parser *resources.Parser +type PullRequestWorker struct { + parsers *resources.ParserFactory lintTemplate *template.Template prevTemplate *template.Template - baseURL *url.URL + urlProvider func(namespace string) string renderer PreviewRenderer } func NewPullRequestWorker( - repo PullRequestRepo, - parser *resources.Parser, + parsers *resources.ParserFactory, renderer PreviewRenderer, - baseURL *url.URL, -) (PullRequestWorker, error) { + urlProvider func(namespace string) string, +) (*PullRequestWorker, error) { lintTemplate, err := template.New("comment").Parse(lintDashboardIssuesTemplate) if err != nil { return nil, fmt.Errorf("parse lint comment template: %w", err) @@ -98,22 +91,46 @@ func NewPullRequestWorker( return nil, fmt.Errorf("parse previews comment template: %w", err) } - return &pullRequestCommenter{ - repo: repo, - parser: parser, + return &PullRequestWorker{ + parsers: parsers, lintTemplate: lintTemplate, prevTemplate: prevTemplate, renderer: renderer, - baseURL: baseURL, + urlProvider: urlProvider, }, nil } -func (c *pullRequestCommenter) Process(ctx context.Context, +func (c *PullRequestWorker) IsSupported(ctx context.Context, job provisioning.Job) bool { + return job.Spec.Action == provisioning.JobActionPullRequest +} + +//nolint:gocyclo +func (c *PullRequestWorker) Process(ctx context.Context, repo repository.Repository, - options provisioning.PullRequestJobOptions, - progress func(provisioning.JobStatus) error, + job provisioning.Job, + progress jobs.ProgressFn, ) (*provisioning.JobStatus, error) { - cfg := c.repo.Config().Spec + cfg := repo.Config().Spec + + options := job.Spec.PullRequest + if options == nil { + return nil, apierrors.NewBadRequest("missing spec.pr") + } + + prRepo, ok := repo.(PullRequestRepo) + if !ok { + return nil, fmt.Errorf("repository is not a github repository") + } + + baseURL, err := url.Parse(c.urlProvider(job.Namespace)) + if err != nil { + return nil, fmt.Errorf("error parsing base url: %w", err) + } + + parser, err := c.parsers.GetParser(ctx, repo) + if err != nil { + return nil, fmt.Errorf("failed to get parser for %s: %w", repo.Config().Name, err) + } // TODO: Figure out how we want to determine this in practice. lintingVal, ok := os.LookupEnv("GRAFANA_LINTING") @@ -136,7 +153,7 @@ func (c *pullRequestCommenter) Process(ctx context.Context, ref := options.Hash // list pull requests changes files - files, err := c.repo.CompareFiles(ctx, base, ref) + files, err := prRepo.CompareFiles(ctx, base, ref) if err != nil { return &provisioning.JobStatus{ State: provisioning.JobStateError, @@ -145,7 +162,7 @@ func (c *pullRequestCommenter) Process(ctx context.Context, } // clear all previous comments - if err := c.repo.ClearAllPullRequestFileComments(ctx, options.PR); err != nil { + if err := prRepo.ClearAllPullRequestFileComments(ctx, options.PR); err != nil { return &provisioning.JobStatus{ State: provisioning.JobStateError, Message: fmt.Sprintf("failed to clear pull request comments: %+v", err), @@ -167,7 +184,7 @@ func (c *pullRequestCommenter) Process(ctx context.Context, } logger := logger.With("file", f.Path) - fileInfo, err := c.repo.Read(ctx, f.Path, ref) + fileInfo, err := prRepo.Read(ctx, f.Path, ref) if err != nil { return &provisioning.JobStatus{ State: provisioning.JobStateError, @@ -175,7 +192,7 @@ func (c *pullRequestCommenter) Process(ctx context.Context, }, nil } - parsed, err := c.parser.Parse(ctx, fileInfo, true) + parsed, err := parser.Parse(ctx, fileInfo, true) if err != nil { if errors.Is(err, resources.ErrUnableToReadResourceBytes) { logger.Debug("file is not a resource", "path", f.Path) @@ -191,7 +208,7 @@ func (c *pullRequestCommenter) Process(ctx context.Context, return nil, fmt.Errorf("execute lint comment template: %w", err) } - if err := c.repo.CommentPullRequestFile(ctx, options.PR, f.Path, ref, buf.String()); err != nil { + if err := prRepo.CommentPullRequestFile(ctx, options.PR, f.Path, ref, buf.String()); err != nil { return nil, fmt.Errorf("comment pull request file %s: %w", f.Path, err) } @@ -207,21 +224,21 @@ func (c *pullRequestCommenter) Process(ctx context.Context, switch f.Action { case repository.FileActionCreated: - preview.PreviewURL = c.previewURL(ref, f.Path, options.URL) + preview.PreviewURL = c.previewURL(baseURL, repo.Config().GetName(), ref, f.Path, options.URL) case repository.FileActionUpdated: - preview.OriginalURL = c.previewURL(base, f.Path, options.URL) - preview.PreviewURL = c.previewURL(ref, f.Path, options.URL) + preview.OriginalURL = c.previewURL(baseURL, repo.Config().GetName(), ref, f.Path, options.URL) + preview.PreviewURL = c.previewURL(baseURL, repo.Config().GetName(), ref, f.Path, options.URL) case repository.FileActionRenamed: - preview.OriginalURL = c.previewURL(base, f.PreviousPath, options.URL) - preview.PreviewURL = c.previewURL(ref, f.Path, options.URL) + preview.OriginalURL = c.previewURL(baseURL, repo.Config().GetName(), base, f.PreviousPath, options.URL) + preview.PreviewURL = c.previewURL(baseURL, repo.Config().GetName(), ref, f.Path, options.URL) case repository.FileActionDeleted: - preview.OriginalURL = c.previewURL(base, f.Path, options.URL) + preview.OriginalURL = c.previewURL(baseURL, repo.Config().GetName(), base, f.Path, options.URL) default: return nil, fmt.Errorf("unknown file action: %s", f.Action) } if cfg.GitHub.GenerateDashboardPreviews && f.Action != repository.FileActionDeleted { - screenshotURL, err := c.renderer.RenderDashboardPreview(ctx, f.Path, ref) + screenshotURL, err := c.renderer.RenderDashboardPreview(ctx, repo.Config().GetNamespace(), repo.Config().GetName(), f.Path, ref) if err != nil { return nil, fmt.Errorf("render dashboard preview: %w", err) } @@ -239,7 +256,7 @@ func (c *pullRequestCommenter) Process(ctx context.Context, return nil, fmt.Errorf("execute previews comment template: %w", err) } - if err := c.repo.CommentPullRequest(ctx, options.PR, buf.String()); err != nil { + if err := prRepo.CommentPullRequest(ctx, options.PR, buf.String()); err != nil { return nil, fmt.Errorf("comment pull request: %w", err) } @@ -253,10 +270,10 @@ func (c *pullRequestCommenter) Process(ctx context.Context, } // previewURL returns the URL to preview the file in Grafana -func (c *pullRequestCommenter) previewURL(ref, filePath, pullRequestURL string) string { +func (c *PullRequestWorker) previewURL(u *url.URL, repoName, ref, filePath, pullRequestURL string) string { // Copy the baseURL to modify path and query - baseURL := *c.baseURL - baseURL = *baseURL.JoinPath("/admin/provisioning", c.repo.Config().GetName(), "dashboard/preview", filePath) + baseURL := *u + baseURL = *baseURL.JoinPath("/admin/provisioning", repoName, "dashboard/preview", filePath) query := baseURL.Query() if ref != "" { diff --git a/pkg/registry/apis/provisioning/jobs/queue.go b/pkg/registry/apis/provisioning/jobs/queue.go index 759d03fe3ea..abb8e622c50 100644 --- a/pkg/registry/apis/provisioning/jobs/queue.go +++ b/pkg/registry/apis/provisioning/jobs/queue.go @@ -2,6 +2,7 @@ package jobs import ( "context" + "errors" "fmt" "net/http" "strconv" @@ -11,26 +12,39 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/endpoints/request" "k8s.io/apiserver/pkg/storage" "github.com/grafana/grafana-app-sdk/logging" + "github.com/grafana/grafana/pkg/apimachinery/identity" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/auth" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/util" ) -func NewJobQueue(capacity int) JobQueue { +type RepoGetter interface { + GetRepository(ctx context.Context, name string) (repository.Repository, error) +} + +func NewJobQueue(capacity int, getter RepoGetter, identities auth.BackgroundIdentityService) JobQueue { return &jobStore{ - rv: 1, - capacity: capacity, - jobs: []provisioning.Job{}, - watchSet: NewWatchSet(), - versioner: &storage.APIObjectVersioner{}, + workers: make([]Worker, 0), + getter: getter, + identities: identities, + rv: 1, + capacity: capacity, + jobs: []provisioning.Job{}, + watchSet: NewWatchSet(), + versioner: &storage.APIObjectVersioner{}, } } type jobStore struct { - capacity int - worker Worker + getter RepoGetter + identities auth.BackgroundIdentityService + capacity int + workers []Worker // All jobs jobs []provisioning.Job @@ -43,7 +57,7 @@ type jobStore struct { // Register a worker (inline for now) func (s *jobStore) Register(worker Worker) { - s.worker = worker + s.workers = append(s.workers, worker) } // Add implements JobQueue. @@ -131,18 +145,18 @@ func (s *jobStore) drainPending() { ctx := logging.Context(ctx, logger) started := time.Now() - var status *provisioning.JobStatus - if s.worker == nil { - status = &provisioning.JobStatus{ - State: provisioning.JobStateError, - Errors: []string{ - "no registered worker supports this job", - }, + var ( + status *provisioning.JobStatus + foundWorker bool + ) + + for _, worker := range s.workers { + if !worker.IsSupported(ctx, *job) { + continue } - } else { - status, err = s.worker.Process(ctx, *job, func(j provisioning.JobStatus) error { - return s.Update(ctx, job.Namespace, job.Name, j) - }) + + foundWorker = true + status, err = s.processByWorker(ctx, worker, *job) if err != nil { logger.Error("error processing job", "error", err) status = &provisioning.JobStatus{ @@ -160,6 +174,18 @@ func (s *jobStore) drainPending() { status.State = provisioning.JobStateSuccess } logger.Debug("job processing finished", "status", status.State) + + // Already found a worker, no need to continue + break + } + + if !foundWorker { + status = &provisioning.JobStatus{ + State: provisioning.JobStateError, + Errors: []string{ + "no registered worker supports this job", + }, + } } status.Started = started.UnixMilli() @@ -173,6 +199,44 @@ func (s *jobStore) drainPending() { } } +func (s *jobStore) processByWorker(ctx context.Context, worker Worker, job provisioning.Job) (*provisioning.JobStatus, error) { + id, err := s.identities.WorkerIdentity(ctx, job.Name) + if err != nil { + return nil, fmt.Errorf("get worker identity: %w", err) + } + + ctx = request.WithNamespace(identity.WithRequester(ctx, id), job.Namespace) + repoName := job.Spec.Repository + + logger := logging.FromContext(ctx) + logger = logger.With("repository", repoName) + ctx = logging.Context(ctx, logger) + + repo, err := s.getter.GetRepository(ctx, repoName) + if err != nil { + return nil, fmt.Errorf("get repository: %w", err) + } + + // TODO: does this really happen? + if repo == nil { + return nil, errors.New("unknown repository") + } + + status, err := worker.Process(ctx, repo, job, func(ctx context.Context, j provisioning.JobStatus) error { + return s.Update(ctx, job.Namespace, job.Name, j) + }) + if err != nil { + return nil, fmt.Errorf("process job: %w", err) + } + + // TODO: does this really happen? + if status == nil { + return nil, errors.New("worker did not return a status") + } + + return nil, nil +} + // Checkout the next "pending" job func (s *jobStore) Next(ctx context.Context) *provisioning.Job { s.mutex.Lock() diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker.go b/pkg/registry/apis/provisioning/jobs/sync/worker.go index 1a956a3c73f..f9706765231 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker.go @@ -23,6 +23,7 @@ import ( folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" client "github.com/grafana/grafana/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath" @@ -48,10 +49,14 @@ func NewSyncWorker( } } +func (r *SyncWorker) IsSupported(ctx context.Context, job provisioning.Job) bool { + return job.Spec.Action == provisioning.JobActionSync +} + func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, job provisioning.Job, - progress func(provisioning.JobStatus) error, + progress jobs.ProgressFn, ) (*provisioning.JobStatus, error) { var err error cfg := repo.Config() @@ -126,7 +131,7 @@ func (r *SyncWorker) Process(ctx context.Context, func (r *SyncWorker) sync(ctx context.Context, repo repository.Repository, options provisioning.SyncJobOptions, - progress func(provisioning.JobStatus) error, + progress jobs.ProgressFn, ) (*provisioning.JobStatus, *provisioning.SyncStatus, error) { cfg := repo.Config() if !cfg.Spec.Sync.Enabled { @@ -190,7 +195,7 @@ func (r *SyncWorker) sync(ctx context.Context, type syncJob struct { repository repository.Repository options provisioning.SyncJobOptions - progress func(provisioning.JobStatus) error + progress jobs.ProgressFn progressInterval time.Duration progressLast time.Time logger logging.Logger @@ -331,7 +336,7 @@ func (r *syncJob) applyChanges(ctx context.Context, changes []ResourceFileChange // Convert git changes into resource file changes func (r *syncJob) maybeNotify(ctx context.Context) { if time.Since(r.progressLast) > r.progressInterval { - err := r.progress(*r.jobStatus) + err := r.progress(ctx, *r.jobStatus) if err != nil { r.logger.Warn("unable to send progress", "err", err) } diff --git a/pkg/registry/apis/provisioning/jobs/types.go b/pkg/registry/apis/provisioning/jobs/types.go index f5598feb6c1..55d70a5fe32 100644 --- a/pkg/registry/apis/provisioning/jobs/types.go +++ b/pkg/registry/apis/provisioning/jobs/types.go @@ -6,6 +6,7 @@ import ( "k8s.io/apiserver/pkg/registry/rest" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" ) var _ JobQueue = (*jobStore)(nil) @@ -29,6 +30,9 @@ type JobQueue interface { } type Worker interface { - // Process a single job, can periodically make progress updates - Process(ctx context.Context, job provisioning.Job, progress func(provisioning.JobStatus) error) (*provisioning.JobStatus, error) + IsSupported(ctx context.Context, job provisioning.Job) bool + Process(ctx context.Context, repo repository.Repository, job provisioning.Job, progress ProgressFn) (*provisioning.JobStatus, error) } + +// ProgressFn is a function that can be called to update the progress of a job +type ProgressFn func(ctx context.Context, status provisioning.JobStatus) error diff --git a/pkg/registry/apis/provisioning/jobs/worker.go b/pkg/registry/apis/provisioning/jobs/worker.go deleted file mode 100644 index e1e17c95917..00000000000 --- a/pkg/registry/apis/provisioning/jobs/worker.go +++ /dev/null @@ -1,155 +0,0 @@ -package jobs - -import ( - "context" - "fmt" - "net/url" - - apierrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apiserver/pkg/endpoints/request" - - "github.com/grafana/grafana-app-sdk/logging" - "github.com/grafana/grafana/pkg/apimachinery/identity" - provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/auth" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/export" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/pullrequest" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" - "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" - "github.com/grafana/grafana/pkg/services/rendering" - "github.com/grafana/grafana/pkg/storage/unified/blob" -) - -type SyncWorker interface { - Process( - ctx context.Context, - repo repository.Repository, - job provisioning.Job, - progress func(provisioning.JobStatus) error, - ) (*provisioning.JobStatus, error) -} - -type RepoGetter interface { - GetRepository(ctx context.Context, name string) (repository.Repository, error) -} - -var _ Worker = (*JobWorker)(nil) - -type JobWorker struct { - getter RepoGetter - parsers *resources.ParserFactory - identities auth.BackgroundIdentityService - render rendering.Service - lister resources.ResourceLister - blobstore blob.PublicBlobStore - urlProvider func(namespace string) string - syncWorker SyncWorker -} - -func NewJobWorker( - getter RepoGetter, - parsers *resources.ParserFactory, - identities auth.BackgroundIdentityService, - render rendering.Service, - lister resources.ResourceLister, - syncWorker SyncWorker, - blobstore blob.PublicBlobStore, - urlProvider func(namespace string) string, -) *JobWorker { - return &JobWorker{ - getter: getter, - parsers: parsers, - identities: identities, - render: render, - lister: lister, - blobstore: blobstore, - urlProvider: urlProvider, - syncWorker: syncWorker, - } -} - -func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress func(provisioning.JobStatus) error) (*provisioning.JobStatus, error) { - logger := logging.FromContext(ctx).With("job", job.GetName(), "namespace", job.GetNamespace()) - ctx = logging.Context(ctx, logger) - - id, err := g.identities.WorkerIdentity(ctx, job.Name) - if err != nil { - return nil, err - } - - ctx = request.WithNamespace(identity.WithRequester(ctx, id), job.Namespace) - repoName := job.Spec.Repository - logger = logger.With("repository", repoName) - ctx = logging.Context(ctx, logger) - - repo, err := g.getter.GetRepository(ctx, repoName) - if err != nil { - return nil, err - } - if repo == nil { - return nil, fmt.Errorf("unknown repository") - } - - parser, err := g.parsers.GetParser(ctx, repo) - if err != nil { - return nil, fmt.Errorf("failed to get parser for %s: %w", repo.Config().Name, err) - } - - switch job.Spec.Action { - case provisioning.JobActionSync: - return g.syncWorker.Process(ctx, repo, job, progress) - case provisioning.JobActionPullRequest: - prRepo, ok := repo.(pullrequest.PullRequestRepo) - if !ok { - return nil, fmt.Errorf("repository is not a github repository") - } - - baseURL, err := url.Parse(g.urlProvider(job.Namespace)) - if err != nil { - return nil, fmt.Errorf("error parsing base url: %w", err) - } - - renderer := pullrequest.NewRenderer(repo.Config(), g.render, g.blobstore, id) - - options := job.Spec.PullRequest - if options == nil { - return nil, apierrors.NewBadRequest("missing spec.pr") - } - - commenter, err := pullrequest.NewPullRequestWorker(prRepo, parser, renderer, baseURL) - if err != nil { - return nil, fmt.Errorf("error creating pull request commenter: %w", err) - } - return commenter.Process(ctx, repo, *options, progress) - - case provisioning.JobActionExport: - if job.Spec.Export == nil { - return &provisioning.JobStatus{ - State: provisioning.JobStateError, - Errors: []string{"Export job missing export settings"}, - }, nil - } - - if repo.Config().Spec.ReadOnly { - return &provisioning.JobStatus{ - State: provisioning.JobStateError, - Errors: []string{"Exporting to a read only repository is not supported"}, - }, nil - } - - var exporter export.ExportWorker - // Test for now... so we have something with long spinners for UI testing!!! - if job.Spec.Export.Branch == "*dummy*" { - exporter = export.NewDummyExportWorker() - } else { - exporter, err = export.NewExportWorker(repo, parser.Client()) - if err != nil { - return nil, err - } - } - return exporter.Export(ctx, repo, *job.Spec.Export, progress) - - default: - return nil, fmt.Errorf("unknown job action: %s", job.Spec.Action) - } -} diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 559cdfbd5a1..623b8cb586f 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -31,6 +31,8 @@ import ( listers "github.com/grafana/grafana/pkg/generated/listers/provisioning/v0alpha1" "github.com/grafana/grafana/pkg/registry/apis/provisioning/auth" "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/pullrequest" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/sync" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository/github" @@ -99,7 +101,6 @@ func NewAPIBuilder( render: render, resourceLister: resources.NewResourceLister(index), blobstore: blobstore, - jobs: jobs.NewJobQueue(50), // in memory for now } } @@ -182,6 +183,7 @@ func (b *APIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.APIGroupI return fmt.Errorf("failed to create repository storage: %w", err) } b.getter = repositoryStorage + b.jobs = jobs.NewJobQueue(50, b, b.identities) // in memory for now repositoryStatusStorage := grafanaregistry.NewRegistryStatusStore(opts.Scheme, repositoryStorage) @@ -433,22 +435,20 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH } b.repositoryLister = repoInformer.Lister() - syncWorker := sync.NewSyncWorker( + b.jobs.Register(export.NewExportWorker(b.parsers)) + b.jobs.Register(sync.NewSyncWorker( c.ProvisioningV0alpha1(), b.parsers, b.resourceLister, - ) - b.jobs.Register(jobs.NewJobWorker( - b, - b.parsers, - b.identities, - b.render, - b.resourceLister, - syncWorker, - b.blobstore, - b.urlProvider, )) + renderer := pullrequest.NewRenderer(b.render, b.blobstore) + pullRequestWorker, err := pullrequest.NewPullRequestWorker(b.parsers, renderer, b.urlProvider) + if err != nil { + return fmt.Errorf("create pull request worker: %w", err) + } + b.jobs.Register(pullRequestWorker) + repoController, err := NewRepositoryController( c.ProvisioningV0alpha1(), repoInformer,