grafana/pkg/registry/apis/provisioning/jobs/progress.go

package jobs

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/grafana/grafana-app-sdk/logging"
	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
	"github.com/grafana/grafana/apps/provisioning/pkg/repository"
	"github.com/grafana/grafana/apps/provisioning/pkg/safepath"
	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
)

// maybeNotifyProgress will only notify if a certain amount of time has passed
// or if the job completed
func maybeNotifyProgress(threshold time.Duration, fn ProgressFn) ProgressFn {
	var last time.Time
	var mu sync.Mutex

	return func(ctx context.Context, status provisioning.JobStatus) error {
		mu.Lock()
		shouldNotify := status.Finished != 0 || last.IsZero() || time.Since(last) > threshold
		if shouldNotify {
			last = time.Now()
		}
		mu.Unlock()

		if shouldNotify {
			return fn(ctx, status)
		}

		return nil
	}
}

// FIXME: ProgressRecorder should be initialized in the queue
type JobResourceResult struct {
	Name    string
	Group   string
	Kind    string
	Path    string
	Action  repository.FileAction
	Error   error
	Warning error
}

type jobProgressRecorder struct {
	mu                  sync.RWMutex
	started             time.Time
	total               int
	maxErrors           int
	message             string
	finalMessage        string
	resultCount         int
	errorCount          int
	errors              []string
	refURLs             *provisioning.RepositoryURLs
	notifyImmediatelyFn ProgressFn
	maybeNotifyFn       ProgressFn
	summaries           map[string]*provisioning.JobResourceSummary
	failedCreations     []string // Tracks folder paths that failed to be created
	failedDeletions     []string // Tracks resource paths that failed to be deleted
}

func newJobProgressRecorder(ProgressFn ProgressFn) JobProgressRecorder {
	return &jobProgressRecorder{
		started: time.Now(),
		// Have a faster notifier for messages and total
		notifyImmediatelyFn: maybeNotifyProgress(500*time.Millisecond, ProgressFn),
		maybeNotifyFn:       maybeNotifyProgress(5*time.Second, ProgressFn),
		summaries:           make(map[string]*provisioning.JobResourceSummary),
	}
}

func (r *jobProgressRecorder) Started() time.Time {
	return r.started
}

func (r *jobProgressRecorder) Record(ctx context.Context, result JobResourceResult) {
	var shouldLogError bool
	var logErr error

	r.mu.Lock()
	r.resultCount++

	if result.Error != nil {
		shouldLogError = true
		logErr = result.Error

		// Don't count ignored actions as errors in error count or error list
		if result.Action != repository.FileActionIgnored {
			if len(r.errors) < 20 {
				r.errors = append(r.errors, result.Error.Error())
			}
			r.errorCount++
		}

		// Automatically track failed operations based on error type and action
		// Check if this is a PathCreationError (folder creation failure)
		var pathErr *resources.PathCreationError
		if errors.As(result.Error, &pathErr) {
			r.failedCreations = append(r.failedCreations, pathErr.Path)
		}

		// Track failed deletions, any deletion will stop the deletion of the parent folder (as it won't be empty)
		if result.Action == repository.FileActionDeleted {
			r.failedDeletions = append(r.failedDeletions, result.Path)
		}
	}

	r.updateSummary(result)
	r.mu.Unlock()

	logger := logging.FromContext(ctx).With("path", result.Path, "group", result.Group, "kind", result.Kind, "action", result.Action, "name", result.Name)
	if shouldLogError {
		logger.Error("job resource operation failed", "err", logErr)
	} else {
		logger.Info("job resource operation succeeded")
	}

	r.maybeNotify(ctx)
}

// ResetResults will reset the results of the job
func (r *jobProgressRecorder) ResetResults() {
	r.mu.Lock()
	defer r.mu.Unlock()

	r.resultCount = 0
	r.errorCount = 0
	r.errors = nil
	r.summaries = make(map[string]*provisioning.JobResourceSummary)
	r.failedCreations = nil
	r.failedDeletions = nil
}

func (r *jobProgressRecorder) SetMessage(ctx context.Context, msg string) {
	r.mu.Lock()
	r.message = msg
	r.mu.Unlock()

	logging.FromContext(ctx).Info("job progress message", "message", msg)
	r.notifyImmediately(ctx)
}

func (r *jobProgressRecorder) SetFinalMessage(ctx context.Context, msg string) {
	r.mu.Lock()
	r.finalMessage = msg
	r.mu.Unlock()

	logging.FromContext(ctx).Info("job final message", "message", msg)
}

func (r *jobProgressRecorder) SetRefURLs(ctx context.Context, refURLs *provisioning.RepositoryURLs) {
	r.mu.Lock()
	r.refURLs = refURLs
	r.mu.Unlock()

	if refURLs != nil {
		logging.FromContext(ctx).Debug("job ref URLs set", "sourceURL", refURLs.SourceURL, "compareURL", refURLs.CompareURL, "newPullRequestURL", refURLs.NewPullRequestURL)
	} else {
		logging.FromContext(ctx).Debug("job ref URLs cleared")
	}
}

func (r *jobProgressRecorder) SetTotal(ctx context.Context, total int) {
	r.mu.Lock()
	r.total = total
	r.mu.Unlock()

	r.notifyImmediately(ctx)
}

func (r *jobProgressRecorder) StrictMaxErrors(maxErrors int) {
	r.mu.Lock()
	r.maxErrors = maxErrors
	r.mu.Unlock()
}

func (r *jobProgressRecorder) TooManyErrors() error {
	if r.maxErrors > 0 && r.errorCount >= r.maxErrors {
		return fmt.Errorf("too many errors: %d", r.errorCount)
	}

	return nil
}

func (r *jobProgressRecorder) summary() []*provisioning.JobResourceSummary {
	if len(r.summaries) == 0 {
		return nil
	}

	summaries := make([]*provisioning.JobResourceSummary, 0, len(r.summaries))
	for _, summary := range r.summaries {
		summaries = append(summaries, summary)
	}

	return summaries
}

func (r *jobProgressRecorder) updateSummary(result JobResourceResult) {
	// Note: This method is called from Record() which already holds the lock
	key := result.Group + ":" + result.Kind
	summary, exists := r.summaries[key]
	if !exists {
		summary = &provisioning.JobResourceSummary{
			Group: result.Group,
			Kind:  result.Kind,
		}
		r.summaries[key] = summary
	}

	if result.Error != nil {
		errorMsg := fmt.Sprintf("%s (file: %s, name: %s, action: %s)", result.Error.Error(), result.Path, result.Name, result.Action)
		summary.Errors = append(summary.Errors, errorMsg)
		summary.Error++
	} else if result.Warning != nil {
		warningMsg := fmt.Sprintf("%s (file: %s, name: %s, action: %s)", result.Warning.Error(), result.Path, result.Name, result.Action)
		summary.Warnings = append(summary.Warnings, warningMsg)
		summary.Warning++
	} else {
		switch result.Action {
		case repository.FileActionDeleted:
			summary.Delete++
		case repository.FileActionUpdated:
			summary.Update++
		case repository.FileActionCreated:
			summary.Create++
		case repository.FileActionIgnored:
			summary.Noop++
		case repository.FileActionRenamed:
			summary.Delete++
			summary.Create++
		}
		summary.Write = summary.Create + summary.Update
	}
}

func (r *jobProgressRecorder) progress() float64 {
	// Note: This method is called from currentStatus() which already holds the lock
	if r.total == 0 {
		return 0
	}

	return float64(r.resultCount) / float64(r.total) * 100
}

func (r *jobProgressRecorder) currentStatus() provisioning.JobStatus {
	r.mu.RLock()
	defer r.mu.RUnlock()

	return provisioning.JobStatus{
		Started:  r.started.UnixMilli(),
		State:    provisioning.JobStateWorking,
		Message:  r.message,
		Errors:   r.errors,
		Progress: r.progress(),
		Summary:  r.summary(),
	}
}

func (r *jobProgressRecorder) notifyImmediately(ctx context.Context) {
	jobStatus := r.currentStatus()
	logger := logging.FromContext(ctx)
	if err := r.notifyImmediatelyFn(ctx, jobStatus); err != nil {
		logger.Warn("error notifying immediate progress", "err", err)
	}
}

func (r *jobProgressRecorder) maybeNotify(ctx context.Context) {
	jobStatus := r.currentStatus()

	logger := logging.FromContext(ctx)
	if err := r.maybeNotifyFn(ctx, jobStatus); err != nil {
		logger.Warn("error notifying progress", "err", err)
	}
}

func (r *jobProgressRecorder) Complete(ctx context.Context, err error) provisioning.JobStatus {
	r.mu.RLock()

	jobStatus := provisioning.JobStatus{
		Started:  r.started.UnixMilli(),
		Finished: time.Now().UnixMilli(),
		State:    provisioning.JobStateSuccess,
		Message:  "completed successfully",
	}

	if err != nil {
		jobStatus.State = provisioning.JobStateError
		jobStatus.Message = err.Error()
	}

	summaries := r.summary()
	jobStatus.Summary = summaries
	jobStatus.Errors = r.errors

	// Extract warnings from summaries
	warnings := make([]string, 0)
	for _, summary := range summaries {
		warnings = append(warnings, summary.Warnings...)
	}
	jobStatus.Warnings = warnings

	jobStatus.URLs = r.refURLs

	tooManyErrors := r.maxErrors > 0 && r.errorCount >= r.maxErrors
	finalMessage := r.finalMessage

	r.mu.RUnlock()

	if len(jobStatus.Errors) > 0 && jobStatus.State != provisioning.JobStateError {
		if tooManyErrors {
			jobStatus.Message = "completed with too many errors"
			jobStatus.State = provisioning.JobStateError
		} else {
			jobStatus.Message = "completed with errors"
			jobStatus.State = provisioning.JobStateWarning
		}
	} else if len(jobStatus.Warnings) > 0 {
		jobStatus.State = provisioning.JobStateWarning
		jobStatus.Message = "completed with warnings"
	}

	// Override message if progress have a more explicit message
	if finalMessage != "" && jobStatus.State != provisioning.JobStateError {
		jobStatus.Message = finalMessage
	}

	return jobStatus
}

// HasDirPathFailedCreation checks if a path is nested under any failed folder creation
func (r *jobProgressRecorder) HasDirPathFailedCreation(path string) bool {
	r.mu.RLock()
	defer r.mu.RUnlock()

	for _, failedCreation := range r.failedCreations {
		if safepath.InDir(path, failedCreation) {
			return true
		}
	}
	return false
}

// HasDirPathFailedDeletion checks if any resource deletions failed under a folder path
func (r *jobProgressRecorder) HasDirPathFailedDeletion(folderPath string) bool {
	r.mu.RLock()
	defer r.mu.RUnlock()

	for _, failedDeletion := range r.failedDeletions {
		if safepath.InDir(failedDeletion, folderPath) {
			return true
		}
	}
	return false
}