Zoals ik aan het einde van de vorige post al schreef, hebben veel workflow engines de mogelijkheid om het systeem te pauzeren tot er een signaal van buiten komt die aangeeft dat we verder kunnen.

Nu is Durable Functions geen echte workflow engine maar we kunnen dit wel hier ook inbouwen. Nu is het op zich niet zo ingewikkeld, maar er zitten wat haken en ogen aan. Immers: op het moment dat je de mens een onderdeel maak van een geautomatiseerd proces moet je rekening houden met het grillige gedrag van mensen. Computers zijn voorspelbaar. Mensen niet.

Het proces is dus vrij recht-toe, recht-aan. Maar we willen graag een robuuste omgeving maken die rekening houdt met alle eventualiteiten.

De proces stap die we hier behandelen is de volgende: stap 3 in het proces, genaamd “Get Approval”

Het idee is dat wij als bestuur van Stichting dotNed toch wel even kijken wie zich allemaal aanmelden zodat we sommige mensen kunnen weigeren (doen we nooit hoor…) Hoe doen we dat? Simpel: op het moment dat iemand zich aanmeldt, krijgen we een mailtje in een centrale mailbox met daarin de tekst:

“Persoon {x} heeft zich aangemeld voor de bijeenkomst. Klik {link A} om deze persoon te accepteren en klik {link B} om deze persoon te weigeren.”

Of woorden van die strekking. Door link A of link B te kiezen kunnen we onze voorkeur aangeven.

Onze worfklow zal dus een mail moeten versturen. Maar: daarna moet de flow gepauzeerd worden tot iemand op een van beide links geklikt heeft. Die link zal dan een trigger moeten hebben die de workflow weer aanzwengelt zodat we verder kunnen gaan met het accepteren of afwijzen.

Nu hebben wij als bestuur ook nog een ‘normale’ baan, we hebben een gezin, we hebben het best druk. Dus we missen nog wel eens een dergelijk mailtje. Wat moeten we dan doen? Het is een beetje flauw om de mensen die zich willen aanmelden heel lang te laten wachten. Dus we hebben afgesproken dat als het bestuur niet binnen 24 uur reageert, het proces er van uitgaat dat we deze persoon wel goed keuren. En dan gaat het proces weer verder.

Dat houdt dus in dat we in de gaten moeten houden wanneer een mail verstuurd wordt en dan moet er een timer gaan lopen. We moeten die 24 uur in de gaten houden.

We hebben nu een paar uitdagingen:

1. Hoe pauzeren we een flow?
2. Hoe triggeren we de flow zodat hij weer gaat lopen?
3. Hoe bouwen we een time out in?
4. Hoe zorgen we er voor dat het systeem veilig blijft en hack-vrij (we hebben te maken met professionele it’ers die langs willen komen, die vinden niets leuker dan het proces hacken).
5. Hoe bouwen we dit zonder het overzicht kwijt te raken?

Als je mee bouwt, ga dan even naar de website van SendGrid. SendGrip is een API waarmee je emais kunt versturen via een eenvoudige REST API. Het account is gratis maar dan heb je een beperking van maximaal 100 berichten per dag. Voor ons is dat meer dan genoeg natuurlijk. Als je je aanmeldt, kun je een API key aanmaken. Ga daarvoor op de SendGrid site naar je Dashboard. In het menu links onderin staat “Settings”. Kies daar voor API Keys:

Klik dan op Create API Key. Hou de key geheim! Deel die niet! Je kunt ze altijd weer revoken maar met deze key kan iedereen op jouw account mails versturen. Er zijn heel veel spammers die dat best fijn zouden vinden.

De key moet een naam hebben, en je kunt bepalen wat je er mee kunt. In dit geval laat ik hem op Full Access staan.

Je krijgt nu een nieuwe key, Dit is de enige keer dat je hem te zien krijgt op de website van SendGrid dus maak er een kopie van (in jouw geval zal er geen zwarte balk over heen staan, maar ik wil mijn key toch geheim houden.) Ik sla de key op in mijn local.settings.json file zodat ik hem later kan uitlezen:

Goed. Dat hebben we gedaan.

Wachten op events: twee stappen

Het proces is op te knippen in twee losse stappen:

1. Het versturen van de email, waarna het proces wacht
2. Het verder gaan van het proces nadat een link is geklikt of de time-out verstreken is.

Laten we eens beginnen met stap 1. Om precies te zijn: we gaan eerst de activity maken.

Aangezien we SendGrid gebruiken voor het versturen van de mail, moeten we de SendGrid Nuget package gebruiken. Laten we die eerst toevoegen. Er zijn er een aantal verschillende in de Nuget gallery, dus pak wel de goede. Namelijk: Microsoft.Azure.Webjobs.Extensions.SendGrid.

Veilig omgaan met workflow instances

Het is tijd voor de activity. Zoals we al gezien hebben, krijgt een activity niet meer informatie dan dat hij nodig heeft om zijn werk te doen. Maar wat hebben we nu nodig in onze email-verzend activity? In ieder geval data over de bezoeker. Ik hou het minimaal en geef alleen het email adres van de bezoeker mee.

Maar: als het proces gepauzeerd wordt en daarna weer verder moet gaan, moet de workflow wel weten welke instance we verder willen laten gaan. Dat moet op de een of andere manier in de link gezet worden die we naar het bestuur versturen. Dus we moeten de instanceId van de workflow ook meegeven. We maken daar dus weer een DTO van:

namespace MeetingRegistration
{
    public class BoardEmailData
    {
        public string InstanceId { get; set; }
        public string Email { get; set; }
    }
}

Er is wel een puntje met betrekking tot de InstanceId. Dat is eigenlijk gevoelige informatie die je niet wilt delen met de buitenwereld. Ok, om iets zinnigs te doen met de workflow (weet je nog die statusQueryGetUri die we gebruiken?) moet je wel een secret key hebben maar als je die hebt EN je hebt de instance Id dan heb je ineens de mogelijkheid om het proces van buiten af te beinvloeden. Dus hoe minder info we delen met de buitenwereld hoe beter. Daarnaast: de InstanceId is een implementatie nummer; die hoor je sowieso niet te delen. We moeten dus wat anders verzinnen.

De oplossing is als volgt:

1. We maken een storage ergens en daarin slaan we de InstanceId op.
2. We gebruiken als key voor die entry in te storage een random, uniek nummer (een GUID zou mooi zijn).
3. Als iemand de link klikt, dan geven we in die link de key mee (die GUID dus). De link verwijst naar een nieuwe Azure Function (niet durable, gewoon een GET based, HTTP Triggered Function zoals onze starter function).
4. Als parameter hebben we in die function nu de GUID. Daarmee halen we de InstanceId van de workflow op
5. We zeggen tegen de Durable Functions Runtime dat de workflow met dit unieke ID verder mag gaan, en we geven de approved of denied status mee.

Klinkt ingewikkeld? Valt heel erg mee!

We gaan met TableStorage werken, dus we moeten ook daarvoor de Nuget packages installeren: we gebruiken Microsoft.Azure..Webjobs.Extensions.Storage.

Goed. Tijd voor wat code. Ik geef de lege activity alvast, dan doen we de implementatie later als ik uitgelegd heb hoe het een en ander werkt.

We hebben een class nodig die de data voorstelt die we gaan opslaan in de TableStorage.

using System;
using System.Collections.Generic;
using System.Text;

namespace MeetingRegistration
{
    public class StoredOrchestration
    {
        public string PartitionKey { get; set; }
        public string RowKey { get; set; }
        public string OrchestrationId { get; set; }
    }
}

Deze class bevat de data die we op gaan slaan in de database. Als je nog niet eerder met TableStorage gewerkt heb: het is redelijk simpel.

De TableStorage heeft een Table (duh) en die tabel bevat records. Ieder record of entry heeft een composite key: een key die bestaat uit twee delen. Als eerste is er de Partitionkey en als tweede de Rowkey. We gebruiken een vaste PartitionKey om aan te geven dat deze entry gaat over een goed te keuren attendee. De Rowkey wordt die random GUID die we gebruiken. Als laatste heb je de mogelijkheid om zelf kolommen toe te voegen. Wij gebruiken er maar 1: de OrchestrationId die aangeeft welke instance van de workflow we gebruiken.

Waarom slaan we de rest van de data niet op, kun je je afvragen? Het ligt voor de hand om ook het email adres en/of de naam van de bezoeker op te slaan. Het antwoord is natuurlijk: we hebben het niet nodig. Als we de orchestrationId hebben, kan de orchestrator daarmee alle data ophalen die nodig is. Weet je nog? Als de activity klaar is, gaat de orchestrator opnieuw beginnen en haalt alle data op uit zijn eigen TableStorage. Daarin hebben we de naam en het email adres.

Laten we een begin maken met de activity zelf:

[FunctionName("A_SendBoardEmail")]
public static Task SendBoardEmail(
    [ActivityTrigger] BoardEmailData boardEmailData,
    [Table("Attendees", Connection = "AzureWebJobsStorage")] 
    out StoredOrchestration storedOrchestration,
    ILogger log)
{
    var uniqueKey = Guid.NewGuid().ToString("N");

    storedOrchestration = new StoredOrchestration()
    {
        PartitionKey = "AttendeeApprovals",
        RowKey = uniqueKey,
        OrchestrationId = boardEmailData.InstanceId
    };

    return Task.CompletedTask;
}

Er gebeurt nogal wat hier. We gaan hem even uit elkaar trekken.

Als eerste de FunctionName. Die kennen we nu wel. De activity zelf is een public static async method die een Task teruggeeft. De eerste parameter is de BoardEmailData die we net gemaakt hebben en die markeren we als ActivityTrigger. Tot zover geen rocket science. Dan de tweede parameter op regel 4 en 5: we maken gebruik van TableStorage. Hoe dit werkt is als volgt:

Door het gebruik van het attribuut [Table] geven we aan dat de data in storedOrchestration in de database moet komen of daar van uit moet komen. We geven de tabel naam mee (“Attendees”) en we geven de Connection mee. Dit is de naam van de setting in je local.settings.json file en bij ons staat dat op UseDevelopmentStorage=true. Dat is de default waarde, in productie verander je dat naar het echte storage account die je wilt gaan gebruiken.

Doordat we de StoredOrchestration als ‘out’ parameter aangeven, zal de runtime een entry in de tabel plaatsen als je een nieuwe instance van die class aanmaakt en de functie verlaten wordt. We hoeven dus niets te doen met connections, insert-statements, database code en de rest. De code die je nu ziet is genoeg om een entry in de TableStorage te plaatsen.

We maken een unieke sleutel aan. Dit is dus iets wat wel in de activity mag maar niet in een orchestration functie. We zitten in de activity dus dat werkt.

We maken een instance van de StoredOrchestration aan. De PartitionKey zet ik op AttendeeApprovals en de Rowkey op de nieuwe key die we net gemaakt hebben. We geven ook de orchestrationId mee die we mee gaan krijgen vanuit de orchestration.

Laten we dat gelijk maar even regelen: in de hoofd orchestration, na het aanroepen van de sub orchestration, plaatsen we de volgende regels:

if(!ctx.IsReplaying)
    log.LogWarning("Sending email to the dotNed board members.");

await ctx.CallActivityAsync("A_SendBoardEmail",
    new BoardEmailData {Email = attendee.Email, InstanceId = ctx.InstanceId});

Niets nieuws hier. We loggen (eenmalig!) en we roepen de activity aan met de data die we nodig hebben.

We moeten nu alleen de email nog versturen. Dat doen we in de activity en dat gaan we nu doen.

Om dat voor elkaar te krijgen moeten we de signature van de method aanpassen. Na de [Table] parameter maken we nu ook een [SendGrid] parameter:

[FunctionName("A_SendBoardEmail")]
public static Task SendBoardEmail(
    [ActivityTrigger] BoardEmailData boardEmailData,
    [Table("Attendees", Connection = "AzureWebJobsStorage")]
    out StoredOrchestration storedOrchestration,
    [SendGrid(ApiKey = "SendGridAPIKey")] 
    out SendGridMessage sendGridMessage,
    ILogger log)
{

We hebben net gezien dat als we een out parameter hebben en die markeren met [Table] dat die dan automatisch een entry aanmaakt in de TableStorage. Voor SendGrid hebben we een zelfde techniek: we definieren een out parameter van het type SendGridMessage (dit is de email) en we markeren deze met [SendGrid]. We geven de API key mee, dat is die in de local.settings.json staat (zie boven in deze post). De runtime zal deze er in zetten.

Dus: als we nu een instance aanmaken van deze SendGridMessage zal SendGrid deze automatisch versturen. Geen gedoe met het aanmaken van REST calls, aanroepen van de API enzovoorts. Komt ie!

[FunctionName("A_SendBoardEmail")]
public static Task SendBoardEmail(
    [ActivityTrigger] BoardEmailData boardEmailData,
    [Table("Attendees", Connection = "AzureWebJobsStorage")]
    out StoredOrchestration storedOrchestration,
    [SendGrid(ApiKey = "SendGridAPIKey")] 
    out SendGridMessage sendGridMessage,
    ILogger log)
{
    var uniqueKey = Guid.NewGuid().ToString("N");

    storedOrchestration = new StoredOrchestration
    {
        PartitionKey = "AttendeeApprovals",
        RowKey = uniqueKey,
        OrchestrationId = boardEmailData.InstanceId
    };

    var myHostName = System.Environment.GetEnvironmentVariable("WEBSITE_HOSTNAME");

    var linkBase = @"http://{0}/api/processApproval/{1}/{2}";
    var approval = string.Format(linkBase, myHostName, uniqueKey, "true");
    var deny = string.Format(linkBase, myHostName, uniqueKey, "false");

    var body =
        "<html>" +
        "<h1>" +
        "Approval needed" +
        "</h1>" +
        "<p>" +
        $"Please approve at<a href= \"{approval}\" > this link</a>." +
        "<br />" +
        $"Or deny at<a href= \"{deny}\" > this link</a>" +
        "<br/>" +
        $"for user with email {boardEmailData.Email}." +
        "</p>" +
        "<p>" +
        "<h3>" +
        "Thanks!" +
        "</h3>" +
        "</p> " +
        "</html> ";

    var from = new EmailAddress("dennis@vroegop.org");
    var to = new EmailAddress("dotneddemo@outlook.com");

    sendGridMessage = new SendGridMessage()
    {
        From = from,
        Subject = "Please approve!",
        HtmlContent = body
    };
    sendGridMessage.AddTo(to);

    return Task.CompletedTask;
}

Ik geef je even de hele activity. Ik maak een paar strings aan voor de links. De host name vraag ik op via System.Environment.GetEnvironmentVariable(“WEBSITE_HOSTNAME”). In ons geval geeft dat de string “localhost:7071” terug. Die plakken we in de base string. Die base string zal er dus als volgt uit zien: http://localhost:7071/api/processApproval/%5BmyGuid%5D/true (of false bij afwijzen. Deze URL gaat verwijzen naar een nieuw te bouwen HTTP triggered Azure Function. Die maken we de volgende post wel.

Daarna maken we de HTML body voor de mail, zetten de adressen goed en maken een instance van de SendGridMessage aan. Dat zorgt ervoor dat de mail verzonden wordt als we de activity verlaten.

Als we dit nu runnen, zal het systeem een entry aanmaken in de tabel Attendees (als die nog niet bestaat, wordt die gelijk maar even aangemaakt) en daarna wordt er een email verstuurd. Ik verstuur hem even vanaf mijn eigen mail adres naar een dummy bestuurs-mailbox, dus pas dat aan.

In de Storage Explorer kun je het resultaat zien:

Je ziet: er is een nieuwe tabel aangemaakt en deze heeft nu de info over de orchestrationId die we moeten gaan goed- of afkeuren.

In mijn mailbox zie ik het volgende:

De link doet het niet natuurlijk, maar zou je er op klikken dan zie je in de URL dat de querystring parameter voor de unique key hetzelfde is als de Rowkey in de tabel. Dus in de nog te schrijven Azure Function kunnen we die data uit de table storage halen, de OrchestrationId ophalen en verder gaan.

Wachten…

Prima. Maar we wachten nog steeds niet en dat was het onderwerp van deze post. Hoe doen we dat? We moeten wachten tot iemand op een van die twee links klikt.

We gaan terug naar onze orchestrator. Na de call naar A_SendBoardEmail schrijven we de volgende code:

var boardHasApproved = await ctx.WaitForExternalEvent<bool>(
    "E_BoardHasApproved", 
    TimeSpan.FromSeconds(30), 
    true);

if (!boardHasApproved)
{
    return new RegistrationResult()
    {
        IsSucces = false,
        Reason = "Board did not approve. Sorry!"
    };
}

We roepen in de context de method WaitForExternalEvent<bool>(“E_BoardHasApproved”) aan. Er moet dus een externe event gebeuren willen we verder gaan. Tot die tijd is de orchestration niet meer levend, net als met activities. We geven een time-out mee van 30 seconden. Dat is wat kort om de board te laten reageren op de email, maar ik heb geen zin om voor iedere test 24 uur te moeten wachten. Idealiter sla je dit op in de Config (en lees die uit in een aparte activity!) maar nu doen we het even zo.

Als die timeout plaatsvind voor het event er is geweest, krijgen we een TimeOutException. Die kan je afvangen en melden dat er geen reactie is geweest op tijd. Maar wij hadden in ons geval al afgesproken dat als het bestuur niet binnen 24 uur (of 30 seconden zoals we hier doen) reageert dat we dan die aanvraag goedkeuren. Dus we geven een default return value van True mee aan onze WaitForExternalEvent. Na de timeout krijgen we dus automatisch true terug.

Laten we dit eens gaan testen.

We starten de flow, openen een browser met de juiste URL en laten hem gaan. Zo gauw de status pagina verschijnt, klikken we op de url bij statusQueryGetUri. Ik krijg dan het volgende resultaat.

// 20191104101214
// http://localhost:7071/runtime/webhooks/durabletask/instances/a05de0c734e64db181028163991c21f9?taskHub=DurableFunctionsHub&connection=Storage&code=rCJjUkbfRFrmoxIzf6CU8mxBnxUeOnLY5qr5DJ04C80FG87HgsDm6w==

{
  "name": "O_PerformRegistration",
  "instanceId": "a05de0c734e64db181028163991c21f9",
  "runtimeStatus": "Running",
  "input": {
    "$type": "MeetingRegistration.Attendee, MeetingRegistration",
    "Name": "Dennis",
    "Email": "dennis@vroegop.org",
    "WantsTweet": true
  },
  "customStatus": null,
  "output": null,
  "createdTime": "2019-11-04T09:12:08Z",
  "lastUpdatedTime": "2019-11-04T09:12:12Z"
}

Je ziet dat de status nu running is. En dat blijft hij ook tot we het proces verder laten gaan (nogmaals: next post!) of tot de timeout plaats vindt. Wacht even, doe een paar keer refresh van deze pagina en je ziet hem op een gegeven moment veranderen van Running naar Completed, met de bijbehorende data.

Je ziet hoe eenvoudig het is om te wachten op externe events en er een time out bij te bouwen. Je hoeft daar zelf eigenlijk niets voor te doen.

Onder water gebeurt er wel heel veel: er wordt een nieuw activity aangemaakt met een timer. Die timer gaat af als de tijd verlopen is. Als dat zo is, dan wordt er een message in de queue gezet, die de orchestrator krijgt. Dan wordt de activity die wacht op de message met approval (of denial) gestopt en het resultaat wordt genegeerd. Mocht er nu wel op tijd gereageerd zijn, dan zal de runtime de timer stoppen en disposen.

Dit is een hoop plumbing. Gelukkig hoeven we dat niet zelf te schrijven. Iemand die de code van de orchestrator ziet en een beetje kan programmeren, ziet meteen wat de flow is en hoe dat allemaal samenhangt.

Er is nog een dingetje: deel 2 van dit verhaal. Hoe gaan we nu zorgen dat dat event verstuurd wordt waar de activity zo met smart op zit te wachten?

Dat vind je in het volgende deel!

We gaan verder met onze inschrijving function workflow. Je zag in de vorige post hoe de engine intern werkt. Dat is op zich goed om te weten, maar we hoeven ons er nu niet al te druk meer over te maken. Echter, je moet wel onthouden dat we in de orchestrator geen code mogen gebruiken die niet deterministisch is. Oftewel: geen code die iedere keer andere resultaten zou kunnen terug geven.

Fan out – Fan in

In ons systeem hebben we besloten de andere user groups op de hoogte te stellen van het voorgenomen bezoek van deze bezoeker. We willen graag van elkaar weten wie nu vaak meetups bezoekt, wie zich altijd aanmeldt maar niet komt, en hoe goed we het nu eigenlijk doen. Even voor de duidelijkheid: dat doen we in het echt dus niet: zelfs al zou het van de wet mogen (en dat mag dus niet) dan nog nemen we privacy heel serieus. Het is maar een voorbeeld…

Nu kunnen we die code relatief makkelijk schrijven. We kunnen een functie maken en daarin een REST api aanroepen. We gaan er even vanuit dat alle usergroups een standaard API hanteren, dus dat is niet zo ingewikkeld.

Maar: Micro Services zijn klein en doen maar 1 ding. Dus een service schrijven die alle REST API’s van alle usergroups aanroept, past daar niet binnen. Daarnaast kan de lijst met usergroups best lang worden, dus waarom zouden we dat niet parallel uitvoeren? Dit zou de hele execution van de totale workflow wel ten goede komen.

Het proces ziet dan als volgt uit:

We hebben ons start punt. Dat is het einde van onze vorige A_CheckInput. Vervolgens gaan we parallel een bericht sturen naar de usergroups SDN, DevNetNoord, DotNetOost en de anderen. Pas als deze berichten allemaal verstuurd zijn, gaan we verder (en in ons voorbeeld tot nu toe is dat het einde van het proces).

Dit is het Fan out – fan in patroon. We waaieren eerst uit naar de verschillende functies, en dan wachten we tot ze klaar zijn waarna we verder gaan. De resultaten worden gebundeld en we kunnen verder gaan.

In ons voorbeeld doen we even niets met de eventuele resultaten van de REST calls. We zouden in dat geval dus een fire-and-forget actie kunnen doen. Wel fan-out maar niet wachten. Maar ik wil graag het hele proces laten zien. Fan out only is makkelijker te bouwen dus dat kun je zelf wel bedenken.

Goed. Laten we eerst de activity eens maken. Ik ga niet echt de implementatie maken van het aanroepen van een REST API: dat is gewoon standaard .net code.

Als eerste moeten we weer bedenken wat we mee gaan geven aan de usergroups. Ik denk dat een email adres genoeg is. De naam is niet relevant. De activity zelf heeft echter ook informatie nodig over welke usergroup we gaan aanroepen. Ik gebruik even een string met de naam van de groep. Dus:

public class UsergroupNotification
{
    public string UsergroupName { get; set; }
    public string AttendeeEmail { get; set; }
}

Geen verrassingen hier.

De activity zelf is ook niet ingewikkeld, zeker omdat we hier niets in het echt gaan doen:

[FunctionName("A_NotifyUsergroup")]
public static void NotifyUsergroup(
    [ActivityTrigger] UsergroupNotification notification, 
    ILogger log)
{
    log.LogWarning($"We are notifying {notification.UsergroupName} about {notification.AttendeeEmail}.");            
}

We maken weer een standaard activity. De FunctionName krijgt de prefix A_. Het is een ActivityTrigger en als data geven we mee de net aangemaakte UsergroupNotification. Uiteraard willen we een logger erbij.

In de body doe ik niets anders dan een log wegschrijven.

Nu naar de orchestrator. Na de call naar A_CheckInput (als dat goed gegaan is) moeten we deze activity starten. Dit is een eerste opzetje:

        [FunctionName("O_PerformRegistration")]
        public static async Task<RegistrationResult> PerformRegistration(
            [OrchestrationTrigger] DurableOrchestrationContext ctx,
            ILogger log)
        {
            if(!ctx.IsReplaying)
                log.LogWarning("We are in the orchestrator! Yeah!");
            var attendee = ctx.GetInput<Attendee>();

            var isValid = await ctx.CallActivityAsync<bool>("A_ValidateInput", attendee.Email);
            if (!isValid)
            {
                return new RegistrationResult
                {
                    IsSucces = false,
                    Reason = "Not a valid email given."
                };
            }

            await ctx.CallActivityAsync("A_NotifyUsergroup", 
                new UsergroupNotification { 
                    AttendeeEmail = attendee.Email, 
                    UsergroupName = "SDN" });

            await ctx.CallActivityAsync("A_NotifyUsergroup",
                new UsergroupNotification
                {
                    AttendeeEmail = attendee.Email,
                    UsergroupName = "DNN"
                });

            await ctx.CallActivityAsync("A_NotifyUsergroup",
                new UsergroupNotification
                {
                    AttendeeEmail = attendee.Email,
                    UsergroupName = "DNO"
                });


            if (!ctx.IsReplaying)
                log.LogWarning("At the end of the workflow.");

            return new RegistrationResult {
                IsSucces = true,
                Reason = "Everything checked out."
            };
        }

Ik heb even de hele orchestrator geplaatst zodat je ziet waar we zijn.

Dit werkt wel maar is niet echt handig: we roepen de verschillende usergroups nu sequentieel aan. We wachten iedere keer tot er een call geweest is en gaan dan verder. Dat willen we niet: we willen het parallel doen.

Ok. Poging 2: ik vervang de 3 calls door het volgende stuk code:

var allUsergroups = new string[] { "SDN", "DNN", "DNO" };
foreach(var usergroupName in allUsergroups)
{
    var notification = new UsergroupNotification { AttendeeEmail = attendee.Email, UsergroupName = usergroupName };
    await ctx.CallActivityAsync("A_NotifyUsergroup", notification);
}

Dat lost niet veel op. De code is leesbaarder maar het is nog steeds sequentieel. Poging 3 dan maar:

var allUsergroups = new string[] { "SDN", "DNN", "DNO" };
var allCalls = new List<Task>();
foreach(var usergroupName in allUsergroups)
{
    var notification = new UsergroupNotification { AttendeeEmail = attendee.Email, UsergroupName = usergroupName };
    allCalls.Add(ctx.CallActivityAsync("A_NotifyUsergroup", notification));
}

await Task.WhenAll(allCalls);

Dit is veel beter. We maken een lijst van de usergroups, dan maken we een lege lijst van Tasks aan. In de loop roepen we CallActivityAsync aan maar we awaiten het niet. Dan krijgen we een Task terug (we hebben geen return type, dus geen Task<T> maar gewoon Task). De verzamelen we in de lijst. Pas als we alle Tasks aangemaakt hebben, roepen we await Task.WhenAll(allCalls); aan.

Deze gaat alle Tasks opstarten en komt pas terug als alle tasks uitgevoerd zijn. Let op: ook hier geldt dat het systeem iedere keer als er een task gestart wordt, de orchestrator weggooid en daarna opnieuw opstart. Met andere woorden: samen met de call naar A_CheckInput wordt de orchestrator nu 5 keer gestart… Een keer voor de eerste run, daarna na het beeindigen van A_Checkinput en daarna 3 keer na het beeindigen van iedere call naar A_NotifyUsergroup. Je ziet: het aantal invocations kan snel oplopen.

We zouden het resultaat van de calls makkelijk kunnen uitlezen hier. Ze geven immers allemaal een result terug. Dit gaan we niet doen, maar ik vraag je wel even om na te denken hoe je dit soort dingen zou bouwen in een normale Azure Function. Realiseer je eens hoeveel controle code je zou moeten schrijven. Nu doen we een fan-out, fan-in in een paar regels code.

Configuration ophalen

Ik ben nog niet tevreden. De lijst met usergroups is nu hardcoded in de orchestrator. Gezien het tempo waarin Meetups verschijnen en weer verdwijnen, is dat niet optimaal. We moeten die data ergens vandaan halen.

Het ligt voor de hand om die lijst in een storage te hebben. We hebben immers al de TableStorage. Maar in dit voorbeeld maak ik het leven wat eenvoudiger en sla ik het op in de config. Daar kan ik later makkelijk bij komen als het aantal usergroups dat we willen benaderen veranderd.

In de file local.settings.json verander ik de huidige settings in:

{
    "IsEncrypted": false,
    "Values": {
        "AzureWebJobsStorage": "UseDevelopmentStorage=true",
      "FUNCTIONS_WORKER_RUNTIME": "dotnet",
      "UserGroups": "SDN,DNN,DNO"
    }
}

Ik heb dus de “UserGroups” key toegevoegd en de waardes erbij gezet in een comma-seperated list.

Als ik dit later wil deployen naar Azure, zal ik die waardes daar ook moeten toevoegen. De file local.settings.json is immers, zoals de naam al zegt, local.

Dit moeten we even uitlezen:

var allUsergroups = System.Environment.GetEnvironmentVariable("Usergroups")
    .Split(',')
    .ToArray(); 
         
var allCalls = new List<Task>();
foreach(var usergroupName in allUsergroups)
{
    var notification = new UsergroupNotification { AttendeeEmail = attendee.Email, UsergroupName = usergroupName };
    allCalls.Add(ctx.CallActivityAsync("A_NotifyUsergroup", notification));
}

await Task.WhenAll(allCalls);

We lezen de data uit, doen een split op de komma en veranderen dit in een array. De rest blijft hetzelfde.

Zijn we er nu?

Nee.

Dit mag dus niet. Orchestrator functions moeten deterministic zijn. Stel dat tussen de invocation van de eerste call naar A_NotifyUsergroup en de tweede de config wijzigt. Dat kan zo maar gebeuren. Wat betekent dat dan? We hebben al eerder gezien dat we geen code mogen gebruiken die eventueel andere resultaten tussen invocations kan opleveren. Dus dit mag niet.

De oplossing is uiteraard simpel: verplaats die code naar een eigen activity. Immers: de eerste keer dat die activity uitgevoerd wordt, lezen we de settings file daadwerkelijk uit. De keren daarna krijgen we het resultaat daarvan terug uit de TableStorage, ongeacht of de onderliggende data veranderd is. Dat maakt voor deze instance van de workflow niet uit.

Dus onze orchestration ziet er als volgt uit:

if (!ctx.IsReplaying)
    log.LogWarning("Getting all usergroups.");

var allUsergroups = await ctx.CallActivityAsync<string[]>("A_GetUsergroupNames", null);
            
var allCalls = new List<Task>();
foreach(var usergroupName in allUsergroups)
{
    var notification = new UsergroupNotification { AttendeeEmail = attendee.Email, UsergroupName = usergroupName };
    allCalls.Add(ctx.CallActivityAsync("A_NotifyUsergroup", notification));
}

await Task.WhenAll(allCalls);

In regel 4 zien we de call naar de Activity. Deze moet data mee krijgen maar we kunnen daar null voor gebruiken.

De activity:

[FunctionName("A_GetUsergroupNames")]
public static string[] GetUsergroupNames(
    [ActivityTrigger] object input, 
    ILogger log)
{
    var allUsergroups = System.Environment.GetEnvironmentVariable("Usergroups")
        .Split(',')
        .ToArray();

    log.LogWarning("Loaded the usergroup list.");

    return allUsergroups;
}

Het is een activity, dus we moeten een parameter meegeven met als attribuut [ActivityTrigger]. Echter, we geven null mee en we doen er niets mee, dus ik heb er object van gemaakt met als naam input. Er moet toch iets staan.

In de body lezen we de config uit. Hier mag dat wel: dit wordt per instance van de workflow maar een keer uitgevoerd. Het resultaat komt voor deze instance in de TableStorage te staan dus we hebben een deterministische orchestrator.

Als we deze runnen kunnen we in de output ook zien dat het werkt:

De volgorde van de output kan varieren, de timings ook. Maar alle usergroups die we hebben staan in de config worden aangeroepen.

Interessant is ook de output. Klik maar op de status query (je weet wel: die in de browser staat onder de naam statusQueryGetUri. Voeg aan het einde van deze url de parameters &showHistory=true&showHistoryOutput=true en zie het resultaat.

// 20191103143449
// http://localhost:7071/runtime/webhooks/durabletask/instances/a9f84fe8ee0f42529b26b9cfeeb68a17?taskHub=DurableFunctionsHub&connection=Storage&code=6ZoZrnZw/qufYwLVultvsgS7RiwKNf1g6t0AjF9YNUMERC84wKrOLA==&showHistory=true&showHistoryOutput=true

{
  "name": "O_PerformRegistration",
  "instanceId": "a9f84fe8ee0f42529b26b9cfeeb68a17",
  "runtimeStatus": "Completed",
  "input": {
    "$type": "MeetingRegistration.Attendee, MeetingRegistration",
    "Name": "Dennis",
    "Email": "dennis@vroegop.org",
    "WantsTweet": true
  },
  "customStatus": null,
  "output": {
    "IsSucces": true,
    "Reason": "Everything checked out."
  },
  "createdTime": "2019-11-03T13:34:20Z",
  "lastUpdatedTime": "2019-11-03T13:34:25Z",
  "historyEvents": [
    {
      "EventType": "ExecutionStarted",
      "Timestamp": "2019-11-03T13:34:20.8306001Z",
      "FunctionName": "O_PerformRegistration"
    },
    {
      "EventType": "TaskCompleted",
      "Result": true,
      "Timestamp": "2019-11-03T13:34:22.6157852Z",
      "ScheduledTime": "2019-11-03T13:34:21.8540948Z",
      "FunctionName": "A_ValidateInput"
    },
    {
      "EventType": "TaskCompleted",
      "Result": [
        "SDN",
        "DNN",
        "DNO"
      ],
      "Timestamp": "2019-11-03T13:34:23.3958736Z",
      "ScheduledTime": "2019-11-03T13:34:23.1352168Z",
      "FunctionName": "A_GetUsergroupNames"
    },
    {
      "EventType": "TaskCompleted",
      "Result": null,
      "Timestamp": "2019-11-03T13:34:24.0132401Z",
      "ScheduledTime": "2019-11-03T13:34:23.7418444Z",
      "FunctionName": "A_NotifyUsergroup"
    },
    {
      "EventType": "TaskCompleted",
      "Result": null,
      "Timestamp": "2019-11-03T13:34:24.158211Z",
      "ScheduledTime": "2019-11-03T13:34:23.7418405Z",
      "FunctionName": "A_NotifyUsergroup"
    },
    {
      "EventType": "TaskCompleted",
      "Result": null,
      "Timestamp": "2019-11-03T13:34:24.5138455Z",
      "ScheduledTime": "2019-11-03T13:34:23.7418457Z",
      "FunctionName": "A_NotifyUsergroup"
    },
    {
      "EventType": "ExecutionCompleted",
      "OrchestrationStatus": "Completed",
      "Result": {
        "IsSucces": true,
        "Reason": "Everything checked out."
      },
      "Timestamp": "2019-11-03T13:34:25.1201956Z"
    }
  ]
}

En zo kunnen we dus zaken parallel laten draaien!

Toch ben ik nog niet helemaal tevreden. Onze orchestrator functie wordt wat groot en er gebeurt te veel in. Daar gaan we in de volgende post wat aan doen.

Goed, In het vorige deel hebben we een starter function, een orchestration function en een activity function gemaakt. Het lijkt een hoop werk voor een relatief eenvoudig systeem, maar dat valt wel mee: de rest van de code wordt niet veel ingewikkelder dan dit. Nou ja, wel een beetje maar de plumbing zit er nu ongeveer wel in.

We hebben ook gezien dat als we onze heel eenvoudige workflow starten, er in de output twee keer de log message van de orchestrator stond. En ik vertelde dat dat geen bug is maar dat dat zo hoort.

In deze post, zal ik uitleggen hoe dat werkt en gelijk hoe stateless Azure Functions ineens toch stateful kunnen lijken.

Als je wilt meekijken, raad ik je aan om de gratis tool Azure Storage Explorer te downloaden vanaf deze site. Deze tool gebruik je om te kijken wat er in je storage accounts staat, ook in je emulated versie van je storage account. Dus: we kunnen kijken hoe onze Durable Functions in het echt werken.

De orchestrator aanpassen

Om een goed beeld te krijgen van de werking van Durable Functions, moeten we even onze orchestrator aanpassen. Dat moesten we toch al: onze validator geeft aan of een attendee een geldig email adres meegeeft of niet. Als dat zo is: prima, we gaan door. Maar als dat niet zo is, dan moeten we stoppen met de functies.

Dit was een van de voorbeelden die ik aanhaalde om aan te geven dat we best veel code moeten schrijven om die case af te handelen. In “traditionele” Azure Functions moet je op de een of andere manier bijhouden wat er gebeurt met het resultaat van die Validator functie. Ik zei al dat die functie messages in de queue post om aan te geven wat het resultaat is. Maar: dat betekent dat de functie zelf kennis heeft van de flow en de messages die er allemaal zijn. Dat wil je niet: die functie is minder herbruikbaar op die manier. De functie zoals we die nu geschreven hebben is proces-agnostisch. Hij heeft geen idee waar hij voor gebruikt wordt, het enige wat hij weet is dat hij en boolean terug geeft om aan te geven of een email adres geldig is of niet (nou ja, of er een @ in zit of niet, maar dat is muggenziften,,,)

De orchestrator regelt per definitie de flow. Dat is wat een orchestrator doet. Dus daar gaan we iets met het return resultaat doen.

Ik introduceer een nieuwe DTO: RegistrationResult. Deze geeft de orchestrator terug: is het goed gegaan of niet?

namespace MeetingRegistration
{
    public class RegistrationResult
    {
        public bool IsSucces { get; set; }
        public string Reason { get; set; }
    }
}

Deze spreekt voor zich, denk ik.

Dit moeten we gaan gebruiken in de orchestrator functie:

[FunctionName("O_PerformRegistration")]
public static async Task<RegistrationResult> PerformRegistration(
    [OrchestrationTrigger] DurableOrchestrationContext ctx,
    ILogger log)
{
    log.LogWarning("We are in the orchestrator! Yeah!");
    var attendee = ctx.GetInput<Attendee>();
            
    var isValid = await ctx.CallActivityAsync<bool>("A_ValidateInput", attendee.Email);
    if (!isValid)
    {
        return new RegistrationResult
        {
            IsSucces = false,
            Reason = "Not a valid email given."
        };
    }

    return new RegistrationResult { 
        IsSucces = true, 
        Reason = "Everything checked out." 
    };
}

De return type is veranderd van Task naar Task<RegistrationResult>. Aan het einde van de method geven we de default terug met de info dat alles gelukt is. Als onze activity aangeeft dat er een probleem was (dus het resultaat is False), geven we die info ook terug aan onze aanroepende client.

Als we dit gaan testen, krijgen we het volgende resultaat met een geldig email adres (dus: in de browser de link van statusQueryGetUri klikken of selecteren en pasten)

En als we een ongeldig adres meegeven:

(ik weet het: waarschijnlijk weet je het wel, maar… je hoeft de app niet iedere keer opnieuw op te starten.. gewoon de URL in de browser aanpassen. Je krijgt toch een nieuwe instance van de workflow)

Je ziet dan in het eerste geval de “output” alle data bevat die aangeeft dat alles goed ging, en in het tweede geval dat er een fout was (not a valid email given).

Dit is het resultaat van de orchestration. Maar kunnen we ook kijken wat er in de activities gebeurt? Ik zou de vraag niet stellen als het antwoord niet ‘ja’ was..

In het scherm met de output van het resultaat moeten we de querystring even aanpassen (let op dat je runtime nog steeds draait.. mocht dit niet zo zijn dan kun je gewoon in Visual Studio op F5 drukken. Je hoeft geen nieuwe request aan te maken: deze oude is nog gewoon beschikbaar).

Voeg aan het einde van de URL in de browser de volgende parameter toe:

&showHistory=true

Je ziet dat je nu voor alle stappen meer informatie krijgt. Meer nog dan je wellicht zou verwachten (komt goed.. moment!)

Voeg nu de volgende parameter toe aan wat we net al hadden:

&showHistoryOutput=true

En ververs de browser

Als je even zoekt, zie je nu per stap de return values. In ons geval zien we dat A_ValidateInput false terug geeft, wat resulteerde in die message die we deze attendee niet kunnen accepteren. We kunnen deze attendee helaas niet laten weten dat hij of zij niet mag komen: we hebben geen geldig email adres.

Let op: deze informatie wordt heel snel heel groot. Vandaar dat we twee opties hebben om het al dan niet te laten zien.

Async / await in durable functions

Maar hoe zit het nu met die dubbele log messages? En al die aangeroepen functies die ik zie? Ik ben blij dat je het vraagt.

We komen nu bij de kern van hoe Durable Functions durable kunnen zijn en toch stateless. Heb je de Storage Explorer open staan? Die hebben we namelijk zo nodig. Kijk even mee!

Het hart van het systeem wordt gedefinieerd door de keywords Async / Await. De meeste .net / c# ontwikkelaars kennen deze wel, hoewel maar weinigen echt weten hoe het intern werkt. Mijn uitleg over de werking hieronder is totaal niet correct, maar geeft wel weer hoe de werking zou kunnen zijn. Met andere woorden: ook al werkt het intern anders, zo zou het kunnen werken…

Async / await is een pattern on asynchrone code synchroon uit te voeren. Kijk eens naar de volgende dummy code:

public async Task DoSomethingUsefull()
{
    // Hier komen we binnen
    Console.WriteLine("Hello world");

    // Nu doen we de volgende stap, maar...
    // dit wordt uitgevoerd in een andere thread (niet dus, maar het
    // idee klopt wel)
    // Pas als die thread klaar is, komen we hier terug in deze thread

    await DoSomethingThatLastsLong();
            
    // Als bovenstaande thread klaar is, gaan we hier verder
    // (nog steeds: niet dus...)

}

Dit is ongeveer hoe de meeste mensen denken dat async/ await werkt. En het idee klopt wel, hoewel dit pattern in het echt niet (altijd) met threads werkt.

Je zou verwachten dat onze code van de orchestrator hetzelfde doet.

Echter… dit is niet wat er gebeurt. Dit is namelijk weer een mooi voorbeeld van hoe keywords in C# hergebruikt worden maar iets anders doen in een andere context.

Laten we de volgende dummy orchestrator eens bekijken:

public static async Task DummyOrchestrator(
    [OrchestrationClient] DurableOrchestrationClient client
    )
{
    // Entry point. Hier beginnen we.

    var result = await client.StartNewAsync<string>("A_Activity", SomeData);
    // De thread wacht hier tot de activity klaar is, 
    // en gaat dan verder

    // Meer code..
}

Dit lijkt op de code die we eerder zagen. Maar de compiler maakt er iets heel anders van. De Activity bijvoorbeeld doet het volgende in het echt:

public static string Activity([ActivityTrigger] SomeData someData)
{
    // Doe wat werk
    // Sla de data op in TableStorage in Azure (of Emulator)
    // Return void... 
}

Hoewel deze activity een string terug moet geven, geeft deze niets terug… het resultaat echter wordt opgeslagen in TableStorage.

De orchestrator doet iets als deze pseudocode:

public static async Task DummyOrchestrator(
    [OrchestrationClient] DurableOrchestrationClient client
    )
{
    // Entry point. Hier beginnen we.

    bool a_activityHasRun = GetRunStatus("A_Activity");
    if (!a_activityHasRun)
    {
        client.StarAsync("A_Activity", SomeData);
        return;
    }
    var result = ReadFromTableStorage("A_ActivityResult");

    // Meer code..
}

Hier moeten we even iets langer bij stilstaan.

De workflow wordt gestart. Als eerste wordt er ergens in een tabel in TableStorage gekeken of de eerste activity (A_Activity) al gedraaid heeft voor deze workflow instance. Als dat niet zo is, wordt deze functie uitgevoerd in een aparte context en de orchestrator verlaat het geheugen. De code wordt opgeruimd. Het app-domain verdwijnt. De VM kan zelfs weggegooid worden, afhankelijk van de tijd en de geheugendruk.

Als de activity klaar is met zijn werk en het resultaat opgeslagen heeft in de TableStorage, wordt er een nieuwe instance van de orchestrator gemaakt. En deze begint van voor af aan…

Alleen: dit keer is het resultaat van GetRunStatus(“A_Activity”) true: we hebben hem al uitgevoerd! Dus de method gaat door en haalt dat resultaat op uit de TableStorage. Nu hebben we het resultaat, en we gaan verder.

Als hier nu weer een activity gestart wordt, gebeurt hetzelfde: de orchestrator stopt en verdwijnt uit het geheugen. Als de activity klaar is, wordt de orchestrator weer opgestart en begint alles weer van voor af aan.. ook A_Activity (maar die is al uitgevoerd dus het enige wat er gebeurt is dat het resultaat uit de TableStorage komt.

Onze log message verschijnt daarom dus 2 keer. Await / Async werkt hier dus heel anders dan we gewend zijn. Await hier betekent: stop met verwerken, verlaat de functie en ruim je resources op. Dat is iets heel anders dan de thread pauzeren!

Regels voor orchestrators

Aangezien de code in de orchestrator iedere stap van de workflow opnieuw uitgevoerd wordt zijn er een aantal beperkingen aan wat je kunt doen in die code. Je kunt niet zomaar alles doen wat je wilt: de flow in de method moet deterministisch zijn. Het systeem moet volledig voorspelbaar zijn. Kijk even naar de volgende dummy orchestrator:

[FunctionName("O_MyOrchestrator"]
public static async Task MyOrchestrator(
    [OrchestrationTrigger] DurableOrchestrationContext ctx, 
    ILogger log)
{
    var invocationDate = System.DateTime.UtcNow;
    log.LogInformation($"Code is invoked at {invocationDate}.");

    var result = await ctx.CallActivityAsync<bool>("A_MyActivity", invocationDate);
    var moreResult = await ctx.CallActivityAsync<Attendee>("A_EnrichAttendee", invocationDate);
}

Op zich gebeurt hier niets raars. We halen de tijd op, we loggen de informatie en we roepen twee activities aan. Prima. In activity A_MyActivity geven we die datum mee. Blijkbaar is dat nodig om een berekening uit te voeren.

Maar wat gebeurt er nu echt? Laten we het eens stap voor stap bekijken.

1. We halen de huidige UTC tijd op
2. We loggen die tijd
3. We zorgen ervoor dat de functie A_MyActivity gestart wordt, met de huidige tijd als data
4. We verlaten deze method.
5. Als A_MyActivity afgerond is, start deze method weer op
6. We halen de huidige UTC tijd op
7. We loggen de nieuwe tijd
8. We halen het resultaat van A_MyActivity uit de TableStorage
9. We starten activity A_EnrichAttendee op, met als parameter een nieuwe waarde voor invocationDate…

Wacht even. Dat kan niet kloppen. Als die twee activities iets met die tijd doen, en daar dus afhankelijk zijn, dan kun je in de flow van deze method aflezen dat ze allebei dezelfde waarde krijgen. Maar die krijgen ze niet. Als A_MyActivity uren er over doet om te runnen (en dat kan makkelijk!) dan zit er een behoorlijk verschil tussen de waardes van invocationDate die aan de twee methods doorgegeven worden.

Dit levert potentieel grote problemen op.

Ander voorbeeld: we maken in de eerste regel een GUID aan. Deze hebben we nodig als key in een storage bijvoorbeeld. De eerste activity krijgt deze GUID mee en slaat data op met die GUID als key. De tweede activity heeft die GUID ook nodig om de data weer op te halen. Maar: omdat de orchestrator opnieuw begint krijgen we een nieuwe GUID.

Dit mag dus niet.

Net als het ophalen van data uit een storage, uit een config, uitlezen van een web-resources als REST server, enzovoorts. Alle data die mogelijk tussen twee calls in kan veranderen, mag je niet gebruiken in een orchestrator functie.

Dat klinkt als een beperking, maar dat valt wel mee. We zullen snel zien hoe we hier mee om moeten gaan, maar ik kan je wel alvast de volgende code meegeven:

public static async Task MyOrchestrator(
    [OrchestrationTrigger] DurableOrchestrationContext ctx, 
    ILogger log)
{
    var invocationDate = ctx.CurrentUtcDateTime;

    if(!ctx.IsReplaying)
        log.LogInformation($"Code is invoked at {invocationDate}.");

    var result = await ctx.CallActivityAsync<bool>("A_MyActivity", invocationDate);
    var moreResult = await ctx.CallActivityAsync<Attendee>("A_EnrichAttendee", null);
}

In plaats van System.DateTime.UtcNow vragen we ct.CurrentUtcDateTime op. De eerste keer dat we in deze orchestrator komen voor deze instance van de workflow, geeft dit de huidige datum en tijd terug. De volgende keer als we hier inkomen, nadat A_MyActivity klaar is, dan krijgen we gegarandeerd dezelfde waarde terug. Hoewel die tweede call dus dagen later kan gebeuren (want activities kunnen heel lang duren) krijgen we altijd die eerste waarde terug.

Hetzelfde kunnen we doen met ctx.NewGuid. Die genereert een GUID die bij het opnieuw uitvoeren hetzelfde resultaat terug geeft.

Verder kunnen we aan de context vragen of dit de eerste keer is dat deze regel uitgevoerd wordt: ctx.IsReplaying geeft aan of dit een replay is of niet. Let wel op: als A_MyActivity klaar is, is op dit punt IsReplaying true, maar zo gauw de code na de call naar A_MyActivity is (dus wanneer eigenlijk het resultaat uit de TableStorage gehaald wordt) dan is IsReplaying weer false. We zijn immers hier nog niet geweest.

Ga nu niet proberen de boel te optimaliseren door dit te doen:

public static async Task MyOrchestrator(
    [OrchestrationTrigger] DurableOrchestrationContext ctx, 
    ILogger log)
{
    var invocationDate = ctx.CurrentUtcDateTime;

    if(!ctx.IsReplaying)
        log.LogInformation($"Code is invoked at {invocationDate}.");

    if(!ctx.IsReplaying)
        var result = await ctx.CallActivityAsync<bool>("A_MyActivity", invocationDate);

    if (!ctx.IsReplaying)
        var moreResult = await ctx.CallActivityAsync<Attendee>("A_EnrichAttendee", null);
}

Je moet echt iedere keer die CallActivityAsync aanroepen. Weet je nog? De eerste keer zorgt dit er voor dat de functie uitgevoerd wordt, de tweede, derde en andere keren haalt die code regel het resultaat op uit de TableStorage. Dus die moet iedere keer opnieuw aangeroepen worden.

Ik gebruik die ctx.IsReplaying dan eigenlijk alleen maar voor het loggen van informatie. Dat hoef ik namelijk niet iedere keer te zien. De data is steeds gelijk dus 1 logregel per keer is genoeg.

Stateless durable functions

En nu weet je gelijk hoe het kan dat we stateless kunnen zijn en toch onze data kunnen behouden. Alle data wordt continue opgeslagen in een TableStorage. Deze storage kunnen we zo configureren dat we zelfs geo-redundant zijn. Dus de data staat keurig ergens in de cloud, de functies worden iedere keer opnieuw gestart (en de app-domain opnieuw aangemaakt, zelfs de VM waar het op draait kan opnieuw aangemaakt worden). We zijn volledig serverless en stateless bezig maar voor ons als ontwikkelaar lijkt het alsof we een eenvoudige flow hebben in een method waarin data bewaard blijft tussen de calls in. Het lijkt wel magie!

Even terug naar onze originele code. Als we deze voor de allereerste keer draaien, maakt de runtime alle structuren voor ons aan. Ik heb het al een paar keer over die TableStorage gehad, laten we eens kijken hoe dat er uit ziet. En daar hebben we die Storage Explorer voor nodig.

Als ik onze code nu een keer ga draaien, krijg ik het resultaat dat we al gezien hebben. Ik geef even een geldig email adres mee zodat we de hele flow (hoe kort ook) doorlopen.

Daarna start ik de Storage Explorer. Die ziet er dan bij mij als volgt uit:

Zoals je ziet, heeft de runtime van alles aangemaakt. We hebben blob containers, queues, en tables.

Die queues (4 control queues en 1 workitems queue, die het lopende werk bevatten) zijn het hart van het systeem. Ik vertelde in de allereerste post hoe je functions kunt laten communiceren: door het versturen van messages in een queue. Later vertelde ik dat we dat kunnen voorkomen door Durable Functions te gebruiken. Maar wat blijkt nu: dat is precies hoe Durable Functions werken. Ze versturen zelf allemaal berichten in allerlei queues om het werk te synchroniseren. Het verschil: we hoeven het dit keer niet zelf te schrijven. Maar uiteindelijk, onder water, zijn alle orchestrator en activity functions gewoon queue-triggered functions. We zien het alleen niet.

Alle data die we nodig hebben voor het runnen van een workflow staat in de tabel DurableFunctionsHistory. Durable Functions zijn een voorbeeld van event-sourcing: alle data wordt stap voor stap opgeslagen zodat we deze later kunnen herhalen.

In het voorbeeld dat ik hier gaf kun je dat duidelijk zien. Je ziet precies welke stappen er genomen worden. In de kolom EventType staat precies welke stap er genomen wordt en daar achter in de andere kolommen input en result (niet zichtbaar in mijn voorbeeld maar bij jou kun je hem wel zien) staat de data die heen en weer gaat.

We kunnen ook mooi zien wat de flow is:

1. Orchestrator started. We beginnen de flow dus
2. Execution started. We gaan een instance uitvoeren
3. Task Scheduled. Onze A_CheckInput method wordt klaar gezet om te beginnen
4. Orchestrator Completed. Tot A_CheckInput klaar is, is de orchestrator voorlopig klaar
5. Orchestrator started. Blijkbaar is A_CheckInput klaar: we gaan opnieuw beginnen
6. Task Completed. We gaan het resultaat van A_CheckInput ophalen (in de result staat nu dan de waarde true)
7. Execution Completed. We zijn aan het einde van deze instance flow
8. Orchestrator Completed. De orchestrator wordt weer afgebroken

Je kunt de flow met alle data, timestamps en dergelijke dus perfect volgen.

En nu?

We hebben weer een hoop besproken. In de volgende post gaan we een wat meer uitdagende activity maken: we gaan de andere usergroups informeren over het feit dat iemand zich aanmeldt.

Tot dan!

In het eerste deel van deze serie heb ik uitgelegd waarom Durable Functions waarde kunnen toevoegen: we hoeven minder plumbing te schrijven. In het tweede deel zijn we begonnen met de code (voornamelijk plumbing, helaas. Nu wordt het tijd om iets Durable te gaan maken!

Ik wil eerst even stilstaan bij wat de term Durable Functions nu precies betekent. In het eerste deel heb ik je al behoorlijk veel theorie gegeven dus ik wilde dit even overslaan. Maar nu is het moment toch daar om uit te leggen waar we het nu precies over hebben.

Azure Functions zijn stateless. Je roept ze aan met data, de runtime maakt een instance van de functie aan (en dus de class waar deze in zit, en dus een app-domain, en eventueel wordt er zelfs een VM opgestart om die app-domain op te draaien), de functie gaat aan het werk, stuurt het resultaat terug en… komt te overlijden. De functie is weg, alle data die erbij hoort is weg. De app-domain kan ook weg en zelfs de VM zou kunnen verdwijnen.

Dit zorgt er natuurlijk voor dat we heel makkelijk een schaalbaar systeem kunnen maken. Als we ineens heel veel requests krijgen, kunnen we makkelijke tientallen, honderden of duizenden instances van onze functies de lucht in krijgen. Ze hebben toch niets met elkaar of met andere systemen te maken dus ze zitten elkaar niet in de weg.

Maar: als je toch state wilt bewaren zul je dat ergens op moeten slaan. SQL Server is een optie, blog storage of table storage zou ook kunnen, het maakt eigenlijk niet uit. Als je er maar voor zorgt dat alle data die je wilt bewaren ergens opgeslagen wordt want de functie verdwijnt uit het geheugen als deze klaar is.

Uiteraard hebben we daar al heel lang oplossingen voor. En met Durable Functions hebben we daar een oplossing bijgekregen. Durable Functions doen alsof ze state hebben. Dat hebben ze niet: de data wordt steeds opgeslagen tussen de calls in zodat deze bewaard blijft, maar het gebeurt achter de schermen. We krijgen dus de illusie van een stateful, oftewel durable, function.

Een van de dingen die je daar mee kunt doen is er een state machine van maken, of een workflow. Als je een workflow heb, moet je bijhouden waar je bent in de flow en de resultaten van alle stappen opslaan en delen met de andere stappen. Dat klinkt als stateful programmeren en dat is dus ook precies waar Durable Functions je mee helpen.

De workflow bestaat uit 3 onderdelen:

1. Een starter function, die de workflow aftrapt
2. Een orchestrator function, die de juiste functies aanroept en de data deelt
3. De activity functions, die het echte werk doen

We hebben een begin gemaakt met de starter functie. Laten we daar eens een echte starter van maken. Daarna zullen we de orchestrator bekijken, gevolgd door de activity.

Komt ‘ie!

De starter function

We nemen onze code uit de vorige post. De functie “StartRegistration” zal de workflow gaan starten. Deze heeft nu immers alle data die we nodig hebben (name, email en wantsTweet).

Om Durable Functions te kunnen gebruiken moeten we wel een Nuget package installeren. Dat is Microsoft.Azure.WebJobs.Extensions.DurableTask

Let op: op dit moment is de laatste stabiele versie v1.8.3. Er is een preview van v2 uit maar die werkt totaal anders. Dus gebruik deze nog even niet. Als deze versie uit preview is, zal ik er uitgebreid over schrijven!

Als dit gebeurt is, gaan we terug naar onze starter function. We voegen een parameter toe aan de method:

[FunctionName("StartRegistration")]
public static async Task<HttpResponseMessage> StartRegistration(
    [HttpTrigger(
        AuthorizationLevel.Function, 
        "get", 
        Route = "dotNed/register/{name}/{email}/{wantsTweet}")] 
    HttpRequestMessage req,
    [OrchestrationClient] DurableOrchestrationClient client,
    string name, 
    string email,
    bool wantsTweet,
    ILogger log)
{
    log.LogWarning($"Received: {name} {email} {wantsTweet}.");

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

}
}

In regel 8 zie je dat ik een DurableOrchestrationClient (met de naam client) heb toegevoegd. Deze heeft als attribuut OrchestrationClient zodat de runtime deze kan injecteren. De client kunnen we gebruiken om de workflow mee te starten.

We daan de body van de method aanpassen, maar voor we dat gaan doen moeten we iets hebben wat de data vast kan houden die we moeten doorgeven. We krijgen deze data mee van de runtime (name, email en wantsTweet) en die wil ik in een class hebben. Dus:

namespace MeetingRegistration
{
    public class Attendee
    {
        public string Name { get; set; }
        public string Email { get; set; }
        public bool WantsTweet { get; set; }
    }
}

Dit is een gewone DTO. We zullen in deze reeks nog veel meer van dit soort classes maken, ieder met precies genoeg informatie voor de function die ze zal krijgen. Want: micro-services zijn zo onafhankelijk mogelijk.

In de body van onze Function kunnen we nu een instance aanmaken van deze Attendee class. Deze kunnen we dan doorgeven aan de orchestrator die deze data zal gebruiken om alle functies uit te voeren.

Ok. Komt ‘ie:

        [FunctionName("StartRegistration")]
        public static async Task<HttpResponseMessage> StartRegistration(
            [HttpTrigger(
                AuthorizationLevel.Function, 
                "get", 
                Route = "dotNed/register/{name}/{email}/{wantsTweet}")] 
            HttpRequestMessage req,
            [OrchestrationClient] DurableOrchestrationClient client,
            string name, 
            string email,
            bool wantsTweet,
            ILogger log)
        {
            log.LogWarning($"Received: {name} {email} {wantsTweet}.");
            var attendee = new Attendee 
            { 
                Name = name, 
                Email = email, 
                WantsTweet = wantsTweet 
            };

            var activityId = await client.StartNewAsync("O_PerformRegistration",attendee);
            return client.CreateCheckStatusResponse(req, activityId);
        }

Ik heb hier nog even de hele method neergezet. Vanaf nu zal ik dat alleen doen als dat nodig is. Maar: dit is alles wat we nodig hebben voor onze starter functie. Hij is af!

Wat doen we nu?

Als eerste maken we die Attendee instance aan.

Daarna roepen we StartNewAsync aan op de client (je weet wel: die geinjecteerde DurableOrchestrationClient). Deze krijgt een paar parameters: de naam van de orchestrator function en de data die we door willen geven.

De naam is een string. Zoals ik al eerder zei, kijkt Azure Functions alleen naar het attribuut FunctionName om te bepalen wat je aan wilt roepen. De C# method name is niet relevant. Ik heb als naming convention overigens de volgende regel:

1. Normale, publieke functions krijgen dezelfde naam als de C# method
2. Orchestration functions krijgen de naam als de method maar met de prefix O_
3. Activity functions krijgen de naam als de method maar met de prefix A_

De reden hiervoor is dat we later, in Azure, bij de Function App een hele lijst krijgen met alle functies die we hebben. We krijgen uiteindelijke meerdere publieke functies, meerdere orchestrators en meerdere activities. Dan is het wel fijn om in die lijst meteen te zien wat nu precies wat is.

De StartNewAsync geeft een string terug. Deze string is de unieke identifier van deze run van de workflow. Dus voor iedere attendee die zich aanmeldt wordt de workflow gestart en krijgen we een unieke ID.

Als laatste heb ik de return aangepast. Eerst stond daar dat we 202 OK teruggeven, maar nu laat ik de client wat debug informatie genereren en die op het scherm tonen. Doe dit nu niet in productie: alle gevoelige informatie die je niet wilt delen komt hier in te staan. We zullen zo zien waar dat uit bestaat.

De orchestrator basis

Voordat we dit kunnen runnen om te testen, moeten we een kale orchestrator hebben. Ook dit is een Azure Function. Ik zet deze even in een aparte class, waar alle orchestrators in zullen komen. Op die manier hou ik een mooie scheiding:

Dit is hem:

using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;

namespace MeetingRegistration
{
    public static class OrchestratorFunctions
    {
        [FunctionName("O_PerformRegistration")]
        public static void PerformRegistration(
            [OrchestrationTrigger] DurableOrchestrationContext ctx,
            ILogger log)
        {
            log.LogWarning("We are in the orchestrator! Yeah!");
        }
    }
}

De class heb ik OrchestratorFunctions genoemd. Net als bij de starter functions is de naam niet relevant, maar C# verreist nu eenmaal dat alle code in classes zit.

De method zelf heet PeformRegistration. Ook hier is de naam niet belangrijk; wel belangrijk is de FunctionName attribuut met daarin de naam “O_PerformOrchestration”. Dit is de naam die we gebruikt hebben in de starter method, dus die moeten overeen komen.

De method krijgt twee parameters: als laatste de logger en als eerste de trigger. Aangezien dit gewoon een Azure Function is, moeten we een trigger opgeven. Onze starter function had een HTTP Trigger, deze heeft een OrchestrationTrigger (type DurableOrchestrationContext, naam ctx).

En in de body doen we niets anders dan loggen.

Ik heb de method niet async gemaakt, we hebben ook geen return value.

Laten we dit eens gaan runnen. Als we de runtime starten, zien we de normale debug informatie en de URL die we kunnen gebruiken om de starter aan te roepen. Daar is niets veranderd. Als we echter in de browser die URL aanroepen, krijgen we een heel ander resultaat. In de output van de runtime zien we dit:

En in onze browser zien we dit:

Ok. Wellicht heb jij in je browser hele andere informatie: ik heb in Google Chrome de extension JSON Viewer geinstalleerd die JSON iets vriendelijker weergeeft. Maar de data moet ongeveer hetzelfde zijn.

Laten we eens beginnen met de debug info.

We zien keurig in het geel boven in dat onze data ontvangen is door de starter functie. Prima. Maar verder op zien we in het geel de log van onze orchestrator. Deze is dus ook gevonden!

De output in de browser is ook veelzeggend. Als eerste de datum en tijd van de aanroep. De tweede regel geeft de volledige request URL weer. En daarna komt er wat extra informatie.

Deze info is afkomstig uit de call naar client.CreateCheckStatusResponse.

Ik zei al dat je dit niet in productie wil. We gaan de verschillende URL’s nog wel nakijken, maar ik kan je wel verklappen dat hier een aantal secret keys in staan die je liever niet deelt met de buitenwereld. Zo is er bijvoorbeeld een URL genaamd terminatePostUri, die je kunt gebruiken om een lopende workflow om zeep te helpen. Dat is iets wat je liever niet deelt met de rest van de wereld. De query parameter “code” is de secret key die je nodig hebt om dit uit te kunnen voeren, deze moet je in principe nooit delen.

Klik voor de gein nu eens op de URL voor “statusQueryGetUri” of kopieer deze uit je JSON als je geen JSON Viewer hebt en plak deze in een nieuw window.

Hier zie je meer informatie over de workflow die je aangeklikt had. Zo zien we hier de naam van de orchestrator (O_PerformRegistration), de instanceId van de workflow, de status (completed) en nog meer. Dit scherm geeft ons, zeker later als we meer interessante dingen gaan doen, heel veel info.

Onze eerste activity

Ik wil je niet achter later zonder een echte activity. In de volgende post gaan we deze uitbreiden en er meer induiken, maar ik wil hem nu toch alvast even geven.

Maak een nieuwe class aan (ik noem hem ActivityFunctions) met daarin een method. Deze ziet er bij mij zo uit:

using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;

namespace MeetingRegistration
{
    public static class ActivityFunctions
    {
        [FunctionName("A_ValidateInput")]
        public static bool ValidateInput([ActivityTrigger] string email, 
            ILogger log)
        {
            var isValid = email.IndexOf('@') != -1;
            log.LogWarning($"Email {email} check result: {isValid}");

            return isValid;
        }
    }
}

Het moet nu redelijk bekend voorkomen: een class, een method en de namen maken niet zo veel uit. Een functionname, (dit keer met A_ ervoor om mij later duidelijk te maken dat dit een activity is).

We hebben een ActivityTrigger hier als trigger. Echter, welk data type we hier doorgeven is niet relevant voor de runtime. We controleren hier alleen maar het email adres, dus we geven alleen die maar door. Weet je nog? Micro services krijgen zo weinig mogelijk data en dus afhankelijkheden.

Deze code is nogal dom: we kijken of het email adres geldig is door te kijken of er een @ teken in staat. Een regular expression zou beter zijn, maar ik beschouw ze als write-once code dus die probeer ik zo veel mogelijk te vermijden. Verder loggen we het resultaat.

Deze functie zullen we verder niet uitbreiden. Hij is goed zo.

Om deze nu te gebruiken, moeten we deze aanroepen. Duh. Omdat te doen moeten we terug naar onze orchestrator.

Als eerste moeten we even de return type en signature van de orchestrator aanpassen. Dit was public static void, maar dat wordt nu public static async Task. We gaan namelijk onze activity async aanroepen (ook al is er niets async aan, maar dat leg ik volgende post uit).

In onze starter function geven we een Attendee instance mee aan de orchestrator. Die moeten we op de een of andere manier uit lezen. Gelukkig is daar een mooie method voor: client.GetInput<T>(). Daarmee kunnen we de parameters uitlezen die in deze instance van de workflow zitten. Het lijkt dus wel stateful, maar let op: dat is het niet! Het lijkt alleen maar zo..

De code:

[FunctionName("O_PerformRegistration")]
public static async Task PerformRegistration(
    [OrchestrationTrigger] DurableOrchestrationContext ctx,
    ILogger log)
{
    log.LogWarning("We are in the orchestrator! Yeah!");
    var attendee = ctx.GetInput<Attendee>();
            
    await ctx.CallActivityAsync<bool>("A_ValidateInput", attendee.Email);
}

De signature is nu dus public static async Task geworden.

Na de log lezen we de Attendee uit met een call naar ctx.GetInput<Attendee>()

En dan: de magie. we roepen onze eerste activity aan! Deze is asynchroon dus die moeten we awaiten. Deze geeft een bool terug (want: true als er een @ in zit en anders False). De function heet A_ValidateInput. En de data is attendee.Email.

Er is dus een verschil tussen orchestration functions en activity functions als het gaat om het doorgeven van data. Hoewel je bij beiden de data doorgeeft in de call naar de client of context (in de starter zeiden we “StartNewAsync(“O_PerformRegistration”, attendee) en hier zeggen we CallActivityAsync<bool>(“A_ValidateInput”, attendee.Email)), lees je ze uit op verschillende manieren: in de orchestration function gebruiken we ctx.GetInput en in de activity function krijgen we hem geinjecteerd in de functie aanroep. Het is maar dat je het weet…

Tijd om te runnen!

De output in de browser is ongeveer hetzelfde als voorgaande keren. De output van de runtime is anders: we zien dat onze activity aangeroepen is en we zien of we een @ in het email veld hebben!

En…. voor de scherpzinnigen onder ons: inderdaad, de regel “We are in the orchestrator! Yeah!” staat er twee keer. En dat is geen bug. Dat is bij design…

Maar dat bespreken we de volgende keer!