Fan-out/fan-in scenario i Durable Functions – Exempel på molnsäkerhetskopiering

Artikel
10/01/2023

Utloggning/utloggning avser mönstret för att köra flera funktioner samtidigt och sedan utföra viss aggregering på resultaten. I den här artikeln förklaras ett exempel som använder Durable Functions för att implementera ett scenario med in-/ut fläkt. Exemplet är en beständig funktion som säkerhetskopierar allt eller en del av en apps webbplatsinnehåll i Azure Storage.

Anteckning

Version 4 av Node.js programmeringsmodell för Azure Functions är allmänt tillgänglig. Den nya v4-modellen är utformad för att ha en mer flexibel och intuitiv upplevelse för JavaScript- och TypeScript-utvecklare. Läs mer om skillnaderna mellan v3 och v4 i migreringsguiden.

I följande kodfragment anger JavaScript (PM4) programmeringsmodellen V4, den nya upplevelsen.

Förutsättningar

Översikt över scenario

I det här exemplet laddar funktionerna upp alla filer under en angiven katalog rekursivt till Blob Storage. De räknar också det totala antalet byte som har laddats upp.

Det går att skriva en enda funktion som tar hand om allt. Det största problemet du stöter på är skalbarhet. En enda funktionskörning kan bara köras på en enda virtuell dator, så dataflödet begränsas av dataflödet för den enskilda virtuella datorn. Ett annat problem är tillförlitlighet. Om det uppstår ett fel halvvägs, eller om hela processen tar mer än 5 minuter, kan säkerhetskopieringen misslyckas i ett delvis slutfört tillstånd. Den skulle då behöva startas om.

En mer robust metod är att skriva två vanliga funktioner: en skulle räkna upp filerna och lägga till filnamnen i en kö, och en annan skulle läsa från kön och ladda upp filerna till Blob Storage. Den här metoden är bättre när det gäller dataflöde och tillförlitlighet, men du måste etablera och hantera en kö. Ännu viktigare är att betydande komplexitet införs när det gäller tillståndshantering och samordning om du vill göra något mer, som att rapportera det totala antalet uppladdade byte.

En Durable Functions metod ger dig alla de nämnda fördelarna med mycket låga omkostnader.

Funktionerna

Den här artikeln förklarar följande funktioner i exempelappen:

E2_BackupSiteContent: En orkestreringsfunktion som anropar E2_GetFileList för att hämta en lista över filer som ska säkerhetskopieras och sedan anropar E2_CopyFileToBlob för att säkerhetskopiera varje fil.
E2_GetFileList: En aktivitetsfunktion som returnerar en lista med filer i en katalog.
E2_CopyFileToBlob: En aktivitetsfunktion som säkerhetskopierar en enda fil för att Azure Blob Storage.

E2_BackupSiteContent orchestrator-funktion

Den här orkestreringsfunktionen gör i princip följande:

Tar ett rootDirectory värde som en indataparameter.
Anropar en funktion för att hämta en rekursiv lista över filer under rootDirectory.
Gör flera parallella funktionsanrop för att ladda upp varje fil till Azure Blob Storage.
Väntar på att alla uppladdningar ska slutföras.
Returnerar summan av totalt antal byte som laddats upp till Azure Blob Storage.

Här är koden som implementerar orchestrator-funktionen:

[FunctionName("E2_BackupSiteContent")]
public static async Task<long> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext backupContext)
{
    string rootDirectory = backupContext.GetInput<string>()?.Trim();
    if (string.IsNullOrEmpty(rootDirectory))
    {
        rootDirectory = Directory.GetParent(typeof(BackupSiteContent).Assembly.Location).FullName;
    }

    string[] files = await backupContext.CallActivityAsync<string[]>(
        "E2_GetFileList",
        rootDirectory);

    var tasks = new Task<long>[files.Length];
    for (int i = 0; i < files.Length; i++)
    {
        tasks[i] = backupContext.CallActivityAsync<long>(
            "E2_CopyFileToBlob",
            files[i]);
    }

    await Task.WhenAll(tasks);

    long totalBytes = tasks.Sum(t => t.Result);
    return totalBytes;
}

Lägg märke till await Task.WhenAll(tasks); raden. Alla enskilda anrop till E2_CopyFileToBlob funktionen väntades inte , vilket gör att de kan köras parallellt. När vi skickar den här matrisen med aktiviteter till Task.WhenAllfår vi tillbaka en uppgift som inte slutförs förrän alla kopieringsåtgärder har slutförts. Om du är bekant med TPL (Task Parallel Library) i .NET är detta inte nytt för dig. Skillnaden är att dessa uppgifter kan köras på flera virtuella datorer samtidigt, och tillägget Durable Functions säkerställer att körningen från slutpunkt till slutpunkt är motståndskraftig mot processåtervinning.

Efter att ha väntat från Task.WhenAllvet vi att alla funktionsanrop har slutförts och har returnerat värden tillbaka till oss. Varje anrop till E2_CopyFileToBlob returnerar antalet uppladdade byte, så att beräkna det totala antalet byte handlar om att lägga till alla dessa returvärden tillsammans.

Funktionen använder standard function.json för orchestrator-funktioner.

{
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

Här är koden som implementerar orchestrator-funktionen:

const df = require("durable-functions");

module.exports = df.orchestrator(function* (context) {
    const rootDirectory = context.df.getInput();
    if (!rootDirectory) {
        throw new Error("A directory path is required as an input.");
    }

    const files = yield context.df.callActivity("E2_GetFileList", rootDirectory);

    // Backup Files and save Promises into array
    const tasks = [];
    for (const file of files) {
        tasks.push(context.df.callActivity("E2_CopyFileToBlob", file));
    }

    // wait for all the Backup Files Activities to complete, sum total bytes
    const results = yield context.df.Task.all(tasks);
    const totalBytes = results.reduce((prev, curr) => prev + curr, 0);

    // return results;
    return totalBytes;
});

Lägg märke till yield context.df.Task.all(tasks); raden. Alla enskilda anrop till E2_CopyFileToBlob funktionen returnerades inte , vilket gör att de kan köras parallellt. När vi skickar den här matrisen med aktiviteter till context.df.Task.allfår vi tillbaka en uppgift som inte slutförs förrän alla kopieringsåtgärder har slutförts. Om du är bekant med Promise.all i JavaScript är detta inte nytt för dig. Skillnaden är att dessa uppgifter kan köras på flera virtuella datorer samtidigt, och tillägget Durable Functions säkerställer att körningen från slutpunkt till slutpunkt är motståndskraftig mot processåtervinning.

Anteckning

Även om uppgifter begreppsmässigt liknar JavaScript-löften bör orchestrator-funktioner använda context.df.Task.all och context.df.Task.any i stället för Promise.all och Promise.race för att hantera uppgiftsparallellisering.

När vi har gett från context.df.Task.allvet vi att alla funktionsanrop har slutförts och har returnerat värden tillbaka till oss. Varje anrop till E2_CopyFileToBlob returnerar antalet uppladdade byte, så att beräkna det totala antalet byte handlar om att lägga till alla dessa returvärden tillsammans.

Här är koden som implementerar orchestrator-funktionen:

const df = require("durable-functions");
const path = require("path");
const getFileListActivityName = "getFileList";
const copyFileToBlobActivityName = "copyFileToBlob";

df.app.orchestration("backupSiteContent", function* (context) {
    const rootDir = context.df.getInput();
    if (!rootDir) {
        throw new Error("A directory path is required as an input.");
    }

    const rootDirAbs = path.resolve(rootDir);
    const files = yield context.df.callActivity(getFileListActivityName, rootDirAbs);

    // Backup Files and save Tasks into array
    const tasks = [];
    for (const file of files) {
        const input = {
            backupPath: path.relative(rootDirAbs, file).replace("\\", "/"),
            filePath: file,
        };
        tasks.push(context.df.callActivity(copyFileToBlobActivityName, input));
    }

    // wait for all the Backup Files Activities to complete, sum total bytes
    const results = yield context.df.Task.all(tasks);
    const totalBytes = results ? results.reduce((prev, curr) => prev + curr, 0) : 0;

    // return results;
    return totalBytes;
});

Lägg märke till yield context.df.Task.all(tasks); raden. Alla enskilda anrop till copyFileToBlob funktionen returnerades inte , vilket gör att de kan köras parallellt. När vi skickar den här matrisen med aktiviteter till context.df.Task.allfår vi tillbaka en uppgift som inte slutförs förrän alla kopieringsåtgärder har slutförts. Om du är bekant med Promise.all i JavaScript är detta inte nytt för dig. Skillnaden är att dessa uppgifter kan köras på flera virtuella datorer samtidigt, och tillägget Durable Functions säkerställer att körningen från slutpunkt till slutpunkt är motståndskraftig mot processåtervinning.

Anteckning

När vi har gett från context.df.Task.allvet vi att alla funktionsanrop har slutförts och har returnerat värden tillbaka till oss. Varje anrop till copyFileToBlob returnerar antalet uppladdade byte, så att beräkna det totala antalet byte handlar om att lägga till alla dessa returvärden tillsammans.

Funktionen använder standard function.json för orchestrator-funktioner.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ]
}

Här är koden som implementerar orchestrator-funktionen:

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


def orchestrator_function(context: df.DurableOrchestrationContext):

    root_directory: str = context.get_input()

    if not root_directory:
        raise Exception("A directory path is required as input")

    files = yield context.call_activity("E2_GetFileList", root_directory)
    tasks = []
    for file in files:
        tasks.append(context.call_activity("E2_CopyFileToBlob", file))
    
    results = yield context.task_all(tasks)
    total_bytes = sum(results)
    return total_bytes

main = df.Orchestrator.create(orchestrator_function)

Lägg märke till yield context.task_all(tasks); raden. Alla enskilda anrop till E2_CopyFileToBlob funktionen returnerades inte , vilket gör att de kan köras parallellt. När vi skickar den här matrisen med aktiviteter till context.task_allfår vi tillbaka en uppgift som inte slutförs förrän alla kopieringsåtgärder har slutförts. Om du är bekant med asyncio.gather i Python är detta inte nytt för dig. Skillnaden är att dessa uppgifter kan köras på flera virtuella datorer samtidigt, och tillägget Durable Functions säkerställer att körningen från slutpunkt till slutpunkt är motståndskraftig mot processåtervinning.

Anteckning

Även om uppgifter begreppsmässigt liknar Python awaitables bör orchestrator-funktioner använda yield samt API:erna context.task_all och context.task_any för att hantera uppgiftsparallellisering.

När vi har gett från context.task_allvet vi att alla funktionsanrop har slutförts och har returnerat värden tillbaka till oss. Varje anrop till returnerar antalet uppladdade byte, så vi kan beräkna det totala antalet byte genom att E2_CopyFileToBlob lägga till alla returvärden tillsammans.

Aktivitetsfunktioner för hjälpkomponenter

Hjälpaktivitetsfunktionerna är precis som med andra exempel bara vanliga funktioner som använder utlösarbindningen activityTrigger .

E2_GetFileList aktivitetsfunktion

[FunctionName("E2_GetFileList")]
public static string[] GetFileList(
    [ActivityTrigger] string rootDirectory, 
    ILogger log)
{
    log.LogInformation($"Searching for files under '{rootDirectory}'...");
    string[] files = Directory.GetFiles(rootDirectory, "*", SearchOption.AllDirectories);
    log.LogInformation($"Found {files.Length} file(s) under {rootDirectory}.");

    return files;
}

Filen function.json för E2_GetFileList ser ut så här:

{
  "bindings": [
    {
      "name": "rootDirectory",
      "type": "activityTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

Och här är implementeringen:

const readdirp = require("readdirp");

module.exports = function (context, rootDirectory) {
    context.log(`Searching for files under '${rootDirectory}'...`);
    const allFilePaths = [];

    readdirp(
        { root: rootDirectory, entryType: "all" },
        function (fileInfo) {
            if (!fileInfo.stat.isDirectory()) {
                allFilePaths.push(fileInfo.fullPath);
            }
        },
        function (err, res) {
            if (err) {
                throw err;
            }

            context.log(`Found ${allFilePaths.length} under ${rootDirectory}.`);
            context.done(null, allFilePaths);
        }
    );
};

Funktionen använder modulen readdirp (version 2.x) för att rekursivt läsa katalogstrukturen.

Här är implementeringen av aktivitetsfunktionen getFileList :

const df = require("durable-functions");
const readdirp = require("readdirp");
const getFileListActivityName = "getFileList";

df.app.activity(getFileListActivityName, {
    handler: async function (rootDirectory, context) {
        context.log(`Searching for files under '${rootDirectory}'...`);

        const allFilePaths = [];
        for await (const entry of readdirp(rootDirectory, { type: "files" })) {
            allFilePaths.push(entry.fullPath);
        }
        context.log(`Found ${allFilePaths.length} under ${rootDirectory}.`);
        return allFilePaths;
    },
});

Funktionen använder modulen readdirp (version 3.x) för att rekursivt läsa katalogstrukturen.

Filen function.json för E2_GetFileList ser ut så här:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "rootDirectory",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

Och här är implementeringen:

import os
from os.path import dirname
from typing import List

def main(rootDirectory: str) -> List[str]:

    all_file_paths = []
    # We walk the file system
    for path, _, files in os.walk(rootDirectory):
        # We copy the code for activities and orchestrators
        if "E2_" in path:
            # For each file, we add their full-path to the list
            for name in files:
                if name == "__init__.py" or name == "function.json":
                    file_path = os.path.join(path, name)
                    all_file_paths.append(file_path)
    
    return all_file_paths

Anteckning

Du kanske undrar varför du inte bara kunde placera den här koden direkt i orchestrator-funktionen. Du kan, men detta skulle bryta en av de grundläggande reglerna för orchestrator-funktioner, vilket är att de aldrig bör göra I/O, inklusive åtkomst till lokala filsystem. Mer information finns i Begränsningar för Orchestrator-funktionskod.

E2_CopyFileToBlob aktivitetsfunktion

[FunctionName("E2_CopyFileToBlob")]
public static async Task<long> CopyFileToBlob(
    [ActivityTrigger] string filePath,
    Binder binder,
    ILogger log)
{
    long byteCount = new FileInfo(filePath).Length;

    // strip the drive letter prefix and convert to forward slashes
    string blobPath = filePath
        .Substring(Path.GetPathRoot(filePath).Length)
        .Replace('\\', '/');
    string outputLocation = $"backups/{blobPath}";

    log.LogInformation($"Copying '{filePath}' to '{outputLocation}'. Total bytes = {byteCount}.");

    // copy the file contents into a blob
    using (Stream source = File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.Read))
    using (Stream destination = await binder.BindAsync<CloudBlobStream>(
        new BlobAttribute(outputLocation, FileAccess.Write)))
    {
        await source.CopyToAsync(destination);
    }

    return byteCount;
}

Anteckning

Du måste installera Microsoft.Azure.WebJobs.Extensions.Storage NuGet-paketet för att köra exempelkoden.

Funktionen använder vissa avancerade funktioner i Azure Functions bindningar (dvs. användningen av parameternBinder), men du behöver inte bekymra dig om informationen i den här genomgången.

Filen function.json för E2_CopyFileToBlob är på samma sätt enkel:

{
  "bindings": [
    {
      "name": "filePath",
      "type": "activityTrigger",
      "direction": "in"
    },
    {
      "name": "out",
      "type": "blob",
      "path": "",
      "connection": "AzureWebJobsStorage",
      "direction": "out"
    }
  ],
  "disabled": false
}

JavaScript-implementeringen använder Azure Storage SDK för Node för att ladda upp filerna till Azure Blob Storage.

const fs = require("fs");
const path = require("path");
const storage = require("azure-storage");

module.exports = function (context, filePath) {
    const container = "backups";
    const root = path.parse(filePath).root;
    const blobPath = filePath.substring(root.length).replace("\\", "/");
    const outputLocation = `backups/${blobPath}`;
    const blobService = storage.createBlobService();

    blobService.createContainerIfNotExists(container, (error) => {
        if (error) {
            throw error;
        }

        fs.stat(filePath, function (error, stats) {
            if (error) {
                throw error;
            }
            context.log(
                `Copying '${filePath}' to '${outputLocation}'. Total bytes = ${stats.size}.`
            );

            const readStream = fs.createReadStream(filePath);

            blobService.createBlockBlobFromStream(
                container,
                blobPath,
                readStream,
                stats.size,
                function (error) {
                    if (error) {
                        throw error;
                    }

                    context.done(null, stats.size);
                }
            );
        });
    });
};

JavaScript-implementeringen av copyFileToBlob använder en Azure Storage-utdatabindning för att ladda upp filerna till Azure Blob Storage.

const df = require("durable-functions");
const fs = require("fs/promises");
const { output } = require("@azure/functions");

const copyFileToBlobActivityName = "copyFileToBlob";

const blobOutput = output.storageBlob({
    path: "backups/{backupPath}",
    connection: "StorageConnString",
});

df.app.activity(copyFileToBlobActivityName, {
    extraOutputs: [blobOutput],
    handler: async function ({ backupPath, filePath }, context) {
        const outputLocation = `backups/${backupPath}`;
        const stats = await fs.stat(filePath);
        context.log(`Copying '${filePath}' to '${outputLocation}'. Total bytes = ${stats.size}.`);

        const fileContents = await fs.readFile(filePath);

        context.extraOutputs.set(blobOutput, fileContents);

        return stats.size;
    },
});

Filen function.json för E2_CopyFileToBlob är på samma sätt enkel:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "filePath",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

Python-implementeringen använder Azure Storage SDK för Python för att ladda upp filerna till Azure Blob Storage.

import os
import pathlib
from azure.storage.blob import BlobServiceClient
from azure.core.exceptions import ResourceExistsError

connect_str = os.getenv('AzureWebJobsStorage')

def main(filePath: str) -> str:
    # Create the BlobServiceClient object which will be used to create a container client
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)
    
    # Create a unique name for the container
    container_name = "backups"
    
    # Create the container if it does not exist
    try:
        blob_service_client.create_container(container_name)
    except ResourceExistsError:
        pass

    # Create a blob client using the local file name as the name for the blob
    parent_dir, fname = pathlib.Path(filePath).parts[-2:] # Get last two path components
    blob_name = parent_dir + "_" + fname
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)

    # Count bytes in file
    byte_count = os.path.getsize(filePath)

    # Upload the created file
    with open(filePath, "rb") as data:
        blob_client.upload_blob(data)

    return byte_count

Implementeringen läser in filen från disken och strömmar innehållet asynkront till en blob med samma namn i containern "säkerhetskopior". Returvärdet är det antal byte som kopieras till lagringen, som sedan används av orchestrator-funktionen för att beräkna aggregeringssumman.

Anteckning

Det här är ett perfekt exempel på hur du flyttar I/O-åtgärder till en activityTrigger funktion. Arbetet kan inte bara distribueras på många olika datorer, utan du får också fördelarna med att kontrollera förloppet. Om värdprocessen avslutas av någon anledning vet du vilka uppladdningar som redan har slutförts.

Kör exemplet

Du kan starta orkestreringen i Windows genom att skicka följande HTTP POST-begäran.

POST http://{host}/orchestrators/E2_BackupSiteContent
Content-Type: application/json
Content-Length: 20

"D:\\home\\LogFiles"

I en Linux-funktionsapp (Python körs för närvarande bara på Linux för App Service) kan du också starta orkestreringen så här:

POST http://{host}/orchestrators/E2_BackupSiteContent
Content-Type: application/json
Content-Length: 20

"/home/site/wwwroot"

Anteckning

Funktionen HttpStart som du anropar fungerar bara med JSON-formaterat innehåll. Därför Content-Type: application/json krävs rubriken och katalogsökvägen kodas som en JSON-sträng. Dessutom förutsätter HTTP-kodfragment att det finns en post i host.json filen som tar bort standardprefixet api/ från alla URL:er för HTTP-utlösarfunktioner. Du hittar koden för den här konfigurationen host.json i filen i exemplen.

Den här HTTP-begäran utlöser orkestreraren E2_BackupSiteContent och skickar strängen D:\home\LogFiles som en parameter. Svaret innehåller en länk för att hämta status för säkerhetskopieringsåtgärden:

HTTP/1.1 202 Accepted
Content-Length: 719
Content-Type: application/json; charset=utf-8
Location: http://{host}/runtime/webhooks/durabletask/instances/b4e9bdcc435d460f8dc008115ff0a8a9?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

(...trimmed...)

Den här åtgärden kan ta flera minuter att slutföra, beroende på hur många loggfiler du har i funktionsappen. Du kan få den senaste statusen genom att fråga URL:en i Location rubriken för föregående HTTP 202-svar.

GET http://{host}/runtime/webhooks/durabletask/instances/b4e9bdcc435d460f8dc008115ff0a8a9?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

HTTP/1.1 202 Accepted
Content-Length: 148
Content-Type: application/json; charset=utf-8
Location: http://{host}/runtime/webhooks/durabletask/instances/b4e9bdcc435d460f8dc008115ff0a8a9?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

{"runtimeStatus":"Running","input":"D:\\home\\LogFiles","output":null,"createdTime":"2019-06-29T18:50:55Z","lastUpdatedTime":"2019-06-29T18:51:16Z"}

I det här fallet körs funktionen fortfarande. Du kan se de indata som sparades i orkestreringstillståndet och den senaste uppdaterade tiden. Du kan fortsätta att använda Location rubrikvärdena för att söka efter slutförande. När statusen är "Slutförd" visas ett HTTP-svarsvärde som liknar följande:

HTTP/1.1 200 OK
Content-Length: 152
Content-Type: application/json; charset=utf-8

{"runtimeStatus":"Completed","input":"D:\\home\\LogFiles","output":452071,"createdTime":"2019-06-29T18:50:55Z","lastUpdatedTime":"2019-06-29T18:51:26Z"}

Nu kan du se att orkestreringen är klar och ungefär hur lång tid det tog att slutföra. Du ser också ett värde för output fältet, vilket indikerar att cirka 450 KB loggar laddades upp.

Nästa steg

Det här exemplet har visat hur du implementerar fan-out/fan-in-mönstret. Nästa exempel visar hur du implementerar övervakningsmönstret med hjälp av hållbara timers.

Kör övervakningsexemplet

Fan-out/fan-in scenario i Durable Functions – Exempel på molnsäkerhetskopiering

Förutsättningar

Översikt över scenario

Funktionerna

E2_BackupSiteContent orchestrator-funktion

Aktivitetsfunktioner för hjälpkomponenter

E2_GetFileList aktivitetsfunktion

E2_CopyFileToBlob aktivitetsfunktion

Kör exemplet

Nästa steg

Ytterligare resurser