Устойчивые оркестрации

Устойчивые функции — это расширение функций Azure. Функцию оркестратора можно использовать для оркестрации выполнения других устойчивых функций в приложении-функции. Функции Оркестратора имеют следующие характеристики:

• Функции Оркестратора определяют рабочие процессы функций с помощью процедурного кода. Декларативные схемы или конструкторы не требуются.
• Функции Оркестратора могут вызывать другие устойчивые функции синхронно и асинхронно. Выходные данные из вызываемой функции можно надежно сохранить в локальных переменных.
• Функции оркестратора являются устойчивыми и надежными. Автоматическая фиксация прогресса выполнения происходит в момент, когда функция "awaits" или "приостанавливает". Локальное состояние никогда не теряется при перезапуске процесса или перезагрузке виртуальной машины.
• Функции оркестратора могут быть длительными. Общий срок службы экземпляра оркестрации может составлять секунды, дни, месяцы или быть бесконечным.

В этой статье представлены общие сведения о функциях оркестратора и способах решения различных задач разработки приложений. Если вы еще не знакомы с типами функций, доступных в приложении устойчивых функций, сначала ознакомьтесь со статьей о типах устойчивых функций .

Идентификация оркестрации

Каждый экземпляр оркестрации имеет идентификатор экземпляра (также известный как идентификатор экземпляра). По умолчанию каждый идентификатор экземпляра является автоматически созданным GUID. Однако идентификаторы экземпляров также могут быть любым пользовательским строковым значением. Каждый идентификатор экземпляра оркестрации должен быть уникальным в узле задач.

Ниже приведены некоторые правила для идентификаторов экземпляров:

• Идентификаторы должны быть от 1 до 100 символов.
• Идентификаторы экземпляров не должны начинаться с @.
• Идентификаторы экземпляров не должны содержать /, \#или ? символы.
• Идентификаторы экземпляров не должны содержать управляющие символы.

Идентификатор экземпляра оркестрации является обязательным параметром для большинства операций управления экземплярами. Они также важны для диагностики, например, как поиск среди данных отслеживания оркестрации в Application Insights для таких задач, как устранение неполадок или проведение анализа. По этой причине рекомендуется сохранить идентификаторы созданных экземпляров в некотором внешнем расположении (например, в базе данных или в журналах приложений), где к ним можно будет легко обратиться позже.

Reliability

Функции Оркестратора надежно поддерживают состояние выполнения с помощью шаблона конструктора событий . Вместо прямого хранения текущего состояния оркестрации фреймворк Durable Task использует хранилище с добавлением для записи полной последовательности действий, выполняемых оркестрацией функции. Хранилище только для добавления имеет множество преимуществ по сравнению со сбросом полного состояния во время выполнения. Преимущества включают повышение производительности, масштабируемости и реагирования. Кроме того, вы получаете постепенное достижение согласованности для транзакционных данных, а также полные следы аудита и истории. Журналы аудита обеспечивают надежные компенсирующие действия.

Устойчивые функции используют обработку событий прозрачно. За кулисами оператор await (C#) или yield (JavaScript/Python) в функции оркестратора передаёт управление потоком оркестратора обратно в диспетчер фреймворка устойчивых задач. В случае Java не существует специального ключевого слова языка. Вместо этого при вызове .await() управление задачей возвращается диспетчеру с помощью пользовательского Throwable. Затем диспетчер фиксирует любые новые действия, запланированные функцией оркестратора (например, вызов одной или нескольких дочерних функций или планирование устойчивого таймера) в хранилище. Действие прозрачной фиксации обновляет историю выполнения экземпляра оркестрации через добавление всех новых событий в хранилище, как ведение журнала только с операцией добавления. Аналогичным образом действие фиксации создает сообщения в хранилище для расписания выполнения фактической работы. На этом этапе функцию оркестратора можно выгрузить из памяти. По умолчанию Устойчивые функции используют службу хранилища Azure в качестве хранилища состояний среды выполнения, но поддерживаются и другие поставщики хранилища.

Если функция оркестрации получает дополнительную работу (например, получено сообщение-ответ или истек срок действия долговременного таймера), оркестратор просыпается и снова выполняет всю функцию с самого начала для того, чтобы перестроить локальное состояние. Во время воспроизведения, если код пытается вызвать функцию (или выполнить любую другую асинхронную работу), фреймворк Durable Task обращается к журналу выполнения текущей оркестрации. Если он обнаружит, что функция действия уже выполнила и дала результат, он воспроизводит результат функции, и код оркестратора продолжает выполняться. Воспроизведение продолжается, пока функциональный код не завершится или пока не будет запланирована новая асинхронная работа.

История оркестрации

Поведение с использованием событий в платформе Durable Task тесно связано с кодом функции оркестратора, который вы пишете. Предположим, у вас есть функция оркестратора с цепочкой действий, например следующая функция оркестратора:

[FunctionName("HelloCities")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Tokyo"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

[Function("HelloCities")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Tokyo"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

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

module.exports = df.orchestrator(function*(context) {
    const output = [];
    output.push(yield context.df.callActivity("SayHello", "Tokyo"));
    output.push(yield context.df.callActivity("SayHello", "Seattle"));
    output.push(yield context.df.callActivity("SayHello", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return output;
});

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

df.app.orchestration("helloSequence", function* (context) {
    const output = [];
    output.push(yield context.df.callActivity(helloActivityName, "Tokyo"));
    output.push(yield context.df.callActivity(helloActivityName, "Seattle"));
    output.push(yield context.df.callActivity(helloActivityName, "London"));

    // Return ["Hello Tokyo!", "Hello Seattle!", "Hello London!"].
    return output;
});

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

# Create a FunctionApp instance.
app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# Orchestrator
@app.orchestration_trigger(context_name="context")
def orchestrator_function(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('SayHello', "Tokyo")
    result2 = yield context.call_activity('SayHello', "Seattle")
    result3 = yield context.call_activity('SayHello', "London")
    return [result1, result2, result3]

# Activity
@app.activity_trigger(input_name="city")
def SayHello(city: str):
    return f"Hello {city}"

param($Context)

$output = @()

$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'Tokyo'
$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'Seattle'
$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'London'

$output

@FunctionName("HelloCities")
public String helloCitiesOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String result = "";
    result += ctx.callActivity("SayHello", "Tokyo", String.class).await() + ", ";
    result += ctx.callActivity("SayHello", "Seattle", String.class).await() + ", ";
    result += ctx.callActivity("SayHello", "London", String.class).await();
    return result;
}

При планировании функции активности платформа устойчивых задач сохраняет состояние выполнения функции в устойчивое серверное хранилище, например, хранилище таблиц Azure по умолчанию. Это состояние известно как история оркестрации.

Таблица журнала

Как правило, платформа устойчивых задач выполняет следующие действия на каждой контрольной точке:

1. Сохраняет журнал выполнения в устойчивом хранилище.
2. Помещает сообщения в очередь для функций, которые оркестратор хочет вызвать.
3. Ставит в очередь сообщения для самого оркестратора — например, долговечные таймерные сообщения.

После завершения контрольной точки функцию оркестратора можно удалить из памяти до тех пор, пока для неё не появится новая работа.

По завершении журнал функции, показанной ранее, выглядит примерно так, как в службе хранилища таблиц Azure (сокращено для иллюстрации):

Несколько заметок о значениях столбцов:

• PartitionKey: содержит идентификатор экземпляра оркестрации.
• EventType: представляет тип события. Подробные описания всех типов событий истории можно найти здесь.
• Метка времени: время UTC события журнала.
• Имя: имя вызываемой функции.
• Входные данные: входные данные в формате JSON функции.
• Результат: выходные данные функции; т. е. его возвращаемое значение.

Каждый раз, когда функция возобновляется после ожидания завершения задачи, платформа устойчивых задач повторно запускает функцию оркестратора с нуля. При каждом повторном запуске он обращается к журналу выполнения, чтобы определить, завершена ли текущая асинхронная задача. Если журнал выполнения показывает, что задача уже завершена, платформа воспроизводит выходные данные этой задачи и переходит к следующей задаче. Этот процесс продолжается до тех пор, пока не будет воспроизведена вся история выполнения. После воспроизведения текущего журнала выполнения локальные переменные будут восстановлены до предыдущих значений.

Функции и шаблоны

В следующих разделах описываются функции и шаблоны функций оркестратора.

Подоркестрации

Функции оркестратора могут вызывать функции активности, а также другие функции оркестратора. Например, можно создать большую оркестрацию из библиотеки функций оркестратора. Кроме того, можно параллельно запускать несколько экземпляров функции оркестратора.

Дополнительные сведения и примеры см. в статье о подоркестрациях.

Устойчивые таймеры

Оркестрации могут запланировать устойчивые таймеры для реализации задержек или настройки обработки времени ожидания для асинхронных действий. Используйте устойчивые таймеры в функциях оркестратора вместо встроенных в язык программирования функций sleep.

Дополнительные сведения и примеры см. в статье о устойчивых таймерах .

Внешние события

Функции оркестратора могут ожидать внешних событий для обновления экземпляра оркестрации. Эта функция устойчивых функций часто полезна для обработки взаимодействия с человеком или других внешних обратных вызовов.

Дополнительные сведения и примеры см. в статье о внешних событиях .

Обработка ошибок

Функции оркестратора могут использовать функции обработки ошибок языка программирования. Существующие шаблоны, например try/catch , поддерживаются в коде оркестрации.

Функции оркестратора также могут добавлять политики повторных попыток в функции активности или в подчиненные функции оркестрации, которые они вызывают. Если действие или функция подоркестратора завершается ошибкой с исключением, указанная политика повторных попыток может автоматически отложить выполнение и повторить его до указанного количества раз.

Дополнительные сведения и примеры см. в статье об обработке ошибок .

Критические разделы (устойчивые функции 2.x, на данный момент только для .NET)

Экземпляры оркестрации являются однопоточными, поэтому не нужно беспокоиться о условиях гонки в оркестрации. Однако условия гонки возможны, когда оркестрации взаимодействуют с внешними системами. Чтобы устранить условия гонки при взаимодействии с внешними системами, функции оркестратора могут определять критические разделы с помощью LockAsync метода в .NET.

В следующем примере кода показана функция оркестратора, определяющая критически важный раздел. Он вводит критически важный раздел с помощью LockAsync метода. Этот метод требует передачи одной или нескольких ссылок на устойчивую сущность, которая может управлять состоянием блокировки. Только один экземпляр этого оркестрации может выполнять код в критическом разделе одновременно.

[FunctionName("Synchronize")]
public static async Task Synchronize(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var lockId = new EntityId("LockEntity", "MyLockIdentifier");
    using (await context.LockAsync(lockId))
    {
        // critical section - only one orchestration can enter at a time
    }
}

Приобретает LockAsync устойчивые блокировки и возвращает IDisposable, которая завершает критическую секцию при освобождении. Этот IDisposable результат можно использовать вместе с блоком using для получения синтаксического представления критического раздела. Когда функция оркестратора входит в критически важный раздел, только один экземпляр может выполнять этот блок кода. Любые другие экземпляры, которые пытаются войти в критическую секцию, будут заблокированы до тех пор, пока предыдущий экземпляр не выйдет из критической секции.

Функция критической секции также полезна для координации изменений долговечных объектов. Дополнительные сведения о критических разделах см. в разделе "Устойчивые сущности: координация сущностей".

Вызов конечных точек HTTP (Durable Functions 2.x)

Функции оркестратора не разрешены для выполнения операций ввода-вывода, как описано в ограничениях кода функции оркестратора. Обычное решение этого ограничения заключается в том, чтобы упаковать любой код, который должен выполнять операции ввода-вывода в функцию действия. Оркестрации, взаимодействующие с внешними системами, часто используют функции активности для выполнения HTTP-вызовов и возврата результата в оркестрацию.

[FunctionName("CheckSiteAvailable")]
public static async Task CheckSiteAvailable(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    Uri url = context.GetInput<Uri>();

    // Makes an HTTP GET request to the specified endpoint
    DurableHttpResponse response = 
        await context.CallHttpAsync(HttpMethod.Get, url);

    if ((int)response.StatusCode == 400)
    {
        // handling of error codes goes here
    }
}

[Function("CheckSiteAvailable")]
public static async Task CheckSiteAvailable(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    Uri url = context.GetInput<Uri>();

    // Makes an HTTP GET request to the specified endpoint
    DurableHttpResponse response = await context.CallHttpAsync(
        method: HttpMethod.Get,
        uri: url,
        content: null,
        retryOptions: null);

    if ((int)response.StatusCode == 400)
    {
        // handling of error codes goes here
    }
}

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

module.exports = df.orchestrator(function*(context) {
    const url = context.df.getInput();
    var res = yield context.df.callHttp("GET", url);
    if (res.statusCode >= 400) {
        // handling of error codes goes here
    }
});

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

df.app.orchestration("checkSiteAvailable", function* (context) {
    const url = context.df.getInput();
    var res = yield context.df.callHttp({ method: "GET", url });
    if (res.statusCode >= 400) {
        // handling of error codes goes here
    }
});

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

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.orchestration_trigger(context_name="context")
def orchestrator_function(context: df.DurableOrchestrationContext):
    url = context.get_input()
    res = yield context.call_http('GET', url)
    if res['statusCode'] >= 400:
        # Handle error codes.

Помимо поддержки базовых шаблонов запросов и ответов, метод поддерживает автоматическую обработку распространенных шаблонов опроса HTTP 202, а также поддерживает проверку подлинности с помощью внешних служб с помощью управляемых удостоверений.

Дополнительные сведения и подробные примеры см. в статье о функциях HTTP .

Передача нескольких параметров

Невозможно передать несколько параметров в функцию действия напрямую. Рекомендация заключается в передаче массива объектов или составных объектов.

[FunctionName("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string major = "ComputerScience";
    int universityYear = context.GetInput<int>();

    object courseRecommendations = await context.CallActivityAsync<object>(
        "CourseRecommendations",
        (major, universityYear));
    return courseRecommendations;
}

[FunctionName("CourseRecommendations")]
public static async Task<object> Mapper([ActivityTrigger] IDurableActivityContext inputs)
{
    // parse input for student's major and year in university
    (string Major, int UniversityYear) studentInfo = inputs.GetInput<(string, int)>();

    // retrieve and return course recommendations by major and university year
    return new
    {
        major = studentInfo.Major,
        universityYear = studentInfo.UniversityYear,
        recommendedCourses = new []
        {
            "Introduction to .NET Programming",
            "Introduction to Linux",
            "Becoming an Entrepreneur"
        }
    };
}

public record CourseInfo(string Major, int UniversityYear);

[Function("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] TaskOrchestrationContext context, int universityYear)
{
    CourseInfo courseInfo = new("ComputerScience", universityYear);
    object courseRecommendations = await context.CallActivityAsync<object>(
        "CourseRecommendations", courseInfo);
    return courseRecommendations;
}

[Function("CourseRecommendations")]
public static async Task<CourseInfo> Mapper(
    [ActivityTrigger] CourseInfo studentInfo, FunctionContext executionContext)
{
    // Retrieve and return course recommendations by major and university year.
    return new
    {
        major = studentInfo.Major,
        universityYear = studentInfo.UniversityYear,
        recommendedCourses = new[]
        {
            "Introduction to .NET Programming",
            "Introduction to Linux",
            "Becoming an Entrepreneur"
        }
    };
}

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

module.exports = df.orchestrator(function*(context) {
    const location = {
        city: "Seattle",
        state: "WA"
    };
    const weather = yield context.df.callActivity("GetWeather", location);

    // ...
});

module.exports = async function (context, location) {
    const {city, state} = location; // destructure properties into variables

    // ...
};

const getWeatherActivityName = "getWeather";

df.app.orchestration("getWeatherOrchestrator", function* (context) {
    const location = {
        city: "Seattle",
        state: "WA",
    };
    const weather = yield context.df.callActivity(getWeatherActivityName, location);

    // ...
});

df.app.activity(getWeatherActivityName, {
    handler: async (input) => {
        const { city, state } = input; // Destructure properties into variables.

        // ...
    }
});

from collections import namedtuple
import azure.functions as func
import azure.durable_functions as df

# Create a FunctionApp instance.
app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# Orchestrator
@app.orchestration_trigger(context_name="context")
def orchestrator_function(context: df.DurableOrchestrationContext):
    Location = namedtuple('Location', ['city', 'state'])
    location = Location(city='Seattle', state= 'WA')

    weather = yield context.call_activity("GetWeather", location)
    # ...

# Activity
@app.activity_trigger(input_name="location")
def GetWeather(location):
    city, state = location
    return f"Hello {city}, {state}!"

param($Context)

$output = @()

$location = @{
    City = 'Seattle'
    State  = 'WA'
}

Invoke-ActivityFunction -FunctionName 'GetWeather' -Input $location

# ...

param($location)

"Hello $($location.City), $($location.State)!"
# ...

@FunctionName("GetWeatherOrchestrator")
public String getWeatherOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    var location = new Location();
    location.city = "Seattle";
    location.state = "WA";
    String weather = ctx.callActivity("GetWeather", location, String.class).await();
    return weather;
}

@FunctionName("GetWeather")
public String getWeather(@DurableActivityTrigger(name = "location") Location location) {
    if (location.city.equals("Seattle") && location.state.equals("WA")) {
        return "Cloudy";
    } else {
        return "Unknown";
    }
}

class Location {
    public String city;
    public String state;
}

Дальнейшие шаги