cdcelasticsearch

Go Mongo CDC Elasticsearch

MongoDB Change Data Capture (CDC) connector for Elasticsearch. This library provides real-time data synchronization from MongoDB to Elasticsearch using MongoDB Change Streams.

Features

• Near real-time synchronization from MongoDB to Elasticsearch using MongoDB Change Streams.
• Custom routing and mapping via pluggable mapper functions that can emit one or many Elasticsearch actions per change event.
• Batch processing controls such as maximum batch size, batch bytes and batch ticker durations for efficient bulk indexing.
• Request body compression support for Elasticsearch bulk requests.
• Easily manageable configurations via YAML files or Go structs.

Installation

go get github.com/Trendyol/go-mongo-cdc-elasticsearch

Quick Start

Basic Usage

package main

import (
    "context"
    cdcelasticsearch "github.com/Trendyol/go-mongo-cdc-elasticsearch"
)

func main() {
    connector, err := cdcelasticsearch.NewConnectorBuilder("config.yml").Build()
    if err != nil {
        panic(err)
    }
    defer connector.Close()
    
    connector.Start(context.Background())
}

Configuration

mongodb:
  connection:
    uri: "localhost:27017"
    database: "mydb"
    collection: "mycollection"
  connectionPool:
    maxPoolSize: 100
    minPoolSize: 5

elasticsearch:
  urls:
    - "http://localhost:9200"
  collectionIndexMapping:
    mycollection: "my-index"
  batchSizeLimit: 1000
  batchTickerDuration: 5s

metric:
  port: 8080

partition:
  totalPartition: 1
  heartbeatInterval: 5s
  consumerGroup: myConsumerGroup

logger:
  logLevel: "info"

Custom Mapper

func customMapper(event mongodb.Event) []document.ESActionDocument {
    if event.IsMutated {
        // Transform your data
        transformedData := transformData(event.Value)
        doc := document.NewIndexAction(event.Key, transformedData, nil)
        return []document.ESActionDocument{doc}
    }
    doc := document.NewDeleteAction(event.Key, nil)
    return []document.ESActionDocument{doc}
}

connector, err := cdcelasticsearch.NewConnectorBuilder("config.yml").
    SetMapper(customMapper).
    Build()

Configuration Options

MongoDB Configuration

Check out on go-mongo-cdc

Elasticsearch Configuration

See example configurations for more details.

Examples

• Simple Example - Basic usage with file-based configuration
• Default Mapper - Using the built-in default mapper
• Struct Config - Programmatic configuration

Exposed metrics

CDC-related metrics are also exposed by the underlying change data capture layer and are available on the same Prometheus endpoint.

Testing

Run All Tests

# Run integration tests
make test-integration

# Run unit tests
make test-unit

# Test all CI checks locally (lint + build + tests)
make test-ci-local

See TESTING.md for detailed testing guide.

Contributing

Contributions are welcome! Please read CONTRIBUTING.md for details.

License

Released under the MIT License.