Understanding Azure Durable Functions - Part 1: Overview

This is the first part in a series of articles.

Durable Functions are built on top of top of the basic Azure Functions platform and provide the ability to define how multiple individual functions can be set up to work together.

You can accomplish a lot with just the basic Azure Functions so the durable extensions are not necessarily required to implement your solutions, however if you find yourself needing to make multiple functions work together in some kind of workflow then durable functions may be able to simply things and as a bonus allow you to document in code how the functions interact.

Using Multiple Functions

Just as you would keep classes and methods functionally cohesive (i.e. do one/small number related things) so should your Azure Functions be “specialized”. There are a number of good reasons for breaking down a problem into multiple individual functions such as:

  • Auto-scaling of functions rather than auto-scaling one massive “god” function
  • Maintainability: individual functions doing one thing are easier to understand/bug fix/test
  • Reusability/composeability: smaller units (functions) of logic could be reused in multiple workflows
  • Time-outs: if an individual function execution exceeds a timeout it will be stopped before finishing

Even if you are not going to use durable functions the above points still make sense.

Common Patterns That Durable Functions Can Help With

There are a number of common logical/architectural/workflow patterns that durable functions can help to orchestrate such as:

  • Chained Functions: the output of one function triggers the next function in the chain (aka functions pipeline)
  • Fan out, fan in: Split data into “chunks” (fanning out), process chunks in parallel,  aggregate results (fan-in)
  • Asynchronous HTTP APIs: Co-ordinate HTTP request with other services, client polls endpoint to check if work complete
  • Monitors: recurring process to perform clean-up, monitor (call) APIs to wait for completion, etc.
  • Human interaction: wait for human to make a decision at some part during the workflow

Implementing the preceding patterns without the use of durable functions may prove difficult, complex, or error prone due to the need to manually manage checkpoints (where are we in the process?) in addition to  other implementation details.

What Do Durable Functions Provide?

When using durable functions, there are a number of implementation details that are taken care of for you such as:

  • Maintains execution position of the workflow
  • When to execute next function
  • Replaying actions
  • Workflow monitoring/diagnostics
  • Workflow state storage
  • Scalability

Durable functions  also provide the ability to specify the workflow (“orchestration”) in code rather than using a visual flowchart style tool for example.

In the next part of this series we’ll see how to create a basic durable orchestration in C#.

SHARE:

Accessing Cosmos DB JSON Properties in Azure Functions with Dynamic C#

This is the eighth part in a series of articles.

When working with the Cosmos DB Microsoft.Azure.Documents.Document class, if you need to get custom properties from the document you can use the GetPropertyValue method as we saw used in part six of this series and replicated as follows:

[FunctionName("PizzaDriverLocationUpdated1")]
public static void RunOperation1([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    LeaseCollectionName = "PizzaDriverLocationUpdated1",
    CreateLeaseCollectionIfNotExists = true,
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            log.LogInformation($"Running operation 1 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

In the preceding code, the Azure Function is triggered from new/updated documents and the data that needs processing is passed to the function by way of the IReadOnlyList<Document> modifiedDrivers parameter. If you have multiple functions that work with a document type you may end up with many duplicated GetPropertyValue calls such as repeatedly getting the driver’s name: var driverName = modifiedDriver.GetPropertyValue<string>("Name"); Also notice the use of the magic string “Name” to refer to the document property to retrieve, this is not type safe and will return null at runtime if there is no property with that name.

Another option to simplify the code and remove the magic string is to use dynamic as the following code demonstrates:

[FunctionName("PizzaDriverLocationUpdated2")]
public static void RunOperation2([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    LeaseCollectionName = "PizzaDriverLocationUpdated2",
    CreateLeaseCollectionIfNotExists = true,
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<dynamic> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.Name;

            log.LogInformation($"Running operation 2 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

Notice in the preceding code that the binding has been changed from IReadOnlyList<Document> modifiedDrivers to IReadOnlyList<dynamic> modifiedDrivers and the code has changed from var driverName = modifiedDriver.GetPropertyValue<string>("Name"); to var driverName = modifiedDriver.Name;

While this removes the magic string, it is still not type safe and an incorrectly spelled property name will not error at compile time. Furthermore if the property does not exist, rather than return null, an exception will be thrown in your function at runtime.

If you’re not familiar with dynamic C# be sure to check out my Dynamic C# Fundamentals Pluralsight course.

SHARE:

How to Schedule Cosmos DB Data Processing With Azure Functions

This is the seventh part in a series of articles.

You can perform scheduled/batch processing of Azure Cosmos DB data by making use of timer triggers in Azure Functions.

Timer triggers allow you to set up a function to execute periodically based on set schedule.

For example the following attribute will cause the function to be executed once per day at  22:00 hours:

[TimerTrigger("0 0 22 * * *")]

As an example, in this series we’ve been using the domain of pizza delivery. Suppose that once per day the manager wanted an SMS with the total sales for the day.

To accomplish this we can combine a timer trigger, cosmos DB input binding, and a Twilio SMS output binding. In this example, the CosmosDB input binding is bound to an instance of DocumentClient. This allows us to perform more complex queries against Cosmos DB as we saw in part 2 of this series.

Once we have a DocumentClient instance, we can use LINQ to query Cosmos DB:

DateTime startOfToday = DateTime.Today;
DateTime endOfToday = startOfToday.AddDays(1).AddTicks(-1); 

decimal totalValueOfTodaysOrders = 
    client.CreateDocumentQuery<Order>(ordersCollectionUri, options)
         .Where(order => order.OrderDate >= startOfToday && order.OrderDate <= endOfToday)
         .Sum(order => order.OrderTotal);

The preceding query gets the total value or orders that have todays date, where order documents look like the following:

[
    {
        "id": "3",
        "StoreId": 2,
        "OrderTotal": 10.25,
        "OrderDate": "2019-06-06T17:17:17.7251173Z",
        "_rid": "Vg08AKOQeVQBAAAAAAAAAA==",
        "_self": "dbs/Vg08AA==/colls/Vg08AKOQeVQ=/docs/Vg08AKOQeVQBAAAAAAAAAA==/",
        "_etag": "\"00000000-0000-0000-1c2d-b2a092fd01d5\"",
        "_attachments": "attachments/",
        "_ts": 1559801067
    },
    {
        "id": "4",
        "StoreId": 2,
        "OrderTotal": 10.25,
        "OrderDate": "2019-06-06T18:18:18.7251173Z",
        "_rid": "Vg08AKOQeVQCAAAAAAAAAA==",
        "_self": "dbs/Vg08AA==/colls/Vg08AKOQeVQ=/docs/Vg08AKOQeVQCAAAAAAAAAA==/",
        "_etag": "\"00000000-0000-0000-1c37-77f607ea01d5\"",
        "_attachments": "attachments/",
        "_ts": 1559805263
    },
    {
        "id": "1",
        "StoreId": 1,
        "OrderTotal": 100,
        "OrderDate": "2019-06-06T14:14:14.7251173Z",
        "_rid": "Vg08AKOQeVQBAAAAAAAACA==",
        "_self": "dbs/Vg08AA==/colls/Vg08AKOQeVQ=/docs/Vg08AKOQeVQBAAAAAAAACA==/",
        "_etag": "\"00000000-0000-0000-1c2d-a87985f501d5\"",
        "_attachments": "attachments/",
        "_ts": 1559801050
    },
    {
        "id": "2",
        "StoreId": 1,
        "OrderTotal": 25.87,
        "OrderDate": "2019-06-06T16:16:16.7251173Z",
        "_rid": "Vg08AKOQeVQCAAAAAAAACA==",
        "_self": "dbs/Vg08AA==/colls/Vg08AKOQeVQ=/docs/Vg08AKOQeVQCAAAAAAAACA==/",
        "_etag": "\"00000000-0000-0000-1c2d-aef57e8c01d5\"",
        "_attachments": "attachments/",
        "_ts": 1559801061
    }
]

Once the total order value has been calculated, a Twilio SMS can be created and written to the Twilio output binding.

The complete listing is as follows:

using System;
using System.Linq;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Twilio.Rest.Api.V2010.Account;
using Twilio.Types;

namespace DontCodeTiredDemosV2.CosmosDemos
{
    public static class DailySales
    {
        [FunctionName("DailySales")]
        public static void Run(
            [TimerTrigger("0 0 22 * * *")]TimerInfo myTimer,            
            [CosmosDB(ConnectionStringSetting = "pizzaConnection")] DocumentClient client,
            [TwilioSms(AccountSidSetting = "TwilioAccountSid", AuthTokenSetting = "TwilioAuthToken", From = "%TwilioFromNumber%")
                ] out CreateMessageOptions messageToSend,
            ILogger log)
        {
            log.LogInformation($"C# Timer trigger function executed at: {DateTime.Now}");

            Uri ordersCollectionUri = UriFactory.CreateDocumentCollectionUri(databaseId: "pizza", collectionId: "orders");

            var options = new FeedOptions { EnableCrossPartitionQuery = true }; // Enable cross partition query

            DateTime startOfToday = DateTime.Today;
            DateTime endOfToday = startOfToday.AddDays(1).AddTicks(-1); 

            decimal totalValueOfTodaysOrders = 
                client.CreateDocumentQuery<Order>(ordersCollectionUri, options)
                      .Where(order => order.OrderDate >= startOfToday && order.OrderDate <= endOfToday)
                      .Sum(order => order.OrderTotal);


            var messageText = $"Total sales for today: {totalValueOfTodaysOrders}";

            log.LogInformation(messageText);

            var managersMobileNumber = new PhoneNumber(Environment.GetEnvironmentVariable("ManagersMobileNumber"));

            var mobileMessage = new CreateMessageOptions(managersMobileNumber)
            {
                Body = messageText
            };

            messageToSend = mobileMessage;
        }
    }
}

In the preceding code, notice the From = "%TwilioFromNumber%" element of the [TwilioSms] binding, the %% means that the from number will be read from configuration, e.g. local.settings.json in the development environment. Similarly, notice the phone number that the SMS is sent to is read from configuration: var managersMobileNumber = new PhoneNumber(Environment.GetEnvironmentVariable("ManagersMobileNumber"));

Now once per day at 10pm the function will run and send an SMS to the manager giving him the days sales figures.

SHARE:

Executing Multiple Azure Functions When Azure Cosmos DB Documents Are Created or Modified

This is the sixth part in a series of articles.

Sometimes you may want more than one Azure Function to execute when a document  is changed or inserted in Cosmos DB.

You could just use one function that performs multiple logical operations on the changed document but there are some things to consider when doing this:

  • What if the function throws an exception during the first logical operation? (operation 2 may not be executed).
  • What scaling do you want, you/Azure won’t be able to scale the 2 logical operations independently.
  • How long will the function execute for if performing multiple operations in a single function, will you risk function timeouts?
  • How will you monitor operations when they are all contained in a single function.
  • How will you update the code/fix bugs: you will have to update the entire function even if the bug is only related to one operation.
  • How will you write automated tests? They will be more complex if there are multiple operations in a single function.

In some cases you may decide the preceding points don’t matter, but if they do you will need to split the operations into multiple separate Azure Functions.

As an example, the following function contains two logical operations in a single function:

[FunctionName("PizzaDriverLocationUpdated")]
public static void RunMultipleOperations([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            // Simulate running logical operation 1
            log.LogInformation($"Running operation 1 for driver {modifiedDriver.Id} {driverName}");

            // Simulate running logical operation 2
            log.LogInformation($"Running operation 2 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

The preceding function could be separated into two separate functions, each one containing only a single logical operation:

[FunctionName("PizzaDriverLocationUpdated1")]
public static void RunOperation1([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            log.LogInformation($"Running operation 1 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

[FunctionName("PizzaDriverLocationUpdated2")]
public static void RunOperation2([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            log.LogInformation($"Running operation 2 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

If you try and run the function app (for example in the local development functions runtime) you will see some error such as the following:

Unhealthiness detected in the operation AcquireLease for localhost_...==_...=..1 Owner='626d5aec...' Continuation="49" Timestamp(local)=...
Unhealthiness detected in the operation AcquireLease for localhost_...==_...=..0 Owner='626d5aec... Continuation="586" Timestamp(local)=...

If you then make updates/inserts you may see that only one of the two functions is executed, rather than both of them. This is due to change feed leases.

Understanding Azure Cosmos DB Change Feed Leases

The Azure Functions Cosmos DB trigger knows when documents are changed/insert ed by way of the Cosmos DB change feed.

The change feed at a simple level listens for changes made in a collection and allows these changes to be passed to other processes (such as Azure Functions) to do work on.

Without a way to keep track of what changes in the underlying collection have been “fed” out to other process(es) there would be no way to know what changed documents have been passed to external process(es). This is where the lease collection comes in.

The lease collection stores a “checkpoint” for an Azure Function that is using the Cosmos DB trigger. Without this checkpoint, the function would not know if it has processed changed documents or not.

When only one function exists for Cosmos DB collection there is not a problem as only one checkpoint needs to be stored, because there is only one function.

When more that one function exists, there needs to be a way to store different checkpoints for different functions.

One way to do this is to use lease prefixes.

Sharing a Single Lease Collection Across Multiple Azure Functions

To use a single lease collection when you have multiple Azure Functions, you can use the LeaseCollectionPrefix property of the [CosmosDBTrigger] attribute. The value for this property needs to be unique for every function, as the following code demonstrates:

[FunctionName("PizzaDriverLocationUpdated1")]
public static void RunOperation1([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    LeaseCollectionPrefix = "PizzaDriverLocationUpdated1",
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            log.LogInformation($"Running operation 1 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

[FunctionName("PizzaDriverLocationUpdated2")]
public static void RunOperation2([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    LeaseCollectionPrefix = "PizzaDriverLocationUpdated2",
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            log.LogInformation($"Running operation 2 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

In the preceding code, notice LeaseCollectionPrefix = "PizzaDriverLocationUpdated1", and LeaseCollectionPrefix = "PizzaDriverLocationUpdated2".

If the function app is run now there is no startup error and changes made to a document trigger both functions:

Executing 'PizzaDriverLocationUpdated2' (Reason='New changes on collection driver at 2019-05-31T02:49:55.6946671Z', Id=b1476848-7f98-4362-a25f-69beb714c379)
Executing 'PizzaDriverLocationUpdated1' (Reason='New changes on collection driver at 2019-05-31T02:49:55.6946679Z', Id=366fa257-3b94-4d41-94f2-2777e0b8249a)
Running operation 2 for driver 1 Amrit
Running operation 1 for driver 1 Amrit
Executed 'PizzaDriverLocationUpdated2' (Succeeded, Id=b1476848-7f98-4362-a25f-69beb714c379)
Executed 'PizzaDriverLocationUpdated1' (Succeeded, Id=366fa257-3b94-4d41-94f2-2777e0b8249a)

If you check the lease collection behind the scenes notice the lease prefixes in use as the following screenshot shows:

Azure Functions Lease Prefixes

 

Using Multiple Azure Cosmos DB Lease Collections with Azure Functions

Rather than sharing a single lease collection, you can instead specify completely separate collections with the LeaseCollectionName property:

[FunctionName("PizzaDriverLocationUpdated1")]
public static void RunOperation1([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    LeaseCollectionName = "PizzaDriverLocationUpdated1",
    CreateLeaseCollectionIfNotExists = true,
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            log.LogInformation($"Running operation 1 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

[FunctionName("PizzaDriverLocationUpdated2")]
public static void RunOperation2([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    LeaseCollectionName = "PizzaDriverLocationUpdated2",
    CreateLeaseCollectionIfNotExists = true,
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");

            log.LogInformation($"Running operation 2 for driver {modifiedDriver.Id} {driverName}");
        }
    }
}

Notice in the preceding code LeaseCollectionName = "PizzaDriverLocationUpdated1", and LeaseCollectionName = "PizzaDriverLocationUpdated2". Also notice the CreateLeaseCollectionIfNotExists = true, as its name suggests this will create the lease collections if they don’t already exist.

Running the function app and once again and changing a document will result in both functions executing.

Because there are now two separate collections being used for leases there will be a cost associated with having these two collections. Also be sure to read up on RUs for your lease collections, especially if sharing a lease collection and using lease prefixes you should keep an eye on the metrics and make sure you are not getting throttled requests on your lease collection(s).

SHARE:

Processing a Single Azure Cosmos DB Document at a Time With Azure Functions

This is the fifth part in a series of articles.

In previous parts of this series the Azure Function code receives one or more Cosmos DB documents as part of the trigger.

The Azure Functions Cosmos DB trigger uses the change feed from Azure Cosmos to determine when to trigger the function and also what changes to send to the function invocation. If there is only a single updated (or inserted) document then the function will be called and the single document passed to it.

Notice in the following code that the document(s) that need processing are contained in the IReadOnlyList<Document> modifiedDriver collection:

[FunctionName("PizzaDriverLocationUpdated")]
public static async Task Run([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,            
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        log.LogInformation($"Total modified drivers: {modifiedDrivers.Count}");

        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");
            var driverLat = modifiedDriver.GetPropertyValue<double>("Latitude");
            var driverLong = modifiedDriver.GetPropertyValue<double>("Longitude");

            log.LogInformation($"Driver {modifiedDriver.Id} {driverName} was updated (lat,long) {driverLat}, {driverLong}");                    
        }
    }
}

The preceding code just outputs what documents changed to the log/console.

The reason that the foreach loop is required is that the function may receive multiple documents from the change feed.

If you only want to work with a single document per invocation of the function you can restrict the number of documents. To do this the MaxItemsPerInvocation property of the trigger attribute can be set as the following code demonstrates:

[FunctionName("PizzaDriverLocationUpdated")]
public static async Task Run([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    MaxItemsPerInvocation = 1,
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,            
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        log.LogInformation($"Total modified drivers: {modifiedDrivers.Count}");

        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");
            var driverLat = modifiedDriver.GetPropertyValue<double>("Latitude");
            var driverLong = modifiedDriver.GetPropertyValue<double>("Longitude");

            log.LogInformation($"Driver {modifiedDriver.Id} {driverName} was updated (lat,long) {driverLat}, {driverLong}");                    
        }
    }
}

Now there will only be one document in the collection, so the function can be simplified to the following (removing the foreach loop):

[FunctionName("PizzaDriverLocationUpdated")]
public static async Task Run([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    MaxItemsPerInvocation = 1,
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    ILogger log)
{
    if (modifiedDrivers != null && modifiedDrivers.Any())
    {
        var modifiedDriver = modifiedDrivers[0];

        var driverName = modifiedDriver.GetPropertyValue<string>("Name");
        var driverLat = modifiedDriver.GetPropertyValue<double>("Latitude");
        var driverLong = modifiedDriver.GetPropertyValue<double>("Longitude");

        log.LogInformation($"Driver {modifiedDriver.Id} {driverName} was updated (lat,long) {driverLat}, {driverLong}");                
    }
}

Now if 2 updates are made to the database, the function will execute twice, once per changed document:

Executing 'PizzaDriverLocationUpdated' (Reason='New changes on collection driver at 2019-05-28T05:42:25.2143459Z', Id=06d1e318-ab26-4f9b-ac94-3c95fba2dc5c)
Driver 1 Amrit was updated (lat,long) 31.2, 300.3
Executed 'PizzaDriverLocationUpdated' (Succeeded, Id=06d1e318-ab26-4f9b-ac94-3c95fba2dc5c)
Executing 'PizzaDriverLocationUpdated' (Reason='New changes on collection driver at 2019-05-28T05:42:25.5350736Z', Id=34629f4a-66f4-4bbc-85d6-8c3d8c17ee62)
Driver 2 Sarah was updated (lat,long) 222.1, 31.2
Executed 'PizzaDriverLocationUpdated' (Succeeded, Id=34629f4a-66f4-4bbc-85d6-8c3d8c17ee62)

In the next part of this series we’ll look at how to execute multiple different functions when a document changes.

SHARE:

How To Notify Clients of Cosmos DB Changes with Azure SignalR and Azure Functions

This is the fourth part in a series of articles.

The Cosmos DB Azure Functions trigger can be used in conjunction with the Azure SignalR Service to create real-time notifications of changes made to data in Cosmos DB, all in a serverless way.

Azure Cosmos DB Change Notifications with Azure Functions and SignalR

If you want to learn more about how to setup the SignalR Service, check out this previous article.

In this series we’ve been using the domain of pizza delivery. In this article we’ll see how  updates in Cosmos DB can trigger a notification on an HTML client – you may have seen this if you’ve ordered pizza online where the delivery driver is tracked by GPS on a map.

The first thing to do is set up an Azure SignalR Service in Azure and add the SignalR Service connection string in the local.settings.json file which was covered in this article.

We can now add the negotiate function:

[FunctionName("negotiate")]
public static SignalRConnectionInfo GetDriverLocationUpdatesSignalRInfo(
    [HttpTrigger(AuthorizationLevel.Anonymous, "post")] HttpRequest req,
    [SignalRConnectionInfo(HubName = "driverlocationnotifications")] SignalRConnectionInfo connectionInfo)
{
    return connectionInfo;
}

The preceding code sets the SignalR hub to be used in this example as “driverlocationnotifications”, we’ll use the same hub name in the function that actually sends updates to clients.

The next thing to do is add a function that is triggered when changes are made to the location of pizza delivery drivers (note the trigger will also fire for new documents that are added):

[FunctionName("PizzaDriverLocationUpdated")]
public static async Task Run([CosmosDBTrigger(
    databaseName: "pizza",
    collectionName: "driver",
    ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> modifiedDrivers,
    [SignalR(HubName = "driverlocationnotifications")] IAsyncCollector<SignalRMessage> signalRMessages,
    ILogger log)
{
    if (modifiedDrivers != null)
    {
        log.LogInformation($"Total modified drivers: {modifiedDrivers.Count}");

        foreach (var modifiedDriver in modifiedDrivers)
        {
            var driverName = modifiedDriver.GetPropertyValue<string>("Name");
            var driverLat = modifiedDriver.GetPropertyValue<double>("Latitude");
            var driverLong = modifiedDriver.GetPropertyValue<double>("Longitude");

            log.LogInformation($"Driver {modifiedDriver.Id} {driverName} was updated (lat,long) {driverLat}, {driverLong}");

            var message = new DriverLocationUpdatedMessage
            {
                DriverId = modifiedDriver.Id,
                DriverName = driverName,
                Latitude = driverLat,
                Longitude = driverLong
            };

            await signalRMessages.AddAsync(new SignalRMessage
            {
                Target = "driverLocationUpdated",
                Arguments = new[] { message }
            });
        }
    }
}

The preceding code uses the [CosmosDBTrigger] to fire the function when there are changes made to the “driver” collection in the “pizza” database.

The [SignalR] binding attribute allows outgoing SignalR messages to be sent to connected clients by adding messages to the IAsyncCollector<SignalRMessage> signalRMessages object. Notice that the type of the Cosmos DB trigger is IReadOnlyList<Document>, to get the actual data items of the document we use the GetPropertyValue method, for example to get the driver name: modifiedDriver.GetPropertyValue<string>("Name").

In the HTML, we can connect to the “driverlocationnotifications” SignalR hub, and then get notifications when any driver location changes, in reality we would probably only want to get messages for the driver that is delivering our pizza but for simplicity we won’t worry about it in this demo code.

The following is the HTML to make this work:

<html>

<head>
     <!--Adapted from: https://azure-samples.github.io/signalr-service-quickstart-serverless-chat/demo/chat-v2/ -->

    <title>SignalR Demo</title>
    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@4.1.3/dist/css/bootstrap.min.css">

    <style>
        .slide-fade-enter-active, .slide-fade-leave-active {
            transition: all 1s ease;
        }

        .slide-fade-enter, .slide-fade-leave-to {
            height: 0px;
            overflow-y: hidden;
            opacity: 0;
        }
    </style>
</head>

<body>
    <p>&nbsp;</p>
    <div id="app" class="container">
        <h3>Your Pizza Is On Its Way!!</h3>

        <div class="row" v-if="!ready">
            <div class="col-sm">
                <div>Loading...</div>
            </div>
        </div>
        <div v-if="ready">
            <transition-group name="slide-fade" tag="div">
                <div class="row" v-for="driver in drivers" v-bind:key="driver.DriverId">
                    <div class="col-sm">
                        <hr />
                        <div>
                            <div style="display: inline-block; padding-left: 12px;">
                                <div>
                                    <span class="text-info small"><strong>{{ driver.DriverName }}</strong> just changed location:</span>
                                </div>
                                <div>
                                    {{driver.Latitude}},{{driver.Longitude}}
                                </div>
                            </div>
                        </div>
                    </div>
                </div>
            </transition-group>
        </div>
    </div>

    <script src="https://cdn.jsdelivr.net/npm/vue@2.5.17/dist/vue.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/@aspnet/signalr@1.1.2/dist/browser/signalr.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/axios@0.18.0/dist/axios.min.js"></script>

    <script>
        const data = {
          drivers: [],
          ready: false
        };

        const app = new Vue({
          el: '#app',
          data: data,
          methods: {
          }
        });

        const connection = new signalR.HubConnectionBuilder()
          .withUrl('http://localhost:7071/api')
          .configureLogging(signalR.LogLevel.Information)
          .build();

        connection.on('driverLocationUpdated', driverLocationUpdated);
        connection.onclose(() => console.log('disconnected'));

        console.log('connecting...');
        connection.start()
            .then(() => data.ready = true)
            .catch(console.error);

        let counter = 0;

        function driverLocationUpdated(driverLocationUpdatedMessage) {
        driverLocationUpdatedMessage.id = counter++; // vue transitions need an id
        data.drivers.unshift(driverLocationUpdatedMessage);
    }
    </script>
</body>

</html>

Now when changes are made to the documents in the driver collection, the Azure Function will execute and send out Azure SignalR Service messages to clients. In this example we’re just writing out the messages in the page, but you can image the latitude and longitude being used to draw a little car image on top of a map, etc.

SHARE:

Writing Azure Cosmos DB Data from Azure Functions

This is the third part in a series of articles.

In part 2 of this series we saw how to read data from Cosmos DB, in this article we’ll see how to write data to Cosmos DB from an Azure Function.

As with input, output is achieved by the use of the [CosmosDB] binding attribute and as before, the database name, collection name, id, partition, and connection string can be defined.

In part 2 we saw how to use binding expressions to take data from the HTTP querystring and use that to select a document to read in (e.g. by Id). In this article we’ll see how we can do a similar thing when the trigger type is for a storage queue.

The following function is triggered when a message is written to the “update-pizza-driver-location” queue:

[FunctionName("UpdatePizzaDriverLocation")]
[return: CosmosDB(databaseName: "pizza", collectionName: "driver", ConnectionStringSetting = "pizzaConnection", Id = "{Id}", PartitionKey = "{StoreId}")]
public static Driver Run(
    [QueueTrigger("update-pizza-driver-location")] PizzaDriverLocationUpdate locationUpdate,
    [CosmosDB(databaseName: "pizza", collectionName: "driver", ConnectionStringSetting = "pizzaConnection", Id = "{Id}", PartitionKey = "{StoreId}")] Driver driver,            
    ILogger log)
{
    if (driver is null)
    {
        log.LogError($"Driver Id/partition {locationUpdate.Id}/{locationUpdate.StoreId} not found in database.");
        return null;
    }

    driver.Latitude = locationUpdate.NewLat;
    driver.Longitude = locationUpdate.NewLong;

    return driver;
}

The first thing to notice in the preceding function that there are 2 instances of the [CosmosDB] binding attribute, one as the return binding for the method (the output binding to perform the update) and the input binding in the Run method parameter (the input binding to read the current state of the Driver).

Another thing to notice is the use of the {Id} and {StoreId} binding expressions. These expressions assume that the incoming queue message contains JSON properties that match these expressions, for example:

{
    "Id" : "1",
    "StoreId" : "101",
    "NewLat" : 111.2,
    "NewLong" : 3110.3
}

The Driver and PizzaDriverLocationUpdate classes look like the following:

public class PizzaDriverLocationUpdate
{
    public string Id { get; set; }
    public string StoreId { get; set; }
    public double NewLat { get; set; }
    public double NewLong { get; set; }
}

 

public class Driver
{
    [JsonProperty(PropertyName = "id")]
    public string Id { get; set; }
    public string StoreId { get; set; }
    public string Name { get; set; }
    public double Latitude { get; set; }
    public double Longitude { get; set; }
}

Note in the Driver class, the [JsonProperty(PropertyName = "id")] attribute is mapping Id to id in Cosmos DB.

If you want to write multiple documents from a single function invocation you make use of an IAsyncCollector, for example IAsyncCollector<Driver> drivers and then use drivers.AddAsync(newDriver) method to write multiple documents.

In the next part of this series we’ll see how we can combine Azure Cosmos DB triggers with SignalR to create notifications to clients when data changes.

SHARE:

Reading Azure Cosmos DB Data In Azure Functions

This is the second part in a series of articles.

In addition to triggering a function when Cosmos DB data changes (as we saw in part one) you can also read data in from Cosmos DB when a function executes. The simplest way to do this is to use an input binding.

The [CosmosDB] binding attribute can be used both as an input binding and an output binding. When used as an input binding it allows one or more documents to be retrieved from the database.

When using the attribute there are a number of ways to configure it including:

  • The Cosmos DB database name
  • The collection name
  • The document Id to retrieve
  • The partition key
  • And the connection string app setting

Reading a Single Cosmos DB Document in an Azure Function

The following code shows an Azure Function that is triggered from an HTTP GET request:

[FunctionName("GetDriver")]
public static async Task<IActionResult> Run(
    [HttpTrigger(AuthorizationLevel.Function, "get", Route = null)] HttpRequest req,
    [CosmosDB(databaseName: "pizza",collectionName: "driver", Id = "{Query.id}", PartitionKey = "{Query.storeId}", ConnectionStringSetting = "pizzaConnection")] Driver driver,
    ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");

    if (driver is null)
    {
        return new NotFoundResult();
    }

    return new OkObjectResult(driver);
}

Notice first the [CosmosDB] binding attribute. The combination of the Id and PartitionKey will determine the Driver object (if any) that will be retrieved. Note the format of these two: {Query.id} and {Query.storeId}.This will look for query string parameters in the incoming HTTP GET called id and storeId, for example: http://localhost:7071/api/GetDriver?id=1&storeId=101

If no document is found, driver will be null.

If the document was found it will be returned as JSON to the caller.

Reading Multiple Cosmos DB Documents in an Azure Function Using SqlQuery

The following function will get the latest 100 drivers (as sorted by the built in timestamp _ts property):

[FunctionName("GetDrivers")]
public static async Task<IActionResult> GetDrivers(
    [HttpTrigger(AuthorizationLevel.Function, "get", Route = null)] HttpRequest req,
    [CosmosDB(databaseName: "pizza", collectionName: "driver", SqlQuery = "SELECT top 100 * FROM driver order by driver._ts desc", ConnectionStringSetting = "pizzaConnection")] IEnumerable<Driver> drivers,
    ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");

    if (drivers is null)
    {
        return new NotFoundResult();
    }

    foreach (var driver in drivers)
    {
        log.LogInformation(driver.Name);
    }

    return new OkObjectResult(drivers);
}

There’s a couple of things to notice in the preceding code. The first is that instead of a single Driver, the binding now returns IEnumerable<Driver> drivers. Also notice that the binding no longer has Id and PartitionKey.

Reading Multiple Cosmos DB Documents Based on Query String Parameter

A more advanced technique is to bind to an instance of DocumentClient. This allows more fine grained/low level/more specific access of data, such as using a LINQ query to perform the search:

[FunctionName("GetDriversForStore")]
public static async Task<IActionResult> GetDriversForStore(
    [HttpTrigger(AuthorizationLevel.Function, "get", Route = null)] HttpRequest req,
    [CosmosDB( ConnectionStringSetting = "pizzaConnection")] DocumentClient client,
    ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");
    
    // Validation and error checking omitted for demo purposes
               
    string storeId = req.Query["storeId"]; // read storeId to get driver for from querystring

    Uri driverCollectionUri = UriFactory.CreateDocumentCollectionUri(databaseId: "pizza", collectionId: "driver");

    var options = new FeedOptions { EnableCrossPartitionQuery = true }; // Enable cross partition query

    IDocumentQuery<Driver> query = client.CreateDocumentQuery<Driver>(driverCollectionUri, options)
                                         .Where(driver => driver.StoreId == storeId)
                                         .AsDocumentQuery();

    var driversForStore = new List<Driver>();

    while (query.HasMoreResults)
    {
        foreach (Driver driver in await query.ExecuteNextAsync())
        {
            driversForStore.Add(driver);
        }
    }                       

    return new OkObjectResult(driversForStore);
}

The preceding code returns all drivers that belong to a specific store passed in as a querystring storeId parameter.

In the next part of this series we’ll see how to write data out to Cosmos DB when a function executes.

SHARE:

Getting Started with Azure Cosmos DB and Azure Functions

This is the first part in a series of articles.

Azure Cosmos DB is a “globally distributed, multi-model database service. With a click of a button, Cosmos DB enables you to elastically and independently scale throughput and storage across any number of Azure regions worldwide. You can elastically scale throughput and storage, and take advantage of fast, single-digit-millisecond data access using your favorite API including SQL, MongoDB, Cassandra, Tables, or Gremlin” [Microsoft]

One way to respond to changes in Cosmos is to use Azure Functions. When changes occur (currently limited to inserts and updates, e.g. not deletions ) the Function can be notified and executes.

Azure Function Cosmos DB triggers under the hood make use of the Azure Cosmos DB change feed to know when to execute functions.

Installing the Azure Cosmos Emulator

You can use the Azure Cosmos Emulator to get started locally without even needing an Azure subscription.

Once the emulator is downloaded and installed on your development PC (a Docker version is also available) you can start the  emulator from the start menu.

When running you can see the icon in the Windows taskbar notification area and it will pop up a browser window pointing to https://localhost:8081/_explorer/index.html as the following screenshot shows:

Azure Cosmos Emulator portal

Notice in the preceding screenshot the Primary Connection String, you will need to copy this for use in the local app settings later.

Creating an Azure Function Triggered from Cosmos DB

Once you have created an Azure Function project in Visual Studio, you need to add the Microsoft.Azure.WebJobs.Extensions.CosmosDB NuGet package to get access to the .NET C# bindings.

Now the package is installed (and with an up to date installation of Visual Studio) you can right click the Azure Functions project and choose Add –> New Azure Function…

Give the function a name and for the trigger type choose Cosmos - this will create a boiler plate function like the following:

public static class PizzaDriverLocationUpdated
{
    [FunctionName("PizzaDriverLocationUpdated")]
    public static void Run([CosmosDBTrigger(
        databaseName: "databaseName",
        collectionName: "collectionName",
        ConnectionStringSetting = "",
        LeaseCollectionName = "leases")]IReadOnlyList<Document> input, ILogger log)
    {
        if (input != null && input.Count > 0)
        {
            log.LogInformation("Documents modified " + input.Count);
            log.LogInformation("First document Id " + input[0].Id);
        }
    }
}

The preceding code uses the [CosmosDBTrigger] attribute to tell this function to execute when there are changes as specified by the following attribute parameters/properties:

  • databaseName - Azure Cosmos DB database with the monitored collection
  • collectionName – Collection being monitored for changes
  • ConnectionStringSetting – App setting name containing the connection string to the Azure Cosmos DB being monitored
  • LeaseCollectionName - name of the collection used to store leases, defaults to “leases” if not specified

The LeaseCollectionName is required by the trigger to store leases over Cosmos DB partitions, one thing to note: “If multiple functions are configured to use a Cosmos DB trigger for the same collection, each of the functions should use a dedicated lease collection or specify a different LeaseCollectionPrefix for each function. Otherwise, only one of the functions will be triggered” [Microsoft]

How to Create a Database and Collection in the Azure Cosmos Emulator

In the emulator portal in the browser, click Explorer.This will allow you to create collections in the emulator.

Click the New Collection button and enter the following:

  • Database id: pizza
  • Collection Id: driverLocation
  • Partition key: /storeId

Creating a new collection in the Azure Cosmos DB Emulator

Save this new collection.

Configuring an Azure Function Cosmos DB Trigger

Modify the function that was created earlier to be as follows:

public static class PizzaDriverLocationUpdated
{
    [FunctionName("PizzaDriverLocationUpdated")]
    public static void Run([CosmosDBTrigger(
        databaseName: "pizza",
        collectionName: "driverLocation",
        ConnectionStringSetting = "pizzaConnection")] IReadOnlyList<Document> input,
        ILogger log)
    {
        if (input != null && input.Count > 0)
        {
            log.LogInformation("Documents modified " + input.Count);
            log.LogInformation("First document Id " + input[0].Id);
        }
    }
}

Notice in the preceding code that the databaseName and collectionName settings match what we just created in the emulator.

The ConnectionStringSetting is set to “pizzaConnection” – this needs to be in the function settings, in the case of local development in the local.settings.json file:

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "dotnet",
    "pizzaConnection": "AccountEndpoint=https://localhost:8081/;AccountKey=C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw=="
  },
  "Host": {
    "LocalHttpPort": 7071,
    "CORS": "http://localhost:3872",
    "CORSCredentials": true
  }
}

Notice in the preceding settings, the value for the pizzaConnection item is the Primary Connection String copied from the Azure Cosmos Emulator.

Testing the Function Locally

Now that the function code is configured and the database and collection exist in the emulator, hit F5 in Visual Studio (or click Run) and the local functions runtime will start.

Once the runtime has started, head back to the Cosmos Emulator in the browser, click on Pizza –> driverLocation –> Documents and click the New Document button.

Creating a new document in the Cosmos DB emulator

Add the following:

{
    "id": "1",
    "storeId": 42,
    "name": "Sarah",
    "lat" : 52,
    "long": 2
}

And click Save.

Head back to the locally running functions runtime window and you will see the function has noticed this new document and executed:

Executing 'PizzaDriverLocationUpdated' (Reason='New changes on collection driverLocation at 2019-05-10T04:10:29.9913171Z', Id=203ae791-c7d4-4270-ac6b-501f313c3805)
Documents modified 1
First document Id 1
Executed 'PizzaDriverLocationUpdated' (Succeeded, Id=203ae791-c7d4-4270-ac6b-501f313c3805)

If you head back to the emulator, modify the document (e.g. change the name to “Amrit”) and click Update, the function will trigger a second time:

Executing 'PizzaDriverLocationUpdated' (Reason='New changes on collection driverLocation at 2019-05-10T04:13:00.8233214Z', Id=778caddd-9f19-42fa-9d9d-6c5dd5892a25)
Documents modified 1
First document Id 1
Executed 'PizzaDriverLocationUpdated' (Succeeded, Id=778caddd-9f19-42fa-9d9d-6c5dd5892a25)

SHARE:

Returning HTTP Status Codes from Azure Functions

(This post refers to Azure Functions v2)

When creating HTTP-triggered Azure Functions there are a number of ways to indicate results back to the calling client.

Returning HTTP Status Codes Manually

To return a specific status code to the client you can create an instance of one of the …Result classes and return that from the function body.

The following example returns an instance of an OkResult or a BadRequestResult:

[FunctionName("AddActor1")]
public static async Task<IActionResult> AddActor1(
    [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequest req,
    ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");

    string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
    dynamic data = JsonConvert.DeserializeObject(requestBody);

    string name = data.actorName; // get name from dynamic/JSON object

    if (name == null)
    {
        // Return a 400 bad request  result to the client
        return new BadRequestResult();
    }

    // Do some processing
    char firstLetter = name[0];

    // Return a 200 OK to the client
    return new OkResult();                
}

If you wanted to provide additional success/failure information you could use the OkObjectResult and BadRequestObjectResult classes instead, these allow you to provide additional contextual information to the client:

[FunctionName("AddActor2")]
public static async Task<IActionResult> AddActor2(
    [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequest req,
    ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");

    string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
    dynamic data = JsonConvert.DeserializeObject(requestBody);

    string name = data.actorName; // get name from dynamic/JSON object

    if (name == null)
    {
        // Return a 400 bad request result to the client with additional information
        return new BadRequestObjectResult("Please pass an actorName in the request body");
    }

    // Do some processing
    char firstLetter = name[0];

    // Return a 200 OK to the client with additional information
    return new OkObjectResult($"Actor {name} was added");
}

Automatically Returning Status Codes

In addition to manually returning status code instances, you can let the functions runtime take care of this for you.

For example, the following code will automatically return a “204 no content” if the function executes without throwing an exception, or a “500 internal server error” if an exception was thrown:

[FunctionName("AddActor3")]
public static async Task AddActor3(
    [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequest req,
    ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");

    string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
    dynamic data = JsonConvert.DeserializeObject(requestBody);

    string name = data.actorName; // get name from dynamic/JSON object

    // Do some processing
    char firstLetter = name[0]; // 500 internal server error if name is null

    // Auto return a 204 no content if no exception was thrown
}

In the preceding code, if the client fails to provide a actorName in the JSON, rather then getting a more helpful “400 bad request” (with optional additional message), they instead get a less useful “500 internal server error” status code and they have no idea what may have gone wrong or how to resolve it.

In this way, automatic status codes can be helpful if you want to write less code or perhaps use the return value of the function in a binding as in the following example:

[FunctionName("AddActor4")]
[return: Queue("new-actor-first-letter")]
public static async Task<string> AddActor4(
    [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequest req,
    ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");

    string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
    dynamic data = JsonConvert.DeserializeObject(requestBody);

    string name = data.actorName;

    return name.Substring(0,1); // add a new message to the queue containing the first letter of the name
}

Once again, the preceding function will return a 500 if there is an exception (e.g. actorName not provided in JSON) but will return a “200 OK” if no exception occurs (rather than the “204 no content” in the earlier example).

SHARE: