diff --git a/pkg/apis/provisioning/v0alpha1/jobs.go b/pkg/apis/provisioning/v0alpha1/jobs.go index 747907e62e5..7e60bafd6bb 100644 --- a/pkg/apis/provisioning/v0alpha1/jobs.go +++ b/pkg/apis/provisioning/v0alpha1/jobs.go @@ -86,8 +86,8 @@ type PullRequestJobOptions struct { } type SyncJobOptions struct { - // Complete forces the sync to overwrite - Complete bool `json:"complete,omitempty"` + // Incremental synchronization for versioned repositories + Incremental bool `json:"incremental"` } type ExportJobOptions struct { diff --git a/pkg/apis/provisioning/v0alpha1/types.go b/pkg/apis/provisioning/v0alpha1/types.go index 02d71ffdf3f..a421acc37a8 100644 --- a/pkg/apis/provisioning/v0alpha1/types.go +++ b/pkg/apis/provisioning/v0alpha1/types.go @@ -185,6 +185,9 @@ type SyncStatus struct { // The repository hash when the last sync ran Hash string `json:"hash,omitempty"` + + // Incremental synchronization for versioned repositories + Incremental bool `json:"incremental,omitempty"` } type WebhookStatus struct { diff --git a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go index 1266c57a65e..94e8e9247af 100644 --- a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go +++ b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go @@ -1606,14 +1606,16 @@ func schema_pkg_apis_provisioning_v0alpha1_SyncJobOptions(ref common.ReferenceCa SchemaProps: spec.SchemaProps{ Type: []string{"object"}, Properties: map[string]spec.Schema{ - "complete": { + "incremental": { SchemaProps: spec.SchemaProps{ - Description: "Complete forces the sync to overwrite", + Description: "Incremental synchronization for versioned repositories", + Default: false, Type: []string{"boolean"}, Format: "", }, }, }, + Required: []string{"incremental"}, }, }, } @@ -1726,6 +1728,13 @@ func schema_pkg_apis_provisioning_v0alpha1_SyncStatus(ref common.ReferenceCallba Format: "", }, }, + "incremental": { + SchemaProps: spec.SchemaProps{ + Description: "Incremental synchronization for versioned repositories", + Type: []string{"boolean"}, + Format: "", + }, + }, }, Required: []string{"state", "message"}, }, diff --git a/pkg/generated/applyconfiguration/provisioning/v0alpha1/syncstatus.go b/pkg/generated/applyconfiguration/provisioning/v0alpha1/syncstatus.go index 76129b21998..9db1d73f054 100644 --- a/pkg/generated/applyconfiguration/provisioning/v0alpha1/syncstatus.go +++ b/pkg/generated/applyconfiguration/provisioning/v0alpha1/syncstatus.go @@ -11,13 +11,14 @@ import ( // SyncStatusApplyConfiguration represents a declarative configuration of the SyncStatus type for use // with apply. type SyncStatusApplyConfiguration struct { - State *provisioningv0alpha1.JobState `json:"state,omitempty"` - JobID *string `json:"job,omitempty"` - Started *int64 `json:"started,omitempty"` - Finished *int64 `json:"finished,omitempty"` - Scheduled *int64 `json:"scheduled,omitempty"` - Message []string `json:"message,omitempty"` - Hash *string `json:"hash,omitempty"` + State *provisioningv0alpha1.JobState `json:"state,omitempty"` + JobID *string `json:"job,omitempty"` + Started *int64 `json:"started,omitempty"` + Finished *int64 `json:"finished,omitempty"` + Scheduled *int64 `json:"scheduled,omitempty"` + Message []string `json:"message,omitempty"` + Hash *string `json:"hash,omitempty"` + Incremental *bool `json:"incremental,omitempty"` } // SyncStatusApplyConfiguration constructs a declarative configuration of the SyncStatus type for use with @@ -83,3 +84,11 @@ func (b *SyncStatusApplyConfiguration) WithHash(value string) *SyncStatusApplyCo b.Hash = &value return b } + +// WithIncremental sets the Incremental field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the Incremental field is set to the value of the last call. +func (b *SyncStatusApplyConfiguration) WithIncremental(value bool) *SyncStatusApplyConfiguration { + b.Incremental = &value + return b +} diff --git a/pkg/registry/apis/provisioning/files.go b/pkg/registry/apis/provisioning/files.go index f28d1fad3be..46847469618 100644 --- a/pkg/registry/apis/provisioning/files.go +++ b/pkg/registry/apis/provisioning/files.go @@ -154,6 +154,9 @@ func (s *filesConnector) doRead(ctx context.Context, repo repository.Repository, if err != nil { return 0, nil, err } + if strings.HasPrefix(info.Path, "/") { + return 0, nil, fmt.Errorf("repository path must be relative to the root") + } parser, err := s.parsers.GetParser(ctx, repo) if err != nil { diff --git a/pkg/registry/apis/provisioning/jobs/changes.go b/pkg/registry/apis/provisioning/jobs/changes.go new file mode 100644 index 00000000000..75247248ee3 --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/changes.go @@ -0,0 +1,93 @@ +package jobs + +import ( + "fmt" + "sort" + "strings" + + folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1" + 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 ResourceFileChange struct { + // Path to the file in a repository with a change + Path string + Action repository.FileAction + + // The current value in the database -- only required for delete + Existing *provisioning.ResourceListItem +} + +func Changes(source []repository.FileTreeEntry, target *provisioning.ResourceList) ([]ResourceFileChange, error) { + lookup := make(map[string]*provisioning.ResourceListItem, len(target.Items)) + for _, item := range target.Items { + if item.Path == "" { + if item.Group != folders.GROUP { + return nil, fmt.Errorf("empty path on a non folder") + } + continue + } + lookup[item.Path] = &item + } + + var keep []string + changes := make([]ResourceFileChange, 0, len(source)) + for _, file := range source { + if !file.Blob { + continue // skip folder references? + } + + check, ok := lookup[file.Path] + if ok { + if check.Hash != file.Hash { + changes = append(changes, ResourceFileChange{ + Action: repository.FileActionUpdated, + Path: check.Path, + Existing: check, + }) + } + keep = append(keep, file.Path) + delete(lookup, file.Path) + continue + } + + if !resources.ShouldIgnorePath(file.Path) { + changes = append(changes, ResourceFileChange{ + Action: repository.FileActionCreated, // or previously ignored/failed + Path: file.Path, + }) + } + } + + // nested nested loop :grimmice: trie? + hasFolder := func(p string) bool { + for _, k := range keep { + if strings.HasPrefix(k, p) { + return true + } + } + return false + } + + // Paths found in grafana, without a matching path in the repository + for _, v := range lookup { + if v.Resource == folders.RESOURCE && hasFolder(v.Path) { + continue + } + + changes = append(changes, ResourceFileChange{ + Action: repository.FileActionDeleted, + Path: v.Path, + Existing: v, + }) + } + + // Longest first (stable sort order) + sort.Slice(changes, func(i, j int) bool { + return len(changes[i].Path) > len(changes[j].Path) + }) + + return changes, nil +} diff --git a/pkg/registry/apis/provisioning/jobs/changes_test.go b/pkg/registry/apis/provisioning/jobs/changes_test.go new file mode 100644 index 00000000000..d06ab791fe4 --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/changes_test.go @@ -0,0 +1,145 @@ +package jobs + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" + + provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" +) + +func TestChanges(t *testing.T) { + t.Run("start the same", func(t *testing.T) { + source, target := getBase(t) + changes, err := Changes(source, target) + require.NoError(t, err) + require.Empty(t, changes) + }) + + t.Run("create a source file", func(t *testing.T) { + source, target := getBase(t) + source = append(source, repository.FileTreeEntry{ + Path: "muta.json", Hash: "xyz", Blob: true, + }) + + changes, err := Changes(source, target) + require.NoError(t, err) + require.Len(t, changes, 1) + require.Equal(t, ResourceFileChange{ + Action: repository.FileActionCreated, + Path: "muta.json", + }, changes[0]) + }) + + t.Run("delete a source file", func(t *testing.T) { + source, target := getBase(t) + source = []repository.FileTreeEntry{source[0]} + + changes, err := Changes(source, target) + require.NoError(t, err) + require.Len(t, changes, 1) + require.Equal(t, ResourceFileChange{ + Action: repository.FileActionDeleted, + Path: "adsl62h.yaml", + Existing: &provisioning.ResourceListItem{ + Path: "adsl62h.yaml", + Group: "dashboard.grafana.app", + Resource: "dashboards", + Name: "adsl62h-hrw-f-fvlt2dghp-gufrc4lisksgmq-c", + Hash: "ce5d497c4deadde6831162ce8509e2b2b1776237", + }, + }, changes[0]) + }) + + t.Run("folder deletion order", func(t *testing.T) { + source := []repository.FileTreeEntry{ + {Path: "x/y/z/ignored.md"}, // ignored + {Path: "aaa/bbb.yaml", Hash: "xyz", Blob: true}, + } + target := &provisioning.ResourceList{ + Items: []provisioning.ResourceListItem{ + {Path: "a.json"}, + {Path: "x/y/file.json"}, + {Path: "aaa/", Resource: "folders"}, // Folder... no action required + {Path: "aaa/bbb.yaml", Hash: "xyz"}, + {Path: "zzz/longest/path/here.json"}, + {Path: "short/file.yml"}, + }, + } + changes, err := Changes(source, target) + require.NoError(t, err) + + order := make([]string, len(changes)) + for i := range changes { + order[i] = changes[i].Path + } + require.Equal(t, []string{ + "zzz/longest/path/here.json", // not sorted yet + "short/file.yml", + "x/y/file.json", + "a.json", + }, order) + }) + + t.Run("modify a file", func(t *testing.T) { + source, target := getBase(t) + source[1].Hash = "different" + + changes, err := Changes(source, target) + require.NoError(t, err) + require.Len(t, changes, 1) + require.Equal(t, ResourceFileChange{ + Action: repository.FileActionUpdated, + Path: "adsl62h.yaml", + Existing: &provisioning.ResourceListItem{ + Path: "adsl62h.yaml", + Group: "dashboard.grafana.app", + Resource: "dashboards", + Name: "adsl62h-hrw-f-fvlt2dghp-gufrc4lisksgmq-c", + Hash: "ce5d497c4deadde6831162ce8509e2b2b1776237", + }, + }, changes[0]) + }) +} + +func getBase(t *testing.T) (source []repository.FileTreeEntry, target *provisioning.ResourceList) { + target = &provisioning.ResourceList{} + err := json.Unmarshal([]byte(`{ + "kind": "ResourceList", + "apiVersion": "provisioning.grafana.app/v0alpha1", + "metadata": {}, + "items": [ + { + "path": "", + "group": "folder.grafana.app", + "resource": "folders", + "name": "simplelocal-3794ab9", + "hash": "" + }, + { + "path": "ad4lwp2.yaml", + "group": "dashboard.grafana.app", + "resource": "dashboards", + "name": "ad4lwp2-xofjsuo-mr5blr1zwimlfi0ds0pyrrpd", + "hash": "ca83d64b9c4a23fed975aacdf47e7de8878b4ae0" + }, + { + "path": "adsl62h.yaml", + "group": "dashboard.grafana.app", + "resource": "dashboards", + "name": "adsl62h-hrw-f-fvlt2dghp-gufrc4lisksgmq-c", + "hash": "ce5d497c4deadde6831162ce8509e2b2b1776237" + } + ] + }`), target) + require.NoError(t, err) + + source = []repository.FileTreeEntry{ + {Path: "ad4lwp2.yaml", Hash: "ca83d64b9c4a23fed975aacdf47e7de8878b4ae0", Blob: true}, + {Path: "adsl62h.yaml", Hash: "ce5d497c4deadde6831162ce8509e2b2b1776237", Blob: true}, + } + + return // named values! +} diff --git a/pkg/registry/apis/provisioning/jobs/pullrequest.go b/pkg/registry/apis/provisioning/jobs/pullrequest.go index 20d7073a58a..476a6bb70c6 100644 --- a/pkg/registry/apis/provisioning/jobs/pullrequest.go +++ b/pkg/registry/apis/provisioning/jobs/pullrequest.go @@ -62,7 +62,7 @@ Click the preview links above to view how your changes will look and compare the type PullRequestRepo interface { Config() *provisioning.Repository Read(ctx context.Context, path, ref string) (*repository.FileInfo, error) - CompareFiles(ctx context.Context, base, ref string) ([]repository.FileChange, error) + CompareFiles(ctx context.Context, base, ref string) ([]repository.VersionedFileChange, error) ClearAllPullRequestFileComments(ctx context.Context, pr int) error CommentPullRequestFile(ctx context.Context, pr int, path string, ref string, comment string) error CommentPullRequest(ctx context.Context, pr int, comment string) error diff --git a/pkg/registry/apis/provisioning/jobs/syncer.go b/pkg/registry/apis/provisioning/jobs/syncer.go index a323c2999db..22508118edc 100644 --- a/pkg/registry/apis/provisioning/jobs/syncer.go +++ b/pkg/registry/apis/provisioning/jobs/syncer.go @@ -1,24 +1,23 @@ package jobs import ( + "bytes" "context" - "errors" "fmt" "path" "sort" "strings" + "time" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/dynamic" "github.com/grafana/grafana-app-sdk/logging" "github.com/grafana/grafana/pkg/apimachinery/utils" - dashboards "github.com/grafana/grafana/pkg/apis/dashboard" - dashboardsv2alpha1 "github.com/grafana/grafana/pkg/apis/dashboard/v2alpha1" + "github.com/grafana/grafana/pkg/apis/dashboard" folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" @@ -26,444 +25,423 @@ import ( "github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath" ) -type Syncer struct { - client *resources.DynamicClient - parser *resources.Parser - lister resources.ResourceLister - folders dynamic.ResourceInterface - repository repository.Repository +// Sync will make grafana look like the contents of the repository +// it will add/remove/update resources in grafana so the results look like a mirror +type Syncer interface { + Sync(ctx context.Context, + repo repository.Repository, + options provisioning.SyncJobOptions, + progress func(provisioning.JobStatus) error, + ) (*provisioning.JobStatus, *provisioning.SyncStatus, error) } -func NewSyncer( +// syncer will start sync jobs +type syncer struct { + parsers *resources.ParserFactory + lister resources.ResourceLister +} + +// start a job and run it +func (r *syncer) Sync(ctx context.Context, repo repository.Repository, - lister resources.ResourceLister, - parser *resources.Parser, -) (*Syncer, error) { - dynamicClient := parser.Client() - folders := dynamicClient.Resource(schema.GroupVersionResource{ - Group: folders.GROUP, - Version: folders.VERSION, - Resource: folders.RESOURCE, - }) - - return &Syncer{ - parser: parser, - client: dynamicClient, - lister: lister, - folders: folders, - repository: repo, - }, nil -} - -// Sync replicates all files in the repository. -func (r *Syncer) Sync(ctx context.Context, complete bool) (string, error) { - // FIXME: how to handle the scenario in which folder changes? - cfg := r.repository.Config() + options provisioning.SyncJobOptions, + progress func(provisioning.JobStatus) error, +) (*provisioning.JobStatus, *provisioning.SyncStatus, error) { + cfg := repo.Config() if !cfg.Spec.Sync.Enabled { - return "", fmt.Errorf("sync is not enabled") + return &provisioning.JobStatus{ + State: provisioning.JobStateError, + Message: "sync is not enabled", + }, nil, nil } - lastCommit := cfg.Status.Sync.Hash - versionedRepo, isVersioned := r.repository.(repository.VersionedRepository) + parser, err := r.parsers.GetParser(ctx, repo) + if err != nil { + return nil, nil, fmt.Errorf("failed to get parser for %s: %w", cfg.Name, err) + } + dynamicClient := parser.Client() + if repo.Config().Namespace != dynamicClient.GetNamespace() { + return nil, nil, fmt.Errorf("namespace mismatch") + } + job := &syncJob{ + repository: repo, + options: options, + progress: progress, + progressInterval: time.Second * 15, // how often we update the status + progressLast: time.Now(), + parser: parser, + lister: r.lister, + logger: logging.FromContext(ctx), + jobStatus: &provisioning.JobStatus{ + State: provisioning.JobStateWorking, + }, + syncStatus: &provisioning.SyncStatus{}, + folders: dynamicClient.Resource(schema.GroupVersionResource{ + Group: folders.GROUP, + Version: folders.VERSION, + Resource: folders.RESOURCE, + }), + dashboards: dynamicClient.Resource(schema.GroupVersionResource{ + Group: dashboard.GROUP, + Version: "v1alpha1", + Resource: dashboard.DASHBOARD_RESOURCE, + }), + } - // Ensure the configured folder exists and it's managed by the repository + // Execute the job + err = job.run(ctx) + if err != nil { + job.logger.Warn("error running job", "err", err) + job.jobStatus.State = provisioning.JobStateError + job.jobStatus.Message = "Sync error: " + err.Error() + job.syncStatus.State = job.jobStatus.State + job.syncStatus.Message = append(job.syncStatus.Message, job.jobStatus.Message) + } else if len(job.jobStatus.Errors) > 0 { + job.jobStatus.State = provisioning.JobStateError + } else if !job.jobStatus.State.Finished() { + job.jobStatus.State = provisioning.JobStateSuccess + } + job.jobStatus.Summary = []provisioning.JobResourceSummary{job.summary} + return job.jobStatus, job.syncStatus, nil +} + +// created once for each sync execution +type syncJob struct { + repository repository.Repository + options provisioning.SyncJobOptions + progress func(provisioning.JobStatus) error + progressInterval time.Duration + progressLast time.Time + logger logging.Logger + + parser *resources.Parser + lister resources.ResourceLister + folders dynamic.ResourceInterface + dashboards dynamic.ResourceInterface + folderLookup *resources.FolderTree + + jobStatus *provisioning.JobStatus + syncStatus *provisioning.SyncStatus + + // generic summary for now (not typed) + summary provisioning.JobResourceSummary +} + +func (r *syncJob) run(ctx context.Context) error { + // Ensure the configured folder exists and is managed by the repository + cfg := r.repository.Config() rootFolder := resources.RootFolder(cfg) if rootFolder != "" { - folder := resources.Folder{ - Path: "", - ID: rootFolder, + if err := r.ensureFolderExists(ctx, resources.Folder{ + ID: rootFolder, // will not change if exists Title: cfg.Spec.Title, - } - - // If the folder already exists, the parent won't be changed - if err := r.ensureFolderExists(ctx, folder, ""); err != nil { - return "", fmt.Errorf("ensure repository folder exists: %w", err) + }, ""); err != nil { + return fmt.Errorf("unable to create root folder: %w", err) } } - logger := logging.FromContext(ctx) + var err error + var currentRef string - if !isVersioned { - logger.Info("replicate tree unversioned repository") - if err := r.replicateTree(ctx, ""); err != nil { - return "", fmt.Errorf("replicate tree: %w", err) - } - return "", nil - } - - latest, err := versionedRepo.LatestRef(ctx) - if err != nil { - return "", fmt.Errorf("latest ref: %w", err) - } - - if lastCommit == "" || complete { - if err := r.replicateTree(ctx, latest); err != nil { - return latest, fmt.Errorf("replicate tree: %w", err) - } - logger.Info("initial replication for versioned repository", "latest", latest) - - return latest, nil - } - - logger.Info("replicate incremental changes for versioned repository", "last_commit", lastCommit, "latest", latest) - changes, err := versionedRepo.CompareFiles(ctx, lastCommit, latest) - if err != nil { - return latest, fmt.Errorf("compare files: %w", err) - } - - if err := r.replicateChanges(ctx, changes); err != nil { - return latest, fmt.Errorf("replicate changes: %w", err) - } - - return latest, nil -} - -func (r *Syncer) cleanUnnecessaryResources(ctx context.Context, tree []repository.FileTreeEntry, list *provisioning.ResourceList) error { - paths := make(map[string]bool) - // Add root path - paths[""] = true - for _, entry := range tree { - paths[entry.Path] = true - } - - sortResourceListForDeletion(list) - - logger := logging.FromContext(ctx) - - var count int - for _, i := range list.Items { - if _, ok := paths[i.Path]; ok { - continue - } - - if err := r.deleteListResource(ctx, i); err != nil { - return fmt.Errorf("delete list resource: %w", err) - } - - count++ - logger.Info("deleted resource not present in repository", "resource", i) - } - - if count == 0 { - logger.Info("resources are in sync with repository") - } else { - logger.Info("deleted resources not present in repository", "count", count) - } - - return nil -} - -func sortResourceListForDeletion(list *provisioning.ResourceList) { - // FIXME: this code should be simplified once unified storage folders support recursive deletion - // Sort by the following logic: - // - Put folders at the end so that we empty them first. - // - Sort folders by depth so that we remove the deepest first - sort.Slice(list.Items, func(i, j int) bool { - switch { - case list.Items[i].Group != folders.RESOURCE: - return true - case list.Items[j].Group != folders.RESOURCE: - return false - default: - return len(strings.Split(list.Items[i].Path, "/")) > len(strings.Split(list.Items[j].Path, "/")) - } - }) -} - -func (r *Syncer) deleteListResource(ctx context.Context, item provisioning.ResourceListItem) error { - // HACK: we need to find a better way to know the API version - var version string - switch item.Resource { - case folders.RESOURCE: - version = folders.VERSION - case dashboards.DASHBOARD_RESOURCE: - version = dashboardsv2alpha1.VERSION - default: - return fmt.Errorf("unknown resource api version: %s", item.Resource) - } - - client := r.client.Resource(schema.GroupVersionResource{ - Group: item.Group, - Version: version, - Resource: item.Resource, - }) - - if err := client.Delete(ctx, item.Name, metav1.DeleteOptions{}); err != nil { - return fmt.Errorf("delete resource: %w", err) - } - - return nil -} - -// replicateTree replicates all files in the repository. -func (r *Syncer) replicateTree(ctx context.Context, ref string) error { - // Load the tree of the repository - fileTree, err := r.repository.ReadTree(ctx, ref) - if err != nil { - return fmt.Errorf("read tree: %w", err) - } - - // Load the list of resources in the cluster - list, err := r.lister.List(ctx, r.client.GetNamespace(), r.repository.Config().Name) - if err != nil { - return fmt.Errorf("list resources: %w", err) - } - - // Remove resources not longer present in the repository - if err := r.cleanUnnecessaryResources(ctx, fileTree, list); err != nil { - return fmt.Errorf("clean up before replicating tree: %w", err) - } - - // Create folders - folderTree := resources.NewFolderTreeFromResourceList(list) - for _, entry := range fileTree { - parent := resources.ParentFolder(path.Dir(entry.Path), r.repository.Config()) - if err := r.ensureFolderPathExists(ctx, path.Dir(entry.Path), parent, folderTree); err != nil { - return fmt.Errorf("ensure folder path exists: %w", err) - } - } - - // Create a list of hashes to identify changed files - hashes := make(map[string]string) - for _, item := range list.Items { - hashes[item.Path] = item.Hash - } - - // Create Dashboards - for _, entry := range fileTree { - logger := logging.FromContext(ctx).With("file", entry.Path) - if hashes[entry.Path] == entry.Hash { - logger.Debug("file is up to date") - continue - } - - if !entry.Blob { - logger.Debug("ignoring non-blob entry") - continue - } - - if resources.ShouldIgnorePath(entry.Path) { - logger.Debug("ignoring file") - continue - } - - info, err := r.repository.Read(ctx, entry.Path, ref) + versionedRepo, _ := r.repository.(repository.VersionedRepository) + if versionedRepo != nil { + currentRef, err = versionedRepo.LatestRef(ctx) if err != nil { - return fmt.Errorf("read file: %w", err) + return fmt.Errorf("getting latest ref: %w", err) } - // The parse function will fill in the repository metadata, so copy it over here - info.Hash = entry.Hash - info.Modified = nil // modified? - - parsed, err := r.parseResource(ctx, info) - if err != nil { - return fmt.Errorf("parse resource: %w", err) - } - - if err := r.replicateFile(ctx, parsed, folderTree); err != nil { - if errors.Is(err, resources.ErrUnableToReadResourceBytes) { - logger.Info("file does not contain a resource") - continue + if cfg.Status.Sync.Hash != "" && r.options.Incremental { + r.syncStatus.Hash = currentRef + if currentRef == cfg.Status.Sync.Hash { + message := "same commit as last sync" + r.syncStatus.Hash = currentRef + r.syncStatus.State = provisioning.JobStateSuccess + r.syncStatus.Message = append(r.syncStatus.Message, message) + r.syncStatus.Incremental = true + r.jobStatus.Message = message + return nil } - return fmt.Errorf("replicate file: %w", err) + return r.applyVersionedChanges(ctx, versionedRepo, cfg.Status.Sync.Hash, currentRef) } } - return nil + // Read the complete change set + target, err := r.lister.List(ctx, cfg.Namespace, cfg.Name) + if err != nil { + return fmt.Errorf("error listing current: %w", err) + } + source, err := r.repository.ReadTree(ctx, currentRef) + if err != nil { + return fmt.Errorf("error reading tree: %w", err) + } + changes, err := Changes(source, target) + if err != nil { + return fmt.Errorf("error calculating changes: %w", err) + } + + r.syncStatus.Hash = currentRef + if len(changes) == 0 { + message := "no changes to sync" + r.syncStatus.State = provisioning.JobStateSuccess + r.syncStatus.Message = append(r.syncStatus.Message, message) + r.jobStatus.Message = message + return nil + } + + // Load any existing folder information + r.folderLookup = resources.NewFolderTreeFromResourceList(target) + + // Now apply the changes + return r.applyChanges(ctx, changes) } -// replicateFile creates a new resource in the cluster. -// If the resource already exists, it will be updated. -func (r *Syncer) replicateFile(ctx context.Context, file *resources.ParsedResource, folderTree *resources.FolderTree) error { - logger := logging.FromContext(ctx).With("file", file.Info.Path, "ref", file.Info.Ref) - logger = logger.With("action", file.Action, "name", file.Obj.GetName(), "file_namespace", file.Obj.GetNamespace(), "namespace", r.client.GetNamespace()) - - parent := resources.ParentFolder(file.Info.Path, r.repository.Config()) - logger = logger.With("folder", parent) - if parent != "" { - if !folderTree.In(parent) { - return fmt.Errorf("failed to find parent in tree for %s in %s", file.Info.Path, parent) - } - file.Meta.SetFolder(parent) - } - - if file.Action == provisioning.ResourceActionCreate { - _, err := file.Client.Create(ctx, file.Obj, metav1.CreateOptions{}) - if err != nil { - return fmt.Errorf("failed to create object: %w", err) - } - } else if file.Action == provisioning.ResourceActionUpdate { - existingMeta, err := utils.MetaAccessor(file.Existing) - if err != nil { - return fmt.Errorf("failed to create meta accessor for the existing object: %w", err) - } - - // Just in case no uid is present on the metadata for some reason. - logger := logger.With("previous_uid", file.Meta.GetUID(), "previous_resource_version", existingMeta.GetResourceVersion()) - if uid, ok, _ := unstructured.NestedString(file.Existing.Object, "spec", "uid"); ok { - logger.Debug("updating file's UID with spec.uid", "uid", uid) - file.Meta.SetUID(types.UID(uid)) - } - if uid := existingMeta.GetUID(); uid != "" { - logger.Debug("updating file's UID with existing meta uid", "uid", uid) - file.Meta.SetUID(uid) - } - if rev := existingMeta.GetResourceVersion(); rev != "" { - logger.Debug("updating file's UID with existing resource version", "version", rev) - file.Meta.SetResourceVersion(rev) - } - if gen := existingMeta.GetGeneration(); gen != 0 { - logger.Debug("updating file's UID with existing generation + 1", "generation", gen, "new_generation", gen+1) - file.Meta.SetGeneration(gen + 1) - } - - _, err = file.Client.Update(ctx, file.Obj, metav1.UpdateOptions{}) - if err != nil { - return fmt.Errorf("failed to update object: %w", err) - } - } else { - logger.Error("bug in Grafana: the file's action is unhandled") - return fmt.Errorf("bug in Grafana: got a file.Action of '%s', which is not defined to be handled", file.Action) - } - - logger.Info("Replicated file") - - return nil -} - -func (r *Syncer) replicateChanges(ctx context.Context, changes []repository.FileChange) error { - // Create an empty tree to avoid loading all folders unnecessarily - folderTree := resources.NewEmptyFolderTree() +func (r *syncJob) applyChanges(ctx context.Context, changes []ResourceFileChange) error { + // Do the longest paths first (important for delete) + sort.Slice(changes, func(i, j int) bool { + return len(changes[i].Path) > len(changes[j].Path) + }) // Create folder structure first for _, change := range changes { + if len(r.jobStatus.Errors) > 20 { + r.jobStatus.Errors = append(r.jobStatus.Errors, "too many errors, stopping") + r.jobStatus.State = provisioning.JobStateError + return nil + } + r.maybeNotify(ctx) + if change.Action == repository.FileActionDeleted { - // FIXME: this will leave empty folder behind until the next sync + if change.Existing == nil || change.Existing.Name == "" { + r.logger.Error("deleted file is missing existing reference", "file", change.Path) + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("unable to delete resource from: %s", change.Path)) + continue + } + + client, err := r.client(change.Existing.Resource) + if err != nil { + r.logger.Warn("unable to get client for deleted object", "file", change.Path, "err", err, "obj", change.Existing) + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("unsupported object: %s / %s", change.Path, change.Existing.Resource)) + continue + } + err = client.Delete(ctx, change.Existing.Name, metav1.DeleteOptions{}) + if err != nil { + r.logger.Warn("deleting error", "file", change.Path, "err", err) + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("error deleting %s: %s // %s", change.Existing.Resource, change.Existing.Name, change.Path)) + } else { + r.summary.Delete++ + } continue } - parent := resources.ParentFolder(path.Dir(change.Path), r.repository.Config()) - if err := r.ensureFolderPathExists(ctx, path.Dir(change.Path), parent, folderTree); err != nil { - return fmt.Errorf("ensure folder path exists: %w", err) + // Write the resource file + err := r.writeResourceFromFile(ctx, change.Path, "", change.Action) + if err != nil { + r.logger.Warn("write resource error", "file", change.Path, "err", err) + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("error writing: %s // %s", change.Path, err.Error())) } } + return nil +} - // Replicate the file changes - for _, change := range changes { - if resources.ShouldIgnorePath(change.Path) { - continue - } - - fileInfo, err := r.repository.Read(ctx, change.Path, change.Ref) +// 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) if err != nil { - return fmt.Errorf("read file: %w", err) + r.logger.Warn("unable to send progress", "err", err) } + } +} - parsed, err := r.parseResource(ctx, fileInfo) - if err != nil { - return fmt.Errorf("parse resource: %w", err) +// Convert git changes into resource file changes +func (r *syncJob) applyVersionedChanges(ctx context.Context, repo repository.VersionedRepository, previousRef, currentRef string) error { + diff, err := repo.CompareFiles(ctx, previousRef, currentRef) + if err != nil { + return fmt.Errorf("compare files error: %w", err) + } + + if len(diff) < 1 { + message := "no changes detected between commits" + r.syncStatus.State = provisioning.JobStateSuccess + r.syncStatus.Message = append(r.syncStatus.Message, message) + r.syncStatus.Incremental = true + r.jobStatus.Message = message + return nil + } + + for _, change := range diff { + if len(r.jobStatus.Errors) > 20 { + r.jobStatus.Errors = append(r.jobStatus.Errors, "too many errors to continue") + return nil } + r.maybeNotify(ctx) switch change.Action { case repository.FileActionCreated, repository.FileActionUpdated: - if err := r.replicateFile(ctx, parsed, folderTree); err != nil { - return fmt.Errorf("replicate file: %w", err) - } - case repository.FileActionRenamed: - // delete in old path - oldPath, err := r.repository.Read(ctx, change.PreviousPath, change.Ref) + err = r.writeResourceFromFile(ctx, change.Path, change.Ref, change.Action) if err != nil { - return fmt.Errorf("read previous path: %w", err) - } - oldParsed, err := r.parseResource(ctx, oldPath) - if err != nil { - return fmt.Errorf("parse old resource: %w", err) - } - if err := r.deleteFile(ctx, oldParsed); err != nil { - return fmt.Errorf("delete file: %w", err) + r.logger.Warn("error writing", "change", change, "err", err) + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("error loading: %s / %s", change.Path, err.Error())) } - if err := r.replicateFile(ctx, parsed, folderTree); err != nil { - return fmt.Errorf("replicate file in new path: %w", err) - } case repository.FileActionDeleted: - if err := r.deleteFile(ctx, parsed); err != nil { - return fmt.Errorf("delete file: %w", err) + err = r.deleteObject(ctx, change.Path, change.PreviousRef) + if err != nil { + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("error deleting file: %s / %s", change.Path, err.Error())) + continue + } + + case repository.FileActionRenamed: + // 1. Delete + err = r.deleteObject(ctx, change.Path, change.PreviousRef) + if err != nil { + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("error deleting renamed file: %s / %s", change.Path, err.Error())) + continue + } + + // 2. Create + err = r.writeResourceFromFile(ctx, change.Path, change.Ref, repository.FileActionCreated) + if err != nil { + r.logger.Warn("error writing", "change", change, "err", err) + r.jobStatus.Errors = append(r.jobStatus.Errors, + fmt.Sprintf("error loading: %s / %s", change.Path, err.Error())) } } } - return nil } -func (r *Syncer) deleteFile(ctx context.Context, file *resources.ParsedResource) error { - _, err := file.Client.Get(ctx, file.Obj.GetName(), metav1.GetOptions{}) - // FIXME: Remove the 'false &&' when .Get returns 404 on 404 instead of 500. Until then, this is a really ugly workaround. - if false && err != nil && !apierrors.IsNotFound(err) { - return fmt.Errorf("failed to check if object already exists: %w", err) - } - - if err != nil { // IsNotFound - return fmt.Errorf("get object to delete: %w", err) - } - - if err = file.Client.Delete(ctx, file.Obj.GetName(), metav1.DeleteOptions{}); err != nil { - return fmt.Errorf("failed to delete object: %w", err) - } - - return nil -} - -func (r *Syncer) parseResource(ctx context.Context, fileInfo *repository.FileInfo) (*resources.ParsedResource, error) { - file, err := r.parser.Parse(ctx, fileInfo, true) +func (r *syncJob) deleteObject(ctx context.Context, path string, ref string) error { + info, err := r.repository.Read(ctx, path, ref) if err != nil { - return nil, fmt.Errorf("failed to parse file %s: %w", fileInfo.Path, err) + return err } - if file.GVR == nil { - return nil, errors.New("parsed file is missing GVR") + obj, gvk, _ := resources.DecodeYAMLObject(bytes.NewBuffer(info.Data)) + if obj == nil { + return fmt.Errorf("no object found in: %s", path) } - if file.Client == nil { - return nil, errors.New("parsed file is missing client") + // Find the referenced file + objName, _ := resources.NamesFromHashedRepoPath(r.repository.Config().Name, path) + + client, err := r.client(gvk.Kind) + if err != nil { + return err + } + err = client.Delete(ctx, objName, metav1.DeleteOptions{}) + if err != nil { + return fmt.Errorf("deleting error: %s, %w", path, err) + } + r.summary.Delete++ + return nil +} + +func (r *syncJob) writeResourceFromFile(ctx context.Context, path string, ref string, action repository.FileAction) error { + if resources.ShouldIgnorePath(path) { + return nil // skip } - return file, nil + // Make sure the parent folders exist + folder, err := r.ensureFolderPathExists(ctx, path) + if err != nil { + return err // fail when we can not make folders + } + + // Read the referenced file + fileInfo, err := r.repository.Read(ctx, path, ref) + if err != nil { + return err + } + + parsed, err := r.parser.Parse(ctx, fileInfo, false) // no validation + if err != nil { + return err + } + + parsed.Meta.SetFolder(folder) + parsed.Meta.SetUID("") // clear identifiers + parsed.Meta.SetResourceVersion("") // clear identifiers + + switch action { + case repository.FileActionCreated: + _, err = parsed.Client.Create(ctx, parsed.Obj, metav1.CreateOptions{}) + if err != nil { + return err + } + r.summary.Create++ + + case repository.FileActionUpdated: + _, err = parsed.Client.Update(ctx, parsed.Obj, metav1.UpdateOptions{}) + if err != nil { + return err + } + r.summary.Update++ + + default: + return fmt.Errorf("unexpected action: %s", action) + } + return nil } // ensureFolderPathExists creates the folder structure in the cluster. -func (r *Syncer) ensureFolderPathExists(ctx context.Context, dirPath, parent string, folderTree *resources.FolderTree) error { - return safepath.Walk(ctx, dirPath, func(ctx context.Context, path string) error { - fid := resources.ParseFolder(path, r.repository.Config().GetName()) - if folderTree.In(fid.ID) { - // already visited - parent = fid.ID - return nil +func (r *syncJob) ensureFolderPathExists(ctx context.Context, filePath string) (parent string, err error) { + cfg := r.repository.Config() + parent = resources.RootFolder(cfg) + + dir := path.Dir(filePath) + if dir == "." { + return parent, nil + } + + if r.folderLookup == nil { + r.folderLookup = resources.NewEmptyFolderTree() + } + + f := resources.ParseFolder(dir, cfg.Name) + if r.folderLookup.In(f.ID) { + return f.ID, nil + } + + traverse := "" + + for i, part := range strings.Split(f.Path, "/") { + if i == 0 { + traverse = part + } else { + traverse, err = safepath.Join(traverse, part) + if err != nil { + return "", fmt.Errorf("unable to make path: %w", err) + } } - if err := r.ensureFolderExists(ctx, fid, parent); err != nil { - return fmt.Errorf("ensure folder exists: %w", err) + f := resources.ParseFolder(traverse, cfg.GetName()) + if r.folderLookup.In(f.ID) { + parent = f.ID + continue } - folderTree.Add(fid, parent) - parent = fid.ID - - return nil - }) + if err := r.ensureFolderExists(ctx, f, parent); err != nil { + return "", fmt.Errorf("ensure folder exists: %w", err) + } + r.folderLookup.Add(f, parent) + parent = f.ID + } + return f.ID, err } // ensureFolderExists creates the folder if it doesn't exist. // If the folder already exists: // - it will error if the folder is not owned by this repository -func (r *Syncer) ensureFolderExists(ctx context.Context, folder resources.Folder, parent string) error { +func (r *syncJob) ensureFolderExists(ctx context.Context, folder resources.Folder, parent string) error { cfg := r.repository.Config() - repoInfo := &utils.ResourceRepositoryInfo{ - Name: cfg.GetName(), - Path: folder.Path, - Hash: "", // FIXME: which hash? - Timestamp: nil, // ???&info.Modified.Time, - } obj, err := r.folders.Get(ctx, folder.ID, metav1.GetOptions{}) if err == nil { current, ok := obj.GetAnnotations()[utils.AnnoKeyRepoName] @@ -496,7 +474,12 @@ func (r *Syncer) ensureFolderExists(ctx context.Context, folder resources.Folder if parent != "" { meta.SetFolder(parent) } - meta.SetRepositoryInfo(repoInfo) + meta.SetRepositoryInfo(&utils.ResourceRepositoryInfo{ + Name: cfg.GetName(), + Path: folder.Path, + Hash: "", // FIXME: which hash? + Timestamp: nil, // ???&info.Modified.Time, + }) if _, err := r.folders.Create(ctx, obj, metav1.CreateOptions{}); err != nil { return fmt.Errorf("failed to create folder: %w", err) @@ -504,3 +487,13 @@ func (r *Syncer) ensureFolderExists(ctx context.Context, folder resources.Folder return nil } + +func (r *syncJob) client(kind string) (dynamic.ResourceInterface, error) { + switch kind { + case dashboard.GROUP, dashboard.DASHBOARD_RESOURCE, "Dashboard": + return r.dashboards, nil + case folders.GROUP, folders.RESOURCE, "Folder": + return r.folders, nil + } + return nil, fmt.Errorf("unsupported resource: %s", kind) +} diff --git a/pkg/registry/apis/provisioning/jobs/worker.go b/pkg/registry/apis/provisioning/jobs/worker.go index 5c0ac388bd4..929e100dcfd 100644 --- a/pkg/registry/apis/provisioning/jobs/worker.go +++ b/pkg/registry/apis/provisioning/jobs/worker.go @@ -38,6 +38,7 @@ type JobWorker struct { lister resources.ResourceLister blobstore blob.PublicBlobStore urlProvider func(namespace string) string + syncer Syncer } func NewJobWorker( @@ -59,6 +60,10 @@ func NewJobWorker( lister: lister, blobstore: blobstore, urlProvider: urlProvider, + syncer: &syncer{ + parsers: parsers, + lister: lister, + }, } } @@ -91,7 +96,7 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress switch job.Spec.Action { case provisioning.JobActionSync: - return g.doSync(ctx, repo, job, parser, progress) + return g.doSync(ctx, repo, job, progress) case provisioning.JobActionPullRequest: prRepo, ok := repo.(PullRequestRepo) @@ -156,20 +161,20 @@ func (g *JobWorker) Process(ctx context.Context, job provisioning.Job, progress } } +// The Syncer will synchronize the external repo with grafana database +// this function updates the status for both the job and the referenced repository func (g *JobWorker) doSync(ctx context.Context, repo repository.Repository, job provisioning.Job, - parser *resources.Parser, - _ func(provisioning.JobStatus) error, + progress func(provisioning.JobStatus) error, ) (*provisioning.JobStatus, error) { var err error cfg := repo.Config() logger := logging.FromContext(ctx).With("job", job.GetName(), "namespace", job.GetNamespace()) - status := job.Status.ToSyncStatus(job.Name) patch, err := json.Marshal(map[string]any{ "status": map[string]any{ - "sync": status, + "sync": job.Status.ToSyncStatus(job.Name), }, }) if err != nil { @@ -182,32 +187,26 @@ func (g *JobWorker) doSync(ctx context.Context, logger.Warn("unable to update repo with job status", "err", err) } - syncer, err := NewSyncer(repo, g.lister, parser) - if err != nil { - return nil, fmt.Errorf("error creating replicator") + // Execute the sync task + jobStatus, syncStatus, syncError := g.syncer.Sync(ctx, repo, *job.Spec.Sync, progress) + if syncStatus == nil { + syncStatus = &provisioning.SyncStatus{} } - - var complete bool - if job.Spec.Sync != nil && job.Spec.Sync.Complete { - complete = job.Spec.Sync.Complete - } - - // Sync the repository - ref, syncError := syncer.Sync(ctx, complete) - status = job.Status.ToSyncStatus(job.Name) - status.Hash = ref - status.Finished = time.Now().UnixMilli() + syncStatus.JobID = job.Name + syncStatus.Started = job.Status.Started + syncStatus.Finished = time.Now().UnixMilli() if syncError != nil { - status.State = provisioning.JobStateError - status.Message = []string{ + syncStatus.State = provisioning.JobStateError + syncStatus.Message = []string{ "error running sync", syncError.Error(), } - } else { - status.State = provisioning.JobStateSuccess + } else if syncStatus.State == "" { + syncStatus.State = provisioning.JobStateSuccess } - // Update the resource stats + // Update the resource stats -- give the index some time to catch up + time.Sleep(1 * time.Second) stats, err := g.lister.Stats(ctx, cfg.Namespace, cfg.Name) if err != nil { logger.Warn("unable to read stats", "error", err) @@ -218,7 +217,7 @@ func (g *JobWorker) doSync(ctx context.Context, patch, err = json.Marshal(map[string]any{ "status": map[string]any{ - "sync": status, + "sync": syncStatus, "stats": stats.Items, }, }) @@ -235,8 +234,5 @@ func (g *JobWorker) doSync(ctx context.Context, if syncError != nil { return nil, syncError } - - return &provisioning.JobStatus{ - State: provisioning.JobStateSuccess, - }, nil + return jobStatus, nil } diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index f1907f87ff2..9f56f616fde 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -662,7 +662,7 @@ spec: MediaTypeProps: spec3.MediaTypeProps{ Schema: &optionsSchema, Example: &provisioning.SyncJobOptions{ - Complete: true, + Incremental: false, }, }, }, diff --git a/pkg/registry/apis/provisioning/repository/changes.go b/pkg/registry/apis/provisioning/repository/changes.go deleted file mode 100644 index 75ff2aa29b9..00000000000 --- a/pkg/registry/apis/provisioning/repository/changes.go +++ /dev/null @@ -1,57 +0,0 @@ -package repository - -import ( - "fmt" - - provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" -) - -func Changes(source []FileTreeEntry, target *provisioning.ResourceList) ([]FileChange, error) { - lookup := make(map[string]*provisioning.ResourceListItem, len(target.Items)) - for _, item := range target.Items { - if item.Path == "" { - if item.Group != "folder.grafana.app" { - return nil, fmt.Errorf("empty path on a non folder") - } - continue - } - lookup[item.Path] = &item - } - - changes := make([]FileChange, 0, len(source)) - for _, file := range source { - if !file.Blob { - continue // skip folder references? - } - - check, ok := lookup[file.Path] - if ok { - if check.Hash != file.Hash { - changes = append(changes, FileChange{ - Action: FileActionUpdated, - Path: file.Path, - Ref: file.Hash, - PreviousRef: check.Hash, - }) - } - delete(lookup, file.Path) - } else { - changes = append(changes, FileChange{ - Action: FileActionCreated, // or previously ignored/failed - Path: file.Path, - Ref: file.Hash, - }) - } - } - - // File that were previously added, but are not in the current list - for _, v := range lookup { - changes = append(changes, FileChange{ - Path: v.Path, - PreviousRef: v.Hash, - Action: FileActionDeleted, - }) - } - - return changes, nil -} diff --git a/pkg/registry/apis/provisioning/repository/changes_test.go b/pkg/registry/apis/provisioning/repository/changes_test.go deleted file mode 100644 index 642c86ef4d0..00000000000 --- a/pkg/registry/apis/provisioning/repository/changes_test.go +++ /dev/null @@ -1,107 +0,0 @@ -package repository - -import ( - "encoding/json" - "testing" - - "github.com/stretchr/testify/require" - - provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" -) - -func TestChanges(t *testing.T) { - t.Run("start the same", func(t *testing.T) { - source, target := getBase(t) - changes, err := Changes(source, target) - require.NoError(t, err) - require.Empty(t, changes) - }) - - t.Run("create a source file", func(t *testing.T) { - source, target := getBase(t) - source = append(source, FileTreeEntry{Path: "muta.json", Hash: "xyz", Blob: true}) - - changes, err := Changes(source, target) - require.NoError(t, err) - require.Len(t, changes, 1) - require.Equal(t, FileChange{ - Action: FileActionCreated, - Path: "muta.json", - Ref: "xyz", - }, changes[0]) - }) - - t.Run("delete a source file", func(t *testing.T) { - source, target := getBase(t) - source = []FileTreeEntry{source[0]} - - changes, err := Changes(source, target) - require.NoError(t, err) - require.Len(t, changes, 1) - require.Equal(t, FileChange{ - Action: FileActionDeleted, - Path: "adsl62h.yaml", - PreviousRef: "ce5d497c4deadde6831162ce8509e2b2b1776237", - }, changes[0]) - }) - - t.Run("modify a file", func(t *testing.T) { - source, target := getBase(t) - source[1].Hash = "different" - - changes, err := Changes(source, target) - require.NoError(t, err) - require.Len(t, changes, 1) - require.Equal(t, FileChange{ - Action: FileActionUpdated, - Path: "adsl62h.yaml", - Ref: "different", - PreviousRef: "ce5d497c4deadde6831162ce8509e2b2b1776237", - }, changes[0]) - }) -} - -func getBase(t *testing.T) (source []FileTreeEntry, target *provisioning.ResourceList) { - target = &provisioning.ResourceList{} - err := json.Unmarshal([]byte(`{ - "kind": "ResourceList", - "apiVersion": "provisioning.grafana.app/v0alpha1", - "metadata": {}, - "items": [ - { - "path": "", - "group": "folder.grafana.app", - "resource": "folders", - "name": "simplelocal-3794ab9", - "hash": "", - "title": "Simple local" - }, - { - "path": "ad4lwp2.yaml", - "group": "dashboard.grafana.app", - "resource": "dashboards", - "name": "ad4lwp2-xofjsuo-mr5blr1zwimlfi0ds0pyrrpd", - "hash": "ca83d64b9c4a23fed975aacdf47e7de8878b4ae0", - "title": "Test two", - "folder": "simplelocal-3794ab9" - }, - { - "path": "adsl62h.yaml", - "group": "dashboard.grafana.app", - "resource": "dashboards", - "name": "adsl62h-hrw-f-fvlt2dghp-gufrc4lisksgmq-c", - "hash": "ce5d497c4deadde6831162ce8509e2b2b1776237", - "title": "Test one", - "folder": "simplelocal-3794ab9" - } - ] - }`), target) - require.NoError(t, err) - - source = []FileTreeEntry{ - {Path: "ad4lwp2.yaml", Hash: "ca83d64b9c4a23fed975aacdf47e7de8878b4ae0", Blob: true}, - {Path: "adsl62h.yaml", Hash: "ce5d497c4deadde6831162ce8509e2b2b1776237", Blob: true}, - } - - return // named values! -} diff --git a/pkg/registry/apis/provisioning/repository/github.go b/pkg/registry/apis/provisioning/repository/github.go index 6f777114891..29d4976c28b 100644 --- a/pkg/registry/apis/provisioning/repository/github.go +++ b/pkg/registry/apis/provisioning/repository/github.go @@ -521,7 +521,7 @@ func (r *githubRepository) parsePushEvent(event *github.PushEvent) (*provisionin Repository: r.Config().GetName(), Action: provisioning.JobActionSync, Sync: &provisioning.SyncJobOptions{ - Complete: false, + Incremental: true, }, }, }, nil @@ -593,7 +593,7 @@ func (r *githubRepository) LatestRef(ctx context.Context) (string, error) { return branch.Sha, nil } -func (r *githubRepository) CompareFiles(ctx context.Context, base, ref string) ([]FileChange, error) { +func (r *githubRepository) CompareFiles(ctx context.Context, base, ref string) ([]VersionedFileChange, error) { if ref == "" { var err error ref, err = r.LatestRef(ctx) @@ -610,24 +610,24 @@ func (r *githubRepository) CompareFiles(ctx context.Context, base, ref string) ( return nil, fmt.Errorf("compare commits: %w", err) } - changes := make([]FileChange, 0) + changes := make([]VersionedFileChange, 0) for _, f := range files { // reference: https://docs.github.com/en/rest/commits/commits?apiVersion=2022-11-28#get-a-commit switch f.GetStatus() { case "added", "copied": - changes = append(changes, FileChange{ + changes = append(changes, VersionedFileChange{ Path: f.GetFilename(), Ref: ref, Action: FileActionCreated, }) case "modified", "changed": - changes = append(changes, FileChange{ + changes = append(changes, VersionedFileChange{ Path: f.GetFilename(), Ref: ref, Action: FileActionUpdated, }) case "renamed": - changes = append(changes, FileChange{ + changes = append(changes, VersionedFileChange{ Path: f.GetFilename(), PreviousPath: f.GetPreviousFilename(), Ref: ref, @@ -635,7 +635,7 @@ func (r *githubRepository) CompareFiles(ctx context.Context, base, ref string) ( Action: FileActionRenamed, }) case "removed": - changes = append(changes, FileChange{ + changes = append(changes, VersionedFileChange{ Ref: base, Path: f.GetFilename(), Action: FileActionDeleted, diff --git a/pkg/registry/apis/provisioning/repository/github_test.go b/pkg/registry/apis/provisioning/repository/github_test.go index d1e4dcfa9a1..10cf8baa5a4 100644 --- a/pkg/registry/apis/provisioning/repository/github_test.go +++ b/pkg/registry/apis/provisioning/repository/github_test.go @@ -84,6 +84,9 @@ func TestParseWebhooks(t *testing.T) { Job: &provisioning.JobSpec{ // we want to always push a sync job Repository: "unit-test-repo", Action: provisioning.JobActionSync, + Sync: &provisioning.SyncJobOptions{ + Incremental: true, + }, }, }}, {"push", "nested", provisioning.WebhookResponse{ @@ -91,6 +94,9 @@ func TestParseWebhooks(t *testing.T) { Job: &provisioning.JobSpec{ Repository: "unit-test-repo", Action: provisioning.JobActionSync, + Sync: &provisioning.SyncJobOptions{ + Incremental: true, + }, }, }}, {"issue_comment", "created", provisioning.WebhookResponse{ @@ -104,6 +110,9 @@ func TestParseWebhooks(t *testing.T) { Name: "unit-test-repo", }, Spec: provisioning.RepositorySpec{ + Sync: provisioning.SyncOptions{ + Enabled: true, // required to accept sync job + }, GitHub: &provisioning.GitHubRepositoryConfig{ Repository: "git-ui-sync-demo", Owner: "grafana", diff --git a/pkg/registry/apis/provisioning/repository/local.go b/pkg/registry/apis/provisioning/repository/local.go index b3c5ffaa3d8..be144525863 100644 --- a/pkg/registry/apis/provisioning/repository/local.go +++ b/pkg/registry/apis/provisioning/repository/local.go @@ -94,6 +94,9 @@ func NewLocal(config *provisioning.Repository, resolver *LocalFolderResolver) *l } if config.Spec.Local != nil { r.path, _ = resolver.LocalPath(config.Spec.Local.Path) + if r.path != "" && !strings.HasSuffix(r.path, "/") { + r.path += "/" + } } return r } @@ -201,7 +204,7 @@ func (r *localRepository) Read(ctx context.Context, filePath string, ref string) return nil, fmt.Errorf("stat file: %w", err) } - path := strings.TrimPrefix(filePath, safepath.Clean(r.path)) + path := strings.TrimPrefix(filePath, r.path) if info.IsDir() { return &FileInfo{ Path: path, diff --git a/pkg/registry/apis/provisioning/repository/repository.go b/pkg/registry/apis/provisioning/repository/repository.go index 8f70ba9844a..9a5b45798ea 100644 --- a/pkg/registry/apis/provisioning/repository/repository.go +++ b/pkg/registry/apis/provisioning/repository/repository.go @@ -47,23 +47,6 @@ type FileTreeEntry struct { Blob bool } -type FileAction string - -const ( - FileActionCreated FileAction = "created" - FileActionUpdated FileAction = "updated" - FileActionRenamed FileAction = "renamed" - FileActionDeleted FileAction = "deleted" -) - -type FileChange struct { - Path string - PreviousPath string - Ref string - PreviousRef string - Action FileAction -} - type Repository interface { // The saved Kubernetes object. Config() *provisioning.Repository @@ -116,12 +99,32 @@ type RepositoryHooks interface { OnDelete(ctx context.Context) error } +type FileAction string + +const ( + FileActionCreated FileAction = "created" + FileActionUpdated FileAction = "updated" + FileActionDeleted FileAction = "deleted" + + // Renamed actions may be reconstructed as delete then create + FileActionRenamed FileAction = "renamed" +) + +type VersionedFileChange struct { + Action FileAction + Path string + + Ref string + PreviousRef string // rename | update + PreviousPath string // rename +} + // VersionedRepository is a repository that supports versioning // this inferface may be extended to make the the original Repository interface // more agnostic to the underlying storage system type VersionedRepository interface { LatestRef(ctx context.Context) (string, error) - CompareFiles(ctx context.Context, base, ref string) ([]FileChange, error) + CompareFiles(ctx context.Context, base, ref string) ([]VersionedFileChange, error) } func writeWithReadThenCreateOrUpdate(ctx context.Context, r Repository, path, ref string, data []byte, comment string) error { diff --git a/pkg/registry/apis/provisioning/repository_controller.go b/pkg/registry/apis/provisioning/repository_controller.go index 7a0d8fb0499..ef61792af00 100644 --- a/pkg/registry/apis/provisioning/repository_controller.go +++ b/pkg/registry/apis/provisioning/repository_controller.go @@ -344,16 +344,16 @@ func (rc *RepositoryController) process(item *queueItem) error { if hooks != nil { webhookStatus, err = hooks.OnCreate(ctx) } - sync = &provisioning.SyncJobOptions{Complete: true} + sync = &provisioning.SyncJobOptions{} case hasSpecChanged: logger.Info("handle repository update") if hooks != nil { webhookStatus, err = hooks.OnUpdate(ctx) } - sync = &provisioning.SyncJobOptions{Complete: hasSpecChanged} + sync = &provisioning.SyncJobOptions{} case shouldResync: logger.Info("handle repository resync") - sync = &provisioning.SyncJobOptions{Complete: false} + sync = &provisioning.SyncJobOptions{Incremental: true} default: logger.Info("handle unknown repository situation") } diff --git a/pkg/registry/apis/provisioning/resources/parser.go b/pkg/registry/apis/provisioning/resources/parser.go index ffacad921f2..52147acddec 100644 --- a/pkg/registry/apis/provisioning/resources/parser.go +++ b/pkg/registry/apis/provisioning/resources/parser.go @@ -162,6 +162,8 @@ func (r *Parser) Parse(ctx context.Context, info *repository.FileInfo, validate objName, folderName := NamesFromHashedRepoPath(cfg.Name, info.Path) obj.SetName(objName) + obj.SetUID("") // clear identifiers + obj.SetResourceVersion("") // clear identifiers parsed.Meta.SetFolder(folderName) // We can not do anything more if no kind is defined diff --git a/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json b/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json index 24fa3130b78..225a22ab0ab 100644 --- a/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json +++ b/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json @@ -2248,15 +2248,19 @@ "application/json": { "schema": { "type": "object", + "required": [ + "incremental" + ], "properties": { - "complete": { - "description": "Complete forces the sync to overwrite", - "type": "boolean" + "incremental": { + "description": "Incremental synchronization for versioned repositories", + "type": "boolean", + "default": false } } }, "example": { - "complete": true + "incremental": false } } } @@ -3789,10 +3793,14 @@ }, "com.github.grafana.grafana.pkg.apis.provisioning.v0alpha1.SyncJobOptions": { "type": "object", + "required": [ + "incremental" + ], "properties": { - "complete": { - "description": "Complete forces the sync to overwrite", - "type": "boolean" + "incremental": { + "description": "Incremental synchronization for versioned repositories", + "type": "boolean", + "default": false } } }, @@ -3840,6 +3848,10 @@ "description": "The repository hash when the last sync ran", "type": "string" }, + "incremental": { + "description": "Incremental synchronization for versioned repositories", + "type": "boolean" + }, "job": { "description": "The ID for the job that ran this sync", "type": "string" diff --git a/public/app/features/provisioning/SyncRepository.tsx b/public/app/features/provisioning/SyncRepository.tsx index c0e600292e9..5e43bc79809 100644 --- a/public/app/features/provisioning/SyncRepository.tsx +++ b/public/app/features/provisioning/SyncRepository.tsx @@ -40,7 +40,7 @@ export function SyncRepository({ repository }: Props) { if (!name) { return; } - syncResource({ name, body: { complete: true } }); + syncResource({ name, body: {} }); // will queue a full resync job setIsModalOpen(false); }; diff --git a/public/app/features/provisioning/api/endpoints.gen.ts b/public/app/features/provisioning/api/endpoints.gen.ts index 60ff07134ce..071e9291961 100644 --- a/public/app/features/provisioning/api/endpoints.gen.ts +++ b/public/app/features/provisioning/api/endpoints.gen.ts @@ -523,7 +523,7 @@ export type CreateRepositorySyncArg = { name: string; body: { /** Complete forces the sync to overwrite */ - complete?: boolean; + incremental?: boolean; }; }; export type CreateRepositoryTestResponse = /** status 200 OK */ TestResults;