vole

Clone

Read only, no credentials:

https://rodent.software/git/vole.git

Write over SSH with a registered key:

git clone ssh://git@0.0.0.0:2222/vole.git

A free and open-source forge that lives in your Git repo.

Stream run logs over the runner's SSH channel

5ece70a50ebda392626b92bdf29d28bc9b444954 · Keenan Nemetz · 10d ago · parent fdf2340f

Stream run logs over the runner's SSH channel

The runner wrote the log to a temp file and uploaded it as one LFS object
when the job ended, so a running job had no log to read and a log was stored
like a shared artifact even though it belongs to one attempt.

The runner sends 128 KiB chunks over the same runner channel that holds the
lease. The server appends them to data/logs/<repo>/<run>/<attempt>.log and
records the size while the run is still running. The finish record carries
the final size; the server checks the file length, syncs it, and stores the
result. A partial or failed stream leaves no log on the record.

Logs are no longer LFS objects: they are not shared, not content-addressed,
and the web serves the file directly, with range and step slices. GC no
longer treats them as LFS roots.

ci/agent/logstream.go +112

@@ -0,0 +1,112 @@
	1	package agent
	2
	3	// The attempt log is streamed to the server while the job runs, over the same
	4	// runner channel that carries leases and heartbeats. The spool file is the source:
	5	// the streamer reads new bytes from it and sends chunks; the file still bounds the
	6	// log and still holds the step offsets.
	7
	8	import (
	9	"errors"
	10	"fmt"
	11	"os"
	12	"time"
	13	)
	14
	15	const (
	16	// logChunk is how much log one message carries. The JSON encoding adds base64,
	17	// and the control channel's line bound has room for this plus a record.
	18	logChunk = 128 << 10
	19	// logPoll is how often the streamer looks for new bytes. It is a reading
	20	// latency, not a durability interval: the bytes are already in the file.
	21	logPoll = 100 * time.Millisecond
	22	)
	23
	24	type logStream struct {
	25	s *runnerSession
	26	job *runnerJob
	27	log *spool
	28	stopc chan struct{}
	29	done chan struct{}
	30	err error
	31	}
	32
	33	func (s runnerSession) startLogStream(job runnerJob, log spool) logStream {
	34	l := &logStream{s: s, job: job, log: log, stopc: make(chan struct{}), done: make(chan struct{})}
	35	go l.run()
	36	return l
	37	}
	38
	39	func (l *logStream) run() {
	40	defer close(l.done)
	41	f, err := os.Open(l.log.Path())
	42	if err != nil {
	43	l.err = err
	44	return
	45	}
	46	defer f.Close()
	47	ticker := time.NewTicker(logPoll)
	48	defer ticker.Stop()
	49	offset := int64(0)
	50	for {
	51	select {
	52	case <-l.stopc:
	53	l.send(f, &offset) // one last pass after seal wrote the truncation note
	54	return
	55	case <-ticker.C:
	56	if !l.send(f, &offset) {
	57	return
	58	}
	59	}
	60	}
	61	}
	62
	63	func (l logStream) send(f os.File, offset *int64) bool {
	64	size := l.log.Len()
	65	for *offset < size {
	66	n := min(int64(logChunk), size-*offset)
	67	buf := make([]byte, n)
	68	if _, err := f.ReadAt(buf, *offset); err != nil {
	69	l.err = err
	70	return false
	71	}
	72	rep, err := l.s.call(runnerRequest{
	73	Op: "log", Run: l.job.ID,
	74	Log: &runnerLogChunk{Attempt: l.job.Attempt, Offset: *offset, Data: buf},
	75	})
	76	if err != nil {
	77	l.err = err
	78	return false
	79	}
	80	if rep.Error != "" {
	81	l.err = errors.New(rep.Error)
	82	return false
	83	}
	84	if !rep.OK {
	85	l.err = errors.New("the server refused the log chunk")
	86	return false
	87	}
	88	if rep.LogSize != *offset+n {
	89	l.err = fmt.Errorf("the server has %d log bytes, expected %d", rep.LogSize, *offset+n)
	90	return false
	91	}
	92	*offset += n
	93	}
	94	return true
	95	}
	96
	97	// drain stops after the final bytes and returns the first error. finish calls it
	98	// after sealing the spool and before sending the record, so the record's log size
	99	// describes bytes the server already has.
	100	func (l *logStream) drain() error {
	101	l.stop()
	102	return l.err
	103	}
	104
	105	func (l *logStream) stop() {
	106	select {
	107	case <-l.stopc:
	108	default:
	109	close(l.stopc)
	110	}
	111	<-l.done
	112	}