42. Capstone: a streaming log analyzer for multi-GB files

Goal: parse and aggregate a large access log in constant memory, concurrently, using everything from Chapters 22, 27, and 29.

package main

import (
	"bufio"
	"flag"
	"fmt"
	"os"
	"regexp"
	"sort"
	"sync"
)

var logLineRe = regexp.MustCompile(`^(\S+) \S+ \S+ \[.*?\] "(\S+) (\S+).*?" (\d{3}) (\d+)`)

type stats struct {
	mu         sync.Mutex
	byStatus   map[string]int
	byPath     map[string]int
	totalBytes int64
}

func newStats() *stats {
	return &stats{byStatus: map[string]int{}, byPath: map[string]int{}}
}

func (s *stats) record(status, path string, bytes int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.byStatus[status]++
	s.byPath[path]++
	s.totalBytes += bytes
}

func processLine(line string, s *stats) {
	m := logLineRe.FindStringSubmatch(line)
	if m == nil {
		return
	}
	status := m[4]
	path := m[3]
	var bytes int64
	fmt.Sscanf(m[5], "%d", &bytes)
	s.record(status, path, bytes)
}

func main() {
	path := flag.String("file", "", "log file to analyze")
	workers := flag.Int("workers", 4, "concurrent parsing workers")
	flag.Parse()

	f, err := os.Open(*path)
	if err != nil {
		fmt.Fprintln(os.Stderr, "error:", err)
		os.Exit(1)
	}
	defer f.Close()

	s := newStats()
	lines := make(chan string, 1000)

	var wg sync.WaitGroup
	for i := 0; i < *workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for line := range lines {
				processLine(line, s)
			}
		}()
	}

	scanner := bufio.NewScanner(f)
	scanner.Buffer(make([]byte, 1024*1024), 1024*1024) // handle long lines
	for scanner.Scan() {
		lines <- scanner.Text()
	}
	close(lines)
	wg.Wait()

	printReport(s)
}

func printReport(s *stats) {
	fmt.Println("=== Status codes ===")
	for status, count := range sortedByCount(s.byStatus) {
		fmt.Printf("%-6s %d\n", status, count)
	}
	fmt.Println("\n=== Top paths ===")
	top := sortedByCount(s.byPath)
	i := 0
	for path, count := range top {
		if i >= 10 {
			break
		}
		fmt.Printf("%-40s %d\n", path, count)
		i++
	}
	fmt.Printf("\nTotal bytes served: %d\n", s.totalBytes)
}

func sortedByCount(m map[string]int) map[string]int {
	type kv struct{ k string; v int }
	pairs := make([]kv, 0, len(m))
	for k, v := range m {
		pairs = append(pairs, kv{k, v})
	}
	sort.Slice(pairs, func(i, j int) bool { return pairs[i].v > pairs[j].v })
	out := make(map[string]int, len(pairs))
	for _, p := range pairs {
		out[p.k] = p.v
	}
	return out
}

go run . -file access.log -workers 8

The file is read line-by-line (never fully loaded), lines fan out across worker goroutines through a buffered channel, and a single mutex guards the shared aggregation maps — the same architecture as Chapter 25's URL checker, applied to a CPU-bound (regex parsing) workload instead of an I/O-bound one. (sortedByCount's Go-map-preserves-insertion-illusion is sidestepped by returning a slice-derived ordering for display, not by relying on map order — see Chapter 7.3.)