Ресивер filelog является компонентом OpenTelemetry Collector с открытым исходным кодом и используется для сбора журналов экземпляра СУБД. За подробной информацией обратитесь к документации OpenTelemetry.

Ресивер filelog нужно добавить в секцию receivers и настроить.

Конфигурация ресивера зависит от настройки экземпляра СУБД и используемого формата журнала (см. параметры logging_collector и log_destination). Коллектор поддерживает сбор журналов в формате CSV и JSON.

Независимо от формата журнала, необходимо указать путь к каталогу журналов и шаблон имён журналов.

Пример конфигурации ресивера для сбора журналов в формате JSON:

receivers:
  filelog:
    include: [ /var/log/postgresql/*.json ]
    start_at: end
    retry_on_failure:
      enabled: true
      initial_interval: 1s
      max_interval: 30s
      max_elapsed_time: 5m
    operators:
      - type: csv_parser
        parse_ints: true
        timestamp:
          parse_from: attributes.timestamp
          layout_type: strptime
          layout: '%Y-%m-%d %H:%M:%S.%L %Z'
        severity:
          parse_from: attributes.error_severity
          mapping:
            debug: [ DEBUG ]
            info:  [ INFO, NOTICE, LOG ]
            warn:  [ WARNING ]
            error: [ ERROR ]
            fatal: [ FATAL, PANIC ]
      - type: remove
        field: attributes.timestamp

В разделе severity определяются правила обработки и классификации уровней важности журналов согласно их значимости и приоритету. За дополнительной информацией об этом параметре обратитесь к документации OpenTelemetry.

Пример конфигурации ресивера для сбора журналов в формате CSV:

receivers:
  filelog:
    include: [ /var/log/postgresql/*.csv ]
    start_at: end
    retry_on_failure:
      enabled: true
      initial_interval: 1s
      max_interval: 30s
      max_elapsed_time: 5m
    multiline:
      line_start_pattern: ^[0-9]{4}-[0-9]{2}-[0-9]{2}
    operators:
      - type: csv_parser
        header: timestamp,user,dbname,pid,connection_from,session_id,line_num,ps,session_start,vxid,txid,error_severity,state_code,message,detail,hint,internal_query,internal_position,context,statement,cursor_position,func_name,application_name,backend_type,leader_pid,query_id
        timestamp:
          parse_from: attributes.timestamp
          layout_type: strptime
          layout: '%Y-%m-%d %H:%M:%S.%L %Z'
        severity:
          parse_from: attributes.error_severity
          mapping:
            debug: [ DEBUG ]
            info:  [ INFO, NOTICE, LOG ]
            warn:  [ WARNING ]
            error: [ ERROR ]
            fatal: [ FATAL, PANIC ]
      - type: remove
        field: attributes.timestamp

Подробное описание параметров конфигурации с примерами можно найти в каталоге /usr/share/doc/pgpro-otel-collector/examples.

Ресивер journald — это компонент OpenTelemetry Collector с открытым исходным кодом для сбора записей журнала systemd, что особенно полезно для Shardman и Postgres Pro Enterprise Manager (PPEM). За подробной информацией обратитесь к документации OpenTelemetry.

Параметр log_destination в Postgres Pro должен быть настроен для отправки журналов в syslog. За подробностями обратитесь к разделу Регистрация ошибок и протоколирование работы сервера.

Пример настройки ресивера journald для отправки журналов в формате journald в PPEM для PostgreSQL:

receivers:
  journald:
    directory: /var/log/journal
    start_at: end
    units:
     - postgresql@17-main
    operators:
      # Переименовать _PID в pid. Это поле обязательно для PPEM
      - type: move
        id: "pid"
        from: body._PID
        to: attributes.pid
      # Переименовать __MONOTONIC_TIMESTAMP в line_num. Это поле обязательно для PPEM
      - type: move
        id: "line_num"
        from: body.__MONOTONIC_TIMESTAMP
        to: attributes.line_num
      # Переименовать MESSAGE в message
      - type: move
        id: "message"
        from: body.MESSAGE
        to: attributes.message
      # Переименовать _SYSTEMD_UNIT в backend_type
      - type: move
        id: "backend_type"
        field: attributes.backend_type
        from: body._SYSTEMD_UNIT
        to: attributes.backend_type
      # Преобразовать числовой PRIORITY в текстовый severity
      - type: severity_parser
        parse_from: body.PRIORITY
        overwrite_text: true
        mapping:
            debug: [ 7 ] # отладка
            info:  [ 5, 6 ] # уведомление, информация
            warn:  [ 4 ] # предупреждение
            error: [ 3 ] # ошибка
            fatal: [ 0, 1, 2 ] # аварийная, тревога, критическая
processors:
  batch/journald:
    send_batch_size: 4096
    timeout: 15s
  resource:
    attributes:
    - action: upsert
      key: service.name
      value: postgresql
    - action: upsert
      key: service.instance.id
      value: address-of-postgres-instance:5432
  attributes/convert:
    actions:
      - key: pid
        action: convert
        converted_type: int
      - key: line_num
        action: convert
        converted_type: int
  transform:
    log_statements:
      - context: log
        statements:
          # Установить атрибут error_severity из поля severity_text. Это поле обязательно для PPEM
          - set(log.attributes["error_severity"], log.severity_text)
          # Установить атрибут session_start. Это поле обязательно для PPEM
          - set(log.attributes["session_start"], FormatTime(log.time, "%Y-%m-%d %H:%M:%S %Z"))
          # Установить поле 'session_id' из _SYSTEMD_INVOCATION_ID или INVOCATION_ID. Это поле обязательно для PPEM
          - set(log.attributes["session_id"], log.body["_SYSTEMD_INVOCATION_ID"]) where log.body["_SYSTEMD_INVOCATION_ID"] != nil
          - set(log.attributes["session_id"], log.body["INVOCATION_ID"]) where (log.attributes["session_id"] == nil and log.body["INVOCATION_ID"] != nil)
exporters:
  otlphttp:
    endpoint: https://logs.example.org:8080
    tls:
      insecure_skip_verify: true
    compression: ""
    headers:
      "X-Ppem-Source-Agent-Name": ppem-agent
      "X-Ppem-Source-Instance-Port": '5432'
service:
  telemetry:
  pipelines:
    logs:
      receivers: [ journald ]
      processors: [ resource,attributes/convert,transform,batch/journald ]
      exporters: [ otlphttp ]

Этот ресивер также можно использовать для чтения журналов Shardman. Ниже приведён пример настройки для чтения таких журналов и их отправки в Elasticsearch.

receivers:
  journald:
    directory: /var/log/journal
    start_at: end
    units:
      - shardmand@*
    operators:
      # Разобрать сообщение shardmand на соответствующие поля журнала
      - type: regex_parser
        parse_from: body.MESSAGE
        regex: '^(?P<timestamp>.+)	(?P<level>.+)	(?P<message>.*)	(?P<fields>.+)$'
        timestamp:
          parse_from: attributes.timestamp
          layout_type: strptime
          layout: '%Y-%m-%dT%H:%M:%S.%f%z'
        severity:
          parse_from: attributes.level
      # Разобрать дополнительные атрибуты из JSON-строки 'fields'
      - type: json_parser
        parse_from: attributes.fields
      # Удалить разобранную временную метку во избежание дублирования
      - type: remove
        id: remove_timestamp
        field: attributes.timestamp
      # Удалить разобранный уровень после извлечения severity
      - type: remove
        id: remove_level
        field: attributes.level
      # Удалить разобранную JSON-строку 'fields' после разбора
      - type: remove
        id: remove_fields
        field: attributes.fields
      # Сохранить только поля, полезные для отладки
      - type: retain
        fields:
          - body._PID
          - body._GID
          - body._UID
          - body._CMDLINE
          - body._EXE
          - body._HOSTNAME
          - body._SYSTEMD_UNIT
          - body._TRANSPORT
processors:
  batch/journald:
    send_batch_size: 4096
    timeout: 15s
  transform:
    log_statements:
      - context: log
        statements:
          # Установить resource.process.pid из body._PID, затем удалить body._PID
          - set(resource.attributes["process.pid"], Int(log.body["_PID"]))
          - delete_key(log.body, "_PID")
          # Установить resource.process.command_line из body._CMDLINE (при наличии), затем удалить body._CMDLINE
          - set(resource.attributes["process.command_line"], log.body["_CMDLINE"]) where log.body["_CMDLINE"] != nil
          - delete_key(log.body, "_CMDLINE")
  resource:
    attributes:
    - action: upsert
      key: service.name
      value: shardmand
    - action: upsert
      key: service.instance.id
      value: address-of-shardman-instance:5432
exporters:
  otlphttp:
    compression: gzip
    endpoint: https://logs.example.org:8080
    tls:
      insecure_skip_verify: true
service:
  pipelines:
    logs:
      receivers: [ journald ]
      processors: [ transform,resource,batch/journald ]
      exporters: [ otlphttp ]

Процессоры attributes, resource и filter являются компонентами OpenTelemetry Collector с открытым исходным кодом.

Конфигурация процессоров также зависит от настройки экземпляра СУБД и используемого формата журнала (см. параметры logging_collector и log_destination).

Процессор resource требует настройки при отправке журналов в Elastic. Независимо от формата журналов, необходимо указать атрибуты service.name и service.instance.id.

Журналы можно фильтровать по уровню важности с помощью процессора filter, как показано в примерах ниже.

Пример конфигурации процессоров для сбора журналов в формате JSON:

processors:
  filter/include:
    logs:
      include:
        match_type: strict # Или regexp
        severity_texts:
          #- "DEBUG"
          - "INFO"
          - "NOTICE"
          - "WARNING"
          - "ERROR"
          - "LOG"
          - "FATAL"
          - "PANIC"
  attributes/convert:
    actions:
      - key: query_id
        action: convert
        converted_type: string
      - key: pid
        action: convert
        converted_type: string
  resource:
    attributes:
      - key: service.name
        action: upsert
        value: postgresql
      - key: service.instance.id
        action: upsert
        value: 1.2.3.4:5432

Пример конфигурации процессоров для сбора журналов в формате CSV:

processors:
  filter/include:
    logs:
      include:
        match_type: strict # Или regexp
        severity_texts:
          #- "DEBUG"
           - "INFO"
           - "NOTICE"
           - "WARNING"
           - "ERROR"
           - "LOG"
           - "FATAL"
           - "PANIC"
  attributes/convert:
    actions:
      - key: pid
        action: convert
        converted_type: int
      - key: line_num
        action: convert
        converted_type: int
      - key: txid
        action: convert
        converted_type: int
      key: remote_port
        action: convert
        converted_type: int
      - key: cursor_position
        action: convert
        converted_type: int
      - key: internal_position
        action: convert
        converted_type: int
      - key: leader_pid
        action: convert
        converted_type: int
  resource:
    attributes:
      - key: service.name
        action: upsert
        value: postgresql
      - key: service.instance.id
        action: upsert
        value: 1.2.3.4:5432

Экспортёр otlphttp является компонентом OpenTelemetry Collector с открытым исходным кодом и используется для отправки собранных журналов в OTLP-совместимую систему хранения или мониторинга, которая предварительно должна быть развёрнута и доступна. За подробной информацией обратитесь к документации OpenTelemetry.

Для настройки экспортёра otlphttp достаточно указать адрес целевой системы, куда следует отправить данные:

exporters:
  otlphttp:
    endpoint: https://otlp.example.org

Экспортёр kafka является компонентом OpenTelemetry Collector с открытым исходным кодом для отправки метрик и журналов в Apache Kafka. За подробной информацией обратитесь к документации OpenTelemetry.

В примере ниже показано, как настроить отправку журналов.

receivers:
  filelog:
    include: [ /var/log/postgresql/*.json ]
    start_at: end
    retry_on_failure:
      enabled: true
      initial_interval: 1s
      max_interval: 30s
      max_elapsed_time: 5m
    operators:
      - type: json_parser
        parse_ints: true
        timestamp:
          parse_from: attributes.timestamp
          layout_type: strptime
          layout: '%Y-%m-%d %H:%M:%S.%L %Z'
        severity:
          parse_from: attributes.error_severity
          mapping:
            debug: [ DEBUG ]
            info:  [ INFO, NOTICE, LOG ]
            warn:  [ WARNING ]
            error: [ ERROR ]
            fatal: [ FATAL, PANIC ]
      - type: remove
        id: remove_timestamp
        field: attributes.timestamp
exporters:
  kafka:
    brokers:
      - localhost:9092
    protocol_version: 2.1.0
    client_id: pgpro-otel-collector
    logs:
      topic: otlp_logs
      encoding: otlp_json # proto supported
    include_metadata_keys:
      - service.name
      - service.instance.id
    tls:
      insecure: true
    timeout: 30s
    producer:
      max_message_bytes: 1000000
      required_acks: 1
      compression: none # gzip, snappy, lz4, and zstd;
processors:
  batch/kafka:
    send_batch_size: 1024
    timeout: 1s
  attributes/convert:
    actions:
      - key: query_id
        action: convert
        converted_type: string
  resource:
    attributes:
      - key: service.name
        action: upsert
        value: postgresql
      - key: service.instance.id
        action: upsert
        value: address-of-postgres-instance:5432
service:
  pipelines:
    logs/kafka:
      receivers: [ filelog ]
      processors: [ batch/kafka,resource,attributes/convert ]
      exporters: [ kafka ]

После того как ресиверы, процессоры и экспортёры добавлены и настроены, нужно объединить их в конвейер. Конвейер настраивается в секции service. Состав конвейера напрямую зависит от предварительно добавленных компонентов (не существует конфигурации по умолчанию).

В примере ниже показано, как настроить конвейер для работы с журналами. Данные собираются ресивером filelog, обрабатываются процессорами resource и attributes и экспортируются otlphttp.

Таким образом, все компоненты, используемые в конвейере, также должны быть добавлены в файл конфигурации и настроены.

service:
  extensions: []
  pipelines:
    logs:
      receivers:
        - filelog
      processors:
        - resource
        - attributes/convert
      exporters:
        - otlphttp