The DynamicScaler examines queue throughput and issues commands to horizontally scale cloud infrastructure to optimize processing. With auto-scaling you can spin up additional nodes during high traffic events, and pare down to a single node during a lull. Beyond optimizing throughput, scaling may save money in environments with little to no usage at off-peak times, e.g. staging.

Horizontal scaling is applied at the node level, not the queue level, so you can distribute processing over more physical hardware.

• Predictive Scaling — The optimal scale is calculated by predicting the future size of a queue based on recent trends. Multiple samples are then used to prevent overreacting to changes in queue depth or throughput. You provide an acceptable range of nodes and auto-scaling takes care of the rest.

• Multi-Cloud — Cloud integration is provided by a simple, flexible, behaviour that you implement for your specific environment and configure for each scaler.

• Queue Filtering — By default, all queues are considered for scale calculations. However, you can restrict calculations to one or more critical queues.

• Multiple Scalers — Some systems may restrict work to specific node types, e.g. generating exports or processing videos. Other hybrid systems may straddle multiple clouds. In either case, you can configure multiple independent scalers driven by distinct queues.

• Non Linear — An optional step parameter allows you to conservatively scale up or down one node at a time, or optimize for responsiveness and jump from the min to the max in a single scaling period.

• Prevent Thrashing — A cooldown period skips scaling when there was recent scale activity to prevent unnecessarily scaling nodes up or down. Nodes may take several minutes to start within an environment, so the default cooldown period is 2 minutes.

Using and Configuring

With a cloud module in hand you're ready to add the DynamicScaler plugin to your Oban config:

config :my_app, Oban,
  plugins: [
    {Oban.Pro.Plugins.DynamicScaler, ...}
  ]

Then, add a scaler with a range to define the minimum and maximum nodes, and your cloud strategy:

{DynamicScaler, scalers: [range: 1..5, cloud: MyApp.Cloud]}

Now, every minute, DynamicScaler will calculate the optimal number of nodes for your queue's throughput and issue scaling commands accordingly.

Configuring Scalers

The required scalers option takes one or more scaler options to run several independent scalers. Each scaler accepts:

• :cloud — A module or {module, options} tuple that interacts with the external cloud during scale events. Required.

• :range — The range of compute units to scale between. For example, 1..3 declares a minimum of 1 node and a maximum of 3. The minimum must be 0 or more, and the maximum must be 1 or at least match the minimum. Required.

• :cooldown — The minimum time between scaling events. Defaults to 120 seconds.

• :lookback — The historic time to check queues. Defaults to 60 seconds.

• :queues — Either :all or a list of queues to consider when measuring throughput and backlog.

• :step — Either :none or the maximum nodes to scale up or down at once. Defaults to :none.

Scaler Examples

Filter throughput queries to the :media queue:

scalers: [queues: :media, range: 1..3, cloud: MyApp.Cloud]

Filter throughput queries to both :audio and :video queues:

scalers: [queues: [:audio, :video], range: 1..3, cloud: MyApp.Cloud]

Configure scalers driven by different queues (note, queues may not overlap):

scalers: [
  [queues: :audio, range: 0..2, cloud: {MyApp.Cloud, asg: "my-audio-asg"}],
  [queues: :video, range: 0..5, cloud: {MyApp.Cloud, asg: "my-video-asg"}]
]

Limit scaling to one node up or down at a time:

scalers: [range: 1..3, step: 1, cloud: MyApp.Cloud]

Wait at least 5 minutes (300 seconds) between scaling events:

scalers: [range: 1..3, cloud: MyApp.Cloud, cooldown: 300]

Increase the period used to calculate historic throughput to 90 seconds:

scalers: [range: 1..3, cloud: MyApp.Cloud, lookback: 90]

Cloud Modules

There are a lot of hosting platforms, aka "clouds" out there and we can't support them all! Even with optional dependencies, it would be a mess of libraries that may not agree with your application decisions. Instead, the Oban.Pro.Cloud behaviour defines two simple callbacks, and integrating with platforms typically takes a single HTTP query or library call.

Rather than implement a cloud from scratch, copy-and-paste a recipe for a popular platform and use it as-is, or as a starting point for your own:

• Fly (scale machine count through the Fly Machines API)
• GCP (resize a managed instance group on Google Compute Engine)
• Kubernetes (patch a deployment's replicas through the API server)

Let us know if a recipe for your platform is missing and you'd like assistance. Otherwise, follow the guide below to write your own integration.

Writing Cloud Modules

Cloud callback modules must define an init/1 function to prepare configuration at runtime, and a scale/2 callback called with the desired number of nodes and the prepared configuration. The scale/2 callback returns {:ok, conf} so the scaler can track the response, or {:error, reason} when scaling fails.

Most integrations are a single HTTP call. This skeleton prepares options into a config map, then resizes by posting the desired count to an API:

defmodule MyApp.Cloud do
  @behaviour Oban.Pro.Cloud

  @impl Oban.Pro.Cloud
  def init(opts), do: Map.new(opts)

  @impl Oban.Pro.Cloud
  def scale(desired, conf) do
    case Req.post(conf.url, json: %{count: desired}) do
      {:ok, %{status: status}} when status in 200..299 -> {:ok, conf}
      {:ok, response} -> {:error, response}
      {:error, reason} -> {:error, reason}
    end
  end
end

You'd then use your cloud module as a scaler option:

{DynamicScaler, scalers: [range: 1..3, cloud: {MyApp.Cloud, url: "https://..."}]}

Clouds can also pull from the application or system environment to build configuration. If your module pulls from the environment exclusively, then you can pass the module name rather than a tuple:

{DynamicScaler, scalers: [range: 1..3, cloud: MyApp.Cloud]}

Optimizing Throughput Queries

While the scaler's throughput queries are optimized for a standard load, high throughput queues, or systems that retain a large volume of jobs, may benefit from an additional index that aids calculating throughput. Use the following migration to add an index if you find that scaling queries are too slow or timing out:

@disable_ddl_transaction true
@disable_migration_lock true

def change do
  create_if_not_exists index(
    :oban_jobs,
    [:state, :queue, :attempted_at, :attempted_by],
    concurrently: true,
    where: "attempted_at IS NOT NULL",
    prefix: "public"
  )
end

Alternatively, you can raise the timeout used for scaler inspection queries, which defaults to 15_000 milliseconds:

{DynamicScaler, timeout: :timer.seconds(30), scalers: ...}

Instrumenting with Telemetry

The DynamicScaler plugin adds the following metadata to the [:oban, :plugin, :stop] event:

• :scaler - details of the active scaler config with recent scaling values
• :skipped - the reason scaling is skipped, either :recently_scaled or :already_scaled
• :error — the reason returned from scale/2 when scaling fails

When multiple scalers are configured one event is emitted for each scaler.