diff --git a/cmd/catp/catp/app.go b/cmd/catp/catp/app.go index 0b155ce..004e83d 100644 --- a/cmd/catp/catp/app.go +++ b/cmd/catp/catp/app.go @@ -109,6 +109,12 @@ 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, 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+",", version.Info().GoVersion, strings.Join(versionExtra, " ")) diff --git a/cmd/catp/catp/app_test.go b/cmd/catp/catp/app_test.go index b71fd84..b10e8d7 100644 --- a/cmd/catp/catp/app_test.go +++ b/cmd/catp/catp/app_test.go @@ -1,12 +1,52 @@ package catp_test import ( + "bufio" + "compress/gzip" + "flag" + "fmt" + "io" "os" + "strconv" + "strings" "testing" "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) +} + func Test_Main(t *testing.T) { os.Args = []string{ "catp", @@ -43,3 +83,347 @@ 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) + 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) + 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) + 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 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) + 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, + } + + 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) + 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) + 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, + } + + 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) + 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..f304352 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,17 @@ 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 +327,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 +496,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 +526,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 +606,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..25d33e7 --- /dev/null +++ b/cmd/catp/catp/samples.go @@ -0,0 +1,208 @@ +package catp + +import ( + "bufio" + "bytes" + "errors" + "io" + "os" + "path" + "strings" + "sync/atomic" + + "github.com/bool64/progress" +) + +// 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") + } + + if strings.HasSuffix(filename, ".gz") || strings.HasSuffix(filename, ".zst") { + return false, "the file is compressed", nil + } + + return true, "", r.sampleSeekable(filename) +} + +// 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 + } + }() + + // 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() + } + + taken := 0 + + for taken < n { + offset := int64(taken) * size / int64(n) + atomic.StoreInt64(&fileBytes, offset) + + matched, pos, err := r.sampleNextMatch(filename, f, out, offset, size, taken, &fileBytes) + if 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 +} + +// 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 { + 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 + } + } + + for { + raw, rerr := br.ReadBytes('\n') + pos += int64(len(raw)) + atomic.StoreInt64(fileBytes, pos) + + if rerr != nil && !errors.Is(rerr, io.EOF) { + return false, pos, rerr + } + + if len(raw) > 0 { + line := bytes.TrimRight(raw, "\n") + + 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 + } + } +} + +// 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() + defer r.mu.Unlock() + } + + _, err := out.Write(append(line, '\n')) + + 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)) +} 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=