Introduction#

Sometimes, the hard part of concurrent programming is not making things run in parallel. The hard part is keeping the flow of data understandable.

A pipeline is a simple way to do that. Instead of putting every responsibility inside one large loop, you split the work into small stages. Each stage receives values from a channel, does one transformation, and sends the result to the next stage.

1
source -> parse -> filter -> enrich -> sink

In Go, this pattern feels natural because goroutines and channels already give us the building blocks. A generator can create the initial stream, each pipeline stage can transform it, and a final consumer can collect or print the result.

This article continues the Go Patterns series after Producer-Consumer, Generator, and Worker Pool. The goal is not to build a framework. The goal is to learn how to structure data processing without turning one function into a pile of responsibilities.

When to Use#

Use the Pipeline Pattern when data moves through multiple steps:

• Processing log streams
• Reading, validating, and transforming CSV rows
• Cleaning API responses before storing them
• Building small ETL-like flows
• Splitting parsing, filtering, enrichment, and reporting into separate responsibilities

The pattern is especially useful when each step can be described as:

take a stream of values, transform or filter it, and return another stream of values.

Why Use It#

Pipelines are useful because they keep each stage focused.

• Separation of concerns: parsing, filtering, and reporting do not live in the same loop
• Composability: stages can be reused and reordered
• Natural backpressure: channels slow down upstream stages when downstream stages cannot keep up
• Testability: each stage can be tested with a small input channel
• Readable concurrency: the data flow is visible from the stage composition

The pattern is not magic. It is mostly discipline: one stage, one responsibility.

How It Works#

A pipeline stage usually has this shape:

1
func stage(in <-chan Input) <-chan Output {
2
    out := make(chan Output)
3

4
    go func() {
5
        defer close(out)
6

7
        for value := range in {
8
            out <- transform(value)
9
        }
10
    }()
11

12
    return out
13
}

The stage receives a read-only channel, creates its own output channel, starts a goroutine, then closes the output channel when the input channel is exhausted.

This gives us a chain:

1
raw := source()
2
parsed := parse(raw)
3
filtered := filter(parsed)
4
enriched := enrich(filtered)
5
sink(enriched)

Simple Example#

Before looking at logs, let’s start with a tiny pipeline:

1
numbers -> square -> keep even -> print

1
package main
2

3
import "fmt"
4

5
func numbers(max int) <-chan int {
6
    out := make(chan int)
7

8
    go func() {
9
        defer close(out)
10

11
        for i := 1; i <= max; i++ {
12
            out <- i
13
        }
14
    }()
15

16
    return out
17
}
18

19
func square(in <-chan int) <-chan int {
20
    out := make(chan int)
21

22
    go func() {
23
        defer close(out)
24

25
        for n := range in {
26
            out <- n * n
27
        }
28
    }()
29

30
    return out
31
}
32

33
func keepEven(in <-chan int) <-chan int {
34
    out := make(chan int)
35

36
    go func() {
37
        defer close(out)
38

39
        for n := range in {
40
            if n%2 == 0 {
41
                out <- n
42
            }
43
        }
44
    }()
45

46
    return out
47
}
48

49
func main() {
50
    values := numbers(10)
51
    squared := square(values)
52
    evenSquares := keepEven(squared)
53

54
    for n := range evenSquares {
55
        fmt.Println(n)
56
    }
57
}

Each function owns one small part of the work. numbers produces values, square transforms them, keepEven filters them, and main consumes the final stream.

That is the pipeline pattern in its smallest useful form.

Real-World Example: Log Processing Pipeline#

Now let’s use a more realistic example.

Imagine we receive raw log lines and want to turn them into useful information. We can model that as a pipeline:

1
raw log lines -> parse logs -> filter errors -> enrich logs -> print report

For a complete runnable version, this example needs:

1
import (
2
    "fmt"
3
    "strings"
4
    "time"
5
)

First, define the data we want to pass between stages:

1
type RawLog string
2

3
type LogEntry struct {
4
    Timestamp time.Time
5
    Level     string
6
    Service   string
7
    Message   string
8
    Alert     bool
9
}

The source stage sends raw log lines:

1
func logSource(lines []string) <-chan RawLog {
2
    out := make(chan RawLog)
3

4
    go func() {
5
        defer close(out)
6

7
        for _, line := range lines {
8
            out <- RawLog(line)
9
        }
10
    }()
11

12
    return out
13
}

The parser turns each raw line into a structured LogEntry:

1
func parseLogs(in <-chan RawLog) <-chan LogEntry {
2
    out := make(chan LogEntry)
3

4
    go func() {
5
        defer close(out)
6

7
        for raw := range in {
8
            parts := strings.SplitN(string(raw), "|", 4)
9
            if len(parts) != 4 {
10
                continue
11
            }
12

13
            timestamp, err := time.Parse(time.RFC3339, parts[0])
14
            if err != nil {
15
                continue
16
            }
17

18
            out <- LogEntry{
19
                Timestamp: timestamp,
20
                Level:     parts[1],
21
                Service:   parts[2],
22
                Message:   parts[3],
23
            }
24
        }
25
    }()
26

27
    return out
28
}

The filter stage keeps only errors:

1
func filterErrors(in <-chan LogEntry) <-chan LogEntry {
2
    out := make(chan LogEntry)
3

4
    go func() {
5
        defer close(out)
6

7
        for entry := range in {
8
            if entry.Level == "ERROR" {
9
                out <- entry
10
            }
11
        }
12
    }()
13

14
    return out
15
}

The enrichment stage adds a small piece of derived information:

1
func enrichLogs(in <-chan LogEntry) <-chan LogEntry {
2
    out := make(chan LogEntry)
3

4
    go func() {
5
        defer close(out)
6

7
        for entry := range in {
8
            entry.Alert = entry.Service == "payment" || entry.Service == "auth"
9
            out <- entry
10
        }
11
    }()
12

13
    return out
14
}

Finally, the sink consumes the enriched entries:

1
func printReport(in <-chan LogEntry) {
2
    for entry := range in {
3
        alert := ""
4
        if entry.Alert {
5
            alert = " [ALERT]"
6
        }
7

8
        fmt.Printf("%s %s: %s%s\n", entry.Service, entry.Level, entry.Message, alert)
9
    }
10
}

The full pipeline becomes very readable:

1
func main() {
2
    lines := []string{
3
        "2026-04-24T10:00:00Z|INFO|api|request completed",
4
        "2026-04-24T10:00:01Z|ERROR|payment|card authorization failed",
5
        "2026-04-24T10:00:02Z|ERROR|worker|job timeout",
6
        "2026-04-24T10:00:03Z|ERROR|auth|invalid token",
7
    }
8

9
    raw := logSource(lines)
10
    parsed := parseLogs(raw)
11
    errors := filterErrors(parsed)
12
    enriched := enrichLogs(errors)
13

14
    printReport(enriched)
15
}

The important part is not the log format. The important part is that the flow is explicit.

Each stage can be read, tested, and replaced independently.

Error Handling#

The log parser above silently skips invalid lines. That keeps the example small, but it is not always what you want in production.

Two common approaches are:

1. Send errors to a separate error channel
2. Pass a result type through the pipeline

For example:

1
type LogResult struct {
2
    Entry LogEntry
3
    Err   error
4
}

This makes failures explicit without panicking inside a goroutine. It also lets the final consumer decide whether to log, count, retry, or ignore invalid records.

Cancellation#

The examples above work for finite inputs. For long-running pipelines, use context.Context so every stage can stop when the caller is done.

The shape usually looks like this:

1
func parseLogs(ctx context.Context, in <-chan RawLog) <-chan LogEntry {
2
    out := make(chan LogEntry)
3

4
    go func() {
5
        defer close(out)
6

7
        for {
8
            select {
9
            case <-ctx.Done():
10
                return
11
            case raw, ok := <-in:
12
                if !ok {
13
                    return
14
                }
15

16
                entry, ok := parseLog(raw)
17
                if ok {
18
                    out <- entry
19
                }
20
            }
21
        }
22
    }()
23

24
    return out
25
}

Without cancellation, a pipeline that reads from a never-ending source can leak goroutines when the consumer stops early.

Best Practices and Pitfalls#

Best Practices:

1. Keep each stage focused on one responsibility
2. Return receive-only channels (<-chan T) from stages
3. Close the output channel from the goroutine that writes to it
4. Use context.Context for long-running or cancellable pipelines
5. Test each stage independently with small input channels

Pitfalls:

1. Forgetting to close output channels
2. Stopping early without cancelling upstream stages
3. Creating too many tiny stages that hide simple logic
4. Mixing parsing, filtering, enrichment, and reporting in one function
5. Assuming ordering will stay the same if you later parallelize a stage

Related Patterns#

• Generator Pattern: Creates the initial stream of values
• Producer-Consumer Pattern: Separates production from consumption
• Worker Pool Pattern: Parallelizes expensive stages
• Fan-Out/Fan-In Pattern: Distributes one stage across multiple workers and merges the results

Summary#

The Pipeline Pattern is one of the most readable ways to structure data processing in Go. It lets you split a flow into small stages, connect them with channels, and keep each responsibility isolated.

It works well when data naturally moves through a sequence: read, parse, filter, enrich, report.

The pattern is also a bridge to more advanced concurrency designs. Once one stage becomes too slow, you can combine a pipeline with a Worker Pool or Fan-Out/Fan-In to parallelize only that part of the flow.

Series Navigation#

This article is part of the Go Patterns series:

• Previous: Mastering the Worker Pool Pattern in Go
• Series: Go Patterns

If you want to experiment with the code examples, you can find them on my GitHub repository.