fix(housekeeping): Fix migrator for replicated astarte keyspaces (#1879)

The previous logic had a two-step process for astarte keyspace creation
and setup.

Unfortunately consistency level for many operations (such as CREATE) is
ignored, so we have two options at our disposal
- re-using a single connection so that all queries are made against
  the same scylladb replica
- after each operation, run a `SELECT` with the desired consistency
  in a retry with backoff fashion

The first methods is chosen as it looks less invasive and less error
prone, but a final `SELECT` with desired consistency is preserved to
guarantee that data was replicated before returning.

As a logic shift was needed anyway, the chance was taken to remove
custom logic from the housekeeping entrypoint and moving it into the
normal supervision tree, in order to avoid future "application not
started" curses.

Signed-off-by: Francesco Noacco <francesco.noacco@secomind.com>

Author: noaccOS

apps/astarte_housekeeping/Dockerfile

@@ -46,10 +46,8 @@ RUN apt-get -qq install openssl ca-certificates
 ARG BUILD_ENV=prod
 
 COPY --from=builder /app/_build/$BUILD_ENV/rel/astarte_housekeeping .
-COPY --from=builder /app/entrypoint.sh .
 
 # Change to non-root user
 USER nobody
 
-ENTRYPOINT ["/bin/bash", "entrypoint.sh"]
-CMD ["start"]
+CMD ["./bin/astarte_housekeeping", "start"]

apps/astarte_housekeeping/entrypoint.sh

@@ -17,9 +17,10 @@
 #
 
 defmodule Astarte.Housekeeping do
+  alias Astarte.DataAccess.Config, as: DataAccessConfig
   alias Astarte.Housekeeping.Config
+  alias Astarte.Housekeeping.ReleaseTasks
   alias Astarte.RPC.Config, as: RPCConfig
-  alias Astarte.DataAccess.Config, as: DataAccessConfig
 
   use Application
 
@@ -34,7 +35,8 @@ defmodule Astarte.Housekeeping do
 
     children = [
       Astarte.HousekeepingWeb.Telemetry,
-      Astarte.Housekeeping.BackendSupervisor
+      Astarte.Housekeeping.BackendSupervisor,
+      ReleaseTasks
     ]
 
     # make amqp supervisors logs less verbose

apps/astarte_housekeeping/lib/astarte_housekeeping.ex

@@ -26,21 +26,20 @@ defmodule Astarte.Housekeeping.Migrator do
   alias Astarte.Housekeeping.Queries
   @query_timeout 60_000
 
-  def run_astarte_keyspace_migrations do
+  def run_astarte_keyspace_migrations(conn) do
     _ = Logger.info("Starting to migrate Astarte keyspace.", tag: "astarte_migration_started")
 
-    with :ok <- ensure_astarte_kv_store(),
-         {:ok, astarte_schema_version} <- get_astarte_schema_version(),
-         :ok <- migrate_astarte_keyspace_from_version(astarte_schema_version) do
+    with {:ok, astarte_schema_version} <- Queries.get_astarte_schema_version(conn),
+         :ok <- migrate_astarte_keyspace_from_version(conn, astarte_schema_version) do
       :ok
     end
   end
 
-  def run_realms_migrations do
+  def run_realms_migrations(conn) do
     _ = Logger.info("Starting to migrate Realms.", tag: "realms_migration_started")
 
-    with {:ok, realms} <- Queries.list_realms(),
-         :ok <- migrate_realms(realms) do
+    with {:ok, realms} <- Queries.list_realms(conn),
+         :ok <- migrate_realms(conn, realms) do
       :ok
     end
   end
@@ -63,107 +62,20 @@ defmodule Astarte.Housekeeping.Migrator do
     version
   end
 
-  defp migrate_realms([]) do
+  defp migrate_realms(_conn, []) do
     _ = Logger.info("Finished migrating Realms.", tag: "realms_migration_finished")
     :ok
   end
 
-  defp migrate_realms([realm | tail]) do
+  defp migrate_realms(conn, [realm | tail]) do
     _ = Logger.info("Starting to migrate realm.", tag: "realm_migration_started", realm: realm)
 
-    with {:ok, realm_astarte_schema_version} <- get_realm_astarte_schema_version(realm),
-         :ok <- migrate_realm_from_version(realm, realm_astarte_schema_version) do
-      migrate_realms(tail)
+    with {:ok, realm_astarte_schema_version} <- Queries.get_realm_schema_version(conn, realm),
+         :ok <- migrate_realm_from_version(conn, realm, realm_astarte_schema_version) do
+      migrate_realms(conn, tail)
     end
   end
 
-  defp ensure_astarte_kv_store do
-    query = """
-    SELECT table_name
-    FROM system_schema.tables
-    WHERE keyspace_name='#{CQLUtils.realm_name_to_keyspace_name("astarte", Config.astarte_instance_id!())}' AND table_name='kv_store'
-    """
-
-    consistency = Consistency.domain_model(:read)
-
-    with {:ok, %Xandra.Page{} = page} <-
-           Xandra.Cluster.execute(:xandra, query, %{}, consistency: consistency) do
-      if Enum.count(page) == 1 do
-        :ok
-      else
-        create_astarte_kv_store()
-      end
-    else
-      {:error, %Xandra.Error{} = err} ->
-        _ = Logger.warning("Database error: #{inspect(err)}.", tag: "database_error")
-        {:error, :database_error}
-
-      {:error, %Xandra.ConnectionError{} = err} ->
-        _ =
-          Logger.warning("Database connection error: #{inspect(err)}.",
-            tag: "database_connection_error"
-          )
-
-        {:error, :database_connection_error}
-    end
-  end
-
-  defp create_astarte_kv_store do
-    query = """
-    CREATE TABLE #{CQLUtils.realm_name_to_keyspace_name("astarte", Config.astarte_instance_id!())}.kv_store (
-      group varchar,
-      key varchar,
-      value blob,
-
-      PRIMARY KEY ((group), key)
-    );
-    """
-
-    case Xandra.Cluster.execute(:xandra, query, %{},
-           consistency: Consistency.domain_model(:write),
-           timeout: @query_timeout
-         ) do
-      {:ok, %Xandra.SchemaChange{}} ->
-        :ok
-
-      {:error, %Xandra.Error{} = err} ->
-        _ = Logger.warning("Database error: #{inspect(err)}.", tag: "database_error")
-        {:error, :database_error}
-
-      {:error, %Xandra.ConnectionError{} = err} ->
-        _ =
-          Logger.warning("Database connection error: #{inspect(err)}.",
-            tag: "database_connection_error"
-          )
-
-        {:error, :database_connection_error}
-    end
-  end
-
-  defp get_astarte_schema_version do
-    Xandra.Cluster.run(:xandra, fn conn ->
-      with :ok <-
-             use_keyspace(
-               conn,
-               "#{CQLUtils.realm_name_to_keyspace_name("astarte", Config.astarte_instance_id!())}"
-             ) do
-        get_keyspace_astarte_schema_version(conn)
-      end
-    end)
-  end
-
-  defp get_realm_astarte_schema_version(realm_name) do
-    Xandra.Cluster.run(:xandra, fn conn ->
-      with :ok <-
-             use_keyspace(
-               conn,
-               CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
-             ) do
-        get_keyspace_astarte_schema_version(conn)
-      end
-    end)
-  end
-
   defp use_keyspace(conn, keyspace) do
     case Xandra.execute(conn, "USE #{keyspace}", %{}, timeout: @query_timeout) do
       {:ok, %Xandra.SetKeyspace{}} ->
@@ -183,86 +95,48 @@ defmodule Astarte.Housekeeping.Migrator do
     end
   end
 
-  defp get_keyspace_astarte_schema_version(keyspace_conn) do
-    query = """
-    SELECT blobAsBigint(value)
-    FROM kv_store
-    WHERE group='astarte' AND key='schema_version'
-    """
-
-    consistency = Consistency.domain_model(:read)
-
-    with {:ok, %Xandra.Page{} = page} <-
-           Xandra.execute(keyspace_conn, query, %{}, consistency: consistency) do
-      case Enum.to_list(page) do
-        [%{"system.blobasbigint(value)" => schema_version}] ->
-          {:ok, schema_version}
-
-        [] ->
-          # If no entry is found, we assume we're at version 0
-          {:ok, 0}
-      end
-    else
-      {:error, %Xandra.Error{} = err} ->
-        _ = Logger.warning("Database error: #{inspect(err)}.", tag: "database_error")
-        {:error, :database_error}
-
-      {:error, %Xandra.ConnectionError{} = err} ->
-        _ =
-          Logger.warning("Database connection error: #{inspect(err)}.",
-            tag: "database_connection_error"
-          )
-
-        {:error, :database_connection_error}
-    end
-  end
-
-  defp migrate_astarte_keyspace_from_version(current_schema_version) do
+  defp migrate_astarte_keyspace_from_version(conn, current_schema_version) do
     _ = Logger.info("Astarte schema version is #{current_schema_version}")
 
     migrations =
       astarte_migrations_path()
       |> collect_migrations()
       |> filter_migrations(current_schema_version)
 
-    Xandra.Cluster.run(:xandra, [timeout: :infinity], fn conn ->
-      with :ok <-
-             use_keyspace(
-               conn,
-               "#{CQLUtils.realm_name_to_keyspace_name("astarte", Config.astarte_instance_id!())}"
-             ),
-           :ok <- execute_migrations(conn, migrations) do
-        _ = Logger.info("Finished migrating Astarte keyspace.", tag: "astarte_migration_finished")
+    with :ok <-
+           use_keyspace(
+             conn,
+             "#{CQLUtils.realm_name_to_keyspace_name("astarte", Config.astarte_instance_id!())}"
+           ),
+         :ok <- execute_migrations(conn, migrations) do
+      _ = Logger.info("Finished migrating Astarte keyspace.", tag: "astarte_migration_finished")
 
-        :ok
-      end
-    end)
+      :ok
+    end
   end
 
-  defp migrate_realm_from_version(realm_name, current_schema_version) do
+  defp migrate_realm_from_version(conn, realm_name, current_schema_version) do
     _ = Logger.info("Realm schema version is #{current_schema_version}", realm: realm_name)
 
     migrations =
       realm_migrations_path()
       |> collect_migrations()
       |> filter_migrations(current_schema_version)
 
-    Xandra.Cluster.run(:xandra, [timeout: :infinity], fn conn ->
-      with :ok <-
-             use_keyspace(
-               conn,
-               CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
-             ),
-           :ok <- execute_migrations(conn, migrations) do
-        _ =
-          Logger.info("Finished migrating realm.",
-            tag: "realm_migration_finished",
-            realm: realm_name
-          )
+    with :ok <-
+           use_keyspace(
+             conn,
+             CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
+           ),
+         :ok <- execute_migrations(conn, migrations) do
+      _ =
+        Logger.info("Finished migrating realm.",
+          tag: "realm_migration_finished",
+          realm: realm_name
+        )
 
-        :ok
-      end
-    end)
+      :ok
+    end
   end
 
   defp astarte_migrations_path do