Merge pull request #11042 from neinteractiveliterature/nbudin/issue11033

Scale up background workers in addition to web workers

Author: nbudin

Gemfile

@@ -189,3 +189,5 @@ end
 
 gem "sentry-ruby", "~> 5.17"
 gem "sentry-rails", "~> 5.17"
+
+gem "openssl", "~> 3.3"

Gemfile.lock

@@ -187,7 +187,7 @@ GEM
     crass (1.0.6)
     csv (3.3.5)
     dalli (3.2.8)
-    date (3.4.1)
+    date (3.5.0)
     dead_end (4.0.0)
     debug (1.11.0)
       irb (~> 1.10)
@@ -240,7 +240,7 @@ GEM
       dotenv (= 3.1.8)
       railties (>= 6.1)
     drb (2.2.3)
-    erb (5.0.3)
+    erb (5.1.3)
     erubi (1.13.1)
     erubis (2.7.0)
     eu_central_bank (1.7.0)
@@ -398,6 +398,7 @@ GEM
     oj (3.16.12)
       bigdecimal (>= 3.0)
       ostruct (>= 0.2)
+    openssl (3.3.2)
     optimist (3.1.0)
     orm_adapter (0.5.0)
     ostruct (0.6.3)
@@ -502,9 +503,10 @@ GEM
       msgpack (>= 0.4.3)
       optimist (>= 3.0.0)
     rbtree (0.4.6)
-    rdoc (6.14.2)
+    rdoc (6.15.1)
       erb
       psych (>= 4.0.0)
+      tsort
     recaptcha (5.21.1)
     redcarpet (3.6.1)
     regexp_parser (2.11.3)
@@ -717,6 +719,7 @@ DEPENDENCIES
   money-rails
   mysql2 (~> 0.5.3)
   oj (~> 3.16.0)
+  openssl (~> 3.3)
   parallel
   pg
   pg_search

app/services/autoscale_servers_service.rb

@@ -92,23 +92,45 @@ def self.scaling_target_for(time)
     scaling_targets.max.clamp(MIN_INSTANCES, MAX_INSTANCES).ceil
   end
 
-  private
+  def self.worker_scaling_target_for(time)
+    web_scaling_target = scaling_target_for(time)
 
-  def inner_call
-    adapter = HostingServiceAdapters::Base.find_adapter
-    return unless adapter
+    if web_scaling_target == MIN_INSTANCES
+      1
+    else
+      2
+    end
+  end
 
-    scaling_target = self.class.scaling_target_for(Time.now)
-    current_instance_count = adapter.fetch_instance_count
+  def self.worker_instance_type_for(time)
+    web_scaling_target = scaling_target_for(time)
 
-    if current_instance_count == scaling_target
-      Rails.logger.info "Currently running #{current_instance_count} #{"instance".pluralize(current_instance_count)}; \
-no autoscaling needed"
+    if web_scaling_target == MIN_INSTANCES
+      :small
     else
-      Rails.logger.info "Autoscaling to #{scaling_target} instances"
-      adapter.update_instance_count(scaling_target)
+      :large
     end
+  end
+
+  private
+
+  def inner_call
+    now = Time.now
+    web_scaling_target = self.class.scaling_target_for(now)
+    worker_scaling_target = self.class.worker_scaling_target_for(now)
+    worker_instance_type = self.class.worker_instance_type_for(now)
+
+    adapter.apply_instance_counts(
+      [
+        { group: :web, type: :small, count: web_scaling_target },
+        { group: :worker, type: worker_instance_type, count: worker_scaling_target }
+      ]
+    )
 
     success
   end
+
+  def adapter
+    @adapter ||= HostingServiceAdapters.find_adapter
+  end
 end

app/services/hosting_service_adapters.rb

@@ -0,0 +1,7 @@
+module HostingServiceAdapters
+  ADAPTER_CLASSES = [HostingServiceAdapters::Fly, HostingServiceAdapters::Heroku]
+
+  def self.find_adapter
+    ADAPTER_CLASSES.map(&:new).find(&:applicable?)
+  end
+end

app/services/hosting_service_adapters/base.rb

@@ -1,19 +1,28 @@
 class HostingServiceAdapters::Base
-  ADAPTER_CLASSES = [HostingServiceAdapters::Fly, HostingServiceAdapters::Heroku, HostingServiceAdapters::Render]
-
-  def self.find_adapter
-    ADAPTER_CLASSES.map(&:new).find(&:applicable?)
-  end
-
   def applicable?
     raise NotImplementedError, "HostingServiceAdapters::Base subclasses must implement #applicable?"
   end
 
-  def fetch_instance_count
-    raise NotImplementedError, "HostingServiceAdapters::Base subclasses must implement #fetch_instance_count"
+  def instances_state
+    raise NotImplementedError, "HostingServiceAdapters::Base subclasses must implement #instances_state"
+  end
+
+  def apply_instance_counts(target_counts)
+    target_groups = Set.new(target_counts.map { |target| { group: target[:group], type: target[:type] } })
+
+    target_counts.each do |target|
+      update_instance_group(group: target[:group], type: target[:type], count: target[:count])
+    end
+
+    instances_state.counts.each do |row|
+      next if target_groups.include?(row.slice(:group, :type))
+      update_instance_group(group: row[:group], type: row[:type], count: 0)
+    end
+
+    instances_state
   end
 
-  def update_instance_count(instance_count)
-    raise NotImplementedError, "HostingServiceAdapters::Base subclasses must implement #update_instance_count"
+  def update_instance_group(group:, type:, count:)
+    raise NotImplementedError, "HostingServiceAdapters::Base subclasses must implement #update_instance_group"
   end
 end

app/services/hosting_service_adapters/fly.rb

@@ -1,74 +1,167 @@
-require "fly.io-rails/machines"
-
 class HostingServiceAdapters::Fly < HostingServiceAdapters::Base
+  SMALL_GUEST = { "cpu_kind" => "shared", "cpus" => 1, "memory_mb" => 512 }
+  LARGE_GUEST = { "cpu_kind" => "performance", "cpus" => 1, "memory_mb" => 2048 }
+
+  class Machine < HostingServiceAdapters::Instance
+    attr_reader :id, :state, :config, :region, :instance_id, :name
+
+    def initialize(id:, instance_id:, state:, config:, region:, name:, **args)
+      super(**args)
+      @id = id
+      @instance_id = instance_id
+      @state = state
+      @config = config
+      @region = region
+      @name = name
+    end
+  end
+
   def applicable?
     ENV["FLY_APP_NAME"].present? && ENV["FLY_API_TOKEN"].present?
   end
 
-  def fetch_instance_count
-    web_machines.count
+  def instances_state
+    @instances_state ||=
+      HostingServiceAdapters::InstancesState.new(
+        instances: current_machines.map { |machine| instance_for_machine(machine) }
+      )
+  end
+
+  def invalidate_state
+    @current_machines = nil
+    @instances_state = nil
+  end
+
+  def force_refresh_state
+    invalidate_state
+    instances_state
   end
 
-  def update_instance_count(instance_count)
-    raise "Instance count must be at least 1" if instance_count < 1
+  def update_instance_group(group:, type:, count:)
+    current_instances = instances_state.find_instances(group:, type:)
 
-    if instance_count > fetch_instance_count
-      scale_up(instance_count - fetch_instance_count)
-    elsif instance_count < fetch_instance_count
-      scale_down(fetch_instance_count - instance_count)
+    if current_instances.size < count
+      create_machines(group:, type:, count: count - current_instances.size)
+      invalidate_state
+    elsif current_instances.size > count
+      destroy_machines(current_instances.first(current_instances.size - count))
+      invalidate_state
     end
   end
 
   private
 
-  def scale_up(amount)
-    config = web_machines.first[:config]
-    created_machines =
-      Array
-        .new(amount)
-        .map do
-          Fly::Machines.create_and_start_machine(
-            ENV.fetch("FLY_APP_NAME"),
-            { config: config, region: web_machines.first[:region] }
-          )
+  def machines_api_base
+    @machines_api_base ||= ENV["FLY_PRIVATE_IP"].present? ? "http://_api.internal:4280" : "https://api.machines.dev"
+  end
+
+  def machines_api
+    @machines_api ||=
+      Faraday.new(url: machines_api_base) do |builder|
+        builder.request :authorization, "Bearer", -> { ENV.fetch("FLY_API_TOKEN") }
+        builder.request :json
+        builder.response :json
+        builder.response :raise_error
+      end
+  end
+
+  def current_machines
+    @current_machines ||= machines_api.get("/v1/apps/#{ENV.fetch("FLY_APP_NAME")}/machines").body
+  end
+
+  def instance_for_machine(machine) # rubocop:disable Metrics/MethodLength
+    guest = machine.dig("config", "guest")
+    process_group = machine.dig("config", "metadata", "fly_process_group")
+
+    Machine.new(
+      id: machine["id"],
+      instance_id: machine["instance_id"],
+      name: machine["name"],
+      state: machine["state"],
+      config: machine["config"],
+      region: machine["region"],
+      group:
+        case process_group
+        when "web"
+          :web
+        when "shoryuken"
+          :worker
+        else
+          :other
+        end,
+      type:
+        case guest
+        when SMALL_GUEST
+          :small
+        when LARGE_GUEST
+          :large
+        else
+          :other
         end
-    Rails.logger.info "Created and started Fly machines: #{created_machines.pluck(:name).join(", ")}"
-    Rails.logger.info "Waiting for all machines to start"
-    created_machines.each do |machine|
-      Fly::Machines.wait_for_machine(ENV.fetch("FLY_APP_NAME"), machine[:id], status: "started")
-    end
-    created_machines
+    )
   end
 
-  def scale_down(amount)
-    to_destroy = web_machines.first(amount)
-    to_stop = to_destroy.filter { |machine| machine[:state] != "stopped" }
-    to_stop.each do |machine|
-      Rails.logger.info "Stopping Fly machine #{machine[:name]}"
-      Fly::Machines.stop_machine(ENV.fetch("FLY_APP_NAME"), machine[:id])
+  def guest_config_for_type(type)
+    case type
+    when :small
+      SMALL_GUEST
+    when :large
+      LARGE_GUEST
     end
-    unless to_stop.empty?
-      Rails.logger.info "Waiting for all machines to stop"
-      to_stop.each do |machine|
-        Fly::Machines.wait_for_machine(ENV.fetch("FLY_APP_NAME"), machine[:id], status: "stopped")
-      end
+  end
+
+  def fly_process_group_for_group(group)
+    case group
+    when :web
+      "web"
+    when :worker
+      "shoryuken"
     end
+  end
+
+  def machine_create_body(group:, type:)
+    template_machine = instances_state.find_instances(group:).first
+    config_template =
+      template_machine.config.merge(
+        "guest" => guest_config_for_type(type),
+        "metadata" => {
+          "fly_process_group" => fly_process_group_for_group(group)
+        }
+      )
+    { config: config_template, region: template_machine.region }
+  end
+
+  def create_machines(group:, type:, count:)
+    create_body = machine_create_body(group:, type:)
 
-    to_destroy.each do |machine|
-      Rails.logger.info "Deleting Fly machine #{machine[:name]}"
-      Fly::Machines.delete_machine(ENV.fetch("FLY_APP_NAME"), machine[:id])
+    Rails.logger.info "Creating #{count} #{type} #{group} machine(s)"
+    created_machines =
+      Array.new(count) { |_n| machines_api.post("/v1/apps/#{ENV.fetch("FLY_APP_NAME")}/machines", create_body).body }
+    Rails.logger.info "Created and started Fly machines: #{created_machines.pluck("name").join(", ")}"
+
+    Rails.logger.info "Waiting for all machines to start"
+    created_machines.each do |machine|
+      machines_api.get("/v1/apps/#{ENV.fetch("FLY_APP_NAME")}/machines/#{machine["id"]}/wait?state=started")
     end
 
-    to_destroy
+    created_machines
   end
 
-  def web_machines
-    @web_machines ||=
-      begin
-        Fly::Machines.fly_api_hostname!
-        Fly::Machines
-          .list_machines(ENV.fetch("FLY_APP_NAME"), nil)
-          .filter { |machine| machine[:config][:metadata][:fly_process_group] == "web" }
-      end
+  def destroy_machines(instances)
+    instances.each do |instance|
+      Rails.logger.info "Stopping Fly machine #{instance.name}"
+      machines_api.post("/v1/apps/#{ENV.fetch("FLY_APP_NAME")}/machines/#{instance.id}/stop")
+    end
+
+    Rails.logger.info "Waiting for all machines to stop"
+    instances.each do |instance|
+      machines_api.get(
+        "/v1/apps/#{ENV.fetch("FLY_APP_NAME")}/machines/#{instance.id}/wait?\
+state=stopped&instance_id=#{instance.instance_id}"
+      )
+
+      Rails.logger.info "Destroying Fly machine #{instance.name}"
+      machines_api.delete("/v1/apps/#{ENV.fetch("FLY_APP_NAME")}/machines/#{instance.id}")
+    end
   end
 end