Система приема Netflow и NEL на базе Clickhouse, Kafka и pmacct.

Предположим, что от маршрутизатора Cisco, который выполняет функции BRAS и NAT подают на наш сервер два потока
1) NetFlow в формате IPFIX, порт UDP 2055, это статистика (Кто, куда, сколько байт передал). Формат полей данных следующий:

{

  "ip_src": "10.10.145.224", 
  "ip_dst": "54.191.232.8",
  "port_src": 44354,
  "port_dst": 443,
  "bytes": 4934,
  "packets": 5,
  "ip_proto": 6,                        
  "timestamp_start": 1787910941512000000,
  "timestamp_end": 1787910941737000000,
  "peer_ip_src": "172.19.0.2"
}
2) NEL в формате NetFlow v9, порт UDP 2056, это события состояния сессии NAT (создалась сессия NAT, удалилась сессия NAT).
NEL может передаваться разными шаблонами. Будем использовать самый простой шаблон, который маршрутизатор отдает постоянно. Формат данных следующий:
{
  "peer_ip_src": "172.19.0.2",
  "ip_src": "10.10.148.170", 
  "port_src": 49534,
  "post_nat_ip_src": "97.221.112.206", 
  "post_nat_port_src": 58001,
  "nat_event": 1,             // 1=CREATE, 2=DELETE
  "ip_proto": "udp",
  "timestamp_start": "2026-09-07 16:36:10.415000"
}
Настроим связку которая будет принимать данные и стабильно складывать в базу Clickhouse.
Данные будут приниматься и разбираются на лету сервисом pmacct. Каждое сообщение будет преобразовываться в JSON формат и передается в агрегатор Kafka. Kafka будет выступать буфером, сглаживающим пики и объединяющим порции данных перед вставкой в Clickhouse.
Clickhouse будет постоянно читать топики Kafka и забирать новые данные в базу. Поток flow и поток nel будут очевидно хранится в разных таблицах базы, так как имеют разную структуру.

*** Установка ClickHouse
# apt-get install -y apt-transport-https ca-certificates curl gnupg
# cd /usr/src/
У меня под рукой был старый сервер с процессором Intel Xeon E5620, который не поддерживает современные версии Clickhouse.
Процессор Intel Xeon E5620 относится к семейству Westmere (2010 год). Поддерживает инструкции SSE4.2, но не поддерживает AVX/AVX2. Это важно, так как современные версии ClickHouse требуют AVX2.
Оптимальным выбором будет 22.3 LTS, которые все еще поддерживаются и работают с SSE4.2.
# wget https://packages.clickhouse.com/tgz/lts/clickhouse-common-static-22.3.19.6-amd64.tgz
# wget https://packages.clickhouse.com/tgz/lts/clickhouse-server-22.3.19.6-amd64.tgz
# wget https://packages.clickhouse.com/tgz/lts/clickhouse-client-22.3.19.6-amd64.tgz
Установка:
# tar -xzvf clickhouse-common-static-22.3.19.6-amd64.tgz
# ./clickhouse-common-static-22.3.19.6/install/doinst.sh
# tar -xzvf clickhouse-server-22.3.19.6-amd64.tgz
# ./clickhouse-server-22.3.19.6/install/doinst.sh
# tar -xzvf clickhouse-client-22.3.19.6-amd64.tgz
# ./clickhouse-client-22.3.19.6/install/doinst.sh
Настройка Clickhouse
1. Задать пользовательские настройки
# nano /etc/clickhouse-server/users.xml
1.1. Задать пароль по умолчанию для доступа к Clickhouse
Секция <users> -> <default> -> <password>
Прописываем пароль:
  <password>*******</password>
1.2. Задать ограничения по использованию памяти и ядер процессора
Секция <profiles> -> <default> 
Размер памяти определяем как 50% от памяти, которую выделим Clickhouse (в моем случае - 32 ГБ)
Для сервера с 16 ядрами и выделенными 32 ГБ RAM под ClickHouse, рекомендация max_threads = 4.
   <max_memory_usage>17179869184</max_memory_usage>
   <max_threads>4</max_threads>
2.Правим конфигурацию сервера:
# nano /etc/clickhouse-server/config.xml
2.1. Отключаем использование IPv6, раскомментируем строку (тем самым отключаем IPv6)
 <listen_host>0.0.0.0</listen_host> 
2.2. Изменяем уровень логирования – оставляем только ошибки
Так же устанавливаем размер лог-файла в 100Мбайт (вместо 1G)
 <logger>
   <level>warning</level> 
   <size>100M</size>
 <logger>
2.3. Ограничиваем сервер полкой в использовании памяти - в 32G, иначе сервер съест другие приложения
 <max_server_memory_usage>34359738368</max_server_memory_usage>
2.4. Отключаем процентное указание в использование памяти
 <max_server_memory_usage_to_ram_ratio>0</max_server_memory_usage_to_ram_ratio>
2.5. Проверить, что Clickhouse работает с директорией, которая для этого была создана в примонтированном хранилище:
 <path>/var/lib/clickhouse/</path>
2.6. Из блока <openSSL> вырезать раздел <server>, оставить только раздел клиента
  <openSSL>
    <client>
  ...
    </client>
  </openSSL>
2.7. Добавляем в конец конфигурации перед закрывающим тегом </clickhouse> директивы, определяющие протокол взаимодействия между Kafka и Clickhouse.
Это необходимо, так как версия 22.3 не совместима с новыми сервисами Kafka.
    <kafka>
        <api_version_request>false</api_version_request>
        <broker_version_fallback>2.4.1</broker_version_fallback>
    </kafka>
</clickhouse>
Создаем пользователя clickhouse и передаем ему права на директорию, в которой он должен работать
# groupadd --system clickhouse
# useradd --system --gid clickhouse --shell /usr/sbin/nologin --home-dir /var/lib/clickhouse clickhouse
Передаем clickhouse права на директорию:
# chown -R clickhouse:clickhouse /var/lib/clickhouse/
# chmod -R 750 /var/lib/clickhouse/
Создаем каталог логов
# mkdir -p /var/log/clickhouse-server
# chown clickhouse:clickhouse /var/log/clickhouse-server
# chmod 0750 /var/log/clickhouse-server
Запускаем сервис:
# service clickhouse-server start
Проверка:
# systemctl status clickhouse-server
# ss -tlnp | grep clickhouse
Подключение к серверу с паролем:
# clickhouse-client --query "SELECT 1" --password
Если что-то идет не так смотрим логи:
# journalctl -u clickhouse-server -b -n 100 --no-pager
# tail -f /var/log/clickhouse-server/clickhouse-server.err.log 
Вносим Clickhouse в автозагрузку
# systemctl enable clickhouse-server


*** Установка Kafka
# apt update
Устанавливаем Java
# apt install -y openjdk-21-jdk
# java --version
openjdk 21.0.12.1 2026-08-18
Затем скачать и распаковать Kafka. Для работы с Clickhouse 22.3 нет возможности использовать новую версию Kafka. Используем версию 3.7.1, которая совместима с Clickhouse 22.3.
# wget https://archive.apache.org/dist/kafka/3.7.1/kafka_2.12-3.7.1.tgz
# tar -xzf kafka_2.12-3.7.1.tgz  -C /opt/
# mv /opt/kafka_2.12-3.7.1 /opt/kafka
Конфигурации kafka тут: /opt/kafka/config/
Нас интересует конфигурация в режиме kraft: /opt/kafka/config/kraft/server.properties
# nano /opt/kafka/config/kraft/server.properties
Тут важно указать директорию, что бы данные были именно в ней и не забивали нужный диск.
Выбираем директорию /home/kafka, так как тут есть место.
Так же необходимо указать формат взаимодействия с Clickhouse - 3.0
Дополнительно указываем размер топиков на дисков и время жизни топиков:
 log.dirs=/home/kafka
 log.message.format.version=3.0
 # Хранить данные не более 24 часов
 log.retention.hours=24
 # Ограничение на один топик - 10G
 log.segment.bytes=10737418240
 # Оганичение размера сегмента
 log.segment.bytes=536870912  # 512MB
 # Время жизни сегмента
 log.segment.ms=3600000  # 1 час
Создаем директорию для данных, которую определили в конфигурации и директорию для логов:
# mkdir -p /home/kafka
# chmod -R 755 /home/kafka
# mkdir -p /var/log/kafka
Генерация идентификатора кластера Kafka и форматирование директории с даными Kafka
# CLUSTER_ID=$(/opt/kafka/bin/kafka-storage.sh random-uuid)
#  /opt/kafka/bin/kafka-storage.sh format \
  -t $CLUSTER_ID \
  -c /opt/kafka/config/kraft/server.properties
Создаем пользователя под которым будет работать kafka и передаем ему права на директории:
# useradd -r -s /bin/false -d /opt/kafka kafka
# chown -R kafka:kafka /opt/kafka /var/log/kafka /home/kafka
Сервис systemd
# nano /etc/systemd/system/kafka.service
Содержание файла:
[Unit]
Description=Apache Kafka Server (KRaft mode)
Documentation=http://kafka.apache.org/documentation.html
Requires=network.target remote-fs.target
After=network.target remote-fs.target
[Service]
Type=simple
User=kafka
Group=kafka
# Переменные окружения
Environment="JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64"
Environment="KAFKA_HEAP_OPTS=-Xmx1G -Xms1G"
Environment="KAFKA_OPTS=-Djava.net.preferIPv4Stack=true"
Environment="LOG_DIR=/var/log/kafka"
# Команда запуска
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties
# Команда остановки (мягкая)
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
# Политика перезапуска
Restart=on-failure
RestartSec=10
# Лимиты
LimitNOFILE=100000
LimitNPROC=100000
[Install]
WantedBy=multi-user.target
# systemctl daemon-reload
# systemctl start kafka
Проверка:
# systemctl status kafka
# journalctl -u kafka -f -n 100
# ss -tlnp | grep -E '9092|9093'
# systemctl enable kafka

=>>> ПРОБНЫЕ ЗАПИСИ <<<=
=>>> Тестовый топик
=>>> # /opt/kafka/bin/kafka-topics.sh --create \
       --topic test-topic \
       --bootstrap-server localhost:9092 \
       --partitions 3 \
       --replication-factor 1
=>>> Список топиков
=>>> # /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092
=>>> Тестовое сообщение
=>>> # /opt/kafka/bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092
    Теперь вводим несколько сообщений и для выхода Ctrl+C
=>>> Информация о топике
=>>> # /opt/kafka/bin/kafka-topics.sh --describe --topic test-topic --bootstrap-server localhost:9092
=>>> Принят сообщение:
=>>> # /opt/kafka/bin/kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server localhost:9092
=>>> Список всех топиков:
=>>> # /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
=>>> Удаление тестового топика:
=>>> # /opt/kafka/bin/kafka-topics.sh --delete --topic test-topic --bootstrap-server localhost:9092

Теперь создаем топики pmacct-flows и pmacct-nel для приема сообщений Netflow (сам поток Netflow и NEL):
# /opt/kafka/bin/kafka-topics.sh --create \
  --topic pmacct-flows \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1 
# /opt/kafka/bin/kafka-topics.sh --create \
  --topic pmacct-nel \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1
  

*** Сборка и установка pmacct с поддержкой kafka
# /usr/src/
# apt install git
# apt update && sudo apt install -y libtool automake autoconf
# apt install -y pkgconf build-essential
# apt install -y libpcap-dev
# apt install -y libcurl4-openssl-dev
# apt install -y libmariadb-dev-compat libmariadb-dev
# apt install -y libnuma-dev
# apt install -y librdkafka-dev
# apt install -y libjansson-dev
# git clone https://github.com/pmacct/pmacct.git
# cd pmacct/
# ./autogen.sh
# ./configure --enable-jansson --enable-kafka --disable-bgp-bins --disable-bmp-bins --disable-st-bins
# make -j$(nproc)
# make install
# nfacctd -V
В выводе мы должны увидеть версию программы и информацию о том, что плагин kafka подключен
 Libs:
  cdada 0.6.4
  libpcap version 1.10.5 (with TPACKET_V3)
  rdkafka 2.8.0
  jansson 2.14
 Plugins:
  ...
  kafka
Создаем директорию для логов:
# mkdir -p /var/log/pmacct
Конфиги лежат тут: /etc/pmacct/
Создаем конфиг для приема основного потока Netflow: pmacct-flows.conf
# nano /etc/pmacct/pmacct-flows.conf
 ##### Для теста с выводом в консоль:
 #daemonize: false
 #pidfile: /var/run/pmacct-flows.pid
 #logfile: /var/log/pmacct/pmacct-flows.log
 #debug: true 
 #nfacctd_port: 2055
 #nfacctd_time_new: true
 #plugins: print[flows]
 #aggregate[flows]: src_host, dst_host, src_port, dst_port, proto, timestamp_start, peer_src_ip
 #print_output[flows]: json
 #print_refresh_time[flows]: 5
 ##### Для записи в Kafka
 #daemonize: true  # Для запуска без systemd
 daemonize: false
 pidfile: /var/run/pmacct-flows.pid
 logfile: /var/log/pmacct/pmacct-flows.log
 debug: false
 nfacctd_port: 2055
 nfacctd_time_new: true
 plugins: kafka[flows]
 aggregate[flows]: src_host, dst_host, src_port, dst_port, proto, timestamp_start, peer_src_ip
 kafka_broker_host[flows]: localhost
 kafka_broker_port[flows]: 9092
 kafka_topic[flows]: pmacct-flows
 kafka_partition_key[flows]: src_host
 kafka_refresh_time[flows]: 10
[Пробный запуск /usr/local/sbin/nfacctd -f /etc/pmacct/pmacct-flows.conf ]

Создаем unit systemd для запуска через systemd
# nano /etc/systemd/system/pmacct-flows.service
 [Unit]
 Description=pmacct NetFlow Accounting Daemon
 Documentation=http://www.pmacct.net/
 After=network.target kafka.service
 Wants=kafka.service
 [Service]
 Type=simple
 User=root
 Group=root
 ExecStart=/usr/local/sbin/nfacctd -f /etc/pmacct/pmacct-flows.conf
 Restart=on-failure
 RestartSec=5
 LimitNOFILE=65536
 [Install]
 WantedBy=multi-user.target

Создаем конфиг для приема потока NEL: pmacct-nel.conf
# nano /etc/pmacct/pmacct-nel.conf
 ##### Для теста с выводом в консоль:
 #daemonize: false
 #pidfile: /var/run/pmacct-nel.pid
 #logfile: /var/log/pmacct/pmacct-nel.log
 #debug: true 
 #nfacctd_port: 2056
 #nfacctd_time_new: true
 #plugins: print[cgn]
 #aggregate[cgn]: src_host, post_nat_src_host, src_port, post_nat_src_port, proto, nat_event, timestamp_start, peer_src_ip
 #aggregate_filter[cgn]: dst_host == ""
 #print_output[cgn]: json
 #print_refresh_time[cgn]: 5
 #print_output_file[cgn]: /dev/stdout
 ##### Для записи в Kafka
 #daemonize: true
 daemonize: false
 pidfile: /var/run/pmacct-nel.pid
 logfile: /var/log/pmacct/pmacct-nel.log
 debug: false
 nfacctd_port: 2056
 nfacctd_time_new: true
 plugins: kafka[cgn]
 aggregate[cgn]: src_host, post_nat_src_host, src_port, post_nat_src_port, proto, nat_event, timestamp_start, peer_src_ip
 aggregate_filter[cgn]: dst_host == ""
 kafka_broker_host[cgn]: localhost
 kafka_broker_port[cgn]: 9092
 kafka_topic[cgn]: pmacct-nel
 kafka_partition_key[cgn]: src_host
 kafka_refresh_time[cgn]: 10
[Пробный запуск /usr/local/sbin/nfacctd -f /etc/pmacct/pmacct-nel.conf ]

Создаем unit systemd для запуска через systemd
# nano /etc/systemd/system/pmacct-nel.service
 [Unit]
 Description=pmacct NEL
 Documentation=http://www.pmacct.net/
 After=network.target kafka.service
 Wants=kafka.service
 [Service]
 Type=simple
 User=root
 Group=root
 ExecStart=/usr/local/sbin/nfacctd -f /etc/pmacct/pmacct-nel.conf
 Restart=on-failure
 RestartSec=5
 LimitNOFILE=65536
 [Install]
 WantedBy=multi-user.target
Запускаем:
[Если запускал руками, то сначала нужно убить процессы nfacctd]
# systemctl daemon-reload
# systemctl start pmacct-flows
# systemctl status pmacct-flows
# systemctl start pmacct-nel
# systemctl status pmacct-nel

Проверка того, что данные залетают в Kafka:
# ps aux | grep nfacctd
# /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic pmacct-flows \
  --from-beginning \
  --max-messages 5
# /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic pmacct-nel \
  --from-beginning \
  --max-messages 5
# systemctl enable pmacct-flows
# systemctl enable pmacct-nel


** Подключаем Kafka к Clickhouse
1 часть. Поток Netflow.
# clickhouse-client --password (нужен пароль)
1. Создаем базу данных:
CREATE DATABASE IF NOT EXISTS netflow;
2. Создание Kafka Engine таблицы для подключения к Kafka (удаление старой таблицы DROP TABLE IF EXISTS netflow.flows_kafka;)
CREATE TABLE netflow.flows_kafka
(
 `value` String
)
ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'localhost:9092',
kafka_topic_list = 'pmacct-flows',
kafka_group_name = 'clickhouse-flows',
kafka_format = 'JSONAsString';
3. Создание таблицы хранения данных: (удаление старой таблицы DROP TABLE IF EXISTS netflow.flows;)
CREATE TABLE netflow.flows
(
    `timestamp_start` DateTime CODEC(Delta(4), ZSTD(1)),
    `ip_src` IPv4,
    `ip_dst` IPv4,
    `port_src` UInt16 DEFAULT 0 CODEC(ZSTD(1)),
    `port_dst` UInt16 DEFAULT 0 CODEC(ZSTD(1)),
    `proto` LowCardinality(String) DEFAULT '' CODEC(ZSTD(1)),
    `peer_ip_src` IPv4,
    `bytes` UInt64 DEFAULT 0 CODEC(ZSTD(1)),
    `packets` UInt32 DEFAULT 0 CODEC(ZSTD(1))
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(timestamp_start)
ORDER BY (timestamp_start, ip_src, ip_dst)
TTL timestamp_start + toIntervalMonth(6)
SETTINGS index_granularity = 8192;
4. Создание Materialized View для форматирования принимаемых из Kafka данных (удаление старой таблицы DROP TABLE IF EXISTS netflow.flows_mv;)
CREATE MATERIALIZED VIEW netflow.flows_mv TO netflow.flows AS
SELECT
    toDateTime(substring(JSONExtractString(value, 'timestamp_start'), 1, 19)) AS timestamp_start,
    toIPv4(nullIf(JSONExtractString(value, 'ip_src'), '')) AS ip_src,
    toIPv4(nullIf(JSONExtractString(value, 'ip_dst'), '')) AS ip_dst,
    toUInt16(JSONExtractInt(value, 'port_src')) AS port_src,
    toUInt16(JSONExtractInt(value, 'port_dst')) AS port_dst,
    JSONExtractString(value, 'ip_proto') AS proto,
    toIPv4(nullIf(JSONExtractString(value, 'peer_ip_src'), '')) AS peer_ip_src,
    toUInt64(JSONExtractInt(value, 'bytes')) AS bytes,
    toUInt32(JSONExtractInt(value, 'packets')) AS packets
FROM netflow.flows_kafka
WHERE (JSONExtractString(value, 'ip_src') != '') AND (JSONExtractString(value, 'ip_dst') != '')
Проверка логов на предмет чего-то критичного:
# tail -n 20 /var/log/clickhouse-server/clickhouse-server.err.log | grep -i kafka
2 часть. Поток NEL
5. Создание Kafka Engine таблицы для подключения к Kafka (удаление старой таблицы DROP TABLE IF EXISTS netflow.nel_kafka;)
CREATE TABLE IF NOT EXISTS netflow.nel_kafka 
(
    `value` String
)
ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'localhost:9092',
    kafka_topic_list = 'pmacct-nel',
    kafka_group_name = 'clickhouse-nel',
    kafka_format = 'JSONAsString';
6. Создание таблицы хранения данных NEL (удаление старой таблицы DROP TABLE IF EXISTS netflow.nel;)
CREATE TABLE netflow.nel
(
    `timestamp_start` DateTime CODEC(Delta(4), ZSTD(1)),
    `ip_src` IPv4,
    `port_src` UInt16 DEFAULT 0 CODEC(ZSTD(1)),
    `proto` LowCardinality(String) DEFAULT '' CODEC(ZSTD(1)),
    `peer_ip_src` IPv4,
    `post_nat_ip_src` IPv4,
    `post_nat_port_src` UInt16 DEFAULT 0 CODEC(ZSTD(1)),
    `nat_event` UInt8 DEFAULT 0 CODEC(ZSTD(1))
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(timestamp_start)
ORDER BY (timestamp_start, ip_src, post_nat_ip_src)
TTL timestamp_start + toIntervalMonth(6)
SETTINGS index_granularity = 8192;
7. Создание Materialized View для форматирования принимаемых из Kafka данных (удаление старой таблицы DROP TABLE IF EXISTS netflow.nel_mv;)
CREATE MATERIALIZED VIEW IF NOT EXISTS netflow.nel_mv TO netflow.nel AS
SELECT
    toDateTime(substring(JSONExtractString(value, 'timestamp_start'), 1, 19)) AS timestamp_start,
    toIPv4(nullIf(JSONExtractString(value, 'ip_src'), '')) AS ip_src,
    toUInt16(JSONExtractInt(value, 'port_src')) AS port_src,
    JSONExtractString(value, 'ip_proto') AS proto,
    toIPv4(nullIf(JSONExtractString(value, 'peer_ip_src'), '')) AS peer_ip_src,
    toIPv4(nullIf(JSONExtractString(value, 'post_nat_ip_src'), '')) AS post_nat_ip_src,
    toUInt16(JSONExtractInt(value, 'post_nat_port_src')) AS post_nat_port_src,
    toUInt8(JSONExtractInt(value, 'nat_event')) AS nat_event
FROM netflow.nel_kafka
WHERE 
    JSONExtractString(value, 'ip_src') != '' AND 
    JSONExtractString(value, 'post_nat_ip_src') != '';

*** Включаем firewall
# apt install ufw
# ufw allow ssh
# ufw allow 2055/udp
# ufw allow 2056/udp
# ufw allow from 10.0.0.0/8 to any port 8123 proto tcp
# ufw allow from 10.0.0.0/8 to any port 9000 proto tcp
# ufw enable
Проверка:
# ufw status verbose
# systemctl status ufw


*** Включение ротации логов

У сlickhouse ротацию логов уже включили через конфиг при насройке clickhouse

Логи демона pmacct храняться тут /var/log/pmacct
Используем стандартный logrotate для ротации логов
# nano /etc/logrotate.d/pmacct
Содержимое файла:
/var/log/pmacct/*.log {
    hourly
    rotate 2
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
}
Этот конфиг копирует текущий лог в архив и очищает исходный файл. pmacct продолжает писать в тот же файловый дескриптор, поэтому перезапуск сервиса не нужен.
Проверка синтаксиса:
# logrotate -d /etc/logrotate.d/pmacct
Принудительная ротация логов прямо сейчас:
# logrotate -f /etc/logrotate.d/pmacct
Ротация будет проводиться раз в сутки. Когда следующая ротация покажет команда:
# systemctl list-timers logrotate.timer

Логи kafka расположены в директории /var/log/kafka
Kafka использует log4j для логирования. Файл конфигурации логирования:  /opt/kafka/config/log4j.properties
А вот удаление старых файлов kafka делать не умеет.
Для удаления старых файлов используем команду в crontab: 
# Очищение логов kafka старше 5 суток
42 *    * * *   root  find /var/log/kafka -type f -name "*.log.20*" -mmin +7200 -delete > /dev/null 2>&1

Установка GoFlow2 в Debian

GoFlow2 принимает данные из потоков Netflow и в стандартном случае передает на стандартный выход в консоль.

# cd /usr/src/
# wget https://github.com/netsampler/goflow2/releases/download/v2.2.6/goflow2_2.2.6_amd64.deb
# apt-get install -f -y ./goflow2_2.2.6_amd64.deb
# goflow2 --v
GoFlow2 v2.2.6 (2025-12-27T05:02:22+0000)

Проверка наличия сервиса systemd
# systemctl cat goflow2

Запуск:  
# goflow2   -listen "netflow://:2055"   -format json
# goflow2   -listen "netflow://:2056"   -format json

Удаление пакета
# dpkg --purge goflow2

Gitea. Установка и базовая настройка

Создаем директорию для контейнера gitea

# mkdir /mnt/data/docker_projects/gitea

В интерфейсе portainer переходим в окружение local.

Выбираем пункты меню Stacks --> + Add Stack. Прописываем имя в поле Name - gitea

Оставляем выбор метода задания параметров Web editor.

В поле редактора конфига вносим конфигурацию:

services:

  gitea:

    image: gitea/gitea:latest

    container_name: gitea

    restart: unless-stopped

    ports:

      - "3000:3000"

      - "2222:22"

    volumes:

      - /mnt/data/docker_projects/gitea/data:/data

    environment:

      - USER_UID=1000

      - USER_GID=1000

      - GITEA__database__DB_TYPE=postgres

      - GITEA__database__HOST=db:5432

      - GITEA__database__NAME=gitea

      - GITEA__database__USER=gitea

      - GITEA__database__PASSWD=gitea_pass

      - GITEA__server__ROOT_URL=http://gitea.example.ru/

      - GITEA__server__LOCAL_ROOT_URL=http://localhost:3000/

      - GITEA__server__SSH_PORT=2222

      - GITEA__server__SSH_DOMAIN=gitea.example.ru

      - GITEA__packages__ENABLED=true

      - GITEA__server__DOMAIN=gitea.example.ru

      - GITEA__service__DISABLE_REGISTRATION=true

      - GITEA__service__REQUIRE_SIGNIN_VIEW=true

    depends_on:

      db:

        condition: service_healthy

  db:

    image: postgres:16-alpine

    container_name: gitea-db

    restart: unless-stopped

    volumes:

      - /mnt/data/docker_projects/gitea/db:/var/lib/postgresql/data

    environment:

      - POSTGRES_USER=gitea

      - POSTGRES_PASSWORD=gitea_pass

      - POSTGRES_DB=gitea

    healthcheck:

      test: ["CMD-SHELL", "pg_isready -U gitea"]

      interval: 5s

      timeout: 5s

      retries: 5

Активируем кнопку "Deploy the stack", Будет создан стек контейнеров.

Открываем порт для работы через web-интерфейс с Gitea

# firewall-cmd --permanent --add-port=3000/tcp

# firewall-cmd --reload

# firewall-cmd --list-all

Открываем WEB-интерфейс GITEA

http://IP:3000/

Будет показан стартовый экран первичных настроек.

Важно написать название сервиса. Например, Gitea smarts.

Прокручиваем вниз и нажимает "Установить Gitea".

Затем нужно задать первого пользователя - администратора системы и нажать "Регистрация аккаунта".

 

Будет создан пользователь и откроется интерфейс gitea.

Создаем репозитарий.

Раздел Репозитарий --> + Новый репозитарий.

Задаем Название репозитория.

Устанавливаем флаг Видимость (Сделать репозиторий приватным).

Другие флажки не устанавливаем.

Нажимаем кнопку "Создать репозитарий"

Теперь на сервере разработки создаем ключ SSH по которому с сервера можно будет коннектиться к Gitea.

Проверка налияи ключа:

# ls -l ~/.ssh/ | grep pub

Показываем ключ:

# cat ~/.ssh/id_ecdsa.pub 

ecdsa-sha2-nistp256 Arx7r.....rlgehf0/XGro4B9A= root@server

И переносим его в Gitea в раздел Аватар (справа вверху) --> Настройки --> Ключи SSH / GPG Keys --> Управление ключами SSH --> Добавить ключ

Задаем имя ключа, вставляем ключ в поле и проводим его добавление.

Сбросить пароль в gitea

Если вдруг забыл пароль:

# docker exec -it -u git gitea gitea admin user change-password \

  --username korolev \

  --password <НОВЫЙ_ВРЕМЕННЫЙ_ПАРОЛЬ>

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

x