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

feat: parallel processChange (#112198)

Browse Source 
* feat: parallel processChange

* all except move work

* fix: tests and order of operations

* fix: tests

* chore: review feedback

* chore: review feedback
This commit is contained in:
Costa Alexoglou
2025-10-16 19:17:04 +02:00
committed by GitHub
parent aa8af6b798
commit 163a88056e
11 changed files with 381 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/provisioning/resources/parser.go
+50 -7
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"path"
"go.opentelemetry.io/otel/attribute"
"gopkg.in/yaml.v3"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -22,6 +23,7 @@ import (
"github.com/grafana/grafana/pkg/apimachinery/apis/common/v0alpha1"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/apimachinery/utils"
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/util"
)
@@ -302,10 +304,15 @@ func (f *ParsedResource) Run(ctx context.Context) error {
}
// Always use the provisioning identity when writing
ctx, _, err := identity.WithProvisioningIdentity(ctx, f.Obj.GetNamespace())
identityCtx, _, err := identity.WithProvisioningIdentity(ctx, f.Obj.GetNamespace())
ctx, identitySpan := tracing.Start(identityCtx, "provisioning.resources.run_resource.set_identity")
if err != nil {
identitySpan.RecordError(err)
identitySpan.End()
return err
}
identitySpan.End()
fieldValidation := "Strict"
if f.GVR == DashboardResource {
@@ -318,42 +325,57 @@ func (f *ParsedResource) Run(ctx context.Context) error {
Identity: f.Repo.Name,
}
actionsCtx, actionsSpan := tracing.Start(ctx, "provisioning.resources.run_resource.actions")
defer actionsSpan.End()
// Handle deletion action
if f.Action == provisioning.ResourceActionDelete {
deleteCtx, deleteSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.delete")
deleteSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName()))
// If we don't have existing resource from DryRun, fetch it now
if f.DryRunResponse == nil {
f.Existing, err = f.Client.Get(ctx, f.Obj.GetName(), metav1.GetOptions{})
f.Existing, err = f.Client.Get(deleteCtx, f.Obj.GetName(), metav1.GetOptions{})
if err != nil {
deleteSpan.RecordError(err)
if apierrors.IsNotFound(err) {
// Resource doesn't exist, nothing to delete - this is fine
deleteSpan.End()
return nil
}
deleteSpan.End()
return fmt.Errorf("failed to get existing resource for delete: %w", err)
}
}
// Check ownership with the existing resource
if err := CheckResourceOwnership(f.Existing, f.Obj.GetName(), requestingManager); err != nil {
deleteSpan.RecordError(err)
deleteSpan.End()
return err
}
// Perform the actual delete
err = f.Client.Delete(ctx, f.Obj.GetName(), metav1.DeleteOptions{})
err = f.Client.Delete(deleteCtx, f.Obj.GetName(), metav1.DeleteOptions{})
if apierrors.IsNotFound(err) {
err = nil // ignorable - resource was already deleted
}
if err != nil {
deleteSpan.RecordError(err)
}
// Set the deleted resource as the result
if err == nil && f.Existing != nil {
f.Upsert = f.Existing.DeepCopy()
}
deleteSpan.End()
return err
}
// If we don't have existing resource from DryRun, fetch it now
if f.DryRunResponse == nil {
f.Existing, _ = f.Client.Get(ctx, f.Obj.GetName(), metav1.GetOptions{})
f.Existing, _ = f.Client.Get(actionsCtx, f.Obj.GetName(), metav1.GetOptions{})
}
// Check ownership with the existing resource (if any)
@@ -364,9 +386,16 @@ func (f *ParsedResource) Run(ctx context.Context) error {
// If we have already tried loading existing, start with create
if f.DryRunResponse != nil && f.Existing == nil {
f.Action = provisioning.ResourceActionCreate
f.Upsert, err = f.Client.Create(ctx, f.Obj, metav1.CreateOptions{
createCtx, createSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.create")
createSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName()))
f.Upsert, err = f.Client.Create(createCtx, f.Obj, metav1.CreateOptions{
FieldValidation: fieldValidation,
})
if err != nil {
createSpan.RecordError(err)
}
createSpan.End()
if err == nil {
return nil // it worked, return
}
@@ -374,14 +403,28 @@ func (f *ParsedResource) Run(ctx context.Context) error {
// Try update, otherwise create
f.Action = provisioning.ResourceActionUpdate
f.Upsert, err = f.Client.Update(ctx, f.Obj, metav1.UpdateOptions{
updateCtx, updateSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.update")
updateSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName()))
f.Upsert, err = f.Client.Update(updateCtx, f.Obj, metav1.UpdateOptions{
FieldValidation: fieldValidation,
})
if err != nil {
updateSpan.RecordError(err)
}
updateSpan.End()
if apierrors.IsNotFound(err) {
f.Action = provisioning.ResourceActionCreate
f.Upsert, err = f.Client.Create(ctx, f.Obj, metav1.CreateOptions{
fallbackCreateCtx, fallbackCreateSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.create_fallback")
fallbackCreateSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName()))
f.Upsert, err = f.Client.Create(fallbackCreateCtx, f.Obj, metav1.CreateOptions{
FieldValidation: fieldValidation,
})
if err != nil {
fallbackCreateSpan.RecordError(err)
}
fallbackCreateSpan.End()
}
return err
}
pkg/registry/apis/provisioning/resources/resources.go
+46 -7
View File
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"slices"
"sync"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -17,6 +18,7 @@ import (
"github.com/grafana/grafana/apps/provisioning/pkg/safepath"
"github.com/grafana/grafana/pkg/apimachinery/utils"
"github.com/grafana/grafana/pkg/infra/slugify"
"github.com/grafana/grafana/pkg/infra/tracing"
)
var (
@@ -55,6 +57,7 @@ type ResourcesManager struct {
parser Parser
clients ResourceClients
resourcesLookup map[resourceID]string // the path with this k8s name
mu sync.RWMutex
}
func NewResourcesManager(repo repository.ReaderWriter, folders *FolderManager, parser Parser, clients ResourceClients) *ResourcesManager {
@@ -67,6 +70,25 @@ func NewResourcesManager(repo repository.ReaderWriter, folders *FolderManager, p
}
}
// findResource checks if a resource exists in the lookup map (read operation)
func (r *ResourcesManager) findResource(id resourceID) (string, bool) {
r.mu.RLock()
defer r.mu.RUnlock()
path, found := r.resourcesLookup[id]
return path, found
}
func (r *ResourcesManager) addResource(id resourceID, path string) {
r.mu.Lock()
defer r.mu.Unlock()
if _, found := r.resourcesLookup[id]; found {
return
}
r.resourcesLookup[id] = path
}
// CheckResourceOwnership validates that the requesting manager can modify the existing resource
// Returns an error if the existing resource is owned by a different manager that doesn't allow edits
// If existingResource is nil, no ownership conflict exists (new resource)
@@ -193,15 +215,23 @@ func (r *ResourcesManager) WriteResourceFileFromObject(ctx context.Context, obj
func (r *ResourcesManager) WriteResourceFromFile(ctx context.Context, path string, ref string) (string, schema.GroupVersionKind, error) {
// Read the referenced file
fileInfo, err := r.repo.Read(ctx, path, ref)
readCtx, readSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.read_file")
fileInfo, err := r.repo.Read(readCtx, path, ref)
if err != nil {
readSpan.RecordError(err)
readSpan.End()
return "", schema.GroupVersionKind{}, fmt.Errorf("failed to read file: %w", err)
}
readSpan.End()
parsed, err := r.parser.Parse(ctx, fileInfo)
parseCtx, parseSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.parse_file")
parsed, err := r.parser.Parse(parseCtx, fileInfo)
if err != nil {
parseSpan.RecordError(err)
parseSpan.End()
return "", schema.GroupVersionKind{}, fmt.Errorf("failed to parse file: %w", err)
}
parseSpan.End()
if parsed.Obj.GetName() == "" {
return "", schema.GroupVersionKind{}, ErrMissingName
@@ -213,27 +243,36 @@ func (r *ResourcesManager) WriteResourceFromFile(ctx context.Context, path strin
Resource: parsed.GVR.Resource,
Group: parsed.GVK.Group,
}
existing, found := r.resourcesLookup[id]
if found {
if existing, found := r.findResource(id); found {
return "", parsed.GVK, fmt.Errorf("duplicate resource name: %s, %s and %s: %w", parsed.Obj.GetName(), path, existing, ErrDuplicateName)
}
r.resourcesLookup[id] = path
r.addResource(id, path)
// For resources that exist in folders, set the header annotation
if slices.Contains(SupportsFolderAnnotation, parsed.GVR.GroupResource()) {
// Make sure the parent folders exist
folder, err := r.folders.EnsureFolderPathExist(ctx, path)
folderCtx, folderSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.ensure_folder")
folder, err := r.folders.EnsureFolderPathExist(folderCtx, path)
if err != nil {
folderSpan.RecordError(err)
folderSpan.End()
return "", parsed.GVK, fmt.Errorf("failed to ensure folder path exists: %w", err)
}
parsed.Meta.SetFolder(folder)
folderSpan.End()
}
// Clear any saved identifiers
parsed.Meta.SetUID("")
parsed.Meta.SetResourceVersion("")
err = parsed.Run(ctx)
runCtx, runSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.run_resource")
err = parsed.Run(runCtx)
if err != nil {
runSpan.RecordError(err)
}
runSpan.End()
return parsed.Obj.GetName(), parsed.GVK, err
}
pkg/registry/apis/provisioning/resources/tree.go
+30 -2
View File
@@ -3,6 +3,7 @@ package resources
import (
"context"
"fmt"
"sync"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -30,11 +31,18 @@ type folderTree struct {
tree map[string]string
folders map[string]Folder
count int
mu sync.RWMutex
}
// In determines if the given folder is in the tree at all. That is, it answers "does the folder even exist in the Grafana instance?"
// An empty folder string means the root folder, and is special-cased to always return true.
func (t *folderTree) In(folder string) bool {
t.mu.RLock()
defer t.mu.RUnlock()
return t.in(folder)
}
func (t *folderTree) in(folder string) bool {
_, ok := t.tree[folder]
return ok || folder == ""
}
@@ -47,7 +55,19 @@ func (t *folderTree) In(folder string) bool {
// If In(folder) or In(baseFolder) is false, this will return ok=false, because it would be undefined behaviour.
// If baseFolder is not a parent of folder, ok=false is returned.
func (t *folderTree) DirPath(folder, baseFolder string) (fid Folder, ok bool) {
if !t.In(folder) || !t.In(baseFolder) {
t.mu.RLock()
defer t.mu.RUnlock()
return t.dirPath(folder, baseFolder)
}
// dirPath is the internal implementation that assumes the mutex is already held
// Needed to avoid deadlock when called from other methods that hold locks like Walk()
func (t *folderTree) dirPath(folder, baseFolder string) (fid Folder, ok bool) {
// Inline In() logic to avoid deadlock when called from other methods that hold locks
folderInTree := t.in(folder)
baseFolderInTree := t.in(baseFolder)
if !folderInTree || !baseFolderInTree {
return Folder{}, false
}
if folder == "" && baseFolder != "" {
@@ -76,21 +96,27 @@ func (t *folderTree) DirPath(folder, baseFolder string) (fid Folder, ok bool) {
}
func (t *folderTree) Add(folder Folder, parent string) {
t.mu.Lock()
defer t.mu.Unlock()
t.tree[folder.ID] = parent
t.folders[folder.ID] = folder
t.count++
}
func (t *folderTree) Count() int {
t.mu.RLock()
defer t.mu.RUnlock()
return t.count
}
type WalkFunc func(ctx context.Context, folder Folder, parent string) error
func (t *folderTree) Walk(ctx context.Context, fn WalkFunc) error {
t.mu.RLock()
defer t.mu.RUnlock()
toWalk := make([]Folder, 0, len(t.folders))
for _, folder := range t.folders {
folder, _ := t.DirPath(folder.ID, "")
folder, _ := t.dirPath(folder.ID, "")
toWalk = append(toWalk, folder)
}
@@ -123,6 +149,8 @@ func (t *folderTree) AddUnstructured(item *unstructured.Unstructured) error {
Title: meta.FindTitle(item.GetName()),
ID: item.GetName(),
}
t.mu.Lock()
defer t.mu.Unlock()
t.tree[folder.ID] = meta.GetFolder()
t.folders[folder.ID] = folder
t.count++