From 00ea22848f9c184ab10e40adfc1902dfcfabccf4 Mon Sep 17 00:00:00 2001 From: dawn Date: Fri, 23 Oct 2026 03:15:07 +0300 Subject: [PATCH] spindle/engine: count refused workflows in the metrics for remote engines too Signed-off-by: dawn --- spindle/engine/engine.go | 41 +++++++++----- spindle/engine/engine_minutes_test.go | 82 ++++++++++++++++++++++++++- 2 files changed, 107 insertions(+), 16 deletions(-) diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go index 2f65c160f..5ca60efd6 100644 --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -517,16 +517,15 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q } }() + quotaIdentity := quota.Identity{OwnerDID: w.OwnerDID, RepoDID: repoDID} + quotaID := quota.WorkflowReservationID(w.RunID, quotaIdentity.OwnerDID, quotaIdentity.RepoDID, string(wid.PipelineId), wid.Name) + if qm != nil && hasQuotaReporter { - resID := quota.WorkflowReservationID(w.RunID, w.OwnerDID, w.RepoDID, string(wid.PipelineId), wid.Name) req := quota.ReserveRequest{ - ID: resID, - Kind: quota.KindWorkflow, - Key: resID, - Identity: quota.Identity{ - OwnerDID: w.OwnerDID, - RepoDID: w.RepoDID, - }, + ID: quotaID, + Kind: quota.KindWorkflow, + Key: quotaID, + Identity: quotaIdentity, Resources: requestedResources, } quotaLease, err = qm.Acquire(wfCtx, req) @@ -552,13 +551,17 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q } // after the slot, or every queued run could pass this before any of them bills - minutesIdentity := quota.Identity{OwnerDID: w.OwnerDID, RepoDID: repoDID} - minutesBilled := qm != nil && minutesIdentity.Valid() + minutesBilled := qm != nil && quotaIdentity.Valid() + if qm != nil && !minutesBilled { + wl.WarnContext(wfCtx, "workflow has no valid owner and repo identity, so its ci minutes are neither checked nor billed") + } if minutesBilled { - if err = qm.CheckMinutes(wfCtx, minutesIdentity); err != nil { + if err = qm.CheckMinutes(wfCtx, quotaIdentity); err != nil { if errors.Is(err, quota.ErrDenied) { err = ClassifiedFailure(FailureClassPolicy, FailureReasonQuotaDenied, err) } + // nothing ran remotely, so the terminal metric is still ours to record + remoteExecutionStarted = false setTerminalError("checking ci minutes", err) return } @@ -620,10 +623,9 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q // a spindle crash bills nothing, the defer never runs if minutesBilled { - usageID := quota.WorkflowReservationID(w.RunID, w.OwnerDID, w.RepoDID, string(wid.PipelineId), wid.Name) defer func() { minutes := quota.BillableMinutes(runStart, time.Now()) - if err := qm.RecordMinutes(context.WithoutCancel(wfCtx), usageID, minutesIdentity, minutes); err != nil { + if err := recordMinutes(context.WithoutCancel(wfCtx), qm, quotaID, quotaIdentity, minutes); err != nil { wl.ErrorContext(wfCtx, "failed to record ci minutes", "wid", wid, "minutes", minutes, "err", err) } }() @@ -788,6 +790,19 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q l.Info("all workflows completed") } +const recordMinutesAttempts = 3 + +// recording is idempotent by usage id, so a failed attempt can simply be tried again +func recordMinutes(ctx context.Context, qm *quota.Manager, usageID string, id quota.Identity, minutes int64) (err error) { + for attempt := 1; attempt <= recordMinutesAttempts; attempt++ { + if err = qm.RecordMinutes(ctx, usageID, id, minutes); err == nil { + return nil + } + time.Sleep(time.Duration(attempt) * 100 * time.Millisecond) + } + return err +} + type logStores interface { Names() []string PutFile(ctx context.Context, ref, sourcePath string) []error diff --git a/spindle/engine/engine_minutes_test.go b/spindle/engine/engine_minutes_test.go index 7cc66c870..275023f8f 100644 --- a/spindle/engine/engine_minutes_test.go +++ b/spindle/engine/engine_minutes_test.go @@ -2,6 +2,7 @@ package engine import ( "context" + "errors" "io" "log/slog" "strings" @@ -30,7 +31,7 @@ func minutesQuotaManager(t *testing.T, database *db.DB, userLimit int64) (*quota return qm, qs } -func runMinutesWorkflow(t *testing.T, qm *quota.Manager, database *db.DB, ctx context.Context, eng *mockEngine, name string) models.WorkflowId { +func runMinutesWorkflow(t *testing.T, qm *quota.Manager, database *db.DB, ctx context.Context, eng models.Engine, name string) models.WorkflowId { t.Helper() logger := slog.New(slog.NewTextHandler(io.Discard, nil)) cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} @@ -89,6 +90,13 @@ func TestStartWorkflows_DeniedWhenMonthlyMinutesAreUsedUp(t *testing.T) { t.Fatalf("status message = %q", msg) } + if !quotaDeniedRecorded(t, metrics) { + t.Fatal("denial was not attributed to quota_denied") + } +} + +func quotaDeniedRecorded(t *testing.T, metrics *observability.Metrics) bool { + t.Helper() families, err := metrics.Registry().Gather() if err != nil { t.Fatal(err) @@ -103,11 +111,44 @@ func TestStartWorkflows_DeniedWhenMonthlyMinutesAreUsedUp(t *testing.T) { labels[label.GetName()] = label.GetValue() } if labels["failure_class"] == string(FailureClassPolicy) && labels["reason"] == string(FailureReasonQuotaDenied) && metric.GetCounter().GetValue() == 1 { - return + return true } } } - t.Fatal("denial was not attributed to quota_denied") + return false +} + +type remoteSlotEngine struct { + *mockEngine + released chan struct{} +} + +func (*remoteSlotEngine) AuthorsRemoteStatus() {} + +func (e *remoteSlotEngine) AcquireWorkflowSlot(context.Context, models.WorkflowId, *models.Workflow, AcquireMode) (WorkflowSlot, error) { + return releaseFunc(func() { close(e.released) }), nil +} + +func TestStartWorkflows_RemoteEngineDenialStillRecordsTheTerminalMetric(t *testing.T) { + database := newTestDB(t) + qm, qs := minutesQuotaManager(t, database, 3) + if err := qs.RecordMinutes(context.Background(), "earlier", quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo}, 3); err != nil { + t.Fatal(err) + } + + metrics := observability.NewMetrics() + ctx := observability.WithMetrics(context.Background(), metrics) + eng := &remoteSlotEngine{mockEngine: &mockEngine{}, released: make(chan struct{})} + runMinutesWorkflow(t, qm, database, ctx, eng, "job1") + + select { + case <-eng.released: + default: + t.Fatal("the remote slot was not released after the denial") + } + if !quotaDeniedRecorded(t, metrics) { + t.Fatal("a denial after the remote slot was taken recorded no quota_denied termination") + } } func TestStartWorkflows_RecordsMinutesWhenAWorkflowFinishes(t *testing.T) { @@ -140,3 +181,38 @@ func TestStartWorkflows_RecordsMinutesWhenAWorkflowFinishes(t *testing.T) { t.Fatalf("minutes used = %v, want 2 on both the owner and the repo", got) } } + +type flakyMinutesStore struct { + spyStore + failures int + recorded []string +} + +func (s *flakyMinutesStore) CheckMinutes(context.Context, quota.Identity) error { return nil } + +func (s *flakyMinutesStore) RecordMinutes(_ context.Context, usageID string, _ quota.Identity, _ int64) error { + if s.failures > 0 { + s.failures-- + return errors.New("database is locked") + } + s.recorded = append(s.recorded, usageID) + return nil +} + +func TestRecordMinutesRetriesAFailedRecord(t *testing.T) { + id := quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo} + + store := &flakyMinutesStore{failures: recordMinutesAttempts - 1} + qm := quota.NewManager(store, 0, nil) + t.Cleanup(qm.Close) + if err := recordMinutes(context.Background(), qm, "wf", id, 1); err != nil || len(store.recorded) != 1 { + t.Fatalf("recordMinutes = %v, recorded %v, want it to succeed on the last attempt", err, store.recorded) + } + + store = &flakyMinutesStore{failures: recordMinutesAttempts} + qm = quota.NewManager(store, 0, nil) + t.Cleanup(qm.Close) + if err := recordMinutes(context.Background(), qm, "wf", id, 1); err == nil || len(store.recorded) != 0 { + t.Fatalf("recordMinutes = %v, recorded %v, want the error once the attempts run out", err, store.recorded) + } +} -- 2.54.0