Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987package engine
import ( "context" "encoding/json" "errors" "log/slog" "math" "os" "path/filepath" "reflect" "slices" "strings" "sync" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "go.opentelemetry.io/otel" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" "tangled.org/core/api/tangled" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/observability" "tangled.org/core/spindle/quota" "tangled.org/core/spindle/secrets")
type mockStep struct { name string command string}
func (m mockStep) Name() string { return m.name }func (m mockStep) Command() string { return m.command }func (m mockStep) Kind() models.StepKind { return models.StepKindUser }
type emptyVault struct{}
func (emptyVault) AddSecret(context.Context, secrets.UnlockedSecret) error { return nil}
func (emptyVault) RemoveSecret(context.Context, secrets.Secret[any]) error { return nil}
func (emptyVault) RemoveAllSecrets(context.Context, secrets.RepoIdentifier) error { return nil}
func (emptyVault) GetSecretsLocked(context.Context, secrets.RepoIdentifier) ([]secrets.LockedSecret, error) { return nil, nil}
func (emptyVault) GetSecretsUnlocked(context.Context, secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) { return nil, nil}
type staticCacheController struct { bindings []models.CacheBinding}
func (c staticCacheController) Plan(context.Context, *models.Pipeline, *models.Workflow) ([]models.CacheBinding, error) { return c.bindings, nil}
func (staticCacheController) Apply(context.Context, CacheUpdate) error { return nil}
type mockEngine struct { mu sync.Mutex setupCalls []models.WorkflowId runStepCalls []models.WorkflowId runStepSecrets [][]secrets.UnlockedSecret setupFunc func(ctx context.Context, wid models.WorkflowId) error runStepFunc func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error destroyFunc func(ctx context.Context, wid models.WorkflowId) error runWorkflowFunc func(*models.Workflow) timeout time.Duration}
func (m *mockEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { owner := "did:web:alice" repo := "did:web:alice/repo" if md := tpl.TriggerMetadata; md != nil && md.Repo != nil { owner = md.Repo.Did if md.Repo.RepoDid != nil { repo = *md.Repo.RepoDid } } return &models.Workflow{ OwnerDID: owner, RepoDID: repo, }, nil}
func (m *mockEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error { m.mu.Lock() m.setupCalls = append(m.setupCalls, wid) fn := m.setupFunc m.mu.Unlock() if fn != nil { return fn(ctx, wid) } return nil}
func (m *mockEngine) WorkflowTimeout() time.Duration { if m.timeout != 0 { return m.timeout } return 5 * time.Second}
func (m *mockEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { if m.destroyFunc != nil { return m.destroyFunc(ctx, wid) } return nil}
func (m *mockEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { m.mu.Lock() m.runStepCalls = append(m.runStepCalls, wid) m.runStepSecrets = append(m.runStepSecrets, slices.Clone(secrets)) runStepFn := m.runStepFunc runWorkflowFn := m.runWorkflowFunc m.mu.Unlock()
if runWorkflowFn != nil { runWorkflowFn(w) } if runStepFn != nil { return runStepFn(ctx, wid, idx, wfLogger) } return nil}
func newTestDB(t *testing.T) *db.DB { t.Helper() d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) if err != nil { t.Fatalf("failed to create test db: %v", err) } return d}
func TestStartWorkflows_CollisionRejection(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
eng := &mockEngine{} pipelineId := models.PipelineId("test-rkey")
// two names that normalize to the same wid must not both run wfColliding1 := models.Workflow{ Name: "test-job", Steps: []models.Step{mockStep{name: "step1"}}, } wfColliding2 := models.Workflow{ Name: "test job", Steps: []models.Step{mockStep{name: "step1"}}, } wfUnique := models.Workflow{ Name: "unique_job", Steps: []models.Step{mockStep{name: "step1"}}, }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {wfColliding1, wfColliding2, wfUnique}, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
eng.mu.Lock() setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...) eng.mu.Unlock()
for _, call := range setupCalls { if call.Name == "test-job" || call.Name == "test job" { t.Fatalf("expected colliding workflow %s to not be started", call.Name) } }
hasUnique := false for _, call := range setupCalls { if call.Name == "unique_job" { hasUnique = true } } if !hasUnique { t.Fatalf("expected unique workflow unique_job to be started") }
widColliding1 := models.WorkflowId{PipelineId: pipelineId, Name: "test-job"} widColliding2 := models.WorkflowId{PipelineId: pipelineId, Name: "test job"} widUnique := models.WorkflowId{PipelineId: pipelineId, Name: "unique_job"}
status1, err := testDB.GetStatus(widColliding1) if err != nil || status1 != models.StatusKindFailed { t.Fatalf("expected colliding1 status to be failed, got status=%v err=%v", status1, err) }
status2, err := testDB.GetStatus(widColliding2) if err != nil || status2 != models.StatusKindFailed { t.Fatalf("expected colliding2 status to be failed, got status=%v err=%v", status2, err) }
statusUnique, err := testDB.GetStatus(widUnique) if err != nil || statusUnique != models.StatusKindSuccess { t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err) }}
func TestCancelWorkflow_NotOverwritten(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
stepStarted := make(chan struct{}) eng := &mockEngine{ runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error { close(stepStarted) <-ctx.Done() return ctx.Err() }, }
pipelineId := models.PipelineId("test-rkey")
wid := models.WorkflowId{ PipelineId: pipelineId, Name: "cancel_test_job", }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "cancel_test_job", Steps: []models.Step{mockStep{name: "step1"}}, }, }, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} doneChan := make(chan struct{}) go func() { StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId) close(doneChan) }()
select { case <-stepStarted: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for step to start") }
_ = testDB.StatusCancelled(wid, "User canceled the workflow", -1, nil) CancelWorkflow(wid)
select { case <-doneChan: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for StartWorkflows to complete") }
// the runner writes StatusCancelled itself when it sees the canceled ctx // the handler writes nothing for a live wf, so nothing lands after to overwrite it st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st != models.StatusKindCancelled { t.Fatalf("expected status to be cancelled, got %s", st) }}
func TestStartWorkflowsScopesSecretsToTrustedPipelineRepo(t *testing.T) { vault, err := secrets.NewSQLiteManager(":memory:") if err != nil { t.Fatal(err) } repoA := syntax.DID("did:plc:workflowrepoa") repoB := syntax.DID("did:plc:workflowrepob") for _, secret := range []secrets.UnlockedSecret{ {Repo: secrets.RepoIdentifier(repoA), Key: "A_SECRET", Value: "alpha"}, {Repo: secrets.RepoIdentifier(repoB), Key: "B_SECRET", Value: "bravo"}, } { if err := vault.AddSecret(context.Background(), secret); err != nil { t.Fatal(err) } } run := func(trusted bool) [][]secrets.UnlockedSecret { t.Helper() database := newTestDB(t) defer database.Close() engine := &mockEngine{} pipeline := &models.Pipeline{ RepoDid: repoA, TrustedSource: trusted, Workflows: map[models.Engine][]models.Workflow{ engine: {{Name: "ci.yml", Steps: []models.Step{mockStep{name: "test"}}}}, }, } StartWorkflows(slog.Default(), vault, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, database, nil, nil, nil, context.Background(), pipeline, models.PipelineId("3mu2xwiorc2xl")) engine.mu.Lock() defer engine.mu.Unlock() return slices.Clone(engine.runStepSecrets) }
trustedSecrets := run(true) if len(trustedSecrets) != 1 || len(trustedSecrets[0]) != 1 || trustedSecrets[0][0].Key != "A_SECRET" || trustedSecrets[0][0].Value != "alpha" { t.Fatalf("trusted secrets = %#v", trustedSecrets) } if untrustedSecrets := run(false); len(untrustedSecrets) != 1 || len(untrustedSecrets[0]) != 0 { t.Fatalf("untrusted secrets = %#v", untrustedSecrets) }}
func TestStartWorkflows_FlushesLogBeforeTerminalStatus(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) logDir := t.TempDir() vault, err := secrets.NewSQLiteManager(filepath.Join(t.TempDir(), "secrets.db")) if err != nil { t.Fatal(err) }
repoDid := syntax.DID("did:plc:test") if err := vault.AddSecret(context.Background(), secrets.UnlockedSecret{ Key: "LONG_SECRET", Value: strings.Repeat("s", 256), Repo: secrets.RepoIdentifier(repoDid.String()), }); err != nil { t.Fatal(err) }
destroyStarted := make(chan struct{}) releaseDestroy := make(chan struct{}) var releaseOnce sync.Once release := func() { releaseOnce.Do(func() { close(releaseDestroy) }) } t.Cleanup(release)
eng := &mockEngine{ runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error { if _, err := wfLogger.DataWriter(idx, "stdout").Write([]byte("ssh debug hint\n")); err != nil { return err } return errors.New("step failed") }, destroyFunc: func(ctx context.Context, wid models.WorkflowId) error { close(destroyStarted) <-releaseDestroy return nil }, }
pipelineId := models.PipelineId("test-rkey") wid := models.WorkflowId{PipelineId: pipelineId, Name: "failed_job"} pipeline := &models.Pipeline{ RepoDid: repoDid, TrustedSource: true, Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: wid.Name, Steps: []models.Step{mockStep{name: "step1"}}}}, }, } cfg := &config.Config{Server: config.Server{LogDir: logDir}}
done := make(chan struct{}) go func() { StartWorkflows(logger, vault, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId) close(done) }()
select { case <-destroyStarted: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for workflow destruction") }
status, err := testDB.GetStatus(wid) if err != nil { t.Fatal(err) } if status != models.StatusKindFailed { t.Fatalf("expected failed status, got %s", status) }
log, err := os.ReadFile(models.LogFilePath(logDir, wid)) if err != nil { t.Fatal(err) } if !strings.Contains(string(log), "ssh debug hint") { t.Fatalf("terminal status was visible before the log was flushed: %s", log) }
release() select { case <-done: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for StartWorkflows to complete") }}
func TestStartWorkflows_EndLinesCarryStepOutcome(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) logDir := t.TempDir()
eng := &mockEngine{ runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error { if idx == 1 { hint := mockStep{name: "Debug shell"} wfLogger.ControlWriter(2, hint, models.StepStatusStart).Write([]byte{0}) wfLogger.ControlWriter(2, hint, models.StepStatusEnd).Write([]byte{0}) return errors.New("exited with code 1") } return nil }, }
pipelineId := models.PipelineId("outcome-rkey") wid := models.WorkflowId{PipelineId: pipelineId, Name: "outcome"} pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: wid.Name, Steps: []models.Step{mockStep{name: "build"}, mockStep{name: "test"}}}}, }, } cfg := &config.Config{Server: config.Server{LogDir: logDir}} StartWorkflows(logger, emptyVault{}, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
raw, err := os.ReadFile(models.LogFilePath(logDir, wid)) if err != nil { t.Fatal(err) } got := map[string]models.StepOutcome{} for _, encoded := range strings.Split(strings.TrimSpace(string(raw)), "\n") { var line models.LogLine if err := json.Unmarshal([]byte(encoded), &line); err != nil { t.Fatalf("decode log line %q: %v", encoded, err) } if line.Kind == models.LogKindControl && line.StepStatus == models.StepStatusEnd { got[line.Content] = line.StepOutcome } } want := map[string]models.StepOutcome{ "build": models.StepOutcomeSuccess, "test": models.StepOutcomeFailed, "Debug shell": "", } if !reflect.DeepEqual(got, want) { t.Fatalf("end line outcomes = %v, want %v", got, want) }}
func TestStartWorkflowsCapsTheWorkflowTimeBudget(t *testing.T) { deadline := make(chan time.Time, 1) eng := &mockEngine{ timeout: 24 * time.Hour, setupFunc: func(ctx context.Context, _ models.WorkflowId) error { got, ok := ctx.Deadline() if !ok { return errors.New("workflow context has no deadline") } deadline <- got return nil }, } pipeline := &models.Pipeline{Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: "bounded", Steps: []models.Step{mockStep{name: "step"}}}}, }}
StartWorkflows(slog.Default(), nil, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, newTestDB(t), nil, nil, nil, context.Background(), pipeline, "bounded")
remaining := time.Until(<-deadline) if remaining > models.MaxWorkflowTimeBudget || remaining < models.MaxWorkflowTimeBudget-time.Minute { t.Fatalf("workflow time budget = %s, want %s", remaining, models.MaxWorkflowTimeBudget) }}
func TestSetupTimeout_ReportsTimeout(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
// setup blocks past the workflow timeout, so it should land as timeout not failed eng := &mockEngine{ timeout: 100 * time.Millisecond, setupFunc: func(ctx context.Context, wid models.WorkflowId) error { <-ctx.Done() return ctx.Err() }, }
pipelineId := models.PipelineId("test-rkey") wid := models.WorkflowId{PipelineId: pipelineId, Name: "timeout_job"}
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: "timeout_job", Steps: []models.Step{mockStep{name: "step1"}}}}, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st != models.StatusKindTimeout { t.Fatalf("expected status to be timeout, got %s", st) }
if len(eng.runStepCalls) != 0 { t.Fatalf("expected no steps to run after setup timeout, got %d", len(eng.runStepCalls)) }}
func TestStartWorkflows_SpanEnrichment(t *testing.T) { oldProvider := otel.GetTracerProvider() recorder := tracetest.NewSpanRecorder() provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder)) otel.SetTracerProvider(provider) t.Cleanup(func() { _ = provider.Shutdown(context.Background()) otel.SetTracerProvider(oldProvider) })
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
eng := &mockEngine{} pipelineId := models.PipelineId("test-rkey")
pipeline := &models.Pipeline{ RepoDid: "did:web:test-repo", Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "test-workflow", Steps: []models.Step{ mockStep{name: "step-1"}, }, OwnerDID: "did:web:test-owner", RepoDID: "did:web:test-repo", }, }, }, TrustedSource: false, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
spans := recorder.Ended() var workflowRunSpan, stepRunSpan sdktrace.ReadOnlySpan for _, span := range spans { if span.Name() == "workflow.run" { workflowRunSpan = span } else if span.Name() == "step.run" { stepRunSpan = span } }
if workflowRunSpan == nil { t.Fatal("workflow.run span not recorded") } if stepRunSpan == nil { t.Fatal("step.run span not recorded") }
workflowAttrs := make(map[string]string) for _, attr := range workflowRunSpan.Attributes() { workflowAttrs[string(attr.Key)] = attr.Value.AsString() }
wid := models.WorkflowId{ PipelineId: pipelineId, Name: "test-workflow", }
expectedWfAttrs := map[string]string{ observability.OwnerDIDKey: "did:web:test-owner", observability.RepoDIDKey: "did:web:test-repo", observability.PipelineIDKey: pipelineId.String(), observability.WorkflowIDKey: wid.String(), } for k, wantVal := range expectedWfAttrs { if gotVal := workflowAttrs[k]; gotVal != wantVal { t.Errorf("workflow attribute %q = %q, want %q", k, gotVal, wantVal) } }
stepAttrs := make(map[string]string) for _, attr := range stepRunSpan.Attributes() { stepAttrs[string(attr.Key)] = attr.Value.AsString() }
expectedStepAttrs := map[string]string{ observability.OwnerDIDKey: "did:web:test-owner", observability.RepoDIDKey: "did:web:test-repo", observability.PipelineIDKey: pipelineId.String(), observability.WorkflowIDKey: wid.String(), observability.StepNameKey: "step-1", } for k, wantVal := range expectedStepAttrs { if gotVal := stepAttrs[k]; gotVal != wantVal { t.Errorf("step attribute %q = %q, want %q", k, gotVal, wantVal) } } hasStepIndex := false for _, attr := range stepRunSpan.Attributes() { if string(attr.Key) == observability.StepIndexKey && attr.Value.AsInterface() == int64(0) { hasStepIndex = true } } if !hasStepIndex { t.Error("step.run span missing step_index attribute or it is not 0") }}
type mockResourceUsageReporter struct { *mockEngine workflows, memory, vcpus, disk int64 usage WorkflowResourceUsage usageOk bool}
func (m *mockResourceUsageReporter) QuotaResources(wf *models.Workflow) quota.Resources { return quota.Resources{ quota.ResourceWorkflows: m.workflows, quota.ResourceMemoryMiB: m.memory, quota.ResourceVCPUs: m.vcpus, quota.ResourceDiskMiB: m.disk, }}
func (m *mockResourceUsageReporter) WorkflowResourceUsage(wf *models.Workflow) (WorkflowResourceUsage, bool) { return m.usage, m.usageOk}
func TestStartWorkflows_ResourceTracking(t *testing.T) { oldProvider := otel.GetTracerProvider() recorder := tracetest.NewSpanRecorder() provider := sdktrace.NewTracerProvider( sdktrace.WithSpanProcessor(recorder), sdktrace.WithSampler(sdktrace.AlwaysSample()), ) otel.SetTracerProvider(provider) t.Cleanup(func() { _ = provider.Shutdown(context.Background()) otel.SetTracerProvider(oldProvider) })
testDB := newTestDB(t)
mu := &sync.Mutex{} records := []slog.Record{} handler := &captureHandler{ mu: mu, records: &records, } logger := slog.New(handler)
eng := &mockResourceUsageReporter{ mockEngine: &mockEngine{}, workflows: 2, memory: 1024, vcpus: 4, disk: 512, usage: WorkflowResourceUsage{ CPUUsec: 12345, MemoryCurrentBytes: 2048, MemoryPeakBytes: 4096, SwapCurrentBytes: 100, SwapPeakBytes: 200, PIDsCurrent: 10, IOReadBytes: 300, IOWriteBytes: 400, IOReadOps: 30, IOWriteOps: 40, VolumeAllocatedBytes: 1000, CgroupAvailable: true, VolumeAvailable: true, }, usageOk: true, }
pipelineId := models.PipelineId("test-rkey")
pipeline := &models.Pipeline{ RepoDid: "did:web:test-repo", Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "test-workflow", Steps: []models.Step{ mockStep{name: "step-1"}, }, OwnerDID: "did:web:test-owner", RepoDID: "did:web:test-repo", }, }, }, TrustedSource: false, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
metrics := observability.NewMetrics() ctx := observability.WithMetrics(context.Background(), metrics)
StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, ctx, pipeline, pipelineId)
handler.mu.Lock() records = *handler.records handler.mu.Unlock()
var startedLog, finishedLog slog.Record for _, r := range records { if r.Message == "workflow started" { startedLog = r } else if r.Message == "workflow finished" { finishedLog = r } }
if startedLog.Message == "" { t.Fatal("workflow started log not found") } if finishedLog.Message == "" { t.Fatal("workflow finished log not found") }
getAttr := func(r slog.Record, key string) any { var val any r.Attrs(func(a slog.Attr) bool { if a.Key == key { val = a.Value.Any() return false } return true }) return val }
if getAttr(startedLog, "requested_workflows") != int64(2) { t.Errorf("started log requested_workflows = %v, want 2", getAttr(startedLog, "requested_workflows")) } if getAttr(startedLog, "requested_vcpus") != int64(4) { t.Errorf("started log requested_vcpus = %v, want 4", getAttr(startedLog, "requested_vcpus")) }
if getAttr(finishedLog, "result") != "success" { t.Errorf("finished log result = %v, want success", getAttr(finishedLog, "result")) } if getAttr(finishedLog, "actual_cpu_usec") != uint64(12345) { t.Errorf("finished log actual_cpu_usec = %v, want 12345", getAttr(finishedLog, "actual_cpu_usec")) } if getAttr(finishedLog, "actual_memory_peak_bytes") != uint64(4096) { t.Errorf("finished log actual_memory_peak_bytes = %v, want 4096", getAttr(finishedLog, "actual_memory_peak_bytes")) } if getAttr(finishedLog, "error") != "" || getAttr(finishedLog, "failure_class") != string(FailureClassNone) { t.Errorf("finished log error = %q, failure_class = %q, want no error", getAttr(finishedLog, "error"), getAttr(finishedLog, "failure_class")) } duration, _ := getAttr(finishedLog, "duration_seconds").(float64) waited, _ := getAttr(finishedLog, "wait_seconds").(float64) ran, _ := getAttr(finishedLog, "run_seconds").(float64) if waited < 0 || ran < 0 || math.Abs(waited+ran-duration) > 1e-6 { t.Errorf("finished log wait %v + run %v, want both >= 0 adding up to duration %v", waited, ran, duration) }
spans := recorder.Ended() var workflowRunSpan sdktrace.ReadOnlySpan for _, span := range spans { if span.Name() == "workflow.run" { workflowRunSpan = span break } }
if workflowRunSpan == nil { t.Fatal("workflow.run span not found") }
attrs := make(map[string]any) for _, attr := range workflowRunSpan.Attributes() { attrs[string(attr.Key)] = attr.Value.AsInterface() }
if attrs[observability.ReqWorkflowsKey] != int64(2) { t.Errorf("span attr %s = %v, want 2", observability.ReqWorkflowsKey, attrs[observability.ReqWorkflowsKey]) } if attrs[observability.ReqVCPUsKey] != int64(4) { t.Errorf("span attr %s = %v, want 4", observability.ReqVCPUsKey, attrs[observability.ReqVCPUsKey]) }
if attrs[observability.ActCPUUsecKey] != int64(12345) { t.Errorf("span attr %s = %v, want 12345", observability.ActCPUUsecKey, attrs[observability.ActCPUUsecKey]) } if attrs[observability.ActMemoryPeakBytesKey] != int64(4096) { t.Errorf("span attr %s = %v, want 4096", observability.ActMemoryPeakBytesKey, attrs[observability.ActMemoryPeakBytesKey]) }}
type captureHandler struct { attrs []slog.Attr records *[]slog.Record mu *sync.Mutex}
func (h *captureHandler) Enabled(ctx context.Context, level slog.Level) bool { return true}
func (h *captureHandler) Handle(ctx context.Context, r slog.Record) error { h.mu.Lock() defer h.mu.Unlock() r.AddAttrs(h.attrs...) *h.records = append(*h.records, r) return nil}
func (h *captureHandler) WithAttrs(attrs []slog.Attr) slog.Handler { newAttrs := append([]slog.Attr(nil), h.attrs...) newAttrs = append(newAttrs, attrs...) return &captureHandler{ attrs: newAttrs, records: h.records, mu: h.mu, }}
func (h *captureHandler) WithGroup(name string) slog.Handler { return h}
func TestStartWorkflowsPassesPlannedCacheBindingsToEngine(t *testing.T) { t.Parallel()
database := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) expected := []models.CacheBinding{{ EntryIndex: 0, Paths: []string{"deps"}, Key: "deps", SaveID: "cache-id", SaveKey: "objects/cache-id", }} var got []models.CacheBinding eng := &mockEngine{ runWorkflowFunc: func(workflow *models.Workflow) { got = append([]models.CacheBinding(nil), workflow.CacheBindings...) }, } pipelineID := models.PipelineId("test-rkey") pipeline := &models.Pipeline{ RepoDid: "did:web:example.com", TrustedSource: true, Workflows: map[models.Engine][]models.Workflow{ eng: {{ Name: "cache-job", Engine: "microvm", Steps: []models.Step{mockStep{name: "step1"}}, Caches: []models.CacheEntry{{Key: "deps", Paths: []string{"deps"}}}, }}, }, } cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} store := &fakeStorage{objects: make(map[string][]byte)}
StartWorkflows( logger, emptyVault{}, cfg, nil, nil, database, nil, store, staticCacheController{bindings: expected}, context.Background(), pipeline, pipelineID, )
if !reflect.DeepEqual(got, expected) { t.Fatalf("cache bindings passed to engine = %#v, want %#v", got, expected) }}
func TestStartWorkflowsDestroysWorkflowWhenSetupIsCancelled(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
setupStarted := make(chan struct{}) destroyed := make(chan struct{}, 1) eng := &mockEngine{ setupFunc: func(ctx context.Context, wid models.WorkflowId) error { close(setupStarted) <-ctx.Done() return ctx.Err() }, destroyFunc: func(_ context.Context, wid models.WorkflowId) error { destroyed <- struct{}{} return nil }, }
pipelineId := models.PipelineId("test-rkey")
wid := models.WorkflowId{ PipelineId: pipelineId, Name: "cancel_setup_job", }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "cancel_setup_job", Steps: []models.Step{mockStep{name: "step1"}}, }, }, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} doneChan := make(chan struct{}) go func() { StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId) close(doneChan) }()
select { case <-setupStarted: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for workflow setup to start") }
CancelWorkflow(wid)
select { case <-doneChan: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for StartWorkflows to complete") }
select { case <-destroyed: case <-time.After(5 * time.Second): t.Fatal("cancelled setup did not destroy the workflow") }
st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st != models.StatusKindCancelled { t.Fatalf("expected status to be cancelled, got %s", st) }}