Orquestrações Duráveis

As Durable Functions são uma extensão do Azure Functions. É possível usar uma função funções de orquestrador para orquestrar a execução de outras Durable Functions em um aplicativo de funções. As funções de orquestrador têm as seguintes características:

• As funções de orquestrador definem fluxos de trabalho de função que usam o código do procedimento. Não são necessários esquemas declarativos ou designers.
• As funções de orquestrador podem chamar outras Durable Functions de maneira síncrona e assíncrona. A saída das funções chamadas pode ser salva em variáveis locais com confiança.
• As funções de orquestrador são duráveis e confiáveis. O progresso da execução é marcado automaticamente com um ponto de verificação quando a função “aguarda” ou “gera”. O estado local nunca é perdido quando o processo é reciclado ou a VM é reiniciada.
• As funções de orquestrador podem ser de longa duração. O ciclo de vida total de uma instância de orquestração pode ser de segundos, dias, meses ou interminável.

Este artigo oferece uma visão geral das funções de orquestrador e como elas podem ajudar a resolver vários desafios de desenvolvimento do aplicativo. Se você ainda não está familiarizado com os tipos de funções disponíveis em um aplicativo Durable Functions, leia primeiro o artigo sobre tipos de Funções Duráveis.

Identidade de orquestração

Cada instância de uma orquestração tem um identificador de instância (também conhecido como ID de uma instância). Por padrão, cada ID de instância é um GUID gerado automaticamente. No entanto, as IDs de instância também podem ser qualquer valor de cadeia de caracteres gerado pelo usuário. Cada ID de instância de orquestração deve ser exclusiva dentro de um hub de tarefas.

A seguir, há algumas regras sobre IDs de instância:

• As IDs de instância devem ter entre 1 e 100 caracteres.
• As IDs de instância não devem começar com @.
• As IDs de instância não devem conter /, \, # ou ? caracteres.
• As IDs de instância não devem conter caracteres de controle.

A ID da instância de uma orquestração é um parâmetro obrigatório para a maioria das operações de gerenciamento de instância. Elas também são importantes para diagnóstico, como pesquisa por meio de dados de acompanhamento de orquestração no Application Insights para fins de solução de problemas ou de análise. Por esse motivo, é recomendável salvar as IDs de instância geradas em alguma localização externa (por exemplo, um banco de dados ou em logs de aplicativo), em que elas podem ser facilmente consultadas mais tarde.

Confiabilidade

As funções de orquestrador mantêm com confiança seu estado de execução usando o padrão de design de fornecimento de eventos. Em vez de armazenar diretamente o estado atual de uma orquestração, a Estrutura de Tarefas Duráveis usa um armazenamento somente de acréscimo para registrar a série completa de ações executadas pela orquestração de função. Um armazenamento somente de acréscimo tem muitos benefícios em comparação com o "despejo" do estado de runtime completo. Os benefícios incluem maior desempenho, escalabilidade e capacidade de resposta. Você também obtém consistência eventual para dados transacionais e trilhas e histórico de auditoria completos. As trilhas de auditoria dão suporte a ações de compensação confiáveis.

As Durable Functions usam o fornecimento de eventos de forma transparente. Nos bastidores, o operador await (C#) ou yield (JavaScript/Python) de uma função de orquestrador leva o controle do thread do orquestrador de volta para o dispatcher da Estrutura de Tarefas Duráveis. No caso do Java, não existe palavra-chave de linguagem especial. Em vez disso, chamar .await() em uma tarefa suspenderá o controle de volta ao dispatcher por meio de um Throwable personalizado. O dispatcher, em seguida, confirma novas ações agendadas pela função orquestradora (como chamar uma ou mais funções filho ou agendar um temporizador durável) no armazenamento. A ação de confirmação transparente atualiza o histórico de execuções da instância de orquestração acrescentando todos os novos eventos ao armazenamento, assim como um log somente de acréscimo. De maneira semelhante, a ação de confirmação cria mensagens no armazenamento para agendar o trabalho real. Neste ponto, a função orquestradora pode ser descarregada da memória. Por padrão, o Durable Functions usa o Armazenamento do Azure como o armazenamento de estado do runtime, mas outros provedores de armazenamento também têm suporte.

Quando uma função de orquestração recebe mais trabalho a fazer (por exemplo, uma mensagem de resposta é recebida ou um temporizador durável expira), o orquestrador é ativado e executa mais uma vez a função inteira, desde o início, para recompilar o estado local. Durante a reprodução, se o código tentar chamar uma função (ou executar qualquer outro trabalho assíncrono), a Estrutura de Tarefas Duráveis consultará o histórico de execução da orquestração atual. Se ele detectar que a função de atividade já foi executada e gerou um resultado, ele reproduzirá o resultado da função e o código do orquestrador continuará em execução. A reprodução continua até que o código da função seja concluído ou até que tenha agendado um novo trabalho assíncrono.

Histórico de orquestração

O comportamento de fornecimento de eventos da Estrutura de Tarefas Duráveis está intimamente acoplado ao código da função funções de orquestrador que você escreve. Suponha que você tenha uma função de orquestrador de encadeamento de atividades, como a seguinte função de orquestrador:

[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, "Cairo"));

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

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

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]

main = df.Orchestrator.create(orchestrator_function)

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

Sempre que uma atividade de função é agendada, o Durable Task Framework verifica o estado da execução da função em algum back-end de armazenamento durável (por padrão, no Armazenamento de Tabelas do Azure). Esse estado é o que chamamos de histórico de orquestração.

Tabela de histórico

De modo geral, o Framework de Tarefa Durável faz o seguinte em cada ponto de verificação:

1. Salva o histórico de execuções no armazenamento durável.
2. Enfileira mensagens para as funções que o orquestrador quer invocar.
3. Enfileira mensagens para o próprio orquestrador, por exemplo, mensagens de temporizador durável.

Quando o ponto de verificação for concluído, a função de orquestrador estará livre para ser removida da memória até que tenha mais trabalho a fazer.

Após a conclusão, o histórico da função mostrado anteriormente se parece com a seguinte tabela no Armazenamento de Tabelas do Azure (abreviado para fins de ilustração):

Algumas observações sobre os valores das colunas:

• PartitionKey: Contém a ID de instância da orquestração.
• EventType: Representa o tipo do evento. Você pode encontrar descrições detalhadas de todos os tipos de eventos de histórico aqui.
• Timestamp: O carimbo de data/hora, em UTC, do evento do histórico.
• Name: O nome da função que foi invocada.
• Entrada: A entrada da função formatada em JSON.
• Result: A saída da função, ou seja, o valor retornado.

Sempre que a função é retomada depois de aguardar a conclusão de uma tarefa, o Durable Task Framework executa novamente a função de orquestrador do zero. A cada nova execução, ele consulta o histórico de execução para determinar se a tarefa assíncrona atual foi concluída. Se o histórico de execução mostrar que a tarefa já foi concluída, a estrutura reproduzirá a saída dessa tarefa e passará para a próxima tarefa. Esse processo continua até que todo o histórico de execução tenha sido reproduzido. Depois que o histórico de execução atual tiver sido reproduzido, as variáveis locais terão sido restauradas para seus valores anteriores.

Recursos e padrões

As seções a seguir descrevem os recursos e padrões de funções de orquestrador.

Subtipo de orquestrações

As funções de orquestrador podem chamar funções de atividade, mas também outras funções de orquestrador. Por exemplo, você pode criar uma orquestração maior usando uma biblioteca de funções de orquestrador. Ou pode executar várias instâncias de uma função de orquestrador em paralelo.

Para obter mais informações e exemplos, confira o artigo Suborquestrações.

Temporizadores duráveis

As orquestrações podem agendar temporizadores duráveis para implementar atrasos ou configurar a manipulação de tempo limite em ações assíncronas. Use temporizadores duráveis em funções de orquestrador em vez de APIs "suspender" nativas da linguagem.

Para obter mais informações e exemplos, confira o artigo Temporizadores duráveis.

Eventos externos

As funções de orquestrador podem aguardar os eventos externos atualizarem uma instância de orquestração. O recurso Durable Functions geralmente é útil para manipular uma interação humana ou outros retornos de chamada externos.

Para obter mais informações e exemplos, confira o artigo Eventos externos.

Tratamento de erros

As funções de orquestrador podem usar os recursos de tratamento de erro da linguagem de programação. No código de orquestração, há suporte para padrões existentes, como try/catch.

As funções de orquestrador também podem adicionar políticas de repetição à atividade ou às funções de suborquestrador que elas chamam. Se uma atividade ou função de suborquestrador falhar com uma exceção, a política de repetição especificada poderá atrasar automaticamente e repetir a execução até um número especificado de vezes.

Para obter mais informações e exemplos, confira o artigo Tratamento de erro.

Seções críticas (Durable Functions 2.x, atualmente somente no .NET)

As instâncias de orquestração têm thread único para que não seja preciso se preocupar com condições de corrida dentro de uma orquestração. No entanto, as condição de corrida são possíveis quando as orquestrações interagem com sistemas externos. Para atenuar as condições de corrida ao interagir com sistemas externos, as funções de orquestrador podem definir seções críticas usando um método LockAsync no .NET.

O exemplo de código a seguir mostra uma função funções de orquestrador que define uma seção crítica. Ele entra na seção crítica usando o método LockAsync. Esse método requer a passagem de uma ou mais referências a uma Entidade Durável, que gerencia de maneira durável o estado de bloqueio. Somente uma única instância dessa orquestração pode executar o código na seção crítica por vez.

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

O LockAsync adquire os bloqueios duráveis e retorna um IDisposable que encerra a seção crítica quando descartado. Esse resultado IDisposable pode ser usado junto com um bloco using para obter uma representação sintática da seção crítica. Quando uma função funções de orquestrador entra em uma seção crítica, apenas uma instância pode executar esse bloco de código. Outras instâncias que tentarem entrar na seção crítica serão bloqueadas até que a instância anterior saia da seção crítica.

O recurso de seção crítica também é útil para coordenar alterações em entidades duráveis. Para obter informações sobre seções críticas, confira o tópico “Coordenação de entidade” de entidades duráveis.

Chamar pontos de extremidade HTTP (Durable Functions 2.x)

As funções de orquestrador não têm permissão para realizar E/S conforme descrito nas restrições de código da função funções de orquestrador. A alternativa típica para essa limitação é encapsular o código que precisa realizar E/S em uma função de atividade. As orquestrações que interagem com sistemas externos frequentemente usam funções de atividade para realizar chamadas HTTP e retornar o resultado à orquestração.

[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(HttpMethod.Get, url);

    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

def orchestrator_function(context: df.DurableOrchestrationContext):
    url = context.get_input()
    res = yield context.call_http('GET', url)
    if res.status_code >= 400:
        # handing of error code goes here

Além de dar suporte a padrões básicos de solicitação/resposta, o método dá suporte ao tratamento automático de padrões de sondagem HTTP 202 assíncronos comuns e à autenticação com serviços externos usando identidades gerenciadas.

Para obter mais informações e exemplos detalhados, confira o artigo Recursos HTTP.

Passando vários parâmetros

Não é possível passar vários parâmetros para uma função de atividade diretamente. A recomendação é transmitir uma matriz de objetos ou objetos de composição.

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

[Function("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] TaskOrchestrationContext context, int universityYear)
{
    string major = "ComputerScience";

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

[FunctionName("CourseRecommendations")]
public static async Task<object> Mapper(
    [ActivityTrigger] (string Major, int UniversityYear) 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, async function (location) {
    const { city, state } = location; // destructure properties into variables

    // ...
});

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

def orchestrator_function(context: df.DurableOrchestrationContext):
    Location = namedtuple('Location', ['city', 'state'])
    location = Location(city='Seattle', state= 'WA')

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

    # ...

from collections import namedtuple

Location = namedtuple('Location', ['city', 'state'])

def main(location: Location) -> str:
    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;
}

Próximas etapas