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

Есть несколько способов сделать это, но ни один из них не является слишком простым и потребует большого количества тестов, чтобы убедиться, что они жизнеспособны. Все они связаны с письмом collectdлокально, а затем с помощью какого-либо метода получить его на центральном сервере.

Я не тестировал ни одно из перечисленных ниже, поэтому некоторые могут вообще не работать.

Без особого порядка легкости или сложности:

1. Вывод сокета / сети в скрипт
   Напишите Collectdвывод в сокет или IP / порт, где сценарий PHP / Perl / Python / Bash прослушивает запись команд в файл.

   Затем эти файлы могут быть отправлены / извлечены центральным сервером и приняты Logstash.

   Плюсы: Простой скрипт для захвата вывода; стандартные команды Linux используются
   Минусы: Не масштабируется, если вы собираете много статистики; нужно поддерживать скрипт; не уверен, будет ли LS обрабатывать простой протокол

2. Redis / AMQP / Kafka / MongoDB Напишите Collectdвывод в один из возможных «буферов». Каждый из них работает немного по-своему и имеет разные варианты развертывания, поэтому я оставлю вам выяснить, что лучше, поскольку это выходит за рамки этого вопроса. Тем не менее, любой из них должен работать.

   Затем вам понадобится способ получить данные из буферного раствора обратно на центральный сервер. Собственная репликация / зеркалирование / кластеризация приложения или сценарий, который запускается каждый интервал X для отправки данных (запускается на любом конце) - это две возможности.

   Плюсы: Очень гибкие варианты развертывания; должен очень хорошо масштабироваться; использует хорошо известные инструменты / программы
   Минусы: Программе Buffer может потребоваться много ресурсов или установлено много пакетов

3. Выход сокета / сети в Logstash Это почти то же самое, что и вариант 1, но вместо collectd вывод в скрипт / программу, вы должны записать его в локальный экземпляр Logstash на каждом удаленном хосте.

   Затем Logstash будет записывать в CSV / JSON локально, и вы можете использовать любые средства для возврата этих файлов на центральный сервер, включая сам LS.

   Плюсы: Единый набор инструментов для всего решения; предоставляет способ преобразования данных на границе, а затем просто централизованного приема; очень мало движущихся частей Минусы: нужен Java / LS на всех удаленных хостах

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

редактировать: Achtung! Предупреждение!

Пожалуйста, используйте это docker-compose вместо того, который я связал (для этого требуется docker и compose и возможно machine но он сделал для вас больше, и вам придется меньше бороться.

CELK: https://github.com/codenamekt/celk-docker-compose/blob/master/logstash/logstash.conf

Также

Итак, начните здесь, чтобы получить хороший обзор работающей системы. Они сделали часть работы за вас, поэтому вам нужно беспокоиться только о своем вопросе, который относится к настройке и развертыванию.

Даже если вы не перестанете использовать Docker, это все равно приведет вас к успеху. и имеют дополнительное преимущество, показывая вам, как они сочетаются друг с другом.

Сначала получите Vagrant и создайте образ Vagrant с vagrant up

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

• (используйте ссылку выше на CELK): https://github.com/pblittle/docker-logstash

Чтобы использовать это, вам необходимо установить Vagrant. Просто сделай это! Тогда вам не нужно устанавливать docker потому что он будет работать на Vagrant VM.

У вас есть четыре варианта использования, но сначала подготовьте Vagrant с командой, выделенной жирным шрифтом ....

vagrant up

Решите, какие программы вам нужно запустить

Ваши варианты:

• Полный люкс или ELK (elasticsearch, logstash, kibana)
• Только агент (сборщик Logstash)
• Только Кибана

Доступны и другие конфигурации, но только для тестирования.

Время для шоу

Пришло время настроить Logstash, который на самом деле является единственной частью со сложным поведением.

Файлы конфигурации Logstash - это простые текстовые файлы, заканчивающиеся на conf и при желании могут быть сгруппированы вместе с помощью tar или gunzipgz.

Вы можете получить файлы конфигурации одним из двух способов:

• вы загружаете их из Интернета, используя переменную окружения LOGSTASH_CONFIG_URL чтобы указать на URL-адрес вашей конфигурации, и ** если вы ошиблись с URL-адресом или возникла проблема, и он не может получить конфигурацию с URL-адреса, он возвращается к известному URL-адресу, иначе
• читать их с диска, вроде - поскольку это докер, вы фактически будете создавать объем один раз (сейчас), и вы будете монтировать том при каждом запуске контейнера.

Вот как это выглядит при запуске с использованием конфигурации из Интернета:

$ docker run -d \
  -e LOGSTASH_CONFIG_URL=https://secretlogstashesstash.com/myconfig.tar.gz \
  -p 9292:9292 \
  -p 9200:9200 \
  pblittle/docker-logstash

Автор docker предупреждает вас:

По умолчанию logstash.conf прослушивает только входные данные stdin и file. Если вы хотите настроить ввод tcp и / или udp, используйте свои собственные файлы конфигурации logstash и самостоятельно открывайте порты. См. Документацию logstash для синтаксиса конфигурации и дополнительной информации.

Примечание. Что такое конфигурация logstash по умолчанию?

Напомним, это файл, который вы получаете, если не указали правильный URL-адрес для требуемой переменной среды. LOGSTASH_CONFIG_URL

Это раздел ввода:

// As the author warned, this is all you get. StdIn and Syslog file inputs.

input {
  stdin {
    type => "stdin-type"
  }

  file {
    type => "syslog"
    path => [ "/var/log/*.log", "/var/log/messages", "/var/log/syslog" ]
  }

  file {
    type => "logstash"
    path => [ "/var/log/logstash/logstash.log" ]
    start_position => "beginning"
  }
}

По умолчанию

Узнать больше о logstash на вебсайте.

Сейчас logstash есть плагины, которые помещают данные в input. Плагины различаются в точности так, как вы ожидаете; вот несколько:

• s3 от amazon (события файловой системы)
• stdin из logstash (по умолчанию читается stdin буфер)
• http из logstash (ваше предположение)
• ...и т.д...

Пример: сокеты UDP

UDP это быстрый протокол без установления соединения, который работает в нижней части L4 (транспорт) и поддерживает мультиплексирование, обрабатывает сбои и обычно является хорошим выбором для регистрации передачи данных.

Вы выбираете порт, который хотите; другие варианты зависят от того, что вы делаете.

TCP работает так же.

udp {порт => 9999 кодек => json buffer_size => 1452}

Пример 2: сокеты UDP из collectd фильтруется и выводится

This is stolen from https://github.com/codenamekt/celk-docker-compose/blob/master/logstash/logstash.conf

input {
  udp {
    port => 25826         # 25826 matches port specified in collectd.conf
    buffer_size => 1452   # 1452 is the default buffer size for Collectd
    codec => collectd { } # specific Collectd codec to invoke
    type => collectd
  }
}
output {
  elasticsearch {
    host => elasticsearch
    cluster  => logstash
    protocol => http
  }
}

И фильтрация - отличный пример: То есть он действительно длинный, и я думаю, что он

filter {
  # TEST implementation of parse for collectd
  if [type] == "collectd" {
    if [plugin] {
      mutate {
        rename => { "plugin" => "collectd_plugin" }
      }
    }
    if [plugin_instance] {
      mutate {
        rename => { "plugin_instance" => "collectd_plugin_instance" }
      }
    }
    if [type_instance] {
      mutate {
        rename => { "type_instance" => "collectd_type_instance" }
      }
    }
    if [value] {
      mutate {
        rename => { "value" => "collectd_value" }
      }
      mutate {
        convert => { "collectd_value" => "float" }
      }
    }
    if [collectd_plugin] == "interface" {
      mutate {
        add_field => {
          "collectd_value_instance" => "rx"
          "collectd_value" => "%{rx}"
        }
      }
      mutate {
        convert => {
          "tx" => "float"
          "collectd_value" => "float"
        }
      }
      # force clone for kibana3
      clone {
        clones => [ "tx" ]
      }
      ##### BUG EXISTS : AFTER clone 'if [type] == "foo"' NOT WORKING : ruby code is working #####
      ruby {
        code => "
          if event['type'] == 'tx'
            event['collectd_value_instance'] = 'tx'
            event['collectd_value'] = event['tx']
          end
        "
      }
      mutate {
        replace => { "_type" => "collectd" }
        replace => { "type" => "collectd" }
        remove_field => [ "rx", "tx" ]
      }
    }
    if [collectd_plugin] == "disk" {
      mutate {
        add_field => {
          "collectd_value_instance" => "read"
          "collectd_value" => "%{read}"
        }
      }
      mutate {
        convert => {
          "write" => "float"
          "collectd_value" => "float"
        }
      }
      # force clone for kibana3
      clone {
        clones => [ "write" ]
      }
      ##### BUG EXISTS : AFTER clone 'if [type] == "foo"' NOT WORKING : ruby code is working #####
      ruby {
        code => "
          if event['type'] == 'write'
             event['collectd_value_instance'] = 'write'
             event['collectd_value'] = event['write']
          end
        "
      }
      mutate {
        replace => { "_type" => "collectd" }
        replace => { "type" => "collectd" }
        remove_field => [ "read", "write" ]
      }
    }
    if [collectd_plugin] == "df" {
      mutate {
        add_field => {
          "collectd_value_instance" => "free"
          "collectd_value" => "%{free}"
        }
      }
      mutate {
        convert => {
          "used" => "float"
          "collectd_value" => "float"
        }
      }
      # force clone for kibana3
      clone {
        clones => [ "used" ]
      }
      ##### BUG EXISTS : AFTER clone 'if [type] == "foo"' NOT WORKING : ruby code is working  #####
      ruby {
        code => "
          if event['type'] == 'used'
            event['collectd_value_instance'] = 'used'
            event['collectd_value'] = event['used']
          end
        "
      }
      mutate {
        replace => { "_type" => "collectd" }
        replace => { "type" => "collectd" }
        remove_field => [ "used", "free" ]
      }
    }
    if [collectd_plugin] == "load" {
      mutate {
        add_field => {
          "collectd_value_instance" => "shortterm"
          "collectd_value" => "%{shortterm}"
        }
      }
      mutate {
        convert => {
          "longterm" => "float"
          "midterm" => "float"
          "collectd_value" => "float"
        }
      }
      # force clone for kibana3
      clone {
        clones => [ "longterm", "midterm" ]
      }
      ##### BUG EXISTS : AFTER clone 'if [type] == "foo"' NOT WORKING : ruby code is working #####
      ruby {
        code => "
          if event['type'] != 'collectd'
            event['collectd_value_instance'] = event['type']
            event['collectd_value'] = event[event['type']]
          end
        "
      }
      mutate {
        replace => { "_type" => "collectd" }
        replace => { "type" => "collectd" }
        remove_field => [ "longterm", "midterm", "shortterm" ]
      }
    }
  }
}

редактировать 3: Я, вероятно, не должен делать вашу работу за вас, но это нормально.

collectd как любое хорошее программное обеспечение КАПСУЛЬТЫ определенные аспекты, которые уродливы или трудны для пользователей, и пытается упростить вам задачу, поскольку это выглядит так, как будто вы отправляете данные (в данном случае кортеж) вместо того, чтобы обмануть сериализацию.

Ваш пример:

(date_time, current_cpu_load), for example ('2016-0-04-24 11:09:12', 12.3)

Я не собираюсь тратить свое время на выяснение того, как вы это формируете. Если вы можете получить эти данные с помощью плагина CPU, отлично. Я собираюсь скопировать и вставить тот, который нашел в Интернете, чтобы облегчить мне задачу.

Тем не менее, подумайте об этом ... немного, это не повредит.

Вы видите, что плагин CPU загружен ниже.

Вы видите интерфейс для collectd в conf файл слишком мал для указания полей.

Итак, если вы просто сделаете это, это сработает, но вы получите намного намного больше данные, чем просто загрузка процессора.

Вот где вы можете использовать фильтр. Но я думаю, это можно сделать и в Кибане. Так что я бы предпочел не тратить время на написание фильтра, который вам а) не нужен и б) можно легко написать, если вы потратите некоторое время.

## In `collectd`:
# For each instance where collectd is running, we define 
# hostname proper to that instance. When metrics from
# multiple instances are aggregated, hostname will tell 
# us were they came from.
Hostname "**YOUR_HOSTNAME**"


    # Fully qualified domain name, false for our little lab
    FQDNLookup false

    # Plugins we are going to use with their configurations,
    # if needed
    LoadPlugin cpu

    LoadPlugin df
    <Plugin df>
            Device "/dev/sda1"
            MountPoint "/"
            FSType "ext4"
            ReportReserved "true"
    </Plugin>

    LoadPlugin interface
    <Plugin interface>
            Interface "eth0"
            IgnoreSelected false
    </Plugin>

    LoadPlugin network
    <Plugin network>
            Server "**YOUR.HOST.IP.ADDR**" "**PORTNUMBER**"
    </Plugin>

    LoadPlugin memory

    LoadPlugin syslog
    <Plugin syslog>
            LogLevel info
    </Plugin>

    LoadPlugin swap

    <Include "/etc/collectd/collectd.conf.d">
            Filter ".conf"
    </Include>

Ваша конфигурация logstash

    input {
      udp {
        port => **PORTNUMBER**         # 25826 matches port specified in collectd.conf
        buffer_size => **1452**   **# 1452 is the default buffer size for Collectd**
        codec => collectd { } # specific Collectd codec to invoke
        type => collectd 
      }
    }
    output {
      elasticsearch {
        cluster  => **ELASTICSEARCH_CLUSTER_NAME** # this matches out elasticsearch cluster.name
        protocol => http
      }
    }

обновление от Аарона на collectd

В Logstash 1.3.x мы представили плагин ввода collectd. Было здорово! Мы могли обрабатывать метрики в Logstash, хранить их в Elasticsearch и просматривать их с помощью Kibana. Единственным недостатком было то, что через плагин можно было получать только около 3100 событий в секунду. В Logstash 1.4.0 мы представили недавно обновленный плагин ввода UDP, который был многопоточным и имел очередь. Я переделал подключаемый модуль ввода collectd в кодек (с некоторой помощью моих коллег и сообщества), чтобы воспользоваться этим огромным увеличением производительности. Теперь, имея всего 3 потока на моем двухъядерном Macbook Air, я могу получать более 45 000 событий в секунду через кодек collectd!

Итак, я хотел предоставить несколько быстрых примеров, которые вы могли бы использовать для изменения конфигурации вашего плагина, чтобы вместо этого использовать кодек.

По старому: input { collectd {} } Новый способ:

input {   udp {
port => 25826         # Must be specified. 25826 is the default for collectd
buffer_size => 1452   # Should be specified. 1452 is the default for recent versions of collectd
codec => collectd { } # This will invoke the default options for the codec
type => "collectd"   } } This new configuration will use 2 threads and a queue size of 2000 by default for the UDP input plugin. With
this you should easily be able to break 30,000 events per second!

Я привел некоторые другие примеры конфигурации. Для получения дополнительной информации, пожалуйста, ознакомьтесь с документацией Logstash для кодека collectd.

Удачного логирования!