Vault+Pydantic: конфигурация с четкой структурой и валидацией

Предисловие

Продолжение саги. Сразу, на всякий случай

В данной статье я расскажу о конфигурации для вашей сервисов с помощью связки Vault (KV и пока только первой версии, т.е. без версионирования секретов) и Pydantic (Settings) под патронажем Sitri.

Итак, допустим, что у нас есть приложение superapp с заведёнными конфигами в Vault и аутентификацией с помощью approle, примерно так настроим (настройку policies для доступа к секрет-энжайнам и к самим секретам я оставлю за кадром, так как это достаточно просто и статья не об этом):

Key                        Value
---                        -----
bind_secret_id             true
local_secret_ids           false
policies                   [superapp_service]
secret_id_bound_cidrs      <nil>
secret_id_num_uses         0
secret_id_ttl              0s
token_bound_cidrs          []
token_explicit_max_ttl     0s
token_max_ttl              30m
token_no_default_policy    false
token_num_uses             50
token_period               0s
token_policies             [superapp_service]
token_ttl                  20m
token_type                 default

Прим.: естественно, что если у вас есть возможность и приложение выходит в боевой режим, то secret_id_ttl лучше делать не бесконечным, выставляя 0 секунд.

SuperApp требует конфигурации: подключения к базе данных, подключение к kafka и faust конфигурации для работы кластера воркеров.

Подготовим Sitri

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

Итак, для начала сконфигурируем vault-провайдер в условном файле provider_config.py:

import hvac  

from sitri.providers.contrib.vault import VaultKVConfigProvider  
from sitri.providers.contrib.system import SystemConfigProvider  

configurator = SystemConfigProvider(prefix="superapp")  
ENV = configurator.get("env")  

def vault_client_factory() -> hvac.Client:  
    client = hvac.Client(url=configurator.get("vault_api"))  

    client.auth_approle(  
        role_id=configurator.get("role_id"),  
  secret_id=configurator.get("secret_id"),  
  )  

    return client  

provider = VaultKVConfigProvider(  
    vault_connector=vault_client_factory, mount_point=f"{configurator.get('app_name')}/{ENV}"  
)

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

export SUPERAPP_ENV=dev
export SUPERAPP_APP_NAME=superapp
export SUPERAPP_VAULT_API=https://your-vault-host.domain
export SUPERAPP_ROLE_ID=535b268d-b858-5fb9-1e3e-79068ca77e27 # Пример
export SUPERAPP_SECRET_ID=243ab423-12a2-63dc-3d5d-0b95b1745ccf # Пример

В примере предполагается, что базовый mount_point к вашим секретам для определённой среды будет содержать имя приложения и имя среды, поэтому мы и экспортировали SUPERAPP_ENV. Путь до секретов отдельных частей приложения мы будем определять в settings-классах далее, поэтому в провайдере secret_path мы оставляем пустым.

Классы настроек

Начнём по пунктам и разнесём три класса настроек (БД, Kafka, Faust) по трём разным файлам.

Настройки БД

from pydantic import Field  

from sitri.settings.contrib.vault import VaultKVSettings  

from superapp.config.provider_config import provider  

class DBSettings(VaultKVSettings):  
    user: str = Field(..., vault_secret_key="username")  
    password: str = Field(...)  
    host: str = Field(...)  
    port: int = Field(...)  

    class Config:  
        provider = provider  
        default_secret_path = "db"

Итак, как видите, конфиг. данные для базы у нас достаточно простые. Этот класс будет по-умолчанию смотреть в секрет superapp/dev/db, так, как мы указали в config классе, в остальном здесь простые pydantic поля, но в одном из них присутствует extra-аргумент vault_secret_key — он нужен тогда, когда ключ в секрете не совпадает по имени с pydantic полем в нашем классе, если его не указывать, то провайдер будет искать ключ по имени поля.

Например, в нашем тестовом приложении, предполагается, что в секрете superapp/dev/db, есть ключи password и username, но мы хотим, чтобы последний был помещён в поле user для удобства и краткости.

Поместим в вышеозначенный секрет следующие данные для примера:

{
  "host": "testhost",
  "password": "testpassword",
  "port": "1234",
  "username": "testuser"
}

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

db_settings = DBSettings()
pprint(db_settings.dict())
# -> 
# {
#     "host": "testhost",
#     "password": "testpassword",
#     "port": 1234,
#     "user": "testuser"
# }

Настройки Kafka

from typing import Dict, Any  

from pydantic import Field  

from sitri.settings.contrib.vault import VaultKVSettings  

from superapp.config.provider_config import provider, configurator  

class KafkaSettings(VaultKVSettings):  
    mechanism: str = Field(..., vault_secret_key="auth_mechanism")  
    brokers: str = Field(...)  
    auth_data: Dict[str, Any] = Field(...)  

    class Config:  
        provider = provider  
        default_secret_path = "kafka"  
        default_mount_point = f"{configurator.get('app_name')}/common"

В данном случае, представим, что инстанс кафки для разных сред нашего сервиса один, поэтому секрет хранится по пути superapp/common/kafka

{
  "auth_data": "{\"password\": \"testpassword\", \"username\": \"testuser\"}",
  "auth_mechanism": "SASL_PLAINTEXT",
  "brokers": "kafka://test"
}

Класс настройки поймёт комплексный тип данных Dict[str, Any] и распарсит его в словарь, то есть при заполнении наших настроек будут следующие данные:

{
    "auth_data":
    {
        "password": "testpassword",
        "username": "testuser"
    },
    "brokers": "kafka://test",
    "mechanism": "SASL_PLAINTEXT"
}

Так же, если секрет будет задан напрямую в json, например так:

{
  "auth_data": {
    "password": "testpassword",
    "username": "testuser"
  },
  "auth_mechanism": "SASL_PLAINTEXT",
  "brokers": "kafka://test"
}

То класс настроек тоже сможет правильно разложить данные.

P.S.
Так же, secret_path и mount_point можно задавать на уровне полей, чтобы провайдер запросил конкретные значения из разных секретов (если это требуется). Приведу цитату с приоритезацией пути секрета и точки монтирования из документации:

Secret path prioritization:

1. vault_secret_path (Field arg)
2. default_secret_path (Config class field)
3. secret_path (provider initialization optional arg)

Mount point prioritization:

1. vault_mount_point (Field arg)
2. default_mount_point (Config class field)
3. mount_point (provider initialization optional arg)

Настройки Faust и отдельных воркеров

from typing import Dict  

from pydantic import Field, BaseModel  

from sitri.settings.contrib.vault import VaultKVSettings  

from superapp.config.provider_config import provider  

class AgentConfig(BaseModel):  
    partitions: int = Field(...)  
    concurrency: int = Field(...)  

class FaustSettings(VaultKVSettings):  
    app_name: str = Field(...)  
    default_partitions_count: int = Field(..., vault_secret_key="partitions_count")  
    default_concurrency: int = Field(..., vault_secret_key="agent_concurrency")  
    agents: Dict[str, AgentConfig] = Field(default=None, vault_secret_key="agents_specification")  

    class Config:  
        provider = provider  
        default_secret_path = "faust"

superapp/dev/faust:

{
  "agent_concurrency": "5",
  "app_name": "superapp-workers",
  "partitions_count": "10"
}

В данном случае, по-умолчанию у нас есть глобальные значения кол-ва партиций в кафке и concurrency для агентов. Таким образом, по-умолчанию наши настройки будут выгружены так:

{
  "agents": None,
  "app_name": "superapp-workers",
  "default_concurrency": 5,
  "default_partitions_count": 10
}

Например, у нас есть агент X с настройками:

{
  "partitions": 5,
  "concurrency": 2
}

Наш секрет в связи с этим должен выглядеть следующим образом:

{
  "agent_concurrency": "5",
  "agents_specification": {
    "X": {
      "concurrency": "2",
      "partitions": "5"
    }
  },
  "app_name": "superapp-workers",
  "partitions_count": "10"
}

Как и ожидалось данные корректно смапились и типы значений были преобразованы так, как указано в модели AgentConfig:

{
    "agents":
    {
        "X":
        {
            "concurrency": 2,
            "partitions": 5
        }
    },
    "app_name": "superapp-workers",
    "default_concurrency": 5,
    "default_partitions_count": 10
}

Совмещаем в единый конфиг класс

from pydantic import BaseModel, Field  

from superapp.config.database_settings import DBSettings  
from superapp.config.faust_settings import FaustSettings  
from superapp.config.kafka_settings import KafkaSettings  

class AppSettings(BaseModel):  
    db: DBSettings = Field(default_factory=DBSettings)  
    faust: FaustSettings = Field(default_factory=FaustSettings)  
    kafka: KafkaSettings = Field(default_factory=KafkaSettings)

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

Давайте запустим наш код и проверим, как всё сработается вместе:

from superapp.config import AppSettings  

config = AppSettings()  

print(config)  
print(config.dict())

Получаем общий вывод всей конфигурации приложения:

db=DBSettings(user='testuser', password='testpassword', host='testhost', port=1234) 
faust=FaustSettings(app_name='superapp-workers', default_partitions_count=10, default_concurrency=5, agents={'X': AgentConfig(partitions=5, concurrency=2)}) 
kafka=KafkaSettings(mechanism='SASL_PLAINTEXT', brokers='kafka://test', auth_data={'password': 'testpassword', 'username': 'testuser'})

{
    "db":
    {
        "host": "testhost",
        "password": "testpassword",
        "port": 1234,
        "user": "testuser"
    },
    "faust":
    {
        "agents":
        {
            "X":
            {
                "concurrency": 2,
                "partitions": 5
            }
        },
        "app_name": "superapp-workers",
        "default_concurrency": 5,
        "default_partitions_count": 10
    },
    "kafka":
    {
        "auth_data":
        {
            "password": "testpassword",
            "username": "testuser"
        },
        "brokers": "kafka://test",
        "mechanism": "SASL_PLAINTEXT"
    }
}

Счастье, радость, восторг!

У нас получилась такая структура тест-проекта:

superapp
├── config
│   ├── app_settings.py
│   ├── database_settings.py
│   ├── faust_settings.py
│   ├── __init__.py
│   ├── kafka_settings.py
│   └── provider_config.py
├── __init__.py
└── main.py

Послесловие

Как видите настройка достаточно проста с Sitri, после неё мы получаем чёткую схему конфигурации с нужными нам типами данных у значений, даже если в vault по-умолчанию они хранились строками.

Пишите комментарии по поводу либы, кода или общие впечатления. Буду рад любому отзыву!

P.S. Код из статьи я залил на github