Q&A Sink Elasticsearch id_key not reading my value

[NyankoSensei-Edbert]

Question

My main issue is that Elasticsearch is using its default auto-generated _id instead of the hash I defined. I’ve attached a screenshot of the data input. While the id field contains the correct value, the metadata _id is incorrect. How can I resolve this? Thank you.

I used this base image timberio/vector:0.51.1-debian

### Vector Config

data_dir = "/var/lib/vector"

# Enable Vector's built-in API for monitoring
[api]
enabled = true
address = "0.0.0.0:8686"

# Stream decompressed WARC files
[sources.warc_stream]
type = "exec"
mode = "streaming"
command = ["sh", "-c", "find /data -name '*.gz' -type f -exec zcat {} \\;"]

# Extract full domain names (including subdomains) from WARC-Target-URI and Host headers
[transforms.extract_domains]
type = "remap"
inputs = ["warc_stream"]
source = '''
    .message = to_string!(.message)

    # Extract domain from WARC-Target-URI:
    if match(.message, r'WARC-Target-URI:\s*') {
        uri_match = parse_regex!(.message, r'WARC-Target-URI:\s*(?P<uri>\S+)')
        if exists(uri_match.uri) {
            parsed, err = parse_url(uri_match.uri)
            if err == null && exists(parsed.host) {
                # Get full hostname including subdomain (e.g., www.example.com, api.sub.example.com)
                host = downcase!(parsed.host)
                # Remove port if present
                host_parts = split(host, ":")
                .domain = host_parts[0]
                # .source = "warc_target_uri"
            } else {
                abort
            }
        } else {
            abort
        }
    
    # Extract domain from Host: header
    } else if match(.message, r'Host:\s*') {
        host_match = parse_regex!(.message, r'Host:\s*(?P<host>[^\s\r\n]+)')
        if exists(host_match.host) {
            host = strip_whitespace(host_match.host)
            # Remove port if present
            host_parts = split(host, ":")
            .domain = downcase!(host_parts[0])
            # .source = "host_header"
        } else {
            abort
        }

    } else {
        abort
    }

    # .timestamp = now()
'''

# Filter valid domains only
[transforms.filter_valid]
type = "filter"
inputs = ["extract_domains"]
condition = '''
    exists(.domain) &&
    length(to_string!(.domain)) > 0 &&
    match(to_string!(.domain), r'^[a-z0-9.-]+\.[a-z]{2,}$')
'''

# Deduplicate domains - keep only unique ones
[transforms.deduplicate]
type = "dedupe"
inputs = ["filter_valid"]
fields.match = ["domain"]

# Format for Elasticsearch with unique domain as document ID
[transforms.format_for_es]
type = "remap"
inputs = ["deduplicate"]
source = '''
    # Use domain as the document ID to ensure uniqueness in Elasticsearch
    .id = sha1(to_string!(.domain))
    .@id = sha1(to_string!(.domain))
    
    . = {
        "domain": .domain,
        "id": .id
        # "source": .source,
        # "first_seen": .timestamp,
        # "last_updated": .timestamp
    }
'''

# Send to Elasticsearch with domain as document ID (ensures uniqueness)
[sinks.elasticsearch]
type = "elasticsearch"
inputs = ["format_for_es"]
endpoints = ["http://elasticsearch:9200"]
mode = "bulk"
bulk.index = "dns-records"
bulk.action = "create"
id_key = "@id"
compression = "gzip"
api_version = "v8" 

[sinks.elasticsearch.buffer]
type = "memory"
max_events = 50000
when_full = "block"

[sinks.elasticsearch.batch]
max_bytes = 52428800   # 50 MB
max_events = 5000
timeout_secs = 3

[sinks.elasticsearch.healthcheck]
enabled = true

[sinks.elasticsearch.request]
retry_attempts = 5
retry_initial_backoff_secs = 1
retry_max_duration_secs = 30

# Console output for monitoring progress
[sinks.console]
type = "console"
inputs = ["format_for_es"]
encoding.codec = "json"

# Internal metrics for monitoring
[sinks.prometheus_exporter]
type = "prometheus_exporter"
inputs = ["format_for_es"]
address = "0.0.0.0:9598"

Vector Logs

[pront]

Hey @NyankoSensei-Edbert, did you figure this out on your own?

[NyankoSensei-Edbert]

yes. turns out there was a typo in my code