Run one deployment orchestrator per deployment per cluster

Makefile

@@ -16,6 +16,10 @@ iex:
 iex-server:
 	iex -S mix phx.server
 
+num = 0
+iex-server-clustered:
+	WEB_PORT="40${num}0" iex --sname dev-$(num) --cookie nomnomnom -S mix phx.server
+
 reset-db:
 	mix ecto.reset
 

config/dev.exs

@@ -10,13 +10,15 @@ config :phoenix, :stacktrace_depth, 20
 ##
 # NervesHub Web
 #
+web_port = String.to_integer(System.get_env("WEB_PORT", "4000"))
+
 config :nerves_hub, NervesHubWeb.Endpoint,
   url: [
     host: System.get_env("WEB_HOST", "localhost"),
     scheme: System.get_env("WEB_SCHEME", "http"),
-    port: String.to_integer(System.get_env("WEB_PORT", "4000"))
+    port: web_port
   ],
-  http: [ip: {0, 0, 0, 0}, port: 4000],
+  http: [ip: {0, 0, 0, 0}, port: web_port],
   debug_errors: true,
   code_reloader: true,
   check_origin: false,
@@ -46,7 +48,7 @@ config :nerves_hub, NervesHubWeb.DeviceEndpoint,
   watchers: [],
   https: [
     ip: {0, 0, 0, 0},
-    port: 4001,
+    port: web_port + 1,
     otp_app: :nerves_hub,
     thousand_island_options: [
       transport_module: NervesHub.DeviceSSLTransport,

config/runtime.exs

@@ -35,6 +35,7 @@ config :nerves_hub,
     String.to_integer(System.get_env("DEVICE_CONNECTION_DELETE_LIMIT", "100000")),
   deployment_calculator_interval_seconds:
     String.to_integer(System.get_env("DEPLOYMENT_CALCULATOR_INTERVAL_SECONDS", "3600")),
+  deployments_orchestrator: System.get_env("DEPLOYMENTS_ORCHESTRATOR", "multi"),
   mapbox_access_token: System.get_env("MAPBOX_ACCESS_TOKEN"),
   dashboard_enabled: System.get_env("DASHBOARD_ENABLED", "false") == "true",
   extension_config: [
@@ -209,34 +210,35 @@ end
 ##
 # Database and Libcluster connection settings
 #
-if config_env() == :prod do
-  database_ssl_opts =
-    if System.get_env("DATABASE_SSL", "true") == "true" do
-      if System.get_env("DATABASE_PEM") do
-        db_hostname_charlist =
-          ~r/.*@(?<hostname>[^:\/]+)(?::\d+)?\/.*/
-          |> Regex.named_captures(System.fetch_env!("DATABASE_URL"))
-          |> Map.get("hostname")
-          |> to_charlist()
-
-        cacerts =
-          System.fetch_env!("DATABASE_PEM")
-          |> Base.decode64!()
-          |> :public_key.pem_decode()
-          |> Enum.map(fn {_, der, _} -> der end)
-
-        [
-          verify: :verify_peer,
-          cacerts: cacerts,
-          server_name_indication: db_hostname_charlist
-        ]
-      else
-        [cacerts: :public_key.cacerts_get()]
-      end
+
+database_ssl_opts =
+  if System.get_env("DATABASE_SSL", "true") == "true" do
+    if System.get_env("DATABASE_PEM") do
+      db_hostname_charlist =
+        ~r/.*@(?<hostname>[^:\/]+)(?::\d+)?\/.*/
+        |> Regex.named_captures(System.fetch_env!("DATABASE_URL"))
+        |> Map.get("hostname")
+        |> to_charlist()
+
+      cacerts =
+        System.fetch_env!("DATABASE_PEM")
+        |> Base.decode64!()
+        |> :public_key.pem_decode()
+        |> Enum.map(fn {_, der, _} -> der end)
+
+      [
+        verify: :verify_peer,
+        cacerts: cacerts,
+        server_name_indication: db_hostname_charlist
+      ]
     else
-      false
+      [cacerts: :public_key.cacerts_get()]
     end
+  else
+    false
+  end
 
+if config_env() == :prod do
   database_socket_options = if System.get_env("DATABASE_INET6") == "true", do: [:inet6], else: []
 
   config :nerves_hub, NervesHub.Repo,
@@ -259,28 +261,40 @@ if config_env() == :prod do
 
   config :nerves_hub,
     database_auto_migrator: System.get_env("DATABASE_AUTO_MIGRATOR", "true") == "true"
+end
 
-  # Libcluster is using Postgres for Node discovery
-  # The library only accepts keyword configs, so the DATABASE_URL has to be
-  # parsed and put together with the ssl pieces from above.
-  postgres_config = Ecto.Repo.Supervisor.parse_url(System.fetch_env!("DATABASE_URL"))
-
-  libcluster_db_config =
-    [port: 5432]
-    |> Keyword.merge(postgres_config)
-    |> Keyword.take([:hostname, :username, :password, :database, :port])
-    |> Keyword.merge(ssl: database_ssl_opts)
-    |> Keyword.merge(parameters: [])
-    |> Keyword.merge(channel_name: "nerves_hub_clustering")
-
-  config :libcluster,
-    topologies: [
-      postgres: [
-        strategy: LibclusterPostgres.Strategy,
-        config: libcluster_db_config
-      ]
+# Libcluster is using Postgres for Node discovery
+# The library only accepts keyword configs, so the DATABASE_URL has to be
+# parsed and put together with the ssl pieces from above.
+#
+# By using the dev database url as the default it allows us to reduce the
+# libcluster config and keep it all here.
+postgres_config =
+  Ecto.Repo.Supervisor.parse_url(
+    System.get_env("DATABASE_URL", "postgres://postgres:postgres@localhost/nerves_hub_dev")
+  )
+
+libcluster_db_config =
+  [port: 5432]
+  |> Keyword.merge(postgres_config)
+  |> Keyword.take([:hostname, :username, :password, :database, :port])
+  |> then(fn keywords ->
+    if config_env() == :prod do
+      Keyword.merge(keywords, ssl: database_ssl_opts)
+    else
+      keywords
+    end
+  end)
+  |> Keyword.merge(parameters: [])
+  |> Keyword.merge(channel_name: "nerves_hub_clustering")
+
+config :libcluster,
+  topologies: [
+    postgres: [
+      strategy: LibclusterPostgres.Strategy,
+      config: libcluster_db_config
     ]
-end
+  ]
 
 ##
 # Firmware upload backend.

lib/nerves_hub/application.ex

@@ -42,7 +42,7 @@ defmodule NervesHub.Application do
           {Task.Supervisor, name: NervesHub.TaskSupervisor},
           {Oban, Application.fetch_env!(:nerves_hub, Oban)}
         ] ++
-        deployments_supervisor(deploy_env()) ++
+        deployments_orchestrator(deploy_env()) ++
         endpoints(deploy_env())
 
     opts = [strategy: :one_for_one, name: NervesHub.Supervisor]
@@ -73,10 +73,31 @@ defmodule NervesHub.Application do
     :ok
   end
 
-  defp deployments_supervisor("test"), do: []
+  defp deployments_orchestrator("test"), do: []
 
-  defp deployments_supervisor(_) do
-    [NervesHub.Deployments.Supervisor]
+  # When running in the default `multi` mode, start a deployments orchestrator supervisor
+  # on every node. But when running in `clustered` mode, run the `ProcessHub` supervisor
+  # on the `web` or `all` nodes only.
+  defp deployments_orchestrator(_) do
+    orchestrator = Application.get_env(:nerves_hub, :deployments_orchestrator)
+    app = Application.get_env(:nerves_hub, :app)
+
+    case [orchestrator, app] do
+      ["multi", _] ->
+        [NervesHub.Deployments.Supervisor]
+
+      ["clustered", "device"] ->
+        []
+
+      ["clustered", _] ->
+        [
+          ProcessHub.child_spec(%ProcessHub{hub_id: :deployment_orchestrators}),
+          NervesHub.Deployments.Distributed.OrchestratorRegistration
+        ]
+
+      [other, _] ->
+        raise "Deployments Orchestrator '#{other}' not supported"
+    end
   end
 
   defp endpoints("test") do

lib/nerves_hub/deployments.ex

@@ -6,6 +6,7 @@
   alias NervesHub.AuditLogs.DeploymentTemplates
   alias NervesHub.AuditLogs.DeviceTemplates
   alias NervesHub.Deployments.Deployment
+  alias NervesHub.Deployments.Distributed.Orchestrator, as: DistributedOrchestrator
   alias NervesHub.Deployments.InflightDeploymentCheck
   alias NervesHub.Devices
   alias NervesHub.Devices.Device
@@ -21,6 +22,13 @@
     Repo.all(Deployment)
   end
 
+  @spec all_active() :: [Deployment.t()]
+  def all_active() do
+    Deployment
+    |> where(is_active: true)
+    |> Repo.all()
+  end
+
   @spec filter(Product.t(), map()) :: {[Product.t()], Flop.Meta.t()}
   def filter(product, opts \\ %{}) do
     opts = Map.reject(opts, fn {_key, val} -> is_nil(val) end)
@@ -156,7 +164,22 @@
     end
   end
 
-  def get_deployment!(deployment_id), do: Repo.get!(Deployment, deployment_id)
+  def get_deployment(%Deployment{id: id}), do: get_deployment(id)
+
+  def get_deployment(deployment_id) do
+    Deployment
+    |> where([d], d.id == ^deployment_id)
+    |> join(:left, [d], f in assoc(d, :firmware), as: :firmware)
+    |> preload([firmware: f], firmware: f)
+    |> Repo.one()
+    |> case do
+      nil ->
+        {:error, :not_found}
+
+      deployment ->
+        {:ok, deployment}
+    end
+  end
 
   @spec get_by_product_and_name!(Product.t(), String.t(), boolean()) :: Deployment.t()
   def get_by_product_and_name!(product, name, with_device_count \\ false)
@@ -208,12 +231,15 @@
 
   @spec delete_deployment(Deployment.t()) :: {:ok, Deployment.t()} | {:error, :not_found}
   def delete_deployment(%Deployment{id: deployment_id}) do
-    case Repo.delete(Repo.get!(Deployment, deployment_id)) do
+    Deployment
+    |> Repo.get!(deployment_id)
+    |> Repo.delete()
+    |> case do
       {:error, _changeset} ->
         {:error, :not_found}
 
       {:ok, deployment} ->
-        _ = broadcast(:monitor, "deployments/delete", %{deployment_id: deployment.id})
+        _ = deployment_deleted_event(deployment)
 
         {:ok, deployment}
     end
@@ -258,6 +284,14 @@
         _ = maybe_trigger_delta_generation(deployment, changeset)
         :ok = broadcast(deployment, "deployments/update")
 
+        if Map.has_key?(changeset.changes, :is_active) do
+          if deployment.is_active do
+            deployment_activated_event(deployment)
+          else
+            deployment_deactivated_event(deployment)
+          end
+        end
+
         {:ok, deployment}
 
       {:error, changeset} ->
@@ -333,7 +367,7 @@
 
     case Repo.insert(changeset) do
       {:ok, deployment} ->
-        _ = broadcast(:monitor, "deployments/new", %{deployment_id: deployment.id})
+        deployment_created_event(deployment)
 
         {:ok, deployment}
 
@@ -346,31 +380,30 @@
   def broadcast(deployment, event, payload \\ %{})
 
   def broadcast(:none, event, payload) do
-    message = %Phoenix.Socket.Broadcast{
-      topic: "deployment:none",
-      event: event,
-      payload: payload
-    }
-
-    Phoenix.PubSub.broadcast(NervesHub.PubSub, "deployment:none", message)
+    Phoenix.Channel.Server.broadcast(
+      NervesHub.PubSub,
+      "deployment:none",
+      event,
+      payload
+    )
   end
 
   def broadcast(:monitor, event, payload) do
-    Phoenix.PubSub.broadcast(
+    Phoenix.Channel.Server.broadcast(
       NervesHub.PubSub,
       "deployment:monitor",
-      %Phoenix.Socket.Broadcast{event: event, payload: payload}
+      event,
+      payload
     )
   end
 
   def broadcast(%Deployment{id: id}, event, payload) do
-    message = %Phoenix.Socket.Broadcast{
-      topic: "deployment:#{id}",
-      event: event,
-      payload: payload
-    }
-
-    Phoenix.PubSub.broadcast(NervesHub.PubSub, "deployment:#{id}", message)
+    Phoenix.Channel.Server.broadcast(
+      NervesHub.PubSub,
+      "deployment:#{id}",
+      event,
+      payload
+    )
   end
 
   @doc """
@@ -555,4 +588,48 @@
     |> where([d, firmware: f], f.architecture == ^device.firmware_metadata.architecture)
     |> Repo.all()
   end
+
+  def deployment_created_event(deployment) do
+    _ =
+      case Application.get_env(:nerves_hub, :deployments_orchestrator) do
+        "multi" -> _ = broadcast(:monitor, "deployments/new", %{deployment_id: deployment.id})
+        "clustered" -> DistributedOrchestrator.start_orchestrator(deployment)
+        other -> raise "Deployments Orchestrator '#{other}' not supported"
+      end
+
+    :ok
+  end
+
+  def deployment_activated_event(deployment) do
+    _ =
+      case Application.get_env(:nerves_hub, :deployments_orchestrator) do
+        "multi" -> :ok
+        "clustered" -> DistributedOrchestrator.start_orchestrator(deployment)
+        other -> raise "Deployments Orchestrator '#{other}' not supported"
+      end
+
+    :ok
+  end
+
+  def deployment_deactivated_event(deployment) do
+    _ =
+      case Application.get_env(:nerves_hub, :deployments_orchestrator) do
+        "multi" -> :ok
+        "clustered" -> DistributedOrchestrator.stop_orchestrator(deployment)
+        other -> raise "Deployments Orchestrator '#{other}' not supported"
+      end
+
+    :ok
+  end
+
+  def deployment_deleted_event(deployment) do
+    _ =
+      case Application.get_env(:nerves_hub, :deployments_orchestrator) do
+        "multi" -> broadcast(:monitor, "deployments/delete", %{deployment_id: deployment.id})
+        "clustered" -> broadcast(deployment, "deleted")
+        other -> raise "Deployments Orchestrator '#{other}' not supported"
+      end
+
+    :ok
+  end
 end