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

split replicator

Browse Source
This commit is contained in:
Ryan McKinley committed 2025-01-22 20:54:16 +03:00
1 parent a92227dd26
commit eab86bb7d2
8 files changed
+297 -226
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/registry/apis/provisioning/jobs/exporter.go
+166 -27
View File
@@ -2,53 +2,192 @@ package jobs
import (
"context"
"encoding/json"
"errors"
"fmt"
"math/rand/v2"
"time"
"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/schema"
"k8s.io/client-go/dynamic"
"github.com/grafana/grafana-app-sdk/logging"
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/util"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
)
type dummyExporter struct{}
type Exporter interface {
Export(ctx context.Context,
repo repository.Repository,
options provisioning.ExportOptions,
progress func(provisioning.JobStatus) error,
) (*provisioning.JobStatus, error)
}
func (s *dummyExporter) Export(ctx context.Context,
type exporter struct {
client *resources.DynamicClient
folders dynamic.ResourceInterface
repository repository.Repository
}
func NewExporter(
repo repository.Repository,
parser *resources.Parser,
) (Exporter, error) {
dynamicClient := parser.Client()
folders := dynamicClient.Resource(schema.GroupVersionResource{
Group: "folder.grafana.app",
Version: "v0alpha1",
Resource: "folders",
})
return &exporter{
client: dynamicClient,
folders: folders,
repository: repo,
}, nil
}
func (r *exporter) Export(ctx context.Context,
repo repository.Repository,
options provisioning.ExportOptions,
progress func(provisioning.JobStatus) error,
) (*provisioning.JobStatus, error) {
logger := logging.FromContext(ctx).With("logger", "exporter", "repository", repo.Config().Name, "namespace", repo.Config().Name)
logger.Info("start export", "folder", options.Folder)
logger := logging.FromContext(ctx)
dashboardIface := r.client.Resource(schema.GroupVersionResource{
Group: "dashboard.grafana.app",
Version: "v2alpha1",
Resource: "dashboards",
})
size := 5 + rand.IntN(15)
status := provisioning.JobStatus{
State: provisioning.JobStateWorking,
Message: "exporting..." + repo.Config().Spec.Title,
Progress: 0,
// TODO: handle pagination
folders, err := r.fetchRepoFolderTree(ctx)
if err != nil {
return nil, fmt.Errorf("failed to list folders: %w", err)
}
err := progress(status)
// TODO: handle pagination
dashboardList, err := dashboardIface.List(ctx, metav1.ListOptions{})
if err != nil {
return nil, fmt.Errorf("failed to list dashboards: %w", err)
}
for _, item := range dashboardList.Items {
if ctx.Err() != nil {
logger.Debug("cancelling replication process due to ctx error", "error", err)
return nil, ctx.Err()
}
name := item.GetName()
logger := logger.With("item", name)
ns := r.repository.Config().GetNamespace()
if item.GetNamespace() != ns {
logger.Debug("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.Debug("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 nil, fmt.Errorf("failed to marshal dashboard %s: %w", name, err)
}
fileName := filepath.Join(folders.DirPath(folder), baseFileName)
logger = logger.With("file", fileName)
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, fileName, ref)
if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {
logger.Error("failed to check if file exists before writing", "error", err)
return nil, fmt.Errorf("failed to check if file exists before writing: %w", err)
} else if err != nil { // ErrFileNotFound
err = r.repository.Create(ctx, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
} else {
err = r.repository.Update(ctx, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
}
if err != nil {
logger.Error("failed to write a file in repository", "error", err)
return nil, fmt.Errorf("failed to write file in repo: %w", err)
}
logger.Debug("successfully exported item")
}
return &provisioning.JobStatus{
State: provisioning.JobStateSuccess,
}, nil
}
func (r *exporter) 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
}
}
func (r *exporter) 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
}
time.Sleep(200 * time.Millisecond)
for i := 0; i < size; i++ {
status.Progress = (float64(i) / float64(size)) * 100
sleep := time.Duration(400+rand.IntN(800)) * time.Millisecond
status.Message, _ = util.GetRandomString(rand.IntN(10 + 15))
time.Sleep(sleep)
err = progress(status)
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)
}
}
status.State = provisioning.JobStateSuccess
status.Message = fmt.Sprintf("pretended to export %d", size)
status.Progress = 0
// folders now only includes the tree of the repoFolder.
logger.Info("finished export", "folder", options.Folder)
return &status, nil
return &folderTree{
tree: folders,
repoFolder: repoFolder,
}, nil
}
pkg/registry/apis/provisioning/jobs/exporter_dummy.go
+54
View File
@@ -0,0 +1,54 @@
package jobs
import (
"context"
"fmt"
"math/rand/v2"
"time"
"github.com/grafana/grafana-app-sdk/logging"
provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
"github.com/grafana/grafana/pkg/util"
)
type dummyExporter struct{}
func (s *dummyExporter) Export(ctx context.Context,
repo repository.Repository,
options provisioning.ExportOptions,
progress func(provisioning.JobStatus) error,
) (*provisioning.JobStatus, error) {
logger := logging.FromContext(ctx).With("logger", "exporter", "repository", repo.Config().Name, "namespace", repo.Config().Name)
logger.Info("start export", "folder", options.Folder)
size := 5 + rand.IntN(15)
status := provisioning.JobStatus{
State: provisioning.JobStateWorking,
Message: "exporting..." + repo.Config().Spec.Title,
Progress: 0,
}
err := progress(status)
if err != nil {
return nil, err
}
time.Sleep(200 * time.Millisecond)
for i := 0; i < size; i++ {
status.Progress = (float64(i) / float64(size)) * 100
sleep := time.Duration(400+rand.IntN(800)) * time.Millisecond
status.Message, _ = util.GetRandomString(rand.IntN(10 + 15))
time.Sleep(sleep)
err = progress(status)
if err != nil {
return nil, err
}
}
status.State = provisioning.JobStateSuccess
status.Message = fmt.Sprintf("pretended to export %d", size)
status.Progress = 0
logger.Info("finished export", "folder", options.Folder)
return &status, nil
}
pkg/registry/apis/provisioning/resources/replicator.go → pkg/registry/apis/provisioning/jobs/syncer.go
+19 -187
View File
@@ -1,15 +1,12 @@
package resources
package jobs
import (
"context"
"encoding/json"
"errors"
"fmt"
"path"
"path/filepath"
"strings"
"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"
@@ -22,23 +19,24 @@ import (
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"
)
type UnsyncOptions struct {
KeepDashboards bool
}
type Replicator struct {
client *DynamicClient
parser *Parser
type Syncer struct {
client *resources.DynamicClient
parser *resources.Parser
folders dynamic.ResourceInterface
repository repository.Repository
}
func NewReplicator(
func NewSyncer(
repo repository.Repository,
parser *Parser,
) (*Replicator, error) {
parser *resources.Parser,
) (*Syncer, error) {
dynamicClient := parser.Client()
folders := dynamicClient.Resource(schema.GroupVersionResource{
Group: "folder.grafana.app",
@@ -46,7 +44,7 @@ func NewReplicator(
Resource: "folders",
})
return &Replicator{
return &Syncer{
parser: parser,
client: dynamicClient,
folders: folders,
@@ -55,7 +53,7 @@ func NewReplicator(
}
// Sync replicates all files in the repository.
func (r *Replicator) Sync(ctx context.Context) (string, error) {
func (r *Syncer) Sync(ctx context.Context) (string, error) {
// FIXME: how to handle the scenario in which folder changes?
cfg := r.repository.Config()
lastCommit := cfg.Status.Sync.Hash
@@ -104,7 +102,7 @@ func (r *Replicator) Sync(ctx context.Context) (string, error) {
return latest, nil
}
func (r *Replicator) Unsync(ctx context.Context, opts UnsyncOptions) error {
func (r *Syncer) Unsync(ctx context.Context, opts UnsyncOptions) error {
cfg := r.repository.Config()
logger := logging.FromContext(ctx)
@@ -138,7 +136,7 @@ func (r *Replicator) Unsync(ctx context.Context, opts UnsyncOptions) error {
}
// replicateTree replicates all files in the repository.
func (r *Replicator) replicateTree(ctx context.Context, ref string) error {
func (r *Syncer) replicateTree(ctx context.Context, ref string) error {
tree, err := r.repository.ReadTree(ctx, ref)
if err != nil {
return fmt.Errorf("read tree: %w", err)
@@ -166,7 +164,7 @@ func (r *Replicator) replicateTree(ctx context.Context, ref string) error {
info.Modified = nil // modified?
if err := r.replicateFile(ctx, info); err != nil {
if errors.Is(err, ErrUnableToReadResourceBytes) {
if errors.Is(err, resources.ErrUnableToReadResourceBytes) {
logger.Info("file does not contain a resource")
continue
}
@@ -179,7 +177,7 @@ func (r *Replicator) replicateTree(ctx context.Context, ref string) error {
// replicateFile creates a new resource in the cluster.
// If the resource already exists, it will be updated.
func (r *Replicator) replicateFile(ctx context.Context, fileInfo *repository.FileInfo) error {
func (r *Syncer) replicateFile(ctx context.Context, fileInfo *repository.FileInfo) error {
logger := logging.FromContext(ctx).With("file", fileInfo.Path, "ref", fileInfo.Ref)
file, err := r.parseResource(ctx, fileInfo)
if err != nil {
@@ -241,7 +239,7 @@ func (r *Replicator) replicateFile(ctx context.Context, fileInfo *repository.Fil
return nil
}
func (r *Replicator) createFolderPath(ctx context.Context, filePath string) (string, error) {
func (r *Syncer) createFolderPath(ctx context.Context, filePath string) (string, error) {
dir := path.Dir(filePath)
parent := r.repository.Config().Spec.Folder
if dir == "." || dir == "/" {
@@ -304,7 +302,7 @@ func (r *Replicator) createFolderPath(ctx context.Context, filePath string) (str
return parent, nil
}
func (r *Replicator) replicateChanges(ctx context.Context, changes []repository.FileChange) error {
func (r *Syncer) replicateChanges(ctx context.Context, changes []repository.FileChange) error {
for _, change := range changes {
if r.parser.ShouldIgnore(change.Path) {
continue
@@ -343,7 +341,7 @@ func (r *Replicator) replicateChanges(ctx context.Context, changes []repository.
return nil
}
func (r *Replicator) deleteFile(ctx context.Context, fileInfo *repository.FileInfo) error {
func (r *Syncer) deleteFile(ctx context.Context, fileInfo *repository.FileInfo) error {
file, err := r.parseResource(ctx, fileInfo)
if err != nil {
return err
@@ -368,7 +366,7 @@ func (r *Replicator) deleteFile(ctx context.Context, fileInfo *repository.FileIn
return nil
}
func (r *Replicator) parseResource(ctx context.Context, fileInfo *repository.FileInfo) (*ParsedResource, error) {
func (r *Syncer) parseResource(ctx context.Context, fileInfo *repository.FileInfo) (*resources.ParsedResource, error) {
file, err := r.parser.Parse(ctx, fileInfo, true)
if err != nil {
return nil, fmt.Errorf("failed to parse file %s: %w", fileInfo.Path, err)
@@ -385,163 +383,7 @@ func (r *Replicator) parseResource(ctx context.Context, fileInfo *repository.Fil
return file, nil
}
func (r *Replicator) Export(ctx context.Context) error {
logger := logging.FromContext(ctx)
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.Debug("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.Debug("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.Debug("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", fileName)
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, fileName, ref)
if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {
logger.Error("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, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
} else {
err = r.repository.Update(ctx, fileName, ref, marshalledBody, "export of dashboard "+name+" in namespace "+ns)
}
if err != nil {
logger.Error("failed to write a file in repository", "error", err)
return fmt.Errorf("failed to write file in repo: %w", err)
}
logger.Debug("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 (r *Replicator) ensureRepositoryFolderExists(ctx context.Context) error {
func (r *Syncer) ensureRepositoryFolderExists(ctx context.Context) error {
if r.repository.Config().Spec.Folder == "" {
return nil
}
@@ -603,13 +445,3 @@ func (r *Replicator) ensureRepositoryFolderExists(ctx context.Context) error {
return 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/jobs/utils.go
+34
View File
@@ -0,0 +1,34 @@
package jobs
import "path"
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
}
pkg/registry/apis/provisioning/resources/replicator_test.go → pkg/registry/apis/provisioning/jobs/utils_test.go
+1 -1
View File
@@ -1,4 +1,4 @@
package resources
package jobs
import (
"testing"
pkg/registry/apis/provisioning/jobs/worker.go
+20 -8
View File
@@ -86,11 +86,6 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress
return nil, fmt.Errorf("failed to get parser for %s: %w", repo.Config().Name, err)
}
replicator, err := resources.NewReplicator(repo, parser)
if err != nil {
return nil, fmt.Errorf("error creating replicator")
}
switch job.Spec.Action {
case provisioning.JobActionSync:
started := time.Now()
@@ -108,8 +103,13 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress
return nil, fmt.Errorf("update repository status: %w", err)
}
syncer, err := NewSyncer(repo, parser)
if err != nil {
return nil, fmt.Errorf("error creating replicator")
}
// Sync the repository
ref, syncError := replicator.Sync(ctx)
ref, syncError := syncer.Sync(ctx)
status = &provisioning.SyncStatus{
State: provisioning.JobStateSuccess,
JobID: job.GetName(),
@@ -168,14 +168,26 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress
return nil, fmt.Errorf("error processing pull request: %w", err)
}
case provisioning.JobActionExport:
dummy := &dummyExporter{}
if job.Spec.Export == nil {
return &provisioning.JobStatus{
State: provisioning.JobStateError,
Errors: []string{"Export job missing export settings"},
}, nil
}
return dummy.Export(ctx, repo, *job.Spec.Export, progress)
var exporter Exporter
// Test for now... so we have something with long spinners for UI testing!!!
if job.Spec.Export.Branch == "*dummy*" {
exporter = &dummyExporter{}
} else {
exporter, err = NewExporter(repo, parser)
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/repository_controller.go
+2 -2
View File
@@ -246,12 +246,12 @@ func (rc *RepositoryController) process(item *queueItem) error {
return fmt.Errorf("failed to get parser for %s: %w", repo.Config().Name, err)
}
replicator, err := resources.NewReplicator(repo, parser)
replicator, err := jobs.NewSyncer(repo, parser)
if err != nil {
return fmt.Errorf("error creating replicator")
}
if err := replicator.Unsync(ctx, resources.UnsyncOptions{}); err != nil {
if err := replicator.Unsync(ctx, jobs.UnsyncOptions{}); err != nil {
return fmt.Errorf("unsync repository: %w", err)
}
public/app/features/provisioning/ExportToRepository.tsx
+1 -1
View File
@@ -86,7 +86,7 @@ export function ExportToRepository({ repo }: Props) {
exportRepo({
name: repo.metadata?.name!,
body: {
branch: 'my-branch',
branch: '*dummy*', // << triggers a fake exporter
folder: 'some-folder',
history: true,
prefix: 'prefix/in/remote/tree',