Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888//go:build torture
package spindle
import ( "context" "database/sql" "fmt" "io/fs" "math" "net" "net/url" "os" "path/filepath" "runtime" "runtime/debug" "strings" "sync" "sync/atomic" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/gorilla/websocket"
"tangled.org/core/api/org_tangled" "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" pipelinecodec "tangled.org/core/spindle/pipeline" "tangled.org/core/workflow")
var tqSeq atomic.Int64
func tqWorld(r *tortureRun, owners, perOwner, workers int) (*tortureWorld, []*tortureRepo) { w := newTortureWorld(r) w.maxJobs = workers k := w.addKnot() repos := w.addRepos(owners*perOwner, func(i int) (*tortureKnot, syntax.DID) { return k, ownerFor(i, perOwner) }, 0, nil) w.boot() w.seed(repos, 3*time.Minute) return w, repos}
func (w *tortureWorld) tqPipeline(repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, workflow.PipelineSpec, *models.PipelineRecord, error) { n := tqSeq.Add(1) id := models.PipelineId(fmt.Sprintf("tq%09d", n)) repoDid := repo.did.String() rkey := repo.name actor := repo.owner.String() tpl := workflow.PipelineSpec{ TriggerMetadata: &workflow.TriggerMetadata{ Kind: "push", Actor: &actor, Push: &workflow.PushTriggerData{NewSha: fmt.Sprintf("%040x", n), OldSha: zeroSha, Ref: "refs/heads/main"}, Repo: &workflow.TriggerRepo{ Did: repo.owner.String(), Knot: repo.knot.host(), Repo: &rkey, RepoDid: &repoDid, DefaultBranch: "main", }, }, } for i := range workflows { name := "ci.yaml" if workflows > 1 { name = fmt.Sprintf("w%d.yaml", i) } tpl.Workflows = append(tpl.Workflows, &workflow.WorkflowSpec{ Name: name, Engine: "torture", Raw: tortureWorkflow(steps...), Clone: &workflow.CloneSpec{Skip: true}, }) } record, err := pipelinecodec.FromTangled(id, time.Now(), tpl) return id, tpl, record, err}
func (w *tortureWorld) tqEnqueueTimed(lat *latencies, repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, error) { id, tpl, record, err := w.tqPipeline(repo, workflows, steps...) if err != nil { return "", err } start := time.Now() err = w.s.processPipeline(context.Background(), repo.did, tpl, record, nil, false) if lat != nil { lat.since(start) } return id, err}
func (w *tortureWorld) tqEnqueue(repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, error) { return w.tqEnqueueTimed(nil, repo, workflows, steps...)}
func (w *tortureWorld) tqEnqueueMany(repos []*tortureRepo, n, par, workflows int, lat *latencies, steps ...string) []models.PipelineId { ids := make([]models.PipelineId, n) var next atomic.Int64 var wg sync.WaitGroup for range par { wg.Go(func() { for { i := int(next.Add(1)) - 1 if i >= n { return } id, err := w.tqEnqueueTimed(lat, repos[i%len(repos)], workflows, steps...) if err != nil { w.r.count("enqueue_error", 1) continue } ids[i] = id } }) } wg.Wait() return ids}
func (w *tortureWorld) tqFinished(within time.Duration) bool { return w.r.waitFor(within, func() bool { return w.unfinishedWorkflows() == 0 })}
func (w *tortureWorld) tqWorkflowDone(id models.PipelineId, name string) bool { st, err := w.s.db.GetStatus(models.WorkflowId{PipelineId: id, Name: name}) return err == nil && st.IsFinish()}
func (w *tortureWorld) tqStartDelays(ids []models.PipelineId, name string, lat *latencies) { for _, id := range ids { if id == "" { continue } d, ok, err := w.s.db.WorkflowStartupDelay(context.Background(), models.WorkflowId{PipelineId: id, Name: name}) if err == nil && ok { lat.add(d) } }}
// must match logWriteTimeout in xrpc/ci_pipeline_subscribe_logs.goconst tqLogWriteTimeout = time.Minute
func tqPeaks(r *tortureRun, fn func(), gauges ...string) []float64 { peaks := make([]float64, len(gauges)) for i, g := range gauges { peaks[i] = r.gaugeValue(g) } done := make(chan struct{}) var wg sync.WaitGroup wg.Go(func() { tick := time.NewTicker(25 * time.Millisecond) defer tick.Stop() for { select { case <-done: return case <-tick.C: for i, g := range gauges { peaks[i] = max(peaks[i], r.gaugeValue(g)) } } } }) fn() close(done) wg.Wait() return peaks}
func tqSettle(r *tortureRun, phase, gauge string, ceiling float64, within time.Duration) { ok := r.waitFor(within, func() bool { return r.gaugeValue(gauge) <= ceiling }) r.check(phase+": "+gauge+" settles", ok, "%s is %.0f after %v, want <= %.0f", gauge, r.gaugeValue(gauge), within, ceiling) if ok { return } switch gauge { case "goroutines": r.dumpGoroutines() case "fds": tqFdSummary(r) }}
func tqSettleBoth(r *tortureRun, phase string, baseGoroutines, baseFds, fdSlack float64, within time.Duration) { tqSettle(r, phase, "goroutines", baseGoroutines+30, within) tqSettle(r, phase, "fds", baseFds+fdSlack, within)}
func tqFdSummary(r *tortureRun) { targets, err := fdTargets() if err != nil { return } kinds := map[string]int{} others := map[string]int{} for _, target := range targets { switch { case target == "": kinds["gone"]++ case strings.HasPrefix(target, "socket:"): kinds["socket"]++ case strings.HasPrefix(target, "pipe:"): kinds["pipe"]++ case strings.HasPrefix(target, "anon_inode:"): kinds[target]++ case strings.HasSuffix(target, ".log"): kinds["workflow log"]++ default: kinds["other"]++ others[filepath.Base(target)]++ } } r.note("fds by kind: %v", kinds) r.note("fds of kind other by file name: %v", others)}
func tqLargestFileMB(root string) float64 { var largest int64 _ = filepath.WalkDir(root, func(_ string, d fs.DirEntry, err error) error { if err != nil || d.IsDir() { return nil } if info, err := d.Info(); err == nil { largest = max(largest, info.Size()) } return nil }) return float64(largest) / (1 << 20)}
func tqOpenLogs(root string) float64 { targets, err := fdTargets() if err != nil { return math.NaN() } n := 0 for _, target := range targets { if strings.HasPrefix(target, root) && strings.HasSuffix(target, ".log") { n++ } } return float64(n)}
func tqFileCount(root string) float64 { entries, err := os.ReadDir(root) if err != nil { return 0 } return float64(len(entries))}
func (w *tortureWorld) tqWSURL(nsid string, repo syntax.DID, pipeline string, workflows ...string) string { q := url.Values{} q.Set("repo", repo.String()) q.Set("pipeline", pipeline) for _, name := range workflows { q.Add("workflows", name) } return "ws" + strings.TrimPrefix(w.api.URL, "http") + "/xrpc/" + nsid + "?" + q.Encode()}
// small recv buffer so a stalled reader backs up the send queue quicklyfunc tqDialer(smallRecvBuf bool) *websocket.Dialer { d := &websocket.Dialer{HandshakeTimeout: 30 * time.Second} if smallRecvBuf { d.NetDial = func(network, addr string) (net.Conn, error) { c, err := net.Dial(network, addr) if err != nil { return nil, err } if tcp, ok := c.(*net.TCPConn); ok { _ = tcp.SetReadBuffer(4096) } return c, nil } } return d}
func tqDrain(conn *websocket.Conn, bytesRead, msgs *atomic.Int64) { for { _, p, err := conn.ReadMessage() if err != nil { return } bytesRead.Add(int64(len(p))) msgs.Add(1) }}
func init() { // the only worker gets stuck here and the backlog piles up behind it registerTorture("queue-backlog", "a big backlog nobody is draining yet, then a restart that works it off", func(r *tortureRun) { depth := r.load("jobs", 4000) probes := 50 par := 4
w, repos := tqWorld(r, 10, 3, 1) if _, err := w.tqEnqueue(repos[0], 1, "hang"); err != nil { r.t.Fatal(err) } if !r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() == 1 }) { r.t.Fatal("the hung pipeline never started") } baseMB := r.gaugeValue("db_mb")
var enqueue latencies var firstMedian, lastMedian, firstEnqueue time.Duration var writes latencies var quiet latencies for range 50 { start := time.Now() if err := w.s.db.StatusPending(models.WorkflowId{PipelineId: "tq-quiet", Name: "ci.yaml"}, nil); err != nil { r.count("status_write", 1) } quiet.since(start) } quietWrite := quiet.quantile(0.99) levels := []int{depth / 8, depth / 4, depth / 2, depth} for _, level := range levels { have := int(w.rows("jobs")) w.tqEnqueueMany(repos, max(level-have, 0), par, 1, &enqueue, "sleep 1ms") if level == levels[0] { firstEnqueue = enqueue.quantile(0.99) }
// probes only pull 50 jobs a level, not enough to shift the depth var lat latencies stopWrites := make(chan struct{}) var wg sync.WaitGroup wg.Go(func() { wid := models.WorkflowId{PipelineId: "tq-probe", Name: "ci.yaml"} for { select { case <-stopWrites: return case <-time.After(2 * time.Millisecond): } start := time.Now() if err := w.s.db.StatusPending(wid, nil); err != nil { r.count("status_write", 1) } writes.since(start) } }) for range probes { start := time.Now() if _, err := w.s.db.DequeueJob(context.Background(), nil); err != nil { r.count("dequeue_probe", 1) } lat.since(start) } close(stopWrites) wg.Wait()
if level == levels[0] { firstMedian = lat.quantile(0.5) } lastMedian = lat.quantile(0.5) } r.latency("enqueue", &enqueue) r.latency("status_write_during_dequeue", &writes) have := int(w.rows("jobs")) r.metric("db_kb_per_queued_job", (r.gaugeValue("db_mb")-baseMB)*1024/float64(max(have, 1)))
r.metric("dequeue_ms_per_1k_queued", float64(lastMedian.Microseconds())/1000/float64(max(have, 1))*1000) r.check("D1 dequeue cost is flat in queue depth", lastMedian <= 3*firstMedian+500*time.Microsecond, "median dequeue %v at depth %d vs %v at depth %d", lastMedian, levels[len(levels)-1], firstMedian, levels[0]) r.check("D1 a dequeue doesn't stall other writers", writes.quantile(0.99) <= 10*quietWrite+50*time.Millisecond, "status write p99 %v while dequeues ran, %v when quiet, want within 10x plus 50ms", writes.quantile(0.99), quietWrite) r.check("enqueue stays fast under the backlog", enqueue.quantile(0.99) <= 10*firstEnqueue+50*time.Millisecond, "enqueue p99 %v over the whole run, %v at the first depth, want within 10x plus 50ms", enqueue.quantile(0.99), firstEnqueue)
w.stop() w.maxJobs = int(r.param("drain_workers", 8)) w.boot()
total := int(w.rows("jobs")) marks := []float64{0.75, 0.5, 0.25, 0} at := make([]time.Duration, len(marks)) drainStart := time.Now() drained := true for i, m := range marks { target := int(float64(total) * m) if !r.waitFor(4*time.Minute, func() bool { return int(w.rows("jobs")) <= target }) { drained = false break } at[i] = time.Since(drainStart) } r.check("queue drains after the restart", drained, "%v jobs still queued", w.rows("jobs")) if drained && at[0] > 0 && at[3] > at[2] { first := float64(total) / 4 / at[0].Seconds() last := float64(total) / 4 / (at[3] - at[2]).Seconds() r.metric("drain_first_quarter_per_s", first) r.metric("drain_last_quarter_per_s", last) r.check("D1 drain rate doesn't depend on how deep the queue is", first >= 0.7*last, "%.0f jobs/s over the first quarter vs %.0f jobs/s over the last", first, last) } r.note("the repo schedule refresh warnings in this run are the records-only repos having no git, harness noise") r.note("the fake engine does no work, drain rates are about dequeue and bookkeeping only. probe-dequeued jobs and the hung pipeline are left unfinished on purpose") })
registerTorture("queue-fairness", "one owner floods the queue while a few small owners push a little", func(r *tortureRun) { flood := r.load("flood", 1500) smallOwners := 5 smallJobs := 3 workers := int(r.param("workers", 4)) floodStep := r.duration("flood_step", 100*time.Millisecond)
w, repos := tqWorld(r, 1+smallOwners, 1, workers) floodRepo, smallRepos := repos[0], repos[1:] floodOwner := floodRepo.owner.String()
w.tqEnqueueMany(repos[:1], flood, 4, 1, nil, "sleep "+floodStep.String()) if !r.waitFor(time.Minute, func() bool { return w.eng.started.Load() >= int64(workers*3) }) { r.t.Fatal("the flood never started running") } ahead := len(w.eng.startOrder()) r.metric("queued_when_smalls_arrive", w.rows("jobs"))
var smallIDs []models.PipelineId for range smallJobs { smallIDs = append(smallIDs, w.tqEnqueueMany(smallRepos, smallOwners, 1, 1, nil, "sleep "+floodStep.String())...) } done := r.waitFor(5*time.Minute, func() bool { for _, id := range smallIDs { if !w.tqWorkflowDone(id, "ci.yaml") { return false } } return true }) var delay latencies w.tqStartDelays(smallIDs, "ci.yaml", &delay) r.latency("small_owner_start_delay", &delay)
order := w.eng.startOrder() firstSmall := -1 for i := ahead; i < len(order); i++ { if order[i] != floodOwner { firstSmall = i break } } r.metric("flood_starts_ahead_of_first_small", float64(firstSmall-ahead)) r.metric("flood_starts_total", float64(flood)) r.check("small owners finish while the flood is still queued", done && w.rows("jobs") > 0, "done=%v queued=%v", done, w.rows("jobs")) r.check("Q3 a small owner is only ever behind about one round of flood jobs", firstSmall >= 0 && firstSmall-ahead <= 2*workers, "%d flood jobs started after the small ones were queued and before the first small one", firstSmall-ahead) r.check("Q3 small owners start within a few job lengths", delay.quantile(0.99) <= 20*floodStep, "p99 wait %v with %v jobs and %d workers", delay.quantile(0.99), floodStep, workers) r.check("the flood drains", w.tqFinished(10*time.Minute), "%v workflows unfinished", w.unfinishedWorkflows()) })
registerTorture("queue-restart-mid-run", "restarts with work in flight, and a worker that wakes into a database error", func(r *tortureRun) { hung := r.load("hung", 6) queued := r.load("queued", 30) recoverWait := 20 * time.Second rounds := r.load("wake_rounds", 20) lockHold := 7 * time.Second
w, repos := tqWorld(r, 3, 1, hung+2) startHung := func(tag string) []models.PipelineId { ids := w.tqEnqueueMany(repos, hung, 2, 1, nil, "hang") if !r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() >= int64(hung) }) { r.t.Fatalf("%s: only %d of %d hung workflows started", tag, w.eng.running.Load(), hung) } return ids }
startHung("clean") w.tqEnqueueMany(repos, queued, 2, 1, nil, "sleep 50ms") w.stop() r.check("a clean stop destroys every running workflow", w.eng.running.Load() == 0, "%d still running", w.eng.running.Load()) w.boot() finished := w.tqFinished(2 * time.Minute) r.check("queued jobs survive a clean restart and finish", finished, "%v workflows pending or running", w.unfinishedWorkflows())
// erase them here, a killed process never writes its terminal row crashed := startHung("crash") w.stop() w.tqEraseTerminals(crashed) w.boot() recovered := r.waitFor(recoverWait, func() bool { return w.unfinishedWorkflows() == 0 }) r.metric("unfinished_after_crash", w.unfinishedWorkflows()) notFailed := 0 for _, id := range crashed { if id == "" { continue } if status, err := w.s.db.GetStatus(models.WorkflowId{PipelineId: id, Name: "ci.yaml"}); err != nil || status != models.StatusKindFailed { notFailed++ } } r.check("Q1 a crash restart marks what it was running as failed", notFailed == 0, "%d of %d crashed workflows are not failed, want 0", notFailed, len(crashed)) r.check("Q1 nothing stays pending or running forever after a crash restart", recovered, "%v workflows pending or running %v after the restart, want 0", w.unfinishedWorkflows(), recoverWait)
if !w.tqFinished(time.Minute) { r.note("workflows from the crash step are still unfinished, wake rounds run on top of them") } stuck := 0 for range rounds { var ids [2]models.PipelineId var wg sync.WaitGroup for i := range ids { wg.Go(func() { id, err := w.tqEnqueue(repos[i%len(repos)], 1, "hang") if err != nil { r.count("enqueue_error", 1) } ids[i] = id }) } wg.Wait() if !r.waitFor(10*time.Second, func() bool { return w.eng.running.Load() >= 2 }) { stuck++ } for _, id := range ids { engine.CancelWorkflow(models.WorkflowId{PipelineId: id, Name: "ci.yaml"}) } r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() == 0 }) } r.metric("rounds_with_a_second_job_not_started", float64(stuck)) r.check("W2 a second enqueue wakes a second idle worker", stuck == 0, "%d of %d rounds had a job that never started with idle workers around", stuck, rounds) r.note("the wake race is narrow, a clean result here doesn't prove the channel is safe")
startedBefore := w.eng.started.Load() id, tpl, record, err := w.tqPipeline(repos[0], 1, "sleep 10ms") if err != nil { r.t.Fatal(err) } if err := w.s.db.EnqueuePipeline(context.Background(), repos[0].did.String(), record, nil, tpl, false, "", "", nil); err != nil { r.t.Fatal(err) } conn, err := w.s.db.Conn(context.Background()) if err != nil { r.t.Fatal(err) } if _, err := conn.ExecContext(context.Background(), "begin immediate"); err != nil { r.t.Fatal(err) } select { case w.s.jobWake <- struct{}{}: default: } time.Sleep(lockHold) _, _ = conn.ExecContext(context.Background(), "rollback") conn.Close() resumed := r.waitFor(15*time.Second, func() bool { return w.eng.started.Load() > startedBefore }) r.check("W2 the queue resumes after a dequeue error with no new enqueue", resumed, "job %s started=%v within %v of the lock being released, want started", id, resumed, 15*time.Second) select { case w.s.jobWake <- struct{}{}: default: } })
registerTorture("logs-flood", "log output without newlines, a lot of output, and many workflows at once", func(r *tortureRun) { rawMB := 64 dribbleKB := 2048 logMB := 24 logWorkflows := 4 many := r.load("many_workflows", 150) slots := int(r.param("slots", 4)) capMB := r.param("log_cap_mb", 33) heapBudget := r.param("heap_budget_mb", 48)
w, repos := tqWorld(r, 1, 1, 4) eng := w.eng repo := repos[0] logDir := filepath.Join(w.dir, "logs") baseGoroutines := r.gaugeValue("goroutines") baseFds := r.gaugeValue("fds")
// no newline means the whole thing waits in the writer's pending buffer rawCmd := fmt.Sprintf("raw %d", rawMB<<20) // tight gc so heap numbers are live memory, not garbage defer debug.SetGCPercent(debug.SetGCPercent(10)) runtime.GC() baseHeap := r.gaugeValue("heap_mb") peaks := tqPeaks(r, func() { if _, err := w.tqEnqueue(repo, 1, rawCmd); err != nil { r.t.Fatal(err) } if !w.tqFinished(5 * time.Minute) { r.check("raw output workflow finishes", false, "still unfinished after 5m") } }, "heap_mb", "rss_mb") debug.SetGCPercent(100) r.metric("raw_heap_growth_mb", peaks[0]-baseHeap) r.metric("raw_rss_peak_mb", peaks[1]) r.check("L1 output without newlines doesn't pile up in memory", peaks[0]-baseHeap <= heapBudget, "heap grew %.0fMB while a step wrote %dMB with no newline, budget %.0fMB", peaks[0]-baseHeap, rawMB, heapBudget)
// doubling the bytes should only double the time, small writes rescan the pending line dribble := func(kb int) time.Duration { cmd := fmt.Sprintf("dribble %d", kb<<10) if _, err := w.tqEnqueue(repo, 1, cmd); err != nil { r.t.Fatal(err) } w.tqFinished(5 * time.Minute) return eng.tookFor(cmd) } once, twice := dribble(dribbleKB), dribble(2*dribbleKB) r.metric("dribble_step_s", once.Seconds()) r.check("L1 a long line written in small chunks costs linear time", twice <= 3*once+50*time.Millisecond, "%dKB in 16 byte writes took %v, %dKB took %v, want the double within 3x plus 50ms", dribbleKB, once, 2*dribbleKB, twice)
logCmd := fmt.Sprintf("log %d", logMB<<20) // sample the size while it still grows, logs get deleted once archived r.gauge("largest_log_mb", func() float64 { return tqLargestFileMB(logDir) }) largest := tqPeaks(r, func() { if _, err := w.tqEnqueue(repo, logWorkflows, logCmd); err != nil { r.t.Fatal(err) } w.tqFinished(5 * time.Minute) }, "largest_log_mb")[0] r.metric("largest_log_file_mb", largest) r.check("L1 a workflow's log file is capped", largest > 1 && largest <= capMB, "largest log file seen while running is %.1fMB, want over 1MB so the flood was seen and at most %.0fMB", largest, capMB) r.note("log_cap_mb is the default LogMaxBytes (32MB) plus a line and a marker, dawn picks the real number")
w.eng.slots = engine.NewSemaphoreSlotter(slots) r.gauge("open_workflow_logs", func() float64 { return tqOpenLogs(logDir) }) peaks = tqPeaks(r, func() { if _, err := w.tqEnqueue(repo, many, "lines 100; sleep 200ms"); err != nil { r.t.Fatal(err) } w.tqFinished(5 * time.Minute) }, "fds", "goroutines", "open_workflow_logs") logPeak := peaks[2] r.metric("many_open_logs_peak", logPeak) r.check("Q2 log files are only open for workflows that hold a slot", logPeak <= float64(slots)+16, "%d workflows on %d slots peaked %.0f open workflow log fds, want <= %d (a few more are archives in flight)", many, slots, logPeak, slots+16)
r.note("many_workflows has no cap in spindle (Q2), this only measures what a pipeline that wide costs. the log engine writes real files, the numbers are about spindle's writer and bookkeeping, nothing an engine does")
tqSettleBoth(r, "after the floods", baseGoroutines, baseFds, 64, 30*time.Second) })
registerTorture("logs-subscribers", "many log subscribers, stalled readers and vanishing clients", func(r *tortureRun) { subs := r.load("subscribers", 300) slow := r.load("slow_readers", 60) churn := r.load("churn", 400) slowWait := tqLogWriteTimeout + 10*time.Second backlogMB := 32
w, repos := tqWorld(r, 1, 1, 4) repo := repos[0] logDir := filepath.Join(w.dir, "logs") dbStat := func(read func(sql.DBStats) int) func() float64 { return func() float64 { return w.fromSpindle(func(s *Spindle) float64 { return float64(read(s.db.Stats())) }) } } r.gauge("db_open_conns", dbStat(func(st sql.DBStats) int { return st.OpenConnections })) r.gauge("db_conns_in_use", dbStat(func(st sql.DBStats) int { return st.InUse })) logSize := func(id models.PipelineId) int64 { fi, err := os.Stat(models.LogFilePath(logDir, models.WorkflowId{PipelineId: id, Name: "ci.yaml"})) if err != nil { return 0 } return fi.Size() }
small, err := w.tqEnqueue(repo, 1, "lines 200", "hang") if err != nil { r.t.Fatal(err) } big, err := w.tqEnqueue(repo, 1, fmt.Sprintf("log %d", backlogMB<<20), "hang") if err != nil { r.t.Fatal(err) } if !r.waitFor(time.Minute, func() bool { return w.eng.running.Load() >= 2 && logSize(big) >= int64(backlogMB)<<20 }) { r.t.Fatal("the running pipelines never got their logs written") } smallURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, repo.did, string(small), "ci.yaml") bigURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, repo.did, string(big), "ci.yaml") // take the baseline after the first viewer warms lazy state if conn, _, err := tqDialer(false).Dial(smallURL, nil); err == nil { _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) _, _, _ = conn.ReadMessage() conn.Close() } time.Sleep(2 * time.Second) baseGoroutines := r.gaugeValue("goroutines") baseFds := r.gaugeValue("fds")
var first latencies var bytesRead, msgs atomic.Int64 conns := make([]*websocket.Conn, subs) sem := make(chan struct{}, 50) var wg sync.WaitGroup for i := range subs { wg.Go(func() { sem <- struct{}{} start := time.Now() conn, _, err := tqDialer(false).Dial(smallURL, nil) <-sem if err != nil { r.count("dial", 1) return } _ = conn.SetReadDeadline(time.Now().Add(30 * time.Second)) _, p, err := conn.ReadMessage() if err != nil { r.count("first_message", 1) conn.Close() return } first.since(start) bytesRead.Add(int64(len(p))) _ = conn.SetReadDeadline(time.Time{}) conns[i] = conn go tqDrain(conn, &bytesRead, &msgs) }) } wg.Wait() r.latency("herd_first_message", &first) r.metric("goroutines_per_subscriber", (r.gaugeValue("goroutines")-baseGoroutines)/float64(subs)) r.metric("fds_per_subscriber", (r.gaugeValue("fds")-baseFds)/float64(subs)) time.Sleep(3 * time.Second) for _, c := range conns { if c != nil { c.Close() } } tqSettleBoth(r, "herd leaves", baseGoroutines, baseFds, 30, 30*time.Second)
// stalled readers, with more backlog than the socket buffers hold baseGoroutines, baseFds = r.gaugeValue("goroutines"), r.gaugeValue("fds") var slowConns []*websocket.Conn for range slow { conn, _, err := tqDialer(true).Dial(bigURL, nil) if err != nil { r.count("slow_dial", 1) continue } slowConns = append(slowConns, conn) } time.Sleep(5 * time.Second) r.metric("slow_goroutines_per_reader_at_5s", (r.gaugeValue("goroutines")-baseGoroutines)/float64(max(len(slowConns), 1))) time.Sleep(max(slowWait-5*time.Second, 0)) r.metric("slow_fds_over_base", r.gaugeValue("fds")-baseFds) for _, c := range slowConns { c.UnderlyingConn().Close() } tqSettleBoth(r, "slow readers leave", baseGoroutines, baseFds, 30, 15*time.Second)
// they leave with backlog still unsent on the server for _, abrupt := range []bool{false, true} { baseGoroutines, baseFds = r.gaugeValue("goroutines"), r.gaugeValue("fds") sem = make(chan struct{}, 25) for range churn { wg.Go(func() { sem <- struct{}{} defer func() { <-sem }() conn, _, err := tqDialer(false).Dial(smallURL, nil) if err != nil { r.count("churn_dial", 1) return } _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) _, _, _ = conn.ReadMessage() if abrupt { conn.UnderlyingConn().Close() } else { conn.Close() } }) } wg.Wait() phase := "churn with a close frame" if abrupt { phase = "churn without a close frame" } tqSettleBoth(r, phase, baseGoroutines, baseFds, 30, 15*time.Second) } })
registerTorture("db-growth", "pipelines and statuses at a steady rate, what grows and what slows", func(r *tortureRun) { soak := 3 * time.Minute perSec := float64(30) workflows := int(r.param("workflows", 3)) owners := r.load("owners", 10)
w, repos := tqWorld(r, owners, 2, 8) logDir := filepath.Join(w.dir, "logs") r.gauge("log_files", func() float64 { return tqFileCount(logDir) }) baseMB := r.gaugeValue("db_mb")
var recentEarly, recentLate, rareEarly, rareLate latencies probe := func(repo *tortureRepo, elapsed time.Duration) { query := func(kinds []string, limit int, early, late *latencies) { start := time.Now() if _, _, _, err := w.s.db.QueryPipelines(context.Background(), repo.did.String(), nil, "", kinds, limit); err != nil { r.count("query", 1) } switch { case elapsed < soak/3: early.since(start) case elapsed > 2*soak/3: late.since(start) } } query(nil, 50, &recentEarly, &recentLate) query([]string{"manual"}, 30, &rareEarly, &rareLate) }
var enqueue latencies interval := time.Duration(float64(time.Second) / perSec) begin := time.Now() next := begin lastProbe := begin sent := 0 for time.Since(begin) < soak { // half go to repos[0] so its history is long enough for the query probes repo := repos[0] if sent%2 == 1 { repo = repos[sent%len(repos)] } if _, err := w.tqEnqueueTimed(&enqueue, repo, workflows, "lines 20"); err != nil { r.count("enqueue_error", 1) } sent++ if time.Since(lastProbe) >= 5*time.Second { probe(repos[0], time.Since(begin)) lastProbe = time.Now() } next = next.Add(interval) time.Sleep(time.Until(next)) } r.latency("enqueue", &enqueue) r.metric("queued_at_end_of_load", w.rows("jobs")) r.check("the queue keeps up with the load", w.rows("jobs") <= 50, "%v jobs still queued at the end of the load", w.rows("jobs")) w.tqFinished(2 * time.Minute)
pipelines := w.rows("pipelines") statuses := w.rows("workflow_statuses") growthMB := r.gaugeValue("db_mb") - baseMB r.metric("pipelines", pipelines) r.metric("status_rows", statuses) r.metric("db_kb_per_pipeline", growthMB*1024/max(pipelines, 1)) r.metric("db_gb_per_month_at_this_rate", growthMB*1024/max(pipelines, 1)*perSec*86400*30/1e6) r.metric("log_dir_files", tqFileCount(logDir)) r.latency("query_recent_early", &recentEarly) r.latency("query_recent_late", &recentLate) r.latency("query_rare_kind_early", &rareEarly) r.latency("query_rare_kind_late", &rareLate)
r.check("QP1 listing a repo's recent pipelines doesn't slow down as history grows", recentLate.len() == 0 || recentLate.quantile(0.5) <= 2*recentEarly.quantile(0.5)+300*time.Microsecond, "median %v late vs %v early", recentLate.quantile(0.5), recentEarly.quantile(0.5)) r.check("QP1 a rare kind filter doesn't walk the whole history", rareLate.len() == 0 || rareLate.quantile(0.5) <= 2*rareEarly.quantile(0.5)+300*time.Microsecond, "median %v late vs %v early with no matching rows", rareLate.quantile(0.5), rareEarly.quantile(0.5)) archived := w.rows("mill_artifacts") r.metric("archived_logs", archived) if archived > 0 { r.check("G1 a log that is archived leaves local disk", tqFileCount(logDir) <= 10, "%.0f log files in the log dir with %.0f archived, want <= 10", tqFileCount(logDir), archived) } else { r.note("no log was archived in this setup, so log disk growth shows in log_dir_files but isn't checked") } r.note("one workflow is %d statuses at least (pending, running, success). the fake engine writes %s of log per workflow, the db numbers are bookkeeping only", 3, "20 lines") })}
func (w *tortureWorld) tqEraseTerminals(ids []models.PipelineId) { d, err := db.Make(context.Background(), w.dbPath()) if err != nil { w.r.t.Fatal(err) } defer d.Close() for _, id := range ids { if id == "" { continue } if _, err := d.Exec(`delete from workflow_statuses where pipeline_id = ? and status not in ('pending', 'running')`, string(id)); err != nil { w.r.t.Fatal(err) } }}