Skip to content
LogchefLogchef

Shipping Logs with Vector

Vector is a lightweight, high-performance observability pipeline that can collect, transform, and route logs to ClickHouse or VictoriaLogs. Logchef queries either backend in place; Vector is the component that gets logs there.

Quick Start

Section titled “Quick Start”

If you’re using the Docker quick start, Vector is already included and configured: demo syslog data flows into ClickHouse automatically. Skip to Adapting for Real Sources when you’re ready to ship your own logs, or add the VictoriaLogs sink below.

For a standalone setup, you need three things:

  1. A ClickHouse table with the right schema, or a reachable VictoriaLogs server
  2. A Vector config that transforms and ships logs
  3. Logchef pointed at the ClickHouse table or VictoriaLogs API

Prerequisites

Section titled “Prerequisites”
  • ClickHouse or VictoriaLogs running and accessible
  • Vector installed (v0.30+)
  • A Logchef instance (for querying)

Create the ClickHouse Table

Section titled “Create the ClickHouse Table”

Logchef works with any ClickHouse table that has a timestamp column, but the built-in OpenTelemetry schema gives you the best experience: severity filtering, service grouping, and flexible log_attributes all work out of the box.

Create the table:

CREATE TABLE IF NOT EXISTS default.logs
(
    timestamp DateTime64(3) CODEC(DoubleDelta, LZ4),
    severity_text LowCardinality(String) CODEC(ZSTD(1)),
    severity_number Int32 CODEC(ZSTD(1)),
    service_name LowCardinality(String) CODEC(ZSTD(1)),
    namespace LowCardinality(String) CODEC(ZSTD(1)),
    body String CODEC(ZSTD(1)),
    log_attributes Map(LowCardinality(String), String) CODEC(ZSTD(1)),


    INDEX idx_severity_text severity_text TYPE set(100) GRANULARITY 4,
    INDEX idx_log_attributes_keys mapKeys(log_attributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_log_attributes_values mapValues(log_attributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_body body TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 1
)
ENGINE = MergeTree()
PARTITION BY toDate(timestamp)
ORDER BY (namespace, service_name, timestamp)
TTL toDateTime(timestamp) + INTERVAL 30 DAY
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;

See Schema Design for details on field types, codecs, and indexing choices.

Configure Vector

Section titled “Configure Vector”

A minimal Vector config that collects syslog and ships to ClickHouse:

[api]
enabled = true


# Source: collect syslog
[sources.syslog_input]
type = "syslog"
address = "0.0.0.0:514"
mode = "udp"


# Transform: map to OTEL schema
[transforms.remap_logs]
inputs = ["syslog_input"]
type = "remap"
source = '''
  structured = parse_syslog!(.message)


  .timestamp = format_timestamp!(structured.timestamp, format: "%Y-%m-%d %H:%M:%S.%f")
  .body = structured.message
  .service_name = structured.appname
  .namespace = "syslog"


  # Map severity to standardized values
  .severity_text = if includes(["emerg", "err", "crit", "alert"], structured.severity) {
    "ERROR"
  } else if structured.severity == "warning" {
    "WARN"
  } else if structured.severity == "debug" {
    "DEBUG"
  } else if includes(["info", "notice"], structured.severity) {
    "INFO"
  } else {
    structured.severity
  }


  # OTEL severity numbers
  # https://opentelemetry.io/docs/specs/otel/logs/data-model/#severity-fields
  .severity_number = if .severity_text == "ERROR" {
    17
  } else if .severity_text == "WARN" {
    13
  } else if .severity_text == "DEBUG" {
    5
  } else {
    9
  }


  # Store source-specific fields in log_attributes
  .log_attributes = {
    "syslog.procid": structured.procid,
    "syslog.facility": structured.facility,
    "syslog.version": structured.version,
    "syslog.hostname": structured.hostname
  }


  del(.message)
  del(.source_type)
'''


# Sink: ship to ClickHouse
[sinks.clickhouse]
type = "clickhouse"
inputs = ["remap_logs"]
endpoint = "http://localhost:8123"
database = "default"
table = "logs"
compression = "gzip"
healthcheck.enabled = false
skip_unknown_fields = true

Run Vector:

Terminal window
vector --config vector.toml

Send the Same Events to VictoriaLogs

Section titled “Send the Same Events to VictoriaLogs”

VictoriaLogs does not need a table schema. Point a second sink at the same remap_logs transform to dual-write the normalized event, or replace the ClickHouse sink if VictoriaLogs is your only store:

[sinks.victorialogs]
type = "elasticsearch"
inputs = ["remap_logs"]
endpoints = ["http://localhost:9428/insert/elasticsearch/"]
api_version = "v8"
mode = "bulk"
compression = "gzip"


[sinks.victorialogs.healthcheck]
enabled = false


[sinks.victorialogs.request.headers]
VL-Stream-Fields = "namespace,service_name"
VL-Time-Field = "timestamp"
VL-Msg-Field = "body"

The three VictoriaLogs headers tell the ingestion API how to map your event:

  • VL-Time-Field becomes VictoriaLogs’ _time
  • VL-Msg-Field becomes _msg
  • VL-Stream-Fields defines stable stream identity

Keep stream fields low-cardinality. Service, environment, namespace, and host are good candidates; request IDs and trace IDs are not. All remaining event fields stay available for LogsQL and LogchefQL filters.

When Vector runs in a container, replace localhost with the VictoriaLogs service name on the shared Docker network. If your VictoriaLogs deployment is multi-tenant, add AccountID and ProjectID headers to this ingestion sink as well as to the Logchef source; writing to one tenant and querying another will look like successful ingestion followed by empty results.

Verify the pipeline at the storage API before configuring Logchef:

Terminal window
curl --fail --data-urlencode 'query=namespace:="syslog"' \
  http://localhost:9428/select/logsql/query

Then add a VictoriaLogs source in Logchef with timestamp field _time and severity field severity_text. See Using VictoriaLogs with Logchef for source authentication, tenancy, immutable scope, and query modes.

Adapting for Real Sources

Section titled “Adapting for Real Sources”

Replace the [sources] and [transforms] sections for your log source. The sink stays the same.

File-based Logs

Section titled “File-based Logs”
[sources.file_logs]
type = "file"
include = ["/var/log/**/*.log"]
read_from = "beginning"


[transforms.remap_file_logs]
inputs = ["file_logs"]
type = "remap"
source = '''
  .timestamp = now()
  .service_name = "file_service"
  .namespace = "files"
  .body = .message


  # Extract severity if present in the log line
  severity_match = parse_regex(.message, r'^(?P<time>\S+)?\s+(?P<level>INFO|DEBUG|WARN|ERROR)') ?? {}
  .severity_text = severity_match.level ?? "INFO"
  .severity_number = if .severity_text == "ERROR" {
    17
  } else if .severity_text == "WARN" {
    13
  } else if .severity_text == "DEBUG" {
    5
  } else {
    9
  }


  .log_attributes = {
    "file.path": .file,
    "host.name": get_hostname()
  }


  del(.message)
  del(.source_type)
'''

Docker Containers

Section titled “Docker Containers”
[sources.docker_logs]
type = "docker_logs"
include_containers = ["*"]


[transforms.remap_docker_logs]
inputs = ["docker_logs"]
type = "remap"
source = '''
  .timestamp = now()
  .service_name = .container_name
  .namespace = "containers"
  .body = .message


  .severity_text = if match(.message, r"(?i)error|exception|fail|critical") {
    "ERROR"
  } else if match(.message, r"(?i)warn|warning") {
    "WARN"
  } else if match(.message, r"(?i)debug") {
    "DEBUG"
  } else {
    "INFO"
  }


  .severity_number = if .severity_text == "ERROR" {
    17
  } else if .severity_text == "WARN" {
    13
  } else if .severity_text == "DEBUG" {
    5
  } else {
    9
  }


  .log_attributes = {
    "container.name": .container_name,
    "container.image": .image,
    "container.id": .container_id
  }


  del(.message)
  del(.source_type)
'''

Journald (systemd)

Section titled “Journald (systemd)”
[sources.journald_logs]
type = "journald"
include_units = ["nginx", "sshd", "my-app"]


[transforms.remap_journald_logs]
inputs = ["journald_logs"]
type = "remap"
source = '''
  .timestamp = now()
  .service_name = ._SYSTEMD_UNIT ?? "unknown"
  .namespace = "systemd"
  .body = .message


  # journald uses numeric priority (0-7)
  priority = to_int(._PRIORITY) ?? 6
  .severity_text = if priority <= 3 {
    "ERROR"
  } else if priority == 4 {
    "WARN"
  } else if priority == 7 {
    "DEBUG"
  } else {
    "INFO"
  }


  .severity_number = if .severity_text == "ERROR" {
    17
  } else if .severity_text == "WARN" {
    13
  } else if .severity_text == "DEBUG" {
    5
  } else {
    9
  }


  .log_attributes = {
    "systemd.unit": ._SYSTEMD_UNIT ?? "",
    "host.name": ._HOSTNAME ?? "",
    "process.pid": to_string(._PID) ?? "",
    "process.uid": to_string(._UID) ?? ""
  }


  del(.message)
  del(.source_type)
'''

Enriching Logs

Section titled “Enriching Logs”

Add a second transform stage to attach environment metadata:

[transforms.enrich_logs]
inputs = ["remap_logs"]  # chain after your remap transform
type = "remap"
source = '''
  .log_attributes.environment = get_env_var!("ENVIRONMENT")
  .log_attributes.version = get_env_var!("APP_VERSION")
  .log_attributes.region = get_env_var!("REGION")
'''

Then update the sink to read from enrich_logs instead:

[sinks.clickhouse]
inputs = ["enrich_logs"]
# ... rest stays the same

If you also configured the VictoriaLogs sink, update its inputs to ["enrich_logs"] too so both backends receive the same enriched event.

Connect Logchef

Section titled “Connect Logchef”

Once logs are flowing into your chosen backend:

  1. Log in to Logchef
  2. Go to Sources > Add Source
  3. Select ClickHouse and enter host/database/table details, or select VictoriaLogs and enter its base URL and optional auth/tenant scope
  4. Create or select a team and assign the source

Start querying:

namespace="syslog" and severity_text="ERROR"
service_name="my-app" and body~"connection refused"
log_attributes.container.name="api-server"

See Search Syntax and Query Examples for more.