From 1ad256cbe3f6998e5ce5e41dd23f397dabdd57e7 Mon Sep 17 00:00:00 2001 From: Viacheslav Poturaev Date: Tue, 4 Aug 2026 14:12:44 +0200 Subject: [PATCH 1/2] Add samples mode --- cmd/catp/catp/app.go | 5 + cmd/catp/catp/app_test.go | 346 ++++++++++++++++++++++++++++++++++++++ cmd/catp/catp/catp.go | 68 +++++++- cmd/catp/catp/samples.go | 139 +++++++++++++++ go.mod | 6 +- go.sum | 8 +- 6 files changed, 562 insertions(+), 10 deletions(-) create mode 100644 cmd/catp/catp/samples.go diff --git a/cmd/catp/catp/app.go b/cmd/catp/catp/app.go index 0b155ce..b79b955 100644 --- a/cmd/catp/catp/app.go +++ b/cmd/catp/catp/app.go @@ -109,6 +109,11 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g flag.IntVar(&r.endLine, "end-line", 0, "stop printing lines at this line (exclusive),\n"+ "default is 0 (no limit), each input file is counted separately") + flag.IntVar(&r.samples, "samples", 0, "collect N evenly distributed sample lines per file instead of full output\n"+ + "for uncompressed files with no -pass/-skip filters, this seeks by byte offset, giving instant access to huge files\n"+ + "otherwise (compressed file, or filters set) this requires a full scan, roughly gated by bytes read so far\n"+ + "(precision isn't guaranteed, and a line must also pass filters to be sampled)") + flag.Usage = func() { fmt.Println("catp", version.Module("github.com/bool64/progress").Version+r.options.VersionLabel+",", version.Info().GoVersion, strings.Join(versionExtra, " ")) diff --git a/cmd/catp/catp/app_test.go b/cmd/catp/catp/app_test.go index b71fd84..d53e5c5 100644 --- a/cmd/catp/catp/app_test.go +++ b/cmd/catp/catp/app_test.go @@ -1,12 +1,23 @@ package catp_test import ( + "bufio" + "compress/gzip" + "flag" + "fmt" "os" + "strconv" + "strings" "testing" "github.com/bool64/progress/cmd/catp/catp" ) +// resetFlags clears the global flag set so catp.Main can be invoked again within the same test binary. +func resetFlags() { + flag.CommandLine = flag.NewFlagSet(os.Args[0], flag.ExitOnError) +} + func Test_Main(t *testing.T) { os.Args = []string{ "catp", @@ -43,3 +54,338 @@ func Test_Main(t *testing.T) { t.Fatal("Unexpected output:\n", string(d)) } } + +func Test_Main_samples_seekable(t *testing.T) { + src, err := os.ReadFile("testdata/release-assets.yml") + if err != nil { + t.Fatal(err) + } + + srcLines := make(map[string]bool) + for l := range strings.SplitSeq(strings.TrimRight(string(src), "\n"), "\n") { + srcLines[l] = true + } + + out := "testdata/samples-seekable.log" + + resetFlags() + + os.Args = []string{ + "catp", + "-samples", "5", + "-no-progress", + "-output", out, + "testdata/release-assets.yml", + } + + if err := catp.Main(); err != nil { + t.Fatal(err) + } + + defer os.Remove(out) //nolint:errcheck + + d, err := os.ReadFile(out) //nolint:gosec + if err != nil { + t.Fatal(err) + } + + lines := strings.Split(strings.TrimRight(string(d), "\n"), "\n") + if len(lines) != 5 { + t.Fatalf("expected 5 sample lines, got %d: %q", len(lines), string(d)) + } + + for _, l := range lines { + if !srcLines[l] { + t.Fatalf("sampled line not found in source file: %q", l) + } + } +} + +// Test_Main_samples_compressed uses a synthetic file large enough, and stored without +// compression, that it clears pgzip's default read-ahead window (16 blocks of 250KB, ~4MB) +// several times over: -samples on compressed files uses compressed bytes read as a coarse proxy +// for position, so a file that fits inside that read-ahead window would have it all read before +// the scanner advances past the first few lines, clustering every sample at the start. That +// read-ahead is negligible against the multi-GB files -samples is meant for, so this only needs +// a large fixture, not a precise one - the assertions below stay loose accordingly. +func Test_Main_samples_compressed(t *testing.T) { + gzPath := "testdata/samples-src.log.gz" + + const totalLines = 2000000 + + func() { + f, err := os.Create(gzPath) //nolint:gosec + if err != nil { + t.Fatal(err) + } + defer f.Close() //nolint:errcheck + + gw, err := gzip.NewWriterLevel(f, gzip.NoCompression) + if err != nil { + t.Fatal(err) + } + defer gw.Close() //nolint:errcheck + + bw := bufio.NewWriterSize(gw, 64*1024) + + for i := range totalLines { + if _, err := fmt.Fprintf(bw, "line %08d\n", i); err != nil { + t.Fatal(err) + } + } + + if err := bw.Flush(); err != nil { + t.Fatal(err) + } + }() + + defer os.Remove(gzPath) //nolint:errcheck + + out := "testdata/samples-compressed.log" + + resetFlags() + + os.Args = []string{ + "catp", + "-samples", "5", + "-no-progress", + "-output", out, + gzPath, + } + + if err := catp.Main(); err != nil { + t.Fatal(err) + } + + defer os.Remove(out) //nolint:errcheck + + d, err := os.ReadFile(out) //nolint:gosec + if err != nil { + t.Fatal(err) + } + + lines := strings.Split(strings.TrimRight(string(d), "\n"), "\n") + if len(lines) != 5 { + t.Fatalf("expected 5 sample lines, got %d: %q", len(lines), string(d)) + } + + prev := -1 + + for _, l := range lines { + n, err := strconv.Atoi(strings.TrimPrefix(l, "line ")) + if err != nil { + t.Fatalf("unexpected sample line %q: %s", l, err) + } + + if n <= prev { + t.Fatalf("samples are not increasing: line %d came after line %d", n, prev) + } + + prev = n + } + + // Loose check that samples spread across the file rather than clustering near the start, + // which is what pgzip's read-ahead outrunning the scanner looked like before the fixture + // was sized past its read-ahead window. + if prev < totalLines/4 { + t.Fatalf("samples did not spread past the first quarter of the file: last sample was line %d of %d", prev, totalLines) + } +} + +func Test_Main_samples_stdin(t *testing.T) { + resetFlags() + + os.Args = []string{ + "catp", + "-samples", "2", + "-no-progress", + "-", + } + + if err := catp.Main(); err == nil { + t.Fatal("expected error sampling stdin") + } +} + +// Test_Main_samples_filtered_uncompressed checks that -pass/-skip filters on an uncompressed +// file steer -samples away from the instant byte-offset path (which can't honor filters) and +// into the same full-scan gating used for compressed files, so only matching lines are sampled. +// The fixture needs to clear progress.CountingReader's 100KB flush granularity (same reasoning +// as Test_Main_samples_compressed), otherwise every sample would land on the last matching line. +func Test_Main_samples_filtered_uncompressed(t *testing.T) { + srcPath := "testdata/samples-filtered-src.log" + + const totalLines = 2000000 + + func() { + f, err := os.Create(srcPath) //nolint:gosec + if err != nil { + t.Fatal(err) + } + defer f.Close() //nolint:errcheck + + bw := bufio.NewWriterSize(f, 64*1024) + + for i := range totalLines { + if i%97 == 0 { + if _, err := fmt.Fprintf(bw, "MATCH line %08d\n", i); err != nil { + t.Fatal(err) + } + } else if _, err := fmt.Fprintf(bw, "noise line %08d\n", i); err != nil { + t.Fatal(err) + } + } + + if err := bw.Flush(); err != nil { + t.Fatal(err) + } + }() + + defer os.Remove(srcPath) //nolint:errcheck + + out := "testdata/samples-filtered-uncompressed.log" + + resetFlags() + + os.Args = []string{ + "catp", + "-samples", "5", + "-pass", "MATCH", + "-no-progress", + "-output", out, + srcPath, + } + + if err := catp.Main(); err != nil { + t.Fatal(err) + } + + defer os.Remove(out) //nolint:errcheck + + d, err := os.ReadFile(out) //nolint:gosec + if err != nil { + t.Fatal(err) + } + + lines := strings.Split(strings.TrimRight(string(d), "\n"), "\n") + if len(lines) != 5 { + t.Fatalf("expected 5 sample lines, got %d: %q", len(lines), string(d)) + } + + prev := -1 + + for _, l := range lines { + if !strings.HasPrefix(l, "MATCH line ") { + t.Fatalf("sample line did not pass -pass filter: %q", l) + } + + n, err := strconv.Atoi(strings.TrimPrefix(l, "MATCH line ")) + if err != nil { + t.Fatalf("unexpected sample line %q: %s", l, err) + } + + if n <= prev { + t.Fatalf("samples are not increasing: line %d came after line %d", n, prev) + } + + prev = n + } + + if prev < totalLines/4 { + t.Fatalf("samples did not spread past the first quarter of the file: last sample was line %d of %d", prev, totalLines) + } +} + +// Test_Main_samples_gap checks that a stretch with no passing lines is simply skipped rather +// than causing the next passing line (wherever it lands) to also grab the very next one right +// after it to "catch up" - the bucket a sample is taken from should track how far bytes read +// has actually progressed, not the count of samples taken so far. +func Test_Main_samples_gap(t *testing.T) { + srcPath := "testdata/samples-gap-src.log" + + const ( + totalLines = 2000000 + matchEvery = 97 + gapStart = totalLines * 6 / 10 + gapEnd = totalLines * 8 / 10 + ) + + func() { + f, err := os.Create(srcPath) //nolint:gosec + if err != nil { + t.Fatal(err) + } + defer f.Close() //nolint:errcheck + + bw := bufio.NewWriterSize(f, 64*1024) + + for i := range totalLines { + if i%matchEvery == 0 && (i < gapStart || i >= gapEnd) { + if _, err := fmt.Fprintf(bw, "MATCH line %08d\n", i); err != nil { + t.Fatal(err) + } + } else if _, err := fmt.Fprintf(bw, "noise line %08d\n", i); err != nil { + t.Fatal(err) + } + } + + if err := bw.Flush(); err != nil { + t.Fatal(err) + } + }() + + defer os.Remove(srcPath) //nolint:errcheck + + out := "testdata/samples-gap.log" + + resetFlags() + + os.Args = []string{ + "catp", + "-samples", "10", + "-pass", "MATCH", + "-no-progress", + "-output", out, + srcPath, + } + + if err := catp.Main(); err != nil { + t.Fatal(err) + } + + defer os.Remove(out) //nolint:errcheck + + d, err := os.ReadFile(out) //nolint:gosec + if err != nil { + t.Fatal(err) + } + + lines := strings.Split(strings.TrimRight(string(d), "\n"), "\n") + + // The 20%-wide gap should cost at least one of the 10 requested samples, and none of the + // samples should land inside the gap. + if len(lines) < 1 || len(lines) >= 10 { + t.Fatalf("expected fewer than 10 sample lines due to the gap, got %d: %q", len(lines), string(d)) + } + + prev := -1 + + for _, l := range lines { + n, err := strconv.Atoi(strings.TrimPrefix(l, "MATCH line ")) + if err != nil { + t.Fatalf("unexpected sample line %q: %s", l, err) + } + + if n >= gapStart && n < gapEnd { + t.Fatalf("sample landed inside the gap: line %d", n) + } + + // The bug this guards against: a gap causing the next passing line to also grab the + // very next matching line right after it, matchEvery lines away, to "catch up". + if prev >= 0 && n-prev == matchEvery { + t.Fatalf("samples grabbed back-to-back matches right after the gap: line %d then %d", prev, n) + } + + prev = n + } +} diff --git a/cmd/catp/catp/catp.go b/cmd/catp/catp/catp.go index c23a617..5159e46 100644 --- a/cmd/catp/catp/catp.go +++ b/cmd/catp/catp/catp.go @@ -75,6 +75,7 @@ type runner struct { startLine int endLine int + samples int hasOptions bool options Options @@ -260,7 +261,15 @@ func (r *runner) linesPush() int { return int(lim) } -func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer) { +// scanFile scans and filters lines from rd to out. When sampleBytes is non-nil, it additionally +// gates writes to r.samples lines evenly spaced across *sampleBytes's range: a line is only +// written once it already passes filters and bytes read so far cross sampleTarget, a plain +// comparison checked on every candidate line. Only once a line actually clears that target is +// the current bucket computed (via one division), so a stretch with no passing lines is simply +// skipped - the line that finally clears a stale target jumps sampleTaken straight to the bucket +// it landed in, instead of also grabbing the very next line to catch up on the buckets that had +// no match. +func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer, sampleBytes *int64) { s := bufio.NewScanner(rd) s.Buffer(make([]byte, 64*1024), 10*1024*1024) @@ -274,6 +283,16 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer) { linesPush := r.linesPush() + sampleTotal := 0 + sampleTaken := 0 + sampleSize := int64(0) + var sampleTarget int64 + + if sampleBytes != nil { + sampleTotal = r.samples + sampleSize = r.sizes[filename] + } + for s.Scan() { fileLines++ @@ -307,6 +326,23 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer) { w = save } + if shouldWrite && sampleBytes != nil { + switch { + case sampleTaken >= sampleTotal: + shouldWrite = false + case atomic.LoadInt64(sampleBytes) < sampleTarget: + shouldWrite = false + default: + bucket := sampleTotal - 1 + if sampleSize > 0 { + bucket = min(int(atomic.LoadInt64(sampleBytes)*int64(sampleTotal)/sampleSize), sampleTotal-1) + } + + sampleTaken = bucket + 1 + sampleTarget = int64(sampleTaken) * sampleSize / int64(sampleTotal) + } + } + if lines >= linesPush { atomic.AddInt64(&r.currentLines, int64(lines)) lines = 0 @@ -459,6 +495,17 @@ func basicAuth(user, pass string, next http.Handler) http.Handler { } func (r *runner) cat(filename string) (err error) { + if r.samples > 0 { + handled, reason, err := r.trySampleSeekable(filename) + if handled { + return err + } + + if _, err := fmt.Fprintf(os.Stderr, "%s: full scan is used to collect samples because %s\n", filename, reason); err != nil { + return err + } + } + var rd io.Reader if filename == "-" { @@ -478,6 +525,21 @@ func (r *runner) cat(filename string) (err error) { rd = io.Reader(file) } + // sampleBytes tracks compressed bytes read for this file only, independent of the shared + // (and -no-progress-gated) progress counters below, so -samples can gate evenly spaced lines + // through a full scan regardless of -no-progress or -parallel. + var sampleBytes *int64 + + if r.samples > 0 { + sampleBytes = new(int64) + + scr := progress.NewCountingReader(rd) + scr.SetBytes(sampleBytes) + scr.SetLines(nil) + + rd = scr + } + if !r.noProgress { cr := progress.NewCountingReader(rd) cr.SetBytes(&r.currentBytes) @@ -543,8 +605,8 @@ func (r *runner) cat(filename string) (err error) { r.countLines = true } - if r.filters.isSet() || r.parallel > 1 || r.hasOptions || r.countLines || r.limiter != nil { - r.scanFile(filename, rd, out) + if r.filters.isSet() || r.parallel > 1 || r.hasOptions || r.countLines || r.limiter != nil || sampleBytes != nil { + r.scanFile(filename, rd, out, sampleBytes) } else { r.readFile(rd, out) } diff --git a/cmd/catp/catp/samples.go b/cmd/catp/catp/samples.go new file mode 100644 index 0000000..5a31b48 --- /dev/null +++ b/cmd/catp/catp/samples.go @@ -0,0 +1,139 @@ +package catp + +import ( + "bufio" + "bytes" + "errors" + "io" + "os" + "path" + "strings" +) + +// trySampleSeekable dispatches filename to the instant, seek-based sampler when it's a plain +// (uncompressed) file with no -pass/-skip filters to honor. It reports handled=true when the +// caller should return err as-is: either the fast path ran, or the file is one we can't sample +// this way (stdin has no known size to compute even offsets from). handled=false means the +// caller should fall back to a full-scan sample instead, with reason explaining why for its +// warning message: the file is compressed, or filters need every line inspected anyway. +func (r *runner) trySampleSeekable(filename string) (handled bool, reason string, err error) { + if filename == "-" { + return true, "", errors.New("samples: stdin is not supported, provide seekable file paths") + } + + if strings.HasSuffix(filename, ".gz") || strings.HasSuffix(filename, ".zst") { + return false, "the file is compressed", nil + } + + if r.filters.isSet() { + return false, "-pass/-skip filters are set", nil + } + + return true, "", r.sampleSeekable(filename) +} + +// sampleSeekable collects samples from an uncompressed file by seeking to evenly spaced byte +// offsets and reading the next full line at each, avoiding a full scan of the file. +func (r *runner) sampleSeekable(filename string) (err error) { + f, err := os.Open(filename) //nolint:gosec + if err != nil { + return err + } + defer func() { + if clErr := f.Close(); clErr != nil && err == nil { + err = clErr + } + }() + + out, closer, err := r.sampleOutput(filename) + if err != nil { + return err + } + defer func() { + if clErr := closer(); clErr != nil && err == nil { + err = clErr + } + }() + + size := r.sizes[filename] + + return r.sampleAtOffsets(size, out, func(offset int64) ([]byte, error) { + return readLineAt(f, size, offset) + }) +} + +// sampleAtOffsets writes a line for each of r.samples evenly spaced byte offsets across +// [0, size), fetched via lineAt. This is the shared distribution strategy for any format that +// can jump straight to a position instead of scanning: plain files today (lineAt seeks the raw +// file), and seekable zstd later (lineAt would resolve offset through the seek table, decompress +// the covering frame, and locate the line within it). size and offset are in whatever position +// space lineAt understands, e.g. uncompressed offsets for seekable zstd. +func (r *runner) sampleAtOffsets(size int64, out io.Writer, lineAt func(offset int64) ([]byte, error)) error { + n := r.samples + if int64(n) > size { + n = int(size) + } + + if n <= 0 { + return nil + } + + chunk := size / int64(n) + + for i := 0; i < n; i++ { + line, err := lineAt(int64(i) * chunk) + if err != nil { + continue + } + + if err := r.writeSample(out, line); err != nil { + return err + } + } + + return nil +} + +// readLineAt returns the next full line at or after offset, skipping a leading partial line when +// offset lands mid-line. ra is read only via ReadAt (via an io.SectionReader), never Seek, so any +// io.ReaderAt-backed source works without a shared, mutable read position: a plain *os.File today, +// and a seekable zstd's decompressed view later. +func readLineAt(ra io.ReaderAt, size, offset int64) ([]byte, error) { + br := bufio.NewReader(io.NewSectionReader(ra, offset, size-offset)) + + if offset > 0 { + if _, err := br.ReadBytes('\n'); err != nil { + return nil, err + } + } + + line, err := br.ReadBytes('\n') + if len(line) == 0 { + return nil, err + } + + return bytes.TrimRight(line, "\n"), nil +} + +// sampleOutput returns where sample lines should be written, honoring -out-dir the same way cat does. +func (r *runner) sampleOutput(filename string) (io.Writer, func() error, error) { + if r.outDir == "" { + return r.output, func() error { return nil }, nil + } + + return makeWriter(r.outDir + "/" + path.Base(filename)) +} + +// writeSample writes a single sample line, synchronizing with other goroutines when output is shared. +func (r *runner) writeSample(out io.Writer, line []byte) error { + synchronize := r.parallel > 1 && r.outDir == "" + + if synchronize { + r.mu.Lock() + defer r.mu.Unlock() + } + + _, err := out.Write(append(line, '\n')) + + return err +} diff --git a/go.mod b/go.mod index 2041398..0fd8d85 100644 --- a/go.mod +++ b/go.mod @@ -1,12 +1,12 @@ module github.com/bool64/progress -go 1.24.0 +go 1.25.0 require ( github.com/DataDog/zstd v1.5.7 github.com/bool64/dev v0.2.45 github.com/cloudflare/ahocorasick v0.0.0-20240916140611-054963ec9396 - github.com/klauspost/compress v1.18.2 + github.com/klauspost/compress v1.19.1 github.com/klauspost/pgzip v1.2.6 - golang.org/x/time v0.14.0 + golang.org/x/time v0.15.0 ) diff --git a/go.sum b/go.sum index 0c780c7..4cae43e 100644 --- a/go.sum +++ b/go.sum @@ -4,9 +4,9 @@ github.com/bool64/dev v0.2.45 h1:3nLKhAS/6Oklk3Mt2lHYSN/Cb4tdAD77KLwzeP+6eYE= github.com/bool64/dev v0.2.45/go.mod h1:iJbh1y/HkunEPhgebWRNcs8wfGq7sjvJ6W5iabL8ACg= github.com/cloudflare/ahocorasick v0.0.0-20240916140611-054963ec9396 h1:W2HK1IdCnCGuLUeyizSCkwvBjdj0ZL7mxnJYQ3poyzI= github.com/cloudflare/ahocorasick v0.0.0-20240916140611-054963ec9396/go.mod h1:tGWUZLZp9ajsxUOnHmFFLnqnlKXsCn6GReG4jAD59H0= -github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk= -github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= +github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/pgzip v1.2.6 h1:8RXeL5crjEUFnR2/Sn6GJNWtSQ3Dk8pq4CL3jvdDyjU= github.com/klauspost/pgzip v1.2.6/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs= -golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= -golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= From 8b6a72227c110c3df3ba4e0613c636b0625db5c7 Mon Sep 17 00:00:00 2001 From: Viacheslav Poturaev Date: Fri, 7 Aug 2026 01:51:25 +0200 Subject: [PATCH 2/2] Add samples mode --- cmd/catp/catp/app.go | 7 +- cmd/catp/catp/app_test.go | 68 +++++++++++--- cmd/catp/catp/catp.go | 1 + cmd/catp/catp/samples.go | 189 ++++++++++++++++++++++++++------------ 4 files changed, 187 insertions(+), 78 deletions(-) diff --git a/cmd/catp/catp/app.go b/cmd/catp/catp/app.go index b79b955..004e83d 100644 --- a/cmd/catp/catp/app.go +++ b/cmd/catp/catp/app.go @@ -110,9 +110,10 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g "default is 0 (no limit), each input file is counted separately") flag.IntVar(&r.samples, "samples", 0, "collect N evenly distributed sample lines per file instead of full output\n"+ - "for uncompressed files with no -pass/-skip filters, this seeks by byte offset, giving instant access to huge files\n"+ - "otherwise (compressed file, or filters set) this requires a full scan, roughly gated by bytes read so far\n"+ - "(precision isn't guaranteed, and a line must also pass filters to be sampled)") + "for uncompressed files, this seeks by byte offset, scanning forward only until a line passing\n"+ + "-pass/-skip filters is found (immediately, if no filters are set), giving near-instant access to huge files\n"+ + "for compressed (.gz/.zst) files this requires a full scan, roughly gated by bytes read so far\n"+ + "(precision isn't guaranteed in either case)") flag.Usage = func() { fmt.Println("catp", version.Module("github.com/bool64/progress").Version+r.options.VersionLabel+",", diff --git a/cmd/catp/catp/app_test.go b/cmd/catp/catp/app_test.go index d53e5c5..b10e8d7 100644 --- a/cmd/catp/catp/app_test.go +++ b/cmd/catp/catp/app_test.go @@ -5,6 +5,7 @@ import ( "compress/gzip" "flag" "fmt" + "io" "os" "strconv" "strings" @@ -13,6 +14,34 @@ import ( "github.com/bool64/progress/cmd/catp/catp" ) +// runCapturingStderr runs fn with os.Stderr redirected, returning whatever it wrote. +func runCapturingStderr(t *testing.T, fn func() error) (string, error) { + t.Helper() + + r, w, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + + orig := os.Stderr + os.Stderr = w + + fnErr := fn() + + os.Stderr = orig + + if err := w.Close(); err != nil { + t.Fatal(err) + } + + captured, err := io.ReadAll(r) + if err != nil { + t.Fatal(err) + } + + return string(captured), fnErr +} + // resetFlags clears the global flag set so catp.Main can be invoked again within the same test binary. func resetFlags() { flag.CommandLine = flag.NewFlagSet(os.Args[0], flag.ExitOnError) @@ -84,7 +113,7 @@ func Test_Main_samples_seekable(t *testing.T) { defer os.Remove(out) //nolint:errcheck - d, err := os.ReadFile(out) //nolint:gosec + d, err := os.ReadFile(out) if err != nil { t.Fatal(err) } @@ -114,7 +143,7 @@ func Test_Main_samples_compressed(t *testing.T) { const totalLines = 2000000 func() { - f, err := os.Create(gzPath) //nolint:gosec + f, err := os.Create(gzPath) if err != nil { t.Fatal(err) } @@ -159,7 +188,7 @@ func Test_Main_samples_compressed(t *testing.T) { defer os.Remove(out) //nolint:errcheck - d, err := os.ReadFile(out) //nolint:gosec + d, err := os.ReadFile(out) if err != nil { t.Fatal(err) } @@ -208,17 +237,16 @@ func Test_Main_samples_stdin(t *testing.T) { } // Test_Main_samples_filtered_uncompressed checks that -pass/-skip filters on an uncompressed -// file steer -samples away from the instant byte-offset path (which can't honor filters) and -// into the same full-scan gating used for compressed files, so only matching lines are sampled. -// The fixture needs to clear progress.CountingReader's 100KB flush granularity (same reasoning -// as Test_Main_samples_compressed), otherwise every sample would land on the last matching line. +// file still use the seek-based sampler: it scans forward from each target offset only until a +// line passing filters is found, then jumps to the next target - so only matching lines are +// sampled, without falling back to a full scan. func Test_Main_samples_filtered_uncompressed(t *testing.T) { srcPath := "testdata/samples-filtered-src.log" const totalLines = 2000000 func() { - f, err := os.Create(srcPath) //nolint:gosec + f, err := os.Create(srcPath) if err != nil { t.Fatal(err) } @@ -256,13 +284,18 @@ func Test_Main_samples_filtered_uncompressed(t *testing.T) { srcPath, } - if err := catp.Main(); err != nil { - t.Fatal(err) + stderr, mainErr := runCapturingStderr(t, func() error { return catp.Main() }) + if mainErr != nil { + t.Fatal(mainErr) + } + + if strings.Contains(stderr, "full scan") { + t.Fatalf("expected the seek-based sampler to run, but got a full-scan warning: %q", stderr) } defer os.Remove(out) //nolint:errcheck - d, err := os.ReadFile(out) //nolint:gosec + d, err := os.ReadFile(out) if err != nil { t.Fatal(err) } @@ -311,7 +344,7 @@ func Test_Main_samples_gap(t *testing.T) { ) func() { - f, err := os.Create(srcPath) //nolint:gosec + f, err := os.Create(srcPath) if err != nil { t.Fatal(err) } @@ -349,13 +382,18 @@ func Test_Main_samples_gap(t *testing.T) { srcPath, } - if err := catp.Main(); err != nil { - t.Fatal(err) + stderr, mainErr := runCapturingStderr(t, func() error { return catp.Main() }) + if mainErr != nil { + t.Fatal(mainErr) + } + + if strings.Contains(stderr, "full scan") { + t.Fatalf("expected the seek-based sampler to run, but got a full-scan warning: %q", stderr) } defer os.Remove(out) //nolint:errcheck - d, err := os.ReadFile(out) //nolint:gosec + d, err := os.ReadFile(out) if err != nil { t.Fatal(err) } diff --git a/cmd/catp/catp/catp.go b/cmd/catp/catp/catp.go index 5159e46..f304352 100644 --- a/cmd/catp/catp/catp.go +++ b/cmd/catp/catp/catp.go @@ -286,6 +286,7 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer, sampleBy sampleTotal := 0 sampleTaken := 0 sampleSize := int64(0) + var sampleTarget int64 if sampleBytes != nil { diff --git a/cmd/catp/catp/samples.go b/cmd/catp/catp/samples.go index 5a31b48..25d33e7 100644 --- a/cmd/catp/catp/samples.go +++ b/cmd/catp/catp/samples.go @@ -8,14 +8,17 @@ import ( "os" "path" "strings" + "sync/atomic" + + "github.com/bool64/progress" ) -// trySampleSeekable dispatches filename to the instant, seek-based sampler when it's a plain -// (uncompressed) file with no -pass/-skip filters to honor. It reports handled=true when the -// caller should return err as-is: either the fast path ran, or the file is one we can't sample -// this way (stdin has no known size to compute even offsets from). handled=false means the -// caller should fall back to a full-scan sample instead, with reason explaining why for its -// warning message: the file is compressed, or filters need every line inspected anyway. +// trySampleSeekable dispatches filename to the seek-based sampler when it's a plain +// (uncompressed) file, filters or no filters. It reports handled=true when the caller should +// return err as-is: either the seek-based path ran, or the file is one we can't sample this way +// (stdin has no known size to compute even offsets from). handled=false means the caller should +// fall back to a full-scan sample instead, because the file is compressed and can't be seeked +// into cheaply. func (r *runner) trySampleSeekable(filename string) (handled bool, reason string, err error) { if filename == "-" { return true, "", errors.New("samples: stdin is not supported, provide seekable file paths") @@ -25,108 +28,165 @@ func (r *runner) trySampleSeekable(filename string) (handled bool, reason string return false, "the file is compressed", nil } - if r.filters.isSet() { - return false, "-pass/-skip filters are set", nil - } - return true, "", r.sampleSeekable(filename) } -// sampleSeekable collects samples from an uncompressed file by seeking to evenly spaced byte -// offsets and reading the next full line at each, avoiding a full scan of the file. +// sampleSeekable collects r.samples lines from an uncompressed file by seeking to evenly spaced +// target offsets and scanning forward from each one for the first line that passes -pass/-skip +// filters (with no filters set, that's simply the next line). Once a line is taken, it jumps +// straight to the next target instead of continuing to read through the rest of the current +// segment - only the (typically short) stretch between a target and its first match is ever +// read, so this stays close to instant even with filters, unlike a full scan. A stretch with no +// passing line before EOF costs that sample rather than blocking for one - fewer samples, not a +// stall - and since all remaining targets are past it too, none of them can find anything either. func (r *runner) sampleSeekable(filename string) (err error) { f, err := os.Open(filename) //nolint:gosec if err != nil { return err } + defer func() { if clErr := f.Close(); clErr != nil && err == nil { err = clErr } }() + size := r.sizes[filename] + + n := r.samples + if int64(n) > size { + n = int(size) + } + + if n <= 0 { + return nil + } + out, closer, err := r.sampleOutput(filename) if err != nil { return err } + defer func() { if clErr := closer(); clErr != nil && err == nil { err = clErr } }() - size := r.sizes[filename] - - return r.sampleAtOffsets(size, out, func(offset int64) ([]byte, error) { - return readLineAt(f, size, offset) - }) -} - -// sampleAtOffsets writes a line for each of r.samples evenly spaced byte offsets across -// [0, size), fetched via lineAt. This is the shared distribution strategy for any format that -// can jump straight to a position instead of scanning: plain files today (lineAt seeks the raw -// file), and seekable zstd later (lineAt would resolve offset through the seek table, decompress -// the covering frame, and locate the line within it). size and offset are in whatever position -// space lineAt understands, e.g. uncompressed offsets for seekable zstd. -func (r *runner) sampleAtOffsets(size int64, out io.Writer, lineAt func(offset int64) ([]byte, error)) error { - n := r.samples - if int64(n) > size { - n = int(size) + // fileBytes tracks our actual position in the file, kept in sync across jumps (set to the + // seek target immediately, then advanced as each segment is read) so progress display + // reflects real position rather than lagging behind at the last linearly-read byte. + var fileBytes int64 + + if r.parallel <= 1 && !r.noProgress { + cr := progress.NewCountingReader(nil) + cr.SetBytes(&fileBytes) + r.currentFile = cr + r.currentTotal = size + + r.pr.Start(func(t *progress.Task) { + t.TotalBytes = func() int64 { return r.totalBytes } + t.CurrentBytes = func() int64 { return atomic.LoadInt64(&fileBytes) } + t.Task = filename + t.Continue = true + t.PrintOnStart = true + }) + + defer r.pr.Stop() } - if n <= 0 { - return nil - } + taken := 0 - chunk := size / int64(n) + for taken < n { + offset := int64(taken) * size / int64(n) + atomic.StoreInt64(&fileBytes, offset) - for i := 0; i < n; i++ { - line, err := lineAt(int64(i) * chunk) + matched, pos, err := r.sampleNextMatch(filename, f, out, offset, size, taken, &fileBytes) if err != nil { - continue + return err } - if err := r.writeSample(out, line); err != nil { - return err + if !matched { + break // nothing more from here to EOF; every remaining target is past it too } + + taken = min(int(pos*int64(n)/size), n-1) + 1 } return nil } -// readLineAt returns the next full line at or after offset, skipping a leading partial line when -// offset lands mid-line. ra is read only via ReadAt (via an io.SectionReader), never Seek, so any -// io.ReaderAt-backed source works without a shared, mutable read position: a plain *os.File today, -// and a seekable zstd's decompressed view later. -func readLineAt(ra io.ReaderAt, size, offset int64) ([]byte, error) { +// sampleNextMatch scans forward from offset for the first line passing filters, writing it +// (honoring -save-matches and -out-dir/-parallel write synchronization the same way scanFile +// does) and reporting the byte position right after it. matched is false if EOF was reached +// with no passing line. +func (r *runner) sampleNextMatch( + filename string, ra io.ReaderAt, out io.Writer, offset, size int64, lineNr int, fileBytes *int64, +) (matched bool, pos int64, err error) { br := bufio.NewReader(io.NewSectionReader(ra, offset, size-offset)) + pos = offset if offset > 0 { - if _, err := br.ReadBytes('\n'); err != nil { - return nil, err + skip, err := br.ReadBytes('\n') + pos += int64(len(skip)) + atomic.StoreInt64(fileBytes, pos) + + if err != nil { + if !errors.Is(err, io.EOF) { + return false, pos, err + } + + return false, pos, nil // no full line left from here to EOF } } - line, err := br.ReadBytes('\n') - if len(line) == 0 { - return nil, err - } + for { + raw, rerr := br.ReadBytes('\n') + pos += int64(len(raw)) + atomic.StoreInt64(fileBytes, pos) - return bytes.TrimRight(line, "\n"), nil -} + if rerr != nil && !errors.Is(rerr, io.EOF) { + return false, pos, rerr + } -// sampleOutput returns where sample lines should be written, honoring -out-dir the same way cat does. -func (r *runner) sampleOutput(filename string) (io.Writer, func() error, error) { - if r.outDir == "" { - return r.output, func() error { return nil }, nil - } + if len(raw) > 0 { + line := bytes.TrimRight(raw, "\n") - return makeWriter(r.outDir + "/" + path.Base(filename)) + save, shouldWrite := r.filters.shouldWrite(line) + if shouldWrite { + if r.hasOptions && r.options.PrepareLine != nil { + var buf []byte + + line = r.options.PrepareLine(filename, lineNr, line, &buf) + } + + if line != nil { + w := out + if save != nil { + w = save + } + + if err := r.writeMatch(w, line, save != nil); err != nil { + return false, pos, err + } + + atomic.AddInt64(&r.matches, 1) + } + + return true, pos, nil + } + } + + if rerr != nil { + return false, pos, nil //nolint:nilerr // only io.EOF reaches here, non-EOF already returned above + } + } } -// writeSample writes a single sample line, synchronizing with other goroutines when output is shared. -func (r *runner) writeSample(out io.Writer, line []byte) error { - synchronize := r.parallel > 1 && r.outDir == "" +// writeMatch writes a single sample line, synchronizing with other goroutines when output is +// shared: either STDOUT/a shared file (no -out-dir), or a shared -save-matches writer. +func (r *runner) writeMatch(out io.Writer, line []byte, saved bool) error { + synchronize := r.parallel > 1 && (r.outDir == "" || saved) if synchronize { r.mu.Lock() @@ -137,3 +197,12 @@ func (r *runner) writeSample(out io.Writer, line []byte) error { return err } + +// sampleOutput returns where sample lines should be written, honoring -out-dir the same way cat does. +func (r *runner) sampleOutput(filename string) (io.Writer, func() error, error) { + if r.outDir == "" { + return r.output, func() error { return nil }, nil + } + + return makeWriter(r.outDir + "/" + path.Base(filename)) +}