11 - Log Pipelines & Alerting

🎁 What if you could find the one error that crashed production at 3am, out of 10GB of daily logs, in seconds?

Logs are the narrative of your system. Every request, every error, every state change. But raw log files on disk are useless at scale. You need pipelines: tail the files, parse the lines, enrich with context, and ship them somewhere you can query and alert on. Go's concurrency model makes it ideal for building log agents that handle high throughput without dropping entries.

Imports in code snippets are trimmed for brevity. See the full example linked at the end for complete, compilable source.

Structured Logging for Machine Consumption

Unstructured logs (INFO: user logged in) are fine for humans staring at a terminal. They're terrible for machines. Structured logging outputs each entry as a JSON object with consistent fields:

package main

import (
	"log/slog"
	"os"
)

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
		Level: slog.LevelInfo,
	}))

	logger.Info("request completed",
		"method", "GET",
		"path", "/api/users",
		"status", 200,
		"duration_ms", 45,
		"user_id", "u-12345",
	)
}
// Output:
// {"time":"2026-07-20T10:00:00Z","level":"INFO","msg":"request completed","method":"GET","path":"/api/users","status":200,"duration_ms":45,"user_id":"u-12345"}

Every field is parseable. You can filter, aggregate, and alert on any dimension. This is the foundation of a log pipeline: your applications produce structured JSON, and downstream systems consume it.

Tail-Based Log Collection

The first stage of a log pipeline is collection. A log agent watches files (like fluentd or promtail) and streams new lines as they appear. In Go, this means seeking to the end of a file and polling for new content:

package tailer

import (
	"bufio"
	"context"
	"io"
	"os"
	"time"
)

type Tailer struct {
	path     string
	lines    chan string
	pollRate time.Duration
}

func New(path string, bufSize int) *Tailer {
	return &Tailer{
		path:     path,
		lines:    make(chan string, bufSize),
		pollRate: 100 * time.Millisecond,
	}
}

func (t *Tailer) Lines() <-chan string {
	return t.lines
}

func (t *Tailer) Run(ctx context.Context) error {
	f, err := os.Open(t.path)
	if err != nil {
		return err
	}
	defer f.Close()

	// Seek to end, only collect new lines
	if _, err := f.Seek(0, io.SeekEnd); err != nil {
		return err
	}

	reader := bufio.NewReader(f)

	for {
		select {
		case <-ctx.Done():
			close(t.lines)
			return nil
		default:
			line, err := reader.ReadString('\n')
			if err != nil {
				// No new data, wait and retry
				time.Sleep(t.pollRate)
				continue
			}
			if line != "" {
				t.lines <- line[:len(line)-1] // trim newline
			}
		}
	}
}

This polls the file for new content. For production, you'd add file rotation detection (inode changes), position checkpointing (resume after restart), and filesystem event watching (fsnotify) to reduce polling overhead.

Log Parsing and Enrichment

Raw lines need parsing. For JSON-structured logs, decode the fields. For unstructured logs, use regex. Then enrich with metadata the original application didn't include:

package pipeline

import (
	"encoding/json"
	"os"
	"regexp"
	"time"
)

type LogEntry struct {
	Timestamp time.Time         `json:"timestamp"`
	Level     string            `json:"level"`
	Message   string            `json:"msg"`
	Fields    map[string]any    `json:"fields"`
	Meta      map[string]string `json:"meta"`
}

// ParseJSON handles structured JSON log lines
func ParseJSON(line string) (*LogEntry, error) {
	var raw map[string]any
	if err := json.Unmarshal([]byte(line), &raw); err != nil {
		return nil, err
	}

	entry := &LogEntry{
		Fields: raw,
		Meta:   make(map[string]string),
	}

	if ts, ok := raw["time"].(string); ok {
		entry.Timestamp, _ = time.Parse(time.RFC3339, ts)
	}
	if lvl, ok := raw["level"].(string); ok {
		entry.Level = lvl
	}
	if msg, ok := raw["msg"].(string); ok {
		entry.Message = msg
	}

	return entry, nil
}

// ParseRegex handles unstructured lines with a regex pattern
var nginxPattern = regexp.MustCompile(
	`^(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) \S+" (?P<status>\d+) (?P<size>\d+)`,
)

func ParseNginx(line string) (*LogEntry, error) {
	matches := nginxPattern.FindStringSubmatch(line)
	if matches == nil {
		return nil, fmt.Errorf("no match")
	}

	entry := &LogEntry{
		Fields: make(map[string]any),
		Meta:   make(map[string]string),
	}
	for i, name := range nginxPattern.SubexpNames() {
		if i != 0 && name != "" {
			entry.Fields[name] = matches[i]
		}
	}
	return entry, nil
}

// Enrich adds host metadata to every entry
func Enrich(entry *LogEntry) {
	hostname, _ := os.Hostname()
	entry.Meta["hostname"] = hostname
	entry.Meta["agent"] = "go-log-agent"
	entry.Meta["collected_at"] = time.Now().UTC().Format(time.RFC3339)
}

Enrichment adds context that makes logs searchable at scale: which host produced this, when was it collected, which pipeline processed it.

Forwarding Logs to a Destination

Once parsed and enriched, entries need to reach a central system. A forwarder batches entries and ships them over HTTP. Batching is critical for throughput. One HTTP request with 100 entries beats 100 individual requests:

package forwarder

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"time"
)

type HTTPForwarder struct {
	endpoint   string
	client     *http.Client
	batch      []*LogEntry
	batchSize  int
	flushEvery time.Duration
}

func New(endpoint string, batchSize int, flushInterval time.Duration) *HTTPForwarder {
	return &HTTPForwarder{
		endpoint:   endpoint,
		client:     &http.Client{Timeout: 10 * time.Second},
		batch:      make([]*LogEntry, 0, batchSize),
		batchSize:  batchSize,
		flushEvery: flushInterval,
	}
}

func (f *HTTPForwarder) Send(entry *LogEntry) {
	f.batch = append(f.batch, entry)
	if len(f.batch) >= f.batchSize {
		f.flush()
	}
}

func (f *HTTPForwarder) RunFlusher(ctx context.Context) {
	ticker := time.NewTicker(f.flushEvery)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			f.flush() // final flush on shutdown
			return
		case <-ticker.C:
			f.flush()
		}
	}
}

func (f *HTTPForwarder) flush() {
	if len(f.batch) == 0 {
		return
	}

	payload, err := json.Marshal(f.batch)
	if err != nil {
		fmt.Printf("marshal error: %v\n", err)
		// Output: marshal error: json: unsupported type
		return
	}

	resp, err := f.client.Post(f.endpoint, "application/json", bytes.NewReader(payload))
	if err != nil {
		fmt.Printf("forward error: %v (will retry)\n", err)
		// Output: forward error: Post "http://loki:3100/api/push": dial tcp: connection refused (will retry)
		return // entries stay in batch for retry on next flush
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		fmt.Printf("flushed %d entries to %s\n", len(f.batch), f.endpoint)
		// Output: flushed 50 entries to http://loki:3100/api/push
		f.batch = f.batch[:0] // clear on success
	} else {
		fmt.Printf("forward failed: status %d\n", resp.StatusCode)
		// Output: forward failed: status 429
	}
}

Alert Rules and Threshold Detection

A log agent can also detect patterns and fire alerts. Count error occurrences in a sliding window and trigger when thresholds are crossed:

package alerting

import (
	"fmt"
	"sync"
	"time"
)

type AlertRule struct {
	Name      string
	Match     func(*LogEntry) bool
	Threshold int
	Window    time.Duration
	OnFire    func(count int)
}

type AlertEngine struct {
	rules   []AlertRule
	counts  map[string][]time.Time
	mu      sync.Mutex
}

func NewAlertEngine(rules []AlertRule) *AlertEngine {
	return &AlertEngine{
		rules:  rules,
		counts: make(map[string][]time.Time),
	}
}

func (e *AlertEngine) Evaluate(entry *LogEntry) {
	e.mu.Lock()
	defer e.mu.Unlock()

	now := time.Now()

	for _, rule := range e.rules {
		if !rule.Match(entry) {
			continue
		}

		// Record match timestamp
		e.counts[rule.Name] = append(e.counts[rule.Name], now)

		// Evict entries outside window
		cutoff := now.Add(-rule.Window)
		timestamps := e.counts[rule.Name]
		i := 0
		for i < len(timestamps) && timestamps[i].Before(cutoff) {
			i++
		}
		e.counts[rule.Name] = timestamps[i:]

		// Check threshold
		if len(e.counts[rule.Name]) >= rule.Threshold {
			rule.OnFire(len(e.counts[rule.Name]))
			e.counts[rule.Name] = nil // reset after firing
		}
	}
}

Usage:

rules := []AlertRule{
	{
		Name:      "high_error_rate",
		Match:     func(e *LogEntry) bool { return e.Level == "ERROR" },
		Threshold: 50,
		Window:    5 * time.Minute,
		OnFire: func(count int) {
			fmt.Printf("ALERT: %d errors in 5 minutes\n", count)
			// Output: ALERT: 50 errors in 5 minutes
		},
	},
}
engine := NewAlertEngine(rules)

Log Shipping Agent

A complete agent that tails log files, parses JSON entries, enriches them with metadata, evaluates alerting rules, and batch-ships entries to a remote endpoint (e.g., Grafana Loki). Uses goroutines for concurrent tailing, flushing, and graceful shutdown via OS signals.

Input: A log file path to tail, a Loki endpoint URL, and alerting rules configured in code.

Output: Continuously ships parsed and enriched log entries to the remote endpoint in batches, fires alerts when threshold rules are breached, and shuts down cleanly on SIGINT/SIGTERM.

Full source: examples/log-shipping-agent

Key Takeaways

Structured JSON logging is required for automated log pipelines. Every field must be machine-parseable.
A Go log agent uses goroutines to tail files, parse concurrently, and batch-flush to HTTP endpoints without blocking.
Batching is critical for throughput. Never send one log entry per HTTP request.
Sliding-window alert rules detect threshold breaches without an external alerting system.
Keep the pipeline stages (tail, parse, enrich, forward) as separate components for testability and reuse.
For production, add position checkpointing, file rotation handling, retry with backoff, and dead-letter queues for failed forwards.

🎁 What if Kubernetes could automatically restart your service when it's deadlocked, and stop sending traffic until it's fully ready?