package main
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"sync"
"testing"
"craftdesign.group/quarry/internal/ci"
)
// laneEngine is dagnats' #624 tail API over a map keyed the way the
// engine keys its hot lane: step, then attempt. A read without
// ?attempt= answers the step's latest attempt, as the engine does.
type laneEngine struct {
mu sync.Mutex
lanes map[string]map[int][]ci.LogChunk
}
func (l *laneEngine) set(step string, attempt int, chunks []ci.LogChunk) {
l.mu.Lock()
defer l.mu.Unlock()
if l.lanes[step] == nil {
l.lanes[step] = map[int][]ci.LogChunk{}
}
l.lanes[step][attempt] = chunks
}
func (l *laneEngine) ServeHTTP(w http.ResponseWriter, r *http.Request) {
l.mu.Lock()
defer l.mu.Unlock()
step := r.URL.Query().Get("step")
attempt, _ := strconv.Atoi(r.URL.Query().Get("attempt"))
if attempt == 0 {
for n := range l.lanes[step] {
attempt = max(attempt, n)
}
}
type wire struct {
Seq int `json:"seq"`
Stream string `json:"stream"`
Data []byte `json:"data"`
}
page := struct {
Chunks []wire `json:"chunks"`
EOF bool `json:"eof"`
}{Chunks: []wire{}, EOF: true}
for _, c := range l.lanes[step][attempt] {
page.Chunks = append(page.Chunks, wire{Seq: c.Seq, Stream: c.Stream, Data: []byte(c.Data)})
}
_ = json.NewEncoder(w).Encode(page)
}
func drainedAttempts(t *testing.T, dir, step string) map[int][]ci.LogChunk {
t.Helper()
all, ok := ciLogStore{dir: dir}.DrainedLogs("alpha", "run-1")
if !ok {
t.Fatal("nothing drained")
}
out := map[int][]ci.LogChunk{}
for _, c := range all {
if c.Step == step {
out[c.Attempt] = append(out[c.Attempt], c)
}
}
return out
}
func hasChunk(chunks []ci.LogChunk, stream, data string) bool {
for _, c := range chunks {
if c.Stream == stream && c.Data == data {
return true
}
}
return false
}
// #269: a step that fails its first attempt and passes its second keeps
// both logs, each stamped with its attempt; the failing one is intact.
func TestDrainKeepsEveryAttempt(t *testing.T) {
lane := &laneEngine{lanes: map[string]map[int][]ci.LogChunk{}}
lane.set("quarry.test", 1, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "--- FAIL: TestFlake"},
{Seq: 2, Stream: "marker", Data: "failed"}})
lane.set("quarry.test", 2, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "ok"},
{Seq: 2, Stream: "marker", Data: "completed"}})
srv := httptest.NewServer(lane)
defer srv.Close()
dir := t.TempDir()
steps := map[string]ci.Step{"quarry.test": {Status: "completed", Attempts: 2}}
if err := drainLogs(ci.New(srv.URL), dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got := drainedAttempts(t, dir, "quarry.test")
if len(got) != 2 {
t.Fatalf("attempts drained = %d, want 2: %+v", len(got), got)
}
if !hasChunk(got[1], "out", "--- FAIL: TestFlake") || !hasChunk(got[1], "marker", "failed") {
t.Fatalf("attempt 1 lost its failure: %+v", got[1])
}
if !hasChunk(got[2], "marker", "completed") {
t.Fatalf("attempt 2 missing: %+v", got[2])
}
}
// #269, run 79670815: the run is marked failed (another step ran out of
// attempts) while this step is mid-retry. The drain keeps finished
// attempt 1 beside running attempt 2, and a later drain — the verdict
// retrying, the hot lane expiring — never replaces the finished
// attempt with anything, while the running one may still grow.
func TestDrainNeverOverwritesAFinishedAttempt(t *testing.T) {
lane := &laneEngine{lanes: map[string]map[int][]ci.LogChunk{}}
lane.set("quarry.test", 1, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "--- FAIL: TestFlake"},
{Seq: 2, Stream: "marker", Data: "failed"}})
lane.set("quarry.test", 2, []ci.LogChunk{{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"}})
srv := httptest.NewServer(lane)
defer srv.Close()
e := ci.New(srv.URL)
dir := t.TempDir()
steps := map[string]ci.Step{"quarry.test": {Status: "running", Attempts: 2}}
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got := drainedAttempts(t, dir, "quarry.test")
if !hasChunk(got[1], "marker", "failed") || !hasChunk(got[2], "out", "=== RUN TestFlake") {
t.Fatalf("first drain: %+v", got)
}
// The hot lane moves on: attempt 1's subject now answers with other
// output, attempt 2 has grown.
lane.set("quarry.test", 1, []ci.LogChunk{{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"}})
lane.set("quarry.test", 2, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"},
{Seq: 2, Stream: "out", Data: "--- PASS: TestFlake"}})
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got = drainedAttempts(t, dir, "quarry.test")
if len(got[1]) != 2 || !hasChunk(got[1], "out", "--- FAIL: TestFlake") || !hasChunk(got[1], "marker", "failed") {
t.Fatalf("finished attempt 1 was overwritten: %+v", got[1])
}
if !hasChunk(got[2], "out", "--- PASS: TestFlake") {
t.Fatalf("running attempt 2 did not grow: %+v", got[2])
}
// And the lane expiring entirely leaves both as they were.
lane.set("quarry.test", 1, nil)
lane.set("quarry.test", 2, nil)
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got = drainedAttempts(t, dir, "quarry.test")
if len(got[1]) != 2 || len(got[2]) != 2 {
t.Fatalf("an expired lane erased the record: %+v", got)
}
}