Компоненты#

Компоненты представляют собой функциональные элементы процесса сбора метрик и входят в состав коллектора метрик.

При необходимости можно расширять существующую функциональность Коллектора путем разработки новых программных компонентов соответствующего типа. Все примеры конфигураций ниже описаны в формате HOCON.

Типы компонентов

Все компоненты делятся на следующие укрупненные типы:

входные input — задают источник для метрик;
выходные output — определяют приемник метрик (хранилище);
обработчики processor — производят обработку транзитных метрик;
агрегаторы aggregator — реализуют агрегацию потока метрик на основе окна обработки.

Каждый компонент можно настроить путем передачи ему конфигурационных параметров.

Общими являются следующие параметры:

name — имя экземпляра компонента (строка);
description — описание экземпляра компонента (строка);
type — тип компонента (строка, возможные варианты: input, output, processor, aggregator);
class — имя класса программной реализации компонента.

Input компоненты#

JmxInput — подключается к JMX-источнику и возвращает в виде метрик значения атрибутов заданного MBean-а;
KafkaAdminClientInput — подключается через AdminClient API к Kafka-кластеру и возвращает в виде метрик ряд параметров кластера;
HttpInput — подключается к серверу и возвращает в виде метрик ответы сервера;
InfluxDbV1Input — подключается к InfluxDb и возвращает в виде метрик результаты ответа на запрос к БД;
ConfigInput — генератор метрик на основе конфигурационного файла;
ListInput — «заглушка» источника.

JmxInput#

Компонент JmxInput — компонент-источник (input), подключается к JMX-источнику и возвращает в полях метрики значения атрибутов указанного Java Management Bean.

Пример конфигурации:

{
  name: SomeJmxInputComponent
  description: "SomeJmxInputComponent description"
  type: input
  class: JmxInput

  metricName: SomeMetricName
  emitRateMs: 30000
  host: IP_ADDRESS
  port: 7010
  user: <user>
  password: <password>
  objName: "kafka.log:type=Log,name=LogEndOffset,topic=*,partition=*"
  attrNames: [Value]
}

Требуемые значения общих параметров конфигурации:

type:  input
class: JmxInput

Обязательные параметры:

metricName — имя генерируемой метрики (строка);
emitRateMs — период опроса JMX-сервера (длинное целое);
host — адрес JMX-сервера (строка);
port — порт JMX-сервера (целое);
objName — значение MBean.ObjectName (строка);
attrNames — перечень имен запрашиваемых атрибутов (список строк).

Необязательные параметры:

user — имя пользователя для аутентификации на JMX-сервере, значение по умолчанию — user (строка);
password — пароль пользователя для аутентификации на JMX-сервере, значение по умолчанию — password (строка).

Особенности использования:

Строковое значение параметра конфигурации objName может содержать символ подстановки *, маскирующий конкретные значения JMX-свойств из строки MBean.ObjectName. В этом случае компонент будет возвращать набор метрик для всех возможных значений маскируемого JMX-свойства.

Компонент добавляет в словарь тегов выходной Метрики следующие пары (ключ, значение):

Ключ	Значение
`MBean.ObjectName`	строка `ObjectName` с подстановками
<имя_свойства_1>	значение JMX-свойства 1
…	…
<имя_свойства_N>	значение JMX-свойства N
<имя_конфиг_парам_1>	значение параметра конфигурации 1
…	…
<имя_конфиг_парам_M>	значение параметра конфигурации M

KafkaAdminClientInput#

Компонент KafkaAdminClientInput — компонент-источник (input), подключается к Kafka-кластеру через AdminClient API и формирует следующий набор метрик Kafka-кластера:

Список брокеров кластера *.BrokersList
Количество брокеров кластера *.BrokersCount
Загрузка DRIVE *.Broker(HOST).HddUsage
Вспомогательные метрики для расчета задержки консьюмер-групп: *.Broker.Topic(TOPIC).Partition(PARTITION).ConsumerGroup(GROUP).CurrentOffset

Пример конфигурации:

{
  name: SomeKafkaAdminClientInputComponent
  description: "SomeKafkaAdminClientComponent description"
  type: input
  class: KafkaAdminClientInput

  metricName: SomeMetricName
  metricSet: [
        {type: BrokersList, description: "Список брокеров кластера"}
        {type: BrokersCount, description: "Количество брокеров кластера"}
        {type: HddUsage, description: "Загрузка DRIVE"}
        {type: CurrentOffset, description: "Текущий offset (для рассчета лага)"}
      ]
  emitRateMs: 30000
  timeoutMs: 10000
  bootstrapServers: "IP_ADDRESS:9092,IP_ADDRESS:9092"
  securityProtocol: SSL
  sslKeyPassword: SomeKeyPassword
  sslKeystoreLocation: /path/to/file.jks
  sslKeystorePassword: SomeKeyStorePassword
  sslTruststoreLocation: /path/to/file.jks
  sslTruststorePassword: SomeTrustStorePassword
  sslEndpointIdentificationAlgorithm: SomeAlgorithm
}

Требуемые значения общих параметров конфигурации:

type:  input
class: KafkaAdminClientInput

Обязательные параметры:

metricName — наименование поля name генерируемой компонентом метрики (строка);
metricSet — множество типов генерируемых метрик с их текстовым описанием:
- type — имя типа (строка - одно из: BrokersList, BrokersCount, HddUsage, CurrentOffset),
- description — текст описания данного типа метрик (строка);
timeoutMs — тайм-аут ответов Kafka-кластера в миллисекундах (длинное целое);
bootstrapServers — список серверов Kafka-кластера (строка);
securityProtocol — протокол безопасности (строка);
sslKeyPassword — пароль Key (строка);
sslKeystoreLocation — путь к Keystore (строка);
sslKeystorePassword — пароль KeyStore (строка);
sslTruststoreLocation — путь к Truststore (строка);
sslTruststorePassword — пароль TrustStore (строка);
sslEndpointIdentificationAlgorithm — алгоритм аутентификации (строка).

Особенности использования:

Компонент добавляет в словарь тегов выходной Метрики следующие пары (ключ, значение):

Ключ	Значение
`name`	«KafkaAdminClient»
`title`	заголовок метрики
`cluster_id`	значение Kafka Cluster ID
`topic`	имя topic (для метрики CurrentOffset)
`partition`	номер партиции (для метрики CurrentOffset)
`consumerGroup`	имя консьюмер-группы (для метрики CurrentOffset)

HttpInput#

Компонент HttpInput — компонент-источник (input), подключается к HTTP-сервису и возвращает в полях метрики результаты ответа.

Пример конфигурации:

{
  name: SomeHttpInputComponent
  description: "HttpInputComponent's description"
  type: input
  class: HttpInput

  emitRateMs: 1000
  address: {
    host: IP_ADDRESS
    port: 8081
  }
  url: "/taskmanagers/metrics"
  tags: ["id", "uuid"]
  fields: [
    {
      fieldName: "min"
      fieldType: "double"
    },
    {
      fieldName: "max"
      fieldType: "int"
    }
  ]
}

Требуемые значения общих параметров конфигурации:

type:  input
class: HttpInput

Обязательные параметры:

emitRateMs — период опроса HTTP-сервиса (длинное целое);
address — адрес HTTP-сервиса:
- host— адрес сервера (строка);
- port — порт сервера (целое);
url — URL HTTP-ресурса (строка);
tags — перечень ключей тегов (список строк);
fields — перечень полей с их типами (список структур):
- fieldName — имя поля (строка);
- fieldType — тип поля (строка).

InfluxDbV1Input#

Компонент InfluxDbV1Input — компонент-источник (input), подключается к БД InfluxDb (v.1) и возвращает в полях метрики результаты запроса к БД.

Пример конфигурации:

{
  name: SomeInfluxDbV1InputComponent
  description: "SomeInfluxDbV1InputComponent description"
  type: input
  class: InfluxDbV1Input

  emitRateMs: 30000
  url: "https://IP_ADDRESS:8086"
  sslKeystoreLocation: /path/to/file.jks
  sslKeystorePassword: SomeKeyStorePassword
  sslTruststoreLocation: /path/to/file.jks
  sslTruststorePassword: SomeTrustStorePassword
  username: <User>
  password: <password>
  query: "SELECT \"Value\" FROM "SomeDatabaseName.SomeRetention."/Federation.Domain.Kafka.Cluster.*/  WHERE time > now() - 10m"
}

Требуемые значения общих параметров конфигурации:

type:  input
class: InfluxDbV1Input

Обязательные параметры:

emitRateMs — период опроса InfluxDb-сервера (длинное целое);
url — адрес InfluxDb-сервера (строка);
sslKeystoreLocation — путь к Keystore (строка);
sslKeystorePassword — пароль KeyStore (строка);
sslTruststoreLocation — путь к Truststore (строка);
sslTruststorePassword — пароль TrustStore (строка);
username — имя пользователя для аутентификации на InfluxDb-сервере (строка);
password — пароль пользователя для аутентификации на InfluxDb-сервере (строка).

Особенности использования:

Запрос может содержать символ подстановки *, маскирующий суффикс имени Метрики (Measurement) в БД Influx.

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

Output компоненты#

В состав Коллектора входят реализации следующих output компонентов:

InfluxDbOutput — подключается к InfluxDb (v.2) и отправляет в БД входной список метрик на хранение;
InfluxDbV1Output — подключается к InfluxDb (v.1) и отправляет в БД входной список метрик на хранение;
JmxOutput — JMX-приемник;
EventDiscoveryOutput — EventDiscovery-приемник;
PrintOutput — «заглушка» приемника;
KafkaOutput — подключается к Kafka topic и отправляет в него метрики;
ElasticsearchOutput — подключается к Elasticsearch и отправляет туда метрики.

InfluxDbOutput#

Компонент InfluxDbOutput — компонент-приемник (output), подключается к БД InfluxDb (v.2) и сохраняет набор полученных из потока Метрик.

Пример конфигурации:

{
  name: SomeInfluxDbOutputComponent
  description: "SomeInfluxDbOutputComponent description"
  type: output
  class: InfluxDbOutput
  detailedLoggingRateMs: 600000

  url: "http://IP_ADDRESS:8086"
  token = SomeToken
  org = SomeOrganization
  bucket = SomeBucket
}

Требуемые значения общих параметров конфигурации:

type:  output
class: InfluxDbOutput

Обязательные параметры:

url — адрес InfluxDb-сервера (строка);
token — токен для аутентификации на InfluxDb-сервере (строка);
org — название организации для аутентификации на InfluxDb-сервере (строка);
bucket — имя сегмента InfluxDb (строка).

Необязательные параметры:

detailedLoggingRateMs — период формирования детальных логов по выходным метрикам в мс (длинное целое).

Особенности использования:

Компонент выполняет подключение к БД Influx, используя аутентификацию на основе токенов. В качестве имени сегмента БД будет использовано значение параметра bucket. Из каждой Метрики входного набора, компонент сделает одну точку (Point) InfluxDb таким образом, что:

measurement = имя Метрики;
tag = теги Метрики;
fields = поля Метрики;
time = метка времени Метрики.

InfluxDbV1Output#

Компонент InfluxDbV1Output — компонент-приемник (output), подключается к БД InfluxDb (v.1) и сохраняет набор полученных из потока Метрик.

Пример конфигурации:

{
  name: SomeInfluxDbV1OutputComponent
  description: "SomeInfluxDbV1OutputComponent description"
  type: output
  class: InfluxDbV1Output
  detailedLoggingRateMs: 600000

  url: "https://IP_ADDRESS:8086"
  sslKeystoreLocation: /path/to/file.jks
  sslKeystorePassword: SomeKeyStorePassword
  sslTruststoreLocation: /path/to/file.jks
  sslTruststorePassword: SomeTrustStorePassword
  username: <User>
  password: <password>
  database: SomeDatabaseName
  retention: SomeRetention
}

Требуемые значения общих параметров конфигурации:

type:  output
class: InfluxDbV1Output

Обязательные параметры:

url — адрес InfluxDb-сервера (строка);
sslKeystoreLocation — путь к Keystore (строка);
sslKeystorePassword — пароль KeyStore (строка);
sslTruststoreLocation — путь к Truststore (строка);
sslTruststorePassword — пароль TrustStore (строка);
username — имя пользователя для аутентификации на InfluxDb-сервере (строка);
password — пароль пользователя для аутентификации на InfluxDb-сервере (строка);
database — имя БД InfluxDb (строка).

Необязательные параметры:

detailedLoggingRateMs — период формирования детальных логов по выходным метрикам в мс (длинное целое);
retention — имя retention policy (строка).

Особенности использования:

Компонент выполняет подключение к БД Influx, используя SSL-подключение и авторизацию по имени и паролю.

В качестве имени БД используется значение параметра database.
В качестве retention policy — значение параметра retention. Если параметр retention опущен, InfluxDb будет использовать текущее значение по умолчанию.

Из каждой Метрики входного набора компонент сделает одну точку (Point) InfluxDb таким образом, что:

measurement = имя Метрики;
tag = теги Метрики;
fields = поля Метрики;
time = метка времени Метрики.

JmxOutput#

Компонент JmxOutput — JMX-приемник.

EventDiscoveryOutput#

Компонент EventDiscoveryOutput — EventDiscovery-приемник.

PrintOutput#

Компонент PrintOutput — «заглушка» приемника.

KafkaOutput#

Компонент KafkaOutput — компонент-приемник (output), подключается к Kafka-кластеру и сохраняет набор полученных из потока Метрик в указанный topic.

Пример конфигурации:

{
  name: SomeKafkaOutputComponent
  description: "SomeKafkaOutputComponent description"
  type: output
  class: KafkaOutput
  detailedLoggingRateMs: 600000
  template: {
    pattern: "some string pattern ... " (see below)
    dateTimeFormat: "yyyy-MM-dd'T'HH:mm:ss.SSS"
  }

  bootstrapServers: "IP_ADDRESS:9092,IP_ADDRESS:9092"
  topic: kafkaTopic
  securityProtocol: SSL
  ssl {
    sslKeyPassword: SomeKeyPassword
    sslKeystoreLocation: /path/to/file.jks
    sslKeystorePassword: SomeKeyStorePassword
    sslTruststoreLocation: /path/to/file.jks
    sslTruststorePassword: SomeTrustStorePassword
    sslEndpointIdentificationAlgorithm: SomeSslEndpointIdentificationAlgorithm
  }
}

Требуемые значения общих параметров конфигурации:

type:  output
class: KafkaOutput

Обязательные параметры:

bootstrapServers — список серверов Kafka-кластера (строка);
topic — имя выходного топика для отправки метрик (строка);
securityProtocol — тип протокола безопасности для подключения к Kafka-кластеру (строка).

Необязательные параметры:

detailedLoggingRateMs — период формирования детальных логов по выходным метрикам в мс (длинное целое);
ssl — секция настроек для доступа к кластеру по SSL:
- sslKeyPassword — пароль Key (строка);
- sslKeystoreLocation — путь к Keystore (строка);
- sslKeystorePassword — пароль KeyStore (строка);
- sslTruststoreLocation — путь к Truststore (строка);
- sslTruststorePassword — пароль TrustStore (строка);
template — секция настроек для формирования сообщения в формате JSON на основе шаблона и отправки его в topic Kafka:
- pattern — строковое представление шаблона сообщения в формате JSON (строка);
- dateTimeFormat — формат даты-времени, например yyyy-MM-dd'T'HH:mm:ss.SSS (строка).

Особенности использования:

Компонент выполняет подключение к кластеру Kafka и осуществляет запись выходных метрик в заданный topic Kafka-кластера. Формирование JSON-сообщений на основе шаблона реализовано на основе JsonTemplate.

Пример шаблона сообщения: {timestamp: $time, metricName:$name, valueType:$valueType, unit:$tag_unit, velocity:$field_velocity, description: "Any text ..."}

, где $ — знак обозначения переменной, значение которой будет подставлено в шаблон.

Существует несколько зарезервированных переменных, которые можно использовать в шаблоне:

$time — время формирования метрики. Представляет собой строковое представление в формате даты-времени. Если формат дата-время не указан в конфигурации, то используется дефолтный формат — YYYY-MM-DD hh:mm:ss.ss
$name — имя метрики;
$valueType — тип метрики;
$tag_tagKey — значение тега метрики с ключом tagKey. Использование префикса tag_ обязательно;
$field_fieldName — значение поля метрики с ключом fieldName. Использование префикса field_ обязательно;
@props($tags) — представляет список всех тегов с их значениями. Использование @props обязательно;
@props($fields) — представляет список всех полей метрики с их значениями. Использование @props обязательно.

Пример_1 конфигурации для заполнения шаблона:

template: {
    pattern: "{timestamp: $time, metricName:$name, valueType:$valueType, unit:$tag_unit, velocity:$field_velocity, description: "Any text ..."}"
    dateTimeFormat: "yyyy-MM-dd'T'HH:mm:ss.SSS"
}

Результирующее JSON-сообщение:

{
    "timestamp":"2021-08-31T17:34:32.186",
    "metricName":"simple_metric_1",
    "valueType":"Counter",
    "unit":"m/s",
    "velocity":"125.6",
    "description":"Any text ..."
}

Пример конфигурации для заполнения шаблона:

template: {
    pattern: "{timestamp: $time, metricName:$name, valueType:$valueType, allTags:@prop($tags), allFields:@props($fields), description: "Any text ..."}"
    dateTimeFormat: "yyyy-MM-dd'T'HH:mm:ss.SSS"
}

Результирующее JSON-сообщение:

{
    "timestamp":"2021-08-31T17:34:32.186",
    "metricName":"simple_metric_1",
    "valueType":"Counter",
    "allTags":[
      {
        "key": "test_tag_1",
        "value": "tag_value_11"
      },
      {
        "key": "test_tag_2",
        "value": "tag_value_12"
      },
      {
        "key": "test_tag_3",
        "value": "tag_value_13"
      }
    ],
    "allFields":[
      {
        "key": "test_field_1",
        "value": "265"
      },
      {
        "key": "test_field_2",
        "value": "343.564"
      },
      {
        "key": "test_field_3",
        "value": "asdf"
      }
    ],
    "description":"Any text ..."
}

Все значения в результирующем JSON-сообщения представляют собой строковые значения.

ElasticsearchOutput#

Компонент ElasticsearchOutput — компонент-приемник (output), подключается к серверу Elasticsearch и сохраняет набор полученных из потока Метрик в поисковый индекс Elasticsearch-сервера.

Пример конфигурации:

{
  name: SomeKafkaOutputComponent
  description: "SomeElasticsearchOutputComponent description"
  type: output
  class: ElasticsearchOutput
  detailedLoggingRateMs: 600000

  host: localhost
  port: 9200
  ssl {
    sslKeystoreLocation: /path/to/file.jks
    sslKeystorePassword: SomeKeyStorePassword
    sslTruststoreLocation: /path/to/file.jks
    sslTruststorePassword: SomeTrustStorePassword
  }
}

Требуемые значения общих параметров конфигурации:

type:  output
class: ElasticsearchOutput

Обязательные параметры:

host — адрес сервера Elasticsearch (строка);
port — порт сервера Elasticsearch (целое).

Необязательные параметры:

detailedLoggingRateMs — период формирования детальных логов по выходным метрикам в мс (длинное целое);
ssl — секция настроек для доступа к кластеру по SSL:
sslKeystoreLocation — путь к Keystore (строка);
sslKeystorePassword — пароль KeyStore (строка);
sslTruststoreLocation — путь к Truststore (строка);
sslTruststorePassword — пароль TrustStore (строка).

Особенности использования:

Компонент выполняет подключение к серверу Elasticsearch и осуществляет запись выходных метрик в Elasticsearch индекс.

Processor компоненты#

В состав Коллектора входят реализации следующих processor компонентов:

обработка на уровне отдельных полей метрики (field):
- FieldCompositeDataProcessor — преобразует поля метрики с типом CompositeData в набор полей с примитивными типами (committed, init, max, used);
- FieldRenameProcessor — переименовывает заданные поля метрики;
- FieldNameTypeSuffixProcessor — добавляет к именам полей суффикс с именем типа текущего значения данного поля;
- FieldFromTagTransferProcessor — осуществляет перенос значений тегов, указанных в конфигурации, в поля метрики.
обработка на уровне метрики целиком (metric):
- MetricAggregationProcessor — агрегирует входной список метрик с группировкой по имени метрики (объединяет теги, суммирует значения указанных полей);
- MetricKafkaLagProcessor — агрегирует входной список метрик из двух источников (JmxInput + KafkaAdminClientInput);
- MetricSplitByFieldsProcessor — тиражирует (сплитит) метрику на несколько отдельных метрик по количеству полей, добавляет к имени выходной метрики имя соответствующего поля;
- MetricFilterProcessor — фильтрует метрики по образцу;
- MetricTransformerProcessor — процессор может преобразовывать любые части метрики.
обработка на уровне имени метрики (name):
- NameSubsByTagsProcessor — заполняет шаблон имени метрики значениями тегов;
- NameSuffixProcessor — добавляет к имени метрики суффикс;
- NamePrefixProcessor — добавляет к имени метрики префикс.
обработка на уровне тегов метрики (tag):
- TagPrefixProcessor — добавляет к именам тегов заданный префикс;
- TagRemoveProcessor — выборочно удаляет теги из метрики;
- TagSuffixProcessor — добавляет к именам тегов заданный суффикс;
- TagTransformProcessor — преобразует значения заданных тегов по регулярному выражению.

Пример типовой конфигурации процессора:

  {
    name: SomeProcessorComponent
    description: "SomeProcessorComponent description"
    type: processor
    class: ClassOfProcessor
    includeOnly: {
      metrics: [
        {name: nameTemplate1, valueType: typeTemplate1}
        {name: nameTemplate2, valueType: typeTemplate2}
      ]
    }
    excludeOnly: {
      metrics: [
        {name: nameTemplate1, valueType: typeTemplate1}
        {name: nameTemplate2, valueType: typeTemplate2}
      ]
    }

  }

Требуемые значения общих параметров конфигурации:

  type:  processor
  class: FieldCompositeDataProcessor

Необязательные параметры:

includeOnly — список шаблонов соответствия для метрик подвергаемых обработке в процессоре (список);
excludeOnly — список шаблонов соответствия для метрик исключаемых из обработки в процессоре (список).

FieldCompositeDataProcessor#

Компонент FieldCompositeDataProcessor — компонент-обработчик (processor), обработка на уровне отдельных полей метрики (field). Преобразует поля метрики с типом CompositeData в набор полей с примитивными типами (committed, init, max, used).

Пример конфигурации:

{
  name: SomeFieldCompositeDataProcessorComponent
  description: "SomeFieldCompositeDataProcessorComponent description"
  type: processor
  class: FieldCompositeDataProcessor
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: FieldCompositeDataProcessor

Особенности использования:

Компонент преобразует каждое поле Метрики, имеющее композитный тип CompositeData, в набор из четырех отдельных полей:

Имя старого поля	Имя нового поля	Значение
<имя_поля>	<имя_поля>Сommitted	`CompositeData.committed`
- // -	<имя_поля>Init	`CompositeData.init`
- // -	<имя_поля>Max	`CompositeData.max`
- // -	<имя_поля>Used	`CompositeData.used`

FieldRenameProcessor#

Компонент FieldRenameProcessor — компонент-обработчик (processor), обработка на уровне отдельных полей метрики (field).

Переименовывает указанные в конфигурации компонента поля метрики.

Пример конфигурации:

{
  name: SomeFieldRenameProcessorComponent
  description: "SomeFieldRenameProcessorComponent description"
  type: processor
  class: FieldRenameProcessor

  namesMap: [ { oldName: SomeOldName, newName: SomeNewName } ]
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: FieldRenameProcessor

Обязательные параметры:

namesMap — словарь соответствия имен полей (список структур):
- oldName — текущее имя переименовываемого поля (строка);
- newName — новое имя для переименовываемого поля (строка).

MetricAggregationProcessor#

Компонент MetricAggregationProcessor — компонент-обработчик (processor), обработка на уровне метрики целиком (metric). Агрегирует входной список метрик с группировкой по имени метрики.

Пример конфигурации:

{
  name: SomeMetricAggregationProcessorComponent
  description: "SomeMetricAggregationProcessorComponent description"
  type: processor
  class: MetricAggregationProcessor

  fields: [Value]
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: MetricAggregationProcessor

Обязательные параметры:

fields — перечень имен полей (список строк).

Особенности использования:

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

MetricKafkaLagProcessor#

Компонент MetricKafkaLagProcessor — компонент-обработчик (processor), обработка на уровне метрики целиком (metric). Агрегирует входной список метрик из двух источников (JmxInput + KafkaAdminClientInput).

Пример конфигурации:

{
  name: SomeMetricKafkaLagProcessorComponent
  description: "SomeMetricKafkaLagProcessorComponent description"
  type: processor
  class: MetricKafkaLagProcessor

  fieldName: SomeField
  metricName: SomeMetricName
  metricDescription: "SomeMetric description"
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: MetricKafkaLagProcessor

Необязательные параметры:

fieldName — имя поля Метрики для получения входного значения (строка);
metricName — суффикс имени выходной Метрики (префиксом будет имя входной метрики) (строка);
metricDescription — описание выходной Метрики (строка).

Особенности использования: Работает в потоке с двумя Источниками (JmxInput + KafkaAdminClientInput).

Из входного набора Метрик отбирает метрики *.EndOffset (JmxInput) и *.CurrentOffset (KafkaAdminClientInput). Остальные метрики проходят через процессор транзитом без изменений. На основе этих метрик производит расчет задержки консьюмер-групп по топикам Kafka.

Используется следующий подход:

Приближенно Lag = (endOffset - currentOffset);
currentOffset можно по консьюмер-группам взять через Kafka Admin Client;
endOffset (другими словами last produced offset) можно по топикам и партициям взять через JMX-метрику LogEndOffset, после все собирается агрегатором и высчитывается разница.

Компонент требует наличия у Метрики следующих тегов:

topic;
partition;
consumerGroup.

MetricSplitByFieldsProcessor#

Компонент MetricSplitByFieldsProcessor — компонент-обработчик (processor), обработка на уровне метрики целиком (metric).

Тиражирует (сплитит) метрику на несколько отдельных метрик по количеству полей.

Пример конфигурации:

{
  name: SomeMetricSplitByFieldsProcessorComponent
  description: "SomeMetricSplitByFieldsProcessorComponent description"
  type: processor
  class: MetricSplitByFieldsProcessor
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: MetricSplitByFieldsProcessor

Особенности использования:

Из каждой Метрики с несколькими полями компонент делает набор Метрик с единственным полем.

При этом к имени новой Метрики имя поля добавляется как суффикс, например:

было:

Имя метрики	Поля метрики
SomeName	[Field1, Field2, Field3]

станет:

Имя метрики	Поля метрики
SomeNameField1	[Field1]
SomeNameField2	[Field2]
SomeNameField3	[Field3]

NameSubsByTagsProcessor компонент#

Компонент Обработчик (processor).

Обработка на уровне имени метрики (name).

Заполняет шаблон имени метрики значениями тегов.

Пример конфигурации:

{
  name: SomeNameSubsByTagsProcessorComponent
  description: "SomeNameSubsByTagsProcessorComponent description"
  type: processor
  class: NameSubsByTagsProcessor
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: NameSubsByTagsProcessor

Особенности использования:

Предполагается, что вместо имени Метрики может использоваться шаблон следующего вида:

Some.MetricName.Template.Host(<host>).Topic_<topic>.Partition<partition>

Здесь в тексте имени содержатся подстановочные переменные, заключенные в угловые скобки:

<host>
<topic>
<partition>

Количество и место расположение подобных подстановочных переменных может быть произвольным (недопустима только их вложенность).

Компонент производит подстановку вместо переменных шаблона в скобках значений тегов Метрики, имена которых совпадают с именами переменных.

Например, если Метрика содержит следующие теги:

Имя тега	Значение тега
host	IP_ADDRESS
topic	SomeTopic
partition	5

Результатом заполнения шаблона имени Метрики из предыдущего примера станет следующая строка:

Some.MetricName.Template.Host(IP_ADDRESS).Topic_SomeTopic.Partition5

TagPrefixProcessor компонент#

Добавляет к имени метрики суффикс.

TagPrefixProcessor компонент#

Добавляет к именам тегов заданный префикс.

TagRemoveProcessor компонент#

Компонент Обработчик (processor).

Обработка на уровне тегов метрики (tag):

Выборочно удаляет теги метрики.

Пример конфигурации:

{
  name: SomeTagRemoveProcessorComponent
  description: "SomeTagRemoveProcessorComponent description"
  type: processor
  class: TagRemoveProcessor

  nonRemovableTags: [ ]
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: TagRemoveProcessor

Необязательные параметры:

nonRemovableTags — перечень имен тегов, запрещенных к удалению (список строк).

Особенности использования:

Из состава тегов Метрики удаляются все теги, имена которых не содержатся в списке параметра nonRemovableTags.

Если данный список пуст или параметр nonRemovableTags не задан в конфигурации, будут удалены все теги Метрики!

TagSuffixProcessor компонент#

Добавляет к именам тегов заданный суффикс.

TagTransformProcessor компонент#

Компонент Обработчик (processor).

Обработка на уровне тегов метрики (tag):

Преобразует значения заданных тегов по регулярному выражению.

Пример конфигурации:

{
  name: SomeTagTransformProcessorComponent
  description: "SomeTagTransformProcessorComponent description"
  type: processor
  class: TagTransformProcessor

  regexPattern: "connector-(?:producer|consumer)-(.*-connector?).*"
  tags: ["client-id"]
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: TagTransformProcessor

Обязательные параметры:

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

Особенности использования:

Над значениями тегов, перечисленными в параметре tags, выполняется преобразование, заданное регулярным выражением regexPattern.

В тексте регулярного выражения должна быть единственная извлекаемая группа, ее результат и принимается за новое значение трансформируемого тега!

Так, в выражении из примера группа (?:producer|consumer) помечена как не извлекаемая и она не учитывается, а результат трансформации формируется группой (.*-connector?).

FieldFromTagTransferProcessor компонент#

Компонент Обработчик (processor).

Обработка на уровне полей метрики (field).

Осуществляет перенос значений тегов, указанных в конфигурации, в поля метрики.

Пример конфигурации:

{
  name: FieldFromTagTransfer
  description: "transfer tag value to metric field"
  type: processor
  class: FieldFromTagTransferProcessor

  transfer:
     [
       { tag: tag_name_1, fieldName: fieldName_1 },
       { tag: tag_name_2, fieldName: fieldName_2 }
     ]
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: FieldFromTagTransferProcessor

Обязательные параметры:

transfer — перечень тегов и имен полей метрики между которыми необходим перенос значений;
transfer.tag — имя тега метрики откуда необходимо перенести значение;
transfer.fieldName — имя поля метрики куда необходимо перенести значение тега.

Особенности использования:

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

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

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

MetricTransformerProcessor компонент#

Компонент Обработчик (processor).

Обработка на уровне метрики целиком (metric).

Процессор может преобразовывать:

имя метрики — name;
тип метрики — valueType;
время формирования метрики — time;
значения заданных тегов и/или формирование новых тегов;
значения заданных полей метрики и/или формирование новых.

Пример конфигурации:

{
  name: MetricTransformer
  description: "transform metrics"
  type: processor
  class: MetricTransformerProcessor

  transformer: {
    metrics: [
      { template: {name: metric_1, valueType: Gauge}, transformValues: { name: new_metric_name, valueType: Histogram, time: 1631024403, tags: [{key: tagName, value: newTagValue}], fields: [{key: fieldName, value: newFieldValue}]}}
    ]
  }
}

Требуемые значения общих параметров конфигурации:

type:  processor
class: MetricTransformerProcessor

Обязательные параметры:

transformer.metrics — перечень метрик, в которых необходимо провести трансформацию, и что именно необходимо изменить/добавить. Каждый элемент перечня metrics состоит из двух полей — template и transformValues;
template — набор параметров, по которым идентифицируется метрика, в которой необходимо провести трансформацию. К этим параметрам относятся - name, valueType, time, tags, fields;
transformValues — набор новых значений, которые необходимо изменить/добавить в метрике.

В параметрах template и transformValues могут присутствовать следующие параметры — name, valueType, time, tags, fields.

name — имя метрики;
valueType — тип метрики;
time — время формирования метрики;
tags — теги метрики;
fields — поля метрики.

Параметры tags и field представляют собой список ключ-значение — [{key: tagName(or fieldName), value: newValue}, {key: ..., value: ..., ...}] , где:

key — имя тега или поля метрики, которое необходимо изменить;
value — новое значение. В поле value можно использовать математические выражения.

Математические выражения, преобразование типов и использование имен тегов/полей метрик в поле value

Для явного указания на математическое выражение необходимо использовать синтаксическую конструкцию — mathExpr(...) Для корректного вычисления выражение должно быть записано в двойных кавычках — {key: fieldName2, value: "mathExpr((1 + 3)*2 + 5)"}

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

tag_ — для обозначения имени тега. Пример: tag_tagName, где tagName — имя тега в метрике.
field_ — для обозначения имени поля метрики. Пример: field_fieldName, где fieldName — имя поля метрики.

Пример выражений с использованием имен тегов/полей метрики:

{key: fieldName3, value: "mathExpr(field_fieldName1 + field_fieldName2)*2 + 5)"}` или `{key: tagName4, value: "mathExpr((tag_tagName - 34 )* 1000)"}

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

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

toInt — преобразование к типу Int — {key: fieldName2, value: "mathExpr(toInt((1 + 3.2)*2 + 5))"}
toLong — преобразование к типу Long —{key: fieldName2, value: "mathExpr(toLong((1 + 3.2)*2 + 5))"}
toString — преобразование к типу String —{key: fieldName2, value: "mathExpr(toString((1 + 3.2)*2 + 5))"} Функции преобразования можно использовать только для преобразования вычисленного значения всего математического выражения. Применять функции преобразования типов к конкретным величинам мат. выражения нельзя.

Также возможно использование имен тегов/полей метрик в самом поле value, используя при этом соответствующие префиксы.

{key: fieldName, value: tag_tagName} — в поле с именем fieldName будет установлено значение тега с именем tagName
{key: tagName, value: field_fieldName} — в поле с именем tagName будет установлено значение поля метрики с именем fieldName

При переносе значений поля метрики в значение тега значение всегда будет преобразовываться к типу String.

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

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

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

Aggregator компоненты#

В состав Коллектора входят реализации следующих aggregator компонентов:

BasicAggregator — выполняет агрегацию входного потока метрик на основе временного окна.

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

Sum — сумма;
Count — количество значений;
Average — среднее значение;
Max — максимум;
Min — минимум;
Delta — разница между max и min значениями.