diff --git a/services/actions/job_emitter.go b/services/actions/job_emitter.go index 544825f045..6c2b7e27c9 100644 --- a/services/actions/job_emitter.go +++ b/services/actions/job_emitter.go @@ -290,7 +290,7 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action } if len(jobs) > oldJobCount { - log.Info("Matrix re-evaluation created %d new jobs for run %d (was %d, now %d)", + log.Debug("Matrix re-evaluation created %d new jobs for run %d (was %d, now %d)", len(jobs)-oldJobCount, run.ID, oldJobCount, len(jobs)) } @@ -407,17 +407,13 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model continue } - // If new matrix jobs were created, add them to the resolver and continue + // If new matrix jobs were created, the placeholder was already claimed and + // skipped inside ReEvaluateMatrixForJobWithNeeds (transactionally). Nothing + // more to do here — just move to the next job. if len(newMatrixJobs) > 0 { resolveMetrics.matrixReevaluated++ log.Info("Matrix re-evaluation succeeded for job %d (JobID: %s): created %d new jobs (duration: %dms)", id, actionRunJob.JobID, len(newMatrixJobs), duration) - - // Mark the original matrix placeholder job as skipped so it won't be run later. - actionRunJob.Status = actions_model.StatusSkipped - if _, err := db.GetEngine(ctx).ID(actionRunJob.ID).Cols("status").Update(actionRunJob); err != nil { - log.Error("Failed to mark matrix placeholder job %d (JobID: %s) as skipped after re-evaluation: %v", id, actionRunJob.JobID, err) - } continue } @@ -461,10 +457,10 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model switch newStatus { case actions_model.StatusWaiting: resolveMetrics.jobsStarted++ - log.Info("Job %d (JobID: %s) transitioned to StatusWaiting", id, actionRunJob.JobID) + log.Debug("Job %d (JobID: %s) transitioned to StatusWaiting", id, actionRunJob.JobID) case actions_model.StatusSkipped: resolveMetrics.jobsSkipped++ - log.Info("Job %d (JobID: %s) transitioned to StatusSkipped", id, actionRunJob.JobID) + log.Debug("Job %d (JobID: %s) transitioned to StatusSkipped", id, actionRunJob.JobID) } } } diff --git a/services/actions/matrix.go b/services/actions/matrix.go index 0a2e41e0c0..cdba5f8375 100644 --- a/services/actions/matrix.go +++ b/services/actions/matrix.go @@ -10,17 +10,19 @@ import ( "maps" "sort" "strings" - "time" actions_model "code.gitea.io/gitea/models/actions" + "code.gitea.io/gitea/models/db" "code.gitea.io/gitea/modules/actions/jobparser" "code.gitea.io/gitea/modules/log" "go.yaml.in/yaml/v4" + "xorm.io/builder" ) -// markMatrixAsEvaluatedAndSkip marks a job's matrix as evaluated and skipped. -// Used when matrix cannot be expanded or dependency checks fail. +// markMatrixAsEvaluatedAndSkip marks a job as evaluated and skipped unconditionally. +// Used when matrix cannot be expanded or dependency checks fail (no concurrent +// insertion has taken place, so no race is possible). func markMatrixAsEvaluatedAndSkip(ctx context.Context, job *actions_model.ActionRunJob, reason string) error { job.IsMatrixEvaluated = true job.Status = actions_model.StatusSkipped @@ -32,6 +34,23 @@ func markMatrixAsEvaluatedAndSkip(ctx context.Context, job *actions_model.Action return nil } +// claimMatrixExpansion atomically marks the placeholder job as evaluated+skipped +// only when it is still in its original state (is_matrix_evaluated=false AND +// status=Blocked). Returns (true, nil) when this caller wins the claim, or +// (false, nil) when another concurrent process already claimed it. +// This is the concurrency guard that prevents double-expansion. +func claimMatrixExpansion(ctx context.Context, job *actions_model.ActionRunJob) (bool, error) { + job.IsMatrixEvaluated = true + job.Status = actions_model.StatusSkipped + n, err := actions_model.UpdateRunJob(ctx, job, + builder.Eq{"is_matrix_evaluated": false, "status": actions_model.StatusBlocked}, + "is_matrix_evaluated", "status") + if err != nil { + return false, err + } + return n == 1, nil +} + // checkTaskNeedsReady verifies if all task dependencies are completed. // Returns (taskNeeds, allDone, error) func checkTaskNeedsReady(ctx context.Context, job *actions_model.ActionRunJob) (map[string]*TaskNeed, bool, error) { @@ -42,7 +61,6 @@ func checkTaskNeedsReady(ctx context.Context, job *actions_model.ActionRunJob) ( log.Debug("Found %d task needs for job %d (JobID: %s)", len(taskNeeds), job.ID, job.JobID) - // Check if any task needs are not done var pendingNeeds []string for jobID, taskNeed := range taskNeeds { if !taskNeed.Result.IsDone() { @@ -52,15 +70,14 @@ func checkTaskNeedsReady(ctx context.Context, job *actions_model.ActionRunJob) ( if len(pendingNeeds) > 0 { log.Debug("Matrix re-evaluation deferred for job %d: pending needs: %v", job.ID, pendingNeeds) - GetMatrixMetrics().RecordDeferred() return taskNeeds, false, nil } return taskNeeds, true, nil } -// ExtractRawStrategies extracts strategy definitions from the raw workflow content -// Returns a map of jobID to strategy YAML for jobs that have matrix dependencies +// ExtractRawStrategies extracts strategy definitions from the raw workflow content. +// Returns a map of jobID to strategy YAML for jobs that have matrix dependencies. func ExtractRawStrategies(content []byte) (map[string]string, error) { var workflowDef struct { Jobs map[string]struct { @@ -79,7 +96,6 @@ func ExtractRawStrategies(content []byte) (map[string]string, error) { continue } - // Check if this job has needs (dependencies) var needsList []string switch needs := jobDef.Needs.(type) { case string: @@ -92,7 +108,6 @@ func ExtractRawStrategies(content []byte) (map[string]string, error) { } } - // Only store strategy for jobs with dependencies if len(needsList) > 0 { if strategyBytes, err := yaml.Marshal(jobDef.Strategy); err == nil { strategies[jobID] = string(strategyBytes) @@ -103,25 +118,86 @@ func ExtractRawStrategies(content []byte) (map[string]string, error) { return strategies, nil } -// HasMatrixWithNeeds checks if a job's strategy contains a matrix that depends on job outputs +// HasMatrixWithNeeds reports whether rawStrategy contains a matrix value whose +// expression tree references needs..outputs.. +// It walks the parsed YAML tree to avoid false positives from values such as +// "os: [needs.review-runner]" that merely contain the substring "needs.". func HasMatrixWithNeeds(rawStrategy string) bool { if rawStrategy == "" { return false } - var strategy map[string]any - if err := yaml.Unmarshal([]byte(rawStrategy), &strategy); err != nil { + var root yaml.Node + if err := yaml.Unmarshal([]byte(rawStrategy), &root); err != nil { return false } - matrix, ok := strategy["matrix"] - if !ok { + // The top-level document node wraps a single mapping node. + doc := &root + if doc.Kind == yaml.DocumentNode && len(doc.Content) == 1 { + doc = doc.Content[0] + } + + // Find the "matrix" key inside the strategy mapping. + var matrixNode *yaml.Node + if doc.Kind == yaml.MappingNode { + for i := 0; i+1 < len(doc.Content); i += 2 { + if doc.Content[i].Value == "matrix" { + matrixNode = doc.Content[i+1] + break + } + } + } + if matrixNode == nil { return false } - // Check if any matrix value contains "needs." reference - matrixStr := fmt.Sprintf("%v", matrix) - return strings.Contains(matrixStr, "needs.") + return yamlNodeContainsNeedsOutputsExpr(matrixNode) +} + +// yamlNodeContainsNeedsOutputsExpr recursively inspects a yaml.Node and +// returns true if any scalar value contains a GitHub Actions expression of +// the form ${{ ... needs..outputs. ... }}. +func yamlNodeContainsNeedsOutputsExpr(node *yaml.Node) bool { + if node == nil { + return false + } + if node.Kind == yaml.ScalarNode { + return containsNeedsOutputsExpr(node.Value) + } + for _, child := range node.Content { + if yamlNodeContainsNeedsOutputsExpr(child) { + return true + } + } + return false +} + +// containsNeedsOutputsExpr returns true when s contains a GitHub Actions +// expression (${{ ... }}) that references needs..outputs.. +// A bare "needs." substring outside an expression block is not a match. +func containsNeedsOutputsExpr(s string) bool { + if !strings.Contains(s, "${{") { + return false + } + for i := 0; i < len(s); { + start := strings.Index(s[i:], "${{") + if start == -1 { + break + } + start += i + end := strings.Index(s[start:], "}}") + if end == -1 { + break + } + end += start + expr := s[start : end+2] + if strings.Contains(expr, "needs.") && strings.Contains(expr, ".outputs.") { + return true + } + i = end + 2 + } + return false } // ReEvaluateMatrixForJobWithNeeds re-evaluates the matrix strategy of a job once all its @@ -129,8 +205,6 @@ func HasMatrixWithNeeds(rawStrategy string) bool { // ActionRunJobs. The original placeholder job is marked as evaluated and skipped. // Returns nil, nil if the job is not ready yet or has nothing to do. func ReEvaluateMatrixForJobWithNeeds(ctx context.Context, job *actions_model.ActionRunJob, vars map[string]string) ([]*actions_model.ActionRunJob, error) { - startTime := time.Now() - if job.IsMatrixEvaluated || job.RawStrategy == "" { return nil, nil } @@ -149,31 +223,26 @@ func ReEvaluateMatrixForJobWithNeeds(ctx context.Context, job *actions_model.Act return nil, origErr } - // Check if dependencies are ready BEFORE doing expensive parsing taskNeeds, allDone, err := checkTaskNeedsReady(ctx, job) if err != nil { log.Error("Matrix re-evaluation error for job %d: check task needs: %v", job.ID, err) return skipWithError(fmt.Sprintf("task needs check failed: %v", err), fmt.Errorf("check task needs: %w", err)) } if !allDone { - return nil, nil // wait for dependencies + return nil, nil } mergedVars := mergeNeedsIntoVars(vars, taskNeeds) - // Load run and its attributes for expression evaluation context if job.Run == nil { if err := job.LoadRun(ctx); err != nil { - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) return nil, fmt.Errorf("load run: %w", err) } } if job.Run == nil { - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) return nil, errors.New("run not found after loading") } if err := job.Run.LoadAttributes(ctx); err != nil { - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) return nil, fmt.Errorf("load run attributes: %w", err) } @@ -186,27 +255,12 @@ func ReEvaluateMatrixForJobWithNeeds(ctx context.Context, job *actions_model.Act jobResults[jobID] = need.Result.String() } - // Build a minimal workflow containing this job and stubs for its dependencies, - // so the jobparser can resolve needs.*.outputs.* expressions. workflowYAML, err := constructWorkflowWithNeeds(job, taskNeeds) if err != nil { log.Error("Matrix re-evaluation error for job %d: construct workflow: %v", job.ID, err) - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) return skipWithError(fmt.Sprintf("workflow construction failed: %v", err), fmt.Errorf("construct workflow: %w", err)) } - // Track cache hit/miss for metrics (actual caching of parsed objects is future work) - cacheKey := computeCacheKey(workflowYAML, taskNeeds) - if cache := getWorkflowParseCache(); cache != nil { - if _, hit := cache.Get(cacheKey); hit { - log.Debug("Cache hit for workflow parse (job %d, key: %.16s)", job.ID, cacheKey) - GetMatrixMetrics().RecordCacheHit() - } else { - GetMatrixMetrics().RecordCacheMiss() - } - } - - parseStart := time.Now() parsedJobs, err := jobparser.Parse( workflowYAML, jobparser.WithVars(mergedVars), @@ -214,12 +268,8 @@ func ReEvaluateMatrixForJobWithNeeds(ctx context.Context, job *actions_model.Act jobparser.WithJobOutputs(jobOutputs), jobparser.WithJobResults(jobResults), ) - GetMatrixMetrics().RecordParseTime(time.Since(parseStart)) - if err != nil { log.Error("Matrix parse error for job %d (RawStrategy: %s): %v", job.ID, job.RawStrategy, err) - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) - // Don't propagate parse errors — mark as evaluated to avoid retry loops if markErr := markMatrixAsEvaluatedAndSkip(ctx, job, fmt.Sprintf("parse failed: %v", err)); markErr != nil { return nil, fmt.Errorf("parse workflow: %w; additionally failed to mark as evaluated: %v", err, markErr) } @@ -228,15 +278,9 @@ func ReEvaluateMatrixForJobWithNeeds(ctx context.Context, job *actions_model.Act if len(parsedJobs) == 0 { log.Debug("No jobs generated from matrix expansion for job %d (JobID: %s)", job.ID, job.JobID) - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) return nil, markMatrixAsEvaluatedAndSkip(ctx, job, "no jobs generated from matrix expansion") } - // Cache successful parse result (YAML payload for metrics tracking) - if cache := getWorkflowParseCache(); cache != nil && len(parsedJobs) > 0 { - cache.Set(cacheKey, workflowYAML) - } - log.Debug("Parsed %d matrix combinations for job %d (JobID: %s)", len(parsedJobs), job.ID, job.JobID) var newJobs []*actions_model.ActionRunJob @@ -247,7 +291,6 @@ func ReEvaluateMatrixForJobWithNeeds(ctx context.Context, job *actions_model.Act continue } if id != job.JobID { - // Skip dependency stubs — we only want matrix-expanded entries for the target job continue } needs := jobDef.Needs() @@ -267,77 +310,74 @@ func ReEvaluateMatrixForJobWithNeeds(ctx context.Context, job *actions_model.Act JobID: id, Needs: needs, RunsOn: jobDef.RunsOn(), - // All dependency jobs are already done at this point; start as Waiting. - Status: actions_model.StatusWaiting, + Status: actions_model.StatusWaiting, }) } if len(newJobs) == 0 { log.Warn("No valid jobs created from matrix expansion for job %d (JobID: %s), total parsed: %d", job.ID, job.JobID, len(parsedJobs)) - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) return nil, markMatrixAsEvaluatedAndSkip(ctx, job, fmt.Sprintf("no valid jobs created (parsed: %d)", len(parsedJobs))) } - insertStart := time.Now() - if err := actions_model.InsertActionRunJobs(ctx, newJobs); err != nil { - log.Error("Matrix insertion error: failed to insert %d new matrix jobs for job %d (JobID: %s): %v", len(newJobs), job.ID, job.JobID, err) - GetMatrixMetrics().RecordInsertTime(time.Since(insertStart)) - GetMatrixMetrics().RecordReevaluation(time.Since(startTime), false, 0) - return nil, fmt.Errorf("insert new jobs: %w", err) + // Atomically claim the placeholder and insert the new jobs in one transaction. + // The conditional WHERE in claimMatrixExpansion (is_matrix_evaluated=false AND + // status=Blocked) ensures only one concurrent caller can win. The second caller + // sees n==0 and rolls back without inserting duplicate jobs. + var claimed bool + if err := db.WithTx(ctx, func(txCtx context.Context) error { + var txErr error + claimed, txErr = claimMatrixExpansion(txCtx, job) + if txErr != nil { + return txErr + } + if !claimed { + return nil + } + return actions_model.InsertActionRunJobs(txCtx, newJobs) + }); err != nil { + log.Error("Matrix expansion transaction failed for job %d (JobID: %s): %v", job.ID, job.JobID, err) + return nil, fmt.Errorf("matrix expansion transaction: %w", err) } - GetMatrixMetrics().RecordInsertTime(time.Since(insertStart)) - - // Mark the placeholder job as evaluated and skipped so it is never run - if err := markMatrixAsEvaluatedAndSkip(ctx, job, fmt.Sprintf("successfully created %d new jobs", len(newJobs))); err != nil { - // Non-fatal: new jobs are already inserted - log.Error("Failed to mark placeholder job %d as evaluated after creating %d new jobs: %v", job.ID, len(newJobs), err) + if !claimed { + log.Warn("Matrix placeholder job %d (JobID: %s) was already claimed by a concurrent process; skipping", + job.ID, job.JobID) + return nil, nil } - totalTime := time.Since(startTime) - GetMatrixMetrics().RecordReevaluation(totalTime, true, int64(len(newJobs))) - log.Info("Matrix re-evaluation complete for job %d (JobID: %s): %d new jobs in %dms", - job.ID, job.JobID, len(newJobs), totalTime.Milliseconds()) + log.Info("Matrix re-evaluation complete for job %d (JobID: %s): created %d new jobs", + job.ID, job.JobID, len(newJobs)) return newJobs, nil } -// mergeNeedsIntoVars converts task needs outputs into variables for expression evaluation +// mergeNeedsIntoVars converts task needs outputs into variables for expression evaluation. func mergeNeedsIntoVars(baseVars map[string]string, taskNeeds map[string]*TaskNeed) map[string]string { merged := make(map[string]string) - - // Copy base vars maps.Copy(merged, baseVars) - - // Add needs outputs as variables in format: needs..outputs. for jobID, taskNeed := range taskNeeds { for outputKey, outputValue := range taskNeed.Outputs { key := fmt.Sprintf("needs.%s.outputs.%s", jobID, outputKey) merged[key] = outputValue } } - return merged } // constructWorkflowWithNeeds creates a workflow YAML that includes the target job -// and stub definitions for its dependencies so the jobparser can resolve needs.*.outputs expressions +// and stub definitions for its dependencies so the jobparser can resolve needs.*.outputs expressions. func constructWorkflowWithNeeds(job *actions_model.ActionRunJob, taskNeeds map[string]*TaskNeed) ([]byte, error) { - // Parse the original job's workflow payload to get the job definition var jobWorkflow map[string]any if err := yaml.Unmarshal(job.WorkflowPayload, &jobWorkflow); err != nil { return nil, fmt.Errorf("unmarshal job workflow: %w", err) } - // Extract the job definition from the parsed workflow jobsSection, ok := jobWorkflow["jobs"].(map[string]any) if !ok { return nil, errors.New("invalid jobs section in workflow") } - // Create a new workflow with the target job and stub jobs for dependencies newJobs := make(map[string]any) - // Add stub jobs for each dependency with their outputs for needJobID, taskNeed := range taskNeeds { stubJob := map[string]any{ "runs-on": "ubuntu-latest", @@ -347,7 +387,6 @@ func constructWorkflowWithNeeds(job *actions_model.ActionRunJob, taskNeeds map[s newJobs[needJobID] = stubJob } - // Add the actual job we want to expand (with matrix and needs) maps.Copy(newJobs, jobsSection) // The WorkflowPayload may contain a normalised/wrapped matrix (e.g. @@ -361,8 +400,6 @@ func constructWorkflowWithNeeds(job *actions_model.ActionRunJob, taskNeeds map[s if targetJobDef, ok := newJobs[job.JobID]; ok { if targetJobMap, ok := targetJobDef.(map[string]any); ok { delete(targetJobMap, "name") - // Restore needs from taskNeeds keys so the expression evaluator - // can resolve needs..outputs.* references. needsKeys := make([]string, 0, len(taskNeeds)) for needJobID := range taskNeeds { needsKeys = append(needsKeys, needJobID) @@ -379,7 +416,6 @@ func constructWorkflowWithNeeds(job *actions_model.ActionRunJob, taskNeeds map[s } } - // Construct the full workflow workflow := map[string]any{ "name": "matrix-expansion", "on": "push", diff --git a/services/actions/matrix_cache.go b/services/actions/matrix_cache.go deleted file mode 100644 index c1bb0f0a61..0000000000 --- a/services/actions/matrix_cache.go +++ /dev/null @@ -1,94 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package actions - -import ( - "crypto/sha256" - "encoding/hex" - "errors" - "sync" - - lru "github.com/hashicorp/golang-lru/v2" -) - -// WorkflowParseCache caches parsed workflow results to avoid redundant YAML parsing -// This is especially useful for matrix re-evaluation where the same workflow might be -// parsed multiple times with the same job outputs -type WorkflowParseCache struct { - cache *lru.TwoQueueCache[string, []byte] // key -> marshaled workflow bytes - mu sync.RWMutex -} - -var ( - workflowParseCache *WorkflowParseCache - workflowParseCacheOnce sync.Once -) - -// getWorkflowParseCache returns the singleton workflow parse cache instance -func getWorkflowParseCache() *WorkflowParseCache { - workflowParseCacheOnce.Do(func() { - // Cache up to 1000 workflow parses - // 2Q cache is more efficient than simple LRU for this use case - cache, err := lru.New2Q[string, []byte](1000) - if err != nil { - // Fallback to no caching if cache creation fails - workflowParseCache = nil - return - } - workflowParseCache = &WorkflowParseCache{ - cache: cache, - } - }) - return workflowParseCache -} - -// computeCacheKey generates a cache key for a workflow parse operation -func computeCacheKey(workflowYAML []byte, taskNeeds map[string]*TaskNeed) string { - // Create a deterministic hash of workflow + all job outputs - h := sha256.New() - h.Write(workflowYAML) - - // Add outputs in deterministic order (sorted by job ID) - for jobID, need := range taskNeeds { - h.Write([]byte(jobID)) - if need.Outputs != nil { - for k, v := range need.Outputs { - h.Write([]byte(k)) - h.Write([]byte(v)) - } - } - } - - return hex.EncodeToString(h.Sum(nil)) -} - -// Get retrieves a cached workflow parse result -func (c *WorkflowParseCache) Get(key string) ([]byte, bool) { - if c == nil || c.cache == nil { - return nil, false - } - c.mu.RLock() - defer c.mu.RUnlock() - return c.cache.Get(key) -} - -// Set stores a workflow parse result in the cache -func (c *WorkflowParseCache) Set(key string, value []byte) { - if c == nil || c.cache == nil { - return - } - c.mu.Lock() - defer c.mu.Unlock() - c.cache.Add(key, value) -} - -// Stats returns cache statistics for monitoring -func (c *WorkflowParseCache) Stats() (size int, err error) { - if c == nil || c.cache == nil { - return 0, errors.New("cache not initialized") - } - c.mu.RLock() - defer c.mu.RUnlock() - return c.cache.Len(), nil -} diff --git a/services/actions/matrix_metrics.go b/services/actions/matrix_metrics.go deleted file mode 100644 index 03586f4881..0000000000 --- a/services/actions/matrix_metrics.go +++ /dev/null @@ -1,181 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package actions - -import ( - "sync" - "time" -) - -// MatrixMetrics tracks performance metrics for matrix re-evaluation operations -type MatrixMetrics struct { - mu sync.RWMutex - - // Counters - TotalReevaluations int64 - SuccessfulReevaluations int64 - FailedReevaluations int64 - JobsCreatedTotal int64 - DeferredReevaluations int64 - CacheHits int64 - CacheMisses int64 - - // Timing - TotalReevaluationTime time.Duration - TotalParseTime time.Duration - TotalInsertTime time.Duration - - // Histograms (for detailed analysis) - ReevaluationTimes []time.Duration - ParseTimes []time.Duration - InsertTimes []time.Duration -} - -var ( - matrixMetricsInstance *MatrixMetrics - metricsOnce sync.Once -) - -// GetMatrixMetrics returns the global matrix metrics instance -func GetMatrixMetrics() *MatrixMetrics { - metricsOnce.Do(func() { - matrixMetricsInstance = &MatrixMetrics{ - ReevaluationTimes: make([]time.Duration, 0, 1000), - ParseTimes: make([]time.Duration, 0, 1000), - InsertTimes: make([]time.Duration, 0, 1000), - } - }) - return matrixMetricsInstance -} - -// appendToHistogram appends a duration to a histogram with rolling window (keep last 1000) -func appendToHistogram(histogram *[]time.Duration, duration time.Duration) { - if len(*histogram) < 1000 { - *histogram = append(*histogram, duration) - } else { - // Shift and add new value - copy(*histogram, (*histogram)[1:]) - (*histogram)[len(*histogram)-1] = duration - } -} - -// RecordReevaluation records a matrix re-evaluation attempt -func (m *MatrixMetrics) RecordReevaluation(duration time.Duration, success bool, jobsCreated int64) { - m.mu.Lock() - defer m.mu.Unlock() - - m.TotalReevaluations++ - m.TotalReevaluationTime += duration - - if success { - m.SuccessfulReevaluations++ - m.JobsCreatedTotal += jobsCreated - } else { - m.FailedReevaluations++ - } - - appendToHistogram(&m.ReevaluationTimes, duration) -} - -// RecordDeferred records a deferred matrix re-evaluation -func (m *MatrixMetrics) RecordDeferred() { - m.mu.Lock() - defer m.mu.Unlock() - m.DeferredReevaluations++ -} - -// RecordParseTime records the time taken to parse a workflow -func (m *MatrixMetrics) RecordParseTime(duration time.Duration) { - m.mu.Lock() - defer m.mu.Unlock() - - m.TotalParseTime += duration - appendToHistogram(&m.ParseTimes, duration) -} - -// RecordInsertTime records the time taken to insert matrix jobs -func (m *MatrixMetrics) RecordInsertTime(duration time.Duration) { - m.mu.Lock() - defer m.mu.Unlock() - - m.TotalInsertTime += duration - appendToHistogram(&m.InsertTimes, duration) -} - -// RecordCacheHit records a cache hit for workflow parsing -func (m *MatrixMetrics) RecordCacheHit() { - m.mu.Lock() - defer m.mu.Unlock() - m.CacheHits++ -} - -// RecordCacheMiss records a cache miss for workflow parsing -func (m *MatrixMetrics) RecordCacheMiss() { - m.mu.Lock() - defer m.mu.Unlock() - m.CacheMisses++ -} - -// GetStats returns a snapshot of the current metrics -func (m *MatrixMetrics) GetStats() map[string]any { - m.mu.RLock() - defer m.mu.RUnlock() - - avgReevaluationTime := time.Duration(0) - if m.TotalReevaluations > 0 { - avgReevaluationTime = m.TotalReevaluationTime / time.Duration(m.TotalReevaluations) - } - - avgParseTime := time.Duration(0) - if len(m.ParseTimes) > 0 { - avgParseTime = m.TotalParseTime / time.Duration(len(m.ParseTimes)) - } - - avgInsertTime := time.Duration(0) - if len(m.InsertTimes) > 0 { - avgInsertTime = m.TotalInsertTime / time.Duration(len(m.InsertTimes)) - } - - successRate := 0.0 - if m.TotalReevaluations > 0 { - successRate = float64(m.SuccessfulReevaluations) / float64(m.TotalReevaluations) * 100 - } - - return map[string]any{ - "total_reevaluations": m.TotalReevaluations, - "successful_reevaluations": m.SuccessfulReevaluations, - "failed_reevaluations": m.FailedReevaluations, - "deferred_reevaluations": m.DeferredReevaluations, - "success_rate_percent": successRate, - "total_jobs_created": m.JobsCreatedTotal, - "total_reevaluation_time_ms": m.TotalReevaluationTime.Milliseconds(), - "avg_reevaluation_time_ms": avgReevaluationTime.Milliseconds(), - "total_parse_time_ms": m.TotalParseTime.Milliseconds(), - "avg_parse_time_ms": avgParseTime.Milliseconds(), - "total_insert_time_ms": m.TotalInsertTime.Milliseconds(), - "avg_insert_time_ms": avgInsertTime.Milliseconds(), - "total_cache_hits": m.CacheHits, - "total_cache_misses": m.CacheMisses, - } -} - -// Reset clears all metrics -func (m *MatrixMetrics) Reset() { - m.mu.Lock() - defer m.mu.Unlock() - - m.TotalReevaluations = 0 - m.SuccessfulReevaluations = 0 - m.FailedReevaluations = 0 - m.JobsCreatedTotal = 0 - m.DeferredReevaluations = 0 - m.CacheHits = 0 - m.CacheMisses = 0 - m.TotalReevaluationTime = 0 - m.TotalParseTime = 0 - m.TotalInsertTime = 0 - m.ReevaluationTimes = m.ReevaluationTimes[:0] - m.ParseTimes = m.ParseTimes[:0] - m.InsertTimes = m.InsertTimes[:0] -} diff --git a/services/actions/matrix_metrics_prometheus.go b/services/actions/matrix_metrics_prometheus.go deleted file mode 100644 index eb27db3bb8..0000000000 --- a/services/actions/matrix_metrics_prometheus.go +++ /dev/null @@ -1,119 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package actions - -import ( - "github.com/prometheus/client_golang/prometheus" -) - -// MatrixMetricsCollector implements the prometheus.Collector interface -// and exposes matrix re-evaluation metrics for prometheus -type MatrixMetricsCollector struct { - // Counters - totalReevaluations prometheus.Gauge - successfulReevaluations prometheus.Gauge - failedReevaluations prometheus.Gauge - deferredReevaluations prometheus.Gauge - jobsCreatedTotal prometheus.Gauge - - // Timing (in milliseconds) - totalReevaluationTime prometheus.Gauge - avgReevaluationTime prometheus.Gauge - totalParseTime prometheus.Gauge - avgParseTime prometheus.Gauge - totalInsertTime prometheus.Gauge - avgInsertTime prometheus.Gauge - - // Rates - successRate prometheus.Gauge -} - -const ( - namespace = "gitea" - subsystem = "matrix" -) - -// newMatrixGauge creates a new Prometheus Gauge with standard matrix metrics naming -func newMatrixGauge(name, help string) prometheus.Gauge { - return prometheus.NewGauge( - prometheus.GaugeOpts{ - Namespace: namespace, - Subsystem: subsystem, - Name: name, - Help: help, - }, - ) -} - -// NewMatrixMetricsCollector creates a new MatrixMetricsCollector -func NewMatrixMetricsCollector() *MatrixMetricsCollector { - return &MatrixMetricsCollector{ - totalReevaluations: newMatrixGauge("total_reevaluations", "Total number of matrix re-evaluation attempts"), - successfulReevaluations: newMatrixGauge("successful_reevaluations", "Number of successful matrix re-evaluations"), - failedReevaluations: newMatrixGauge("failed_reevaluations", "Number of failed matrix re-evaluations"), - deferredReevaluations: newMatrixGauge("deferred_reevaluations", "Number of deferred matrix re-evaluations (waiting for dependencies)"), - jobsCreatedTotal: newMatrixGauge("jobs_created_total", "Total number of jobs created from matrix expansion"), - totalReevaluationTime: newMatrixGauge("total_reevaluation_time_ms", "Total time spent on matrix re-evaluations in milliseconds"), - avgReevaluationTime: newMatrixGauge("avg_reevaluation_time_ms", "Average time per matrix re-evaluation in milliseconds"), - totalParseTime: newMatrixGauge("total_parse_time_ms", "Total time spent parsing workflow payloads in milliseconds"), - avgParseTime: newMatrixGauge("avg_parse_time_ms", "Average time per workflow parse in milliseconds"), - totalInsertTime: newMatrixGauge("total_insert_time_ms", "Total time spent inserting jobs into database in milliseconds"), - avgInsertTime: newMatrixGauge("avg_insert_time_ms", "Average time per database insert in milliseconds"), - successRate: newMatrixGauge("success_rate_percent", "Success rate of matrix re-evaluations as percentage (0-100)"), - } -} - -// Describe returns the metrics descriptions -func (c *MatrixMetricsCollector) Describe(ch chan<- *prometheus.Desc) { - c.totalReevaluations.Describe(ch) - c.successfulReevaluations.Describe(ch) - c.failedReevaluations.Describe(ch) - c.deferredReevaluations.Describe(ch) - c.jobsCreatedTotal.Describe(ch) - c.totalReevaluationTime.Describe(ch) - c.avgReevaluationTime.Describe(ch) - c.totalParseTime.Describe(ch) - c.avgParseTime.Describe(ch) - c.totalInsertTime.Describe(ch) - c.avgInsertTime.Describe(ch) - c.successRate.Describe(ch) -} - -// Collect collects the current metric values and sends them to the channel -func (c *MatrixMetricsCollector) Collect(ch chan<- prometheus.Metric) { - metrics := GetMatrixMetrics() - stats := metrics.GetStats() - - // Set counter values - c.totalReevaluations.Set(float64(stats["total_reevaluations"].(int64))) - c.successfulReevaluations.Set(float64(stats["successful_reevaluations"].(int64))) - c.failedReevaluations.Set(float64(stats["failed_reevaluations"].(int64))) - c.deferredReevaluations.Set(float64(stats["deferred_reevaluations"].(int64))) - c.jobsCreatedTotal.Set(float64(stats["total_jobs_created"].(int64))) - - // Set timing values (already in milliseconds) - c.totalReevaluationTime.Set(float64(stats["total_reevaluation_time_ms"].(int64))) - c.avgReevaluationTime.Set(float64(stats["avg_reevaluation_time_ms"].(int64))) - c.totalParseTime.Set(float64(stats["total_parse_time_ms"].(int64))) - c.avgParseTime.Set(float64(stats["avg_parse_time_ms"].(int64))) - c.totalInsertTime.Set(float64(stats["total_insert_time_ms"].(int64))) - c.avgInsertTime.Set(float64(stats["avg_insert_time_ms"].(int64))) - - // Set success rate - c.successRate.Set(stats["success_rate_percent"].(float64)) - - // Collect all metrics - c.totalReevaluations.Collect(ch) - c.successfulReevaluations.Collect(ch) - c.failedReevaluations.Collect(ch) - c.deferredReevaluations.Collect(ch) - c.jobsCreatedTotal.Collect(ch) - c.totalReevaluationTime.Collect(ch) - c.avgReevaluationTime.Collect(ch) - c.totalParseTime.Collect(ch) - c.avgParseTime.Collect(ch) - c.totalInsertTime.Collect(ch) - c.avgInsertTime.Collect(ch) - c.successRate.Collect(ch) -} diff --git a/services/actions/matrix_metrics_test.go b/services/actions/matrix_metrics_test.go deleted file mode 100644 index db81000414..0000000000 --- a/services/actions/matrix_metrics_test.go +++ /dev/null @@ -1,88 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package actions - -import ( - "testing" - "time" - - "github.com/prometheus/client_golang/prometheus" - "github.com/stretchr/testify/assert" -) - -// Essential Prometheus Collector Tests - -func TestNewMatrixMetricsCollector(t *testing.T) { - collector := NewMatrixMetricsCollector() - assert.NotNil(t, collector) - assert.NotNil(t, collector.totalReevaluations) - assert.NotNil(t, collector.successRate) -} - -func TestMatrixMetricsCollectorDescribe(t *testing.T) { - collector := NewMatrixMetricsCollector() - ch := make(chan *prometheus.Desc, 100) - collector.Describe(ch) - assert.NotEmpty(t, ch) -} - -func TestMatrixMetricsCollectorCollect(t *testing.T) { - matrixMetricsInstance = nil - metrics := GetMatrixMetrics() - metrics.RecordReevaluation(10*time.Millisecond, true, 5) - metrics.RecordParseTime(8 * time.Millisecond) - - collector := NewMatrixMetricsCollector() - ch := make(chan prometheus.Metric, 100) - collector.Collect(ch) - assert.NotEmpty(t, ch) - - matrixMetricsInstance = nil -} - -func TestMatrixMetricsGetStats(t *testing.T) { - metrics := &MatrixMetrics{ - ReevaluationTimes: make([]time.Duration, 0, 1000), - ParseTimes: make([]time.Duration, 0, 1000), - InsertTimes: make([]time.Duration, 0, 1000), - } - - metrics.RecordReevaluation(10*time.Millisecond, true, 3) - metrics.RecordReevaluation(15*time.Millisecond, true, 2) - metrics.RecordReevaluation(5*time.Millisecond, false, 0) - - stats := metrics.GetStats() - assert.Equal(t, int64(3), stats["total_reevaluations"]) - assert.Equal(t, int64(2), stats["successful_reevaluations"]) - assert.Equal(t, int64(1), stats["failed_reevaluations"]) - assert.Greater(t, stats["success_rate_percent"].(float64), 60.0) -} - -func BenchmarkMatrixMetricsCollectorCollect(b *testing.B) { - metrics := &MatrixMetrics{ - ReevaluationTimes: make([]time.Duration, 0, 1000), - ParseTimes: make([]time.Duration, 0, 1000), - InsertTimes: make([]time.Duration, 0, 1000), - } - matrixMetricsInstance = metrics - - for range 100 { - metrics.RecordReevaluation(10*time.Millisecond, true, 5) - metrics.RecordParseTime(5 * time.Millisecond) - } - - collector := NewMatrixMetricsCollector() - - b.ResetTimer() - for b.Loop() { - // Use a fresh channel each iteration to avoid filling up the buffer - ch := make(chan prometheus.Metric, 100) - collector.Collect(ch) - // Drain the channel to prevent blocking - close(ch) - for range ch { - // discard metrics - } - } -} diff --git a/services/actions/matrix_test.go b/services/actions/matrix_test.go new file mode 100644 index 0000000000..5d8c9a7fc2 --- /dev/null +++ b/services/actions/matrix_test.go @@ -0,0 +1,77 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package actions + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestHasMatrixWithNeeds(t *testing.T) { + tests := []struct { + name string + strategy string + want bool + }{ + { + name: "dynamic matrix referencing job output", + strategy: ` +matrix: + version: ${{ fromJson(needs.generate.outputs.matrix) }} +`, + want: true, + }, + { + name: "static matrix — no expression", + strategy: ` +matrix: + os: [ubuntu-latest, windows-latest] +`, + want: false, + }, + { + name: "value contains needs. but not inside expression", + strategy: ` +matrix: + os: [needs.review-runner] +`, + want: false, + }, + { + name: "needs. outside expression block", + strategy: ` +matrix: + runner: needs.something-but-no-braces +`, + want: false, + }, + { + name: "expression with needs but no .outputs.", + strategy: ` +matrix: + version: ${{ needs.job1 }} +`, + want: false, + }, + { + name: "empty strategy", + strategy: "", + want: false, + }, + { + name: "strategy without matrix key", + strategy: ` +fail-fast: false +`, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, HasMatrixWithNeeds(tt.strategy)) + }) + } +}