diff --git a/ccrawl/journal.go b/ccrawl/journal.go new file mode 100644 index 0000000..c34f524 --- /dev/null +++ b/ccrawl/journal.go @@ -0,0 +1,547 @@ +package ccrawl + +import ( + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +// ProgressMode selects how a long running command narrates itself on stderr. +type ProgressMode string + +const ( + // ProgressText writes one human readable line per update. + ProgressText ProgressMode = "text" + // ProgressJSON writes the same events the journal records, one JSON object + // per line, so whatever collects the logs does not need a regex. + ProgressJSON ProgressMode = "json" + // ProgressNone says nothing. The journal is still written. + ProgressNone ProgressMode = "none" +) + +// ParseProgressMode resolves the --progress flag. An empty value picks text on a +// terminal and json anywhere else, which is what a log file or a supervisor +// wants without having to be told. +func ParseProgressMode(s string, tty bool) (ProgressMode, error) { + switch strings.ToLower(strings.TrimSpace(s)) { + case "", "auto": + if tty { + return ProgressText, nil + } + return ProgressJSON, nil + case "text": + return ProgressText, nil + case "json": + return ProgressJSON, nil + case "none", "off": + return ProgressNone, nil + default: + return "", fmt.Errorf("invalid --progress %q, want text, json, or none", s) + } +} + +// Event kinds written to the journal. +const ( + EventStart = "start" // the run began, with what it was asked to do + EventShard = "shard" // one shard finished, committed, skipped, or failed + EventItem = "item" // one file or record finished, for the streaming commands + EventPhase = "phase" // a multi-phase command entered a new phase + EventTick = "tick" // periodic snapshot of rates and resources + EventEnd = "end" // the run finished, with the final totals +) + +// Outcome values for the status field. +const ( + StatusOK = "ok" + StatusFailed = "failed" + StatusSkipped = "skipped" +) + +// RunEvent is one line of run.jsonl. +// +// Four fields are always there: ts, event, run and pipeline. The rest depend on +// the event. A shard or item event carries one unit of work's outcome, a tick +// carries the run's live rates and resource use, and start and end bracket the +// whole thing. Everything a long run needs to answer afterwards is a jq filter +// over this file: +// +// which shards failed +// jq -r 'select(.event=="shard" and .status=="failed") | "\(.shard) \(.error)"' run.jsonl +// when the rate dropped +// jq -r 'select(.event=="tick") | [.ts, .rate_per_hour, .inflight] | @tsv' run.jsonl +type RunEvent struct { + TS string `json:"ts"` + Event string `json:"event"` + Run string `json:"run"` + Pipeline string `json:"pipeline"` + + Crawl string `json:"crawl,omitempty"` + Phase string `json:"phase,omitempty"` + Name string `json:"name,omitempty"` // the file or URL an item event is about + + // Shard is a pointer because shard 0 is a real shard and omitempty would + // drop it from the line. + Shard *int `json:"shard,omitempty"` + Status string `json:"status,omitempty"` + Error string `json:"error,omitempty"` + + Rows int64 `json:"rows,omitempty"` + Bytes int64 `json:"bytes,omitempty"` + WARCBytes int64 `json:"warc_bytes,omitempty"` + HTMLBytes int64 `json:"html_bytes,omitempty"` + MDBytes int64 `json:"md_bytes,omitempty"` + ParquetBytes int64 `json:"parquet_bytes,omitempty"` + + ExtractS float64 `json:"extract_s,omitempty"` + FetchS float64 `json:"fetch_s,omitempty"` + ConvertS float64 `json:"convert_s,omitempty"` + ExportS float64 `json:"export_s,omitempty"` + PublishS float64 `json:"publish_s,omitempty"` + + Done int `json:"done,omitempty"` + Total int `json:"total,omitempty"` + Committed int `json:"committed,omitempty"` + Skipped int `json:"skipped,omitempty"` + Failed int `json:"failed,omitempty"` // units of work that failed, never URLs + Inflight int64 `json:"inflight,omitempty"` + // FetchFailed counts URLs a refetch shard could not fetch. It is deliberately + // not folded into Failed: a shard with a thousand dead hosts still succeeded. + FetchFailed int64 `json:"fetch_failed,omitempty"` + + Rate float64 `json:"rate_per_hour,omitempty"` + ETAS float64 `json:"eta_s,omitempty"` + ElapsedS float64 `json:"elapsed_s,omitempty"` + FreeDisk int64 `json:"free_disk_bytes,omitempty"` + RSS int64 `json:"rss_bytes,omitempty"` +} + +// Journal appends run events to a JSON Lines file. It is safe for concurrent +// use, and every event reaches the kernel before Write returns, so a run that is +// killed still leaves a journal that reads to the last thing that happened. +// +// A nil *Journal is a working no-op, so callers never have to check. +type Journal struct { + mu sync.Mutex + f *os.File + enc *json.Encoder + path string +} + +// OpenJournal opens the journal at path, creating the directory if it is not +// there and appending if the file already exists. Resuming a run appends to the +// same file; the run id on every event keeps the attempts apart. +func OpenJournal(path string) (*Journal, error) { + if path == "" { + return nil, nil + } + if dir := filepath.Dir(path); dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return nil, err + } + return &Journal{f: f, enc: json.NewEncoder(f), path: path}, nil +} + +// Write appends one event. +func (j *Journal) Write(ev RunEvent) error { + if j == nil { + return nil + } + j.mu.Lock() + defer j.mu.Unlock() + return j.enc.Encode(ev) +} + +// Path is where the journal is being written, or empty when there is none. +func (j *Journal) Path() string { + if j == nil { + return "" + } + return j.path +} + +// Close closes the underlying file. +func (j *Journal) Close() error { + if j == nil || j.f == nil { + return nil + } + return j.f.Close() +} + +// RunReporter fans one event out to the three places a long run reports to: the +// journal on disk, the progress stream on stderr, and the metrics a scraper +// reads. Commands report once and every sink gets the right shape. +// +// A nil *RunReporter is a working no-op. +type RunReporter struct { + pipeline string + mode ProgressMode + journal *Journal + metrics *Metrics + runID string + + mu sync.Mutex + out io.Writer +} + +// NewRunReporter builds a reporter for one command run. The journal and the +// metrics registry may both be nil, which turns those sinks off. +func NewRunReporter(pipeline string, mode ProgressMode, j *Journal, m *Metrics) *RunReporter { + return &RunReporter{ + pipeline: pipeline, + mode: mode, + journal: j, + metrics: m, + runID: newRunID(time.Now()), + out: os.Stderr, + } +} + +// newRunID names one attempt at a run. Wall-clock plus pid is enough to tell two +// attempts apart in a journal they both appended to, and it sorts. +func newRunID(t time.Time) string { + return t.UTC().Format("20060102T150405Z") + "-" + strconv.Itoa(os.Getpid()) +} + +// orDefault returns r, or a plain text reporter on stderr when the caller did +// not configure one. Library callers that never set a reporter keep the stderr +// narration the pipelines have always had. +func (r *RunReporter) orDefault(pipeline string) *RunReporter { + if r != nil { + return r + } + return NewRunReporter(pipeline, ProgressText, nil, nil) +} + +// SetOutput redirects the progress stream, for tests. +func (r *RunReporter) SetOutput(w io.Writer) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + r.out = w +} + +// RunID identifies this attempt, and is stamped on every event it writes. +func (r *RunReporter) RunID() string { + if r == nil { + return "" + } + return r.runID +} + +// JournalPath is where events are being recorded, or empty when nowhere. +func (r *RunReporter) JournalPath() string { + if r == nil { + return "" + } + return r.journal.Path() +} + +// Mode is the resolved progress mode. +func (r *RunReporter) Mode() ProgressMode { + if r == nil { + return ProgressNone + } + return r.mode +} + +// Metrics is the registry this reporter updates, or nil when metrics are off. +func (r *RunReporter) Metrics() *Metrics { + if r == nil { + return nil + } + return r.metrics +} + +// Event stamps an event and hands it to every sink. +func (r *RunReporter) Event(ev RunEvent) { + if r == nil { + return + } + ev.TS = time.Now().UTC().Format("2006-01-02T15:04:05.000Z") + ev.Run = r.runID + if ev.Pipeline == "" { + ev.Pipeline = r.pipeline + } + _ = r.journal.Write(ev) + r.record(ev) + if r.mode == ProgressJSON { + r.mu.Lock() + _ = json.NewEncoder(r.out).Encode(ev) + r.mu.Unlock() + } +} + +// Textf writes a human readable progress line, and only in text mode. The +// machine modes get the same information from the matching Event, so a caller +// reports each thing once and both modes come out right. +func (r *RunReporter) Textf(format string, a ...any) { + if r == nil || r.mode != ProgressText { + return + } + r.mu.Lock() + defer r.mu.Unlock() + fmt.Fprintf(r.out, format, a...) +} + +// record folds an event into the metrics registry. +func (r *RunReporter) record(ev RunEvent) { + m := r.metrics + if m == nil { + return + } + p := ev.Pipeline + switch ev.Event { + case EventShard, EventItem: + status := ev.Status + if status == "" { + status = StatusOK + } + unit := MetricShards + if ev.Event == EventItem { + unit = MetricItems + } + m.Add(unit, 1, "pipeline", p, "status", status) + if ev.Rows > 0 { + m.Add(MetricRows, float64(ev.Rows), "pipeline", p) + } + for _, b := range []struct { + kind string + n int64 + }{ + {"fetched", ev.Bytes}, + {"warc", ev.WARCBytes}, + {"html", ev.HTMLBytes}, + {"markdown", ev.MDBytes}, + {"parquet", ev.ParquetBytes}, + } { + if b.n > 0 { + m.Add(MetricBytes, float64(b.n), "pipeline", p, "kind", b.kind) + } + } + for _, d := range []struct { + phase string + s float64 + }{ + {"extract", ev.ExtractS}, + {"fetch", ev.FetchS}, + {"convert", ev.ConvertS}, + {"export", ev.ExportS}, + {"publish", ev.PublishS}, + } { + if d.s > 0 { + m.Observe(MetricPhaseSeconds, d.s, "pipeline", p, "phase", d.phase) + } + } + case EventTick, EventEnd: + m.Set(MetricInflight, float64(ev.Inflight), "pipeline", p) + m.Set(MetricRate, ev.Rate, "pipeline", p) + m.Set(MetricElapsed, ev.ElapsedS, "pipeline", p) + m.Set(MetricDone, float64(ev.Done), "pipeline", p) + if ev.Total > 0 { + m.Set(MetricTotal, float64(ev.Total), "pipeline", p) + } + if ev.FreeDisk > 0 { + m.Set(MetricFreeDisk, float64(ev.FreeDisk), "pipeline", p) + } + if ev.RSS > 0 { + m.Set(MetricRSS, float64(ev.RSS)) + } + } +} + +// Close releases the journal file. +func (r *RunReporter) Close() error { + if r == nil { + return nil + } + return r.journal.Close() +} + +// DefaultTickInterval is how often a run writes a tick event when the caller +// does not pick something else. Half a minute is often enough to see a stall on +// a 72 hour run and rare enough that the journal stays small. +const DefaultTickInterval = 30 * time.Second + +// StreamProgress reports a counting command (download, export, host enrich, +// crawl seed) on the same schedule and in the same event shape as the shard +// pipelines, so one jq query works across all of them. Every method is safe to +// call from several goroutines. +type StreamProgress struct { + r *RunReporter + unit string // what one item is, for the text line: "files", "records" + start time.Time + interval time.Duration + + // total is atomic because a command that has to read its input before it + // knows the size of the job (export reading locations on stdin) fills it in + // after the ticker is already running. + total atomic.Int64 + done atomic.Int64 + failed atomic.Int64 + bytes atomic.Int64 + rows atomic.Int64 + phase atomic.Pointer[string] + + stopOnce sync.Once + stop chan struct{} + ticked chan struct{} +} + +// StartStreamProgress writes the start event and begins ticking. total is 0 when +// the size of the job is not known up front, which turns the ETA off. The caller +// must call Stop, which writes the end event. +func StartStreamProgress(r *RunReporter, unit string, total int, interval time.Duration) *StreamProgress { + if interval <= 0 { + interval = DefaultTickInterval + } + p := &StreamProgress{ + r: r.orDefault("stream"), + unit: unit, + start: time.Now(), + interval: interval, + stop: make(chan struct{}), + ticked: make(chan struct{}), + } + p.total.Store(int64(total)) + p.r.Event(RunEvent{Event: EventStart, Total: total}) + go p.loop() + return p +} + +// SetTotal fills in the size of the job for a command that only learns it after +// the run has started, which turns the ETA on from the next tick. +func (p *StreamProgress) SetTotal(total int) { + if p == nil { + return + } + p.total.Store(int64(total)) +} + +// Add records progress: n items finished, carrying rows and bytes between them. +// It does not write an event, so a command that handles millions of records can +// call it per record and still get a journal of a sane size. +func (p *StreamProgress) Add(n, rows, bytes int64) { + if p == nil { + return + } + p.done.Add(n) + p.rows.Add(rows) + p.bytes.Add(bytes) +} + +// Item records one named unit of work, with an event of its own. Use it where +// the count is bounded (files in a download) and skip it where it is not. +func (p *StreamProgress) Item(name string, bytes int64, err error) { + if p == nil { + return + } + ev := RunEvent{Event: EventItem, Name: name, Bytes: bytes, Status: StatusOK} + if err != nil { + ev.Status = StatusFailed + ev.Error = err.Error() + p.failed.Add(1) + } else { + p.done.Add(1) + p.bytes.Add(bytes) + } + ev.Done = int(p.done.Load()) + ev.Total = int(p.total.Load()) + p.r.Event(ev) +} + +// Fail counts one failure without an item event. +func (p *StreamProgress) Fail() { + if p == nil { + return + } + p.failed.Add(1) +} + +// Phase names the stage the command is in, for the commands that have more than +// one. It writes a phase event and shows up on every tick after it. +func (p *StreamProgress) Phase(name string) { + if p == nil { + return + } + p.phase.Store(&name) + p.r.Event(RunEvent{Event: EventPhase, Phase: name, Done: int(p.done.Load())}) + p.r.Textf("%s\n", name) +} + +// Stop ends the ticker and writes the end event. It is safe to call twice. +func (p *StreamProgress) Stop() { + if p == nil { + return + } + p.stopOnce.Do(func() { + close(p.stop) + <-p.ticked + p.emit(EventEnd) + }) +} + +func (p *StreamProgress) loop() { + defer close(p.ticked) + t := time.NewTicker(p.interval) + defer t.Stop() + for { + select { + case <-p.stop: + return + case <-t.C: + p.emit(EventTick) + } + } +} + +// emit builds one tick or end event from the live counters. +func (p *StreamProgress) emit(kind string) { + done := p.done.Load() + total := int(p.total.Load()) + elapsed := time.Since(p.start) + ev := RunEvent{ + Event: kind, + Done: int(done), + Total: total, + Failed: int(p.failed.Load()), + Rows: p.rows.Load(), + Bytes: p.bytes.Load(), + ElapsedS: elapsed.Seconds(), + RSS: currentRSSBytes(), + } + if ph := p.phase.Load(); ph != nil { + ev.Phase = *ph + } + if elapsed > 0 { + ev.Rate = float64(done) / elapsed.Hours() + } + if total > 0 && ev.Rate > 0 && int(done) < total { + ev.ETAS = float64(total-int(done)) / ev.Rate * 3600 + } + p.r.Event(ev) + + of := "" + if total > 0 { + of = fmt.Sprintf("/%d", total) + } + eta := "" + if ev.ETAS > 0 { + eta = " | ETA " + fmtETA(time.Duration(ev.ETAS*float64(time.Second))) + } + p.r.Textf("%d%s %s | %s | %.0f %s/hour%s\n", + done, of, p.unit, fmtBytes(ev.Bytes), ev.Rate, p.unit, eta) +} diff --git a/ccrawl/journal_test.go b/ccrawl/journal_test.go new file mode 100644 index 0000000..27e0416 --- /dev/null +++ b/ccrawl/journal_test.go @@ -0,0 +1,348 @@ +package ccrawl + +import ( + "bufio" + "bytes" + "encoding/json" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + "time" +) + +func TestParseProgressMode(t *testing.T) { + cases := []struct { + in string + tty bool + want ProgressMode + bad bool + }{ + {in: "", tty: true, want: ProgressText}, + {in: "", tty: false, want: ProgressJSON}, + {in: "auto", tty: true, want: ProgressText}, + {in: "AUTO", tty: false, want: ProgressJSON}, + {in: "text", tty: false, want: ProgressText}, + {in: " json ", tty: true, want: ProgressJSON}, + {in: "none", tty: true, want: ProgressNone}, + {in: "off", tty: true, want: ProgressNone}, + {in: "verbose", bad: true}, + } + for _, c := range cases { + got, err := ParseProgressMode(c.in, c.tty) + if c.bad { + if err == nil { + t.Fatalf("ParseProgressMode(%q) = %q, want an error", c.in, got) + } + continue + } + if err != nil { + t.Fatalf("ParseProgressMode(%q, %v): %v", c.in, c.tty, err) + } + if got != c.want { + t.Fatalf("ParseProgressMode(%q, %v) = %q, want %q", c.in, c.tty, got, c.want) + } + } +} + +// readJournal parses every line of a journal file back into events. +func readJournal(t *testing.T, path string) []RunEvent { + t.Helper() + f, err := os.Open(path) + if err != nil { + t.Fatalf("open journal: %v", err) + } + defer func() { _ = f.Close() }() + var out []RunEvent + sc := bufio.NewScanner(f) + for sc.Scan() { + var ev RunEvent + if err := json.Unmarshal(sc.Bytes(), &ev); err != nil { + t.Fatalf("journal line %q: %v", sc.Text(), err) + } + out = append(out, ev) + } + if err := sc.Err(); err != nil { + t.Fatalf("scan journal: %v", err) + } + return out +} + +func TestJournalWritesOneLinePerEvent(t *testing.T) { + // A nested directory, so the mkdir path is covered too. + path := filepath.Join(t.TempDir(), "runs", "run.jsonl") + j, err := OpenJournal(path) + if err != nil { + t.Fatalf("OpenJournal: %v", err) + } + rep := NewRunReporter("markdown export", ProgressNone, j, nil) + + shard := 0 + rep.Event(RunEvent{Event: EventStart, Total: 2}) + rep.Event(RunEvent{Event: EventShard, Shard: &shard, Status: StatusOK, Rows: 1200}) + shard2 := 7 + rep.Event(RunEvent{Event: EventShard, Shard: &shard2, Status: StatusFailed, Error: "connection reset"}) + rep.Event(RunEvent{Event: EventEnd, Done: 1, Failed: 1}) + if err := rep.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + evs := readJournal(t, path) + if len(evs) != 4 { + t.Fatalf("journal has %d events, want 4", len(evs)) + } + for _, ev := range evs { + if ev.TS == "" || ev.Run == "" { + t.Fatalf("event %q missing ts or run: %+v", ev.Event, ev) + } + if ev.Pipeline != "markdown export" { + t.Fatalf("event %q pipeline = %q, want the reporter's", ev.Event, ev.Pipeline) + } + if ev.Run != rep.RunID() { + t.Fatalf("event %q run = %q, want %q", ev.Event, ev.Run, rep.RunID()) + } + } + + // This is the query the runbook tells an operator to reach for. + if evs[2].Event != EventShard || evs[2].Status != StatusFailed || evs[2].Error != "connection reset" { + t.Fatalf("failed shard event wrong: %+v", evs[2]) + } + if evs[2].Shard == nil || *evs[2].Shard != 7 { + t.Fatalf("failed shard index wrong: %+v", evs[2].Shard) + } + // Shard 0 is a real shard, so it has to survive the round trip. + if evs[1].Shard == nil || *evs[1].Shard != 0 { + t.Fatalf("shard 0 dropped from the line: %+v", evs[1]) + } +} + +func TestJournalAppendsAcrossRuns(t *testing.T) { + path := filepath.Join(t.TempDir(), "run.jsonl") + var ids []string + for i := 0; i < 2; i++ { + j, err := OpenJournal(path) + if err != nil { + t.Fatalf("OpenJournal: %v", err) + } + rep := NewRunReporter("refetch", ProgressNone, j, nil) + rep.Event(RunEvent{Event: EventStart}) + ids = append(ids, rep.RunID()) + if err := rep.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + } + evs := readJournal(t, path) + if len(evs) != 2 { + t.Fatalf("resumed journal has %d events, want 2", len(evs)) + } + if evs[0].Run != ids[0] || evs[1].Run != ids[1] { + t.Fatalf("run ids not carried through: %q %q", evs[0].Run, evs[1].Run) + } +} + +func TestNilJournalAndReporterAreNoOps(t *testing.T) { + var j *Journal + if err := j.Write(RunEvent{Event: EventTick}); err != nil { + t.Fatalf("nil journal Write: %v", err) + } + if p := j.Path(); p != "" { + t.Fatalf("nil journal Path = %q, want empty", p) + } + if err := j.Close(); err != nil { + t.Fatalf("nil journal Close: %v", err) + } + + opened, err := OpenJournal("") + if err != nil || opened != nil { + t.Fatalf("OpenJournal(\"\") = %v, %v, want nil, nil", opened, err) + } + + var r *RunReporter + r.Event(RunEvent{Event: EventEnd}) + r.Textf("nothing %d\n", 1) + if r.Mode() != ProgressNone || r.RunID() != "" || r.JournalPath() != "" || r.Metrics() != nil { + t.Fatal("nil reporter did not behave as a no-op") + } + if got := r.orDefault("stream"); got == nil || got.Mode() != ProgressText { + t.Fatal("orDefault did not build a text reporter") + } +} + +func TestReporterTextModeWritesOnlyText(t *testing.T) { + var buf bytes.Buffer + rep := NewRunReporter("download", ProgressText, nil, nil) + rep.SetOutput(&buf) + rep.Event(RunEvent{Event: EventItem, Name: "a.warc.gz"}) + rep.Textf("[1/2] ok a.warc.gz\n") + if got := buf.String(); got != "[1/2] ok a.warc.gz\n" { + t.Fatalf("text mode wrote %q", got) + } + + buf.Reset() + rep = NewRunReporter("download", ProgressJSON, nil, nil) + rep.SetOutput(&buf) + rep.Textf("this line is for humans\n") + rep.Event(RunEvent{Event: EventItem, Name: "a.warc.gz"}) + out := buf.String() + if strings.Contains(out, "for humans") { + t.Fatalf("json mode leaked a text line: %q", out) + } + var ev RunEvent + if err := json.Unmarshal([]byte(strings.TrimSpace(out)), &ev); err != nil { + t.Fatalf("json mode line %q: %v", out, err) + } + if ev.Name != "a.warc.gz" || ev.Pipeline != "download" { + t.Fatalf("json mode event wrong: %+v", ev) + } + + buf.Reset() + rep = NewRunReporter("download", ProgressNone, nil, nil) + rep.SetOutput(&buf) + rep.Textf("quiet\n") + rep.Event(RunEvent{Event: EventItem}) + if buf.Len() != 0 { + t.Fatalf("none mode wrote %q", buf.String()) + } +} + +func TestStreamProgressCountsAndBrackets(t *testing.T) { + path := filepath.Join(t.TempDir(), "run.jsonl") + j, err := OpenJournal(path) + if err != nil { + t.Fatalf("OpenJournal: %v", err) + } + rep := NewRunReporter("download", ProgressNone, j, NewMetrics()) + + // A tick interval short enough that the run cannot finish without one. + sp := StartStreamProgress(rep, "files", 3, 10*time.Millisecond) + sp.Item("a.warc.gz", 100, nil) + sp.Item("b.warc.gz", 0, os.ErrNotExist) + sp.Add(1, 42, 7) + time.Sleep(40 * time.Millisecond) + sp.Stop() + sp.Stop() // idempotent, and must not write a second end event + if err := rep.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + evs := readJournal(t, path) + var start, end, ticks int + for _, ev := range evs { + switch ev.Event { + case EventStart: + start++ + if ev.Total != 3 { + t.Fatalf("start total = %d, want 3", ev.Total) + } + case EventEnd: + end++ + if ev.Done != 2 || ev.Failed != 1 { + t.Fatalf("end done=%d failed=%d, want 2 and 1", ev.Done, ev.Failed) + } + if ev.Rows != 42 || ev.Bytes != 107 { + t.Fatalf("end rows=%d bytes=%d, want 42 and 107", ev.Rows, ev.Bytes) + } + if ev.ElapsedS <= 0 { + t.Fatalf("end elapsed_s = %v, want positive", ev.ElapsedS) + } + case EventTick: + ticks++ + } + } + if start != 1 || end != 1 { + t.Fatalf("start=%d end=%d, want one of each", start, end) + } + if ticks == 0 { + t.Fatal("no tick events, the ticker never ran") + } + + m := rep.Metrics() + if got := m.Value(MetricItems, "pipeline", "download", "status", StatusOK); got != 1 { + t.Fatalf("ok items = %v, want 1", got) + } + if got := m.Value(MetricItems, "pipeline", "download", "status", StatusFailed); got != 1 { + t.Fatalf("failed items = %v, want 1", got) + } +} + +func TestStreamProgressSetTotalTurnsOnETA(t *testing.T) { + path := filepath.Join(t.TempDir(), "run.jsonl") + j, err := OpenJournal(path) + if err != nil { + t.Fatalf("OpenJournal: %v", err) + } + rep := NewRunReporter("export", ProgressNone, j, nil) + + sp := StartStreamProgress(rep, "records", 0, time.Hour) + sp.SetTotal(100) + sp.Add(1, 0, 0) + sp.Stop() + if err := rep.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + evs := readJournal(t, path) + end := evs[len(evs)-1] + if end.Event != EventEnd { + t.Fatalf("last event is %q, want end", end.Event) + } + if end.Total != 100 { + t.Fatalf("end total = %d, want the total set after the start", end.Total) + } + if end.ETAS <= 0 { + t.Fatalf("end eta_s = %v, want positive once the total is known", end.ETAS) + } +} + +func TestStreamProgressPhaseEvents(t *testing.T) { + path := filepath.Join(t.TempDir(), "run.jsonl") + j, err := OpenJournal(path) + if err != nil { + t.Fatalf("OpenJournal: %v", err) + } + rep := NewRunReporter("host enrich", ProgressNone, j, nil) + + sp := StartStreamProgress(rep, "hosts", 0, time.Hour) + sp.Phase("vertices") + sp.Add(2, 2, 0) + sp.Phase("rank join") + sp.Stop() + if err := rep.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + evs := readJournal(t, path) + var phases []string + for _, ev := range evs { + if ev.Event == EventPhase { + phases = append(phases, ev.Phase) + } + } + if len(phases) != 2 || phases[0] != "vertices" || phases[1] != "rank join" { + t.Fatalf("phase events = %v, want vertices then rank join", phases) + } + // The phase sticks, so a tick or end event says where a stalled run is. + if end := evs[len(evs)-1]; end.Phase != "rank join" { + t.Fatalf("end phase = %q, want the last phase entered", end.Phase) + } +} + +func TestCurrentRSSBytes(t *testing.T) { + got := currentRSSBytes() + switch runtime.GOOS { + case "linux", "darwin": + if got <= 0 { + t.Fatalf("currentRSSBytes on %s = %d, want positive", runtime.GOOS, got) + } + // A Go test binary is megabytes, not gigabytes. This catches a unit + // mix-up, which is the failure this reader is actually prone to. + if got < 1<<20 || got > 8<<30 { + t.Fatalf("currentRSSBytes = %d bytes, outside a believable range", got) + } + default: + if got != 0 { + t.Fatalf("currentRSSBytes on %s = %d, want 0", runtime.GOOS, got) + } + } +} diff --git a/ccrawl/markdown_run.go b/ccrawl/markdown_run.go index ffa9550..ea010a2 100644 --- a/ccrawl/markdown_run.go +++ b/ccrawl/markdown_run.go @@ -9,6 +9,7 @@ import ( "runtime" "strconv" "sync" + "sync/atomic" "time" ) @@ -112,6 +113,14 @@ type MarkdownExportConfig struct { // Progress is called once per committed batch with a snapshot of the run. // It may be nil. Progress func(MarkdownRunStats) + + // Reporter records the run: one event per shard, one per tick, and the start + // and end. Nil narrates to stderr in text and records nothing, which is what + // the library did before there was a journal. + Reporter *RunReporter + // TickInterval is how often a tick event is written. 0 selects + // DefaultTickInterval. + TickInterval time.Duration } // MarkdownRunStats is a live snapshot of a parallel export run. @@ -131,6 +140,7 @@ type MarkdownRunStats struct { ShardsPerHour float64 ETA time.Duration // estimated time to finish the remaining shards FreeDiskBytes int64 + RSSBytes int64 // resident set size at the last progress update } // packShardFn is the function the orchestrator uses to convert one shard. It @@ -174,6 +184,7 @@ func RunMarkdownExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Mar if minFree <= 0 { minFree = 2 << 30 } + cfg.Reporter = cfg.Reporter.orDefault("markdown export") if err := os.MkdirAll(cfg.OutDir, 0o755); err != nil { return MarkdownRunStats{}, err @@ -186,13 +197,48 @@ func RunMarkdownExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Mar start := time.Now() run := MarkdownRunStats{Total: len(cfg.Indices)} + cfg.Reporter.Event(RunEvent{ + Event: EventStart, + Crawl: cfg.CrawlID, + Total: run.Total, + }) + + // The committer owns run, so the ticker cannot read it directly. It publishes + // a copy after every change instead, which the ticker reads with no lock and + // no chance of catching a half-updated struct. + var snap atomic.Pointer[MarkdownRunStats] + snap.Store(&MarkdownRunStats{Total: run.Total}) + var inflight atomic.Int64 + // Committer: the only goroutine that touches run's cumulative fields and the // ledger, so no extra locking is needed there. committerDone := make(chan struct{}) var commitErr error go func() { defer close(committerDone) - commitErr = runCommitter(ctx, hf, cfg, k, start, finished, &run) + commitErr = runCommitter(ctx, hf, cfg, k, start, finished, &run, &snap) + }() + + // Ticker: a periodic heartbeat so a stalled run shows up in the journal even + // when no shard has finished for an hour. + stopTicks := make(chan struct{}) + ticksDone := make(chan struct{}) + go func() { + defer close(ticksDone) + interval := cfg.TickInterval + if interval <= 0 { + interval = DefaultTickInterval + } + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-stopTicks: + return + case <-t.C: + cfg.Reporter.Event(markdownTickEvent(EventTick, snap.Load(), cfg, inflight.Load(), time.Since(start))) + } + } }() // Shard workers. @@ -208,6 +254,7 @@ func RunMarkdownExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Mar waitForDisk(ctx, cfg.OutDir, minFree) outPath := filepath.Join(cfg.OutDir, fmt.Sprintf("%06d.parquet", idx)) t0 := time.Now() + inflight.Add(1) stats, err := packShardFn(ctx, h, MarkdownPackConfig{ CrawlID: cfg.CrawlID, ShardIdx: idx, @@ -216,6 +263,7 @@ func RunMarkdownExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Mar Workers: c, ConvertSem: sem, }) + inflight.Add(-1) // Per-shard wall-clock is the useful convert figure for a parallel // run; the streamed download is folded into it. stats.DurConvert = time.Since(t0) @@ -248,24 +296,73 @@ func RunMarkdownExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Mar wg.Wait() close(finished) <-committerDone + close(stopTicks) + <-ticksDone run.Elapsed = time.Since(start) run.FreeDiskBytes = freeDiskBytes(cfg.OutDir) + run.RSSBytes = currentRSSBytes() + cfg.Reporter.Event(markdownTickEvent(EventEnd, &run, cfg, 0, run.Elapsed)) if commitErr != nil { return run, commitErr } return run, ctx.Err() } +// markdownTickEvent renders a run snapshot as a tick or end event. Free disk, +// RSS and the elapsed clock are sampled here rather than taken from the +// snapshot: a tick is exactly the moment to ask, and the snapshot's copies are +// as old as the last committed batch, which on a stalled run is the thing you +// are trying to see. +func markdownTickEvent(kind string, s *MarkdownRunStats, cfg MarkdownExportConfig, inflight int64, elapsed time.Duration) RunEvent { + if s == nil { + s = &MarkdownRunStats{} + } + rate := 0.0 + if elapsed > 0 { + rate = float64(s.Committed) / elapsed.Hours() + } + eta := 0.0 + if remaining := s.Total - s.Skipped - s.Committed - s.Failed; remaining > 0 && rate > 0 { + eta = float64(remaining) / rate * 3600 + } + return RunEvent{ + Event: kind, + Crawl: cfg.CrawlID, + Done: s.Committed + s.Skipped + s.Failed, + Total: s.Total, + Committed: s.Committed, + Skipped: s.Skipped, + Failed: s.Failed, + Inflight: inflight, + Rows: s.Rows, + WARCBytes: s.WARCBytes, + HTMLBytes: s.HTMLBytes, + MDBytes: s.MDBytes, + ParquetBytes: s.ParquetBytes, + Rate: rate, + ETAS: eta, + ElapsedS: elapsed.Seconds(), + FreeDisk: freeDiskBytes(cfg.OutDir), + RSS: currentRSSBytes(), + } +} + // errAlreadyDone marks a shard the ledger already recorded so the committer can // count it as skipped without trying to commit it. var errAlreadyDone = fmt.Errorf("already committed") // runCommitter drains finished shards, batches K parquets per HF commit, deletes // the local files, records the ledger, and reports progress with an ETA. -func runCommitter(ctx context.Context, hf *HFClient, cfg MarkdownExportConfig, k int, start time.Time, finished <-chan shardResult, run *MarkdownRunStats) error { +func runCommitter(ctx context.Context, hf *HFClient, cfg MarkdownExportConfig, k int, start time.Time, finished <-chan shardResult, run *MarkdownRunStats, snap *atomic.Pointer[MarkdownRunStats]) error { var batch []shardResult + // publish hands the ticker a stable copy of the run so far. + publish := func() { + s := *run + snap.Store(&s) + } + flush := func() error { if len(batch) == 0 { return nil @@ -333,10 +430,30 @@ func runCommitter(ctx context.Context, hf *HFClient, cfg MarkdownExportConfig, k } } run.Committed += len(batch) + + // The shard events go out after the commit, so a shard event in the + // journal means the parquet is on the hub, not that it was written + // locally. + for _, r := range batch { + idx := r.idx + cfg.Reporter.Event(RunEvent{ + Event: EventShard, + Crawl: cfg.CrawlID, + Shard: &idx, + Status: StatusOK, + Rows: r.stats.Rows, + WARCBytes: r.stats.WARCBytes, + HTMLBytes: r.stats.HTMLBytes, + MDBytes: r.stats.MDBytes, + ParquetBytes: r.stats.ParquetBytes, + ConvertS: r.stats.DurConvert.Seconds(), + }) + } batch = batch[:0] updateRunRates(run, start) logProgress(cfg, run) + publish() if cfg.Progress != nil { cfg.Progress(*run) } @@ -344,12 +461,20 @@ func runCommitter(ctx context.Context, hf *HFClient, cfg MarkdownExportConfig, k } for r := range finished { + idx := r.idx switch { case r.err == errAlreadyDone: run.Skipped++ + cfg.Reporter.Event(RunEvent{Event: EventShard, Crawl: cfg.CrawlID, Shard: &idx, Status: StatusSkipped}) + publish() case r.err != nil: run.Failed++ - fmt.Fprintf(os.Stderr, "markdown: shard %06d failed: %v\n", r.idx, r.err) + cfg.Reporter.Event(RunEvent{ + Event: EventShard, Crawl: cfg.CrawlID, Shard: &idx, + Status: StatusFailed, Error: r.err.Error(), + }) + cfg.Reporter.Textf("markdown: shard %06d failed: %v\n", r.idx, r.err) + publish() default: batch = append(batch, r) if len(batch) >= k { @@ -385,7 +510,8 @@ func logProgress(cfg MarkdownExportConfig, run *MarkdownRunStats) { pct = float64(done) / float64(run.Total) * 100 } run.FreeDiskBytes = freeDiskBytes(cfg.OutDir) - fmt.Fprintf(os.Stderr, + run.RSSBytes = currentRSSBytes() + cfg.Reporter.Textf( "markdown: %d/%d shards (%.1f%%) | %.1f rows total | %.1f shards/hour | ETA %s | disk free %s\n", done, run.Total, pct, float64(run.Rows), run.ShardsPerHour, fmtETA(run.ETA), fmtBytes(run.FreeDiskBytes)) diff --git a/ccrawl/metrics.go b/ccrawl/metrics.go new file mode 100644 index 0000000..1a4de3c --- /dev/null +++ b/ccrawl/metrics.go @@ -0,0 +1,278 @@ +package ccrawl + +import ( + "fmt" + "io" + "net" + "net/http" + "sort" + "strconv" + "strings" + "sync" + "time" +) + +// The metric names a run publishes. They are constants because the reporter and +// the registry have to agree on the spelling, and a test walks this list to +// check that every name the reporter writes is one the registry knows. +const ( + MetricShards = "ccrawl_shards_total" + MetricItems = "ccrawl_items_total" + MetricRows = "ccrawl_rows_total" + MetricBytes = "ccrawl_bytes_total" + MetricPhaseSeconds = "ccrawl_phase_duration_seconds" + MetricInflight = "ccrawl_inflight" + MetricDone = "ccrawl_done" + MetricTotal = "ccrawl_total" + MetricRate = "ccrawl_rate_per_hour" + MetricElapsed = "ccrawl_run_elapsed_seconds" + MetricFreeDisk = "ccrawl_free_disk_bytes" + MetricRSS = "ccrawl_rss_bytes" +) + +// phaseBuckets covers a shard phase from a second to an hour. A markdown shard +// converts in a couple of minutes and a refetch shard can take most of an hour, +// so the tail matters more than fine resolution at the short end. +var phaseBuckets = []float64{1, 5, 15, 30, 60, 120, 300, 600, 1200, 1800, 3600} + +// Metrics is a small Prometheus registry: counters, gauges and histograms +// rendered in the text exposition format. +// +// client_golang would do this too, and it would do more of it. It is a large +// dependency for twelve metrics in a binary whose selling point is that it has +// no dependencies at all, and the exposition format is a dozen lines of +// printing, so we write those instead. +// +// A nil *Metrics is a working no-op. +type Metrics struct { + mu sync.Mutex + defs map[string]*metricDef + order []string +} + +type metricDef struct { + name string + help string + typ string // counter, gauge, histogram + buckets []float64 + series map[string]*metricSeries + order []string +} + +type metricSeries struct { + value float64 // counters and gauges + counts []uint64 // histogram: one per bucket, plus the +Inf slot + sum float64 // histogram + count uint64 // histogram +} + +// NewMetrics returns the registry a run publishes, with every metric declared so +// the exposition carries the right HELP and TYPE from the first scrape, before +// anything has happened. +func NewMetrics() *Metrics { + m := &Metrics{defs: map[string]*metricDef{}} + m.register(MetricShards, "Shards finished, by pipeline and outcome.", "counter", nil) + m.register(MetricItems, "Files or records finished, by pipeline and outcome.", "counter", nil) + m.register(MetricRows, "Dataset rows written.", "counter", nil) + m.register(MetricBytes, "Bytes moved, by pipeline and kind.", "counter", nil) + m.register(MetricPhaseSeconds, "Wall-clock seconds per shard phase.", "histogram", phaseBuckets) + m.register(MetricInflight, "Units of work in flight right now.", "gauge", nil) + m.register(MetricDone, "Units of work finished so far.", "gauge", nil) + m.register(MetricTotal, "Units of work this run was asked to do.", "gauge", nil) + m.register(MetricRate, "Units of work finished per hour, averaged over the run.", "gauge", nil) + m.register(MetricElapsed, "Seconds since the run started.", "gauge", nil) + m.register(MetricFreeDisk, "Free bytes on the filesystem the run writes to.", "gauge", nil) + m.register(MetricRSS, "Resident set size of the ccrawl process, in bytes.", "gauge", nil) + return m +} + +func (m *Metrics) register(name, help, typ string, buckets []float64) { + m.defs[name] = &metricDef{ + name: name, + help: help, + typ: typ, + buckets: buckets, + series: map[string]*metricSeries{}, + } + m.order = append(m.order, name) +} + +// Add increments a counter. Labels are alternating names and values. +func (m *Metrics) Add(name string, v float64, labels ...string) { + m.with(name, labels, func(s *metricSeries) { s.value += v }) +} + +// Set writes a gauge. +func (m *Metrics) Set(name string, v float64, labels ...string) { + m.with(name, labels, func(s *metricSeries) { s.value = v }) +} + +// Observe records one histogram sample. +func (m *Metrics) Observe(name string, v float64, labels ...string) { + m.with(name, labels, func(s *metricSeries) { + d := m.defs[name] + for i, b := range d.buckets { + if v <= b { + s.counts[i]++ + } + } + s.counts[len(d.buckets)]++ // +Inf + s.sum += v + s.count++ + }) +} + +// Value reads one series back, for tests and for the summary a run prints. +func (m *Metrics) Value(name string, labels ...string) float64 { + if m == nil { + return 0 + } + m.mu.Lock() + defer m.mu.Unlock() + d := m.defs[name] + if d == nil { + return 0 + } + s := d.series[renderLabels(labels)] + if s == nil { + return 0 + } + if d.typ == "histogram" { + return s.sum + } + return s.value +} + +// with locates (or creates) one series and applies f under the lock. An unknown +// metric name is dropped: every name is a constant in this file, so a miss is a +// typo, and a typo should not take a 72 hour run down. +func (m *Metrics) with(name string, labels []string, f func(*metricSeries)) { + if m == nil { + return + } + m.mu.Lock() + defer m.mu.Unlock() + d := m.defs[name] + if d == nil { + return + } + key := renderLabels(labels) + s := d.series[key] + if s == nil { + s = &metricSeries{} + if d.typ == "histogram" { + s.counts = make([]uint64, len(d.buckets)+1) + } + d.series[key] = s + d.order = append(d.order, key) + sort.Strings(d.order) + } + f(s) +} + +// renderLabels turns alternating names and values into the exposition label set, +// sorted by name so the same labels always render the same way. An odd trailing +// name is dropped rather than printed with a missing value. +func renderLabels(labels []string) string { + if len(labels) < 2 { + return "" + } + pairs := make([]string, 0, len(labels)/2) + for i := 0; i+1 < len(labels); i += 2 { + pairs = append(pairs, labels[i]+`="`+escapeLabel(labels[i+1])+`"`) + } + sort.Strings(pairs) + return strings.Join(pairs, ",") +} + +// escapeLabel escapes the three characters the exposition format reserves inside +// a label value. +func escapeLabel(v string) string { + return strings.NewReplacer(`\`, `\\`, `"`, `\"`, "\n", `\n`).Replace(v) +} + +// WriteTo renders the registry in the Prometheus text exposition format. +func (m *Metrics) WriteTo(w io.Writer) (int64, error) { + if m == nil { + return 0, nil + } + m.mu.Lock() + defer m.mu.Unlock() + + var b strings.Builder + for _, name := range m.order { + d := m.defs[name] + if len(d.series) == 0 { + continue + } + fmt.Fprintf(&b, "# HELP %s %s\n# TYPE %s %s\n", d.name, d.help, d.name, d.typ) + for _, key := range d.order { + s := d.series[key] + if d.typ != "histogram" { + fmt.Fprintf(&b, "%s%s %s\n", d.name, braces(key), formatFloat(s.value)) + continue + } + for i, bucket := range d.buckets { + fmt.Fprintf(&b, "%s_bucket%s %d\n", d.name, + braces(withLabel(key, "le", formatFloat(bucket))), s.counts[i]) + } + fmt.Fprintf(&b, "%s_bucket%s %d\n", d.name, + braces(withLabel(key, "le", "+Inf")), s.counts[len(d.buckets)]) + fmt.Fprintf(&b, "%s_sum%s %s\n", d.name, braces(key), formatFloat(s.sum)) + fmt.Fprintf(&b, "%s_count%s %d\n", d.name, braces(key), s.count) + } + } + n, err := io.WriteString(w, b.String()) + return int64(n), err +} + +func braces(labels string) string { + if labels == "" { + return "" + } + return "{" + labels + "}" +} + +// withLabel appends one label to a rendered set. le goes last by convention, and +// Prometheus does not care about label order on the wire. +func withLabel(labels, name, value string) string { + pair := name + `="` + escapeLabel(value) + `"` + if labels == "" { + return pair + } + return labels + "," + pair +} + +func formatFloat(v float64) string { + return strconv.FormatFloat(v, 'g', -1, 64) +} + +// Handler serves the registry at whatever path it is mounted on. +func (m *Metrics) Handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + _, _ = m.WriteTo(w) + }) +} + +// ServeMetrics starts an HTTP server exposing m at /metrics on addr. It binds +// before returning, so a busy port or a bad address fails the command at +// startup instead of silently in a goroutine an hour later. The caller closes +// the returned server when the run ends. +// +// The returned server's Addr is the address that was actually bound, which is +// what port 0 resolved to when the caller asked for one. +func ServeMetrics(addr string, m *Metrics) (*http.Server, error) { + ln, err := net.Listen("tcp", addr) + if err != nil { + return nil, fmt.Errorf("--metrics-addr %s: %w", addr, err) + } + mux := http.NewServeMux() + mux.Handle("/metrics", m.Handler()) + mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "ccrawl metrics are at /metrics\n") + }) + srv := &http.Server{Addr: ln.Addr().String(), Handler: mux, ReadHeaderTimeout: 5 * time.Second} + go func() { _ = srv.Serve(ln) }() + return srv, nil +} diff --git a/ccrawl/metrics_test.go b/ccrawl/metrics_test.go new file mode 100644 index 0000000..dc8c4f9 --- /dev/null +++ b/ccrawl/metrics_test.go @@ -0,0 +1,160 @@ +package ccrawl + +import ( + "io" + "net/http" + "strings" + "testing" +) + +// exposition is the metrics text for m, as a scraper would see it. +func exposition(t *testing.T, m *Metrics) string { + t.Helper() + var b strings.Builder + n, err := m.WriteTo(&b) + if err != nil { + t.Fatalf("WriteTo: %v", err) + } + if int(n) != b.Len() { + t.Fatalf("WriteTo reported %d bytes, wrote %d", n, b.Len()) + } + return b.String() +} + +// hasLine reports whether the exposition contains exactly this line. +func hasLine(out, want string) bool { + for _, line := range strings.Split(out, "\n") { + if line == want { + return true + } + } + return false +} + +func TestMetricsExposition(t *testing.T) { + m := NewMetrics() + + // Nothing has happened yet, so nothing is exposed. A series that only exists + // because it was declared would report a zero that is not true. + if out := exposition(t, m); out != "" { + t.Fatalf("empty registry rendered %q", out) + } + + m.Add(MetricShards, 1, "pipeline", "markdown export", "status", StatusOK) + m.Add(MetricShards, 2, "pipeline", "markdown export", "status", StatusOK) + m.Add(MetricShards, 1, "pipeline", "markdown export", "status", StatusFailed) + m.Set(MetricInflight, 3, "pipeline", "markdown export") + m.Set(MetricRSS, 123456) + + out := exposition(t, m) + for _, want := range []string{ + "# HELP ccrawl_shards_total Shards finished, by pipeline and outcome.", + "# TYPE ccrawl_shards_total counter", + `ccrawl_shards_total{pipeline="markdown export",status="failed"} 1`, + `ccrawl_shards_total{pipeline="markdown export",status="ok"} 3`, + "# TYPE ccrawl_inflight gauge", + `ccrawl_inflight{pipeline="markdown export"} 3`, + "ccrawl_rss_bytes 123456", + } { + if !hasLine(out, want) { + t.Fatalf("exposition missing %q\ngot:\n%s", want, out) + } + } + + // An unlabelled series carries no braces at all. + if strings.Contains(out, "ccrawl_rss_bytes{}") { + t.Fatalf("unlabelled series rendered empty braces:\n%s", out) + } +} + +func TestMetricsHistogram(t *testing.T) { + m := NewMetrics() + for _, s := range []float64{0.5, 10, 45, 5000} { + m.Observe(MetricPhaseSeconds, s, "pipeline", "refetch", "phase", "fetch") + } + out := exposition(t, m) + + for _, want := range []string{ + "# TYPE ccrawl_phase_duration_seconds histogram", + `ccrawl_phase_duration_seconds_bucket{phase="fetch",pipeline="refetch",le="1"} 1`, + `ccrawl_phase_duration_seconds_bucket{phase="fetch",pipeline="refetch",le="15"} 2`, + `ccrawl_phase_duration_seconds_bucket{phase="fetch",pipeline="refetch",le="60"} 3`, + `ccrawl_phase_duration_seconds_bucket{phase="fetch",pipeline="refetch",le="3600"} 3`, + `ccrawl_phase_duration_seconds_bucket{phase="fetch",pipeline="refetch",le="+Inf"} 4`, + `ccrawl_phase_duration_seconds_sum{phase="fetch",pipeline="refetch"} 5055.5`, + `ccrawl_phase_duration_seconds_count{phase="fetch",pipeline="refetch"} 4`, + } { + if !hasLine(out, want) { + t.Fatalf("histogram missing %q\ngot:\n%s", want, out) + } + } +} + +func TestMetricsLabelEscaping(t *testing.T) { + m := NewMetrics() + m.Add(MetricItems, 1, "pipeline", "download", "status", "a\"b\\c\nd") + out := exposition(t, m) + want := `ccrawl_items_total{pipeline="download",status="a\"b\\c\nd"} 1` + if !hasLine(out, want) { + t.Fatalf("escaping wrong, want %q\ngot:\n%s", want, out) + } +} + +func TestMetricsNilAndUnknownName(t *testing.T) { + var m *Metrics + m.Add(MetricShards, 1, "pipeline", "x") + m.Set(MetricDone, 1) + m.Observe(MetricPhaseSeconds, 1) + if got := m.Value(MetricShards, "pipeline", "x"); got != 0 { + t.Fatalf("nil registry Value = %v, want 0", got) + } + if n, err := m.WriteTo(io.Discard); n != 0 || err != nil { + t.Fatalf("nil registry WriteTo = %d, %v", n, err) + } + + // A typo in a metric name is dropped, not panicked on: it must never take a + // long run down. + live := NewMetrics() + live.Add("ccrawl_not_a_metric", 1, "pipeline", "x") + if out := exposition(t, live); out != "" { + t.Fatalf("unknown metric was recorded:\n%s", out) + } + if got := live.Value("ccrawl_not_a_metric"); got != 0 { + t.Fatalf("unknown metric Value = %v, want 0", got) + } +} + +func TestServeMetrics(t *testing.T) { + m := NewMetrics() + m.Set(MetricDone, 7, "pipeline", "download") + + srv, err := ServeMetrics("127.0.0.1:0", m) + if err != nil { + t.Fatalf("ServeMetrics: %v", err) + } + defer func() { _ = srv.Close() }() + + // The listener is bound before ServeMetrics returns, so a second server on + // the same address has to fail here rather than in a goroutine later. + addr := srv.Addr + if dup, derr := ServeMetrics(addr, m); derr == nil { + _ = dup.Close() + t.Fatal("ServeMetrics on a busy port did not fail") + } + + resp, err := http.Get("http://" + addr + "/metrics") + if err != nil { + t.Fatalf("GET /metrics: %v", err) + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/plain") { + t.Fatalf("Content-Type = %q, want text/plain", ct) + } + if !hasLine(string(body), `ccrawl_done{pipeline="download"} 7`) { + t.Fatalf("scrape missing the gauge:\n%s", body) + } +} diff --git a/ccrawl/refetch.go b/ccrawl/refetch.go index c27cb4c..c530375 100644 --- a/ccrawl/refetch.go +++ b/ccrawl/refetch.go @@ -8,7 +8,6 @@ import ( "os" "path/filepath" "runtime" - "strconv" "strings" "sync" "time" @@ -432,24 +431,3 @@ func newRefetchParquetWriter(path string) (*ParquetWriter[RefetchRow], error) { func HFRefetchPath(crawlID string, shardIdx int) string { return fmt.Sprintf("data/crawl=%s/%06d.parquet", crawlID, shardIdx) } - -// currentRSSBytes reads the current process RSS from /proc/self/status. -// Returns 0 on any error or non-Linux platform. -func currentRSSBytes() int64 { - data, err := os.ReadFile("/proc/self/status") - if err != nil { - return 0 - } - for _, line := range strings.Split(string(data), "\n") { - if strings.HasPrefix(line, "VmRSS:") { - fields := strings.Fields(line) - if len(fields) >= 2 { - kb, err := strconv.ParseInt(fields[1], 10, 64) - if err == nil { - return kb * 1024 - } - } - } - } - return 0 -} diff --git a/ccrawl/refetch_run.go b/ccrawl/refetch_run.go index 154c311..a69b105 100644 --- a/ccrawl/refetch_run.go +++ b/ccrawl/refetch_run.go @@ -9,6 +9,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "time" "github.com/tamnd/ami/config" @@ -49,6 +50,13 @@ type RefetchExportConfig struct { // Progress is called once per committed batch with a snapshot of the run. Progress func(RefetchRunStats) + + // Reporter records the run: one event per shard, one per tick, and the start + // and end. Nil narrates to stderr in text and records nothing. + Reporter *RunReporter + // TickInterval is how often a tick event is written. 0 selects + // DefaultTickInterval. + TickInterval time.Duration } // RefetchRunStats is a live snapshot of a parallel refetch export run. @@ -81,6 +89,7 @@ type RefetchRunStats struct { ShardsPerHour float64 ETA time.Duration FreeDiskBytes int64 + RSSBytes int64 // resident set size at the last progress update } // packRefetchFn is the function the orchestrator uses to refetch one shard. @@ -120,6 +129,7 @@ func RunRefetchExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Refe if minFree <= 0 { minFree = 2 << 30 } + cfg.Reporter = cfg.Reporter.orDefault("markdown refetch") if err := os.MkdirAll(cfg.OutDir, 0o755); err != nil { return RefetchRunStats{}, err @@ -132,11 +142,39 @@ func RunRefetchExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Refe start := time.Now() run := RefetchRunStats{Total: len(cfg.Indices)} + cfg.Reporter.Event(RunEvent{Event: EventStart, Crawl: cfg.CrawlID, Total: run.Total}) + + // The committer owns run, so it publishes a copy for the ticker to read. + var snap atomic.Pointer[RefetchRunStats] + snap.Store(&RefetchRunStats{Total: run.Total}) + var inflight atomic.Int64 + committerDone := make(chan struct{}) var commitErr error go func() { defer close(committerDone) - commitErr = runRefetchCommitter(ctx, hf, cfg, k, start, finished, &run) + commitErr = runRefetchCommitter(ctx, hf, cfg, k, start, finished, &run, &snap) + }() + + // Ticker: a heartbeat so a run that fetches nothing for an hour still says so. + stopTicks := make(chan struct{}) + ticksDone := make(chan struct{}) + go func() { + defer close(ticksDone) + interval := cfg.TickInterval + if interval <= 0 { + interval = DefaultTickInterval + } + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-stopTicks: + return + case <-t.C: + cfg.Reporter.Event(refetchTickEvent(EventTick, snap.Load(), cfg, inflight.Load(), time.Since(start))) + } + } }() var wg sync.WaitGroup @@ -150,6 +188,7 @@ func RunRefetchExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Refe } waitForDisk(ctx, cfg.OutDir, minFree) outPath := filepath.Join(cfg.OutDir, fmt.Sprintf("%06d.parquet", idx)) + inflight.Add(1) stats, err := packRefetchFn(ctx, h, RefetchPackConfig{ CrawlID: cfg.CrawlID, ShardIdx: idx, @@ -160,6 +199,7 @@ func RunRefetchExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Refe CacheDir: cfg.CacheDir, FetchOnly: cfg.FetchOnly, }) + inflight.Add(-1) if err != nil { _ = os.Remove(outPath) finished <- refetchShardResult{idx: idx, err: err} @@ -188,20 +228,68 @@ func RunRefetchExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Refe wg.Wait() close(finished) <-committerDone + close(stopTicks) + <-ticksDone run.Elapsed = time.Since(start) run.FreeDiskBytes = freeDiskBytes(cfg.OutDir) + run.RSSBytes = currentRSSBytes() + cfg.Reporter.Event(refetchTickEvent(EventEnd, &run, cfg, 0, run.Elapsed)) if commitErr != nil { return run, commitErr } return run, ctx.Err() } +// refetchTickEvent renders a run snapshot as a tick or end event. It samples the +// clock, the free disk and the RSS at the moment of the tick rather than reusing +// the snapshot's copies, which are as old as the last committed batch. +func refetchTickEvent(kind string, s *RefetchRunStats, cfg RefetchExportConfig, inflight int64, elapsed time.Duration) RunEvent { + if s == nil { + s = &RefetchRunStats{} + } + rate := 0.0 + if elapsed > 0 { + rate = float64(s.Committed) / elapsed.Hours() + } + eta := 0.0 + if remaining := s.Total - s.Skipped - s.Committed - s.Failed; remaining > 0 && rate > 0 { + eta = float64(remaining) / rate * 3600 + } + return RunEvent{ + Event: kind, + Crawl: cfg.CrawlID, + Done: s.Committed + s.Skipped + s.Failed, + Total: s.Total, + Committed: s.Committed, + Skipped: s.Skipped, + Failed: s.Failed, + Inflight: inflight, + Rows: s.Rows, + Bytes: s.FetchBytes, + WARCBytes: s.WARCBytes, + HTMLBytes: s.HTMLBytes, + MDBytes: s.MDBytes, + ParquetBytes: s.ParquetBytes, + Rate: rate, + ETAS: eta, + ElapsedS: elapsed.Seconds(), + FreeDisk: freeDiskBytes(cfg.OutDir), + RSS: currentRSSBytes(), + } +} + // runRefetchCommitter drains finished shards, batches K parquets per HF commit, // deletes the local files, records the ledger, and reports progress with an ETA. -func runRefetchCommitter(ctx context.Context, hf *HFClient, cfg RefetchExportConfig, k int, start time.Time, finished <-chan refetchShardResult, run *RefetchRunStats) error { +func runRefetchCommitter(ctx context.Context, hf *HFClient, cfg RefetchExportConfig, k int, start time.Time, finished <-chan refetchShardResult, run *RefetchRunStats, snap *atomic.Pointer[RefetchRunStats]) error { var batch []refetchShardResult + // publish hands the ticker a stable copy of the run so far. + publish := func() { + s := *run + snap.Store(&s) + } + flush := func() error { if len(batch) == 0 { return nil @@ -275,10 +363,33 @@ func runRefetchCommitter(ctx context.Context, hf *HFClient, cfg RefetchExportCon } } run.Committed += len(batch) + + // After the commit, so a shard event means the parquet is on the hub. + for _, r := range batch { + idx := r.idx + cfg.Reporter.Event(RunEvent{ + Event: EventShard, + Crawl: cfg.CrawlID, + Shard: &idx, + Status: StatusOK, + Rows: r.stats.Rows, + Bytes: r.stats.FetchBytes, + WARCBytes: r.stats.WARCBytes, + HTMLBytes: r.stats.HTMLBytes, + MDBytes: r.stats.MDBytes, + ParquetBytes: r.stats.ParquetBytes, + FetchFailed: r.stats.Failed, + ExtractS: r.stats.DurExtract.Seconds(), + FetchS: r.stats.DurFetch.Seconds(), + ConvertS: r.stats.DurConvert.Seconds(), + ExportS: r.stats.DurExport.Seconds(), + }) + } batch = batch[:0] updateRefetchRunRates(run, start) logRefetchProgress(cfg, run) + publish() if cfg.Progress != nil { cfg.Progress(*run) } @@ -286,12 +397,20 @@ func runRefetchCommitter(ctx context.Context, hf *HFClient, cfg RefetchExportCon } for r := range finished { + idx := r.idx switch { case r.err == errAlreadyDone: run.Skipped++ + cfg.Reporter.Event(RunEvent{Event: EventShard, Crawl: cfg.CrawlID, Shard: &idx, Status: StatusSkipped}) + publish() case r.err != nil: run.Failed++ - fmt.Fprintf(os.Stderr, "refetch: shard %06d failed: %v\n", r.idx, r.err) + cfg.Reporter.Event(RunEvent{ + Event: EventShard, Crawl: cfg.CrawlID, Shard: &idx, + Status: StatusFailed, Error: r.err.Error(), + }) + cfg.Reporter.Textf("refetch: shard %06d failed: %v\n", r.idx, r.err) + publish() default: batch = append(batch, r) if len(batch) >= k { @@ -327,6 +446,7 @@ func logRefetchProgress(cfg RefetchExportConfig, run *RefetchRunStats) { pct = float64(done) / float64(run.Total) * 100 } run.FreeDiskBytes = freeDiskBytes(cfg.OutDir) + run.RSSBytes = currentRSSBytes() // Per-phase average over committed shards (0 guard avoids div-by-zero). n := int64(run.Committed) @@ -342,15 +462,15 @@ func logRefetchProgress(cfg RefetchExportConfig, run *RefetchRunStats) { if run.FetchS > 0 { fetchPerS = float64(run.URLsFound) / float64(run.FetchS) } - fmt.Fprintf(os.Stderr, + cfg.Reporter.Textf( "refetch: %d/%d shards (%.1f%%) | %d rows | %.1f shards/hr | ETA %s | disk %s\n", done, run.Total, pct, run.Rows, run.ShardsPerHour, fmtETA(run.ETA), fmtBytes(run.FreeDiskBytes)) - fmt.Fprintf(os.Stderr, + cfg.Reporter.Textf( " phases/shard avg: extract=%ds fetch=%ds convert=%ds export=%ds | fetch %.0f pages/s (%d urls)\n", run.ExtractS/n, run.FetchS/n, run.ConvertS/n, run.ExportS/n, fetchPerS, run.URLsFound) if run.Failures > 0 { - fmt.Fprintf(os.Stderr, + cfg.Reporter.Textf( " failures: %d total | dns=%d timeout=%d refused=%d skip=%d other=%d\n", run.Failures, run.ErrDNS, run.ErrTimeout, run.ErrRefused, run.ErrSkip, run.ErrOther) } diff --git a/ccrawl/rss_darwin.go b/ccrawl/rss_darwin.go new file mode 100644 index 0000000..c0cc163 --- /dev/null +++ b/ccrawl/rss_darwin.go @@ -0,0 +1,35 @@ +//go:build darwin + +package ccrawl + +import ( + "os" + "os/exec" + "strconv" + "strings" + "syscall" +) + +// currentRSSBytes returns this process's resident set size, or 0 when it cannot +// be read. +// +// macOS has no /proc, and the call that answers this directly (task_info) is a +// mach trap that needs cgo, which this binary does not use. ps is in the base +// system and reports the same number in kilobytes, and a progress tick asks for +// it every 30 seconds, so the subprocess is not on any hot path. If ps is +// missing the peak RSS from getrusage stands in: it overstates a long run, but +// it is never zero, which is the failure this replaces. +func currentRSSBytes() int64 { + out, err := exec.Command("ps", "-o", "rss=", "-p", strconv.Itoa(os.Getpid())).Output() + if err == nil { + if kb, perr := strconv.ParseInt(strings.TrimSpace(string(out)), 10, 64); perr == nil { + return kb * 1024 + } + } + var ru syscall.Rusage + if err := syscall.Getrusage(syscall.RUSAGE_SELF, &ru); err != nil { + return 0 + } + // Darwin reports maxrss in bytes, unlike Linux which uses kilobytes. + return int64(ru.Maxrss) +} diff --git a/ccrawl/rss_linux.go b/ccrawl/rss_linux.go new file mode 100644 index 0000000..60db49c --- /dev/null +++ b/ccrawl/rss_linux.go @@ -0,0 +1,33 @@ +//go:build linux + +package ccrawl + +import ( + "os" + "strconv" + "strings" +) + +// currentRSSBytes returns this process's resident set size, or 0 when it cannot +// be read. Linux publishes it as VmRSS in /proc/self/status, in kilobytes. +func currentRSSBytes() int64 { + data, err := os.ReadFile("/proc/self/status") + if err != nil { + return 0 + } + for line := range strings.SplitSeq(string(data), "\n") { + if !strings.HasPrefix(line, "VmRSS:") { + continue + } + fields := strings.Fields(line) + if len(fields) < 2 { + return 0 + } + kb, err := strconv.ParseInt(fields[1], 10, 64) + if err != nil { + return 0 + } + return kb * 1024 + } + return 0 +} diff --git a/ccrawl/rss_other.go b/ccrawl/rss_other.go new file mode 100644 index 0000000..2ef364e --- /dev/null +++ b/ccrawl/rss_other.go @@ -0,0 +1,8 @@ +//go:build !linux && !darwin + +package ccrawl + +// currentRSSBytes returns 0 on platforms where we have no way to read the +// resident set size without cgo. The journal writes rss_bytes as 0 there rather +// than reporting a number it made up. +func currentRSSBytes() int64 { return 0 } diff --git a/cli/crawl_cmd.go b/cli/crawl_cmd.go index 5a56a31..0d53e83 100644 --- a/cli/crawl_cmd.go +++ b/cli/crawl_cmd.go @@ -60,6 +60,16 @@ Examples: if maxTier <= 0 || maxTier > 5 { maxTier = 5 // emit all tiers by default } + // Seeding the full rank table is a ten million row stream, so the ticks + // carry an ETA against --max-seeds and one line per seed would be noise. + rep, stopRun, err := in.App.StartRun("crawl seed", "") + if err != nil { + return err + } + defer stopRun() + sp := ccrawl.StartStreamProgress(rep, "seeds", maxSeeds, 0) + defer sp.Stop() + count := 0 return ccrawl.RankStream(ctx, in.App.HTTP, g.HostRankURL(), "", func(r ccrawl.Rank) error { if count >= maxSeeds { @@ -72,6 +82,7 @@ Examples: return nil // skip hosts below the requested tier ceiling } count++ + sp.Add(1, 1, 0) return emit(SeedRecord{ Host: r.Key, URL: "https://" + r.Key + "/", diff --git a/cli/download.go b/cli/download.go index 98b956c..928778e 100644 --- a/cli/download.go +++ b/cli/download.go @@ -103,13 +103,25 @@ func runDownload(ctx context.Context, app *App, kind, outDir, segment string, sa return usageErr("aborted") } + // A download knows how many files it is going to move, so the progress it + // reports carries a real ETA. One journal event per file is affordable here + // because the file count is what the user asked for. + rep, stopRun, err := app.StartRun("download", "") + if err != nil { + return err + } + defer stopRun() + sp := ccrawl.StartStreamProgress(rep, "files", len(paths), 0) + defer sp.Stop() + var done int64 var bytes int64 progress := func(r ccrawl.DownloadResult) { n := atomic.AddInt64(&done, 1) atomic.AddInt64(&bytes, r.Bytes) + sp.Item(r.Path, r.Bytes, r.Err) if r.Err != nil { - _, _ = fmt.Fprintf(cmdErr, "[%d/%d] FAIL %s: %v\n", n, len(paths), r.Path, r.Err) + rep.Textf("[%d/%d] FAIL %s: %v\n", n, len(paths), r.Path, r.Err) return } state := "ok" @@ -117,14 +129,15 @@ func runDownload(ctx context.Context, app *App, kind, outDir, segment string, sa state = "skip" } if stderrTTY() || !r.Skipped { - _, _ = fmt.Fprintf(cmdErr, "[%d/%d] %-4s %s (%s)\n", n, len(paths), state, r.LocalPath, humanBytes(r.Bytes)) + rep.Textf("[%d/%d] %-4s %s (%s)\n", n, len(paths), state, r.LocalPath, humanBytes(r.Bytes)) } } - err := ccrawl.DownloadFiles(ctx, app.HTTP, app.Cfg.Source, paths, outDir, app.Workers, flat, progress) - _, _ = fmt.Fprintf(cmdErr, "downloaded %s across %d files\n", humanBytes(atomic.LoadInt64(&bytes)), len(paths)) - if err != nil { - return errs.Wrap(errs.KindNetwork, err, "download failed") + dlErr := ccrawl.DownloadFiles(ctx, app.HTTP, app.Cfg.Source, paths, outDir, app.Workers, flat, progress) + sp.Stop() + rep.Textf("downloaded %s across %d files\n", humanBytes(atomic.LoadInt64(&bytes)), len(paths)) + if dlErr != nil { + return errs.Wrap(errs.KindNetwork, dlErr, "download failed") } return nil } diff --git a/cli/engine.go b/cli/engine.go index 5f2e612..55a4453 100644 --- a/cli/engine.go +++ b/cli/engine.go @@ -2,7 +2,9 @@ package cli import ( "context" + "fmt" "io" + "net/http" "os" "github.com/tamnd/any-cli/kit" @@ -32,18 +34,27 @@ type App struct { Workers int UseLibrary bool LibraryDir string + + // Run reporting, from --progress, --journal and --metrics-addr. StartRun + // turns these into the reporter a long command hands to the pipeline. + Progress string + JournalPath string + MetricsAddr string } // domainGlobals holds the ccrawl-specific persistent flags that are not part of // the kit framework baseline. Root binds them on the root command; the client // factory and Finalize hook read them back when building the App. type domainGlobals struct { - crawl string - source string - workers int - library bool - libraryDir string - yes bool + crawl string + source string + workers int + library bool + libraryDir string + yes bool + progress string + journal string + metricsAddr string } // buildApp is the client factory kit calls once per run. It folds the resolved @@ -64,15 +75,63 @@ func buildApp(kc kit.Config, dom *domainGlobals) *App { cfg.Source = ccrawl.SourceS3 } return &App{ - Cfg: cfg, - HTTP: ccrawl.NewHTTPClient(cfg), - Cache: ccrawl.NewCache(cfg.CacheDir, !kc.NoCache), - Workers: dom.workers, - yes: dom.yes, - dryRun: kc.DryRun, - UseLibrary: dom.library, - LibraryDir: dom.libraryDir, + Cfg: cfg, + HTTP: ccrawl.NewHTTPClient(cfg), + Cache: ccrawl.NewCache(cfg.CacheDir, !kc.NoCache), + Workers: dom.workers, + yes: dom.yes, + dryRun: kc.DryRun, + UseLibrary: dom.library, + LibraryDir: dom.libraryDir, + Progress: dom.progress, + JournalPath: dom.journal, + MetricsAddr: dom.metricsAddr, + } +} + +// StartRun wires up everything a long running command reports through: the +// journal on disk, the progress mode on stderr, and the metrics endpoint. Every +// such command calls it the same way, so --progress, --journal and +// --metrics-addr mean the same thing everywhere. +// +// defaultJournal is where the journal goes when --journal was not given. Pass "" +// for the commands where a journal only makes sense if it was asked for. +// +// The returned stop closes the journal and the metrics server, and must be +// called when the run ends. +func (a *App) StartRun(pipeline, defaultJournal string) (*ccrawl.RunReporter, func(), error) { + mode, err := ccrawl.ParseProgressMode(a.Progress, stderrTTY()) + if err != nil { + return nil, nil, usageErr(err.Error()) + } + path := a.JournalPath + if path == "" { + path = defaultJournal + } + journal, err := ccrawl.OpenJournal(path) + if err != nil { + return nil, nil, fmt.Errorf("open run journal: %w", err) } + + var metrics *ccrawl.Metrics + var srv *http.Server + if a.MetricsAddr != "" { + metrics = ccrawl.NewMetrics() + // Bind now, so a busy port fails the command here rather than an hour in. + if srv, err = ccrawl.ServeMetrics(a.MetricsAddr, metrics); err != nil { + _ = journal.Close() + return nil, nil, err + } + } + + rep := ccrawl.NewRunReporter(pipeline, mode, journal, metrics) + rep.SetOutput(cmdErr) // the swappable stderr, so tests can capture progress + return rep, func() { + if srv != nil { + _ = srv.Close() + } + _ = journal.Close() + }, nil } // appFromCtx returns the run's App for an escape-hatch command, with the renderer diff --git a/cli/export.go b/cli/export.go index 0e36b3a..bcb6227 100644 --- a/cli/export.go +++ b/cli/export.go @@ -2,6 +2,7 @@ package cli import ( "context" + "errors" "fmt" "os" "strings" @@ -29,6 +30,11 @@ type exportCmd struct { mime string lang string filter []string + + // sp reports the run. An export can write millions of records, so it counts + // them with Add rather than an event each, and only the ticks land in the + // journal. + sp *ccrawl.StreamProgress } func newExportCmd() kit.Command { @@ -104,6 +110,17 @@ func (c *exportCmd) run(ctx context.Context, args []string) error { app := appFromCtx(ctx) exp := ccrawl.NewWARCExporter(c.outDir, c.prefix, c.subprefix, c.size, c.info()) + rep, stopRun, err := app.StartRun("export", "") + if err != nil { + return err + } + defer stopRun() + // A pattern export does not know how many captures it will match until the + // query is drained, so the total stays 0 and the ticks report a rate with no + // ETA. The stdin form fills the total in once it has read the locations. + c.sp = ccrawl.StartStreamProgress(rep, "records", 0, 0) + defer c.sp.Stop() + var runErr error if args[0] == "-" { runErr = c.exportStdin(ctx, app, exp) @@ -113,6 +130,7 @@ func (c *exportCmd) run(ctx context.Context, args []string) error { if cerr := exp.Close(); cerr != nil && runErr == nil { runErr = cerr } + c.sp.Stop() if runErr != nil { return runErr } @@ -137,12 +155,14 @@ func (c *exportCmd) writeLoc(ctx context.Context, app *App, exp *ccrawl.WARCExpo } raw, err := ccrawl.FetchWARCRecordRaw(ctx, app.HTTP, loc.Filename, loc.Offset, loc.Length) if err != nil { + c.sp.Fail() _, _ = fmt.Fprintln(cmdErr, "warn: "+err.Error()) return true, nil } if err := exp.Write(raw); err != nil { return false, err } + c.sp.Add(1, 0, int64(len(raw))) if app.Limit > 0 && exp.Records() >= app.Limit { return false, nil } @@ -173,7 +193,9 @@ func (c *exportCmd) exportQuery(ctx context.Context, app *App, exp *ccrawl.WARCE } return nil }) - if err != nil && err != errStopExport { + // CDXStream wraps whatever the callback returned with the page it came + // from, so the sentinel has to be unwrapped rather than compared. + if err != nil && !errors.Is(err, errStopExport) { return err } if stop { @@ -193,6 +215,7 @@ func (c *exportCmd) exportStdin(ctx context.Context, app *App, exp *ccrawl.WARCE }); err != nil { return err } + c.sp.SetTotal(len(locs)) for _, loc := range locs { cont, err := c.writeLoc(ctx, app, exp, loc) if err != nil { diff --git a/cli/host.go b/cli/host.go index f3a8781..d74dd60 100644 --- a/cli/host.go +++ b/cli/host.go @@ -296,7 +296,19 @@ func (e *hostEnrichCmd) run(ctx context.Context, _ []string) error { return err } + // Enrichment reads tens of gigabytes over four phases and none of them know + // their size up front, so progress here is about which phase a stalled run is + // stuck in rather than an ETA. + rep, stopRun, err := app.StartRun("host enrich", "") + if err != nil { + return err + } + defer stopRun() + sp := ccrawl.StartStreamProgress(rep, "hosts", 0, 0) + defer sp.Stop() + // Phase 2: vertex ID → host map (needed for degree join) + sp.Phase("vertices") hostToID := make(map[string]int64, 1<<18) maxID := int64(0) if err := ccrawl.VertexStream(ctx, app.HTTP, g.HostVerticesManifestURL(), func(v ccrawl.VertexRecord) error { @@ -312,6 +324,7 @@ func (e *hostEnrichCmd) run(ctx context.Context, _ []string) error { // Phase 3: edge degrees (optional) var inDeg, outDeg []int32 if e.degrees { + sp.Phase("edge degrees") inDeg, outDeg, err = ccrawl.ComputeEdgeDegrees(ctx, app.HTTP, g.HostEdgesManifestURL(), maxID+1) if err != nil { return fmt.Errorf("phase 3 edges: %w", err) @@ -321,6 +334,7 @@ func (e *hostEnrichCmd) run(ctx context.Context, _ []string) error { // Phase 4: CDX aggregation (optional) cdxStats := make(map[string]ccrawl.HostCDXStats) if e.cdx { + sp.Phase("cdx aggregate") if !ccrawl.DuckDBAvailable() { return fmt.Errorf("--cdx requires duckdb binary on PATH") } @@ -341,6 +355,7 @@ func (e *hostEnrichCmd) run(ctx context.Context, _ []string) error { } // Phase 1 + 5: stream rank table, join phases, emit + sp.Phase("rank join") return ccrawl.RankStream(ctx, app.HTTP, g.HostRankURL(), "", func(r ccrawl.Rank) error { rec := ccrawl.HostFromRank(r) rec.RegisteredDomain = registeredDomain(rec.Host) @@ -358,6 +373,7 @@ func (e *hostEnrichCmd) run(ctx context.Context, _ []string) error { applyHostCDXStats(&rec, s) } + sp.Add(1, 1, 0) return app.Out.Emit(rec) }) } diff --git a/cli/markdown_cmd.go b/cli/markdown_cmd.go index 82a007f..e5b15d8 100644 --- a/cli/markdown_cmd.go +++ b/cli/markdown_cmd.go @@ -147,6 +147,17 @@ func (v *markdownExportCmd) run(ctx context.Context, _ []string) error { fmt.Fprintf(os.Stderr, "markdown: ledger %s already records %d committed shards\n", ledgerPath, done) } + // The journal lands beside the ledger, so a resumed run's events sit next to + // the record of what it resumed from. + rep, stopRun, err := app.StartRun("markdown export", filepath.Join(outDir, "run.jsonl")) + if err != nil { + return err + } + defer stopRun() + if p := rep.JournalPath(); p != "" { + fmt.Fprintf(os.Stderr, "markdown: run journal %s (run %s)\n", p, rep.RunID()) + } + run, runErr := ccrawl.RunMarkdownExport(ctx, app.HTTP, hf, ccrawl.MarkdownExportConfig{ CrawlID: crawlID, Indices: indices, @@ -160,9 +171,10 @@ func (v *markdownExportCmd) run(ctx context.Context, _ []string) error { KeepParquet: v.keepParquet, MinFreeBytes: int64(v.minFreeGB) << 30, Ledger: ledger, + Reporter: rep, }) - fmt.Fprintf(os.Stderr, + rep.Textf( "\nmarkdown: %d committed, %d skipped, %d failed of %d | %d rows | html=%s md=%s parquet=%s | %s elapsed (%.1f shards/hour)\n", run.Committed, run.Skipped, run.Failed, run.Total, run.Rows, humanBytes(run.HTMLBytes), humanBytes(run.MDBytes), humanBytes(run.ParquetBytes), diff --git a/cli/markdown_refetch_cmd.go b/cli/markdown_refetch_cmd.go index 51394af..ea6b9c5 100644 --- a/cli/markdown_refetch_cmd.go +++ b/cli/markdown_refetch_cmd.go @@ -233,6 +233,17 @@ func (v *markdownRefetchCmd) run(ctx context.Context, _ []string) error { fmt.Fprintf(os.Stderr, "refetch: fetch-only mode, storing raw HTML and skipping convert\n") } + // The journal lands beside the ledger, so a resumed run's events sit next to + // the record of what it resumed from. + rep, stopRun, err := app.StartRun("markdown refetch", filepath.Join(outDir, "run.jsonl")) + if err != nil { + return err + } + defer stopRun() + if p := rep.JournalPath(); p != "" { + fmt.Fprintf(os.Stderr, "refetch: run journal %s (run %s)\n", p, rep.RunID()) + } + run, runErr := ccrawl.RunRefetchExport(ctx, app.HTTP, hf, ccrawl.RefetchExportConfig{ CrawlID: crawlID, Indices: indices, @@ -249,6 +260,7 @@ func (v *markdownRefetchCmd) run(ctx context.Context, _ []string) error { Ledger: ledger, CacheDir: warcCacheDir, FetchOnly: v.fetchOnly, + Reporter: rep, }) n := int64(run.Committed) @@ -259,19 +271,19 @@ func (v *markdownRefetchCmd) run(ctx context.Context, _ []string) error { if run.FetchS > 0 { fetchPagesPerS = float64(run.URLsFound) / float64(run.FetchS) } - fmt.Fprintf(os.Stderr, + rep.Textf( "\nrefetch: %d committed, %d skipped, %d failed of %d | %d rows | urls=%d html=%s md=%s parquet=%s | %s elapsed (%.1f shards/hour)\n", run.Committed, run.Skipped, run.Failed, run.Total, run.Rows, run.URLsFound, humanBytes(run.HTMLBytes), humanBytes(run.MDBytes), humanBytes(run.ParquetBytes), run.Elapsed.Round(time.Second), run.ShardsPerHour) - fmt.Fprintf(os.Stderr, + rep.Textf( "phase totals: extract=%ds fetch=%ds convert=%ds export=%ds publish=%ds\n", run.ExtractS, run.FetchS, run.ConvertS, run.ExportS, run.PublishS) - fmt.Fprintf(os.Stderr, + rep.Textf( "phase avg/shard: extract=%ds fetch=%ds convert=%ds export=%ds | fetch-only %.0f pages/s (urls/fetch-sec)\n", run.ExtractS/n, run.FetchS/n, run.ConvertS/n, run.ExportS/n, fetchPagesPerS) if run.Failures > 0 { - fmt.Fprintf(os.Stderr, + rep.Textf( "failures: %d total | dns=%d timeout=%d refused=%d skip=%d other=%d\n", run.Failures, run.ErrDNS, run.ErrTimeout, run.ErrRefused, run.ErrSkip, run.ErrOther) } diff --git a/cli/root.go b/cli/root.go index d6fb6b7..c699ac4 100644 --- a/cli/root.go +++ b/cli/root.go @@ -95,6 +95,9 @@ func (b *builder) globals(f *kit.FlagSet) { f.BoolVar(&b.dom.library, "library", false, "read and write under the structured dataset library") f.StringVar(&b.dom.libraryDir, "library-dir", ccrawl.LibraryDir(), "root of the dataset library") f.BoolVarP(&b.dom.yes, "yes", "y", false, "assume yes to prompts") + f.StringVar(&b.dom.progress, "progress", "", "progress reporting for long runs: text|json|none (default: text on a terminal, json otherwise)") + f.StringVar(&b.dom.journal, "journal", "", "append run events as JSON Lines to this file (default: run.jsonl beside the ledger)") + f.StringVar(&b.dom.metricsAddr, "metrics-addr", "", "serve Prometheus metrics for the run on this address, e.g. :9090") } // client is the factory kit calls once per run to build the shared engine from diff --git a/docs/content/reference/_index.md b/docs/content/reference/_index.md index f6fb627..4c1b517 100644 --- a/docs/content/reference/_index.md +++ b/docs/content/reference/_index.md @@ -5,4 +5,4 @@ description: "Every command, flag, output format, and setting, plus a troublesho weight: 30 --- -The complete surface, for when you know what you want and just need the exact spelling: the [CLI](/reference/cli/) command tree and flags, the [Markdown pipelines](/reference/markdown/) and their schemas, [configuration](/reference/configuration/) and environment, [output formats](/reference/output/), [exit codes](/reference/exit-codes/), and [troubleshooting](/reference/troubleshooting/). +The complete surface, for when you know what you want and just need the exact spelling: the [CLI](/reference/cli/) command tree and flags, the [Markdown pipelines](/reference/markdown/) and their schemas, [configuration](/reference/configuration/) and environment, the [run journal](/reference/run-journal/) a long run writes, [output formats](/reference/output/), [exit codes](/reference/exit-codes/), and [troubleshooting](/reference/troubleshooting/). diff --git a/docs/content/reference/cli.md b/docs/content/reference/cli.md index 1008051..e9035e2 100644 --- a/docs/content/reference/cli.md +++ b/docs/content/reference/cli.md @@ -673,6 +673,11 @@ These apply to every command. | `--no-header` | | Omit the header row in table output | false | | `--db` | | Tee every record into a store (e.g. `out.db`, `postgres://...`) | | | `--profile` | | Named profile to load | | +| `--progress` | | Progress reporting for long runs: `text`, `json`, `none` | text on a terminal, json otherwise | +| `--journal` | | Append run events as JSON Lines to this file | `run.jsonl` beside the ledger | +| `--metrics-addr` | | Serve Prometheus metrics for the run on this address, e.g. `:9090` | off | + +See [run journal](/reference/run-journal/) for the event schema, the metric names, and the queries worth keeping. ## Exit codes diff --git a/docs/content/reference/run-journal.md b/docs/content/reference/run-journal.md new file mode 100644 index 0000000..eeb6f84 --- /dev/null +++ b/docs/content/reference/run-journal.md @@ -0,0 +1,110 @@ +--- +title: "Run journal" +linkTitle: "Run journal" +description: "The JSON Lines event log a long run writes, the metrics it exposes, and the queries to ask of both." +weight: 25 +--- + +A publish run takes hours or days. When one of them finishes short, or stalls, or gets killed by a supervisor, the question afterwards is always the same: which shards failed, and when did it slow down. The run journal answers both without a regex over stderr. + +Every long running command writes one JSON object per line to a journal file, narrates itself on stderr in whichever shape you asked for, and can serve Prometheus metrics for the duration of the run. Three global flags control all of it. + +| Flag | Meaning | Default | +|---|---|---| +| `--progress` | `text`, `json`, or `none` | text on a terminal, json otherwise | +| `--journal` | Where the JSON Lines go | `run.jsonl` beside the ledger | +| `--metrics-addr` | Address to serve `/metrics` on | off | + +The default for `--progress` is the useful one: run it in a terminal and you get human lines, redirect it into a log file or run it under a supervisor and you get JSON, without anybody having to remember a flag. + +## Which commands report + +`markdown export` and `markdown refetch` report per shard, since a shard is the unit of work they resume on. `download`, `export`, `host enrich` and `crawl seed` report per item or per phase. The event shape is the same for all of them, so one query works everywhere. + +`markdown export` and `markdown refetch` default their journal to `run.jsonl` beside the ledger, so a resumed run's events land next to the record of what it resumed from. The rest write a journal only when `--journal` asks for one. + +## Event schema + +Four fields are on every line: `ts`, `event`, `run`, and `pipeline`. The rest depend on the event. + +| Field | Meaning | +|---|---| +| `ts` | UTC timestamp, millisecond resolution | +| `event` | `start`, `shard`, `item`, `phase`, `tick`, or `end` | +| `run` | This attempt, timestamp plus pid | +| `pipeline` | `markdown export`, `markdown refetch`, `download`, `export`, `host enrich`, `crawl seed` | +| `crawl` | Crawl ID, where the command has one | +| `phase` | The stage a multi-phase command is in | +| `name` | The file or URL an item event is about | +| `shard` | Shard index, on shard events | +| `status` | `ok`, `failed`, or `skipped` | +| `error` | Why it failed | +| `rows`, `bytes` | Rows written and bytes moved | +| `warc_bytes`, `html_bytes`, `md_bytes`, `parquet_bytes` | Bytes by kind, on the markdown pipelines | +| `extract_s`, `fetch_s`, `convert_s`, `export_s`, `publish_s` | Seconds in each phase of one shard | +| `done`, `total`, `committed`, `skipped`, `failed` | Progress counts | +| `inflight` | Units of work in flight at that moment | +| `fetch_failed` | URLs a refetch shard could not fetch | +| `rate_per_hour`, `eta_s`, `elapsed_s` | Live rates | +| `free_disk_bytes`, `rss_bytes` | Resources at that moment | + +`fetch_failed` is separate from `failed` on purpose. A refetch shard that hit a thousand dead hosts still succeeded, and folding the two together would make a healthy run look like a broken one. + +A `tick` lands every thirty seconds. That is often enough to see a stall on a three day run and rare enough that the journal stays small. + +## Queries worth keeping + +Which shards failed, and why: + +```bash +jq -r 'select(.event=="shard" and .status=="failed") | "\(.shard) \(.error)"' run.jsonl +``` + +When the rate dropped: + +```bash +jq -r 'select(.event=="tick") | [.ts, .rate_per_hour, .inflight] | @tsv' run.jsonl +``` + +What the last attempt did, when the journal holds several: + +```bash +jq -r --arg run "$(jq -r .run run.jsonl | tail -1)" 'select(.run==$run)' run.jsonl +``` + +Bytes committed per shard, largest first: + +```bash +jq -r 'select(.event=="shard" and .status=="ok") | [.parquet_bytes, .shard] | @tsv' run.jsonl | sort -rn | head +``` + +Whether the run ran out of disk before it ran out of work: + +```bash +jq -r 'select(.event=="tick") | [.ts, .free_disk_bytes, .rss_bytes] | @tsv' run.jsonl +``` + +## Metrics + +`--metrics-addr :9090` serves the same numbers at `/metrics` in the Prometheus text format, for the runs you want on a dashboard rather than in a shell. The address is bound before the command starts working, so a busy port fails immediately instead of an hour in. + +| Metric | Type | Labels | +|---|---|---| +| `ccrawl_shards_total` | counter | `pipeline`, `status` | +| `ccrawl_items_total` | counter | `pipeline`, `status` | +| `ccrawl_rows_total` | counter | `pipeline` | +| `ccrawl_bytes_total` | counter | `pipeline`, `kind` | +| `ccrawl_phase_duration_seconds` | histogram | `pipeline`, `phase` | +| `ccrawl_inflight` | gauge | `pipeline` | +| `ccrawl_done` | gauge | `pipeline` | +| `ccrawl_total` | gauge | `pipeline` | +| `ccrawl_rate_per_hour` | gauge | `pipeline` | +| `ccrawl_run_elapsed_seconds` | gauge | `pipeline` | +| `ccrawl_free_disk_bytes` | gauge | `pipeline` | +| `ccrawl_rss_bytes` | gauge | | + +The endpoint lives for the length of the run and goes away with it, so scrape it with a short interval and treat a gap as the run having ended. + +## Resident set size + +`rss_bytes` is read from `/proc/self/status` on Linux and from `ps` on macOS, and is 0 on anything else. It is sampled on the tick, not on a hot path.