go-stash

go-stash — это высокопроизводительный, бесплатный конвейер обработки данных с открытым исходным кодом на стороне сервера, который принимает данные из Kafka, обрабатывает их и затем отправляет в ElasticSearch.

Производительность go-stash примерно в 5 раз выше, чем у logstash, и его легко развернуть, нужен только один исполняемый файл.

Быстрый старт

Установка

cd stash && go build stash.go

Быстрый старт

• С помощью бинарного файла:

./stash -f etc/config.yaml

• Через Docker, убедитесь, что путь к файлу конфигурации указан правильно.

docker run -d -v `pwd`/etc:/app/etc kevinwan/go-stash

Пример config.yaml:

Clusters:
- Input:
    Kafka:
      Name: go-stash
      Log:
        Mode: file
      Brokers:
      - "172.16.48.41:9092"
      - "172.16.48.42:9092"
      - "172.16.48.43:9092"
      Topic: ngapplog
      Group: stash
      Conns: 3
      Consumers: 10
      Processors: 60
      MinBytes: 1048576
      MaxBytes: 10485760
      Offset: first
  Filters:
  - Action: drop
    Conditions:
      - Key: status
        Value: 503
        Type: contains
      - Key: type
        Value: "app"
        Type: match
        Op: and
  - Action: remove_field
    Fields:
    - message
    - source
    - beat
    - fields
    - input_type
    - offset
    - "@version"
    - _score
    - _type
    - clientip
    - http_host
    - request_time
  Output:
    ElasticSearch:
      Hosts:
      - "http://172.16.188.73:9200"
      - "http://172.16.188.74:9200"
      - "http://172.16.188.75:9200"
      Index: "go-stash-{{yyyy.MM.dd}}"
      MaxChunkBytes: 5242880
      GracePeriod: 10s
      Compress: false
      TimeZone: UTC

Детали

Ввод

Conns: 3
Consumers: 10
Processors: 60
MinBytes: 1048576
MaxBytes: 10485760
Offset: first

Conns

• Количество ссылок на Kafka, количество ссылок основано на количестве ядер процессора, обычно <= количеству ядер процессора.

Consumers

• Число открытых потоков на соединение, правило расчёта — Conns * Consumers, не рекомендуется превышать общее количество срезов, например, если срез темы равен 30, Conns * Consumers <= 30.

Processors

• Количество потоков для обработки данных, зависит от количества ядер процессора, может быть увеличено соответствующим образом, рекомендуемая конфигурация: Conns * Consumers * 2 или Conns * Consumers * 3, например: 60 или 90.

MinBytes MaxBytes

• По умолчанию размер блока данных из Kafka составляет 1M~10M. Если сеть и IO лучше, можно настроить его выше.

Offset

• Необязательно last и false, по умолчанию — last, что означает чтение данных из kafka с начала.

Фильтры

- Action: drop
  Conditions:
    - Key: k8s_container_name
      Value: "-rpc"
      Type: contains
    - Key: level
      Value: info
      Type: match
      Op: and
- Action: remove_field
  Fields:
    - message
    - _source
    - _type
    - _score
    - _id
    - "@version"
    - topic
    - index
    - beat
    - docker_container
    - offset
    - prospector
    - source
    - stream
- Action: transfer
  Field: message
  Target: data

- Action: drop

• Флаг удаления: данные, соответствующие этому условию, будут удалены при обработке и не будут введены в es.
• Согласно условию удаления, укажите значение поля ключа и значение, поле типа может быть содержит (contains) или соответствует (match).
• Условие соединения Op: and, также можно записать или.

- Действие: remove_field

• Идентификатор поля для удаления, просто перечислите его ниже.

- Действие: transfer

• Идентификатор поля переноса: например, поле сообщения можно переопределить как поле данных.

Вывод

Индекс

• Имя индекса, индексное имя — {{yyyy.MM.dd}} или {{yyyy-MM-dd}}, в вашем собственном формате.

MaxChunkBytes

• Размер пакета, отправляемого в ES каждый раз, по умолчанию равен 5M, можно отрегулировать в соответствии с ситуацией ввода-вывода ES.

GracePeriod

• Значение по умолчанию равно 10 с, которое используется для обработки оставшегося потребления и данных в течение 10 секунд после закрытия программы и выхода.