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

[Provisioning] Use Job Worker abstraction (#100292)

Browse Source
This commit is contained in:
Roberto Jiménez Sánchez authored and GitHub committed 2025-02-07 23:25:12 +03:00
1 parent bd261794b2
commit 5d8f57a4a7
9 files changed
+236 -305
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/registry/apis/provisioning/jobs/export/dummyworker.go → pkg/registry/apis/provisioning/jobs/export/dummy.go
+9 -13
View File
@@ -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
}
}
pkg/registry/apis/provisioning/jobs/export/worker.go
+50 -46
View File
@@ -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)
pkg/registry/apis/provisioning/jobs/pullrequest/render.go
+8 -12
View File
@@ -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,
})
pkg/registry/apis/provisioning/jobs/pullrequest/worker.go
+58 -41
View File
@@ -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 != "" {
pkg/registry/apis/provisioning/jobs/queue.go
+84 -20
View File
@@ -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()
pkg/registry/apis/provisioning/jobs/sync/worker.go
+9 -4
View File
@@ -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)
}
pkg/registry/apis/provisioning/jobs/types.go
+6 -2
View File
@@ -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
pkg/registry/apis/provisioning/jobs/worker.go
-155
View File
@@ -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)
}
}
pkg/registry/apis/provisioning/register.go
+12 -12
View File
@@ -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,