From de068116685bc4849ff29c2daa7900bbb5ba7d0a Mon Sep 17 00:00:00 2001 From: Viacheslav Poturaev Date: Thu, 22 Oct 2026 15:45:29 +0200 Subject: [PATCH 1/3] Add split chunks mode, add http charts, remove sorting --- cmd/catp/catp/app.go | 24 ++++- cmd/catp/catp/app_test.go | 59 ++++++++++++ cmd/catp/catp/catp.go | 187 ++++++++++++++++++++++++++++++++++++-- cmd/catp/catp/chunks.go | 144 +++++++++++++++++++++++++++++ go.mod | 11 ++- go.sum | 22 +++-- progress.go | 9 +- 7 files changed, 430 insertions(+), 26 deletions(-) create mode 100644 cmd/catp/catp/chunks.go diff --git a/cmd/catp/catp/app.go b/cmd/catp/catp/app.go index 004e83d..8f2a5cc 100644 --- a/cmd/catp/catp/app.go +++ b/cmd/catp/catp/app.go @@ -14,7 +14,6 @@ import ( "path/filepath" "runtime/pprof" "slices" - "sort" "strings" "sync/atomic" "syscall" @@ -87,6 +86,12 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g "lines from different files will go to output simultaneously (out of order of files, but in order of lines in each file)\n"+ "use 0 for multi-threaded zst decoder (slightly faster at cost of more CPU)") + flag.BoolVar(&r.splitChunks, "split-chunks", false, "experimental: with -parallel N (N >= 2), split each uncompressed file (>= 1 MB per chunk)\n"+ + "into up to N byte-range chunks processed in parallel as virtual files named file.chunkNN,\n"+ + "(line order is lost in STDOUT/-output, with -out-dir each chunk goes to its own file, so concatenating them in name order restores the order),\n"+ + "line numbers are per chunk, .gz and .zst files are not split,\n"+ + "incompatible with -start-line, -end-line, -samples and STDIN") + cpuProfile := flag.String("dbg-cpu-prof", "", "write first 10 seconds of CPU profile to file") memProfile := flag.String("dbg-mem-prof", "", "write heap profile to file after 10 seconds") output := flag.String("output", "", "output to file (can have .gz or .zst ext for compression) instead of STDOUT") @@ -181,7 +186,16 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g } } - sort.Strings(files) + if r.splitChunks { + switch { + case r.parallel < 2: + return errors.New("-split-chunks requires -parallel 2 or more") + case r.startLine != 0 || r.endLine != 0 || r.samples != 0: + return errors.New("-split-chunks is incompatible with -start-line, -end-line and -samples") + case isStdin: + return errors.New("-split-chunks is incompatible with STDIN") + } + } if *output != "" && r.outDir == "" { w, closer, err := makeWriter(*output) @@ -280,7 +294,7 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g sem := make(chan struct{}, r.parallel) errs := make(chan error, r.parallel) - for _, fn := range files { + for _, j := range r.jobs(files) { select { case err := <-errs: return err @@ -292,8 +306,8 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g <-sem }() - if err := r.cat(fn); err != nil { - errs <- fmt.Errorf("%s: %w", fn, err) + if err := r.catChunk(j.name, j.chunk); err != nil { + errs <- fmt.Errorf("%s: %w", j.name, err) } }() } diff --git a/cmd/catp/catp/app_test.go b/cmd/catp/catp/app_test.go index b10e8d7..e2c7479 100644 --- a/cmd/catp/catp/app_test.go +++ b/cmd/catp/catp/app_test.go @@ -427,3 +427,62 @@ func Test_Main_samples_gap(t *testing.T) { prev = n } } + +func Test_Main_splitChunks(t *testing.T) { + dir := t.TempDir() + src := dir + "/foo.log" + + var sb strings.Builder + + var want strings.Builder + + for i := 0; sb.Len() < 5<<20; i++ { + line := "line " + strconv.Itoa(i) + strings.Repeat("x", i%97) + if i%7 == 0 { + line += " xyz" + want.WriteString(line + "\n") + } + + sb.WriteString(line + "\n") + } + + if err := os.WriteFile(src, []byte(sb.String()), 0o600); err != nil { + t.Fatal(err) + } + + out := dir + "/out" + if err := os.Mkdir(out, 0o700); err != nil { + t.Fatal(err) + } + + resetFlags() + + os.Args = []string{"catp", "-no-progress", "-split-chunks", "-parallel", "4", "-pass", "xyz", "-out-dir", out, src} + + if err := catp.Main(); err != nil { + t.Fatal(err) + } + + var got strings.Builder + + for i := 0; i < 4; i++ { + b, err := os.ReadFile(out + "/foo.log.chunk" + strconv.Itoa(i)) + if err != nil { + t.Fatal(err) + } + + got.Write(b) + } + + if got.String() != want.String() { + t.Fatalf("chunked output differs: got %d bytes, want %d", got.Len(), want.Len()) + } + + resetFlags() + + os.Args = []string{"catp", "-split-chunks", "-start-line", "1", "-parallel", "2", src} + + if err := catp.Main(); err == nil { + t.Fatal("expected incompatible flags error") + } +} diff --git a/cmd/catp/catp/catp.go b/cmd/catp/catp/catp.go index f304352..34701c5 100644 --- a/cmd/catp/catp/catp.go +++ b/cmd/catp/catp/catp.go @@ -7,12 +7,14 @@ import ( "encoding/csv" "encoding/json" "fmt" + "html" "io" "log" "net" "net/http" "os" "path" + "regexp" "runtime/pprof" "strconv" "strings" @@ -23,6 +25,7 @@ import ( "github.com/bool64/progress" "github.com/klauspost/compress/zstd" gzip "github.com/klauspost/pgzip" + "github.com/vearutop/plot-go" "golang.org/x/time/rate" ) @@ -48,6 +51,10 @@ type runner struct { currentBytesUncompressed int64 currentLines int64 + // consumed counts uncompressed bytes of the current file passed through scanFile (nil if not tracked), + // it is more precise than bytes read by a buffered reader. + consumed *int64 + filters filters currentFile *progress.CountingReader @@ -69,6 +76,7 @@ type runner struct { statusMu sync.Mutex lastStatus string + speed []speedSample // Kept only with -http, for the status page charts. noProgress bool countLines bool @@ -81,6 +89,8 @@ type runner struct { options Options hasCompression bool + + splitChunks bool } type flagFunc func(v string) error @@ -215,7 +225,7 @@ func (r *runner) st(s progress.Status) string { if r.progressJSON != "" { pr.ElapsedSeconds = pr.Elapsed.Truncate(time.Second).Seconds() pr.RemainingSeconds = pr.Remaining.Round(time.Second).Seconds() - pr.BytesCompleted = currentBytes + pr.BytesCompleted = s.BytesCompleted // The shared r.currentBytes is not fed with a single reader. pr.BytesTotal = atomic.LoadInt64(&r.totalBytes) if j, err := json.Marshal(pr); err == nil { @@ -227,6 +237,11 @@ func (r *runner) st(s progress.Status) string { r.statusMu.Lock() r.lastStatus = res + + if r.httpAddr != "" { + r.addSample(speedSample{ts: time.Now().Unix(), lps: s.SpeedLPS, mbps: s.SpeedMBPS}) + } + r.statusMu.Unlock() return res @@ -241,10 +256,19 @@ func (r *runner) readFile(rd io.Reader, out io.Writer) { } } +// smallInputSize is the total input size below which progress is updated for every line. +const smallInputSize = 10 << 20 + // linesPush picks how many lines to batch before updating shared progress counters, // keeping a batch's fill time under the progress status interval so speed/remaining // estimates don't alias against a rate-limited throughput. func (r *runner) linesPush() int { + // Small inputs are cheap to update per line and would otherwise show no progress for a long time. + // Compressed input may expand a lot and STDIN has unknown size (-1), those keep batching. + if total := atomic.LoadInt64(&r.totalBytes); total >= 0 && total < smallInputSize && !r.hasCompression { + return 1 + } + if r.limiter == nil { return 1000 } @@ -294,8 +318,11 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer, sampleBy sampleSize = r.sizes[filename] } + var consumedLocal int64 + for s.Scan() { fileLines++ + consumedLocal += int64(len(s.Bytes()) + 1) if atomic.LoadInt64(&r.closed) > 0 { break @@ -347,6 +374,11 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer, sampleBy if lines >= linesPush { atomic.AddInt64(&r.currentLines, int64(lines)) lines = 0 + + if r.consumed != nil { + atomic.AddInt64(r.consumed, consumedLocal) + consumedLocal = 0 + } linesPush = r.linesPush() if flusher, ok := w.(interface { @@ -406,6 +438,10 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer, sampleBy atomic.AddInt64(&r.currentLines, int64(lines)) + if r.consumed != nil { + atomic.AddInt64(r.consumed, consumedLocal) + } + if err := s.Err(); err != nil { r.mu.Lock() defer r.mu.Unlock() @@ -414,6 +450,55 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer, sampleBy } } +const maxSamples = 7200 + +type speedSample struct { + ts int64 + lps float64 + mbps float64 +} + +// addSample must be called with statusMu held, one sample per second is kept. +func (r *runner) addSample(s speedSample) { + if n := len(r.speed); n > 0 && r.speed[n-1].ts >= s.ts { + return + } + + if len(r.speed) >= maxSamples { + r.speed = append(r.speed[:0], r.speed[len(r.speed)/2:]...) + } + + r.speed = append(r.speed, s) +} + +// speedJSON returns samples newer than ?since= as [ts, lines/s, MB/s] rows. +func (r *runner) speedJSON(w http.ResponseWriter, req *http.Request) { + since, _ := strconv.ParseInt(req.FormValue("since"), 10, 64) + rows := [][]any{} + + r.statusMu.Lock() + for _, s := range r.speed { + if s.ts > since { + rows = append(rows, []any{s.ts, s.lps, s.mbps}) + } + } + r.statusMu.Unlock() + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(rows) //nolint:errchkjson +} + +var donePercent = regexp.MustCompile(`(\d+(?:\.\d+)?)% bytes read`) + +// pageTitle shows completion in the tab title. +func pageTitle(status string) string { + if m := donePercent.FindStringSubmatch(status); m != nil { + return m[1] + "% catp" + } + + return "catp" +} + func formatLimit(lim rate.Limit) string { if lim == rate.Inf { return "unlimited" @@ -448,19 +533,68 @@ func (r *runner) startControlServer() error { _, _ = fmt.Fprintf(w, "%s\n", formatLimit(r.limiter.Limit())) }) + + mux.Handle("/plot/", http.StripPrefix("/plot/", plot.Handler())) + mux.HandleFunc("/speed.json", r.speedJSON) + + if r.options.HTTPMux != nil { + r.options.HTTPMux(mux) + } + mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) { r.statusMu.Lock() status := r.lastStatus r.statusMu.Unlock() - _, _ = fmt.Fprintf(w, ` -
%s
-

rate limit: %s lines/sec

+ if r.options.StatusPage != nil { + _, _ = io.WriteString(w, r.options.StatusPage(status, formatLimit(r.limiter.Limit()))) + + return + } + + _, _ = fmt.Fprintf(w, `%s%s + +
%s
+

rate limit: %s lines/sec

-`, status, formatLimit(r.limiter.Limit())) + +
+ +`, html.EscapeString(pageTitle(status)), plot.HeadTags("/plot/"), html.EscapeString(status), formatLimit(r.limiter.Limit())) }) var handler http.Handler = mux @@ -495,7 +629,13 @@ func basicAuth(user, pass string, next http.Handler) http.Handler { }) } -func (r *runner) cat(filename string) (err error) { +func (r *runner) cat(filename string) error { + return r.catChunk(filename, nil) +} + +// catChunk processes a file, or a byte range of it if c is not nil, in which case filename is +// the virtual name of the chunk. +func (r *runner) catChunk(filename string, c *chunk) (err error) { if r.samples > 0 { handled, reason, err := r.trySampleSeekable(filename) if handled { @@ -512,7 +652,12 @@ func (r *runner) cat(filename string) (err error) { if filename == "-" { rd = os.Stdin } else { - file, err := os.Open(filename) //nolint:gosec + path := filename + if c != nil { + path = c.path + } + + file, err := os.Open(path) //nolint:gosec if err != nil { return err } @@ -524,6 +669,12 @@ func (r *runner) cat(filename string) (err error) { }() rd = io.Reader(file) + + if c != nil { + if rd, err = newChunkReader(file, c.start, c.end); err != nil { + return err + } + } } // sampleBytes tracks compressed bytes read for this file only, independent of the shared @@ -588,6 +739,10 @@ func (r *runner) cat(filename string) (err error) { out = w } + if r.parallel <= 1 && !r.noProgress && !r.hasCompression { + r.consumed = new(int64) + } + if r.parallel <= 1 && !r.noProgress { r.pr.Start(func(t *progress.Task) { t.TotalBytes = func() int64 { @@ -595,7 +750,14 @@ func (r *runner) cat(filename string) (err error) { } t.CurrentBytes = r.currentFile.Bytes - t.CurrentLines = func() int64 { return atomic.LoadInt64(&r.currentLines) } + + // Counters of this file, Progress adds counters of previous files (Continue) on its own. + if consumed := r.consumed; consumed != nil { + t.CurrentBytes = func() int64 { return atomic.LoadInt64(consumed) } + } + + linesBase := atomic.LoadInt64(&r.currentLines) + t.CurrentLines = func() int64 { return atomic.LoadInt64(&r.currentLines) - linesBase } t.Task = filename t.Continue = true t.PrintOnStart = true @@ -672,12 +834,19 @@ func startProfiling(cpuProfile string, memProfile string) { }() } -// Options allows behavior customisations. +// Options allows behavior customizations. type Options struct { // PrepareLine is invoked for every line, if result is nil, line is skipped. // You can use buf to avoid allocations for a result, and change its capacity if needed. PrepareLine func(filename string, lineNr int, line []byte, buf *[]byte) []byte + // HTTPMux registers extra routes on the -http control server. + HTTPMux func(mux *http.ServeMux) + + // StatusPage replaces the -http status page, it receives plain text progress status (not HTML-escaped) + // and current rate limit, and returns a complete HTML page. + StatusPage func(status, rateLimit string) string + // VersionLabel is added to version message. VersionLabel string } diff --git a/cmd/catp/catp/chunks.go b/cmd/catp/catp/chunks.go new file mode 100644 index 0000000..b8b09b9 --- /dev/null +++ b/cmd/catp/catp/chunks.go @@ -0,0 +1,144 @@ +package catp + +import ( + "bufio" + "fmt" + "io" + "os" + "strconv" + "strings" +) + +// minChunkSize avoids splitting small files into pointless tiny chunks. +const minChunkSize = 1 << 20 + +type chunk struct { + path string + start, end int64 +} + +type job struct { + name string + chunk *chunk +} + +// jobs returns files as is, or, with -split-chunks, uncompressed files split into virtual +// "file.chunkNN" chunks. +func (r *runner) jobs(files []string) []job { + var res []job + + for _, fn := range files { + size := r.sizes[fn] + n := min(int64(r.parallel), size/minChunkSize) + + if !r.splitChunks || n < 2 || strings.HasSuffix(fn, ".gz") || strings.HasSuffix(fn, ".zst") { + res = append(res, job{name: fn}) + + continue + } + + width := len(strconv.FormatInt(n-1, 10)) + + for i := int64(0); i < n; i++ { + res = append(res, job{ + name: fmt.Sprintf("%s.chunk%0*d", fn, width, i), + chunk: &chunk{path: fn, start: size * i / n, end: size * (i + 1) / n}, + }) + } + } + + return res +} + +// chunkReader yields whole lines that start within [start, end) of a file: a line that +// straddles start belongs to the previous chunk, a line that straddles end belongs to this one. +type chunkReader struct { + br *bufio.Reader + left int64 // Bytes left before end. + last byte + tailDone bool +} + +func newChunkReader(f *os.File, start, end int64) (*chunkReader, error) { + cr := &chunkReader{last: '\n'} // Nothing read yet: no tail to complete. + pos := start + + if start > 0 { + // Seek one byte back to see whether start is exactly at the line beginning. + if _, err := f.Seek(start-1, io.SeekStart); err != nil { + return nil, err + } + + pos-- + } + + cr.br = bufio.NewReaderSize(f, 64*1024) + + if start > 0 { + for { + b, err := cr.br.ReadSlice('\n') + pos += int64(len(b)) + + if err == bufio.ErrBufferFull { + continue + } + + if err != nil && err != io.EOF { + return nil, err + } + + break + } + } + + cr.left = max(end-pos, 0) + + return cr, nil +} + +func (c *chunkReader) Read(p []byte) (int, error) { + if c.left > 0 { + n, err := c.br.Read(p[:min(int64(len(p)), c.left)]) + c.left -= int64(n) + + if n > 0 { + c.last = p[n-1] + } + + if err == io.EOF { + c.left, c.last = 0, '\n' + } + + return n, err + } + + if c.last == '\n' || c.tailDone { + return 0, io.EOF + } + + // Complete the line that straddles end. + n := 0 + for n < len(p) { + b, err := c.br.ReadByte() + if err != nil { + c.tailDone = true + + break + } + + p[n] = b + n++ + + if b == '\n' { + c.tailDone = true + + break + } + } + + if n == 0 { + return 0, io.EOF + } + + return n, nil +} diff --git a/go.mod b/go.mod index 0fd8d85..15d2653 100644 --- a/go.mod +++ b/go.mod @@ -1,12 +1,15 @@ module github.com/bool64/progress -go 1.25.0 +go 1.26.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.19.1 - github.com/klauspost/pgzip v1.2.6 - golang.org/x/time v0.15.0 + github.com/klauspost/compress v1.20.3 + github.com/klauspost/pgzip v1.3.0 + github.com/vearutop/plot-go v0.1.1 + golang.org/x/time v0.16.0 ) + +require github.com/vearutop/statigz v1.5.0 // indirect diff --git a/go.sum b/go.sum index 4cae43e..07b8d66 100644 --- a/go.sum +++ b/go.sum @@ -1,12 +1,22 @@ github.com/DataDog/zstd v1.5.7 h1:ybO8RBeh29qrxIhCA9E8gKY6xfONU9T6G6aP9DTKfLE= github.com/DataDog/zstd v1.5.7/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw= +github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA= +github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= 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.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.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= -golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +github.com/klauspost/compress v1.20.3 h1:zWNoFqq/lSuYQRim3dUXlXeA0IqwuZGicdG5ZTfY5j4= +github.com/klauspost/compress v1.20.3/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/klauspost/pgzip v1.3.0 h1:8t+KjjzD/0n/8vNlJGGZmTeK4njuwffOQvJm8tOSc8o= +github.com/klauspost/pgzip v1.3.0/go.mod h1:Xx7tAv5DH8ObLHUDQKXoE/nd8i/FYdNDTVMvi3SLTKI= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= +github.com/vearutop/plot-go v0.1.1 h1:TU9XyjsA6zbanlpjNS/Hq80HZNao4kJAfr1k+z9k3fU= +github.com/vearutop/plot-go v0.1.1/go.mod h1:u/PnA1M0q8hVeruxfyF9rgqlOdU+BT7z1L20OARhgTY= +github.com/vearutop/statigz v1.5.0 h1:FuWwZiT82yBw4xbWdWIawiP2XFTyEPhIo8upRxiKLqk= +github.com/vearutop/statigz v1.5.0/go.mod h1:oHmjFf3izfCO804Di1ZjB666P3fAlVzJEx2k6jNt/Gk= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= diff --git a/progress.go b/progress.go index 22ca2a6..4654e14 100644 --- a/progress.go +++ b/progress.go @@ -232,10 +232,15 @@ func (p *Progress) speedStatus(s *Status) { } } - if dc > 0 { + switch { + case dc > 0: s.Remaining = time.Duration((100.0 - s.DonePercent) * float64(ela) / dc) s.Remaining = s.Remaining.Truncate(time.Second) - } else { + case s.DonePercent > 0: + // No progress during this interval (coarse counters), fall back to the average since start. + s.Remaining = time.Duration(float64(100*s.Elapsed)/s.DonePercent) - s.Elapsed + s.Remaining = s.Remaining.Truncate(time.Second) + default: s.Remaining = 0 } From 607c031605724cb891f3929b03b7562d904c4d15 Mon Sep 17 00:00:00 2001 From: Viacheslav Poturaev Date: Thu, 22 Oct 2026 15:49:32 +0200 Subject: [PATCH 2/3] Fix lint --- .github/workflows/golangci-lint.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/golangci-lint.yml b/.github/workflows/golangci-lint.yml index 34c9ed4..8d24825 100644 --- a/.github/workflows/golangci-lint.yml +++ b/.github/workflows/golangci-lint.yml @@ -21,7 +21,7 @@ jobs: steps: - uses: actions/setup-go@v5 with: - go-version: stable + go-version: oldstable - uses: actions/checkout@v4 - name: golangci-lint uses: golangci/golangci-lint-action@v9.2.0 From 3aadf4f9bbf85ab46a3957ec4b2c603f2c590edc Mon Sep 17 00:00:00 2001 From: Viacheslav Poturaev Date: Thu, 22 Oct 2026 15:56:20 +0200 Subject: [PATCH 3/3] Fix lint --- cmd/catp/catp/app_test.go | 2 +- cmd/catp/catp/catp.go | 3 ++- cmd/catp/catp/chunks.go | 5 +++-- 3 files changed, 6 insertions(+), 4 deletions(-) diff --git a/cmd/catp/catp/app_test.go b/cmd/catp/catp/app_test.go index e2c7479..0167b6b 100644 --- a/cmd/catp/catp/app_test.go +++ b/cmd/catp/catp/app_test.go @@ -465,7 +465,7 @@ func Test_Main_splitChunks(t *testing.T) { var got strings.Builder - for i := 0; i < 4; i++ { + for i := range 4 { b, err := os.ReadFile(out + "/foo.log.chunk" + strconv.Itoa(i)) if err != nil { t.Fatal(err) diff --git a/cmd/catp/catp/catp.go b/cmd/catp/catp/catp.go index 34701c5..b899a28 100644 --- a/cmd/catp/catp/catp.go +++ b/cmd/catp/catp/catp.go @@ -379,6 +379,7 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer, sampleBy atomic.AddInt64(r.consumed, consumedLocal) consumedLocal = 0 } + linesPush = r.linesPush() if flusher, ok := w.(interface { @@ -473,7 +474,7 @@ func (r *runner) addSample(s speedSample) { // speedJSON returns samples newer than ?since= as [ts, lines/s, MB/s] rows. func (r *runner) speedJSON(w http.ResponseWriter, req *http.Request) { - since, _ := strconv.ParseInt(req.FormValue("since"), 10, 64) + since, _ := strconv.ParseInt(req.FormValue("since"), 10, 64) //nolint:gosec rows := [][]any{} r.statusMu.Lock() diff --git a/cmd/catp/catp/chunks.go b/cmd/catp/catp/chunks.go index b8b09b9..b592702 100644 --- a/cmd/catp/catp/chunks.go +++ b/cmd/catp/catp/chunks.go @@ -2,6 +2,7 @@ package catp import ( "bufio" + "errors" "fmt" "io" "os" @@ -39,7 +40,7 @@ func (r *runner) jobs(files []string) []job { width := len(strconv.FormatInt(n-1, 10)) - for i := int64(0); i < n; i++ { + for i := range n { res = append(res, job{ name: fmt.Sprintf("%s.chunk%0*d", fn, width, i), chunk: &chunk{path: fn, start: size * i / n, end: size * (i + 1) / n}, @@ -79,7 +80,7 @@ func newChunkReader(f *os.File, start, end int64) (*chunkReader, error) { b, err := cr.br.ReadSlice('\n') pos += int64(len(b)) - if err == bufio.ErrBufferFull { + if errors.Is(err, bufio.ErrBufferFull) { continue }