지속성 함수의 팬아웃/팬인 시나리오 - 클라우드 백업 예제

• 빠른 시작 문서 완료
• GitHub에서 샘플 프로젝트 복제 또는 다운로드

• 빠른 시작 문서 완료
• GitHub에서 샘플 프로젝트 복제 또는 다운로드

• 빠른 시작 문서 완료
• GitHub에서 샘플 프로젝트 복제 또는 다운로드

• 빠른 시작 문서 완료
• GitHub에서 샘플 프로젝트 복제 또는 다운로드

다음은 오케스트레이터 함수를 구현하는 코드입니다.

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

await Task.WhenAll(tasks); 줄에 유의하세요. E2_CopyFileToBlob 함수에 대한 모든 개별 호출은 대기하지 않았으며 병렬로 실행할 수 있습니다. 이 작업 배열을 Task.WhenAll에 전달하면 모든 복사 작업이 완료될 때까지 완료되지 않는 작업을 다시 가져옵니다. .NET의 TPL(작업 병렬 라이브러리)에 익숙하다면 이러한 작업은 새로운 것이 아닙니다. 차이점은 이러한 작업이 여러 가상 머신에서 동시에 실행될 수 있으며, Durable Functions 확장은 엔드투엔드 실행이 프로세스 재활용에 탄력적으로 수행되도록 보장한다는 것입니다.

Task.WhenAll에서 기다린 후에 모든 함수 호출이 완료되고 값을 다시 반환했습니다. E2_CopyFileToBlob을 호출할 때마다 업로드된 바이트 수가 반환되므로 총 바이트 수를 계산하는 것은 이러한 반환 값을 모두 추가하는 문제입니다.

함수는 오케스트레이터 함수에 표준 function.json을 사용합니다.

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

다음은 오케스트레이터 함수를 구현하는 코드입니다.

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

yield context.df.Task.all(tasks); 줄에 유의하세요. E2_CopyFileToBlob 함수에 대한 모든 개별 호출은 일시 중단되지 않았으며 병렬로 실행할 수 있습니다. 이 작업 배열을 context.df.Task.all에 전달하면 모든 복사 작업이 완료될 때까지 완료되지 않는 작업을 다시 가져옵니다. JavaScript의 Promise.all에 대해 잘 알고 있는 경우 낯설지 않을 것입니다. 차이점은 이러한 작업이 여러 가상 머신에서 동시에 실행될 수 있으며, Durable Functions 확장은 엔드투엔드 실행이 프로세스 재활용에 탄력적으로 수행되도록 보장한다는 것입니다.

context.df.Task.all에서 일시 중단된 후에 모든 함수 호출이 완료되고 값을 다시 반환했습니다. E2_CopyFileToBlob을 호출할 때마다 업로드된 바이트 수가 반환되므로 총 바이트 수를 계산하는 것은 이러한 반환 값을 모두 추가하는 문제입니다.

다음은 오케스트레이터 함수를 구현하는 코드입니다.

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

yield context.df.Task.all(tasks); 줄에 유의하세요. copyFileToBlob 함수에 대한 모든 개별 호출은 일시 중단되지 않았으며 병렬로 실행할 수 있습니다. 이 작업 배열을 context.df.Task.all에 전달하면 모든 복사 작업이 완료될 때까지 완료되지 않는 작업을 다시 가져옵니다. JavaScript의 Promise.all에 대해 잘 알고 있는 경우 낯설지 않을 것입니다. 차이점은 이러한 작업이 여러 가상 머신에서 동시에 실행될 수 있으며, Durable Functions 확장은 엔드투엔드 실행이 프로세스 재활용에 탄력적으로 수행되도록 보장한다는 것입니다.

context.df.Task.all에서 일시 중단된 후에 모든 함수 호출이 완료되고 값을 다시 반환했습니다. copyFileToBlob을 호출할 때마다 업로드된 바이트 수가 반환되므로 총 바이트 수를 계산하는 것은 이러한 반환 값을 모두 추가하는 문제입니다.

함수는 오케스트레이터 함수에 표준 function.json을 사용합니다.

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

다음은 오케스트레이터 함수를 구현하는 코드입니다.

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)

yield context.task_all(tasks); 줄에 유의하세요. E2_CopyFileToBlob 함수에 대한 모든 개별 호출은 일시 중단되지 않았으며 병렬로 실행할 수 있습니다. 이 작업 배열을 context.task_all에 전달하면 모든 복사 작업이 완료될 때까지 완료되지 않는 작업을 다시 가져옵니다. Python의 asyncio.gather에 대해 잘 알고 있는 경우 낯설지 않을 것입니다. 차이점은 이러한 작업이 여러 가상 머신에서 동시에 실행될 수 있으며, Durable Functions 확장은 엔드투엔드 실행이 프로세스 재활용에 탄력적으로 수행되도록 보장한다는 것입니다.

context.task_all에서 일시 중단된 후에 모든 함수 호출이 완료되고 값을 다시 반환했습니다. E2_CopyFileToBlob에 대한 각 호출은 업로드된 바이트 수를 반환하므로 모든 반환 값을 함께 추가하여 총 바이트 수를 계산할 수 있습니다.

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

E2_GetFileList에 대한 function.json 파일은 다음과 같습니다.

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

그리고 구현은 다음과 같습니다.

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

함수는 readdirp 모듈(버전 2.x)을 사용하여 디렉터리 구조를 재귀적으로 읽습니다.

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

함수는 readdirp 모듈(버전 3.x)을 사용하여 디렉터리 구조를 재귀적으로 읽습니다.

E2_GetFileList에 대한 function.json 파일은 다음과 같습니다.

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

그리고 구현은 다음과 같습니다.

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

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

이 함수는 Azure Functions 바인딩의 몇 가지 고급 기능(즉, Binder 매개 변수 사용)을 사용하지만 이 연습에서는 이러한 세부 정보에 대해 걱정할 필요가 없습니다.

E2_CopyFileToBlob에 대한 function.json 파일도 마찬가지로 다음과 같이 간단합니다.

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

JavaScript 구현은 Azure Storage SDK for Node를 사용하여 파일을 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);
                }
            );
        });
    });
};

copyFileToBlob의 JavaScript 구현은 Azure Storage 출력 바인딩을 사용하여 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;
    },
});

E2_CopyFileToBlob에 대한 function.json 파일도 마찬가지로 다음과 같이 간단합니다.

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

Python 구현은 Azure Storage SDK for Python을 사용하여 파일을 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