@Vonng I am a huge fan of the 🐷 -- it keeps proving its value in flexibility across multiple domains.

Implementing a migration for a mature Rails app off an overly-expensive hosting provider; ended up using VictoriaMetrics + VictoriaLogs + Vector to feed Grafana since they are so lightweight and performant.

Personally think Victoria + Vextor is a better direction even though VictoriaLogs isn't drop-in compatible with Loki. VM being a drop-in replacement for Prometheus, though, makes that an easy choice.

Here is the vector.yml that evolved and is working to consolidate logs and forward them to the infra VictoriaLogs instances. Feel free to ignore the Rails-specific stuff in the transforms, but thought it might be helpful.

#==============================================================#
# Vector Configuration - Log Collection Agent  
# Path: {{ vector_config_dir }}/vector.yaml
# Generated by Ansible - DO NOT EDIT MANUALLY
# Aligned with Promtail configuration in node_monitor role
#==============================================================#

# Global settings
data_dir: "{{ vector_data_dir }}"

# API disabled - using only Prometheus metrics exporter
api:
  enabled: false

################################################################
#                        SOURCES                              #
################################################################

sources:
  ################################################################
  #                        Vector Metrics                        #
  ################################################################
  vector_metrics:
    type: internal_metrics

  vector_logs:
    type: internal_logs

  ################################################################
  #                        Node Logs                            #
  ################################################################
  # System logs (equivalent to Promtail nodes job)
  syslog:
    type: file
    include:
{% if os_package|default('rpm') == 'deb' %}
      - "/var/log/syslog"
{% else %}
      - "/var/log/messages"
{% endif %}
    read_from: beginning

  # Kernel messages
  dmesg:
    type: file
    include:
      - "/var/log/dmesg"
    read_from: beginning

  # Cron logs
  cron:
    type: file
    include:
      - "/var/log/cron"
    read_from: beginning

{% if docker_socket_stat.stat.exists %}
  # Docker container logs (only if socket exists)
  docker:
    type: docker_logs
    docker_host: "{{ vector_docker_socket }}"
{% endif %}

{% if inventory_hostname in groups['infra']|default([]) and (nginx_enabled is not defined or nginx_enabled|bool) %}
  ################################################################
  #                       Infra Logs                            #
  ################################################################
  # Nginx access logs
  nginx_access:
    type: file
    include:
      - "/var/log/nginx/*.log"
    read_from: beginning

  # Nginx error logs
  nginx_error:
    type: file
    include:
      - "/var/log/nginx-error.log"
    read_from: beginning
{% endif %}

{% if pg_cluster is defined and pg_seq is defined %}
  ################################################################
  #                      PostgreSQL Logs                        #
  ################################################################
  # PostgreSQL logs
  postgres:
    type: file
    include:
      - "{{ pg_log_dir }}/*.json"
    read_from: beginning

{% if patroni_enabled is defined and patroni_enabled|bool %}
  # Patroni logs
  patroni:
    type: file
    include:
      - "{{ patroni_log_dir }}/patroni.log"
    read_from: beginning
{% endif %}

{% if pgbackrest_enabled is defined and pgbackrest_enabled|bool %}
  # pgBackRest logs
  pgbackrest:
    type: file
    include:
      - "{{ pgbackrest_log_dir }}/*.log"
    read_from: beginning
    fingerprint:
      strategy: checksum
      lines: 2
{% endif %}

{% if pgbouncer_enabled is defined and pgbouncer_enabled|bool %}
  # PgBouncer logs
  pgbouncer:
    type: file
    include:
      - "{{ pgbouncer_log_dir }}/pgbouncer.log"
    read_from: beginning
{% endif %}
{% endif %}

{% if redis_cluster is defined and redis_node is defined %}
  ################################################################
  #                         Redis Logs                          #
  ################################################################
  # Redis logs
  redis:
    type: file
    include:
      - "/var/log/redis/*.log"
    read_from: beginning
{% endif %}

################################################################
#                      TRANSFORMS                             #
################################################################

transforms:
  ################################################################
  #                     Vector Metrics Enrichment               #
  ################################################################
  # Add missing labels to vector metrics for dashboard compatibility
  vector_metrics_enriched:
    type: remap
    inputs:
      - vector_metrics
    source: |
      .tags.job = "vector"
      .tags.nodename = "{{ nodename }}"
      .tags.instance = "{{ inventory_hostname }}:{{ vector_port }}"

  ################################################################
  #                     Vector Logs Enrichment                  #
  ################################################################
  # Parse and enrich Vector internal logs
  vector_logs_parsed:
    type: remap
    inputs:
      - vector_logs
    source: |
      .cls = "{{ node_cluster|default('nodes') }}"    
      .job = "vector"
      .src = "vector"
      .ins = "{{ nodename }}"      
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }

  ################################################################
  #                        Node Transforms                      #
  ################################################################
  # Parse and enrich syslog (equivalent to Promtail nodes job)
  syslog_parsed:
    type: remap
    inputs:
      - syslog
    source: |
      .cls = "{{ node_cluster|default('nodes') }}"    
      .job = "node"
      .src = "syslog"
      .ins = "{{ nodename }}"      
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }

  dmesg_parsed:
    type: remap
    inputs:
      - dmesg
    source: |
      .cls = "{{ node_cluster|default('nodes') }}"    
      .job = "node"
      .src = "dmesg"
      .ins = "{{ nodename }}"      
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }

  cron_parsed:
    type: remap
    inputs:
      - cron
    source: |
      .cls = "{{ node_cluster|default('nodes') }}"    
      .job = "node"
      .src = "cron"
      .ins = "{{ nodename }}"
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }

{% if docker_socket_stat.stat.exists %}
  docker_parsed:
    type: remap
    inputs:
      - docker
    source: |
      # Essential stream identification
      .cls = "{{ node_cluster|default('nodes') }}"    
      .job = "docker"
      if exists(.label.src) {
        .src = .label.src
      } else {
        .src = "docker"
      }
      .ins = "{{ nodename }}"      
      .ip = "{{ inventory_hostname }}"

      # Container identification (simplified)
      .container = .container_name
      if exists(.image) && is_string(.image) {
        image_parts, split_err = split(.image, ":")
        if split_err == null {
          .image_name = image_parts[0] || .image
          .image_tag = image_parts[1] || "latest"
        } else {
          .image_name = .image
          .image_tag = "latest"
        }
      }
      
      # Application metadata (only useful labels)
      if exists(.label.app) {
        .application = .label.app
      }
      if exists(.label.service) {
        .service = .label.service
      } else {
        .service = .container_name
      }
      if exists(.label.environment) {
        .environment = .label.environment
      } else {
        .environment = "unknown"
      }
      if exists(.label.version) {
        .version = .label.version
      }

      # Parse application-specific logs
      message_str = to_string(.message) ?? ""
      
      # Parse Rails application logs
      if .container == "fmp-app" && message_str != "" && match(message_str, r'^\w+,\s+\[.*?\]\s+\w+\s+--\s+:') {
        # Rails log format: "I, [2025-09-03T15:01:32.090855 #27]  INFO -- : [request_id] (duration) message"
        parsed, err = parse_regex(message_str, r'(?P<severity>\w+),\s+\[(?P<timestamp>[^\]]+)\]\s+(?P<level>\w+)\s+--\s+:\s+(?:\[(?P<request_id>[^\]]+)\]\s+)?(?:\((?P<duration>[\d.]+)s\)\s+)?(?P<query>.*)')
        if err == null && parsed != null {
          if exists(parsed.level) {
            .log_level, downcase_err = downcase(parsed.level)
            if downcase_err != null {
              .log_level = parsed.level
            }
          }
          if exists(parsed.request_id) {
            .request_id = parsed.request_id
          }
          if exists(parsed.duration) {
            .query_duration_seconds, duration_err = to_float(parsed.duration)
          }
          if exists(parsed.query) {
            .query = parsed.query
          }
          .parsed = true
        }
      }

      # Parse Caddy JSON logs (for caddy containers)
      if .container == "fmp-caddy" && message_str != "" {
        caddy_json, json_err = parse_json(message_str)
        if json_err == null && caddy_json != null {
          # Caddy log level and metadata
          if exists(caddy_json.level) {
            .log_level = caddy_json.level
          }
          if exists(caddy_json.logger) {
            .logger = caddy_json.logger
          }
          if exists(caddy_json.msg) {
            .message = caddy_json.msg
          }
          
          # HTTP request details (for access logs)
          if exists(caddy_json.request) {
            .http_method = caddy_json.request.method
            .http_host = caddy_json.request.host
            .http_uri = caddy_json.request.uri
            .http_proto = caddy_json.request.proto
            .client_ip = caddy_json.request.client_ip
          }

          # HTTP response details
          if exists(caddy_json.status) {
            .http_status = caddy_json.status
          }
          if exists(caddy_json.duration) {
            .response_duration_seconds = caddy_json.duration
          }
          if exists(caddy_json.size) {
            .response_size_bytes = caddy_json.size
          }
          if exists(caddy_json.bytes_read) {
            .request_size_bytes = caddy_json.bytes_read
          }
          
          # TLS information
          if exists(caddy_json.request.tls) {
            .tls_version = caddy_json.request.tls.version
            .tls_cipher = caddy_json.request.tls.cipher_suite
            .tls_proto = caddy_json.request.tls.proto
            .tls_server_name = caddy_json.request.tls.server_name
          }
          
          # Extract useful headers
          if exists(caddy_json.request.headers) {
            if exists(caddy_json.request.headers."Cf-Connecting-Ip") {
              .cf_connecting_ip = caddy_json.request.headers."Cf-Connecting-Ip"[0]
            }
            if exists(caddy_json.request.headers."User-Agent") {
              .user_agent = caddy_json.request.headers."User-Agent"[0]
            }
            if exists(caddy_json.request.headers."Cf-Ray") {
              .cloudflare_ray = caddy_json.request.headers."Cf-Ray"[0]
            }
            if exists(caddy_json.request.headers."Cf-Ipcountry") {
              .country = caddy_json.request.headers."Cf-Ipcountry"[0]
            }
          }
          
          # Extract Rails request ID from response headers
          if exists(caddy_json.resp_headers) && exists(caddy_json.resp_headers."X-Request-Id") {
            .request_id = caddy_json.resp_headers."X-Request-Id"[0]
          }
          
          .parsed = true
        }
      }

      # Clean up - remove noisy fields
      del(.label)
      del(.container_id)
      del(.container_created_at)
      del(.source_type)
      del(.image)

      # Ensure required fields exist
      if !exists(.timestamp) {
        .timestamp = .created_at || now()
      }
      if !exists(.message) {
        .message = ""
      }
{% endif %}

{% if inventory_hostname in groups["infra"]|default([]) and (nginx_enabled is not defined or nginx_enabled|bool) %}
  ################################################################
  #                       Infra Transforms                      #
  ################################################################
  # Parse and enrich Nginx access logs
  nginx_access_parsed:
    type: remap
    inputs:
      - nginx_access
    source: |
      .cls = "{{ node_cluster|default('nodes') }}"    
      .job = "infra"
      .src = "nginx"
      .ins = "{{ nodename }}"
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }

  # Parse and enrich Nginx error logs
  nginx_error_parsed:
    type: remap
    inputs:
      - nginx_error
    source: |
      .cls = "{{ node_cluster|default('nodes') }}"    
      .job = "infra"
      .src = "nginx-error"
      .ins = "{{ nodename }}"      
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }
{% endif %}

{% if pg_cluster is defined and pg_seq is defined %}
  ################################################################
  #                    PostgreSQL Transforms                    #
  ################################################################
  # Parse and enrich PostgreSQL logs
  postgres_parsed:
    type: remap
    inputs:
      - postgres
    source: |
      # Parse JSON first
      parsed, err = parse_json(.message)
      if err == null {
        # Merge parsed JSON fields
        . = merge!(., parsed)

        # Set stream identification fields after successful parse
        .cls = "{{ pg_cluster }}" 
        .job = "pgsql"
        .src = "postgres"
        .ins = "{{ pg_cluster }}-{{ pg_seq }}"
        .ip = "{{ inventory_hostname }}"

        # Set the message field for searchability
        .message = parsed.message || ""

        # Use PostgreSQL's timestamp if available
        if exists(parsed.timestamp) {
          ts, ts_err = parse_timestamp(parsed.timestamp, format: "%Y-%m-%d %H:%M:%S%.3f %Z")
          if ts_err == null {
            .timestamp = ts
          } else {
            .timestamp = now()
          }
        } else {
          .timestamp = now()
        }
      } else {
        # If JSON parsing fails, set basic fields
        .cls = "{{ pg_cluster }}" 
        .job = "pgsql"
        .src = "postgres"
        .ins = "{{ pg_cluster }}-{{ pg_seq }}"
        .ip = "{{ inventory_hostname }}"
        .message = .message || ""
        .timestamp = now()
      }

{% if patroni_enabled is defined and patroni_enabled|bool %}
  # Parse and enrich Patroni logs
  patroni_parsed:
    type: remap
    inputs:
      - patroni
    source: |
      .cls = "{{ pg_cluster }}"    
      .job = "pgsql"
      .src = "patroni"
      .ins = "{{ pg_cluster }}-{{ pg_seq }}"
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }
{% endif %}

{% if pgbackrest_enabled is defined and pgbackrest_enabled|bool %}
  # Parse and enrich pgBackRest logs
  pgbackrest_parsed:
    type: remap
    inputs:
      - pgbackrest
    source: |
      .cls = "{{ pg_cluster }}"    
      .job = "pgsql"
      .src = "pgbackrest"
      .ins = "{{ pg_cluster }}-{{ pg_seq }}"
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }
{% endif %}

{% if pgbouncer_enabled is defined and pgbouncer_enabled|bool %}
  # Parse and enrich PgBouncer logs
  pgbouncer_parsed:
    type: remap
    inputs:
      - pgbouncer
    source: |
      .cls = "{{ pg_cluster }}"    
      .job = "pgsql"
      .src = "pgbouncer"
      .ins = "{{ pg_cluster }}-{{ pg_seq }}"
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }
{% endif %}
{% endif %}

{% if redis_cluster is defined and redis_node is defined %}
  ################################################################
  #                       Redis Transforms                      #
  ################################################################
  # Parse and enrich Redis logs
  redis_parsed:
    type: remap
    inputs:
      - redis
    source: |
      .cls = "{{ redis_cluster }}"    
      .job = "redis"
      .src = "redis"
      .ins = "{{ redis_cluster }}-{{ redis_node }}"
      .ip = "{{ inventory_hostname }}"

      if !exists(.timestamp) {
        .timestamp = now()
      }
      if !exists(.message) {
        .message = ""
      }
{% endif %}

################################################################
#                         SINKS                               #
################################################################

sinks:
{% for infra_host in groups['infra']|default([]) %}
  # Send directly to VictoriaLogs instance {{ loop.index }}
  victorialogs_{{ loop.index }}:
    type: http
    inputs:
      - syslog_parsed
      - dmesg_parsed
      - cron_parsed
      - vector_logs_parsed
{% if docker_socket_stat.stat.exists %}
      - docker_parsed
{% endif %}
{% if inventory_hostname in groups["infra"]|default([]) and (nginx_enabled is not defined or nginx_enabled|bool) %}
      - nginx_access_parsed
      - nginx_error_parsed
{% endif %}
{% if pg_cluster is defined and pg_seq is defined %}
      - postgres_parsed
{% if patroni_enabled is defined and patroni_enabled|bool %}
      - patroni_parsed
{% endif %}
{% if pgbackrest_enabled is defined and pgbackrest_enabled|bool %}
      - pgbackrest_parsed
{% endif %}
{% if pgbouncer_enabled is defined and pgbouncer_enabled|bool %}
      - pgbouncer_parsed
{% endif %}
{% endif %}
{% if redis_cluster is defined and redis_node is defined %}
      - redis_parsed
{% endif %}

    # VictoriaLogs endpoint
    uri: "http://{{ infra_host }}:{{ victoria_logs_port|default(3100) }}/insert/jsonline"

    # Headers for VictoriaLogs (matching Promtail external_labels)
    request:
      headers:
        VL-Stream-Fields: "{{ vector_stream_fields }}"
        VL-Time-Field: "{{ vector_time_field }}"
        VL-Msg-Field: "{{ vector_msg_field }}"
        AccountID: "0"
        ProjectID: "0"

    # Encoding
    encoding:
      codec: json

    framing:
      method: newline_delimited

    # Performance settings
    compression: "{{ vector_compression }}"
    healthcheck:
      enabled: {{ vector_healthcheck_enabled | lower }}

    buffer:
      type: memory
      max_events: {{ vector_buffer_max_events }}
      when_full: "{{ vector_buffer_when_full }}"

    batch:
      max_events: {{ vector_batch_max_events }}
      timeout_ms: {{ vector_batch_timeout_ms }}

{% endfor %}

  # Expose Vector metrics for Prometheus scraping
  vector_metrics_http:
    type: prometheus_exporter
    inputs:
      - vector_metrics_enriched
    address: "0.0.0.0:{{ vector_port }}"
    default_namespace: vector

Replace promtail with vector #575

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions