DEV Community: Chris Gillum

Resetting your Durable Functions Task Hub state

Chris Gillum — Fri, 03 Dec 2021 21:35:20 +0000

I was recently asked in our GitHub Discussions board how to delete all storage resources associated with a Durable Functions app. I posted an answer that I was happy with but thought it might also be worth sharing here for broader reach.

Why would I want to delete all my Durable storage resources?

There are several reasons you might want to delete all your Durable state.

You want to deploy a new version of your app from scratch and don't intend to preserve any existing orchestrations (check out our Versioning and Zero Downtime Deployments docs if you're interested in this topic).
You have an environment that's been sitting around and want to quickly clean up the storage resources to avoid incurring storage costs.
You're doing rapid prototyping and don't want the debugger to be triggered by old orchestrations from previous debug sessions.

So while there are a few reasons to do this, Durable Functions doesn't actually provide a first-class API for this.

What's the "official" way to clean up state?

The "official" way to delete your Durable state would be to use our Terminate APIs to terminate all still-running orchestrations, followed by our Purge APIs to delete all state associated with these instances. This works and is fully supported, but can be very time consuming, especially if you have a lot of orchestration instances. It can also be expensive if you're using the default Azure Storage backend since each transaction is billed individually.

If you're curious about why this approach can be slow, consider the following:

The Purge APIs will only clear data for orchestration instances that are in a non-running state (Completed, Failed, or Terminated). If you want to purge everything, you need to make sure all your running instances are completed or terminated.
Scanning for all running instances can be time consuming. We have APIs that allow you to query for running instances, but this is a very slow API when using Azure Storage because they rely on multi-partition table scans.
The Terminate API works by sending a message to a specific orchestration instance. Only after the orchestration receives the message will it transition into the Terminated state. This means any existing queue backlogs will further slow down the process.

Read-on if you need a faster way to clean up your Durable state and you're willing to try something that's "less official".

What's the "fast" way to clean up Durable state?

The fast and less official way to clean up state is to use the underlying Durable Task Framework storage APIs directly. Durable Functions has an internal dependency on the Durable Task Framework, so no new packages need to be added to your app to access them. Here's an example function that demonstrates how to clean up Azure Storage state quickly (disclaimer: use any of my code samples at your own risk).

class ResetDurableStateHelper
{
    readonly INameResolver nameResolver;

    // INameResolver is a service of the Functions host that can
    // be used to look up app settings.
    public ResetDurableStateHelper(INameResolver nameResolver)
    {
        this.nameResolver = nameResolver;
    }

    [FunctionName("ResetDurableState")]
    public async Task<HttpResponseMessage> ResetDurableState(
        [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestMessage req,
        [DurableClient] IDurableClient client,
        ILogger log)
    {
        // Get the connection string from app settings
        string connString = this.nameResolver.Resolve("AzureWebJobsStorage");
        var settings = new AzureStorageOrchestrationServiceSettings
        {
            StorageConnectionString = connString,
            TaskHubName = client.TaskHubName,
        };

        // AzureStorageOrchestrationService is defined in
        // Microsoft.Azure.DurableTask.AzureStorage, which is an
        // implicit dependency of the Durable Functions extension.
        var storageService = new AzureStorageOrchestrationService(settings);

        // Delete all Azure Storage tables, blobs, and queues in the task hub
        log.LogInformation(
            "Deleting all storage resources for task hub {taskHub}...",
            settings.TaskHubName);
        await storageService.DeleteAsync();

        // Wait for a minute since Azure Storage won't let us immediately
        // recreate resources with the same names as before.
        log.LogInformation(
            "The delete operation completed. Waiting one minute before recreating...");
        await Task.Delay(TimeSpan.FromMinutes(1));

        // Optional: Recreate all the Azure Storage resources for this task hub.
        // This happens automatically whenever the function app restarts, so it's
        // not a required step.
        log.LogInformation(
            "Recreating storage resources for task hub {taskHub}...",
            settings.TaskHubName);
        await storageService.CreateIfNotExistsAsync();

        return req.CreateResponse(System.Net.HttpStatusCode.OK);
    }
}

In short, the above HTTP-triggered function uses the Azure Storage-based orchestration service APIs to delete all the task hub state and recreate it. Refer to the inline code comments for more details on each step.

Unfortunately, these APIs are only available in .NET, so users of other language stacks will need to either use the traditional techniques or create a new .NET-based app and exercise their copy/paste skills.

What if I'm using a backend other than Azure Storage?

Great question! For example, you might be using the Netherite or MSSQL backends for storing your Durable Task state instead of Azure Storage. It turns out that the process is exactly the same (or perhaps slightly simpler). Here's an example with the MSSQL provider, since I happened to have some code for this handy (again, use at your own risk).

class ResetDurableStateHelper
{
    readonly INameResolver nameResolver;

    // INameResolver is a service of the Functions host that can be used
    // to look up app settings.
    public ResetDurableStateHelper(INameResolver nameResolver)
    {
        this.nameResolver = nameResolver;
    }

    [FunctionName("ResetDurableState")]
    public async Task<HttpResponseMessage> ResetDurableState(
        [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestMessage req,
        ILogger log)
    {
        // Get the connection string from app settings.
        string cnString = this.nameResolver.Resolve("SQLDB_Connection");
        string dbName = new SqlConnectionStringBuilder(cnString).InitialCatalog;
        var settings = new SqlOrchestrationServiceSettings(cnString);

        // SqlOrchestrationService is defined in the
        // Microsoft.DurableTask.SqlServer nuget package, which is an
        // implicit dependency of Microsoft.DurableTask.SqlServer.AzureFunctions.
        var storageService = new SqlOrchestrationService(settings);

        // This will delete all Durable Task resources associated with
        // the database. It's not scoped to any specific task hub.
        // Non-Durable Task schema objects will not be impacted.
        log.LogInformation(
            "Deleting all Durable Task resources in the {database} database...",
            dbName);
        await storageService.DeleteAsync();

        // Optional: Recreate all the Durable Task SQL resources for this
        // task hub. This is done automatically whenever the function app
        // restarts, so it's not a required step if you plan to restart.
        log.LogInformation(
            "Recreating Durable Task resources in the {database} database...",
            dbName);
        await storageService.CreateIfNotExistsAsync();

        return req.CreateResponse(System.Net.HttpStatusCode.OK);
    }
}

Conclusion

I hope this post puts a useful tool in your toolbox when using Durable Functions. The above procedure may not be appropriate for some production environments, but it may be useful for those that want a quick way to clean up their storage and don't have strict data retention policies.

Also, if you happen to run this cleanup function in the same Durable function app that you're cleaning up, you may want to restart the app just to make 100% sure that everything gets back into a clean state (running threads may not be happy about having their storage deleted and recreated right from underneath them). 😉

Scheduling tons of orchestrator functions concurrently in C#

Chris Gillum — Mon, 26 Apr 2021 18:44:57 +0000

I had a call with an engineer on another team who wanted advice for how to schedule a large number of concurrent orchestration functions in Durable Functions. I shared with him some sample code for how we do it in our own internal performance tests and decided it might be useful to share publicly too.

First, here is the orchestration that we're running for the test. It's just a basic sequential orchestrator that calls a SayHello activity function 5 times.

[FunctionName(nameof(HelloSequence))]
public static async Task<List<string>> HelloSequence(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var outputs = new List<string>
    {
        await context.CallActivityAsync<string>(nameof(SayHello), "Tokyo"),
        await context.CallActivityAsync<string>(nameof(SayHello), "Seattle"),
        await context.CallActivityAsync<string>(nameof(SayHello), "London"),
        await context.CallActivityAsync<string>(nameof(SayHello), "Amsterdam"),
        await context.CallActivityAsync<string>(nameof(SayHello), "Mumbai")
    };

    return outputs;
}

[FunctionName(nameof(SayHello))]
public static string SayHello([ActivityTrigger] string name) => $"Hello {name}!";

And here is the HTTP trigger function we use to trigger a performance run. It takes a count parameter from the query string as the number of concurrent "HelloSequence" orchestrations to run. In our tests, we'll often run more than 100K orchestrations concurrently.

[FunctionName(nameof(StartManySequences))]
public static async Task<IActionResult> StartManySequences(
    [HttpTrigger(AuthorizationLevel.Function, "post", Route = null)] HttpRequest req,
    [DurableClient] IDurableClient starter,
    ILogger log)
{
    if (!int.TryParse(req.Query["count"], out int count) || count < 1)
    {
        return new BadRequestObjectResult("A 'count' query string parameter is required and it must contain a positive number.");
    }

    string prefix = await ScheduleManyInstances(starter, nameof(HelloSequence), count, log);
    return new OkObjectResult($"Scheduled {count} orchestrations prefixed with '{prefix}'.");
}

This method calls into a ScheduleManyInstances helper method, which does the actual scheduling. Before I get into our implementation, however, I think it would be useful to describe what not to do.

Naïve implementation #1: Sequential

The most common naïve implementation is to use a for-loop with an await in each iteration.

public static async Task<string> ScheduleManyInstances(
    IDurableOrchestrationClient client,
    string orchestrationName,
    int count,
    ILogger log)
{
    log.LogWarning($"Scheduling {count} orchestration(s)...");
    DateTime utcNow = DateTime.UtcNow;
    string prefix = utcNow.ToString("yyyyMMdd-hhmmss");

    for (int i = 0; i < count; i++)
    {
        // Start each instance one-at-a-time
        string instanceId = $"{prefix}-{i:X16}";
        await client.StartNewAsync(orchestrationName, instanceId);
    }

    log.LogWarning($"All {count} orchestrations were scheduled successfully!");
    return prefix;
}

The above is really slow because you're only enqueuing a single orchestration start message at a time. If you're scheduling a large number of orchestrations, then the client that invoked this HTTP function will probably time-out before all the orchestrations are scheduled. Ideally we'd schedule orchestrations in parallel, which leads us to the next bad practice.

Naïve implementation #2: Too much parallelism

Next we try using Task.WhenAll to schedule all the orchestrations in parallel. This is better than scheduling orchestrations sequentially because it allows us to queue up new work much more quickly. However, it has a major scalability problem, which I'll describe below.

public static async Task<string> ScheduleManyInstances(
    IDurableOrchestrationClient client,
    string orchestrationName,
    int count,
    ILogger log)
{
    log.LogWarning($"Scheduling {count} orchestration(s)...");
    DateTime utcNow = DateTime.UtcNow;
    string prefix = utcNow.ToString("yyyyMMdd-hhmmss");

    // Run all StartNewAsync tasks concurrently
    var startTasks = new Task[count];
    for (int i = 0; i < count; i++)
    {
        string instanceId = $"{prefix}-{i:X16}";
        startTasks[i] = client.StartNewAsync(orchestrationName, instanceId);
    }

    await Task.WhenAll(startTasks);

    log.LogWarning($"All {count} orchestrations were scheduled successfully!");
    return prefix;
}

The problem is that if count is a large number (like 100K), you will very quickly exhaust both threads and outbound TCP connections on your VM. This is because the .NET thread scheduler will try to aggressively allocate a huge number of threads to satisfy your concurrency demands. Each of those threads will also try to open a connection to Azure Storage concurrently, requiring a new TCP connection. The result is often that the function will fail, making this implementation highly unreliable.

Naïve implementation #3: Throttled Parallelism for Parallel.For

This next approach uses Parallel.For to use a throttled concurrency approach.

public static async Task<string> ScheduleManyInstances(
    IDurableOrchestrationClient client,
    string orchestrationName,
    int count,
    ILogger log)
{
    log.LogWarning($"Scheduling {count} orchestration(s)...");
    DateTime utcNow = DateTime.UtcNow;
    string prefix = utcNow.ToString("yyyyMMdd-hhmmss");

    // Use up to 200 threads to schedule orchestrations concurrently
    var maxConcurrencyOptions = new ParallelOptions { MaxDegreeOfParallelism = 200 };
    Parallel.For(0, count, maxConcurrencyOptions, i =>
    {
        string instanceId = $"{prefix}-{i:X16}";

        // Use GetAwaiter().GetResult() to block since Parallel.For() doesn't support async
        client.StartNewAsync(orchestrationName, instanceId).GetAwaiter().GetResult();
    });

    log.LogWarning($"All {count} orchestrations were scheduled successfully!");
    return prefix;
}

This works much better than the first two solutions. It fixes the TCP connection exhaustion issue by giving the system enough time to reuse existing TCP connections. It also addresses the thread starvation issue by ensuring we don't use more than 200 threads at the same time.

However, the Parallel.For solution is still inefficient because each degree of parallelism is occupying a dedicated thread, and those threads will get blocked waiting for the StartNewAsync call to complete. Threads are expensive in terms of CPU and memory and take time to allocate. Ideally we'd do this work in a non-blocking way that allows us to aggressively reuse threads.

Final implementation: Async throttled parallelism

The solution we use is a variation of the above that provides much better thread reuse by using a fully async implementation. I defined a ParallelForEachAsync helper extension method to achieve this.

public static async Task<string> ScheduleManyInstances(
    IDurableOrchestrationClient client,
    string orchestrationName,
    int count,
    ILogger log)
{
    log.LogWarning($"Scheduling {count} orchestration(s)...");
    DateTime utcNow = DateTime.UtcNow;
    string prefix = utcNow.ToString("yyyyMMdd-hhmmss");

    await Enumerable.Range(0, count).ParallelForEachAsync(200, i =>
    {
        string instanceId = $"{prefix}-{i:X16}";
        return client.StartNewAsync(orchestrationName, instanceId);
    });

    log.LogWarning($"All {count} orchestrations were scheduled successfully!");
    return prefix;
}

Here is the ParallelForEachAsync extension method implementation, which includes the async parallel throttling behavior.

public static async Task ParallelForEachAsync<T>(this IEnumerable<T> items, int maxConcurrency, Func<T, Task> action)
{
    List<Task> tasks;
    if (items is ICollection<T> itemCollection)
    {
        // optimization to reduce the number of memory allocations
        tasks = new List<Task>(itemCollection.Count);
    }
    else
    {
        tasks = new List<Task>();
    }

    using var semaphore = new SemaphoreSlim(maxConcurrency);
    foreach (T item in items)
    {
        tasks.Add(InvokeThrottledAction(item, action, semaphore));
    }

    await Task.WhenAll(tasks);
}

static async Task InvokeThrottledAction<T>(T item, Func<T, Task> action, SemaphoreSlim semaphore)
{
    await semaphore.WaitAsync();
    try
    {
        await action(item);
    }
    finally
    {
        semaphore.Release();
    }
}

As you can see, we use a SemaphoreSlim to ensure we don't execute more than maxConcurrency concurrent actions at the same time. Because the code path is fully async, we can now run many operations in parallel with a much smaller number of threads. Right now we have maxConcurrency set to 200, but it's very possible that a larger number could have worked as well.

Note that this ParallelForEachAsync extension method is generic and can be used for any .NET code that needs to execute asynchronous tasks with a cap on concurrency. In fact, we use it inside parts of the Durable Task Framework to reduce the number of concurrent TCP connections we open when making calls to Azure Storage.

Anyways, I hope this is helpful for anyone doing performance work with Durable Functions. If there is a simple way to do this in .NET and I just didn't know about it, I'm interested in learning about that too!

Durable Functions vs. Apache Airflow

Chris Gillum — Sun, 29 Nov 2020 22:49:43 +0000

Recently I've been looking at Apache Airflow since I've noticed it getting a lot of attention from Python developers and cloud providers for supporting "workflow" scenarios. For context, I'm the creator of Durable Functions on Azure and we'll soon be announcing the general availability (GA) of Durable Functions for Python. Both Airflow and Durable Functions support building workflows in Python, so I thought it would be worth doing some investigation to understand the differences between the two technologies. This blog post is my attempt at doing this comparison and I hope folks find it somewhat useful.

In the end, what I've learned is that Durable Functions and Apache Airflow are trying to solve different problems using different approaches, in spite of the fact that they both support implementing "workflows" in Python. The main differences are around the types of workflows supported and the role that Python plays in the authoring process. I go into more details on these differences in this post.

Orchestrators vs DAGs

The most important technical difference I've found is the programming model. Durable Functions allows you to build workflows by writing orchestrator functions. Orchestrator functions describe how actions are executed and the order in which actions are executed. To illustrate, the following is a simple sequential orchestration in Python that calls three tasks known as "activities", t1, t2, and t3, in sequence using Python's generator syntax.

import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    x = yield context.call_activity("t1", None)
    y = yield context.call_activity("t2", x)
    z = yield context.call_activity("t3", y)
    return z

main = df.Orchestrator.create(orchestrator_function)

Each yield expression causes the orchestrator function to wait for the scheduled task to complete and then save the result to a local variable for later use. When yielded in this way, the orchestrator function can be unloaded from memory and the progress of the workflow is persisted.

An orchestration can have many different types of actions, including activity functions, sub-orchestrations, they can wait for external events, make HTTP calls, and sleep using durable timers. Orchestrator functions can also interact with durable actor-like objects known as entity functions. All of this is done using normal procedural coding constructs, which means you can use programming language features like conditionals, loops, function calls, exception handling via try/except/finally (for implementing compensation logic), etc. For developers, this model is very natural and scales nicely to even highly complex workflows. Reliability and distributed execution is handled for you by the underlying framework.

The Apache Airflow programming model is very different in that it uses a more declarative syntax to define a DAG (directed acyclic graph) using Python. To illustrate, let's assume again that we have three tasks defined, t1, t2, and t3. You could implement a similar sequential workflow as above using the following code in Airflow:

dag = DAG('hello_world', description='Sequential DAG',
          schedule_interval='0 12 * * *',
          start_date=datetime(2020, 11, 28), catchup=False)

t1 = PythonOperator(task_id='t1', dag=dag, python_callable=t1)
t2 = PythonOperator(task_id='t2', dag=dag, python_callable=t2)
t3 = PythonOperator(task_id='t3', dag=dag, python_callable=t3)

t1 >> t2 >> t3

It's important to note that Airflow Python scripts are really just configuration files specifying the DAG’s structure as code. Unlike with normal Python scripts, you are not able to pass dynamic inputs, inspect outputs, or do control flow with conditionals, loops, error handling, or other features of the language. It's therefore best to think of Airflow DAG authoring as a workflow DSL that happens to use Python for configuration.

To illustrate this point further, consider a common "approval workflow" use-case. In this scenario, someone submits a purchase order that needs to be approved by a manager. The workflow waits for the approval and moves onto the processing step immediately after it is received. However, if no approval is received in 72 hours (maybe the approver is on vacation), an escalation task is scheduled to help resolve the pending approval. Using Durable Functions, we can implement this workflow using code like the following:

import azure.durable_functions as df
from datetime import timedelta 


def orchestrator_function(context: df.DurableOrchestrationContext):
    yield context.call_activity("RequestApproval", None)

    # create a timer task that expires 72 hours from now
    due_time = context.current_utc_datetime + timedelta(hours=72)
    timeout_task = context.create_timer(due_time)

    # create a task that completes when an "Approval" event is received
    approval_task = context.wait_for_external_event("Approval")

    # context.task_any() waits until any one task completes and returns it
    winning_task = yield context.task_any([approval_task, timeout_task])

    if approval_task == winning_task:
        timeout_task.cancel()
        yield context.call_activity("Process", approval_task.result)
    else:
        yield context.call_activity("Escalate", None)


main = df.Orchestrator.create(orchestrator_function)

As you can see, tasks are scheduled dynamically, have inputs and outputs, and the outputs can be used to decide the next step to take in the workflow. Airflow DAGs, on the other hand, are more optimized for defining static data pipelines that don't necessarily require passing data around.

There are important tradeoffs between these two workflow authoring models. In the "imperative code" model of Durable Functions where task scheduling is dynamic, you can express a much larger set of possible workflows. However, you have to be careful to ensure your orchestrator code is deterministic and doesn't violate the orchestrator code constraints. The "declarative code" model of Airflow is much more static and constrained, but those constraints make it easier to build tools that analyze the DAGs and do things like create visualizations (more on this in subsequent sections).

Activities vs. Operators

Another important distinction is that of activities and operators. Both represent workflow tasks that can be scheduled. In Durable Functions, everything is a function. This includes the orchestrators as well as the activities being orchestrated. However, it's up to you to write the business logic for each of your activity functions. As mentioned earlier, Durable Functions offers several low-level primitive task types, like timers, external event handlers, and HTTP actions, but doesn't offer an extensive library "pre-cooked" tasks in its current version.

Operators in Apache Airflow are more like traditional workflow actions in the sense that there is a preexisting library of operators that you choose from. The most basic is PythonOperator, which allows you to run a Python function as a task. In addition, a broad set of operators exist for interacting with external systems, like databases, HTTP endpoints, cloud data services like S3, etc. You can find the full list of supported operators here. You can also leverage other operators developed by the OSS community, or create your own via Airflow plugins. Because of this library of existing operators, it's possible that many Airflow-authored workflows won't involve any custom code at all.

Event-driven vs. CRON

Apache Airflow DAGs are primarily triggered on predefined CRON schedules. More info here. It's also possible to do one-off runs using the CLI. The Airflow CRON model is especially useful if you need to do backfilling or do "catchup" executions as part of your data processing pipeline.

Durable Functions, on the other hand, relies on the event-driven mechanisms of Azure Functions for triggering orchestrations. For example, you would create a client function that starts one or more orchestration instances with dynamic inputs when an HTTP request is received, when a queue message arrives, or based on an event from one of many other supported trigger types, including CRON-based timer trigger events.

Do-it-yourself vs. built-in management tools

I suspect one of the biggest draws to Apache Airflow is the UI tools that allow you to easily manage and inspect your DAGs. You can see the full list of Airflow UI screenshots here. Some especially useful views, in my opinion, are the graph view that visualizes your DAG and shows its progress in real-time, and the Gantt chart that shows you how long each step in your workflow has taken. As I mentioned earlier, the static nature of DAGs makes it easy to build visual tooling for management and monitoring purposes. You can find a great video showing off some of these features here.

Durable Functions exposes a set of management APIs over HTTP and language-specific SDK APIs. However, there are no built-in tools for doing this kind of visualization (though some 3rd party tools are available). In terms of monitoring, Durable Functions emits detailed telemetry into Application Insights, which is quite powerful but has its own learning curve for creating alerts, visualizations, etc. More information on diagnostics and monitoring for Durable Functions can be found here.

Serverless vs. VMs

If at this point you believe that both Durable Functions and Apache Airflow could potentially meet your needs, then another important difference to consider is the range of supported hosting environments. Durable Functions has a significant advantage in that it can run in a completely serverless environment (via Azure Functions). This means there are no servers or VMs to set up, no databases to configure, scale-out is automatic and elastic, and you only pay when work is being done. This often translates into pennies per month or less when you have very light workloads.

On the other hand, Apache Airflow has managed offerings available in other clouds, including Astronomer.io, Google Cloud Composer, and Amazon Managed Workflows for Apache Airflow (MWAA). These managed offerings provide automatic scaling and management of infrastructure but you are still left with per-hour VM billing models. At the current time of writing, the cheapest configurations on these clouds will cost between $100 and $400 USD per month, regardless of how many workflows you execute.

Conclusion

The most important conclusion I arrived at when examining Apache Airflow is that it is designed and optimized specifically for static data processing pipelines. If you need to build ETL pipelines with a known set of steps and dependencies, then Airflow might be a great option because of the various built-in tools and the broad array of plugins created by the Airflow community.

However, if you're looking for a platform that supports a broader range of workflow primitives and scenarios, and if you need the flexibility that code-first orchestrations provide, then something like Durable Functions may be more suitable.