Part 20 - Observability: Seeing What the Pipeline Is Actually Doing

Part of the series: Production-Grade Concurrent AI Systems in Go

→ Full code for this post: github.com/madmmas/go-concurrent-ai-systems/tree/part-20 → Diff from Part 19: compare/part-19...part-20 → Run it: go run ./cmd/news-processor -articles=15 -workers=5 inside arc-2-production/part-20-observability

Parts 13 through 15 added three resilience patterns: retries, rate limiting, and circuit breaking. Part 16 added backpressure. Part 18 showed how to detect goroutine leaks. All of these mechanisms exist to handle failure, and they work — but they work silently. When the circuit breaker opens, nothing in the pipeline tells you. When the retry rate climbs from 2% to 18%, nothing alerts you. When latency at the embed stage doubles over two days, nothing shows you the trend.

A concurrent pipeline that cannot be observed is a system you cannot operate. You know it is running. You do not know if it is healthy.

Part 20 adds the three metrics that matter most: throughput, latency percentiles, and error breakdown by type. These are not the only metrics a production system needs, but they are the ones that answer the three questions that matter at 3 a.m.: Is it processing articles? How fast? What is failing?

The Three Metrics

Throughput — articles per second processed. This tells you whether the pipeline is keeping up with the rate of incoming articles. If throughput drops while article volume stays constant, something is wrong: a slow provider, a circuit breaker that opened, a worker pool that is undersized.

Latency percentiles — p50, p95, p99 measured per article. The median (p50) tells you what a typical article experiences. The p95 tells you what 5% of articles experience that the median does not. The p99 is the tail — the worst 1% of articles, which often reveals timeout behaviour, slow retries, or a provider that is inconsistently slow.

The reason you need percentiles rather than averages: a pipeline where 90% of articles finish in 100ms and 10% hit a 3-second timeout has an average latency of around 390ms. That average looks acceptable. The p95 of 3 seconds does not.

Error breakdown by type — counts grouped by error message. Seeing 4 failures is useful. Seeing map[llm: server error (503): 4] is actionable — you know exactly which provider error type is occurring and can correlate it with provider status pages, recent deployments, or time-of-day patterns.

The Metrics Struct

type Metrics struct {
    mu           sync.Mutex
    processed    int64
    failed       int64
    latencies    []float64 // milliseconds per article
    errorsByType map[string]int64
    startTime    time.Time
}

processed and failed use sync/atomic operations — atomic.AddInt64 — rather than a mutex. Atomic operations on integers are cheaper than mutex lock/unlock for counters that are incremented from many goroutines simultaneously. The tradeoff is that atomic operations only work on primitive types; the latencies slice and errorsByType map still need a mutex.

Record is called after every article completes:

func (m *Metrics) Record(dur time.Duration, err error) {
    if err != nil {
        atomic.AddInt64(&m.failed, 1)
        m.mu.Lock()
        m.errorsByType[err.Error()]++
        m.mu.Unlock()
    } else {
        atomic.AddInt64(&m.processed, 1)
    }
    m.mu.Lock()
    m.latencies = append(m.latencies, float64(dur.Milliseconds()))
    m.mu.Unlock()
}

The duration recorded is the wall-clock time for the full article — from when the worker picks it up to when the result is produced, including any LLM call latency, timeout waits, or retry delays. This gives a complete picture of what the article experienced, not just the LLM call time in isolation.

Computing Percentiles

func percentile(sorted []float64, p float64) float64 {
    if len(sorted) == 0 { return 0 }
    idx := int(math.Ceil(p*float64(len(sorted)))) - 1
    if idx < 0 { idx = 0 }
    if idx >= len(sorted) { idx = len(sorted) - 1 }
    return sorted[idx]
}

The percentile is computed from a sorted slice. p=0.95 with 15 articles means the index is ceil(0.95 × 15) - 1 = ceil(14.25) - 1 = 15 - 1 = 14 — the 15th article in sorted order, which is the maximum. With more articles, the p95 and p99 separate from the maximum more meaningfully.

The Summary method sorts the latencies, computes all three percentiles, and formats a report:

func (m *Metrics) Summary() string {
    // ... sort latencies, compute throughput, compute percentiles ...
    return fmt.Sprintf(
        "Processed: %d | Failed: %d (%.1f%%) | Throughput: %.1f/s\n"+
            "Latency p50: %.0fms | p95: %.0fms | p99: %.0fms\n"+
            "Errors: %v",
        processed, failed, errRate, throughput,
        p50, p95, p99,
        errTypes,
    )
}

What the Output Shows

Healthy pipeline — no errors, all articles succeed:

═════════════════ METRICS ════════════════════
Processed: 15 | Failed: 0 (0.0%) | Throughput: 14.7/s
Latency p50: 333ms | p95: 398ms | p99: 398ms
Errors: map[]
══════════════════════════════════════════════

14.7 articles per second with 5 workers. Latency is consistent — the p95 and p99 are both 398ms, close to the p50 of 333ms. This is a healthy pipeline: fast, consistent, no errors.

With 30% server error rate:

═════════════════ METRICS ════════════════════
Processed: 11 | Failed: 4 (26.7%) | Throughput: 25.7/s
Latency p50: 161ms | p95: 355ms | p99: 355ms
Errors: map[llm: server error (503):4]
══════════════════════════════════════════════

Four things stand out. Throughput is higher (25.7/s vs 14.7/s) even though 4 articles failed — failed articles return immediately with an error rather than waiting for a simulated LLM call, so they complete faster. This is the correct behaviour: the pipeline is not slower because some articles failed.

Latency is lower at p50 (161ms vs 333ms) for the same reason — some articles completed in near-zero time by failing fast. But the p95 climbs to 355ms, showing the successful articles are running at normal latency while the failures skew the median down.

The error map map[llm: server error (503):4] is the most actionable output. You know the failure mode (503), the count (4), and the error string (which maps directly to the error type in the simulator and, in production, to the HTTP status code). If this were rate limit exceeded (429) instead, you would be looking at rate limiter configuration. If it were context deadline exceeded, you would be looking at timeout configuration or a slow provider.

Wiring Metrics Into the Pool

The ObservablePool wraps the worker pool and records metrics for every article:

func (p *ObservablePool) ProcessAll(ctx context.Context, articles []model.Article) ([]model.AIResult, time.Duration) {
    // ... worker pool setup ...

    go func() {
        defer wg.Done()
        for article := range jobs {
            // ...
            articleStart := time.Now()
            articleCtx, cancel := context.WithTimeout(ctx, p.Timeout)
            result := p.processArticle(articleCtx, article)
            cancel()

            // Record after every article — success or failure
            p.Metrics.Record(time.Since(articleStart), result.Err)
            resultsCh <- result
        }
    }()
    // ...
}

The Record call happens inside the worker goroutine, after processArticle returns. This means the latency includes everything the article experienced: LLM call time, any waits inside the call, context timeout expiry. Workers are concurrent, so Record may be called from multiple goroutines simultaneously — which is why the mutex in Metrics is necessary.

The Path to Prometheus

The Metrics struct in Part 20 is an in-process implementation. In production you would replace it with Prometheus counters and histograms:

// Production equivalent
var (
    articlesProcessed = prometheus.NewCounterVec(
        prometheus.CounterOpts{Name: "articles_processed_total"},
        []string{"status"},
    )
    articleLatency = prometheus.NewHistogram(
        prometheus.HistogramOpts{
            Name:    "article_processing_seconds",
            Buckets: []float64{0.1, 0.25, 0.5, 1.0, 2.5, 5.0},
        },
    )
)

// In Record():
if err != nil {
    articlesProcessed.WithLabelValues("error").Inc()
} else {
    articlesProcessed.WithLabelValues("success").Inc()
}
articleLatency.Observe(dur.Seconds())

The API surface is the same — Record(dur, err) — but the data is now scraped by Prometheus every 15 seconds, stored in time series, and queryable via PromQL. The percentiles are computed by Prometheus from the histogram buckets rather than sorted in-process. Grafana dashboards show the trends over time rather than the snapshot at the end of a run.

OpenTelemetry follows the same pattern with slightly different types. The structural change — adding a Record call after every article — is identical. Changing from in-process metrics to Prometheus to OpenTelemetry requires changing only the Metrics implementation, not the pipeline code.

Arc 2 Complete

Part 20 closes Arc 2. Looking back at the full arc:

The pipeline that comes out of Arc 2 is meaningfully different from the one that entered it. It can process articles concurrently at the task level and the stage level. It recovers from transient failures. It respects provider rate limits proactively. It stops hammering broken providers. It keeps memory bounded. It handles streaming responses. It does not leak goroutines. It produces a grounded answer via RAG. And it can tell you exactly what it is doing.

Arc 3 keeps the pipeline in a single process and goes deeper. It covers the scheduler that runs every goroutine, the sync primitives we haven't used yet, semaphores, singleflight, structured logging with live profiling, and tracing. Everything still runs with go run and no infrastructure. Moving the pipeline into Kafka and Kubernetes waits until Arc 4.

This is Part 20 of the series "Production-Grade Concurrent AI Systems in Go," and the final part of Arc 2 — Production Concurrent AI Systems. Read Part 19 — Concurrent RAG Pipeline or continue to Part 21 — Go Scheduler Internals, the start of Arc 3 — Advanced Go Concurrency Engineering.