POC/Provisioning: Export: Move to a job worker architecture (#97927)

* POC/Provisioning: Export: Move to a job worker architecture

* fix: move export logic to replicator

The logic changed in the last rebase, which wasn't taken into account.
This should make us use a single worker again.

pkg/registry/apis/provisioning/export.go

@@ -2,37 +2,28 @@ package provisioning
 
 import (
 	"context"
-	"encoding/json"
-	"errors"
 	"fmt"
 	"log/slog"
 	"net/http"
-	"path"
-	"path/filepath"
 
-	"gopkg.in/yaml.v3"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/registry/rest"
-	"k8s.io/client-go/dynamic"
 
-	apiutils "github.com/grafana/grafana/pkg/apimachinery/utils"
 	provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
-	"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/jobs"
 )
 
 type exportConnector struct {
 	repoGetter RepoGetter
-	client     *resources.ClientFactory
 	logger     *slog.Logger
+	queue      jobs.JobQueue
 }
 
 func (*exportConnector) New() runtime.Object {
 	// This is added as the "ResponseType" regardless what ProducesObject() returns
-	return &provisioning.ResourceWrapper{}
+	return provisioning.JobResourceInfo.NewFunc()
 }
 
 func (*exportConnector) Destroy() {}
@@ -67,192 +58,42 @@ func (c *exportConnector) Connect(
 	opts runtime.Object,
 	responder rest.Responder,
 ) (http.Handler, error) {
-	logger := c.logger.With("repository_name", name)
 	repo, err := c.repoGetter.GetRepository(ctx, name)
 	if err != nil {
 		return nil, err
 	}
 	ns := repo.Config().GetNamespace()
+	logger := c.logger.With("repository", name, "namespace", ns)
 
 	// TODO: We need some way to filter what we export.
 
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		client, _, err := c.client.New(ns)
-		if err != nil {
-			responder.Error(apierrors.NewInternalError(fmt.Errorf("failed to create a dynamic client: %w", err)))
-			return
-		}
-
-		dashboardIface := client.Resource(schema.GroupVersionResource{
-			Group:    "dashboard.grafana.app",
-			Version:  "v2alpha1",
-			Resource: "dashboards",
+		job, err := c.queue.Add(r.Context(), &provisioning.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: ns,
+				Labels: map[string]string{
+					"repository": repo.Config().GetName(),
+				},
+			},
+			Spec: provisioning.JobSpec{
+				Action: provisioning.JobActionExport,
+			},
 		})
-		folderIface := client.Resource(schema.GroupVersionResource{
-			Group:    "folder.grafana.app",
-			Version:  "v0alpha1",
-			Resource: "folders",
-		})
-
-		// TODO: handle pagination
-		folders, err := c.fetchRepoFolderTree(ctx, folderIface, repo.Config().Spec.Folder)
 		if err != nil {
-			responder.Error(apierrors.NewInternalError(fmt.Errorf("failed to list folders: %w", err)))
-			return
-		}
-
-		// TODO: handle pagination
-		dashboardList, err := dashboardIface.List(ctx, metav1.ListOptions{})
-		if err != nil {
-			responder.Error(apierrors.NewInternalError(fmt.Errorf("failed to list dashboards: %w", err)))
-			return
-		}
-
-		for _, item := range dashboardList.Items {
-			if ctx.Err() != nil {
-				// FIXME: use a proper error code when context is cancelled
-				responder.Error(apierrors.NewInternalError(ctx.Err()))
-				return
-			}
-
-			logger := logger.With("name", item.GetName(), "namespace", item.GetNamespace())
-			logger.DebugContext(ctx, "got item in dashboard list")
-			name := item.GetName()
-			if namespace := item.GetNamespace(); namespace != ns {
-				logger.DebugContext(ctx, "skipping dashboard in export due to mismatched namespace",
-					"expected", ns)
-				continue
-			}
-
-			folder := item.GetAnnotations()[apiutils.AnnoKeyFolder]
-			if !folders.In(folder) {
-				logger.DebugContext(ctx, "skipping dashboard in export due to folder being out-of-tree of repository")
-				continue
-			}
-
-			delete(item.Object, "metadata")
-			marshalledBody, baseFileName, err := c.marshalPreferredFormat(item.Object, name, repo)
-			if err != nil {
-				logger.ErrorContext(ctx, "failed to marshal dashboard into preferred format", "err", err)
-				responder.Error(apierrors.NewInternalError(fmt.Errorf("failed to marshal dashboard %s: %w", name, err)))
-				return
-			}
-
-			var ref string
-			if repo.Config().Spec.Type == provisioning.GitHubRepositoryType {
-				ref = repo.Config().Spec.GitHub.Branch
-			}
-
-			fileName := filepath.Join(folders.DirPath(folder), baseFileName)
-			_, err = repo.Read(ctx, logger, fileName, ref)
-			if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {
-				responder.Error(apierrors.NewInternalError(fmt.Errorf("failed to check if file exists before writing: %w", err)))
-				return
-			} else if err != nil { // ErrFileNotFound
-				err = repo.Create(ctx, logger, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
+			if _, ok := err.(apierrors.APIStatus); ok {
+				responder.Error(err)
 			} else {
-				err = repo.Update(ctx, logger, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
-			}
-			if err != nil {
-				logger.ErrorContext(ctx, "failed to write dashboard model to repository", "err", err)
-				responder.Error(apierrors.NewInternalError(fmt.Errorf("failed to write file in repo: %w", err)))
-				return
+				responder.Error(apierrors.NewInternalError(fmt.Errorf("failed to create a job from request: %w", err)))
 			}
+		} else {
+			logger.InfoContext(ctx, "created an export job from request",
+				"job", job.GetName(),
+				"jobns", job.GetNamespace())
+			responder.Object(http.StatusOK, job)
 		}
-
-		responder.Object(http.StatusOK, &provisioning.ResourceWrapper{})
 	}), nil
 }
 
-type folderTree struct {
-	tree       map[string]string
-	repoFolder string
-}
-
-func (t *folderTree) In(folder string) bool {
-	_, ok := t.tree[folder]
-	return ok
-}
-
-// DirPath creates the path to the directory with slashes.
-// The repository folder is not included in the path.
-// If In(folder) is false, this will panic, because it would be undefined behaviour.
-func (t *folderTree) DirPath(folder string) string {
-	if folder == t.repoFolder {
-		return ""
-	}
-	if !t.In(folder) {
-		panic("undefined behaviour")
-	}
-
-	dirPath := folder
-	parent := t.tree[folder]
-	for parent != "" && parent != t.repoFolder {
-		dirPath = path.Join(parent, dirPath)
-		parent = t.tree[parent]
-	}
-	// Not using Clean here is intentional. We don't want `.` or similar.
-	return dirPath
-}
-
-func (c *exportConnector) fetchRepoFolderTree(
-	ctx context.Context,
-	iface dynamic.ResourceInterface,
-	repoFolder string,
-) (*folderTree, error) {
-	// TODO: handle pagination
-	rawFolders, err := iface.List(ctx, metav1.ListOptions{})
-	if err != nil {
-		return nil, err
-	}
-
-	folders := make(map[string]string, len(rawFolders.Items))
-	for _, rf := range rawFolders.Items {
-		name := rf.GetName()
-		// TODO: Can I use MetaAccessor here?
-		parent := rf.GetAnnotations()[apiutils.AnnoKeyFolder]
-		folders[name] = parent
-	}
-
-	// folders now includes a map[folder name]parent name
-	// The top-most folder has a parent of "". Any folders below have parent refs.
-	// We want to find only folders which are or start in repoFolder.
-	for folder, parent := range folders {
-		if folder == repoFolder {
-			continue
-		}
-
-		hasRepoRoot := false
-		for parent != "" {
-			if parent == repoFolder {
-				hasRepoRoot = true
-				break
-			}
-			parent = folders[parent]
-		}
-		if !hasRepoRoot {
-			delete(folders, folder)
-		}
-	}
-
-	// folders now only includes the tree of the repoFolder.
-
-	return &folderTree{
-		tree:       folders,
-		repoFolder: repoFolder,
-	}, nil
-}
-
-func (c *exportConnector) marshalPreferredFormat(obj any, name string, repo repository.Repository) (body []byte, fileName string, err error) {
-	if repo.Config().Spec.PreferYAML {
-		body, err = yaml.Marshal(obj)
-		return body, name + ".yaml", err
-	} else {
-		body, err := json.MarshalIndent(obj, "", "    ")
-		return body, name + ".json", err
-	}
-}
-
 var (
 	_ rest.Connecter            = (*exportConnector)(nil)
 	_ rest.Storage              = (*exportConnector)(nil)

pkg/registry/apis/provisioning/import.go

@@ -64,7 +64,7 @@ func (c *importConnector) Connect(
 	}
 	cfg := repo.Config()
 	ns := cfg.GetNamespace()
-	replicatorFactory := resources.NewReplicatorFactory(c.client, ns, repo, c.ignore)
+	replicatorFactory := resources.NewReplicatorFactory(c.client, ns, repo, c.ignore, logger)
 
 	// TODO: We need some way to filter what we import.
 

pkg/registry/apis/provisioning/jobs/queue.go

@@ -117,6 +117,7 @@ func (s *jobStore) drainPending() {
 		if job == nil {
 			return // done
 		}
+		logger := s.logger.With("job", job.GetName(), "namespace", job.GetNamespace())
 
 		var status *provisioning.JobStatus
 		if s.worker == nil {
@@ -129,7 +130,7 @@ func (s *jobStore) drainPending() {
 		} else {
 			status, err = s.worker.Process(ctx, *job)
 			if err != nil {
-				s.logger.Error("error processing job", "job", job.Name, "error", err)
+				logger.ErrorContext(ctx, "error processing job", "error", err)
 				status = &provisioning.JobStatus{
 					State:  provisioning.JobStateError,
 					Errors: []string{err.Error()},
@@ -137,12 +138,14 @@ func (s *jobStore) drainPending() {
 			} else if status.State == "" {
 				status.State = provisioning.JobStateFinished
 			}
+			logger.DebugContext(ctx, "job processing finished", "status", status.State)
 		}
 
 		err = s.Complete(ctx, job.Namespace, job.Name, *status)
 		if err != nil {
-			s.logger.Error("error running job", "job", job.Name, "error", err)
+			logger.ErrorContext(ctx, "error running job", "error", err)
 		}
+		logger.DebugContext(ctx, "job has been fully completed")
 	}
 }
 

pkg/registry/apis/provisioning/jobs/worker.go

@@ -39,6 +39,7 @@ func NewJobWorker(getter RepoGetter, resourceClient *resources.ClientFactory, id
 }
 
 func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisioning.JobStatus, error) {
+	logger := g.logger.With("job", job.GetName(), "namespace", job.GetNamespace())
 	id, err := g.identities.WorkerIdentity(ctx, job.Name)
 	if err != nil {
 		return nil, err
@@ -49,6 +50,7 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisi
 	if !ok {
 		return nil, fmt.Errorf("missing repository name in label")
 	}
+	logger = logger.With("repository", repoName)
 
 	repo, err := g.getter.GetRepository(ctx, repoName)
 	if err != nil {
@@ -58,7 +60,7 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisi
 		return nil, fmt.Errorf("unknown repository")
 	}
 
-	factory := resources.NewReplicatorFactory(g.resourceClient, job.Namespace, repo, g.ignore)
+	factory := resources.NewReplicatorFactory(g.resourceClient, job.Namespace, repo, g.ignore, logger)
 	replicator, err := factory.New()
 	if err != nil {
 		return nil, fmt.Errorf("error creating replicator")
@@ -82,6 +84,11 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job) (*provisi
 		if err != nil {
 			return nil, err
 		}
+	case provisioning.JobActionExport:
+		err := replicator.Export(ctx)
+		if err != nil {
+			return nil, err
+		}
 	default:
 	}
 

pkg/registry/apis/provisioning/register.go

@@ -227,8 +227,8 @@ func (b *ProvisioningAPIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserv
 	}
 	storage[provisioning.RepositoryResourceInfo.StoragePath("export")] = &exportConnector{
 		repoGetter: b,
-		client:     b.client,
 		logger:     b.logger.With("connector", "export"),
+		queue:      b.jobs,
 	}
 	apiGroupInfo.VersionedResourcesStorageMap[provisioning.VERSION] = storage
 	return nil

pkg/registry/apis/provisioning/repository/repository.go

@@ -139,6 +139,7 @@ type FileReplicator interface {
 	ReplicateTree(ctx context.Context, ref string) error
 	DeleteFile(ctx context.Context, fileInfo *FileInfo) error
 	Sync(ctx context.Context) error
+	Export(ctx context.Context) error
 }
 
 // FileReplicatorFactory is an interface for creating FileReplicators

pkg/registry/apis/provisioning/resources/replicator.go

@@ -7,11 +7,13 @@ import (
 	"fmt"
 	"log/slog"
 	"path"
+	"path/filepath"
 	"strings"
 
 	apiutils "github.com/grafana/grafana/pkg/apimachinery/utils"
 	provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
 	"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
+	"gopkg.in/yaml.v3"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -24,14 +26,16 @@ type ReplicatorFactory struct {
 	client    *ClientFactory
 	namespace string
 	ignore    provisioning.IgnoreFile
+	logger    *slog.Logger
 }
 
-func NewReplicatorFactory(client *ClientFactory, namespace string, repo repository.Repository, ignore provisioning.IgnoreFile) *ReplicatorFactory {
+func NewReplicatorFactory(client *ClientFactory, namespace string, repo repository.Repository, ignore provisioning.IgnoreFile, logger *slog.Logger) *ReplicatorFactory {
 	return &ReplicatorFactory{
 		client:    client,
 		namespace: namespace,
 		repo:      repo,
 		ignore:    ignore,
+		logger:    logger,
 	}
 }
 
@@ -50,7 +54,7 @@ func (f *ReplicatorFactory) New() (repository.FileReplicator, error) {
 	})
 
 	return &replicator{
-		logger:     slog.Default().With("logger", "replicator", "namespace", f.namespace),
+		logger:     f.logger,
 		parser:     parser,
 		client:     dynamicClient,
 		folders:    folders,
@@ -324,3 +328,173 @@ func (r *replicator) parseResource(ctx context.Context, fileInfo *repository.Fil
 
 	return file, nil
 }
+
+func (r *replicator) Export(ctx context.Context) error {
+	logger := r.logger
+	dashboardIface := r.client.Resource(schema.GroupVersionResource{
+		Group:    "dashboard.grafana.app",
+		Version:  "v2alpha1",
+		Resource: "dashboards",
+	})
+
+	// TODO: handle pagination
+	folders, err := r.fetchRepoFolderTree(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to list folders: %w", err)
+	}
+
+	// TODO: handle pagination
+	dashboardList, err := dashboardIface.List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to list dashboards: %w", err)
+	}
+
+	for _, item := range dashboardList.Items {
+		if ctx.Err() != nil {
+			logger.DebugContext(ctx, "cancelling replication process due to ctx error", "error", err)
+			return ctx.Err()
+		}
+
+		name := item.GetName()
+		logger := logger.With("item", name)
+		ns := r.repository.Config().GetNamespace()
+		if item.GetNamespace() != ns {
+			logger.DebugContext(ctx, "skipping dashboard item due to mismatching namespace", "got", ns)
+			continue
+		}
+
+		folder := item.GetAnnotations()[apiutils.AnnoKeyFolder]
+		logger = logger.With("folder", folder)
+		if !folders.In(folder) {
+			logger.DebugContext(ctx, "folder of item was not in tree of repository")
+			continue
+		}
+
+		delete(item.Object, "metadata")
+		marshalledBody, baseFileName, err := r.marshalPreferredFormat(item.Object, name, r.repository)
+		if err != nil {
+			return fmt.Errorf("failed to marshal dashboard %s: %w", name, err)
+		}
+		fileName := filepath.Join(folders.DirPath(folder), baseFileName)
+		logger = logger.With("file_name", fileName)
+		if logger.Enabled(ctx, slog.LevelDebug) {
+			bodyStr := string(marshalledBody)
+			logger.DebugContext(ctx, "got marshalled body for item", "body", bodyStr)
+		}
+
+		var ref string
+		if r.repository.Config().Spec.Type == provisioning.GitHubRepositoryType {
+			ref = r.repository.Config().Spec.GitHub.Branch
+		}
+		logger = logger.With("ref", ref)
+
+		_, err = r.repository.Read(ctx, r.logger, fileName, ref)
+		if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {
+			logger.ErrorContext(ctx, "failed to check if file exists before writing", "error", err)
+			return fmt.Errorf("failed to check if file exists before writing: %w", err)
+		} else if err != nil { // ErrFileNotFound
+			err = r.repository.Create(ctx, r.logger, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
+		} else {
+			err = r.repository.Update(ctx, r.logger, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
+		}
+		if err != nil {
+			logger.ErrorContext(ctx, "failed to write a file in repository", "error", err)
+			return fmt.Errorf("failed to write file in repo: %w", err)
+		}
+		logger.DebugContext(ctx, "successfully exported item")
+	}
+
+	return nil
+}
+
+type folderTree struct {
+	tree       map[string]string
+	repoFolder string
+}
+
+func (t *folderTree) In(folder string) bool {
+	_, ok := t.tree[folder]
+	return ok
+}
+
+// DirPath creates the path to the directory with slashes.
+// The repository folder is not included in the path.
+// If In(folder) is false, this will panic, because it would be undefined behaviour.
+func (t *folderTree) DirPath(folder string) string {
+	if folder == t.repoFolder {
+		return ""
+	}
+	if !t.In(folder) {
+		panic("undefined behaviour")
+	}
+
+	dirPath := folder
+	parent := t.tree[folder]
+	for parent != "" && parent != t.repoFolder {
+		dirPath = path.Join(parent, dirPath)
+		parent = t.tree[parent]
+	}
+	// Not using Clean here is intentional. We don't want `.` or similar.
+	return dirPath
+}
+
+func (r *replicator) fetchRepoFolderTree(ctx context.Context) (*folderTree, error) {
+	iface := r.client.Resource(schema.GroupVersionResource{
+		Group:    "folder.grafana.app",
+		Version:  "v0alpha1",
+		Resource: "folders",
+	})
+
+	// TODO: handle pagination
+	rawFolders, err := iface.List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	folders := make(map[string]string, len(rawFolders.Items))
+	for _, rf := range rawFolders.Items {
+		name := rf.GetName()
+		// TODO: Can I use MetaAccessor here?
+		parent := rf.GetAnnotations()[apiutils.AnnoKeyFolder]
+		folders[name] = parent
+	}
+
+	// folders now includes a map[folder name]parent name
+	// The top-most folder has a parent of "". Any folders below have parent refs.
+	// We want to find only folders which are or start in repoFolder.
+	repoFolder := r.repository.Config().Spec.Folder
+	for folder, parent := range folders {
+		if folder == repoFolder {
+			continue
+		}
+
+		hasRepoRoot := false
+		for parent != "" {
+			if parent == repoFolder {
+				hasRepoRoot = true
+				break
+			}
+			parent = folders[parent]
+		}
+		if !hasRepoRoot {
+			delete(folders, folder)
+		}
+	}
+
+	// folders now only includes the tree of the repoFolder.
+
+	return &folderTree{
+		tree:       folders,
+		repoFolder: repoFolder,
+	}, nil
+}
+
+func (*replicator) marshalPreferredFormat(obj any, name string, repo repository.Repository) (body []byte, fileName string, err error) {
+	if repo.Config().Spec.PreferYAML {
+		body, err = yaml.Marshal(obj)
+		return body, name + ".yaml", err
+	} else {
+		body, err := json.MarshalIndent(obj, "", "    ")
+		return body, name + ".json", err
+	}
+}

pkg/registry/apis/provisioning/resources/replicator_test.go

@@ -1,4 +1,4 @@
-package provisioning
+package resources
 
 import (
 	"testing"