Configuration

ElasticRelay uses a JSON configuration file to define data sources, Elasticsearch connection, and various options.

Basic Configuration Structure

{
  "version": "3.0",
  "data_sources": [...],
  "elasticsearch": {...},
  "checkpoint": {...},
  "dlq": {...},
  "performance": {...}
}

Configuration Sections

Version

The configuration version. Always use "3.0" for the latest version.

{
  "version": "3.0"
}

Data Sources

Define one or more data sources to sync from.

MySQL Data Source

{
  "data_sources": [
    {
      "id": "mysql-main",
      "type": "mysql",
      "host": "localhost",
      "port": 3306,
      "user": "elasticrelay",
      "password": "your_password",
      "database": "mydb",
      "tables": ["users", "orders", "products"],
      "exclude_tables": ["temp_*", "cache_*"],
      "server_id": 1001
    }
  ]
}

Options:

id (required): Unique identifier for this data source
type (required): Must be "mysql"
host (required): MySQL host
port (optional): MySQL port (default: 3306)
user (required): Database user
password (required): Database password
database (required): Database name
tables (optional): List of tables to sync (default: all tables)
exclude_tables (optional): List of tables to exclude (supports wildcards)
server_id (optional): MySQL server ID for replication (default: auto-generated)

PostgreSQL Data Source

{
  "data_sources": [
    {
      "id": "postgres-main",
      "type": "postgresql",
      "host": "localhost",
      "port": 5432,
      "user": "elasticrelay",
      "password": "your_password",
      "database": "mydb",
      "schema": "public",
      "tables": ["users", "orders"],
      "publication_name": "elasticrelay_pub",
      "slot_name": "elasticrelay_slot"
    }
  ]
}

Options:

id (required): Unique identifier for this data source
type (required): Must be "postgresql"
host (required): PostgreSQL host
port (optional): PostgreSQL port (default: 5432)
user (required): Database user
password (required): Database password
database (required): Database name
schema (optional): Schema name (default: “public”)
tables (optional): List of tables to sync (default: all tables)
publication_name (optional): Publication name (default: “elasticrelay_pub”)
slot_name (optional): Replication slot name (default: “elasticrelay_slot”)

MongoDB Data Source

{
  "data_sources": [
    {
      "id": "mongodb-main",
      "type": "mongodb",
      "connection_string": "mongodb://localhost:27017/?replicaSet=rs0",
      "database": "mydb",
      "collections": ["users", "orders"],
      "exclude_collections": ["temp_*"]
    }
  ]
}

Options:

id (required): Unique identifier for this data source
type (required): Must be "mongodb"
connection_string (required): MongoDB connection string
database (required): Database name
collections (optional): List of collections to sync (default: all collections)
exclude_collections (optional): List of collections to exclude (supports wildcards)

Elasticsearch Configuration

{
  "elasticsearch": {
    "addresses": ["http://localhost:9200"],
    "username": "elastic",
    "password": "your_password",
    "cloud_id": "",
    "api_key": "",
    "index_prefix": "",
    "bulk_size": 1000,
    "bulk_flush_interval": "5s",
    "number_of_shards": 1,
    "number_of_replicas": 1
  }
}

Options:

addresses (required): List of Elasticsearch nodes
username (optional): Basic auth username
password (optional): Basic auth password
cloud_id (optional): Elastic Cloud ID
api_key (optional): API key for authentication
index_prefix (optional): Prefix for index names
bulk_size (optional): Number of documents per bulk request (default: 1000)
bulk_flush_interval (optional): Max time between bulk requests (default: “5s”)
number_of_shards (optional): Number of shards for new indices (default: 1)
number_of_replicas (optional): Number of replicas for new indices (default: 1)

Checkpoint Configuration

{
  "checkpoint": {
    "storage": "file",
    "path": "./checkpoints",
    "interval": "10s"
  }
}

Options:

storage (optional): Storage type - “file” or “elasticsearch” (default: “file”)
path (optional): Path for file-based checkpoints (default: “./checkpoints”)
interval (optional): Checkpoint save interval (default: “10s”)

Dead Letter Queue (DLQ) Configuration

{
  "dlq": {
    "enabled": true,
    "storage": "file",
    "path": "./dlq",
    "max_retries": 3,
    "retry_interval": "1m"
  }
}

Options:

enabled (optional): Enable DLQ (default: true)
storage (optional): Storage type - “file” or “elasticsearch” (default: “file”)
path (optional): Path for file-based DLQ (default: “./dlq”)
max_retries (optional): Max retry attempts (default: 3)
retry_interval (optional): Time between retries (default: “1m”)

Performance Configuration

{
  "performance": {
    "snapshot_threads": 4,
    "snapshot_batch_size": 10000,
    "cdc_buffer_size": 10000,
    "memory_limit": "1GB"
  }
}

Options:

snapshot_threads (optional): Number of parallel threads for snapshot (default: 4)
snapshot_batch_size (optional): Batch size for snapshot reads (default: 10000)
cdc_buffer_size (optional): Buffer size for CDC events (default: 10000)
memory_limit (optional): Memory limit (default: “1GB”)

Complete Example

Here’s a complete configuration example with multiple data sources:

{
  "version": "3.0",
  "data_sources": [
    {
      "id": "mysql-users",
      "type": "mysql",
      "host": "mysql.example.com",
      "port": 3306,
      "user": "elasticrelay",
      "password": "mysql_password",
      "database": "users_db",
      "tables": ["users", "profiles"]
    },
    {
      "id": "postgres-orders",
      "type": "postgresql",
      "host": "postgres.example.com",
      "port": 5432,
      "user": "elasticrelay",
      "password": "postgres_password",
      "database": "orders_db",
      "schema": "public",
      "tables": ["orders", "order_items"]
    },
    {
      "id": "mongodb-products",
      "type": "mongodb",
      "connection_string": "mongodb://mongo.example.com:27017/?replicaSet=rs0",
      "database": "products_db",
      "collections": ["products", "categories"]
    }
  ],
  "elasticsearch": {
    "addresses": ["http://elasticsearch:9200"],
    "username": "elastic",
    "password": "es_password",
    "bulk_size": 1000,
    "bulk_flush_interval": "5s"
  },
  "checkpoint": {
    "storage": "elasticsearch",
    "interval": "10s"
  },
  "dlq": {
    "enabled": true,
    "storage": "elasticsearch",
    "max_retries": 3,
    "retry_interval": "1m"
  },
  "performance": {
    "snapshot_threads": 8,
    "snapshot_batch_size": 10000,
    "cdc_buffer_size": 10000,
    "memory_limit": "2GB"
  }
}

Environment Variables

You can use environment variables in your configuration:

{
  "data_sources": [
    {
      "id": "mysql-main",
      "type": "mysql",
      "host": "${MYSQL_HOST}",
      "user": "${MYSQL_USER}",
      "password": "${MYSQL_PASSWORD}"
    }
  ]
}

Transform Configuration

ElasticRelay includes a built-in Transform Engine for data transformation between your databases and Elasticsearch. Transform rules are defined in a separate JSON configuration file.

Command-Line Parameter

./elasticrelay \
  -config ./config/config.json \
  -port 50051 \
  -transform-config ./config/transform.json

Parameter	Description	Default
`-config`	Data source configuration file	`config.json`
`-port`	gRPC service port	`50051`
`-transform-config`	Transform configuration file (optional)	empty (pass-through)

When -transform-config is not specified, events pass through unchanged (pass-through mode).

Transform Configuration Structure

{
  "transform_rules": [...],
  "global_settings": {...},
  "masking_templates": {...}
}

Transform Rules

Each rule matches tables by pattern and applies transformations:

{
  "transform_rules": [
    {
      "id": "rule-1",
      "name": "user-data-transform",
      "source_id": "mysql-main",
      "table_pattern": "users",
      "priority": 1,
      "field_mappings": [...],
      "field_configs": [...],
      "masking_rules": [...],
      "filters": [...],
      "computed_fields": [...]
    }
  ]
}

Options:

id (required): Unique rule identifier
name (required): Human-readable rule name
source_id (optional): Match specific data source ID (empty matches all)
table_pattern (required): Table name pattern (supports wildcards like user_*)
priority (optional): Rule priority (lower = higher priority, default: 0)

Field Mapping

Rename, copy, or move fields:

{
  "field_mappings": [
    {"source": "user_name", "target": "username", "operation": "rename"},
    {"source": "created_at", "target": "create_time", "operation": "copy"},
    {"source": "old_field", "target": "new_field", "operation": "move"}
  ]
}

Nested paths are supported with dot notation:

{
  "field_mappings": [
    {"source": "user.profile.name", "target": "display_name", "operation": "copy"}
  ]
}

Type Conversion

Convert field types:

{
  "field_configs": [
    {"field": "is_vip", "target_type": "bool"},
    {"field": "age", "target_type": "int"},
    {"field": "price", "target_type": "float64"},
    {"field": "created_at", "target_type": "date"},
    {"field": "phone", "target_type": "keyword"},
    {"field": "metadata", "target_type": "object"}
  ]
}

Supported types: string, keyword, text, int, int64, float, float64, bool, date, timestamp, object

Data Masking

Anonymize sensitive fields:

{
  "masking_rules": [
    {"field": "phone", "strategy": "mask", "preset": "phone"},
    {"field": "id_card", "strategy": "mask", "preset": "id_card"},
    {"field": "email", "strategy": "mask", "preset": "email"},
    {"field": "bank_card", "strategy": "mask", "preset": "bank_card"},
    {"field": "name", "strategy": "mask", "preset": "name"},
    {"field": "password", "strategy": "hash"}
  ]
}

Masking strategies:

Strategy	Description	Example
`mask`	Character masking with presets	`138****5678`
`hash`	SHA256 hash	`a1b2c3d4...`
`token`	Tokenization	`TOKEN_abc123`
`regex`	Custom regex replacement	Custom pattern

Preset templates:

Template	Input	Output
`phone`	`13812345678`	`138****5678`
`id_card`	`110101199001011234`	`1101**********1234`
`email`	`john@example.com`	`jo***@example.com`
`bank_card`	`6222021234567890`	`6222********7890`
`name`	`John`	`J***`

Conditional Filtering

Filter records based on conditions:

{
  "filters": [
    {"field": "status", "operator": "ne", "value": "deleted", "action": "exclude"},
    {"field": "is_test", "operator": "eq", "value": 1, "action": "exclude"},
    {"field": "age", "operator": "gte", "value": 18, "action": "include"},
    {"field": "type", "operator": "in", "value": ["premium", "vip"], "action": "include"},
    {"field": "email", "operator": "regex", "value": ".*@example\\.com", "action": "include"}
  ]
}

Operators: eq, ne, gt, gte, lt, lte, in, nin, regex, exists

Actions: include (keep if matched), exclude (remove if matched), route (route to specific target)

Expression Engine

Compute new fields using expressions:

{
  "computed_fields": [
    {
      "name": "full_name",
      "expression": "concat($.last_name, $.first_name)",
      "type": "string"
    },
    {
      "name": "age_group",
      "expression": "$.age < 18 ? 'minor' : 'adult'",
      "type": "string"
    },
    {
      "name": "total_price",
      "expression": "$.price * $.quantity",
      "type": "float64"
    },
    {
      "name": "processed_at",
      "expression": "now()",
      "type": "date"
    }
  ]
}

Built-in functions:

Category	Functions
String	`concat()`, `substr()`, `upper()`, `lower()`, `trim()`, `replace()`, `length()`
Math	`round()`, `abs()`, `floor()`, `ceil()`, `min()`, `max()`
Date	`now()`, `formatDate()`, `parseDate()`
Conditional	`ifNull()`, `ifEmpty()`, `coalesce()`

Complete Transform Example

{
  "transform_rules": [
    {
      "id": "user-transform",
      "name": "user-data-transform",
      "source_id": "mysql-main",
      "table_pattern": "users",
      "priority": 1,
      "field_mappings": [
        {"source": "user_name", "target": "username", "operation": "rename"},
        {"source": "created_at", "target": "create_time", "operation": "copy"}
      ],
      "field_configs": [
        {"field": "is_vip", "target_type": "bool"},
        {"field": "phone", "target_type": "keyword"},
        {"field": "internal_notes", "exclude": true},
        {"field": "debug_info", "exclude": true}
      ],
      "masking_rules": [
        {"field": "phone", "strategy": "mask", "preset": "phone"},
        {"field": "email", "strategy": "mask", "preset": "email"},
        {"field": "id_card", "strategy": "mask", "preset": "id_card"},
        {"field": "password", "strategy": "hash"}
      ],
      "filters": [
        {"field": "status", "operator": "ne", "value": "deleted", "action": "exclude"},
        {"field": "is_test", "operator": "eq", "value": 1, "action": "exclude"}
      ],
      "computed_fields": [
        {
          "name": "full_name",
          "expression": "concat($.last_name, $.first_name)",
          "type": "string"
        },
        {
          "name": "age_group",
          "expression": "$.age < 18 ? 'minor' : 'adult'",
          "type": "string"
        }
      ]
    }
  ]
}

Configuration Validation

ElasticRelay validates your configuration on startup. If there are errors, you’ll see detailed messages:

elasticrelay --config config.json

What’s Next?

MySQL CDC - MySQL-specific configuration
PostgreSQL CDC - PostgreSQL-specific configuration
MongoDB CDC - MongoDB-specific configuration
Configuration Reference - Complete configuration reference