Add simple load-balancing with use libcluster (#57)

* Add simple load-balancing with use libcluster

* Update mix.exs

Co-authored-by: Przemysław Rożnawski <48837433+roznawsk@users.noreply.github.com>

* Update test/jellyfish/cluster/load_balancing_test.exs

Co-authored-by: Przemysław Rożnawski <48837433+roznawsk@users.noreply.github.com>

* Update lib/jellyfish/room_service.ex

Co-authored-by: Przemysław Rożnawski <48837433+roznawsk@users.noreply.github.com>

* Fix typo

* Fix another typo

* Remove unused variable

* Modify how cluster configuration is passed

* Code review fixes

* Fix typo

* Fix one more typo

* Move config jellyfish address outside if

* Code review fixes

* Use rpc.multicall for gather resources on other nodes

* Fix multicall

* Update openapi.yaml

Co-authored-by: Michał Śledź <michalsledz34@gmail.com>

* Update lib/jellyfish/room_service.ex

Co-authored-by: Michał Śledź <michalsledz34@gmail.com>

* More renaming

* Change module tag in load-balancing tests

* Modify docker-compose

---------

Co-authored-by: Przemysław Rożnawski <48837433+roznawsk@users.noreply.github.com>
Co-authored-by: Michał Śledź <michalsledz34@gmail.com>

.circleci/config.yml

@@ -3,7 +3,21 @@ orbs:
   elixir: membraneframework/elixir@1
   codecov: codecov/codecov@3.2.4
 
+executors:
+ machine_executor_amd64:
+   machine:
+     image: ubuntu-2204:2022.04.2
+   environment:
+     architecture: "amd64"
+     platform: "linux/amd64"
+
 jobs:
+  test_load_balancing:
+    executor: machine_executor_amd64
+    steps:
+      - checkout
+      - run: docker compose run test
+
   test:
     docker:
       - image: membraneframeworklabs/docker_membrane:latest
@@ -23,6 +37,7 @@ workflows:
       - elixir/build_test:
           cache-version: 3
       - test
+      - test_load_balancing
       - elixir/lint:
           cache-version: 3
           docs: false

config/ci.exs

@@ -0,0 +1,16 @@
+import Config
+
+config :jellyfish, server_api_token: "development"
+
+# We don't run a server during test. If one is required,
+# you can enable the server option below.
+config :jellyfish, JellyfishWeb.Endpoint,
+  http: [ip: {127, 0, 0, 1}, port: 4002],
+  secret_key_base: "DtVd7qfpae0tk5zRgAM75hOaCc+phk38gDFVvLPyqVN/vvVg0EPmksTSm5JcyjoJ",
+  server: false
+
+# Print only warnings and errors during test
+config :logger, level: :warning
+
+# Initialize plugs at runtime for faster test compilation
+config :phoenix, :plug_init_mode, :runtime

config/config.exs

@@ -20,4 +20,8 @@ config :logger, :console,
 
 config :phoenix, :json_library, Jason
 
+config :jellyfish,
+  divo: "docker-compose.yaml",
+  divo_wait: [dwell: 1_500, max_tries: 50]
+
 import_config "#{config_env()}.exs"

config/runtime.exs

@@ -63,6 +63,25 @@ defmodule ConfigParser do
   end
 end
 
+hosts =
+  System.get_env("NODES", "")
+  |> String.split(" ")
+  |> Enum.reject(&(&1 == ""))
+  |> Enum.map(&String.to_atom(&1))
+
+unless Enum.empty?(hosts) do
+  config :libcluster,
+    topologies: [
+      epmd_cluster: [
+        strategy: Cluster.Strategy.Epmd,
+        config: [hosts: hosts]
+      ]
+    ]
+end
+
+host = System.get_env("VIRTUAL_HOST") || "example.com"
+port = String.to_integer(System.get_env("PORT") || "4000")
+
 config :jellyfish,
   webrtc_used: String.downcase(System.get_env("WEBRTC_USED", "true")) not in ["false", "f", "0"],
   integrated_turn_ip:
@@ -77,7 +96,8 @@ config :jellyfish,
     System.get_env("INTEGRATED_TURN_TCP_PORT")
     |> ConfigParser.parse_port_number("INTEGRATED_TURN_TCP_PORT"),
   jwt_max_age: 24 * 3600,
-  output_base_path: System.get_env("OUTPUT_BASE_PATH", "jellyfish_output") |> Path.expand()
+  output_base_path: System.get_env("OUTPUT_BASE_PATH", "jellyfish_output") |> Path.expand(),
+  address: "#{host}:#{port}"
 
 config :opentelemetry, traces_exporter: :none
 
@@ -104,9 +124,6 @@ if config_env() == :prod do
       You can generate one by calling: mix phx.gen.secret
       """
 
-  host = System.get_env("VIRTUAL_HOST") || "example.com"
-  port = String.to_integer(System.get_env("PORT") || "4000")
-
   config :jellyfish, JellyfishWeb.Endpoint,
     url: [host: host, port: 443, scheme: "https"],
     http: [

docker-compose.yaml

@@ -0,0 +1,69 @@
+version: "3"
+
+x-jellyfish-template: &jellyfish-template
+  build: .
+  environment: &jellyfish-environment
+    ERLANG_COOKIE: "panuozzo-pollo-e-pancetta"
+    SERVER_API_TOKEN: "development"
+    SECRET_KEY_BASE: "super-secret-key"
+    VIRTUAL_HOST: "localhost"
+    NODES: "app@app1 app@app2"
+  networks:
+    - net1
+  restart: on-failure
+  healthcheck:
+    test:
+      [
+        "CMD",
+        "sh",
+        "-c",
+        "curl --fail -H \"authorization: Bearer $$SERVER_API_TOKEN\" http://localhost:$$PORT/room || exit 1"
+      ]
+    interval: 3s
+    retries: 2
+    timeout: 2s
+    start_period: 15s
+
+services:
+  test:
+    image: membraneframeworklabs/docker_membrane
+    command:
+      - sh
+      - -c
+      - |
+        cd app/ 
+        mix deps.get 
+        MIX_ENV=ci mix test --only cluster
+    volumes:
+      - .:/app
+      - /app/_build
+      - /app/deps
+    networks:
+      - net1
+    depends_on:
+      - app1
+      - app2
+
+  app1:
+    <<: *jellyfish-template
+    environment:
+      <<: *jellyfish-environment
+      RELEASE_NODE: app@app1
+      NODE_NAME: app@app1
+      PORT: 4001
+    ports:
+      - 4001:4001
+
+  app2:
+    <<: *jellyfish-template
+    environment:
+      <<: *jellyfish-environment
+      RELEASE_NODE: app@app2
+      NODE_NAME: app@app2
+      PORT: 4002
+    ports:
+      - 4002:4002
+
+networks:
+  net1:
+    driver: bridge

lib/jellyfish/application.ex

@@ -9,6 +9,8 @@ defmodule Jellyfish.Application do
   def start(_type, _args) do
     scrape_interval = Application.fetch_env!(:jellyfish, :metrics_scrape_interval)
 
+    topologies = Application.get_env(:libcluster, :topologies) || []
+
     children = [
       {Phoenix.PubSub, name: Jellyfish.PubSub},
       # Start the Telemetry supervisor
@@ -22,6 +24,13 @@ defmodule Jellyfish.Application do
       {Registry, keys: :unique, name: Jellyfish.RoomRegistry}
     ]
 
+    children =
+      if topologies == [] do
+        children
+      else
+        [{Cluster.Supervisor, [topologies, [name: Jellyfish.ClusterSupervisor]]} | children]
+      end
+
     # See https://hexdocs.pm/elixir/Supervisor.html
     # for other strategies and supported options
     opts = [strategy: :one_for_one, name: Jellyfish.Supervisor]

lib/jellyfish/room_service.ex

@@ -55,19 +55,47 @@ defmodule Jellyfish.RoomService do
   end
 
   @spec create_room(Room.max_peers(), String.t()) ::
-          {:ok, Room.t()} | {:error, :invalid_max_peers | :invalid_video_codec}
+          {:ok, Room.t(), String.t()} | {:error, :invalid_max_peers | :invalid_video_codec}
   def create_room(max_peers, video_codec) do
-    GenServer.call(__MODULE__, {:create_room, max_peers, video_codec})
+    {node_resources, failed_nodes} =
+      :rpc.multicall(Jellyfish.RoomService, :get_resource_usage, [])
+
+    if Enum.count(failed_nodes) > 0 do
+      Logger.warn(
+        "Couldn't get resource usage of the following nodes. Reason: nodes don't exist. Nodes: #{inspect(failed_nodes)}"
+      )
+    end
+
+    {min_node, _room_size} =
+      Enum.min_by(node_resources, fn {_node_name, room_num} -> room_num end)
+
+    if Enum.count(node_resources) > 1 do
+      Logger.info("Node with least used resources is #{inspect(min_node)}")
+      GenServer.call({__MODULE__, min_node}, {:create_room, max_peers, video_codec})
+    else
+      GenServer.call(__MODULE__, {:create_room, max_peers, video_codec})
+    end
   end
 
   @spec delete_room(Room.id()) :: :ok | {:error, :room_not_found}
   def delete_room(room_id) do
     GenServer.call(__MODULE__, {:delete_room, room_id})
   end
 
+  @spec get_resource_usage() :: {Node.t(), integer()}
+  def get_resource_usage() do
+    list_rooms() |> Enum.count() |> then(&{Node.self(), &1})
+  end
+
   @impl true
   def init(_opts) do
-    {:ok, %{rooms: %{}}}
+    {:ok, %{rooms: %{}}, {:continue, nil}}
+  end
+
+  @impl true
+  def handle_continue(_continue_arg, state) do
+    :ok = Phoenix.PubSub.subscribe(Jellyfish.PubSub, "jellyfishes")
+    {:noreply, state}
   end
 
   @impl true
@@ -89,7 +117,7 @@ defmodule Jellyfish.RoomService do
         {:room_created, room_id}
       )
 
-      {:reply, {:ok, room}, state}
+      {:reply, {:ok, room, Application.fetch_env!(:jellyfish, :address)}, state}
     else
       {:error, :max_peers} ->
         {:reply, {:error, :invalid_max_peers}, state}
@@ -114,6 +142,11 @@ defmodule Jellyfish.RoomService do
     {:reply, response, state}
   end
 
+  @impl true
+  def handle_info({:resources, _node_name, _resources}, state) do
+    {:noreply, state}
+  end
+
   @impl true
   def handle_info({:DOWN, _ref, :process, pid, :normal}, state) do
     {room_id, state} = pop_in(state, [:rooms, pid])

lib/jellyfish_web/controllers/room_controller.ex

@@ -69,11 +69,11 @@ defmodule JellyfishWeb.RoomController do
   def create(conn, params) do
     with max_peers <- Map.get(params, "maxPeers"),
          video_codec <- Map.get(params, "videoCodec"),
-         {:ok, room} <- RoomService.create_room(max_peers, video_codec) do
+         {:ok, room, jellyfish_address} <- RoomService.create_room(max_peers, video_codec) do
       conn
       |> put_resp_content_type("application/json")
       |> put_status(:created)
-      |> render("show.json", room: room)
+      |> render("show.json", room: room, jellyfish_address: jellyfish_address)
     else
       {:error, :invalid_max_peers} ->
         {:error, :bad_request, "maxPeers must be a number"}

lib/jellyfish_web/controllers/room_json.ex

@@ -7,6 +7,10 @@ defmodule JellyfishWeb.RoomJSON do
     %{data: rooms |> Enum.map(&data(&1))}
   end
 
+  def show(%{room: room, jellyfish_address: jellyfish_address}) do
+    %{data: data(room), jellyfish_address: jellyfish_address}
+  end
+
   def show(%{room: room}) do
     %{data: data(room)}
   end

mix.exs

@@ -23,7 +23,9 @@ defmodule Jellyfish.MixProject do
         "coveralls.detail": :test,
         "coveralls.post": :test,
         "coveralls.html": :test,
-        "coveralls.json": :test
+        "coveralls.json": :test,
+        "test.cluster": :test,
+        "test.cluster.ci": :test
       ]
     ]
   end
@@ -39,7 +41,7 @@ defmodule Jellyfish.MixProject do
   end
 
   # Specifies which paths to compile per environment.
-  defp elixirc_paths(:test), do: ["lib", "test/support"]
+  defp elixirc_paths(env) when env in [:test, :ci], do: ["lib", "test/support"]
   defp elixirc_paths(_env), do: ["lib"]
 
   defp deps do
@@ -74,16 +76,24 @@ defmodule Jellyfish.MixProject do
       {:dialyxir, ">= 0.0.0", only: :dev, runtime: false},
       {:credo, ">= 0.0.0", only: :dev, runtime: false},
 
+      # Load balancing
+      {:libcluster, "~> 3.3"},
+      {:httpoison, "~> 2.0"},
+
       # Test deps
-      {:websockex, "~> 0.4.3", only: :test, runtime: false},
-      {:excoveralls, "~> 0.15.0", only: :test, runtime: false}
+      {:websockex, "~> 0.4.3", only: [:test, :ci], runtime: false},
+      {:excoveralls, "~> 0.15.0", only: :test, runtime: false},
+      {:divo, "~> 1.3.1", only: [:test, :ci]}
     ]
   end
 
   defp aliases do
     [
       setup: ["deps.get"],
-      "api.spec": ["openapi.spec.yaml --spec JellyfishWeb.ApiSpec"]
+      "api.spec": ["openapi.spec.yaml --spec JellyfishWeb.ApiSpec"],
+      test: ["test --exclude cluster"],
+      "test.cluster": ["test --only cluster"],
+      "test.cluster.ci": ["cmd docker compose run test; docker compose stop"]
     ]
   end