Vector with Kafka

[quwassar]

Something strange with Vector:
I have 3 VM with Vector container and 1-st can't push logs to Kafka.
My config:

data_dir = "/var/lib/vector"

[api]
    enabled = true # optional, default
    address = "0.0.0.0:8686" # optional, default
    playground = true # optional, default

[sources.vector_metrics]
    type = "internal_metrics"
    scrape_interval_secs = 15

[sinks.my_prometheus_exporter_sink]
    type = "prometheus_exporter"
    inputs = [ "vector_metrics" ]
    flush_period_secs = 30
    address = "0.0.0.0:8086"

[sources.internal_logs_source]
    type = "internal_logs"

[transforms.internal_logs_remap]
    type = "remap"
    inputs = ["internal_logs_source"]
    source = '''
    .dc = "my_dc"
    .env = "my_env"
    .group = "my_group"
    .inst = "my_instance"
    if exists(.system) && .system != null {
      .system = downcase!(.system)
    }
    if !exists(.@timestamp) {
      if exists(.timestamp) && .timestamp != null {
        .@timestamp = .timestamp
      } else {
        .@timestamp = now()
      }
    }
    del(.timestamp)

    .system = "vector"
    del(.level)
    '''

[sources.netdata_docker_source]
    type = "docker_logs"
    include_containers = ["netdata"]

[transforms.netdata_logs_remap]
    type = "remap"
    inputs = ["netdata_docker_source"]
    source = '''
    .dc = "my_dc"
    .env = "my_env"
    .group = "my_group"
    .inst = "my_instance"
    if exists(.system) && .system != null {
      .system = downcase!(.system)
    }
    if !exists(.@timestamp) {
      if exists(.timestamp) && .timestamp != null {
        .@timestamp = .timestamp
      } else {
        .@timestamp = now()
      }
    }
    del(.timestamp)

    .system = "netdata"
    '''

[sources.dockerd_logs]
    type = "journald" # required
    current_boot_only = false # optional, default
    journalctl_path = "/etc/vector/journalctl"
    exclude_units = [] # optional, default
    include_units = ["docker.service"] # optional, default

[transforms.dockerd_logs_add_fields]
    type = "remap"
    inputs = ["dockerd_logs"]
    source = '''
    .dc = "ds"
    .env = "qa"
    .group = "my_group"
    .inst = "my_instance"
    if exists(.system) && .system != null {
      .system = downcase!(.system)
    }
    if !exists(.@timestamp) {
      if exists(.timestamp) && .timestamp != null {
        .@timestamp = .timestamp
      } else {
        .@timestamp = now()
      }
    }
    del(.timestamp)

    .system = "dockerd"
    '''

[sources.sshd_logs]
    type = "journald" # required
    current_boot_only = false # optional, default
    journalctl_path = "/etc/vector/journalctl"
    exclude_units = [] # optional, default
    include_units = ["ssh.service"] # optional, default

[transforms.sshd_logs_add_fields]
    type = "remap"
    inputs = ["sshd_logs"]
    source = '''
    .dc = "my_dc"
    .env = "my_env"
    .group = "my_group"
    .inst = "my_instance"
    if exists(.system) && .system != null {
      .system = downcase!(.system)
    }
    if !exists(.@timestamp) {
      if exists(.timestamp) && .timestamp != null {
        .@timestamp = .timestamp
      } else {
        .@timestamp = now()
      }
    }
    del(.timestamp)

    .system = "sshd"
    '''

[sources.keydb_docker_source]
  type = "docker_logs"
  include_containers = ["keydb"]

[sources.sentinel_docker_source]
  type = "docker_logs"
  include_containers = ["sentinel"]

[transforms.keydb_logs_remap]
  type = "remap"
  inputs = ["keydb_docker_source"]
  source = '''
    .dc = "my_dc"
    .env = "my_env"
    .group = "my_group"
    .inst = "my_instance"
    if exists(.system) && .system != null {
      .system = downcase!(.system)
    }
    if !exists(.@timestamp) {
      if exists(.timestamp) && .timestamp != null {
        .@timestamp = .timestamp
      } else {
        .@timestamp = now()
      }
    }
    del(.timestamp)

  .system = "my_system"
  .cluster = "my_cluster"
  '''

[transforms.sentinel_logs_remap]
  type = "remap"
  inputs = ["sentinel_docker_source"]
  source = '''
    .dc = "my_dc"
    .env = "my_env"
    .group = "my_group"
    .inst = "my_instance"
    if exists(.system) && .system != null {
      .system = downcase!(.system)
    }
    if !exists(.@timestamp) {
      if exists(.timestamp) && .timestamp != null {
        .@timestamp = .timestamp
      } else {
        .@timestamp = now()
      }
    }
    del(.timestamp)

  .system = "my_system"
  .cluster = "My_cluster"
  '''


[sinks.log_sink_kafka]
    type = "kafka"
                                            inputs = ["internal_logs_remap","netdata_logs_remap","sshd_logs_add_fields","dockerd_logs_add_fields","sentinel_logs_remap","keydb_logs_remap"]
        bootstrap_servers = "Kafka-nodes"
    key_field = "my_key"
    topic = "my_topik"
    message_timeout_ms = 300000 
    socket_timeout_ms = 60000 
    compression = "gzip"

    buffer.max_size = 1049000000
    buffer.type = "disk"

    librdkafka_options."client.id" = "client_id"
    librdkafka_options."message.max.bytes" = "1000000"
    librdkafka_options."max.in.flight.requests.per.connection" = "5"
    librdkafka_options."metadata.request.timeout.ms" = "300000"
    librdkafka_options."socket.timeout.ms" = "60000"
    librdkafka_options."socket.keepalive.enable" = "true"
    librdkafka_options."socket.max.fails" = "1"
    librdkafka_options."api.version.request" = "false"
    librdkafka_options."transaction.timeout.ms" = "2147483647"
    librdkafka_options."enable.idempotence" = "false"
    librdkafka_options."queue.buffering.max.messages" = "100000"
    librdkafka_options."queue.buffering.max.kbytes" = "100100000000000000"
    librdkafka_options."queue.buffering.max.ms" = "50"
    librdkafka_options."message.send.max.retries" = "10000"
    librdkafka_options."compression.codec" = "gzip"
    librdkafka_options."compression.type" = "gzip"
    librdkafka_options."compression.level" = "4"
    librdkafka_options."batch.num.messages" = "100"
    librdkafka_options."batch.size" = "2147483647"
    librdkafka_options."request.required.acks" = "-1"
    librdkafka_options."request.timeout.ms" = "900000"
    librdkafka_options."message.timeout.ms" = "300000"
    librdkafka_options."log_level" = "0"

    encoding.codec = "json"
    encoding.except_fields = ["label","container_created_at", "container_id", "container_name", "image", "source_type", "stream","timestamp","LVL"]

And my error:

{"timestamp":"2024-01-11T11:23:31.149786Z","level":"ERROR","message":"Configuration error.","error":"Sink \"log_sink_kafka\": error occurred when building buffer: failed to build individual stage 0: failed to load/create ledger: failed to lock buffer.lock; is another Vector process running and using this buffer?","target":"vector::topology::builder"}

Before I use version image 0.32.0-debian and now upgrade to 0.35.0-debian

[jszwedko]

Hi @quwassar ,

Are you possibly sharing the data directory between those three Vector containers? The error indicates that one of the data directories was locked when Vector attempts to load it.

[quwassar]

Hm, directory on host for only one container. Maybe problem in stop and delete early version before run new? Nothing was not change, only image and container

[quwassar]

Oh, sorry! That's it - I forgot reboot host. After rebooting problem lost. Perhaps something wrong and only this machine I get problems