refactor: move AMQP events producer and supervisor into astarte events library (#1522)

Extracted the AMQP events producer and its supervision tree into a separate
library to improve modularity and reuse across services. The original files
in astarte_data_updater_plant have been removed and all references updated
to point to the external library.

Signed-off-by: Eddy Babetto <eddy.babetto@secomind.com>

Author: eddbbt

apps/astarte_data_updater_plant/config/config.exs

@@ -21,4 +21,6 @@ config :astarte_data_updater_plant, ecto_repos: [Astarte.DataAccess.Repo]
 
 config :astarte_data_updater_plant, Astarte.DataAccess.Repo, []
 
+config :astarte_events, :connection_backoff, 10_000
+
 import_config "#{config_env()}.exs"

apps/astarte_data_updater_plant/config/test.exs

@@ -12,3 +12,5 @@ config :astarte_data_updater_plant, :astarte_instance_id, "test"
 config :astarte_data_updater_plant,
        :vernemq_plugin_rpc_client,
        Astarte.DataUpdaterPlant.RPC.VMQPlugin.ClientMock
+
+config :astarte_events, :connection_backoff, 0

apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/data_pipeline_supervisor.ex

@@ -19,7 +19,6 @@
 defmodule Astarte.DataUpdaterPlant.DataPipelineSupervisor do
   use Supervisor
 
-  alias Astarte.DataUpdaterPlant.ProducersSupervisor
   alias Astarte.DataUpdaterPlant.ConsumersSupervisor
 
   def start_link(init_arg) do
@@ -50,7 +49,6 @@ defmodule Astarte.DataUpdaterPlant.DataPipelineSupervisor do
          members: :auto,
          distribution_strategy: Horde.UniformDistribution
        ]},
-      ProducersSupervisor,
       ConsumersSupervisor,
       Astarte.DataUpdaterPlant.RPC.Supervisor
     ]

apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/producers_supervisor.ex

@@ -19,7 +19,7 @@
 defmodule Astarte.DataUpdaterPlant.TriggersHandler do
   import Bitwise, only: [<<<: 2]
   require Logger
-  alias Astarte.DataUpdaterPlant.AMQPEventsProducer
+  alias Astarte.Events.AMQPEventsProducer
   alias Astarte.DataUpdaterPlant.Config
 
   @moduledoc """

apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/triggers_handler.ex

@@ -22,7 +22,7 @@ defmodule Astarte.DataUpdaterPlant.AMQPEventsProducerTest do
   use Mimic
 
   alias AMQP.Channel
-  alias Astarte.DataUpdaterPlant.AMQPEventsProducer
+  alias Astarte.Events.AMQPEventsProducer
 
   @tag :regression
   test "events producer reconnects in case of error" do

apps/astarte_data_updater_plant/test/astarte_data_updater_plant/amqp_events_producer_test.exs

@@ -18,11 +18,23 @@
 
 defmodule Astarte.Events.Application do
   use Application
+  alias Astarte.Events.AMQPEventsProducer
+  alias Astarte.Events.AMQPTriggers.VHostSupervisor
+  alias Astarte.Events.Config
 
   def start(_type, _args) do
+    events_pool =
+      Supervisor.child_spec(
+        {ExRabbitPool.PoolSupervisor,
+         rabbitmq_config: Config.amqp_options!(), connection_pools: [Config.events_pool_config!()]},
+        id: :events_producer_pool
+      )
+
     children = [
+      events_pool,
+      AMQPEventsProducer,
       {Registry, keys: :unique, name: Astarte.Events.AMQPTriggers.Registry},
-      Astarte.Events.AMQPTriggers.VHostSupervisor
+      VHostSupervisor
     ]
 
     opts = [strategy: :rest_for_one, name: Astarte.Events.Supervisor]

libs/astarte_events/lib/application.ex

@@ -16,17 +16,15 @@
 # limitations under the License.
 #
 
-defmodule Astarte.DataUpdaterPlant.AMQPEventsProducer do
+defmodule Astarte.Events.AMQPEventsProducer do
   require Logger
   use GenServer
 
-  alias Astarte.DataUpdaterPlant.Config
+  alias Astarte.Events.Config
   alias AMQP.Channel
 
-  @connection_backoff if Mix.env() == :test, do: 0, else: 10000
-  @adapter Config.amqp_adapter!()
-
-  # API
+  @connection_backoff Application.compile_env(:astarte_events, :connection_backoff, 10_000)
+  alias ExRabbitPool.RabbitMQ
 
   def start_link(args \\ []) do
     GenServer.start_link(__MODULE__, args, name: __MODULE__)
@@ -57,15 +55,15 @@ defmodule Astarte.DataUpdaterPlant.AMQPEventsProducer do
 
   @impl true
   def handle_call({:publish, exchange, routing_key, payload, opts}, _from, chan) do
-    reply = @adapter.publish(chan, exchange, routing_key, payload, opts)
+    reply = RabbitMQ.publish(chan, exchange, routing_key, payload, opts)
 
     {:reply, reply, chan}
   end
 
   def handle_call({:declare_exchange, exchange}, _from, chan) do
     # TODO: we need to decide who is responsible of deleting the exchange once it is
     # no longer needed
-    reply = @adapter.declare_exchange(chan, exchange, type: :direct, durable: true)
+    reply = RabbitMQ.declare_exchange(chan, exchange, type: :direct, durable: true)
 
     {:reply, reply, chan}
   end
@@ -74,7 +72,7 @@ defmodule Astarte.DataUpdaterPlant.AMQPEventsProducer do
   def handle_info({:DOWN, _, :process, _pid, reason}, _state) do
     # Track channel crash
     :telemetry.execute(
-      [:astarte, :data_updater_plant, :amqp_events_producer, :channel_crash],
+      [:astarte, :astarte_events, :amqp_events_producer, :channel_crash],
       %{},
       %{reason: inspect(reason)}
     )
@@ -139,7 +137,7 @@ defmodule Astarte.DataUpdaterPlant.AMQPEventsProducer do
 
   defp declare_default_events_exchange(channel, conn) do
     with {:error, reason} <-
-           @adapter.declare_exchange(channel, Config.events_exchange_name!(),
+           RabbitMQ.declare_exchange(channel, Config.amqp_events_exchange_name!(),
              type: :direct,
              durable: true
            ) do

…ta_updater_plant/amqp_events_producer.ex …b/astarte_events/amqp_events_producer.exapps/astarte_data_updater_plant/lib/astarte_data_updater_plant/amqp_events_producer.ex renamed to libs/astarte_events/lib/astarte_events/amqp_events_producer.ex apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/amqp_events_producer.ex renamed to libs/astarte_events/lib/astarte_events/amqp_events_producer.ex

@@ -117,10 +117,16 @@ defmodule Astarte.Events.Config do
 
   @envdoc "The total number of data queues in all the Astarte cluster."
   app_env :data_queue_total_count, :astarte_events, :amqp_data_queue_total_count,
-    os_env: "ASTARTE_EVENTS_AMQP_DATA_QUEUE_TOTAL_COUNT",
+    os_env: "ASTARTE_EVENTS_PRODUCER_AMQP_DATA_QUEUE_TOTAL_COUNT",
     type: :integer,
     default: 128
 
+  @envdoc "The exchange used by the AMQP producer to publish events."
+  app_env :amqp_events_exchange_name, :astarte_events, :amqp_events_exchange_name,
+    os_env: "ASTARTE_EVENTS_PRODUCER_AMQP_EVENTS_EXCHANGE_NAME",
+    type: :binary,
+    default: "astarte_events"
+
   # Since we have one channel per queue, this is not configurable
   def amqp_channels_per_connection_number!() do
     ceil(data_queue_total_count!() / amqp_connection_number!())
@@ -169,6 +175,18 @@ defmodule Astarte.Events.Config do
     end
   end
 
+  # Since we have only one producer, this is not configurable
+  def events_connection_number!(), do: 1
+
+  def events_pool_config!() do
+    [
+      name: {:local, :events_producer_pool},
+      worker_module: ExRabbitPool.Worker.RabbitConnection,
+      size: events_connection_number!(),
+      max_overflow: 0
+    ]
+  end
+
   defdelegate astarte_instance_id!, to: DataAccessConfig
   defdelegate astarte_instance_id, to: DataAccessConfig
 end