Описание

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

Для сбора данных в реальном времени в больших объемах хорошо подходят базы данных временных рядов. На основе одной из таких БД c открытым исходным кодом ClickHouse, в сборке 5.2.774 добавлена система удаленного сбора логов RELS (Remote Event Logging System).

Архитектура

Каждый WCS сервер отправляет данные в ClickHouse независимо, используя JDBC-драйвер и HTTP-соединение. Для оптимизации работы с БД ClickHouse, данные буферизуются и отправляются пачками по времени

Описание таблиц данных

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

Данные соединений (таблица ConnectionEvent)

Данные о событиях потоков (таблица StreamEvent)

Данные CDN (таблица CDNEvent)

Данные о метриках потоков (таблица MediaSessionEvent)

В сборке 5.2.1896 добавлен сбор данных о метриках определенных потоков

Данные о нарезке HLS потока (таблица HlsStreamEvent)

В сборке 5.2.1917 добавлен сбор данных о событиях определенных HLS потоков

Данные о сегментах HLS потока (таблица HlsSegmenterEvent)

В сборке 5.2.1917 добавлен сбор данных о сегментах определенных HLS потоков

Данные о подписчиках HLS потока (таблица HlsClientEvent)

В сборке 5.2.1929 добавлен сбор данных о подписчиках определенных HLS потоков.  В сборке 5.2.2041добавлены дополнительные поля

Данные о метриках микшера (таблица MixerEvent)

В сборке 5.2.1923 добавлен сбор данных о метриках определенных микшеров

Данные о восстановлении потерянных аудио пакетов (таблица AudioRecoveryEvent)

В сборке 5.2.1969 добавлен сбор данных о восстановлении потерянных аудио пакетов

Данные о метриках буфера для входящих RTMP  потоков (таблица RtmpInBufferEvent)

В сборке 5.2.1978 добавлен сбор данных о метриках буфера входящих RTMP  потоков

Данные об отправленных REST хуках (таблица RestHooksEvent)

В сборке 5.2.2005 добавлен сбор данных об отправленных REST хуках

Данные о событиях, связанных с изменением RTP синхронизации аудио и видео (таблица RtpSynchronizationEvent)

В сборке 5.2.2017 добавлен сбор данных о событиях, связанных с изменением RTP синхронизации аудио и видео в опубликованном потоке

Данные о потерях пакетов, отправленных зрителю (таблица PlayerPacketEvent)

В сборке 5.2.2024 добавлен сбор данных о потерях пакетов, отправленных зрителю

Настройка

Установка и настройка ClickHouse

Требования к серверу

• CPU не менее 4 физических ядер, частотой не менее 3 ГГц, например Intel(R) Xeon(R) CPU E3-1246 v3 @ 3.50GHz
• RAM не менее 32 Гб
• HDD не менее 2 Тб

Установка ClickHouse из rpm пакетов (CentOS, Red Hat etc)

1. Подключите официальный репозиторий

sudo yum install -y yum-utils
sudo yum-config-manager --add-repo https://packages.clickhouse.com/rpm/clickhouse.repo

2. Установите ClickHouse

sudo yum install -y clickhouse-server clickhouse-client

3. Запустите ClickHouse

sudo systemctl enable clickhouse-server
sudo systemctl start clickhouse-server

Установка ClickHouse из deb пакетов (Debian, Ubuntu etc)

1. Подключите официальный репозиторий

sudo apt-get install -y apt-transport-https ca-certificates dirmngr
sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 8919F6BD2B48D754
echo "deb https://packages.clickhouse.com/deb stable main" | sudo tee /etc/apt/sources.list.d/clickhouse.list
sudo apt-get update

2. Установите ClickHouse

sudo apt-get install -y clickhouse-server clickhouse-client

3. Запустите ClickHouse

sudo systemctl enable clickhouse-server
sudo systemctl start clickhouse-server

Настройка ClickHouse для сборок WCS до 5.2.1999

1. Для того, чтобы прослушивать входящие запросы на всех интерфесах сервера, раскомментируйте строку в файле /etc/clickhouse-server/config.xml 

    <listen_host>::</listen_host>

2. Для того, чтобы создать пользователя, укажите для пользователя default в файле /etc/clickhouse-server/users.xml параметр

            <access_management>1</access_management>

3. Перезапустите ClickHouse

systemctl restart clickhouse-server

4. Создайте базу данных wcs и таблицы в ней

cat wcs_clickhouse.sql | clickhouse-client -mn

CREATE DATABASE IF NOT EXISTS wcs;

DROP DICTIONARY IF EXISTS wcs.DictionaryStreamEvents;
 
DROP DICTIONARY IF EXISTS wcs.DictionaryConnectionEvents;
 
DROP DICTIONARY IF EXISTS wcs.DictionaryCDNEvents;

DROP DICTIONARY IF EXISTS wcs.DictionaryHlsStreamEventType;

DROP DICTIONARY IF EXISTS wcs.DictionaryHlsStreamEventSeverity;
 
DROP DICTIONARY IF EXISTS wcs.DictionaryMixerEvents;

DROP DICTIONARY IF EXISTS wcs.DictionaryBufferStateTypes;
 
DROP TABLE IF EXISTS wcs.StreamEvent;
 
DROP TABLE IF EXISTS wcs.ConnectionEvent;
 
DROP TABLE IF EXISTS wcs.CDNEvent;
 
DROP TABLE IF EXISTS wcs.StreamEventTypes;
 
DROP TABLE IF EXISTS wcs.ConnectionEventTypes;
 
DROP TABLE IF EXISTS wcs.CDNEventTypes;

DROP TABLE IF EXISTS wcs.MediaSessionEvents;

DROP TABLE IF EXISTS wcs.HlsStreamEvents;

DROP TABLE IF EXISTS wcs.HlsSegmenterEvents;

DROP TABLE IF EXISTS wcs.HlsStreamEventSeverity;

DROP TABLE IF EXISTS wcs.HlsStreamEventType;

DROP TABLE IF EXISTS wcs.HlsClientEvents;

DROP TABLE IF EXISTS wcs.MixerEvent;

DROP TABLE IF EXISTS wcs.MixerEventTypes;

DROP TABLE IF EXISTS wcs.RtmpInBufferEvent;

DROP TABLE IF EXISTS wcs.AudioRecoveryEvent;

DROP TABLE IF EXISTS wcs.BufferStateTypes;

CREATE TABLE wcs.ConnectionEventTypes
(
    `id` UInt32,
    `type` String
)
ENGINE = MergeTree()
ORDER BY id
SETTINGS index_granularity = 8192;
 
INSERT INTO wcs.ConnectionEventTypes VALUES (0, 'CONNECTED'), (1, 'DISCONNECTED');
 
CREATE TABLE wcs.StreamEventTypes
(
    `id` UInt32,
    `type` String
)
ENGINE = MergeTree()
ORDER BY id
SETTINGS index_granularity = 8192;
 
INSERT INTO wcs.StreamEventTypes VALUES (0,'CREATED'),(1,'LOCAL_SDP_CREATED'),(2,'REMOTE_SDP_RECEIVED'),(3,'ICE_STARTED'),(4,'ICE_COMPLETE'),(5,'DTLS_STARTED'),(6,'DTLS_COMPLETE'),(7,'INITIALIZED'),(8,'DISPOSING'),(9,'DISPOSED'),(10,'AUDIO_RECEIVED'),(11,'VIDEO_RECEIVED'),(12,'VIDEO_KFRAME_RECEIVED'),(13,'AUDIO_RTCP_RECEIVED'),(14,'VIDEO_RTCP_RECEIVED'),(15,'RESOLUTION_RECEIVED'),(16,'VIDEO_ENCODER_CREATED'),(17,'AUDIO_ENCODER_CREATED'),(18,'VIDEO_ENCODER_DISPOSED'),(19,'AUDIO_ENCODER_DISPOSED'),(20,'TERMINATED'),(21,'AUDIO_SENT'),(22,'VIDEO_SENT'),(23,'VIDEO_JITTER_BUFFER_STALL'),(24,'SENT_PLI'),(25,'RECEIVED_PLI'),(26,'SYNC_BUFFER_FULL'),(27,'SYNC_FORCE_FAILED'),(28,'SYNC_SHIFT'),(29,'SYNC_DEVIATION'),(30,'VIDEO_STATS'),(31,'RECORD');
 
CREATE TABLE wcs.CDNEventTypes
(
    `id` UInt32,
    `type` String
)
ENGINE = MergeTree()
ORDER BY id
SETTINGS index_granularity = 8192;
 
INSERT INTO wcs.CDNEventTypes VALUES (0, 'STATE'), (1, 'CDN_STATE'), (2, 'VERSION'), (3, 'ACL_REFRESH'), (4, 'ACL_UPDATE');
 
CREATE DICTIONARY wcs.DictionaryStreamEvents (
    `id` UInt16,
    `type` String DEFAULT ''
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(
 host 'localhost'
 port 9000
 user 'default'
 password ''
 db 'wcs'
 table 'StreamEventTypes'
))   
LAYOUT(FLAT())
LIFETIME(300);
 
CREATE DICTIONARY wcs.DictionaryConnectionEvents (
    `id` UInt16,
    `type` String DEFAULT ''
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(
 host 'localhost'
 port 9000
 user 'default'
 password ''
 db 'wcs'
 table 'ConnectionEventTypes'
))   
LAYOUT(FLAT())
LIFETIME(300);
 
CREATE DICTIONARY wcs.DictionaryCDNEvents (
    `id` UInt16,
    `type` String DEFAULT ''
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(
 host 'localhost'
 port 9000
 user 'default'
 password ''
 db 'wcs'
 table 'CDNEventTypes'
))   
LAYOUT(FLAT())
LIFETIME(300);
 
CREATE TABLE wcs.StreamEvent
(
    `timestamp` UInt64,
    `ip` IPv4,
    `sessionId` String,
    `mediaSessionId` String,
    `streamName` String,
    `eventType` UInt64,
    `eventPayload` String
)
ENGINE = MergeTree()
ORDER BY (sessionId, mediaSessionId, streamName)
SETTINGS index_granularity = 8192;
 
CREATE TABLE wcs.ConnectionEvent
(
    `timestamp` UInt64,
    `ip` IPv4,
    `sessionId` String,
    `eventType` UInt64,
    `eventPayload` String
)
ENGINE = MergeTree()
ORDER BY (timestamp, sessionId)
SETTINGS index_granularity = 8192;
 
CREATE TABLE wcs.CDNEvent
(
    `timestamp` UInt64,
    `ip` IPv4,
    `nodeId` String,
    `eventType` UInt64,
    `eventPayload` String
)
ENGINE = MergeTree()
ORDER BY (nodeId, eventType)
SETTINGS index_granularity = 8192;

CREATE TABLE wcs.MediaSessionEvents
(
    `timestamp` UInt64,
    `ip` IPv4,
    `mediaSessionId` String,
    `streamName` String,
    `videoProfileId` UInt32,
    `videoWidth` UInt32,
    `videoHeight` UInt32,
    `videoFrameRate` UInt32,
    `videoKframes` UInt64,
    `videoPframes` UInt64,
    `videoBframes` UInt64,
    `videoRate` UInt64,
    `audioRate` UInt64,
    `videoSyncTime` UInt64,
    `audioSyncTime` UInt64,
    `videoTimestamp` UInt64,
    `audioTimestamp` UInt64,
    `lastKeyFrameSyncTime` UInt64,
    `sendNACK` UInt64,
    `recvNACK` UInt64,
    `videoFramesLost` UInt64,
    `audioPacketsLost` UInt64,
    `audioPlaybackSpeed` Float32,
    `videoPlaybackSpeed` Float32
)
ENGINE = MergeTree()
ORDER BY (mediaSessionId, streamName)
SETTINGS index_granularity = 8192;

CREATE TABLE wcs.HlsSegmenterEvents
(
    `timestamp` UInt64,
    `ip` IPv4,
    `streamId` String,
    `variantName` String,
    `segmentId` String,
    `segmentStartPts` UInt64,
    `videoStartPts` UInt64,
    `audioStartPts` UInt64,
    `videoWidth` UInt32,
    `videoHeight` UInt32,
    `videoFrameCount` UInt32,
    `audioPacketCount` UInt32,
    `segmentDuration` UInt64,
    `independent` Bool,
    `gap` Bool,
    `discontinuity` Bool,
    `segmentInterval` UInt64,
    `partial` Bool,
    `playbackSpeed` Float32
)
ENGINE = MergeTree()
ORDER BY (streamId)
SETTINGS index_granularity = 8192;

CREATE TABLE wcs.HlsStreamEventSeverity
(
    `id` UInt8,
    `type` String
)
ENGINE = MergeTree()
ORDER BY id
SETTINGS index_granularity = 8192;

INSERT INTO wcs.HlsStreamEventSeverity VALUES (0, 'INFO'), (1, 'WARNING'), (2, 'ERROR');

CREATE TABLE wcs.HlsStreamEventType
(
    `id` UInt16,
    `type` String
)
ENGINE = MergeTree()
ORDER BY id
SETTINGS index_granularity = 8192;

INSERT INTO wcs.HlsStreamEventType
VALUES (0, 'PLAYBACK_SPEED'), (1, 'FPS_CHANGED'), (2, 'GAP'), (3, 'RESOLUTION_CHANGED'), (4, 'DISCONTINUITY'), (5, 'TASK_SKIPPED'), (6, 'NO_KYE_FRAME'), (7, 'NO_VIDEO'), (8, 'NO_AUDIO'), (9, 'SEGMENT_INTERVAL'), (10, 'OTHER');

CREATE DICTIONARY wcs.DictionaryHlsStreamEventSeverity
(
   `id` UInt8,
   `type` String DEFAULT ''
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(
 host 'localhost'
 port 9000
 user 'default'
 password ''
 db 'wcs'
 table 'HlsStreamEventSeverity'
))
LAYOUT(FLAT())
LIFETIME(300);

CREATE DICTIONARY wcs.DictionaryHlsStreamEventType
(
   `id` UInt16,
   `type` String DEFAULT ''
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(
 host 'localhost'
 port 9000
 user 'default'
 password ''
 db 'wcs'
 table 'HlsStreamEventType'
))
LAYOUT(FLAT())
LIFETIME(300);

CREATE TABLE wcs.HlsStreamEvents
(
    `timestamp` UInt64,
    `ip` IPv4,
    `severity` UInt8,
    `messageType` UInt16,
    `streamId` String,
    `variantName` String,
    `segmentId` String,
    `message` String
)
ENGINE = MergeTree()
ORDER BY (streamId)
SETTINGS index_granularity = 8192; 

CREATE TABLE wcs.HlsClientEvents
(
    `creationTime` UInt64,
    `responseTime` UInt64,
    `streamId` String,
    `variantName` String,
    `uri` String,
    `localIp` IPv4,
    `remoteIp` IPv4,
    `remotePort` UInt32,
    `userAgent` String,
    `httpStatus` UInt32,
    `clientId` UInt64
)
ENGINE = MergeTree()
ORDER BY (creationTime)
SETTINGS index_granularity = 8192;

CREATE TABLE wcs.MixerEvent
(
    `timestamp` UInt64,
    `mixerMediaSessionId` String,
    `mixerStreamName` String,
    `mediaSessionId` String,
    `streamName` String,
    `mixerAverageTickTimeInMs` Int64,
    `audioMixerSync` Int64,
    `videoMixerSync` Int64,
    `nextAudioDataTime` Int64,
    `nextVideoDataTime` Int64,
    `audioBuffered` Int64,
    `videoBuffered` Int64,
    `audioDropsCounter` Int64,
    `audioDropsSizeInBytes` Int64,
    `videoDropsCounter` Int64,
    `videoDropsSizeInBytes` Int64,
    `videoFps` Int64,
    `audioRate` DOUBLE,
    `videoRate` DOUBLE,
    `eventType` UInt32,
    `eventPayload` String
)
ENGINE = MergeTree()
ORDER BY timestamp
SETTINGS index_granularity = 8192;

CREATE TABLE wcs.MixerEventTypes
(
    `id` UInt32,
    `type` String
)
ENGINE = MergeTree()
ORDER BY id
SETTINGS index_granularity = 8192;

INSERT INTO wcs.MixerEventTypes VALUES (0, 'nullEvent'), (1, 'dropBallastAudio'), (2, 'dropBallastVideo'), (3, 'audioNotBuffered'), (4, 'videoNotBuffered'), (5, 'audioBufferExhausted'), (6, 'videoBufferExhausted'), (7, 'alignStreamFailed'), (8, 'alignStreamDropAudio'), (9, 'alignStreamDropVideo'), (10, 'rateOutOfBoundsAudio'), (11, 'rateOutOfBoundsVideo');

CREATE DICTIONARY wcs.DictionaryMixerEvents (
    `id` UInt16,
    `type` String DEFAULT ''
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(
 host 'localhost'
 port 9000
 user 'default'
 password ''
 db 'wcs'
 table 'MixerEventTypes'
))
LAYOUT(FLAT())
LIFETIME(300);

CREATE TABLE wcs.RtmpInBufferEvent
(
    `timestamp` UInt64,
    `streamClockTime` UInt64,
    `mediaSessionId` String,
    `streamName` String,
    `nextAudioDataTime` Int64,
    `nextVideoDataTime` Int64,
    `audioBuffered` Int64,
    `videoBuffered` Int64,
    `maximumAllowedBuffer` Int64,
    `bufferingCounter` Int64,
    `lastAudioDataTime` Int64,
    `lastVideoDataTime` Int64,
    `bufferState` UInt32
)
ENGINE = MergeTree()
ORDER BY timestamp
SETTINGS index_granularity = 8192;

CREATE TABLE wcs.BufferStateTypes
(
    `id` UInt32,
    `type` String
)
ENGINE = MergeTree()
ORDER BY id
SETTINGS index_granularity = 8192;

INSERT INTO wcs.BufferStateTypes VALUES (0, 'BUFFERING'), (1, 'HOLD'), (2, 'TERMINATED'), (3, 'OVERFLOW'), (4, 'PASSTHROUGH');

CREATE DICTIONARY wcs.DictionaryBufferStateTypes (
    `id` UInt16,
    `type` String DEFAULT ''
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(
 host 'localhost'
 port 9000
 user 'default'
 password ''
 db 'wcs'
 table 'BufferStateTypes'
))
LAYOUT(FLAT())
LIFETIME(300);

CREATE TABLE wcs.AudioRecoveryEvent
(
    `timestamp` UInt64,
    `mediaSessionId` String,
    `type` UInt32,
    `rtpTimestamp` UInt64
)
ENGINE = MergeTree()
ORDER BY rtpTimestamp
SETTINGS index_granularity = 8192;

5. Создайте пользователя wcs и дайте ему права на таблицы в базе данных wcs

cat wcs_clickhouse_users.sql | clickhouse-client -mn

CREATE USER IF NOT EXISTS wcs IDENTIFIED BY 'wcs';
GRANT ALL ON wcs.* TO wcs WITH GRANT OPTION;

6. Отключите управление пользователями для пользователя default, указав в файле /etc/clickhouse-server/users.xml параметр

            <access_management>0</access_management>

7. Перезапустите ClickHouse

systemctl restart clickhouse-server

Настройка ClickHouse для сборок WCS 5.2.1999 и новее

В сборке 5.2.1999 добавлено автоматическое создание БД при подключении определенной версии WCS с определенного IP адреса. Данные о всех БД хранятся в таблице WcsMetadata.metadata

SELECT *
FROM WcsMetadata.metadata

┌─dbName───────────────────────────────┬─ip────────────┬─hostname──────────────┬─versionNumber─┬─versionHash──────────────────────────────┐
│ test1flashphonercom_192168065_521999 │ 192.168.0.65  │ test1.flashphoner.com │ 5.2.1999      │ 93e39647113e0121dabc4283ef700814c355568f │
│ test2flashphonercom_192168039_521999 │ 192.168.0.39  │ test2.flashphoner.com │ 5.2.1999      │ 93e39647113e0121dabc4283ef700814c355568f │
└──────────────────────────────────────┴───────────────┴───────────────────────┴───────────────┴──────────────────────────────────────────┘

Поэтому для настройки ClickHouse необходимо только создать пользователя и дать ему права на создание Баз данных, таблиц и вставку данных в них

1. Для того, чтобы прослушивать входящие запросы на всех интерфесах сервера, раскомментируйте строку в файле /etc/clickhouse-server/config.xml 

    <listen_host>::</listen_host>

2. Для того, чтобы создать пользователя, укажите для пользователя default в файле /etc/clickhouse-server/users.xml параметры

            <access_management>1</access_management>
            <named_collection_control>1</named_collection_control>
            <show_named_collections>1</show_named_collections>
            <show_named_collections_secrets>1</show_named_collections_secrets>

3. Перезапустите ClickHouse

systemctl restart clickhouse-server

4. Создайте пользователя wcs и дайте ему необходимые права

cat wcs_clickhouse_users.sql | clickhouse-client -mn

CREATE USER IF NOT EXISTS wcs IDENTIFIED BY 'wcs';
SET allow_introspection_functions = 1;
GRANT ALL ON *.* TO wcs WITH GRANT OPTION;

5. Отключите управление пользователями для пользователя default, указав в файле /etc/clickhouse-server/users.xml параметр

            <access_management>0</access_management>

6. Перезапустите ClickHouse

systemctl restart clickhouse-server

Настройка WCS

Сбор данных в БД ClickHouse включается настройкой, в которой перечисляются типы отправляемых данных

rels_enabled=CONNECTION,STREAM,CDN,MEDIA_SESSION

Доступны следующие типы данных:

Настройка подключения к ClickHouse до сборки WCS 5.2.1999

Адрес сервера ClickHouse, база данных и протокол задаются настройками

rels_client_type=HTTP
rels_database_address=http://clickhouseserver:8123/wcs?user=wcs&password=wcs

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

rels_client_type=JDBC
rels_database_address=jdbc:clickhouse://clickhouseserver:8123/wcs?user=wcs&password=wcs

Настройка подключения к ClickHouse в сборках WCS 5.2.1999 и новее

Адрес сервера ClickHouse, пользователь и протокол задаются настройками

rels_client_type=HTTP
rels_database_address=clickhouseserver:8123
rels_database_properties=user=wcs&password=wcs

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

rels_client_type=JDBC

Остановка сбора данных без перезапуска WCS

При необходимости, передача данных с конкретного WCS сервера в ClickHouse может быть остановлена без перезапуска WCS. Для этого:

1. Отключите сбор данных в настройках сервера

rels_enabled=

2. Перезагрузите настройки сервера из интерфейса командной строки

reload node-settings

Изменение адреса сервера ClickHouse без перезапуска WCS

Адрес сервера ClickHouse может быть изменен без перезапуска WCS. Для этого:

1. Измените адрес в настройках сервера

rels_database_address=jdbc:clickhouse://newclickhouseserver:8123/wcs?user=wcs&password=wcs

2. Отключите сбор данных в настройках сервера

#rels_enabled=CONNECTION,STREAM,CDN,MEDIA_SESSION
rels_enabled=

3. Перезагрузите настройки сервера из интерфейса командной строки

reload node-settings

4. Включите сбор данных в настройках сервера

rels_enabled=CONNECTION,STREAM,CDN,MEDIA_SESSION

5. Перезагрузите настройки сервера из интерфейса командной строки

reload node-settings

Количество процессорных потоков, используемых клиентом ClickHouse

В сборке 5.2.2005 добавлена настройка количества процессорных потоков, используемых клиентом ClickHouse. По умолчанию, клиент будет использовать 2 потока, которые будет запускать для отправки данных и останавливать после отправки

rels_database_thread_pool_size=2

Если установить значение этого параметра в 0

rels_database_thread_pool_size=0

клиент ClickHouse создаст постоянный пул процессорных потоков размером

(CPU count * 2) + 1

Сжатие данных при отправке

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

rels_enable_compression=false

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

Управление сбором данных по REST API

Типы данных CONNECTION, STREAM, CDN, HLS_STREAM, REST_HOOKS собираются безусловно, для всех клиентских сессий и всех опубликованных потоков. Все остальные типы данных собираются только по запросу, поскольку объем отсылаемых данных может быть очень велик.

Сбор данных для определенного потока включается по REST API.

REST-запрос должен быть HTTP/HTTPS POST запросом в таком виде:

• HTTP: http://streaming.flashphoner.com:8081/rest-api/rels/startup
• HTTPS: https://streaming.flashphoner.com:8444/rest-api/rels/startup

Здесь:

• streaming.flashphoner.com - адрес WCS-сервера
• 8081 - стандартный REST / HTTP порт WCS-сервера
• 8444 - стандартный HTTPS порт
• rest-api - обязательный префикс
• /rels/startup - используемый REST-вызов

REST методы и статусы ответа

/rels/startup

Запустить сбор определенных типов данных для определенного потока

Request example

POST /rest-api/rels/startup HTTP/1.1
Host: localhost:8081
Content-Type: application/json

{
    "mediaSession": {
        "frequency":100,
        "ids": [
            "d7d6b6e4-b137-461a-8a32-d6abf2b8666e",
            "39cbf770-128a-11ef-b839-d1a1f53f8bd2"
        ] 
    }
}

Response example

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Content-Type: application/json

Return codes

/rels/find_all

Получить статистику по отправляемым данным

Request example

POST /rest-api/rels/find_all HTTP/1.1
Host: localhost:8081
Content-Type: application/json

Response example

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Content-Type: application/json

[
  {
    "type": "CONNECTION",
    "sentBytes": 1989,
    "bitrateKbps": 0,
    "sentEvents": 6,
    "queueEvents": 0
  },
  {
    "type": "STREAM",
    "sentBytes": 70766,
    "bitrateKbps": 0,
    "sentEvents": 73,
    "queueEvents": 1
  },
  {
    "type": "CDN",
    "sentBytes": 0,
    "bitrateKbps": 0,
    "sentEvents": 0,
    "queueEvents": 0
  },
  {
    "type": "MEDIA_SESSION",
    "ids": [
      "d7d6b6e4-b137-461a-8a32-d6abf2b8666e",
      "39cbf770-128a-11ef-b839-d1a1f53f8bd2"
    ],
    "sentBytes": 205794,
    "bitrateKbps": 143,
    "sentEvents": 999,
    "queueEvents": 119
  }
]

Return codes

/rels/terminate

Остановить сбор определенных типов данных для определенного потока

Request example

POST /rest-api/rels/terminate HTTP/1.1
Host: localhost:8081
Content-Type: application/json

{
    "mediaSession": {
        "ids": [
            "d7d6b6e4-b137-461a-8a32-d6abf2b8666e"
        ] 
    }
}

Response example

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Content-Type: application/json

Return codes

/rels/terminate_all

Остановить сбор определенных типов данных для всех потоков

Request example

POST /rest-api/rels/terminate_all HTTP/1.1
Host: localhost:8081
Content-Type: application/json

{
    "types": [
        "MEDIA_SESSION"
    ]
}

Response example

HTTP/1.1 200 OK
Access-Control-Allow-Origin: *
Content-Type: application/json

{
  "MEDIA_SESSION": [
    "39cbf770-128a-11ef-b839-d1a1f53f8bd2"
  ]
}

Return codes

Параметры

Автоматический сбор данных по условиям

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

Поддерживаются следующие типы данных:

• MEDIA_SESSION
• MIXER
• HLS_SEGMENTER

• HLS_CLIENT (начиная со сборки 5.2.2091)

• RTMP_IN_BUFFER (начиная со сборки 5.2.2091)
• RTP_SYNCHRONIZATION (начиная со сборки 5.2.2091)

Условия задаются в виде фильтров в файле /usr/local/FlashphonerWebCallServer/conf/rels_trap.json 

{
    "mediaSession":
    [
        {
            "ips": ["127.0.0.1"],
            "frequency": "1000ms"
        },
        {
            "ips": ["192.168.0.0/24", "192.168.2.0/24"],
            "streamName": ".*\\-screen"
        },
        {
            "ips": ["192.168.0.101/32"],
            "streamName": "test.*",
            "frequency": "100th"
        },
        {
            "streamName": "test.*",
            "mediaType": ["publish", "play"]
        }
    ],

    "mixer":
    [
        {
            "name": "conference\\-.*",
            "streamName": "user.*",
            "frequency": "1000ms"
        }
    ],

    "hlsSegmenter":
    [
        {
            "streamId": "test.*"
        }
    ],

    "hlsClient":[
        {
            "streamId": "test.*"
        }
    ],
    
    "rtpSynchronization":[
        {
            "ips":["192.168.0.0/24"],
            "streamName": "test.*"
        }
    ],

    "rtmpInBuffer":[
        {
            "ips":["192.168.0.0/24"],
            "streamName": "test.*"
        }
    ]
}

Здесь:

• mediaSession - блок фильтров для отправки данных MEDIA_SESSION
  • ips - список IP адресов клиентов
  • streamName - маска имени в виде регулярного выражения
  • type - тип сессии: "publish", "play"
  • frequency - частота сбора данных для потока, подпадающего под фильтр
• mixer - блок фильтров для отправки данных MIXER
  • name - маска имени микшера в виде регулярного выражения
  • streamName - маска имени входящего потока микшера в виде регулярного выражения
  • frequency - частота сбора данных для микшера, подпадающего под фильтр
• hlsSegmenter - блок фильтров для отправки данных HLS_SEGMENTER
  
  • streamId - маска имени HLS потока в виде регулярного выражения
• hlsClient - блок фильтров для отправки данных HLS_CLIENT
  • streamId - маска имени HLS потока в виде регулярного выражения
• rtpSynchronization - блок фильтров для отправки данных RTP_SYNCHRONIZATION
  • ips - список IP адресов публикующих клиентов
  • streamName - маска имени публикуемого потока в виде регулярного выражения
• rtmpInBuffer - блок фильтров для отправки данных RTMP_IN_BUFFER
  • ips - список IP адресов публикующих клиентов
  • streamName - маска имени публикуемого потока в виде регулярного выражения

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

{
    "mediaSession":
    [
        {
            "ips": ["127.0.0.1", "192.168.0.0/24"],
            ...
        },
        ...
    ],
    ...
}

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

{
    "mediaSession":
    [
        ...,
        {
            "ips": ["192.168.2.0/24"],
            "streamName": ".*\\-screen",
            ...
        },
        ...
    ],
    ...
}

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

{
    ...,

    "mixer":
    [
        {
            "name": "conference\\-.*",
            "streamName": "user.*",
            "frequency": "1000ms"
        }
    ],

    ...
}

Параметр frequency определяет частоту сбора данных для этого типа:

• "frequency": "100ms" - данные собираются каждые 100 мс
• "frequency": "100th" - данные собираются каждые 100 полученных пакетов

Изменения в файле rels_trap.json не требуют перезапуска сервера, но применяются к новым публикациям, микшерам и HLS потокам.

Выборки информации из БД

Выборки информации из БД производятся при помощи SQL запросов в клиенте ClickHouse

Примеры запросов для сборок WCS до 5.2.1999

select timestamp,ip,sessionId,mediaSessionId,streamName,dictGetString('wcs.DictionaryStreamEvents','type', eventType) as eventType from wcs.StreamEvent where streamName = 'test'

select timestamp,ip,sessionId,dictGetString('wcs.DictionaryConnectionEvents','type', eventType) as eventType from wcs.ConnectionEvent

select timestamp,ip,nodeId,dictGetString('wcs.DictionaryCDNEvents','type', eventType) as eventType,eventPayload from wcs.CDNEvent

Примеры запросов для сборок WCS 5.2.1999 и новее

select timestamp,ip,sessionId,mediaSessionId,streamName,dictGetString('wcs.DictionaryStreamEvents','type', eventType) as eventType from test1flashphonercom_192168065_521999.StreamEvent where streamName = 'test'

select timestamp,ip,sessionId,dictGetString('wcs.DictionaryConnectionEvents','type', eventType) as eventType from test1flashphonercom_192168065_521999.ConnectionEvent

select timestamp,ip,nodeId,dictGetString('wcs.DictionaryCDNEvents','type', eventType) as eventType,eventPayload from test1flashphonercom_192168065_521999.CDNEvent

Запись данных в файлы

Не всегда имеется возможность развернуть сервер ClickHouse для мониторинга в уже работающей инфраструктуре. В связи с этим, в сборке 5.2.2058 добавлена возможность записи данных в текстовые файлы, которые затем могут быть импортированы в БД ClickHouse для дальнейшего анализа.

Запись файлов включена по умолчанию настройкой

rels_client_type=FILE

Данные записываются в каталог /usr/local/FlashphonerWebCallServer/logs/rels. В этом каталоге создается подкаталог с именем БД, сформированным из имени хоста, IP адреса сервера и версии WCS. В свою очередь, в этот каталог помещаются текстовые файлы для каждой таблицы данных

/usr/local/FlashphonerWebCallServer/logs/rels/
└── testflashphonercom_192168039_522060
    ├── ConnectionEvent.csv
    ├── ConnectionEvent.json
    ├── StreamEvent.csv
    └── StreamEvent.json

Каждая таблица состоит из двух файлов:

• json файл содержит описание типов данных таблицы, которое требуется для импорта в ClickHouse
• csv файл содержит сами данные, колонки данных разделяются запятыми

Импорт данных из файлов в БД ClickHouse

Для импорта данных в БД ClickHouse требуется установленный WCS сервер (например, в тестовой среде). В состав сервера входит инструмент для импорта, запускаемый следующим образом

/usr/local/FlashphonerWebCallServer/tools/rels_import.sh --dbpath=/usr/local/FlashphonerWebCallServer/logs/rels/testflashphonercom_192168039_522060

Здесь

• dbpath - путь до каталога с файлами таблиц, которые нужно импортировать

Инструмент создаст БД с именем, соответствующим имени каталога с файлами таблиц, в данном примере testflashphonercom_192168039_522060.

Инструмент использует настройки подключения к БД ClickHouse из файла /usr/local/FlashphonerWebCallServer/conf/flashphoner.properties, например

rels_client_type=HTTP
rels_database_address=database.flashphoner.com:8123
rels_database_properties=user=wcs&password=wcs

При использовании БД Clickhouse 24 и новее, для импорта будет использован механизм дедупликации данных, чтобы повторно не добавлять в таблицы полностью совпадающие строки, если инструмент импорта был дважды запущен на одних и тех же данных.