Cenário de fan-out/fan-in em funções duráveis - Exemplo de backup na nuvem

Artigo
07/27/2024

Fan-out/fan-in refere-se ao padrão de executar várias funções simultaneamente e, em seguida, executar alguma agregação nos resultados. Este artigo explica um exemplo que usa funções duráveis para implementar um cenário de fan-in/fan-out. O exemplo é uma função durável que faz backup de todo ou parte do conteúdo do site de um aplicativo no Armazenamento do Azure.

Nota

A versão 4 do modelo de programação Node.js para o Azure Functions está disponível em geral. O novo modelo v4 foi projetado para ter uma experiência mais flexível e intuitiva para desenvolvedores de JavaScript e TypeScript. Saiba mais sobre as diferenças entre v3 e v4 no guia de migração.

Nos trechos de código a seguir, JavaScript (PM4) indica o modelo de programação V4, a nova experiência.

Pré-requisitos

Descrição geral do cenário

Neste exemplo, as funções carregam todos os arquivos em um diretório especificado recursivamente no armazenamento de blob. Eles também contam o número total de bytes que foram carregados.

É possível escrever uma única função que cuida de tudo. O principal problema que você encontraria é a escalabilidade. Uma única execução de função só pode ser executada em uma única máquina virtual, portanto, a taxa de transferência será limitada pela taxa de transferência dessa única VM. Outro problema é a fiabilidade. Se houver uma falha no meio do caminho, ou se todo o processo demorar mais de 5 minutos, o backup poderá falhar em um estado parcialmente concluído. Teria então de ser reiniciado.

Uma abordagem mais robusta seria escrever duas funções regulares: uma enumeraria os arquivos e adicionaria os nomes de arquivo a uma fila e outra leria a fila e carregaria os arquivos para o armazenamento de blobs. Essa abordagem é melhor em termos de taxa de transferência e confiabilidade, mas exige que você provisione e gerencie uma fila. Mais importante ainda, uma complexidade significativa é introduzida em termos de gestão e coordenação do estado se você quiser fazer algo mais, como relatar o número total de bytes carregados.

Uma abordagem de funções duráveis oferece todos os benefícios mencionados com despesas gerais muito baixas.

As funções

Este artigo explica as seguintes funções no aplicativo de exemplo:

E2_BackupSiteContent: Uma função orchestrator que chama E2_GetFileList para obter uma lista de arquivos para backup e, em seguida, chama E2_CopyFileToBlob para fazer backup de cada arquivo.
E2_GetFileList: Uma função de atividade que retorna uma lista de arquivos em um diretório.
E2_CopyFileToBlob: Uma função de atividade que faz backup de um único arquivo no Armazenamento de Blobs do Azure.

E2_BackupSiteContent função orquestradora

Esta função de orquestrador faz essencialmente o seguinte:

Usa um rootDirectory valor como um parâmetro de entrada.
Chama uma função para obter uma lista recursiva de arquivos em rootDirectory.
Faz várias chamadas de função paralelas para carregar cada arquivo no Armazenamento de Blobs do Azure.
Aguarda a conclusão de todos os carregamentos.
Retorna a soma total de bytes que foram carregados no Armazenamento de Blob do Azure.

Aqui está o código que implementa a função orchestrator:

[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;
}

Observe a await Task.WhenAll(tasks); linha. Todas as chamadas individuais para a função não foram esperadas, o E2_CopyFileToBlob que lhes permite correr em paralelo. Quando passamos essa matriz de tarefas para Task.WhenAllo , recebemos de volta uma tarefa que não será concluída até que todas as operações de cópia tenham sido concluídas. Se você estiver familiarizado com a TPL (Task Parallel Library) no .NET, isso não é novo para você. A diferença é que essas tarefas podem ser executadas em várias máquinas virtuais simultaneamente, e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Depois de aguardar do , sabemos que todas as chamadas de Task.WhenAllfunção foram concluídas e retornaram os valores de volta para nós. Cada chamada para E2_CopyFileToBlob retorna o número de bytes carregados, portanto, calcular a contagem total de bytes da soma é uma questão de adicionar todos esses valores de retorno juntos.

A função usa o function.json padrão para funções de orquestrador.

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

Aqui está o código que implementa a função orchestrator:

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;
});

Observe a yield context.df.Task.all(tasks); linha. Todas as chamadas individuais para a função não foram produzidas, o E2_CopyFileToBlob que permite que elas sejam executadas em paralelo. Quando passamos essa matriz de tarefas para context.df.Task.allo , recebemos de volta uma tarefa que não será concluída até que todas as operações de cópia tenham sido concluídas. Se você está familiarizado com Promise.all JavaScript, então isso não é novo para você. A diferença é que essas tarefas podem ser executadas em várias máquinas virtuais simultaneamente, e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Nota

Embora as tarefas sejam conceitualmente semelhantes às promessas do JavaScript, as funções do orquestrador devem usar context.df.Task.all e context.df.Task.any em vez de e Promise.race para gerenciar a paralelização de Promise.all tarefas.

Depois de ceder do , sabemos que todas as chamadas de context.df.Task.allfunção foram concluídas e retornaram valores para nós. Cada chamada para E2_CopyFileToBlob retorna o número de bytes carregados, portanto, calcular a contagem total de bytes da soma é uma questão de adicionar todos esses valores de retorno juntos.

Aqui está o código que implementa a função orchestrator:

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;
});

Observe a yield context.df.Task.all(tasks); linha. Todas as chamadas individuais para a função não foram produzidas, o copyFileToBlob que permite que elas sejam executadas em paralelo. Quando passamos essa matriz de tarefas para context.df.Task.allo , recebemos de volta uma tarefa que não será concluída até que todas as operações de cópia tenham sido concluídas. Se você está familiarizado com Promise.all JavaScript, então isso não é novo para você. A diferença é que essas tarefas podem ser executadas em várias máquinas virtuais simultaneamente, e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Nota

Embora as tarefas sejam conceitualmente semelhantes às promessas de JavaScript, as funções do orquestrador devem usar context.df.Task.all e context.df.Task.any em vez de e Promise.race para gerenciar a paralelização de Promise.all tarefas.

Depois de ceder do , sabemos que todas as chamadas de context.df.Task.allfunção foram concluídas e retornaram valores para nós. Cada chamada para copyFileToBlob retorna o número de bytes carregados, portanto, calcular a contagem total de bytes da soma é uma questão de adicionar todos esses valores de retorno juntos.

A função usa o function.json padrão para funções de orquestrador.

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

Aqui está o código que implementa a função orchestrator:

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)

Observe a yield context.task_all(tasks); linha. Todas as chamadas individuais para a função não foram produzidas, o E2_CopyFileToBlob que permite que elas sejam executadas em paralelo. Quando passamos essa matriz de tarefas para context.task_allo , recebemos de volta uma tarefa que não será concluída até que todas as operações de cópia tenham sido concluídas. Se você está familiarizado com asyncio.gather Python, então isso não é novo para você. A diferença é que essas tarefas podem ser executadas em várias máquinas virtuais simultaneamente, e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Nota

Embora as tarefas sejam conceitualmente semelhantes aos waitables do Python, as funções do orquestrador devem ser usadasyield, bem como as APIs e context.task_any para gerenciar a context.task_all paralelização de tarefas.

Depois de ceder do , sabemos que todas as chamadas de context.task_allfunção foram concluídas e retornaram valores para nós. Cada chamada para E2_CopyFileToBlob retorna o número de bytes carregados, para que possamos calcular a contagem total de bytes somando todos os valores de retorno juntos.

Funções de atividade auxiliar

As funções de atividade auxiliar, como com outros exemplos, são apenas funções regulares que usam a ligação de activityTrigger gatilho.

E2_GetFileList função de atividade

[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;
}

O arquivo function.json para E2_GetFileList tem a seguinte aparência:

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

E aqui está a implementação:

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);
        }
    );
};

A função usa o readdirp módulo (versão 2.x) para ler recursivamente a estrutura de diretórios.

Aqui está a implementação da função de getFileList atividade:

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;
    },
});

A função usa o readdirp módulo (versão 3.x) para ler recursivamente a estrutura de diretórios.

O arquivo function.json para E2_GetFileList tem a seguinte aparência:

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

E aqui está a implementação:

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

Nota

Você pode estar se perguntando por que você não poderia simplesmente colocar esse código diretamente na função de orquestrador. Você poderia, mas isso quebraria uma das regras fundamentais das funções do orquestrador, que é que eles nunca devem fazer E/S, incluindo acesso ao sistema de arquivos local. Para obter mais informações, consulte Restrições de código de função do Orchestrator.

E2_CopyFileToBlob função de atividade

[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;
}

Nota

Você precisará instalar o Microsoft.Azure.WebJobs.Extensions.Storage pacote NuGet para executar o código de exemplo.

A função usa alguns recursos avançados das associações do Azure Functions (ou seja, o uso do Binder parâmetro), mas você não precisa se preocupar com esses detalhes para o propósito deste passo a passo.

O arquivo function.json para E2_CopyFileToBlob é igualmente simples:

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

A implementação JavaScript usa o SDK de Armazenamento do Azure para Nó para carregar os arquivos no Armazenamento de Blobs do Azure.

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);
                }
            );
        });
    });
};

A implementação JavaScript do usa uma associação de saída do Armazenamento do Azure para carregar os arquivos no armazenamento de copyFileToBlob Blob do Azure.

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;
    },
});

O arquivo function.json para E2_CopyFileToBlob é igualmente simples:

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

A implementação Python usa o SDK de Armazenamento do Azure para Python para carregar os arquivos no Armazenamento de Blobs do Azure.

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

A implementação carrega o arquivo do disco e transmite o conteúdo de forma assíncrona em um blob de mesmo nome no contêiner "backups". O valor de retorno é o número de bytes copiados para o armazenamento, que é usado pela função orchestrator para calcular a soma agregada.

Nota

Este é um exemplo perfeito de como mover operações de E/S para uma activityTrigger função. Não só o trabalho pode ser distribuído por muitas máquinas diferentes, mas você também obtém os benefícios de verificar o progresso. Se o processo do anfitrião for encerrado por qualquer motivo, sabe quais os carregamentos que já foram concluídos.

Executar o exemplo

Você pode iniciar a orquestração, no Windows, enviando a seguinte solicitação HTTP POST.

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

"D:\\home\\LogFiles"

Como alternativa, em um aplicativo de função Linux (Python atualmente só é executado no Linux for App Service), você pode iniciar a orquestração assim:

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

"/home/site/wwwroot"

Nota

A HttpStart função que você está invocando só funciona com conteúdo formatado em JSON. Por esse motivo, o Content-Type: application/json cabeçalho é necessário e o caminho do diretório é codificado como uma cadeia de caracteres JSON. Além disso, o trecho HTTP pressupõe que há uma entrada no arquivo que remove o prefixo padrão api/ de todas as URLs de funções de gatilho host.json HTTP. Você pode encontrar a marcação para essa configuração no host.json arquivo nos exemplos.

Essa solicitação HTTP aciona o E2_BackupSiteContent orquestrador e passa a string D:\home\LogFiles como um parâmetro. A resposta fornece um link para obter o status da operação de backup:

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...)

Dependendo de quantos arquivos de log você tem em seu aplicativo de função, essa operação pode levar vários minutos para ser concluída. Você pode obter o status mais recente consultando a Location URL no cabeçalho da resposta HTTP 202 anterior.

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"}

Neste caso, a função ainda está em execução. Você pode ver a entrada que foi salva no estado do orquestrador e a última hora atualizada. Você pode continuar a usar os valores de Location cabeçalho para sondar para conclusão. Quando o status é "Concluído", você vê um valor de resposta HTTP semelhante ao seguinte:

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"}

Agora você pode ver que a orquestração está completa e aproximadamente quanto tempo levou para ser concluída. Você também vê um valor para o output campo, que indica que cerca de 450 KB de logs foram carregados.

Próximos passos

Este exemplo mostrou como implementar o padrão fan-out/fan-in. O próximo exemplo mostra como implementar o padrão de monitor usando temporizadores duráveis.

Executar o exemplo de monitor

Partilhar via

Cenário de fan-out/fan-in em funções duráveis - Exemplo de backup na nuvem

Pré-requisitos

Descrição geral do cenário

As funções

E2_BackupSiteContent função orquestradora

Funções de atividade auxiliar

E2_GetFileList função de atividade

E2_CopyFileToBlob função de atividade

Executar o exemplo

Próximos passos

Comentários

Recursos adicionais