11 - Log Pipelines & Alerting
📋 Jump to Takeaways🎁 What if you could find the one error that crashed production at 3am, out of 10GB of daily logs, in seconds?
It's 3am. Your phone is going off. The service is throwing errors. You SSH in and run tail -f /var/log/app.log. It scrolls too fast to read. You ctrl-C and try grep ERROR app.log. The file is 4GB. grep takes three minutes.
By the time you find the relevant lines, the incident is already 45 minutes old. Users have been getting errors for almost an hour. Your incident timeline will show a 45-minute gap between "first error" and "engineer began investigating" because you were reading a flat text file with grep.
Centralized structured logging would have let you filter by service, by error code, by user ID in a query that takes two seconds. You would have seen the first error in real time. You would have been investigating at minute two, not minute 45.
Building that pipeline is what this lesson is about.
This lesson adds devctl logs tail <file> to the tool — one command to tail a structured log file, enrich entries with metadata, and ship them to a central store with alerting built in.
Imports in code snippets are trimmed for brevity. See the full example linked at the end for complete, compilable source.
Why Not fmt.Println?
The simplest approach is also the worst one for production:
// Unstructured logging: fast to write, impossible to query
fmt.Println("ERROR: something went wrong")
fmt.Printf("user %s failed to authenticate: %v\n", userID, err)Those lines land in your log file as strings. To find all authentication errors for a specific user across 4GB, you need grep and regex and hope. To aggregate error rates, you need awk or a custom parser. To alert on error count, you need a tool that can parse your specific format.
The failure mode scales with your traffic. At 100 requests per second, one day's log file is multiple gigabytes. At 1000 requests per second, you cannot grep in real time. The log file becomes write-only storage: you put things in but can't get them out fast enough to matter.
Structured logging fixes this. Every line is a JSON object with consistent fields. Any tool can parse it. Any query can filter on any dimension. The cost is a few extra keystrokes per log call.
Structured Logging for Machine Consumption
Go 1.21 shipped log/slog, a structured logger in the standard library. Use it.
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 now a named key in a JSON object. Your log system can filter on user_id. Grafana Loki can answer "show me all errors from user u-12345 in the last 10 minutes" in milliseconds. grep cannot.
The authentication error from before becomes:
logger.Error("authentication failed",
"user_id", userID,
"reason", err.Error(),
"ip", r.RemoteAddr,
)
// Output: {"time":"2026-07-20T03:00:00Z","level":"ERROR","msg":"authentication failed","user_id":"u-12345","reason":"invalid token","ip":"203.0.113.42"}Gotcha: Never use slog.SetDefault in library code. If you call slog.SetDefault(logger), you change the global logger for every package in the binary. Libraries should accept a *slog.Logger as a parameter. Only applications should set the default.
Tail-Based Log Collection
Once your application writes structured JSON, something needs to read it and ship it somewhere central. That something is a log agent.
A log agent watches log files and streams new lines as they appear. The pattern: open the file, seek to the end, poll for new content, emit lines on a channel.
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, not the history
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 is the basic shape. It works until the log file rotates.
Gotcha: file rotation. Logrotate renames /var/log/app.log to /var/log/app.log.1 and creates a new /var/log/app.log. Your file descriptor still points to the old inode, the renamed file. You read EOF and stop polling. New entries are going to the new file, but your tailer is reading the old one and seeing nothing. Your pipeline silently drops every log line written after the rotation.
The fix is to check whether the file has been rotated by comparing inodes:
func (t *Tailer) Run(ctx context.Context) error {
f, err := os.Open(t.path)
if err != nil {
return err
}
defer f.Close()
info, err := f.Stat()
if err != nil {
return err
}
currentInode := inode(info)
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 == io.EOF {
// Check if the file was rotated
newInfo, statErr := os.Stat(t.path)
if statErr == nil && inode(newInfo) != currentInode {
// File was replaced. Reopen.
f.Close()
f, err = os.Open(t.path)
if err != nil {
return err
}
currentInode = inode(newInfo)
reader = bufio.NewReader(f)
} else {
time.Sleep(t.pollRate)
}
continue
}
if err != nil {
return err
}
if line != "" {
t.lines <- line[:len(line)-1]
}
}
}
}The inode() helper reads the inode number from os.FileInfo. This uses syscall.Stat_t which is Linux/macOS only — log agents typically run on Linux servers so this is fine, but it won't compile on Windows:
//go:build !windows
import "syscall"
func inode(info os.FileInfo) uint64 {
return info.Sys().(*syscall.Stat_t).Ino
}When the path's inode no longer matches your open file's inode, you reopen. No log lines lost.
Log Parsing and Enrichment
Raw lines need parsing. For JSON logs from slog, decode the fields. For nginx or other unstructured formats, use regex. After parsing, add metadata the original application didn't include.
package pipeline
import (
"encoding/json"
"fmt"
"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
}
// ParseNginx handles unstructured nginx access log lines
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 and pipeline 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)
}The hostname field is the one you'll be grateful for at 3am. When you're looking at errors in Grafana Loki, you want to know which host they came from. Without enrichment, all you have is what the application wrote.
Forwarding Logs to a Destination
Parsed and enriched entries need to reach a central store like Grafana Loki. One HTTP request per log entry would overwhelm any endpoint. Batch instead.
The forwarder receives *LogEntry values produced by the parser — in a real codebase these would be separate packages; in the example they share a single package for simplicity.
package forwarder
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"sync"
"time"
)
type HTTPForwarder struct {
mu sync.Mutex
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.mu.Lock()
defer f.mu.Unlock()
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.mu.Lock()
f.flush() // flush whatever remains before shutting down
f.mu.Unlock()
return
case <-ticker.C:
f.mu.Lock()
f.flush()
f.mu.Unlock()
}
}
}
// flush sends the current batch. Caller must hold f.mu.
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, 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]
} else {
fmt.Printf("forward failed: status %d\n", resp.StatusCode)
// Output: forward failed: status 429
}
}The final flush in RunFlusher is easy to miss, but critical. When your agent receives SIGTERM, the ctx is cancelled. Without the final flush, whatever is in the buffer at that moment is dropped. A brief network blip right before a deploy rotation can mean losing the last 30 seconds of pre-crash logs.
Alert Rules and Threshold Detection
A log agent can watch for patterns and fire alerts without a separate alerting system. Count matching entries in a sliding window and trigger when a threshold is 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
}
e.counts[rule.Name] = append(e.counts[rule.Name], now)
// Evict entries outside the 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:]
if len(e.counts[rule.Name]) >= rule.Threshold {
rule.OnFire(len(e.counts[rule.Name]))
e.counts[rule.Name] = nil // reset after firing to avoid alert storm
}
}
}Wire up a rule that fires on 50 errors in 5 minutes:
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)This runs inside your agent process, before logs even reach Loki. Latency from error to alert can be seconds, not the scrape interval of your monitoring system.
Putting It Together: the Pipeline
Each stage in the pipeline has a single job: tail, parse, enrich, forward, alert. Separate stages means you can test each one in isolation and swap out the destination without touching the parser.
Wire run as the devctl logs tail subcommand in Cobra:
func newLogsTailCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "tail <file>",
Short: "Tail and ship a log file",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
return run(ctx, args[0], defaultRules())
},
}
return cmd
}
var logsCmd = &cobra.Command{Use: "logs", Short: "Log pipeline commands"}
func init() {
logsCmd.AddCommand(newLogsTailCmd())
rootCmd.AddCommand(logsCmd)
}The agent wires them together with goroutines:
func run(ctx context.Context, logPath string, rules []alerting.AlertRule) error {
t := tailer.New(logPath, 1000)
fwd := forwarder.New("http://loki:3100/api/push", 50, 5*time.Second)
alertEngine := alerting.NewAlertEngine(rules)
go func() {
if err := t.Run(ctx); err != nil {
log.Printf("tailer: %v", err)
}
}()
go fwd.RunFlusher(ctx)
for line := range t.Lines() {
entry, err := pipeline.ParseJSON(line)
if err != nil {
continue // skip unparseable lines
}
pipeline.Enrich(entry)
alertEngine.Evaluate(entry)
fwd.Send(entry)
}
return nil
}Three goroutines. One for tailing (blocked on file I/O), one for flushing (blocked on timer), one for the main pipeline loop (CPU work, channel reads). They communicate through channels and a mutex-protected batch. No shared mutable state across goroutines.
func main() {
if len(os.Args) < 2 {
fmt.Fprintln(os.Stderr, "usage: log-agent <log-file>")
os.Exit(1)
}
logPath := os.Args[1]
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
rules := []alerting.AlertRule{
{
Name: "high_error_rate",
Match: func(e *pipeline.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)
},
},
}
if err := run(ctx, logPath, rules); err != nil {
fmt.Fprintf(os.Stderr, "agent error: %v\n", err)
os.Exit(1)
}
}The 3am scenario from the top: with this pipeline running, you'd have seen the first error in Loki within seconds. The alert would have fired at error 50, probably within two minutes of the incident starting. By the time your phone rang, you'd have had a pre-filtered view of exactly the errors that were occurring, grouped by error code, traceable by user ID.
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.
log/slogis the standard library choice in Go 1.21+. Don't useslog.SetDefaultin library code. - Flat text logs with grep fail at scale. At 4GB per day, finding one error takes minutes, not seconds.
- A tail-based log agent opens a file, seeks to the end, and polls for new content. It must handle file rotation by comparing inodes, or it silently stops reading when logrotate renames the file.
- Enrich every entry with hostname and collection timestamp. These fields are invisible when everything works and invaluable when something doesn't.
- Batch HTTP forwards. One request with 50 entries beats 50 requests with 1 entry.
- Always flush on shutdown. Entries in the buffer are lost if you don't drain before exit.
- Keep pipeline stages separate (tail, parse, enrich, forward) for testability. Each stage is one goroutine with one job.
🎁 devctl logs tail now ships your logs. Next up: devctl health check <service> — before you deploy, verify that every dependency your service needs is actually reachable.