[SPEC] Scale-out w/Azure Storage provider

[cgillum]

INTRODUCTION

The default Durable Functions configuration supports massive scale-out across multiple nodes in all supported hosting environments, including the "serverless" Azure Functions Consumption plan. By default, all Durable Functions deployments in Azure use the Azure Storage provider, which is a storage provider extension for the Durable Task Framework (DTFx). This provider implements the scale-out mechanisms used by Durable Functions.

TERMINOLOGY

• Durable Task Framework (DTFx): The underlying framework which is used to implement the Durable Functions triggers.
• DTFx Client: The output binding used to create, query, and manage orchestration instances. Note that in some cases the client can be invoked directly via an HTTP request to the binding extension.
• Storage Provider: A DTFx extension for implementing durable storage and messaging on top of Azure Storage.
• Task Hub: A logical grouping of storage resources for providing durability to one or more durable function applications. Orchestrator and activity functions must be configured to run in the same task hub in order for them to communicate with each other.
• Control Queue: An Azure Storage queue which is used to drive orchestrator function execution. Note that orchestrator function execution is generally non-blocking, single-threaded, and light on CPU usage. Orchestration instances have an affinity with a single control queue based on a hashing algorithm. Each control queue may process messages from any and all orchestrator functions.
• Work-Item Queue: An Azure Storage queue which drives activity function execution. Note that unlike orchestrator functions, activity function execution will vary widely in the type and amount of compute resources required. The max throughput of this queue is expected to be significantly higher than that of control queues (which are bounded by single-threaded execution).
• Worker: A compute instance. This term may be used interchangeably with VM, host, node, compute instance, or Function App instance.
• History Table: An Azure Storage Table which stores orchestration history partitioned by orchestration instance IDs.
• Scale Controller: The Azure Functions infrastructure component which monitors trigger inputs (queues, etc.) and assigns function apps to compute instances based on trigger throughput. In this context, the scale controller monitors the queues in a given task hub, which is expected to be shared across multiple triggers.

GOALS

The following are the high-level scale goals for scale-out when using Durable Functions with the Azure Storage provider.

• Scale-out to hundreds of VMs/compute instances
• Scale-out to millions of orchestration instances
• Hundreds of combined activity and orchestrator executions per-second (when scaled-out)
• Efficiently maximize Azure Storage account resource limits under full load.

SCENARIOS

The following are high-level scenarios that describe when and how scale-out applies.

Azure Consumption Plan

Customer creates an Azure Functions application that uses the Durable Functions extension to implement durable orchestrations. The default configuration uses four partitions for scaling out orchestrator function load. The Azure Functions Scale Controller detects that there are four partitions and automatically starts monitoring the four control queues and the one work-item queue for unprocessed messages.

Once the function app is woken up by the Scale Controller, the Durable Functions extension starts up on a single worker and processes queue messages from all four control queues as well as the global work-item queue. As load on these queues increases, the Scale Controller assigns additional compute instances (workers) to the function app and the extension automatically load balances control queue processing across these nodes until there is a 1:1 ratio of compute instance and control queue. At this point, no further compute instances will be added to accommodate additional control queue load. However, if the work item queue continues to see an increase in throughput, the Scale Controller continues adding compute instances in addition to the four existing instances until the application is able to keep up with the work-item queue load.

Azure App Service Plan

Customer creates an Azure Functions application that uses the Durable Functions extension in the Azure App Service plan with a fixed number of VMs (e.g. 10 Large/A3 VMs). The default configuration uses four partitions for scaling out orchestrator function load. The customer uses the AlwaysOn feature of App Service to ensure all VMs are always warmed up, regardless of the load. The customer may or may not be using AutoScale rules to add or remove VMs based on load - for the sake of simplicity, we assume AutoScale is not being used.

On startup, the Durable Functions extension automatically load balances the partitions across four VMs, even before any messages arrive. As control queue messages arrive in busts across the four control queues, they are immediately picked up and processed concurrently by the four VMs which are monitoring these queues. The remaining six VMs actively wait to help process stateless work item messages.

Adding Partitions

The customer decides that their workload requires more than four partitions for processing orchestrator functions. By updating host.json, the number of partitions is increased from 4 to 16. This doesn't involve any downtime. Durable Functions clients and workers detect the new partitions and new control queues are created automatically. Also, new orchestration instances are load balanced to the new partitions. Existing instances continue to run in their previous partitions.

Removing Partitions

The customer is noticing performance problems because they've added too many partitions, yet are only processing orchestrator load on a single VM. The remedy this, they update host.json to drop the number of partitions down to 4. Clients detect this and start creating new instances on the first four partitions. The remaining partitions go into "deprovisioning" mode, and instances on those partitions get automatically migrated to the first four partitions. No further processing of messages occurs on the deprovisioning partitions.

REQUIREMENTS

These requirements are intended to formally describe the key assumptions for Durable Functions scale-out. The requirements are prioritized accordingly in support of the previously mentioned goals.

HIGH-LEVEL REQUIREMENTS

Category	Priority	Requirement	Notes
History	0	Orchestration history is equally distributed across table storage partitions
Trigger	0	Function processing is automatically load-balanced across all available workers
Trigger	0	Workers can be added or removed at any time and control-queue partitions are load-balanced automatically
Trigger	0	No two workers will ever process the same control queue
Trigger	1	The number of concurrent orchestrator function executions is configurable per-node
Trigger	1	The number of concurrent activity function executions is configurable per-node
Trigger	1	Customers can add additional partitions to the task hub with minimal downtime
Trigger	2	Customers can remove partitions from the task hub with minimal downtime
Trigger	2	The lease lock timeout and renewal intervals are configurable
Consumption Plan	1	Scale controller activates a function app when one or more messages appear in either a control queue or a work item queue
Consumption Plan	1	Scale controller adds compute instances up to the partition count as control queue load increases
Consumption Plan	1	Scale controller adds compute instances indefinitely as work-item queue load increases

PARTITION MANAGEMENT

The design of partition management is influenced by the design of partition management done by the Azure Event Hubs processing host (and in fact, borrows much of the implementation). Here we go over the details of partition management as they relate to Durable Functions.

1. The number of partitions defaults to four (this is not a hard requirement, but represents a good starting point).
2. If more partitions are needed later, the customer can update the task hub configuration in host.json with a new partition count. The host will detect this change after it has been restarted.
3. The host ensures that there is one control queue for every partition within the task hub in Azure Storage and will create them if necessary.
4. An Instances table exists (separate from the History table) that contains instance ID-to-partition mapping information. This table is periodically used by DTFx clients to know which queue to send control messages to. It may also be used for purposes unrelated to scale-out, like instance query.
5. When DTFx clients create new instances, they randomly choose an available partition, update the Instances table, and then enqueue the start message in the appropriate control queue. In the future, we could consider a smarter load balancing algorithm than random.
6. DTFx clients periodically scan the list of partitions (e.g. every 30 seconds) to learn about any changes.
7. Workers take ownership of partitions using a blob lease. The blob lease contains the name of the associated queue as well as the status of the partition - e.g. "active" or "deprovisioning".
8. Workers can steal partitions from other workers if the partition distribution is not balanced - e.g. if Worker A has two partitions and Worker B has none, Worker B can steal one partition from Worker A.
9. All workers will try to acquire all partition leases in order to handle cases where workers lose their lease for any reason.
10. If a worker acquires a "deprovisioning" lease, it will copy all messages in the deprovisioning queue into an active queue. When all messages have been removed from the deprovisioning queue for more than 60 seconds, the queue will be deleted. Waiting for 60 seconds ensures that all clients have time to learn that the deprovisioning queue should no longer be used for new orchestration instances.
11. The work-item queue is stateless and global across the task hub. All work-item messages flow through this queue regardless of which orchestration instance they belong to.

SCALE CONTROLLER INTERACTION

1. The scale controller will activate the function app if any visible messages exist in the work-item or API queues.
2. The scale controller will add new VMs if the work-item queue length increases over time.
3. The scale controller will add new VMs if both a partition's control queue length increases and the partition is on a VM that is also processing other partitions (otherwise adding VMs won't help).

CONCURRENCY MANAGEMENT

The owner of the function app can make the following changes in host.json:

• Number of concurrent activity functions per node
• Number of concurrent orchestrator functions per node
• Lease expiration time for partitions
• Lease renewal interval for partitions
• Lease acquire interval for partitions
• Number of partitions

All changes to host.json require all workers to be restarted.

[christopheranderson]

Notes:

• High level notes

  • I wouldn't mind some Activity or State diagrams here. Personally, visualizing the workflow helps me better understand what's going on.
  • Could use an FMA diagram for this as well
  • This diagram would also help diagnose issues like control queue/work item queue throughput mismatches.
  • Consider scaling out Work Item queues
  • Avoid hard caps in partition count, but there should be a recommended max based on table throughput. Avoiding hard caps makes it simpler if Storage gets faster or we replace Storage with Cosmos/etc. to reconfigure.
  • Consider allowing configuring Task Hub at the trigger level

• Terminology

  • Other words I had to think before I grok'd:
    • Node

• Goals

  • I wouldn't use VM, I'd use "Function App Instance" or something along those lines. There's no reason to assume Functions will be on different VMs, though that is often the case.
  • I feel like we should safely be able to have the goal that we can process as many items as the underlying storage service can offer. The high level cap we have is what Storage will limit us to. We should test that we can hit that.

• Scenarios

  • Is "on-premises" Azure Stack or "Functions Runtime"? I think for Stack, this is fine. For Functions Runtime, I'd break it out into it's own section. It might be out of scope (should be explicit), because of the hard dependency on Azure Storage for this spec.
  • I'm not super knowledgable about the Scale Controller at the low level, but it seems to me that both the control queue and activity queue are independent Functions/Triggers. If I create a control queue that creates activities for Functions that don't exist, it would never scale out, for instance. Does the scale controller even need to be aware that the control queue and activity queue are related? Or are you essentially saying that the control queue can never have more than X number of instances weight added to the resource allocation algorithm, but that activity queues aren't bounded by the partition?
  • For the App Service Plan, it might not be a fixed number of instances, still. Customers could even set up auto-scale rules based on the Activity queue length/etc.

• Partition Management

  • Nit: Event Hubs is not part of Service Bus anymore, although the old SDKs still have that namespace, I believe they are breaking that relationship going forward.
  • Number of partitions are decided when the function app is created? Or the Orchestration trigger?
  • I understand that the partitioning strategy requires a known set of partitions, but I feel like we should think hard about adding a P2 requirement that we'll help "migrate" you to a new set of partitions. Deleting and recreating is a bad place to leave customers. I think we could attempt to solve this in the same way we may attempt to solve versioning updates. Regardless, I don't think the scale controller should make any assumptions that the number of partitions will remain static.
  • I'm not sure I understand the instance ID concerns.

• Scale Controller Interaction

  • I'm not sure that we should make any assumptions between the work item or controller queues. Treat each trigger as a separate entity. (I'm not sure you're making different assumptions, I just want to reassert that idea)
  • I believe the scale controller logic for the work-item queue should be the same as the normal Storage Queue, correct? To that end, same thing for the control queue, expect that one should be capped at partition count.

• Concurrency Management

  • This is something I'd love to see a design meeting on. Lots of ideas in this space for Functions as a whole.
  • Node == Instance?
  • Lease renewal time should be == Function timeout time + C, yes?
  • Are we going to have defaults here?

[cgillum]

Thanks for the detailed feedback. One thing that needed clarification is that the control queue(s) is shared by all orchestrator functions in the task hub. I added more detail in the definitions section to help emphasize this.

Re: High-level notes:

• Agreed diagrams would be very helpful. Let me figure out how to add images in GitHub. I have one in a PPT deck that might help.
• Re: FMA, I'd like to understand what specifically this looks like. From a scale perspective, a machine failure results in partition rebalancing, much like Event Hubs. I tried to capture this in the P0 requirement which states Nodes can be added or removed at any time and control-queue partitions are load-balanced automatically.
• I'll make scale-out of work-item queues P2. Right now the high-level goal is hundreds actions/sec and one work-item queue should be able to achieve this. In general, the throughput on this queue should be significantly higher than that of the control queues since control queues are single-threaded.
• I added a note that there is no hard cap on the maximum partition count. As we discussed, we'll need to provide guidance here as more partitions will result in more overhead for the listener.
• Per-function task hub configuration can be added. I've added a new P2 line item to track.

Re: Terminology:

• Clarified the behaviors of the queues
• Clarified that "Node", "VM", and "compute instance" are interchangeable. I wanted to avoid using the term instance because that could easily be confused with orchestration instance.

Re: Goals:

• Added note about being able to exhaust storage limits, though this might conflict with the throughput goal of "hundreds of actions/sec" (I would hope we could do better than this if we max out storage limits).

Re: Scenarios & Scale Controller Interaction:

• Dropped my reference to "on-premise". Let's consider this separately.
• Let me know if I need more clarify on the relationship between the scale controller, control queues, and work-item queues. A diagram might help, but I feel fairly confident that they cannot be considered independently. The thing to understand is that a task hub is really the unit that the scale controller monitors - not necessarily individual queues/triggers. I added a note to clarify this.
• Added a note about auto-scale, but I don't think it's particularly important to mention in this spec.

Re: Partition Management:

• Partitions are decided when the task hub is created. I added a TODO to define when the task hub is considered created (the current implementation creates it automatically during app startup, which may not always work).
• I added partition reconfiguration as a P2. It would certainly make this a step-up from the EH experience. We should figure out how fancy we want to get with this. The cheapest thing we could do is require all load to stop, manually update the partition count in storage (i.e. update a blob that stores config info), and then resume load. Everything should just rebalance on its own without any issues. The thing to avoid is adding/removing partitions while messages are being actively processed.
• Re: instance IDs - we use hashing to map orchestration instances to partitions. The partition ID is derived from the orchestration instance ID. If the instance ID generation algorithm is not random, it's possible that the hash will result in unbalanced scaling of orchestration instances across partitions, resulting in some control-queues being hot while others are idle.

Re: Concurrency management:

• I'm not super confident about these requirements. In fact, I'm tempted to remove them as configurable options and just add knobs as we discover need for them through perf/scale testing. The defaults I'm starting out with are the same defaults used by the Event Hub Processing Host.

[cgillum]

Per our discussion, I've updated the design and requirements to allow for manually changing partition counts on existing task hubs.