Разворачиваем СУБД временных рядов InfluxDB

22 октября 2023

На первый взгляд может показаться, что такие данные вполне себе неплохо хранятся и в реляционных базах данных. Однако, потом начинаешь понимать, насколько же это неудобно. Простой пример: давайте представим систему мониторинга компьютерной техники в организации, в базу данных которой периодически пишутся показатели температуры компонентов, нагрузки, состояния накопителей и т.д. И всё прекрасно, пока парк техники более-менее одинаковый: просто создаём отдельный столбец в таблице под каждый параметр мониторинга. Но тут происходит апгрейд и в некоторые компьютеры добавляется несколько дополнительных накопителей, которые тоже необходимо мониторить. У нас есть два варианта: либо создать дополнительные столбцы (тогда для остальных хостов они будут пустыми), либо менять структуру всей базы и выносить накопители в отдельную таблицу (и писать каждый накопитель отдельной строкой).

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

Плюсы и минусы TSDB по сравнению с реляционными СУБД:

Плюсы:

• Гибкость добавления новых элементов мониторинга.
• Нет необходимости отправки текущего времени в запросе при записи в базу — время автоматически будет подставлено самой СУБД. Таким образом решается проблема при одновременной записи данных с разных устройств с несинхронизированными часами.
• Большая скорость добавления новых данных.
• Для каждого значения можно указать один или несколько тегов (например, модель накопителя и/или его наименование в системе).
• Автоматическое удаление устаревших данных (при необходимости).

Минусы:

• Медленное удаление данных
• Ограниченность NoSQL запросов

Разворачиваем InfluxDB в докере:

docker run -d --name influxdb-test \
      -p 8086:8086 \
      -v myInfluxVolume-test:/var/lib/influxdb2 \
      -v /var/influxbackup:/backup \
      influxdb:latest

Монтирование каталога /var/influxbackup не является обязательным, но значительно упростит резервное копирование базы данных.

Затем переходим в браузере по адресу http://IP:8086 и приступаем к первоначальной настройке:

Необходимо ввести имя пользователя и придумать пароль. Также нас просят ввести название организации (можно ввести что угодно, например, «home«). В последнем поле придумываем название бакета («ведро») — аналог базы данных в SQL базах.

На следующей странице необходимо скопировать и сохранить API токен. Он показывается только один раз и нигде не сохраняется. Данный токен понадобится, когда необходимо будет получить доступ к СУБД на уровне админа, например, при резервном копировании.

Затем нажимаем кнопку «Configure later»:

Попадаем в симпатичную админку InfluxDB:

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

• Arduino
• C#
• Dart
• GO
• Java
• Kotlin
• Node.js
• PHP
• Python
• R
• Ruby
• Scala
• Swift

Причём сразу в админке можно получить пример кода:

Готовый пример на Python для записи значений от 0 до 30 каждую секунду:

influx-sample.py

import influxdb_client, os, time
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
 
INFLUXDB_TOKEN = "*************************"
#token = os.environ.get("INFLUXDB_TOKEN")  # При чтении токена из переменных окружения
token = INFLUXDB_TOKEN
org = "home"
url = "http://**********:8086"
 
write_client = influxdb_client.InfluxDBClient(url=url, token=token, org=org)
 
bucket = "mybucket"
 
write_api = write_client.write_api(write_options=SYNCHRONOUS)
 
for value in range(30):
    point = (
        Point("measurement1")
        .tag("tagname1", "tagvalue1")
        .field("field1", value)
    )
    write_api.write(bucket=bucket, org="home", record=point)
    time.sleep(1)  # separate points by 1 second

Всё очень просто. При необходимости можно менять имена тегов и их значения, а также добавлять несколько тегов. Примеры для мониторинга:

point = (Point("loadavg").tag("host", "Host1").field("cpu1", value1) )
point = (Point("loadavg").tag("host", "Host1").field("cpu2", value2) )
...
point = (Point("drive").tag("host", "Host1").tag("mountpoint", "/mnt/hdd").field("free", freeValue) )
point = (Point("drive").tag("host", "Host1").tag("mountpoint", "/mnt/hdd").field("size", sizeValue) )
point = (Point("drive").tag("host", "Host1").tag("mountpoint", "/").field("size", sizeValueRoot) )

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

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

А так, например, выглядят данные о температуре на улице:

Если возникает необходимость перехода на InfluxDB с других СУБД, то перенести данные можно довольно легко с помощью CSV файлов или специального «Line Protocol». Рассмотрим как раз второй способ:

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

Каждая строка такого файла должна быть оформлена следующим образом:

// Syntax
<measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]

// Example
myMeasurement,tag1=value1,tag2=value2 fieldKey="fieldValue" 1556813561098000000

Например, для себя я написал вот такой простой скрипт для формирования таких файлов из данных базы MySQL:

convert-mysql-to-influx.py

from mysql.connector import connect, Error
 
DB_HOST = "********"
DB_USER = "********"
DB_PASS = "********"
DB_NAME = "********"
 
f = open("data0.txt", "w")
 
try:
    with connect(host=DB_HOST, user=DB_USER, password=DB_PASS,
                 database=DB_NAME) as connection:
        with connection.cursor() as cursor:
            sql = 'SELECT Time, Temperature FROM history WHERE Source=0;'
            cursor.execute(sql)
            result = cursor.fetchall()
            current = -1
            count = 0
            if len(result) != 0:
                for row in result:
                    s = 'meteo,place=home temp={} {}'.format(row[2], row[1])
                    count = count + 1
                    if count // 100000 != current:
                        current = count // 100000
                        f.close()
                        f = open(f"data{current}.txt", "w")
                    f.write(s+"\n")
 
except Exception as err:
    print(f"DB unexpected error: {err=}, {type(err)=}")
 
f.close()

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

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

sudo docker exec -it influxdb-test influx backup /backup -t INFLUXDB_TOKEN

где:

• influxdb-test — имя контейнера docker;
• /backup — имя каталога внутри контейнера, куда необходимо сохранить файлы бэкапа. При этом в нашем случае он ремапится на каталог /var/influxbackup хоста.
• INFLUXDB_TOKEN — токен доступа к InfluxDB, который мы записали (так ведь?) при установке.

Также, при необходимости, можно указать параметр –bucket для создания бэкапа только одной БД.

Для полного восстановления всех БД выполним следующую команду:

sudo docker exec -it influxdb-test influx restore /backup --full -t INFLUXDB_TOKEN

Внимание! В этом случае все существующие данные будут уничтожены и перезаписаны из резервной копии.

Если вы хотите восстановить только одну БД, то предварительно её необходимо удалить и создать заново, а затем выполнить следующую команду:

sudo docker exec -it influxdb-test influx restore /backup --bucket mybucket -t INFLUXDB_TOKEN

Восстановление в существующую БД не допускается. Но можно создать новую из резервной копии:

sudo docker exec -it influxdb-test influx restore /backup --bucket mybucket --new-bucket new-example-bucket -t INFLUXDB_TOKEN