Vector: отправка логов в Yandex SQS (AWS SQS)

Введение

Понадобилось собирать логи приложений из кластера k8s, который размещён в yandex cloud. В качестве хранилища могут выступать разные варианты: облачный PV и какой-нибудь loki, s3 bucket или ещё что-нибудь привычное.

Но в моём случае доступ к полученным логам должен осуществляться из другой системы для последующей обработки уже внутри контура заказчика, а потому нужен был какой-то публичный инструмент, в который можно сходить изнутри и забрать данные.

Изначально было испробовано складировать логи в s3 бакет и забирать оттуда, но из-за отсутствия механизма offset для понимания, был прочитан лог или нет, схема работала плохо – было много дублей.

Для решения такой задачи отлично подходит Vector: у него есть возможность собирать логи из почти любого источника. И передавать тоже много куда можно.

А в качестве хранилища с публичным доступом в облаке яндекса есть aws-совместимая очередь:

Yandex Message Queue — это сервис очередей сообщений, который позволяет распределенным приложениям и микросервисам взаимодействовать при помощи обмена сообщениями.

Очередь создается в облаке и достаточно гибко настраивается кликопсом в UI или тем же terraform при необходимости. Поддерживается стандартная очередь или FIFO, дедупликация.

Настройка Vector

Важный момент! Amazon перешла с XML на JSON в протоколе доступа к SQS. Это означает, что все версии Vector, использующие AWS-SDK свежее 3.285.0, вызывают ошибки авторизации.

Могут быть ошибки вида: “<Message>Action param was not found.</Message>” при отправке сообщений в yandex sqs с помощью vector 0.28.2+ версий.

Для работы понадобится версия Vector < 0.28.2.

Установка была выполнена helm-чартом с ролью “Agent” и запуском инстансов вектора через daemonset.

Конфиг vector для сбора логов в формате json от nginx:

sources:
    nginx-ext:
      type: kubernetes_logs
      auto_partial_merge: true
      # data_dir: /var/local/lib/vector/
      delay_deletion_ms: 60000
      extra_label_selector: app=nginx
      extra_namespace_label_selector: kubernetes.io/metadata.name=nginx
      node_annotation_fields:
        node_labels: ""
      namespace_annotation_fields:
        namespace_labels: ""
      pod_annotation_fields:
        container_id: ""
        container_image: ""
        # container_image_id: ""
        container_name: ""
        pod_annotations: ""
        pod_ip: ""
        pod_ips: ""
        pod_labels: ""
        pod_name: ""
        pod_namespace: ""
        pod_node_name: ""
        pod_owner: ""
        pod_uid: ""

  transforms:
    drop-fields:
      inputs:
        - nginx-ext
      type: "remap"
      drop_on_abort: true
      source: |
        . = parse_json!(.message)
        # accept only 200 OK messages
        http_status_code = parse_int!(.status)

        if http_status_code != 200 {
            abort
        }
        
  sinks:
    sqs:
      type: aws_sqs
      inputs:
        - nginx-ext
      endpoint: "https://message-queue.api.cloud.yandex.net"
      queue_url: "https://....."
      region: "ru-central1"
      healthcheck: false
      encoding: 
        codec: "json"
      auth:
        access_key_id: "////"
        secret_access_key: "////"

Схема работы следующая:

В source блоке указывается парсинг логов от приложения с селектором app=nginx и неймспейса nginx
портянка из pod_annotation_fields и другими секциями обнуляет поля, которые заполняются мета-информацией из pod и кластера в целом, в данном случае они были не нужны и сразу отсекались
в transform блоке используется VRL-язык, с помощью которого можно преобразовывать события логов, в данном случае исходный json разбирается на секции и по 200-му коду ответу сообщения передаются в SQS, а иначе просто отбрасываются.

Очередь по дефолту хранит данные 4 дня, при вычитывании события удаляются. В кач-ве consumer к SQS также может выступать Vector.

При первом заходе с VRL может быть сложно разобраться, пригодится плейграунд, где можно поупражняться на примерах: https://playground.vrl.dev/