Building a Reliable Data Pipeline with Go, PostgreSQL, and Elasticsearch: From Raw Data to Real-Time Search

Introduction: The Data Synchronization Challenge

Modern applications increasingly combine relational databases with full-text search engines. PostgreSQL reliably stores transactional data, but its full-text search capabilities are limited compared to Elasticsearch. Synchronizing PostgreSQL and Elasticsearch is one of the key architectural challenges for backend developers in 2026.

The straightforward approach of "writing to both systems simultaneously from the application" is unreliable: if one operation fails, the data diverges. The proper solution is a dedicated data pipeline that tracks changes in PostgreSQL and atomically replicates them to Elasticsearch. That is exactly the kind of pipeline we will build in this article using Go.

Pipeline Architecture: Polling, Triggers, and CDC

There are three main approaches to synchronizing data between PostgreSQL and Elasticsearch:

• Polling — periodically querying tables using an updated_at field. Simple to implement, but introduces latency and misses deletions.
• Database Triggers — PostgreSQL triggers write changes to an outbox table. More reliable than polling, but increases database load.
• Change Data Capture (CDC) — reading the PostgreSQL WAL (Write-Ahead Log). Minimal load, full support for all operations (INSERT, UPDATE, DELETE), and millisecond-level latency.

For production systems, the optimal choice is CDC via PostgreSQL logical replication. It is the only approach that guarantees data completeness without adding overhead to the application. This is what we will implement in Go.

Change Data Capture with PostgreSQL: Setting Up Logical Replication

To work with CDC, you need to enable logical replication in PostgreSQL. Edit postgresql.conf:

wal_level = logical\nmax_replication_slots = 4\nmax_wal_senders = 4

Create a replication slot and a publication for the required tables:

-- Create publication\nCREATE PUBLICATION products_pub FOR TABLE products, categories;\n\n-- Create logical replication slot\nSELECT pg_create_logical_replication_slot('pipeline_slot', 'pgoutput');

To read WAL events in Go, we use the pglogrepl library — a pure Go client for the PostgreSQL logical replication protocol. An alternative is Debezium (a JVM-based solution), which is convenient if your infrastructure already uses Kafka. In our case, Go + pglogrepl provides minimal dependencies and maximum control.

Implementing the Pipeline in Go

Project Structure

We organize the code following clean architecture principles:

pipeline/\n├── cmd/\n│   └── pipeline/\n│       └── main.go\n├── internal/\n│   ├── cdc/\n│   │   ├── reader.go       # Reading WAL events\n│   │   └── decoder.go      # Decoding pgoutput\n│   ├── transform/\n│   │   └── mapper.go       # Mapping PG → ES documents\n│   ├── indexer/\n│   │   └── elasticsearch.go # Batch indexing\n│   └── metrics/\n│       └── prometheus.go   # Metrics\n├── docker-compose.yml\n└── Dockerfile

Reading WAL Events

The main worker connects to PostgreSQL via the replication protocol and reads the stream of changes:

package cdc\n\nimport (\n    "context"\n    "fmt"\n    "time"\n\n    "github.com/jackc/pglogrepl"\n    "github.com/jackc/pgx/v5/pgconn"\n    "github.com/jackc/pgx/v5/pgproto3"\n)\n\ntype WALReader struct {\n    conn        *pgconn.PgConn\n    slotName    string\n    publication string\n    lsn         pglogrepl.LSN\n}\n\nfunc NewWALReader(dsn, slotName, publication string) (*WALReader, error) {\n    conn, err := pgconn.Connect(context.Background(), dsn+" replication=database")\n    if err != nil {\n        return nil, fmt.Errorf("connect replication: %w", err)\n    }\n    return &WALReader{\n        conn:        conn,\n        slotName:    slotName,\n        publication: publication,\n    }, nil\n}\n\nfunc (r *WALReader) Start(ctx context.Context, events chan<- *WALEvent) error {\n    opts := pglogrepl.StartReplicationOptions{\n        PluginArgs: []string{\n            "proto_version '1'",\n            fmt.Sprintf("publication_names '%s'", r.publication),\n        },\n    }\n    if err := pglogrepl.StartReplication(ctx, r.conn, r.slotName, r.lsn, opts); err != nil {\n        return fmt.Errorf("start replication: %w", err)\n    }\n\n    standbyDeadline := time.Now().Add(10 * time.Second)\n    for {\n        if time.Now().After(standbyDeadline) {\n            if err := pglogrepl.SendStandbyStatusUpdate(ctx, r.conn,\n                pglogrepl.StandbyStatusUpdate{WALWritePosition: r.lsn}); err != nil {\n                return fmt.Errorf("standby status: %w", err)\n            }\n            standbyDeadline = time.Now().Add(10 * time.Second)\n        }\n\n        ctx2, cancel := context.WithDeadline(ctx, standbyDeadline)\n        msg, err := r.conn.ReceiveMessage(ctx2)\n        cancel()\n        if err != nil {\n            if pgconn.Timeout(err) {\n                continue\n            }\n            return fmt.Errorf("receive message: %w", err)\n        }\n\n        switch m := msg.(type) {\n        case *pgproto3.CopyData:\n            if m.Data[0] == pglogrepl.XLogDataByteID {\n                xld, err := pglogrepl.ParseXLogData(m.Data[1:])\n                if err != nil {\n                    return fmt.Errorf("parse xlog: %w", err)\n                }\n                event, err := DecodeWALData(xld.WALData)\n                if err == nil && event != nil {\n                    events <- event\n                    r.lsn = xld.WALStart + pglogrepl.LSN(len(xld.WALData))\n                }\n            }\n        }\n    }\n}

Batch Indexing in Elasticsearch

For efficient indexing, we use the Elasticsearch Bulk API. Events are accumulated in a buffer and flushed based on size or a timeout:

package indexer\n\nimport (\n    "bytes"\n    "context"\n    "encoding/json"\n    "fmt"\n    "time"\n\n    "github.com/elastic/go-elasticsearch/v8"\n    "github.com/elastic/go-elasticsearch/v8/esapi"\n)\n\ntype BulkIndexer struct {\n    client    *elasticsearch.Client\n    index     string\n    batchSize int\n    flushInterval time.Duration\n    buf       []BulkAction\n}\n\ntype BulkAction struct {\n    ID     string\n    Doc    map[string]interface{}\n    Delete bool\n}\n\nfunc (bi *BulkIndexer) Run(ctx context.Context, actions <-chan BulkAction) error {\n    ticker := time.NewTicker(bi.flushInterval)\n    defer ticker.Stop()\n\n    for {\n        select {\n        case action, ok := <-actions:\n            if !ok {\n                return bi.flush(ctx)\n            }\n            bi.buf = append(bi.buf, action)\n            if len(bi.buf) >= bi.batchSize {\n                if err := bi.flush(ctx); err != nil {\n                    return err\n                }\n            }\n        case <-ticker.C:\n            if len(bi.buf) > 0 {\n                if err := bi.flush(ctx); err != nil {\n                    return err\n                }\n            }\n        case <-ctx.Done():\n            return ctx.Err()\n        }\n    }\n}\n\nfunc (bi *BulkIndexer) flush(ctx context.Context) error {\n    if len(bi.buf) == 0 {\n        return nil\n    }\n    var body bytes.Buffer\n    for _, action := range bi.buf {\n        if action.Delete {\n            meta := map[string]interface{}{"delete": map[string]interface{}{"_index": bi.index, "_id": action.ID}}\n            line, _ := json.Marshal(meta)\n            body.Write(line)\n            body.WriteByte('\n')\n        } else {\n            meta := map[string]interface{}{"index": map[string]interface{}{"_index": bi.index, "_id": action.ID}}\n            line, _ := json.Marshal(meta)\n            body.Write(line)\n            body.WriteByte('\n')\n            doc, _ := json.Marshal(action.Doc)\n            body.Write(doc)\n            body.WriteByte('\n')\n        }\n    }\n    req := esapi.BulkRequest{Body: &body}\n    res, err := req.Do(ctx, bi.client)\n    if err != nil {\n        return fmt.Errorf("bulk request: %w", err)\n    }\n    defer res.Body.Close()\n    if res.IsError() {\n        return fmt.Errorf("bulk response error: %s", res.Status())\n    }\n    bi.buf = bi.buf[:0]\n    return nil\n}

Delivery Guarantees: At-Least-Once and Idempotency

WAL replication in PostgreSQL guarantees at-least-once delivery: when the pipeline restarts, events may be read more than once. The Elasticsearch Bulk API with an explicit document _id ensures idempotency — re-indexing the same document is safe.

It is critical to save the LSN (Log Sequence Number) after a batch has been successfully sent to Elasticsearch, not after reading from the WAL. Store the LSN in a dedicated PostgreSQL table or in Redis:

// Save LSN after a successful flush\nfunc saveLSN(ctx context.Context, db *pgxpool.Pool, slotName string, lsn pglogrepl.LSN) error {\n    _, err := db.Exec(ctx,\n        `INSERT INTO pipeline_checkpoints (slot_name, lsn, updated_at)\n         VALUES ($1, $2, now())\n         ON CONFLICT (slot_name) DO UPDATE SET lsn = $2, updated_at = now()`,\n        slotName, lsn.String(),\n    )\n    return err\n}

Error handling during flush should include exponential backoff with jitter to avoid overwhelming Elasticsearch during transient failures:

func retryFlush(ctx context.Context, fn func() error, maxRetries int) error {\n    backoff := 100 * time.Millisecond\n    for i := 0; i < maxRetries; i++ {\n        if err := fn(); err != nil {\n            if i == maxRetries-1 {\n                return err\n            }\n            jitter := time.Duration(rand.Int63n(int64(backoff)))\n            time.Sleep(backoff + jitter)\n            backoff *= 2\n            continue\n        }\n        return nil\n    }\n    return nil\n}

Data Transformation: Mapping PostgreSQL → Elasticsearch

Data structures in PostgreSQL and Elasticsearch documents often differ: normalized tables need to be denormalized, types converted, and fields filtered or enriched.

package transform\n\nimport "time"\n\n// PostgreSQL row\ntype ProductRow struct {\n    ID          int64\n    Name        string\n    Description string\n    Price       float64\n    CategoryID  int64\n    Category    string\n    Tags        []string\n    CreatedAt   time.Time\n    UpdatedAt   time.Time\n    Deleted     bool\n}\n\n// Elasticsearch document\ntype ProductDoc struct {\n    ID          string    `json:"id"`\n    Name        string    `json:"name"`\n    Description string    `json:"description"`\n    Price       float64   `json:"price"`\n    Category    string    `json:"category"`\n    Tags        []string  `json:"tags"`\n    UpdatedAt   time.Time `json:"updated_at"`\n}\n\nfunc MapProductToDoc(row ProductRow) ProductDoc {\n    return ProductDoc{\n        ID:          fmt.Sprintf("%d", row.ID),\n        Name:        row.Name,\n        Description: row.Description,\n        Price:       row.Price,\n        Category:    row.Category,\n        Tags:        row.Tags,\n        UpdatedAt:   row.UpdatedAt,\n    }\n}\n\nfunc DocToMap(doc ProductDoc) map[string]interface{} {\n    return map[string]interface{}{\n        "id":          doc.ID,\n        "name":        doc.Name,\n        "description": doc.Description,\n        "price":       doc.Price,\n        "category":    doc.Category,\n        "tags":        doc.Tags,\n        "updated_at":  doc.UpdatedAt,\n    }\n}

Note: when a DELETE event is received from the WAL, the pipeline must delete the document from Elasticsearch by _id without querying PostgreSQL for data (the row has already been deleted).

Pipeline Monitoring: Lag, Throughput, and Error Metrics

For a production pipeline, three key metrics must be tracked:

• Replication lag — the delay between a change in PostgreSQL and indexing in Elasticsearch (in seconds).
• Throughput — the number of events per second processed by the pipeline.
• Error rate — the proportion of indexing errors.

We export metrics via Prometheus:

package metrics\n\nimport "github.com/prometheus/client_golang/prometheus"\n\nvar (\n    EventsProcessed = prometheus.NewCounterVec(\n        prometheus.CounterOpts{\n            Name: "pipeline_events_total",\n            Help: "Total WAL events processed",\n        },\n        []string{"table", "operation"},\n    )\n    ReplicationLag = prometheus.NewGauge(\n        prometheus.GaugeOpts{\n            Name: "pipeline_replication_lag_seconds",\n            Help: "Replication lag in seconds",\n        },\n    )\n    IndexErrors = prometheus.NewCounter(\n        prometheus.CounterOpts{\n            Name: "pipeline_index_errors_total",\n            Help: "Total Elasticsearch indexing errors",\n        },\n    )\n    BatchSize = prometheus.NewHistogram(\n        prometheus.HistogramOpts{\n            Name:    "pipeline_batch_size",\n            Help:    "Size of Elasticsearch bulk batches",\n            Buckets: prometheus.LinearBuckets(10, 10, 10),\n        },\n    )\n)\n\nfunc init() {\n    prometheus.MustRegister(EventsProcessed, ReplicationLag, IndexErrors, BatchSize)\n}

Replication lag is calculated as the difference between the current time and the transaction commit time from the WAL event. Configure alerts in Alertmanager: if lag exceeds 30 seconds — a warning; if it exceeds 2 minutes — a critical alert.

Deploying with Docker and Kubernetes

Dockerfile

FROM golang:1.22-alpine AS builder\nWORKDIR /app\nCOPY go.mod go.sum ./\nRUN go mod download\nCOPY . .\nRUN CGO_ENABLED=0 GOOS=linux go build -o pipeline ./cmd/pipeline\n\nFROM alpine:3.19\nRUN apk add --no-cache ca-certificates tzdata\nWORKDIR /app\nCOPY --from=builder /app/pipeline .\nENTRYPOINT ["./pipeline"]

Kubernetes Deployment

The pipeline runs as a single Pod (not horizontally scaled — one replication slot per one reader):

apiVersion: apps/v1\nkind: Deployment\nmetadata:\n  name: data-pipeline\n  namespace: production\nspec:\n  replicas: 1\n  selector:\n    matchLabels:\n      app: data-pipeline\n  template:\n    metadata:\n      labels:\n        app: data-pipeline\n      annotations:\n        prometheus.io/scrape: "true"\n        prometheus.io/port: "9090"\n    spec:\n      containers:\n      - name: pipeline\n        image: your-registry/data-pipeline:latest\n        env:\n        - name: PG_DSN\n          valueFrom:\n            secretKeyRef:\n              name: pipeline-secrets\n              key: pg-dsn\n        - name: ES_ADDR\n          valueFrom:\n            secretKeyRef:\n              name: pipeline-secrets\n              key: es-addr\n        - name: SLOT_NAME\n          value: "pipeline_slot"\n        - name: PUBLICATION\n          value: "products_pub"\n        - name: BATCH_SIZE\n          value: "500"\n        - name: FLUSH_INTERVAL\n          value: "1s"\n        resources:\n          requests:\n            cpu: 100m\n            memory: 128Mi\n          limits:\n            cpu: 500m\n            memory: 256Mi\n        livenessProbe:\n          httpGet:\n            path: /health\n            port: 9090\n          initialDelaySeconds: 10\n          periodSeconds: 30

All configuration parameters are passed via environment variables — no hardcoded values in the image. Secrets are stored in Kubernetes Secrets and mounted via secretKeyRef.

Testing the Pipeline

Unit Tests for Transformation

Mapping functions are tested without any external dependencies:

package transform_test\n\nimport (\n    "testing"\n    "time"\n\n    "github.com/stretchr/testify/assert"\n    "your/project/internal/transform"\n)\n\nfunc TestMapProductToDoc(t *testing.T) {\n    row := transform.ProductRow{\n        ID:          42,\n        Name:        "Test Product",\n        Description: "Description",\n        Price:       99.99,\n        Category:    "Electronics",\n        Tags:        []string{"new", "sale"},\n        UpdatedAt:   time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC),\n    }\n    doc := transform.MapProductToDoc(row)\n    assert.Equal(t, "42", doc.ID)\n    assert.Equal(t, "Test Product", doc.Name)\n    assert.Equal(t, 99.99, doc.Price)\n    assert.Equal(t, []string{"new", "sale"}, doc.Tags)\n}

Integration Tests

For integration tests, use testcontainers-go: spin up real PostgreSQL and Elasticsearch instances in Docker containers directly within the tests:

func TestPipelineIntegration(t *testing.T) {\n    ctx := context.Background()\n\n    pgContainer, err := testcontainers.GenericContainer(ctx,\n        testcontainers.GenericContainerRequest{\n            ContainerRequest: testcontainers.ContainerRequest{\n                Image:        "postgres:16",\n                ExposedPorts: []string{"5432/tcp"},\n                Env: map[string]string{\n                    "POSTGRES_PASSWORD": "test",\n                    "POSTGRES_DB":       "testdb",\n                },\n                Cmd: []string{"postgres", "-c", "wal_level=logical"},\n                WaitingFor: wait.ForListeningPort("5432/tcp"),\n            },\n            Started: true,\n        },\n    )\n    require.NoError(t, err)\n    defer pgContainer.Terminate(ctx)\n\n    // ... similarly for Elasticsearch\n    // Start the pipeline, perform an INSERT into PostgreSQL\n    // Verify the document exists in Elasticsearch using polling with a timeout\n}

The test should: create the table and publication, start the pipeline as a goroutine, execute INSERT/UPDATE/DELETE operations in PostgreSQL, and after a few seconds verify the state of the Elasticsearch index. This covers the entire critical path.

Conclusion

Building a reliable data pipeline with Go, PostgreSQL, and Elasticsearch is achievable through the right combination of proven tools: PostgreSQL logical replication guarantees data completeness, Go provides efficiency and code simplicity, and the Elasticsearch Bulk API makes indexing scalable.

Key principles that make a pipeline production-ready:

• CDC via WAL instead of polling — minimal latency and support for all operations.
• Saving the LSN after a successful flush — correct at-least-once semantics.
• Idempotent indexing via an explicit _id — safe handling of duplicate events.
• Batch sending with flush by size and timeout — a balance between latency and throughput.
• Exporting metrics to Prometheus — real-time visibility into pipeline health.
• Configuration via environment variables — portability across environments.

The next step for scaling: if the volume of events exceeds the capacity of a single replication slot, consider table partitioning with multiple publications and separate pipeline instances, or introduce Kafka as an intermediate broker with a Debezium connector.