add ability to bulk migrate agents to new versions (#502)

config/prod.exs

@@ -33,6 +33,7 @@ config :console, Console.Cron,
     {"0 0 1-31/2 * *", {Console.Deployments.Cron, :backfill_deprecations, []}},
     {"20 * * * *", {Console.Deployments.Cron, :backfill_global_services, []}},
     {"45 * * * *", {Console.Deployments.Cron, :migrate_kas, []}},
+    {"30 * * * *", {Console.Deployments.Cron, :migrate_agents, []}},
     {"@daily", {Console.Deployments.Cron, :rotate_deploy_tokens, []}},
     {"@daily", {Console.Deployments.Cron, :prune_revisions, []}},
     {"@daily", {Console.Cron.Jobs, :prune_notifications, []}},

lib/console/deployments/cron.ex

@@ -2,7 +2,7 @@ defmodule Console.Deployments.Cron do
   use Console.Services.Base
   alias Console.Deployments.{Services, Clusters, Global}
   alias Console.Services.Users
-  alias Console.Schema.{Cluster, Service, ServiceComponent, GlobalService, PipelineStage, PipelinePromotion}
+  alias Console.Schema.{Cluster, Service, ServiceComponent, GlobalService, PipelineStage, PipelinePromotion, AgentMigration}
   alias Console.Deployments.Pipelines.Discovery
 
   require Logger
@@ -137,6 +137,26 @@ defmodule Console.Deployments.Cron do
     |> Stream.run()
   end
 
+  def migrate_agents() do
+    AgentMigration.incomplete()
+    |> Repo.all()
+    |> Stream.each(fn migration ->
+      Cluster.installable()
+      |> Cluster.stream()
+      |> Cluster.preloaded()
+      |> Repo.stream(method: :keyset)
+      |> Stream.each(fn cluster ->
+        Logger.info "installing operator on #{cluster.id}"
+        Clusters.install(cluster)
+      end)
+      |> Stream.run()
+      AgentMigration.changeset(migration, %{completed: true})
+      |> Repo.update()
+      Logger.info "migration #{migration.id} completed"
+    end)
+    |> Stream.run()
+  end
+
   def scan_pending_promotions() do
     PipelinePromotion.pending()
     |> Repo.stream(method: :keyset)

lib/console/schema/agent_migration.ex

@@ -0,0 +1,16 @@
+defmodule Console.Schema.AgentMigration do
+  use Piazza.Ecto.Schema
+
+  schema "agent_migrations" do
+    field :completed, :boolean, default: false
+
+    timestamps()
+  end
+
+  def incomplete(query \\ __MODULE__), do: from(am in query, where: not am.completed)
+
+  def changeset(model, attrs \\ %{}) do
+    model
+    |> cast(attrs, [:completed])
+  end
+end

lib/console/schema/cluster.ex

@@ -196,6 +196,10 @@ defmodule Console.Schema.Cluster do
     from(c in query, where: not is_nil(c.deleted_at))
   end
 
+  def installable(query \\ __MODULE__) do
+    from(c in query, where: (not is_nil(c.provider_id) or c.self) and is_nil(c.deleted_at))
+  end
+
   def uninstalled(query \\ __MODULE__) do
     from(c in query, where: is_nil(c.pinged_at) and (not is_nil(c.provider_id) or c.self) and is_nil(c.deleted_at))
   end

priv/repo/migrations/20231114173755_agent_migrations.exs

@@ -0,0 +1,12 @@
+defmodule Console.Repo.Migrations.AgentMigrations do
+  use Ecto.Migration
+
+  def change do
+    create table(:agent_migrations, primary_key: false) do
+      add :id, :uuid, primary_key: true
+      add :completed, :boolean, default: false
+
+      timestamps()
+    end
+  end
+end

priv/repo/seeds/05_agent.exs

@@ -0,0 +1,9 @@
+import Botanist
+
+alias Console.Schema.AgentMigration
+
+seed do
+  %AgentMigration{}
+  |> AgentMigration.changeset()
+  |> Console.Repo.insert()
+end