Provisioning: Refactor export to simplify loading from legacy storage (#100335)
This commit is contained in:
1 parent
ec987901d6
commit
ea38dfcbdc
21 files changed
+371
-256
No files matched your search
@@ -119,7 +119,7 @@ type JobStatus struct {
|
||||
Progress float64 `json:"progress,omitempty"`
|
||||
|
||||
// Summary of processed actions
|
||||
Summary []JobResourceSummary `json:"summary,omitempty"`
|
||||
Summary []*JobResourceSummary `json:"summary,omitempty"`
|
||||
}
|
||||
|
||||
// Convert a JOB to a
|
||||
|
||||
@@ -304,9 +304,13 @@ func (in *JobStatus) DeepCopyInto(out *JobStatus) {
|
||||
}
|
||||
if in.Summary != nil {
|
||||
in, out := &in.Summary, &out.Summary
|
||||
*out = make([]JobResourceSummary, len(*in))
|
||||
*out = make([]*JobResourceSummary, len(*in))
|
||||
for i := range *in {
|
||||
(*in)[i].DeepCopyInto(&(*out)[i])
|
||||
if (*in)[i] != nil {
|
||||
in, out := &(*in)[i], &(*out)[i]
|
||||
*out = new(JobResourceSummary)
|
||||
(*in).DeepCopyInto(*out)
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
|
||||
@@ -723,8 +723,7 @@ func schema_pkg_apis_provisioning_v0alpha1_JobStatus(ref common.ReferenceCallbac
|
||||
Items: &spec.SchemaOrArray{
|
||||
Schema: &spec.Schema{
|
||||
SchemaProps: spec.SchemaProps{
|
||||
Default: map[string]interface{}{},
|
||||
Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobResourceSummary"),
|
||||
Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobResourceSummary"),
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
@@ -63,7 +63,7 @@ func (c *exportConnector) Connect(
|
||||
options := &provisioning.ExportJobOptions{}
|
||||
err := json.NewDecoder(r.Body).Decode(options)
|
||||
if err != nil {
|
||||
responder.Error(apierrors.NewBadRequest("error decoding request"))
|
||||
responder.Error(apierrors.NewBadRequest("error decoding ExportJobOptions from request"))
|
||||
return
|
||||
}
|
||||
job, err := c.jobs.Add(ctx, &provisioning.Job{
|
||||
|
||||
@@ -25,7 +25,7 @@ func dummyExport(
|
||||
|
||||
size := 5 + rand.IntN(15)
|
||||
|
||||
dashboards := provisioning.JobResourceSummary{
|
||||
dashboards := &provisioning.JobResourceSummary{
|
||||
Group: "dashboard.grafana.app",
|
||||
Resource: "dashboards",
|
||||
}
|
||||
@@ -58,7 +58,7 @@ func dummyExport(
|
||||
dashboards.Noop++
|
||||
}
|
||||
}
|
||||
status.Summary = []provisioning.JobResourceSummary{
|
||||
status.Summary = []*provisioning.JobResourceSummary{
|
||||
dashboards,
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
package export
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"golang.org/x/net/context"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/client-go/dynamic"
|
||||
|
||||
apiutils "github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
"github.com/grafana/grafana/pkg/cmd/grafana-cli/logger"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
|
||||
)
|
||||
|
||||
func readFolders(ctx context.Context, client dynamic.ResourceInterface, skip string) (*resources.FolderTree, error) {
|
||||
// TODO: handle pagination
|
||||
rawList, err := client.List(ctx, metav1.ListOptions{Limit: 10000})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to list folders: %w", err)
|
||||
}
|
||||
if rawList.GetContinue() != "" {
|
||||
return nil, fmt.Errorf("unable to list all folders in one request: %s", rawList.GetContinue())
|
||||
}
|
||||
|
||||
// filter out the folders we already own
|
||||
rawFolders := make([]unstructured.Unstructured, 0, len(rawList.Items))
|
||||
for _, f := range rawList.Items {
|
||||
repoName := f.GetAnnotations()[apiutils.AnnoKeyRepoName]
|
||||
if repoName == skip {
|
||||
logger.Info("skip as folder is already in repository", "folder", f.GetName())
|
||||
continue
|
||||
}
|
||||
|
||||
rawFolders = append(rawFolders, f)
|
||||
}
|
||||
|
||||
return resources.NewFolderTreeFromUnstructure(ctx, rawFolders), nil
|
||||
}
|
||||
@@ -0,0 +1,252 @@
|
||||
package export
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"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"
|
||||
|
||||
"github.com/grafana/grafana-app-sdk/logging"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1"
|
||||
provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
|
||||
"github.com/grafana/grafana/pkg/cmd/grafana-cli/logger"
|
||||
"github.com/grafana/grafana/pkg/infra/slugify"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath"
|
||||
)
|
||||
|
||||
// ExportJob holds all context for a running job
|
||||
type exportJob struct {
|
||||
logger logging.Logger
|
||||
client *resources.DynamicClient // Read from
|
||||
target repository.Repository // Write to
|
||||
|
||||
progress jobs.ProgressFn
|
||||
progressInterval time.Duration
|
||||
progressLast time.Time
|
||||
foldersTree *resources.FolderTree
|
||||
|
||||
prefix string // from options (now clean+safe)
|
||||
ref string // from options (only git)
|
||||
keepIdentifier bool
|
||||
|
||||
jobStatus *provisioning.JobStatus
|
||||
summary map[string]*provisioning.JobResourceSummary
|
||||
}
|
||||
|
||||
func newExportJob(ctx context.Context,
|
||||
target repository.Repository,
|
||||
options provisioning.ExportJobOptions,
|
||||
client *resources.DynamicClient,
|
||||
progress jobs.ProgressFn,
|
||||
) *exportJob {
|
||||
prefix := options.Prefix
|
||||
if prefix != "" {
|
||||
prefix = safepath.Clean(prefix)
|
||||
}
|
||||
return &exportJob{
|
||||
target: target,
|
||||
client: client,
|
||||
logger: logging.FromContext(ctx),
|
||||
progress: progress,
|
||||
progressLast: time.Now(),
|
||||
progressInterval: time.Second * 10,
|
||||
|
||||
prefix: prefix,
|
||||
ref: options.Branch,
|
||||
keepIdentifier: options.Identifier,
|
||||
|
||||
jobStatus: &provisioning.JobStatus{
|
||||
State: provisioning.JobStateWorking,
|
||||
},
|
||||
summary: make(map[string]*provisioning.JobResourceSummary),
|
||||
}
|
||||
}
|
||||
|
||||
// Send progress messages to any listeners
|
||||
func (r *exportJob) maybeNotify(ctx context.Context) {
|
||||
if time.Since(r.progressLast) > r.progressInterval {
|
||||
err := r.progress(ctx, *r.jobStatus)
|
||||
if err != nil {
|
||||
r.logger.Warn("unable to send progress", "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Register summary information for a group/resource
|
||||
func (r *exportJob) getSummary(gr schema.GroupResource) *provisioning.JobResourceSummary {
|
||||
summary, ok := r.summary[gr.String()]
|
||||
if !ok {
|
||||
summary = &provisioning.JobResourceSummary{
|
||||
Group: gr.Group,
|
||||
Resource: gr.Resource,
|
||||
}
|
||||
r.summary[gr.String()] = summary
|
||||
r.jobStatus.Summary = append(r.jobStatus.Summary, summary)
|
||||
}
|
||||
return summary
|
||||
}
|
||||
|
||||
func (r *exportJob) loadFolders(ctx context.Context) error {
|
||||
logger := r.logger
|
||||
targetRepoName := r.target.Config().Name
|
||||
status := r.jobStatus
|
||||
status.Message = "reading folder tree"
|
||||
foldersTree, err := readFolders(ctx, r.client.Resource(schema.GroupVersionResource{
|
||||
Group: folders.GROUP,
|
||||
Version: folders.VERSION,
|
||||
Resource: folders.RESOURCE,
|
||||
}), targetRepoName)
|
||||
|
||||
summary := r.getSummary(schema.GroupResource{
|
||||
Group: folders.GROUP,
|
||||
Resource: folders.RESOURCE,
|
||||
})
|
||||
|
||||
// first create folders
|
||||
// TODO! this should not be necessary if writing to a path also makes the parents
|
||||
status.Message = "writing folders"
|
||||
err = foldersTree.Walk(ctx, func(ctx context.Context, folder resources.Folder) error {
|
||||
p := folder.Path + "/"
|
||||
if r.prefix != "" {
|
||||
p = r.prefix + "/" + p
|
||||
}
|
||||
logger := logger.With("path", p)
|
||||
|
||||
_, err = r.target.Read(ctx, p, r.ref)
|
||||
if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {
|
||||
logger.Error("failed to check if folder exists before writing", "error", err)
|
||||
return fmt.Errorf("failed to check if folder exists before writing: %w", err)
|
||||
} else if err == nil {
|
||||
logger.Info("folder already exists")
|
||||
summary.Noop++
|
||||
return nil
|
||||
}
|
||||
|
||||
// Create with an empty body will make a folder (or .keep file if unsupported)
|
||||
if err := r.target.Create(ctx, p, r.ref, nil, "export folder `"+p+"`"); err != nil {
|
||||
logger.Error("failed to write a folder in repository", "error", err)
|
||||
return fmt.Errorf("failed to write folder in repo: %w", err)
|
||||
}
|
||||
summary.Create++
|
||||
logger.Debug("successfully exported folder")
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write folders: %w", err)
|
||||
}
|
||||
r.foldersTree = foldersTree
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *exportJob) export(ctx context.Context, kind schema.GroupVersionResource) error {
|
||||
r.jobStatus.Message = "Exporting " + kind.Resource + "..."
|
||||
r.maybeNotify(ctx)
|
||||
client := r.client.Resource(kind)
|
||||
summary := r.getSummary(kind.GroupResource())
|
||||
|
||||
continueToken := ""
|
||||
for {
|
||||
list, err := client.List(ctx, metav1.ListOptions{Limit: 100, Continue: continueToken})
|
||||
if err != nil {
|
||||
return fmt.Errorf("error executing list: %w", err)
|
||||
}
|
||||
|
||||
for _, item := range list.Items {
|
||||
if err = r.add(ctx, summary, &item); err != nil {
|
||||
return fmt.Errorf("error adding value: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
continueToken = list.GetContinue()
|
||||
if continueToken == "" {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *exportJob) add(ctx context.Context, summary *provisioning.JobResourceSummary, obj *unstructured.Unstructured) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
r.maybeNotify(ctx)
|
||||
|
||||
item, err := utils.MetaAccessor(obj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
commitMessage := item.GetMessage()
|
||||
if commitMessage == "" {
|
||||
commitMessage = "exported from grafana"
|
||||
}
|
||||
|
||||
name := item.GetName()
|
||||
repoName := item.GetRepositoryName()
|
||||
if repoName == r.target.Config().GetName() {
|
||||
r.logger.Info("skip dashboard since it is already in repository", "dashboard", name)
|
||||
return nil
|
||||
}
|
||||
|
||||
title := item.FindTitle("")
|
||||
if title == "" {
|
||||
title = name
|
||||
}
|
||||
folder := item.GetFolder()
|
||||
|
||||
// Get the absolute path of the folder
|
||||
fid, ok := r.foldersTree.DirPath(folder, "")
|
||||
if !ok {
|
||||
logger.Error("folder of item was not in tree of repository")
|
||||
return fmt.Errorf("folder of item was not in tree of repository")
|
||||
}
|
||||
|
||||
delete(obj.Object, "metadata")
|
||||
if r.keepIdentifier {
|
||||
item.SetName(name) // keep the identifier in the metadata
|
||||
}
|
||||
|
||||
body, err := json.MarshalIndent(obj.Object, "", " ")
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal dashboard %s: %w", name, err)
|
||||
}
|
||||
|
||||
fileName := slugify.Slugify(title) + ".json"
|
||||
if fid.Path != "" {
|
||||
fileName, err = safepath.Join(fid.Path, fileName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error adding file path %s: %w", title, err)
|
||||
}
|
||||
}
|
||||
if r.prefix != "" {
|
||||
fileName, err = safepath.Join(r.prefix, fileName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error adding path prefix %s: %w", r.prefix, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Write the file
|
||||
err = r.target.Write(ctx, fileName, r.ref, body, commitMessage)
|
||||
if err != nil {
|
||||
summary.Error++
|
||||
logger.Error("failed to write a file in repository", "error", err)
|
||||
if len(summary.Errors) < 20 {
|
||||
summary.Errors = append(summary.Errors, fmt.Sprintf("error writing: %s", fileName))
|
||||
}
|
||||
} else {
|
||||
summary.Write++
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -2,39 +2,30 @@ package export
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
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"
|
||||
|
||||
"github.com/grafana/grafana-app-sdk/logging"
|
||||
apiutils "github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
folders "github.com/grafana/grafana/pkg/apis/folder/v0alpha1"
|
||||
dashboards "github.com/grafana/grafana/pkg/apis/dashboard"
|
||||
provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
|
||||
"github.com/grafana/grafana/pkg/infra/slugify"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/repository"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath"
|
||||
)
|
||||
|
||||
type ExportWorker struct {
|
||||
parsers *resources.ParserFactory
|
||||
clients *resources.ClientFactory
|
||||
}
|
||||
|
||||
func NewExportWorker(parsers *resources.ParserFactory) *ExportWorker {
|
||||
return &ExportWorker{parsers: parsers}
|
||||
func NewExportWorker(clients *resources.ClientFactory) *ExportWorker {
|
||||
return &ExportWorker{clients}
|
||||
}
|
||||
|
||||
func (r *ExportWorker) IsSupported(ctx context.Context, job provisioning.Job) bool {
|
||||
return job.Spec.Action == provisioning.JobActionExport
|
||||
}
|
||||
|
||||
//nolint:gocyclo
|
||||
// Process will start a job
|
||||
func (r *ExportWorker) Process(ctx context.Context, repo repository.Repository, job provisioning.Job, progress jobs.ProgressFn) (*provisioning.JobStatus, error) {
|
||||
if repo.Config().Spec.ReadOnly {
|
||||
return &provisioning.JobStatus{
|
||||
@@ -56,187 +47,36 @@ func (r *ExportWorker) Process(ctx context.Context, repo repository.Repository,
|
||||
return dummyExport(ctx, repo, job, progress)
|
||||
}
|
||||
|
||||
parser, err := r.parsers.GetParser(ctx, repo)
|
||||
dynamicClient, _, err := r.clients.New(repo.Config().Namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get parser for %s: %w", repo.Config().GetName(), err)
|
||||
}
|
||||
dynamicClient := parser.Client()
|
||||
if repo.Config().Namespace != dynamicClient.GetNamespace() {
|
||||
return nil, fmt.Errorf("namespace mismatch")
|
||||
}
|
||||
|
||||
foldersClient := dynamicClient.Resource(schema.GroupVersionResource{
|
||||
Group: folders.GROUP,
|
||||
Version: folders.VERSION,
|
||||
Resource: folders.RESOURCE,
|
||||
})
|
||||
worker := newExportJob(ctx, repo, *options, dynamicClient, progress)
|
||||
|
||||
dashboardsClient := dynamicClient.Resource(schema.GroupVersionResource{
|
||||
Group: "dashboard.grafana.app",
|
||||
// Load and write all folders
|
||||
err = worker.loadFolders(ctx)
|
||||
if err != nil {
|
||||
return worker.jobStatus, err
|
||||
}
|
||||
|
||||
kinds := []schema.GroupVersionResource{{
|
||||
Group: dashboards.GROUP,
|
||||
Version: "v1alpha1",
|
||||
Resource: "dashboards",
|
||||
})
|
||||
|
||||
logger := logging.FromContext(ctx)
|
||||
status := provisioning.JobStatus{
|
||||
State: provisioning.JobStateWorking,
|
||||
Message: "reading folder tree...",
|
||||
}
|
||||
if err := progress(ctx, status); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ref := options.Branch // only valid for git (defaults to the configured repo)
|
||||
if options.Prefix != "" {
|
||||
options.Prefix = safepath.Clean(options.Prefix)
|
||||
}
|
||||
|
||||
// TODO: handle pagination
|
||||
rawList, err := foldersClient.List(ctx, metav1.ListOptions{Limit: 10000})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to list folders: %w", err)
|
||||
}
|
||||
if rawList.GetContinue() != "" {
|
||||
return nil, fmt.Errorf("unable to list all folders in one request: %s", rawList.GetContinue())
|
||||
}
|
||||
|
||||
// filter out the folders we already own
|
||||
rawFolders := make([]unstructured.Unstructured, 0, len(rawList.Items))
|
||||
for _, f := range rawList.Items {
|
||||
repoName := f.GetAnnotations()[apiutils.AnnoKeyRepoName]
|
||||
if repoName == repo.Config().GetName() {
|
||||
logger.Info("skip as folder is already in repository", "folder", f.GetName())
|
||||
continue
|
||||
}
|
||||
rawFolders = append(rawFolders, f)
|
||||
}
|
||||
|
||||
status.Message = fmt.Sprintf("exporting folders (%d)...", len(rawFolders))
|
||||
if err := progress(ctx, status); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
folderTree := resources.NewFolderTreeFromUnstructure(ctx, rawFolders)
|
||||
if options.Folder != "" {
|
||||
return nil, fmt.Errorf("non-root folder not yet supported")
|
||||
}
|
||||
|
||||
// first create folders
|
||||
summary := provisioning.JobResourceSummary{
|
||||
Group: folders.GROUP,
|
||||
Resource: folders.RESOURCE,
|
||||
}
|
||||
err = folderTree.Walk(ctx, func(ctx context.Context, folder resources.Folder) error {
|
||||
p := folder.Path + "/"
|
||||
if options.Prefix != "" {
|
||||
p = options.Prefix + "/" + p
|
||||
}
|
||||
logger := logger.With("path", p)
|
||||
|
||||
_, err = repo.Read(ctx, p, ref)
|
||||
if err != nil && !(errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {
|
||||
logger.Error("failed to check if folder exists before writing", "error", err)
|
||||
return fmt.Errorf("failed to check if folder exists before writing: %w", err)
|
||||
} else if err == nil {
|
||||
logger.Info("folder already exists")
|
||||
summary.Noop++
|
||||
return nil
|
||||
}
|
||||
|
||||
// Create with an empty body will make a folder (or .keep file if unsupported)
|
||||
if err := repo.Create(ctx, p, ref, nil, "export folder `"+p+"`"); err != nil {
|
||||
logger.Error("failed to write a folder in repository", "error", err)
|
||||
return fmt.Errorf("failed to write folder in repo: %w", err)
|
||||
}
|
||||
summary.Create++
|
||||
logger.Debug("successfully exported folder")
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to write folders: %w", err)
|
||||
}
|
||||
status.Summary = append(status.Summary, summary)
|
||||
status.Message = "writing dashboards..."
|
||||
|
||||
if err := progress(ctx, status); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// TODO: handle pagination
|
||||
dashboardList, err := dashboardsClient.List(ctx, metav1.ListOptions{Limit: 1000})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to list dashboards: %w", err)
|
||||
}
|
||||
|
||||
commitMessage := fmt.Sprintf("grafana export to: %s", repo.Config().Name)
|
||||
summary = provisioning.JobResourceSummary{
|
||||
Group: "dashboard.grafana.app",
|
||||
Resource: "dashboards",
|
||||
}
|
||||
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()
|
||||
repoName := item.GetAnnotations()[apiutils.AnnoKeyRepoName]
|
||||
if repoName == repo.Config().GetName() {
|
||||
logger.Info("skip dashboard since it is already in repository", "dashboard", name)
|
||||
continue
|
||||
}
|
||||
|
||||
title, _, _ := unstructured.NestedString(item.Object, "spec", "title")
|
||||
if title == "" {
|
||||
title = name
|
||||
}
|
||||
folder := item.GetAnnotations()[apiutils.AnnoKeyFolder]
|
||||
|
||||
// Get the absolute path of the folder
|
||||
fid, ok := folderTree.DirPath(folder, "")
|
||||
if !ok {
|
||||
logger.Error("folder of item was not in tree of repository")
|
||||
return nil, fmt.Errorf("folder of item was not in tree of repository")
|
||||
}
|
||||
|
||||
delete(item.Object, "metadata")
|
||||
if options.Identifier {
|
||||
item.SetName(name) // keep the identifier in the metadata
|
||||
}
|
||||
|
||||
body, err := json.MarshalIndent(item.Object, "", " ")
|
||||
Resource: dashboards.DASHBOARD_RESOURCE,
|
||||
}}
|
||||
for _, kind := range kinds {
|
||||
err = worker.export(ctx, kind)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal dashboard %s: %w", name, err)
|
||||
}
|
||||
|
||||
fileName := slugify.Slugify(title) + ".json"
|
||||
if fid.Path != "" {
|
||||
fileName, err = safepath.Join(fid.Path, fileName)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error adding file path %s: %w", title, err)
|
||||
}
|
||||
}
|
||||
if options.Prefix != "" {
|
||||
fileName, err = safepath.Join(options.Prefix, fileName)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error adding path prefix %s: %w", options.Prefix, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Write the file
|
||||
err = repo.Write(ctx, fileName, ref, body, commitMessage)
|
||||
if err != nil {
|
||||
summary.Error++
|
||||
logger.Error("failed to write a file in repository", "error", err)
|
||||
if len(summary.Errors) < 20 {
|
||||
summary.Errors = append(summary.Errors, fmt.Sprintf("error writing: %s", fileName))
|
||||
}
|
||||
} else {
|
||||
summary.Write++
|
||||
return worker.jobStatus, err
|
||||
}
|
||||
}
|
||||
status.Summary = append(status.Summary, summary)
|
||||
status.State = provisioning.JobStateSuccess
|
||||
status.Message = ""
|
||||
return &status, nil
|
||||
|
||||
// Add summary info to response
|
||||
status := worker.jobStatus
|
||||
if !status.State.Finished() && err == nil {
|
||||
status.State = provisioning.JobStateSuccess
|
||||
status.Message = ""
|
||||
}
|
||||
return status, err
|
||||
}
|
||||
@@ -156,10 +156,11 @@ func (s *jobStore) drainPending() {
|
||||
status, err = s.processByWorker(ctx, worker, *job)
|
||||
if err != nil {
|
||||
logger.Error("error processing job", "error", err)
|
||||
status = &provisioning.JobStatus{
|
||||
State: provisioning.JobStateError,
|
||||
Errors: []string{err.Error()},
|
||||
if status == nil {
|
||||
status = &provisioning.JobStatus{}
|
||||
}
|
||||
status.State = provisioning.JobStateError
|
||||
status.Errors = append(status.Errors, err.Error())
|
||||
}
|
||||
if status == nil {
|
||||
status = &provisioning.JobStatus{
|
||||
@@ -167,8 +168,9 @@ func (s *jobStore) drainPending() {
|
||||
Errors: []string{"no status response from worker"},
|
||||
}
|
||||
}
|
||||
if status.State == "" {
|
||||
if !status.State.Finished() {
|
||||
status.State = provisioning.JobStateSuccess
|
||||
status.Message = ""
|
||||
}
|
||||
logger.Debug("job processing finished", "status", status.State)
|
||||
|
||||
@@ -218,19 +220,9 @@ func (s *jobStore) processByWorker(ctx context.Context, worker Worker, job provi
|
||||
return nil, errors.New("unknown repository")
|
||||
}
|
||||
|
||||
status, err := worker.Process(ctx, repo, job, func(ctx context.Context, j provisioning.JobStatus) error {
|
||||
return worker.Process(ctx, repo, job, func(ctx context.Context, j provisioning.JobStatus) error {
|
||||
return s.Update(ctx, job.Namespace, job.Name, j)
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("process job: %w", err)
|
||||
}
|
||||
|
||||
// TODO: does this really happen?
|
||||
if status == nil {
|
||||
return nil, errors.New("worker did not return a status")
|
||||
}
|
||||
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Checkout the next "pending" job
|
||||
|
||||
@@ -187,7 +187,7 @@ func (r *SyncWorker) sync(ctx context.Context,
|
||||
} else if !job.jobStatus.State.Finished() {
|
||||
job.jobStatus.State = provisioning.JobStateSuccess
|
||||
}
|
||||
job.jobStatus.Summary = []provisioning.JobResourceSummary{job.summary}
|
||||
job.jobStatus.Summary = []*provisioning.JobResourceSummary{&job.summary}
|
||||
return job.jobStatus, job.syncStatus, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -427,7 +427,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
|
||||
}
|
||||
b.repositoryLister = repoInformer.Lister()
|
||||
|
||||
b.jobs.Register(export.NewExportWorker(b.parsers))
|
||||
b.jobs.Register(export.NewExportWorker(b.client))
|
||||
b.jobs.Register(sync.NewSyncWorker(
|
||||
c.ProvisioningV0alpha1(),
|
||||
b.parsers,
|
||||
|
||||
@@ -63,7 +63,7 @@ func (c *syncConnector) Connect(
|
||||
var options provisioning.SyncJobOptions
|
||||
err := json.NewDecoder(r.Body).Decode(&options)
|
||||
if err != nil {
|
||||
responder.Error(apierrors.NewBadRequest("error decoding request"))
|
||||
responder.Error(apierrors.NewBadRequest("error decoding SyncJobOptions from request"))
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -3117,12 +3117,7 @@
|
||||
"description": "Summary of processed actions",
|
||||
"type": "array",
|
||||
"items": {
|
||||
"default": {},
|
||||
"allOf": [
|
||||
{
|
||||
"$ref": "#/components/schemas/com.github.grafana.grafana.pkg.apis.provisioning.v0alpha1.JobResourceSummary"
|
||||
}
|
||||
]
|
||||
"$ref": "#/components/schemas/com.github.grafana.grafana.pkg.apis.provisioning.v0alpha1.JobResourceSummary"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,6 +11,14 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.org/x/exp/rand"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
|
||||
dashboard "github.com/grafana/grafana/pkg/apis/dashboard/v1alpha1"
|
||||
folder "github.com/grafana/grafana/pkg/apis/folder/v0alpha1"
|
||||
provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
|
||||
@@ -21,13 +29,6 @@ import (
|
||||
"github.com/grafana/grafana/pkg/tests/apis"
|
||||
"github.com/grafana/grafana/pkg/tests/testinfra"
|
||||
"github.com/grafana/grafana/pkg/tests/testsuite"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.org/x/exp/rand"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
@@ -155,23 +156,6 @@ func TestIntegrationProvisioning(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("creating repository creates folder", func(t *testing.T) {
|
||||
cleanSlate(t)
|
||||
|
||||
_, err := client.Resource.Update(ctx,
|
||||
helper.LoadYAMLOrJSONFile("testdata/github-example.json"),
|
||||
metav1.UpdateOptions{},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.EventuallyWithT(t, func(collect *assert.CollectT) {
|
||||
resp, err := folderClient.Resource.Get(ctx, "github-example", metav1.GetOptions{})
|
||||
if assert.NoError(collect, err) {
|
||||
assert.Equal(collect, "github-example", mustNestedString(resp.Object, "metadata", "name"))
|
||||
}
|
||||
}, time.Second*2, time.Millisecond*20)
|
||||
})
|
||||
|
||||
t.Run("creating GitHub repository syncs from branch selected", func(t *testing.T) {
|
||||
cleanSlate(t)
|
||||
|
||||
@@ -313,7 +297,7 @@ func TestIntegrationProvisioning(t *testing.T) {
|
||||
require.NoError(t, err, "valid path should be fine")
|
||||
|
||||
// But the dashboard shouldn't exist yet
|
||||
const allPanels = "all-panels-5Y4ReX6LwL7d"
|
||||
const allPanels = "all-panels-hallaxjov44rbumtikbi1sbzroco9"
|
||||
_, err = dashboardClient.Resource.Get(ctx, allPanels, metav1.GetOptions{})
|
||||
require.Error(t, err, "no all-panels dashboard should exist")
|
||||
|
||||
@@ -323,6 +307,9 @@ func TestIntegrationProvisioning(t *testing.T) {
|
||||
Resource("repositories").
|
||||
Name(repo).
|
||||
SubResource("sync").
|
||||
Body(asJSON(provisioning.SyncJobOptions{
|
||||
Incremental: false,
|
||||
})).
|
||||
Do(ctx)
|
||||
|
||||
obj, err := result.Get()
|
||||
@@ -369,6 +356,11 @@ func mustNested(obj map[string]interface{}, fields ...string) interface{} {
|
||||
return v
|
||||
}
|
||||
|
||||
func asJSON(obj any) []byte {
|
||||
jj, _ := json.Marshal(obj)
|
||||
return jj
|
||||
}
|
||||
|
||||
func randomAsciiStr(n int) string {
|
||||
const alphabet string = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
||||
b := strings.Builder{}
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
"readOnly": false,
|
||||
"sync": {
|
||||
"enabled": true,
|
||||
"target": "mirror",
|
||||
"target": "folder",
|
||||
"intervalSeconds": 60
|
||||
},
|
||||
"github": {
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
"readOnly": false,
|
||||
"sync": {
|
||||
"enabled": false,
|
||||
"target": "",
|
||||
"target": "folder",
|
||||
"intervalSeconds": 60
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -9,7 +9,7 @@
|
||||
"description": "Load /devenv/dev-dashboards (from root of repository)",
|
||||
"sync": {
|
||||
"enabled": false,
|
||||
"target": "mirror",
|
||||
"target": "folder",
|
||||
"intervalSeconds": 60
|
||||
},
|
||||
"readOnly": false,
|
||||
|
||||
+2
-2
@@ -10,8 +10,8 @@
|
||||
"path": "the test will replace this string"
|
||||
},
|
||||
"sync": {
|
||||
"enabled": false,
|
||||
"target": "mirror",
|
||||
"enabled": true,
|
||||
"target": "folder",
|
||||
"intervalSeconds": 60
|
||||
},
|
||||
"readOnly": false,
|
||||
|
||||
+1
-1
@@ -11,7 +11,7 @@
|
||||
},
|
||||
"sync": {
|
||||
"enabled": false,
|
||||
"target": "mirror",
|
||||
"target": "folder",
|
||||
"intervalSeconds": 60
|
||||
},
|
||||
"readOnly": false,
|
||||
|
||||
+1
-1
@@ -12,7 +12,7 @@
|
||||
},
|
||||
"sync": {
|
||||
"enabled": false,
|
||||
"target": "mirror",
|
||||
"target": "folder",
|
||||
"intervalSeconds": 60
|
||||
},
|
||||
"readOnly": false,
|
||||
|
||||
@@ -522,8 +522,8 @@ export type CreateRepositorySyncArg = {
|
||||
/** name of the Job */
|
||||
name: string;
|
||||
body: {
|
||||
/** Complete forces the sync to overwrite */
|
||||
incremental?: boolean;
|
||||
/** Incremental synchronization for versioned repositories */
|
||||
incremental: boolean;
|
||||
};
|
||||
};
|
||||
export type CreateRepositoryTestResponse = /** status 200 OK */ TestResults;
|
||||
@@ -658,8 +658,8 @@ export type PullRequestJobOptions = {
|
||||
url?: string;
|
||||
};
|
||||
export type SyncJobOptions = {
|
||||
/** Complete forces the sync to overwrite */
|
||||
complete?: boolean;
|
||||
/** Incremental synchronization for versioned repositories */
|
||||
incremental: boolean;
|
||||
};
|
||||
export type JobSpec = {
|
||||
/** Possible enum values:
|
||||
@@ -808,6 +808,8 @@ export type SyncStatus = {
|
||||
finished?: number;
|
||||
/** The repository hash when the last sync ran */
|
||||
hash?: string;
|
||||
/** Incremental synchronization for versioned repositories */
|
||||
incremental?: boolean;
|
||||
/** The ID for the job that ran this sync */
|
||||
job?: string;
|
||||
/** Summary messages (will be shown to users) */
|
||||
|
||||
Reference in new issue
Block a user