package pipeline
import (
"bufio"
"errors"
"io"
)
var ErrSkipLine = errors.New("skip line")
type LineFunc func(line []byte, out *bufio.Writer) error
const maxLine = 1 << 20 // 1MiB
func Transform(src io.Reader, dst io.Writer, fn LineFunc) error {
r := bufio.NewReaderSize(src, 64*1024)
w := bufio.NewWriterSize(dst, 64*1024)
scanner := bufio.NewScanner(r)
scanner.Buffer(make([]byte, 0, 64*1024), maxLine)
for scanner.Scan() {
err := fn(scanner.Bytes(), w)
if errors.Is(err, ErrSkipLine) {
continue
}
if err != nil {
return err
}
}
if err := scanner.Err(); err != nil {
return err
}
return w.Flush()
}package pipeline
import (
"bufio"
"strconv"
"strings"
)
func NewApacheParser(minMillis int) LineFunc {
return func(line []byte, out *bufio.Writer) error {
s := strings.TrimSpace(string(line))
if s == "" {
return ErrSkipLine
}
fields := strings.Fields(s)
if len(fields) < 11 {
return ErrSkipLine
}
status, err := strconv.Atoi(fields[8])
if err != nil || status < 500 {
return ErrSkipLine
}
millis, err := strconv.Atoi(strings.Trim(fields[len(fields)-1], `"`))
if err != nil || millis < minMillis {
return ErrSkipLine
}
path := strings.Trim(fields[6], `"`)
out.WriteString(`{"ip":"`)
out.WriteString(fields[0])
out.WriteString(`","status":`)
out.WriteString(strconv.Itoa(status))
out.WriteString(`,"path":"`)
out.WriteString(path)
out.WriteString(`","ms":`)
out.WriteString(strconv.Itoa(millis))
out.WriteByte('}')
return out.WriteByte('\n')
}
}package main
import (
"flag"
"fmt"
"io"
"os"
"example.com/logs/pipeline"
)
func main() {
slow := flag.Int("slow-ms", 1000, "only emit requests slower than this")
flag.Parse()
var src io.Reader = os.Stdin
if path := flag.Arg(0); path != "" {
f, err := os.Open(path)
if err != nil {
fmt.Fprintln(os.Stderr, "open:", err)
os.Exit(1)
}
defer f.Close()
src = f
}
parse := pipeline.NewApacheParser(*slow)
if err := pipeline.Transform(src, os.Stdout, parse); err != nil {
fmt.Fprintln(os.Stderr, "transform:", err)
os.Exit(1)
}
}This snippet shows how to process an arbitrarily large file without loading it into memory, by streaming it line-by-line through bufio.Scanner and pushing each transformed record straight to an output writer. The core idea is that a file of any size can be handled in constant memory as long as the program only ever holds one line (plus its scan buffer) at a time, rather than an entire slice of lines.
In transform.go, the Transform function wires a bufio.Reader to a bufio.Writer and drives a bufio.Scanner over the input. A key detail is scanner.Buffer(...): the default Scanner caps a token at 64KB and returns bufio.ErrTooLong on longer lines, which is a common surprise when parsing verbose log lines. Setting an explicit buffer and a larger maxLine raises that ceiling deliberately. Each line is passed to a pluggable LineFunc; returning ErrSkipLine lets a transform drop a record without aborting the whole stream, while any other error stops processing immediately. After the loop, scanner.Err() is checked separately because Scan() returning false means either clean EOF or a real error, and the two must be distinguished. The final w.Flush() is essential — a buffered writer holds bytes that would otherwise be silently lost.
In parser.go, NewApacheParser returns a concrete LineFunc that parses common-log-format access lines and re-emits only slow 5xx requests as compact JSON. It shows the realistic shape of a transform: guard clauses that return ErrSkipLine for blank or malformed lines, cheap field extraction with strings.Fields, and a per-line allocation kept small. Because the transform is just a function, it stays trivially unit-testable in isolation from the I/O plumbing.
In main.go, the command opens the input file (or reads os.Stdin), wires stdout as the sink, and applies the parser. Deferring f.Close() and surfacing the error from Transform gives a clean CLI. The trade-off of this design is that it is inherently sequential; if the transform were CPU-heavy, a worker pool fed by the scanner would parallelize it, but for I/O-bound log rewriting this single-goroutine pipeline is simpler and already saturates the disk. Reach for this pattern whenever data is line-oriented and too large to fit comfortably in RAM.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
Rails.application.configure do
config.after_initialize do
Bullet.enable = true
Bullet.alert = false
Bullet.bullet_logger = true
Bullet.console = trueN+1 query detection with Bullet gem
use clap::Parser;
#[derive(Parser, Debug)]
#[command(author, version, about)]
struct Args {
#[arg(short, long)]clap for CLI argument parsing with derive macros
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
