diff --git a/pkg/registry/apis/provisioning/export.go b/pkg/registry/apis/provisioning/export.go index 988aec21191..adbb6872da9 100644 --- a/pkg/registry/apis/provisioning/export.go +++ b/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) diff --git a/pkg/registry/apis/provisioning/import.go b/pkg/registry/apis/provisioning/import.go index a2c74725ac6..1e945a5baa8 100644 --- a/pkg/registry/apis/provisioning/import.go +++ b/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. diff --git a/pkg/registry/apis/provisioning/jobs/queue.go b/pkg/registry/apis/provisioning/jobs/queue.go index 489a07c2265..a4c45606381 100644 --- a/pkg/registry/apis/provisioning/jobs/queue.go +++ b/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") } } diff --git a/pkg/registry/apis/provisioning/jobs/worker.go b/pkg/registry/apis/provisioning/jobs/worker.go index f6cfad44c54..5d01abef695 100644 --- a/pkg/registry/apis/provisioning/jobs/worker.go +++ b/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: } diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 937c497da77..ee6e9ac549b 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/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 diff --git a/pkg/registry/apis/provisioning/repository/repository.go b/pkg/registry/apis/provisioning/repository/repository.go index 83b37a681b2..b305c26f9c9 100644 --- a/pkg/registry/apis/provisioning/repository/repository.go +++ b/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 diff --git a/pkg/registry/apis/provisioning/resources/replicator.go b/pkg/registry/apis/provisioning/resources/replicator.go index bd3ad22c6b5..fab553e81c7 100644 --- a/pkg/registry/apis/provisioning/resources/replicator.go +++ b/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 + } +} diff --git a/pkg/registry/apis/provisioning/export_test.go b/pkg/registry/apis/provisioning/resources/replicator_test.go similarity index 97% rename from pkg/registry/apis/provisioning/export_test.go rename to pkg/registry/apis/provisioning/resources/replicator_test.go index 8ed203ea67e..12bcd19a5df 100644 --- a/pkg/registry/apis/provisioning/export_test.go +++ b/pkg/registry/apis/provisioning/resources/replicator_test.go @@ -1,4 +1,4 @@ -package provisioning +package resources import ( "testing"