Перейти к основному содержимому
Версия: 3.2

Безагентский сбор данных

Общие рекомендации​

Безагентский способ сбора данных подразумевает собой использование Logstash и его плагинов.

Все нижеуказанные конфигурации работают при условии установленных плагинов сбора данных.

При разборе каких-либо данных на составляющие, рекомендуется придерживаться определённых стандартов для поддержания универсальности и улучшения миграции между версиями. Общие названия таких стандартов в Elastic - Elastic Common Schema (ECS). C версии Logstash 8.x по умолчанию используется стандарт v8. Для версий Logstash 7.x применение ECS опционально.

примечание

Мы рекомендуем придерживаться стандарта ECS v8.

Обращайте внимание на используемую версию Logstash при просмотре параметров для определённого плагина на официальном сайте.

В одном pipeline может быть несколько источников. Рекомендуется так делать только если собираемые данные из разных источников имеют одинаковый формат. Например, вы собираете в одном месте логи веб сервера nginx и требуется указать несколько файлов для обработки.

Некоторые плагины имеют ряд обязательных параметров. Список таких параметров приведен на официальной странице документации каждого плагина.

Зачастую плагины принимают информацию, представленную в различных форматах данных. Чтобы их нормально обработать и разбить для удобства дальнейшего анализа, необходимо использовать фильтры. Если получаемые данные вас устраивают и не требуют дополнительной обработки, то блок filter можно опустить.

Все плагины Logstash включены в дистрибутив инсталлятора Smart Monitor и устанавливаются автоматически. Мы используем Logstash версии 8.4, поэтому все примеры приведены для нее.

Сохранение учетных данных для подключения к источникам​

Хранение учетных данных для подключения к источникам осуществляется при помощи зашифрованного хранилища logstash-keystore.

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

# Переход в каталог исполняемых файлов, путь по умолчанию
cd /app/logstash/bin/

# Добавление в хранилище записи пользователя с именем source.user. Ввод осуществляется с клавиатуры
sudo -u logstash ./logstash-keystore add source.user

# Добавление в хранилище записи пароля с именем source.password. Ввод осуществляется с клавиатуры
sudo -u logstash ./logstash-keystore add source.password

# Добавление в хранилище записи SNMP Community с именем source.snmp_comm. Ввод осуществляется с клавиатуры
sudo -u logstash ./logstash-keystore add source.snmp_comm

# Проверка списка хранимых записей
sudo -u logstash ./logstash-keystore list

Далее в конфигурационных файлах сбора данных сохраненные учетные данные указываются в виде токенов keystore:

user => "${source.user}"
password => "${source.password}"
community => "${source.snmp_comm}"
Обратите внимание

Сохранение любых идентификационных данных (имя пользователя, пароль, имя БД, имя SNMP community и т.д.) в конфигурационных файлах в открытом виде небезопасно.

Часто используемые плагины​

file​

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

Обратите внимание

Для нормальной работы плагина нужно убедиться, что Logstash (обычно используется одноименные пользователь и группа "logstash") имеют доступ к директории, где находится указанный файл и есть права на чтение самого файла.

В примере ниже считываем данные с файла HR2m.csv и модифицируем полученные данные перед отправкой дальше. Ниже приведены часто используемые параметры плагина:

  • path - путь к файлу, является обязательным параметром
  • start_position - с какого места читать файл. beginning - чтение файла с начала, end - чтение только последних записей, значением по умолчанию является end;
  • sincedb_path - файл базы данных sincedb для хранения последней прочитанной позиции из файла.
input {
  file {
    path => "/path/to/HR2m.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

filter {
    csv{
        columns => [ "id", "name_prefix", "first_name", "middle_initial", "last_name", "gender", "e_mail", "fathers_name", "mothers_name", "mothers_maiden_name", "date_of_birth", "time_of_birth", "age_in", "weight", "date_of_joining", "quarter_of_joining", "half_of_joining", "year_of_joining", "month_of_joining", "month_name_of_joining","short_month", "day_of_joining", "dow_of_joining", "short_dow", "age", "salary", "last_hike", "ssn", "phone", "place_name", "county", "city", "state", "zip", "region", "username", "password"]
        separator => ","
        skip_header => true
        source => "message"
    }
}

Обратите внимание, что в данном примере в блоке filter производится дополнительная манипуляция с данными - использование плагина фильтра csv, который позволит разбить каждое сообщение по полям.

tcp​

Чтение событий через сокет tcp в режиме сервера или клиента. Подробнее можно прочитать на официальной странице документации.

В примере конфигурации ниже считываем события через tcp сокет по порту 5411 в режиме сервера (стороннее приложение будет отправлять данные на порт 5411 данного сервера), данные будут поступать в json формате, после считывания без предварительной обработки будут отправлены дальше с помощью плагинов в блоке output.

Обратите внимание, что в данном примере не используется дополнительная обработка получаемых данных, как следствие опущен блок filter. Также не указан необязательный параметр host. Для данного примера параметр mode также можно опустить.

  • port - порт для прослушивания, обязательный параметр
  • mode - режим работы плагина, может быть "server" или "client", по умолчанию используется "server"
  • host - для режима работы сервер - указывается адрес, на котором принимать данные, для режима клиент - указывается адрес, откуда забирать данные. Значением по умолчанию является "0.0.0.0"
  • codec - определяет формат данных источника
input {
  tcp {
    port => 5411
    mode => "server"
    codec => "json"
  }
}

snmp​

SNMP опрашивает источники для сбора информации, связанной с текущим состоянием работы и системной информации. Подробнее можно прочитать на официальной странице документации.

В примере ниже считываются метрики с нескольких рабочих станций. Обратите внимание, что в одном pipeline можно указать несколько плагинов, в том числе одинаковых. Стоит обратить внимание, что плагин не имеет обязательных параметров. Часто используемые параметры приведены ниже:

  • get - запрос значений заданных конечных OID, указывается в виде списка (массив)
  • walk - запрос значений всех вложенных (нижестоящих) OID, указывается в виде списка (массив)
  • tables - запрос значений из нескольких walk-подзапросов (столбцов), указывается в виде таблицы (двухмерный массив)
  • hosts - список хостов для запроса настроенных опций, указывается в виде списка (массив)
  • type - добавляется поле "type" для всех данных из данного источника, позволяет легче обрабатывать данные по "маркировке", например, в блоке filter. Параметр не является уникальным параметром данного фильтра, его можно использовать для всех плагинов в блоке input
input {
  snmp {
    get => ["1.3.6.1.2.1.2.2.1.8.1", "1.3.6.1.2.1.2.2.1.10.1", "1.3.6.1.2.1.2.2.1.16.1", "1.3.6.1.2.1.2.2.1.2.1", "1.3.6.1.2.1.2.2.1.6.1"]
    walk => ["1.3.6.1.2.1.2.2.1.2.1", "1.3.6.1.2.1.4.20.1.1", "1.3.6.1.2.1.4.20.1.2", "1.3.6.1.2.1.4.20.1.3"]
    tables => [{"name" => "table_name" "columns" => ["1.3.6.1.2.1.2.2.1.1", "1.3.6.1.2.1.2.2.1.2", "1.3.6.1.2.1.2.2.1.3"]}]
    hosts => [{host => "udp:172.16.0.110/161" community => "${community.name}" version => "2c"  retries => 2  timeout => 1000},
    {host => "udp:172.16.0.146/161" community => "${community.name}" version => "2c"  retries => 2  timeout => 1000},
    {host => "udp:172.16.0.148/161" community => "${community.name}" version => "2c"  retries => 2  timeout => 1000}]
    type => "1"
  }
  snmp {
    get => ["1.3.6.1.2.1.2.2.1.8.2", "1.3.6.1.2.1.2.2.1.10.2", "1.3.6.1.2.1.2.2.1.16.2", "1.3.6.1.2.1.2.2.1.2.2", "1.3.6.1.2.1.2.2.1.6.2"]
    walk => ["1.3.6.1.2.1.2.2.1.2.1", "1.3.6.1.2.1.4.20.1.1", "1.3.6.1.2.1.4.20.1.2", "1.3.6.1.2.1.4.20.1.3"]
    hosts => [{host => "udp:172.16.0.110/161" community => "${community.name}" version => "2c"  retries => 2  timeout => 1000},
    {host => "udp:172.16.0.146/161" community => "${community.name}" version => "2c"  retries => 2  timeout => 1000},
    {host => "udp:172.16.0.148/161" community => "${community.name}" version => "2c"  retries => 2  timeout => 1000}]
    type => "2"
  }
}

jdbc​

JDBC-плагин позволяет извлекать данные из любой базы данных с интерфейсом JDBC. Подробнее можно прочитать на официальной странице документации. Часто используемые параметры представлены ниже:

  • jdbc_driver_class - драйвер JDBC к соответствующей базе данных. Обязательный параметр
  • jdbc_connection_string - строка подключения JDBC, может быть указан логин и пароль к базе данных здесь, но мы не рекомендуем так делать. Обязательный параметр
  • jdbc_user - пользователь базы данных, должен иметь права на выполнение соответствующего запроса. Обязательный параметр
  • jdbc_password - пароль пользователя базы данных. Мы рекомендуем не указывать пароль в открытом виде, а использовать keystore
  • statement - запрос к БД для выполнения, позволяет использовать параметры, которые можно задать в параметре parameters. Параметр не явно является обязательным, вместо этого параметра можно использовать statement_filepath
  • schedule - расписание запуска плагина с помощью rufus-scheduler, синтаксис похож на cron, но имеет гораздо больше возможностей
  • use_column_value - определяет значение переменной sql_last_value, если установлено true - используется значение параметра tracking_column, иначе - время последнего выполнения запроса
  • tracking_column - столбец, значение которого должно отслеживаться, если параметр use_column_value установлен true
  • tracking_column_type - тип столбца, указанного в параметре tracking_column. Может принимать одно из двух значений - "numeric", "timestamp". По умолчанию используется "numeric"
  • last_run_metadata_path - путь к файлу с последним временем выполнения запроса
input {
  jdbc {
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://<ip>\\sqlexpress:<port>"
    jdbc_user => "${db_username}"
    jdbc_password => "${db_password}"
    statement => "SELECT MNAME, LASTACTIVITY, CLIENTID, VERSION FROM [SN7_SERVER_SCHEMA].[dbo].[CLIENT] WHERE CLIENTCLASS = 'Service' GROUP BY MNAME, LASTACTIVITY, CLIENTID, VERSION"
    schedule => "1-59/5 * * * *"
    use_column_value => true
    tracking_column => clientid
    tracking_column_type => "numeric"
    last_run_metadata_path => "/path/to/metadata/file"
  }
}

kafka​

Позволяет считывать события из Apache Kafka. Подробнее можно прочитать в нашей документации или на официальной странице документации плагина. Плагин не имеет обязательных параметров, основные используемые параметры представлены ниже:

  • topics - список (массив) topic для чтения, значение по умолчанию ["logstash"]
  • bootstrap_servers - адрес брокера Apache Kafka. Значение по умолчанию "localhost:9092"
  • consumer_threads - число потоков для считывания данных из Apache Kafka, по умолчанию 1
  • decorate_events - позволяет добавлять к считанным данным дополнительную информацию из Apache Kafka. Добавляется поле kafka, включающая параметры topic, consumer_group, partition, offset, key. Может принимать одно из значений - "none", "basic", "extended". по умолчанию имеет значение "none"
  • auto_offset_reset - выбор поведения в случае отсутствия значения offset или если оно не корректное. Может принимать значения "earlist", "latest", "none". Если группа, в которой состоит consumer, ранее не использовала данный topic, то при параметре "earliest" будут считаны все существующие записи из Kafka
  • client_id и group_id — имя и группа consumer. Стоит отметить, что если в одном pipeline используется несколько входных плагинов Kafka, то рекомендуется задать для каждого своего пользователя и свою группу
  • type - добавляется поле "type" для всех данных из данного источника, позволяет легче обрабатывать данные по "маркировке", например, в блоке filter. Параметр не является уникальным параметром данного фильтра, его можно использовать для всех плагинов в блоке input. Рекомендуется использовать данный параметр если считываются данные из Kafka несколькими logstash серверами для дополнительных возможностей при анализе в дальнейшем
input {
  kafka {
    topics => ["volgablob"]
    bootstrap_servers => '172.16.0.26:9092'
    consumer_threads => 3
    decorate_events => "basic"
    auto_offset_reset => "earliest"
    client_id => "manylogstash_2"
    group_id => "manylogstash"
    type => "logstash2"
  }
}

http_poller​

Позволяет вызывать HTTP API и обработать полученные данные. Читает из списка URL адресов. Подробнее можно прочитать на официальной странице документации.

urls - hash URL-адресов в формате "name" => "url". Оба значения будут переданы вместе с данными. "url" может быть как строкой, так и hash. Обязательный параметр. В случае "url" как hash может содержать параметры поддерживаемые Manticore, следующие параметры используются чаще всего:

  • url - строка url, обязательный параметр
  • method - HTTP метод для использования, по умолчанию используется GET
  • user - пользователь для авторизации по схеме HTTP Basic Auth
  • password - пароль пользователя для авторизации по схеме HTTP Basic Auth
  • headers - hash полей - значений
  • schedule - расписание запуска плагина с помощью rufus-scheduler, синтаксис похож на cron, но имеет гораздо больше возможностей. Обязательный параметр
  • request_timeout - время ожидания (в секундах) для всего запроса, значение по умолчанию 60 секунд;
  • codec - позволяет преобразовать получаемые данные перед приёмом по какому-нибудь стандарту, является параметром универсальным для всех плагинов в блоке input. Значение по умолчанию "json". Список доступных кодеков можно посмотреть в официальной документации
input {
  http_poller {
    urls => {
      serverclasses => {
        method => get
        user => "${username}"
        password => "${password}"
        url => "https://<opensearch_ip>:<opensearch_port>/serverclasses"
     }
    }
    request_timeout => 60
    schedule => { cron => "* * * * * UTC"}
    codec => "json"
  }
}

http​

Позволяет получать однострочные или многострочные события через HTTP(S), может как получать, так и принимать webhook запросы. Подробнее можно прочитать на официальной странице документации. Плагин не имеет обязательных параметров. Часто используемые параметры представлены ниже:

  • host - хост сервера или IP адрес, на котором принимать данные, значение по умолчанию "0.0.0.0"
  • port - tcp порт если плагин работает в режиме приёма. По умолчанию используется порт 8080
input {
  http {
    host => "localhost"
    port => 8080
  }
}

LDAPSearch​

Позволяет отфильтровать множество записей с сервера LDAP и сгенерировать событие по извлеченной записи. Каждый набор данных содержит атрибуты, определённые в параметре attrsПлагин разрабатывался сторонней компанией, поэтому документацию можно почитать на официальной странице разработчиков. Данный плагин использует не много параметров, но почти все обязательные. Ниже представлены все параметры:

  • host - IP-адрес сервера, _обязательный параметр
  • password - пароль
  • dn - уникальное имя, идентифицирующее пользователя или группу, обязательный параметр
  • base - компоненты домена, обязательный параметр
  • filter - поле для фильтрации записей, обязательный параметр
  • attrs - список (массив) из атрибутов каждой записи, по умолчанию ['uid']
  • port - порт для подключения, по умолчанию "389"
  • usessl - использовать ли SSL, по умолчанию "false"
  • schedule - расписание запуска плагина с помощью rufus-scheduler, синтаксис похож на cron, но имеет гораздо больше возможностей
input {
  LDAPSearch {
    host => "<host_ip>"
    password => "${ldap_password}"
    dn => "dn"
    base => 'dc'
    attrs => []
    filter => "(objectClass=computer)"
    schedule => "14 * * * *"
  }
}

opensearch​

Позволяет выполнить переиндексирование данных для дополнительной обработки с последующей записью в другой индекс. OpenSearch является fork-ом для ElasticSearch, поэтому можно смотреть информацию в плагине elasticsearch на официальной странице документации.. Плагин не имеет обязательных параметров, часто используемые параметры представлены ниже:

  • hosts - список (массив) url адресов сервера OpenSearch. Каждый url может включать IP, HOST, IP:port, HOST:port
  • index - с какого индекса выполняется чтение. Важно указать конкретный индекс без использования index pattern
  • user - имя пользователя для подключения к OpenSearch
  • password - пароль пользователя для подключения к OpenSearch
  • query - выполняемый запрос для чтения. Подробнее можно прочитать в официально документации
  • ca_file - путь до файла - сертификата Центра Сертификации, включать должен все необходимые цепочки сертификатов
  • schedule - расписание запуска плагина с помощью rufus-scheduler, синтаксис похож на cron, но имеет гораздо больше возможностей
input {
  opensearch {
    hosts => ["localhost:9200"]
    index => "linux_deamon-*"
    user => "${opensearch_username}"
    password => "${opensearch_password}"
    query => '{ "size": 10, "query": { "match_all": {}} }'
    ca_file => "/path/to/cert"
    schedule => "* * * * *"
  }
}