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

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

В этом руководстве рассматривается, как реализовать оба подхода.

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

Поскольку учётные данные основной базы данных и реплик известны заранее, добавить поддержку основной базы данных и реплик в приложение Ecto относительно просто. Представим, что у вас есть MyApp.Repo и вы хотите добавить четыре реплики для чтения. Это можно сделать в три шага.

Сначала определите репозитории основной базы данных и реплик в lib/my_app/repo.ex:

defmodule MyApp.Repo do
  use Ecto.Repo,
    otp_app: :my_app,
    adapter: Ecto.Adapters.Postgres

  @replicas [
    MyApp.Repo.Replica1,
    MyApp.Repo.Replica2,
    MyApp.Repo.Replica3,
    MyApp.Repo.Replica4
  ]

  def replica do
    Enum.random(@replicas)
  end

  for repo <- @replicas do
    defmodule repo do
      use Ecto.Repo,
        otp_app: :my_app,
        adapter: Ecto.Adapters.Postgres,
        read_only: true
    end
  end
end

Приведённый выше код определяет обычный MyApp.Repo и четыре реплики с именами от MyApp.Repo.Replica1 до MyApp.Repo.Replica4. Мы передаём параметр :read_only репозиториям реплик, чтобы такие операции, как insert, update и другие подобные функции, были недоступны. Мы также определяем функцию replica, которая возвращает случайную реплику.

Затем нужно убедиться, что основная база данных и реплики правильно настроены в ваших файлах config/config.exs. В средах разработки и тестирования можно использовать одинаковые учётные данные для всех репозиториев, направив их на один и тот же адрес базы данных:

replicas = [
  MyApp.Repo,
  MyApp.Repo.Replica1,
  MyApp.Repo.Replica2,
  MyApp.Repo.Replica3,
  MyApp.Repo.Replica4
]

for repo <- replicas do
  config :my_app, repo,
    username: "postgres",
    password: "postgres",
    database: "my_app_prod",
    hostname: "localhost",
    pool_size: 10
end

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

repos = %{
  MyApp.Repo => "prod-primary",
  MyApp.Repo.Replica1 => "prod-replica-1",
  MyApp.Repo.Replica2 => "prod-replica-2",
  MyApp.Repo.Replica3 => "prod-replica-3",
  MyApp.Repo.Replica4 => "prod-replica-4"
}

for {repo, hostname} <- repos do
  config :my_app, repo,
    username: "postgres",
    password: "postgres",
    database: "my_app_prod",
    hostname: hostname,
    pool_size: 10
end

Наконец, убедитесь, что все репозитории запускаются в дереве супервизии:

children = [
  MyApp.Repo,
  MyApp.Repo.Replica1,
  MyApp.Repo.Replica2,
  MyApp.Repo.Replica3,
  MyApp.Repo.Replica4
]

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

MyApp.Repo.replica().all(query)

Теперь вы готовы работать с основной базой данных и репликами — без обходных решений и сложных зависимостей!

Тестирование реплик

Хотя всё, что мы сделали до этого момента, должно корректно работать в средах разработки и производства, этого может быть недостаточно для тестирования. Большинство разработчиков, тестирующих приложения Ecto, используют песочницу, например песочницу Ecto SQL.

При использовании песочницы каждый тест выполняется в изолированной независимой транзакции. После завершения теста транзакция откатывается. Это позволяет легко и с высокой производительностью отменить все изменения, выполненные в тесте.

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

user = Repo.insert!(%User{name: "jane doe"})
assert Repo.replica().get!(User, user.id)

Дело в том, что Repo.insert! выполнит запись через одно соединение с базой данных, а репозиторий, возвращённый Repo.replica(), выполнит чтение через другое соединение. Поскольку запись выполняется в транзакции, её содержимое не будет доступно другим соединениям до фиксации транзакции, а для тестовых соединений этого никогда не произойдёт.

Есть два способа решить эту проблему: изменить реплики или использовать динамические репозитории.

Пользовательское определение replica

Одно из простых решений описанной выше проблемы — использовать во время тестирования собственную реализацию replica, которая всегда возвращает основной репозиторий, например:

if Mix.env() == :test do
  def replica, do: __MODULE__
else
  def replica, do: Enum.random(@replicas)
end

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

Использование :default_dynamic_repo

Другой подход к тестированию — задать параметр :default_dynamic_repo при определении репозитория. Давайте разберёмся, что это значит.

Когда вы добавляете репозиторий в дерево супервизии, например MyApp.Repo, за кулисами запускается дерево супервизии с процессом под именем MyApp.Repo. По умолчанию процесс называется так же, как модуль репозитория. Каждый раз при вызове функции в MyApp.Repo, например MyApp.Repo.insert/2, Ecto использует пул соединений процесса с именем MyApp.Repo.

Начиная с версии 3.0 Ecto позволяет запускать несколько процессов из одного и того же репозитория. Единственное требование — у них должны быть разные имена процессов, например:

children = [
  MyApp.Repo,
  {MyApp.Repo, name: :another_instance_of_repo}
]

Хотя конкретно этот пример не очень осмыслен (далее мы рассмотрим реальный вариант использования этой возможности), идея в том, что теперь запущены два репозитория: один называется MyApp.Repo, другой — :another_instance_of_repo. У каждого из этих процессов собственный пул соединений. Указать Ecto, какой процесс использовать для операций репозитория, можно так:

MyApp.Repo.put_dynamic_repo(MyApp.Repo)
MyApp.Repo.put_dynamic_repo(:another_instance_of_repo)

После вызова MyApp.Repo.put_dynamic_repo(name) все вызовы, сделанные в MyApp.Repo, будут использовать пул соединений, указанный в name.

Как это поможет при тестировании реплик? Если вернуться к дереву супервизии, определённому ранее в этом руководстве, вы увидите следующее:

children = [
  MyApp.Repo,
  MyApp.Repo.Replica1,
  MyApp.Repo.Replica2,
  MyApp.Repo.Replica3,
  MyApp.Repo.Replica4
]

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

@replicas [
  MyApp.Repo.Replica1,
  MyApp.Repo.Replica2,
  MyApp.Repo.Replica3,
  MyApp.Repo.Replica4
]

setup do
  for replica <- @replicas do
    replica.put_dynamic_repo(MyApp.Repo)
  end

  :ok
end

Обратите внимание: put_dynamic_repo действует только для текущего процесса. Поэтому при каждом создании нового процесса значение dynamic_repo сбрасывается до значения по умолчанию, пока вы снова не вызовете put_dynamic_repo.

К счастью, есть способ ещё лучше! При определении репозитория можно передать параметр :default_dynamic_repo. В нашем случае нужно задать :default_dynamic_repo равным MyApp.Repo только в тестовой среде. В файле lib/my_app/repo.ex сделайте следующее:

  for repo <- @replicas do
    default_dynamic_repo =
      if Mix.env() == :test do
        MyApp.Repo
      else
        repo
      end

    defmodule repo do
      use Ecto.Repo,
        otp_app: :my_app,
        adapter: Ecto.Adapters.Postgres,
        read_only: true,
        default_dynamic_repo: default_dynamic_repo
    end
  end

Теперь тесты будут работать как раньше, при этом вы по-прежнему сможете обнаружить случайное выполнение операции записи в реплике.

Динамические репозитории

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

Например, можно динамически запустить репозиторий с заданными учётными данными:

MyApp.Repo.start_link(
  name: :some_client,
  hostname: "client.example.com",
  username: "...",
  password: "...",
  pool_size: 1
)

Иными словами, start_link принимает те же параметры, что и конфигурация базы данных. Теперь выполним запрос через динамически запущенный репозиторий. Если просто вызвать MyApp.Repo.all(Post), это может привести к ошибке, поскольку по умолчанию будет предпринята попытка использовать процесс с именем MyApp.Repo, который может быть запущен, а может и нет. Поэтому не забудьте сначала вызвать put_dynamic_repo/1:

MyApp.Repo.put_dynamic_repo(:some_client)
MyApp.Repo.all(Post)

Ecto также позволяет запускать репозиторий без имени (прямо как того знаменитого коня). В таком случае нужно явно передать name: nil и сопоставить результат MyApp.Repo.start_link/1, чтобы получить PID, который затем следует передать в put_dynamic_repo. Заодно воспользуемся этой возможностью, чтобы корректно очистить базу данных: остановим новый репозиторий и вернём значение put_dynamic_repo:

default_dynamic_repo = MyApp.Repo.get_dynamic_repo()

{:ok, repo} =
  MyApp.Repo.start_link(
    name: nil,
    hostname: "client.example.com",
    username: "...",
    password: "...",
    pool_size: 1
  )

try do
  MyApp.Repo.put_dynamic_repo(repo)
  MyApp.Repo.all(Post)
after
  MyApp.Repo.put_dynamic_repo(default_dynamic_repo)
  Supervisor.stop(repo)
end

Всё это также можно объединить в функцию, определённую в репозитории:

defmodule MyApp.Repo do
  use Ecto.Repo, ...

  def with_dynamic_repo(credentials, callback) do
    default_dynamic_repo = get_dynamic_repo()
    start_opts = [name: nil, pool_size: 1] ++ credentials
    {:ok, repo} = MyApp.Repo.start_link(start_opts)

    try do
      MyApp.Repo.put_dynamic_repo(repo)
      callback.()
    after
      MyApp.Repo.put_dynamic_repo(default_dynamic_repo)
      Supervisor.stop(repo)
    end
  end
end

Теперь используйте её так:

credentials = [
  hostname: "client.example.com",
  username: "...",
  password: "..."
]

MyApp.Repo.with_dynamic_repo(credentials, fn ->
  MyApp.Repo.all(Post)
end)

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