Escenario de supervisión en Durable Functions: ejemplo de supervisión meteorológica

Artículo
06/01/2023

El patrón de supervisión hace referencia a un proceso periódico flexible de un flujo de trabajo; por ejemplo, realizar un sondeo hasta que se cumplan determinadas condiciones. En este artículo se explica un ejemplo que usa Durable Functions para implementar la supervisión.

Información general de escenario

En este ejemplo se supervisan las condiciones meteorológicas actuales de una ubicación y alerta al usuario por SMS cuando el cielo está despejado. Puede utilizar una función regular desencadenada por temporizador para comprobar el tiempo y enviar alertas. Sin embargo, un problema con este enfoque es la administración del ciclo de vida. Si solo se debe enviar una alerta, la supervisión necesita deshabilitarse después de que se detecte un clima despejado. El patrón de supervisión puede finalizar su propia ejecución, entre otras ventajas:

Las supervisiones se ejecutan en intervalos, no en programaciones: un desencadenador en temporizador se ejecuta cada hora; una supervisión espera una hora entre acciones. Las acciones de una supervisión no se solaparán a menos que se especifique, lo que puede ser importante para tareas de ejecución prolongada.
Las supervisiones pueden tener intervalos dinámicos: el tiempo de espera puede cambiar en función de alguna condición.
Las supervisiones pueden finalizar cuando se cumple alguna condición o cuando otro proceso las finaliza.
Las supervisiones pueden aceptar parámetros. El ejemplo muestra cómo se puede aplicar el mismo proceso de supervisión meteorológica a cualquier ubicación y número de teléfono solicitados.
Las supervisiones son escalables. Debido a que cada supervisión es una instancia de orquestación, se pueden crear múltiples supervisiones sin tener que crear nuevas funciones ni definir más código.
Las supervisiones se integran fácilmente en flujos de trabajo mayores. Una supervisión puede ser una sección de una función de orquestación más compleja, o una suborquestación.

Configuración

Configuración de la integración de Twilio

En este ejemplo se usa el servicio Twilio para enviar mensajes SMS a un teléfono móvil. Azure Functions ya compatible con Twilio a través del enlace de Twilio, el ejemplo utiliza esa característica.

Lo primero que necesita es una cuenta de Twilio. Puede crear una gratis en https://www.twilio.com/try-twilio. Cuando tenga una cuenta, agregue los tres valores de configuración de la aplicación siguientes a la aplicación de función.

Nombre del valor de configuración de la aplicación	Descripción del valor
TwilioAccountSid	Identificador de seguridad de la cuenta de Twilio
TwilioAuthToken	Token de autenticación de la cuenta de Twilio
TwilioPhoneNumber	Número de teléfono asociado a la cuenta de Twilio, que se utiliza para enviar mensajes SMS.

Configuración de la integración de Weather Underground

Este ejemplo implica el uso de la API de Weather Underground para comprobar las condiciones meteorológicas actuales de una ubicación.

Lo primero que necesita es una cuenta de Weather Underground. Puede crear una gratis en https://www.wunderground.com/signup. Cuando tenga la cuenta, tendrá que adquirir una clave de API. Puede hacerlo si visita https://www.wunderground.com/weather/api y, después, si selecciona la configuración de la clave. El plan para desarrolladores Stratus es gratis y suficiente para ejecutar este ejemplo.

Cuando tenga una clave de API, agregue la siguiente configuración de la aplicación a la aplicación de función.

Nombre del valor de configuración de la aplicación	Descripción del valor
WeatherUndergroundApiKey	La clave de API de Weather Underground.

Funciones

En este artículo se explican las funciones siguientes en la aplicación de ejemplo:

E3_Monitor: una función de orquestador que llama periódicamente a E3_GetIsClear. Llama a E3_SendGoodWeatherAlert si E3_GetIsClear devuelve true.
E3_GetIsClear: una función de la actividad que comprueba las condiciones meteorológicas actuales de una ubicación.
E3_SendGoodWeatherAlert: una función de la actividad que envía un mensaje SMS por Twilio.

Función de orquestador E3_Monitor

C#
JavaScript

[FunctionName("E3_Monitor")]
public static async Task Run([OrchestrationTrigger] IDurableOrchestrationContext monitorContext, ILogger log)
{
    MonitorRequest input = monitorContext.GetInput<MonitorRequest>();
    if (!monitorContext.IsReplaying) { log.LogInformation($"Received monitor request. Location: {input?.Location}. Phone: {input?.Phone}."); }

    VerifyRequest(input);

    DateTime endTime = monitorContext.CurrentUtcDateTime.AddHours(6);
    if (!monitorContext.IsReplaying) { log.LogInformation($"Instantiating monitor for {input.Location}. Expires: {endTime}."); }

    while (monitorContext.CurrentUtcDateTime < endTime)
    {
        // Check the weather
        if (!monitorContext.IsReplaying) { log.LogInformation($"Checking current weather conditions for {input.Location} at {monitorContext.CurrentUtcDateTime}."); }

        bool isClear = await monitorContext.CallActivityAsync<bool>("E3_GetIsClear", input.Location);

        if (isClear)
        {
            // It's not raining! Or snowing. Or misting. Tell our user to take advantage of it.
            if (!monitorContext.IsReplaying) { log.LogInformation($"Detected clear weather for {input.Location}. Notifying {input.Phone}."); }

            await monitorContext.CallActivityAsync("E3_SendGoodWeatherAlert", input.Phone);
            break;
        }
        else
        {
            // Wait for the next checkpoint
            var nextCheckpoint = monitorContext.CurrentUtcDateTime.AddMinutes(30);
            if (!monitorContext.IsReplaying) { log.LogInformation($"Next check for {input.Location} at {nextCheckpoint}."); }

            await monitorContext.CreateTimer(nextCheckpoint, CancellationToken.None);
        }
    }

    log.LogInformation($"Monitor expiring.");
}

[Deterministic]
private static void VerifyRequest(MonitorRequest request)
{
    if (request == null)
    {
        throw new ArgumentNullException(nameof(request), "An input object is required.");
    }

    if (request.Location == null)
    {
        throw new ArgumentNullException(nameof(request.Location), "A location input is required.");
    }

    if (string.IsNullOrEmpty(request.Phone))
    {
        throw new ArgumentNullException(nameof(request.Phone), "A phone number input is required.");
    }
}

El orquestador requiere una ubicación para supervisar y un número de teléfono al que enviar un mensaje cuando el tiempo aparece despejado en la ubicación. Estos datos se pasan al orquestador como un objeto MonitorRequest con un tipo inflexible.

La función E3_Monitor utiliza la norma function.json para las funciones de orquestador.

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

Este es el código que implementa la función:

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

module.exports = df.orchestrator(function* (context) {
    const input = context.df.getInput();
    context.log(
        "Received monitor request. location: " +
            (input ? input.location : undefined) +
            ". phone: " +
            (input ? input.phone : undefined) +
            "."
    );

    verifyRequest(input);

    const endTime = moment.utc(context.df.currentUtcDateTime).add(6, "h");
    context.log(
        "Instantiating monitor for " +
            input.location.city +
            ", " +
            input.location.state +
            ". Expires: " +
            endTime +
            "."
    );

    while (moment.utc(context.df.currentUtcDateTime).isBefore(endTime)) {
        // Check the weather
        context.log(
            "Checking current weather conditions for " +
                input.location.city +
                ", " +
                input.location.state +
                " at " +
                context.df.currentUtcDateTime +
                "."
        );
        const isClear = yield context.df.callActivity("E3_GetIsClear", input.location);

        if (isClear) {
            // It's not raining! Or snowing. Or misting. Tell our user to take advantage of it.
            context.log(
                "Detected clear weather for " +
                    input.location.city +
                    ", " +
                    input.location.state +
                    ". Notifying " +
                    input.phone +
                    "."
            );

            yield context.df.callActivity("E3_SendGoodWeatherAlert", input.phone);
            break;
        } else {
            // Wait for the next checkpoint
            const nextCheckpoint = moment.utc(context.df.currentUtcDateTime).add(30, "s");
            context.log(
                "Next check for " +
                    input.location.city +
                    ", " +
                    input.location.state +
                    " at " +
                    nextCheckpoint.toString()
            );

            yield context.df.createTimer(nextCheckpoint.toDate()); // accomodate cancellation tokens
        }
    }

    context.log("Monitor expiring.");
});

function verifyRequest(request) {
    if (!request) {
        throw new Error("An input object is required.");
    }
    if (!request.location) {
        throw new Error("A location input is required.");
    }
    if (!request.phone) {
        throw new Error("A phone number input is required.");
    }
}

Esta función de orquestador realiza las acciones siguientes:

Obtiene la función MonitorRequest que consta de la ubicación para supervisar y el número de teléfono al que se enviará una notificación por SMS.
Determina el tiempo de expiración de la supervisión. El ejemplo utiliza un valor modificable por brevedad.
Llama a E3_GetIsClear para determinar si el cielo está despejado en la ubicación solicitada.
Si está despejado, llama a E3_SendGoodWeatherAlert para enviar una notificación por SMS al número de teléfono solicitado.
Crea un temporizador duradero para reanudar la orquestación en el siguiente intervalo de sondeo. El ejemplo utiliza un valor modificable por brevedad.
Continúa ejecutándose hasta que la hora universal coordinada actual pasa la hora de expiración de la supervisión o se envía una alerta por SMS.

Se pueden ejecutar varias instancias de orquestador de forma simultanea llamando varias veces a la función de orquestador. Se puede especificar la ubicación que se va a supervisar, así como el número de teléfono para enviar una alerta por SMS. Por último, tenga en cuenta que la función de orquestador no se ejecuta mientras se espera el temporizador, por lo que no se aplica ningún cargo por ello.

Función de actividad E3_GetIsClear

Al igual que otros ejemplos, las funciones auxiliares de actividad son básicamente funciones normales que usan el enlace del desencadenador activityTrigger. La función E3_GetIsClear obtiene las condiciones meteorológicas actuales utilizando la API de Weather Underground y determina si el cielo está despejado.

C#
JavaScript

[FunctionName("E3_GetIsClear")]
public static async Task<bool> GetIsClear([ActivityTrigger] Location location)
{
    var currentConditions = await WeatherUnderground.GetCurrentConditionsAsync(location);
    return currentConditions.Equals(WeatherCondition.Clear);
}

function.json se define como sigue:

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

Y esta es la implementación.

const request = require("request");

const clearWeatherConditions = [
    "Overcast",
    "Clear",
    "Partly Cloudy",
    "Mostly Cloudy",
    "Scattered Clouds",
];

module.exports = function (context, location) {
    getCurrentConditions(location)
        .then(function (data) {
            const isClear = clearWeatherConditions.includes(data.weather);
            context.done(null, isClear);
        })
        .catch(function (err) {
            context.log(`E3_GetIsClear encountered an error: ${err}`);
            context.done(err);
        });
};

function getCurrentConditions(location) {
    return new Promise(function (resolve, reject) {
        const options = {
            url: `https://api.wunderground.com/api/${process.env["WeatherUndergroundApiKey"]}/conditions/q/${location.state}/${location.city}.json`,
            method: "GET",
            json: true,
        };
        request(options, function (err, res, body) {
            if (err) {
                reject(err);
            }
            if (body.error) {
                reject(body.error);
            }
            if (body.response.error) {
                reject(body.response.error);
            }
            resolve(body.current_observation);
        });
    });
}

Función de actividad E3_SendGoodWeatherAlert

La función E3_SendGoodWeatherAlert usa el enlace de Twilio para enviar un mensaje SMS que notifica al usuario final que es un buen momento para dar un paseo.

C#
JavaScript

    [FunctionName("E3_SendGoodWeatherAlert")]
    public static void SendGoodWeatherAlert(
        [ActivityTrigger] string phoneNumber,
        ILogger log,
        [TwilioSms(AccountSidSetting = "TwilioAccountSid", AuthTokenSetting = "TwilioAuthToken", From = "%TwilioPhoneNumber%")]
            out CreateMessageOptions message)
    {
        message = new CreateMessageOptions(new PhoneNumber(phoneNumber));
        message.Body = $"The weather's clear outside! Go take a walk!";
    }

internal class WeatherUnderground
{
    private static readonly HttpClient httpClient = new HttpClient();
    private static IReadOnlyDictionary<string, WeatherCondition> weatherMapping = new Dictionary<string, WeatherCondition>()
    {
        { "Clear", WeatherCondition.Clear },
        { "Overcast", WeatherCondition.Clear },
        { "Cloudy", WeatherCondition.Clear },
        { "Clouds", WeatherCondition.Clear },
        { "Drizzle", WeatherCondition.Precipitation },
        { "Hail", WeatherCondition.Precipitation },
        { "Ice", WeatherCondition.Precipitation },
        { "Mist", WeatherCondition.Precipitation },
        { "Precipitation", WeatherCondition.Precipitation },
        { "Rain", WeatherCondition.Precipitation },
        { "Showers", WeatherCondition.Precipitation },
        { "Snow", WeatherCondition.Precipitation },
        { "Spray", WeatherCondition.Precipitation },
        { "Squall", WeatherCondition.Precipitation },
        { "Thunderstorm", WeatherCondition.Precipitation },
    };

    internal static async Task<WeatherCondition> GetCurrentConditionsAsync(Location location)
    {
        var apiKey = Environment.GetEnvironmentVariable("WeatherUndergroundApiKey");
        if (string.IsNullOrEmpty(apiKey))
        {
            throw new InvalidOperationException("The WeatherUndergroundApiKey environment variable was not set.");
        }

        var callString = string.Format("http://api.wunderground.com/api/{0}/conditions/q/{1}/{2}.json", apiKey, location.State, location.City);
        var response = await httpClient.GetAsync(callString);
        var conditions = await response.Content.ReadAsAsync<JObject>();

        JToken currentObservation;
        if (!conditions.TryGetValue("current_observation", out currentObservation))
        {
            JToken error = conditions.SelectToken("response.error");

            if (error != null)
            {
                throw new InvalidOperationException($"API returned an error: {error}.");
            }
            else
            {
                throw new ArgumentException("Could not find weather for this location. Try being more specific.");
            }
        }

        return MapToWeatherCondition((string)(currentObservation as JObject).GetValue("weather"));
    }

    private static WeatherCondition MapToWeatherCondition(string weather)
    {
        foreach (var pair in weatherMapping)
        {
            if (weather.Contains(pair.Key))
            {
                return pair.Value;
            }
        }

        return WeatherCondition.Other;
    }
}

Nota

Tendrá que instalar el paquete NuGet Microsoft.Azure.WebJobs.Extensions.Twilio para ejecutar el código de ejemplo.

La function.json es simple:

{
    "bindings": [
      {
        "name": "phoneNumber",
        "type": "activityTrigger",
        "direction": "in"
      },
      {
        "type": "twilioSms",
        "name": "message",
        "from": "%TwilioPhoneNumber%",
        "accountSidSetting": "TwilioAccountSid",
        "authTokenSetting": "TwilioAuthToken",
        "direction": "out"
      }
    ],
    "disabled": false
  }

Y este es el código que envía el mensaje SMS:

module.exports = function (context, phoneNumber) {
    context.bindings.message = {
        body: `The weather's clear outside! Go take a walk!`,
        to: phoneNumber,
    };

    context.done();
};

Ejecución del ejemplo

Con las funciones desencadenadas mediante HTTP del ejemplo, puede iniciar la orquestación con el envío de la siguiente solicitud HTTP POST:

POST https://{host}/orchestrators/E3_Monitor
Content-Length: 77
Content-Type: application/json

{ "location": { "city": "Redmond", "state": "WA" }, "phone": "+1425XXXXXXX" }

HTTP/1.1 202 Accepted
Content-Type: application/json; charset=utf-8
Location: https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635?taskHub=SampleHubVS&connection=Storage&code={SystemKey}
RetryAfter: 10

{"id": "f6893f25acf64df2ab53a35c09d52635", "statusQueryGetUri": "https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635?taskHub=SampleHubVS&connection=Storage&code={systemKey}", "sendEventPostUri": "https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635/raiseEvent/{eventName}?taskHub=SampleHubVS&connection=Storage&code={systemKey}", "terminatePostUri": "https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635/terminate?reason={text}&taskHub=SampleHubVS&connection=Storage&code={systemKey}"}

Se inicia la instancia de E3_Monitor y consulta las condiciones meteorológicas actuales para la ubicación solicitada. Si el tiempo está despejado, llama a una función de actividad para enviar una alerta; de lo contrario, establece un temporizador. Cuando expire el temporizador, se reanudará la orquestación.

Puede ver los resultados de la actividad de orquestación al examinar los registros de la función en el portal de Azure Functions.

2018-03-01T01:14:41.649 Function started (Id=2d5fcadf-275b-4226-a174-f9f943c90cd1)
2018-03-01T01:14:42.741 Started orchestration with ID = '1608200bb2ce4b7face5fc3b8e674f2e'.
2018-03-01T01:14:42.780 Function completed (Success, Id=2d5fcadf-275b-4226-a174-f9f943c90cd1, Duration=1111ms)
2018-03-01T01:14:52.765 Function started (Id=b1b7eb4a-96d3-4f11-a0ff-893e08dd4cfb)
2018-03-01T01:14:52.890 Received monitor request. Location: Redmond, WA. Phone: +1425XXXXXXX.
2018-03-01T01:14:52.895 Instantiating monitor for Redmond, WA. Expires: 3/1/2018 7:14:52 AM.
2018-03-01T01:14:52.909 Checking current weather conditions for Redmond, WA at 3/1/2018 1:14:52 AM.
2018-03-01T01:14:52.954 Function completed (Success, Id=b1b7eb4a-96d3-4f11-a0ff-893e08dd4cfb, Duration=189ms)
2018-03-01T01:14:53.226 Function started (Id=80a4cb26-c4be-46ba-85c8-ea0c6d07d859)
2018-03-01T01:14:53.808 Function completed (Success, Id=80a4cb26-c4be-46ba-85c8-ea0c6d07d859, Duration=582ms)
2018-03-01T01:14:53.967 Function started (Id=561d0c78-ee6e-46cb-b6db-39ef639c9a2c)
2018-03-01T01:14:53.996 Next check for Redmond, WA at 3/1/2018 1:44:53 AM.
2018-03-01T01:14:54.030 Function completed (Success, Id=561d0c78-ee6e-46cb-b6db-39ef639c9a2c, Duration=62ms)

La orquestación se completa cuando se alcanza el tiempo de espera o se detectan cielos despejados. También puede usar la API terminate dentro de otra función o invocar el webhook HTTP POST terminatePostUri al que se hace referencia en la respuesta 202 anterior. Para usar el webhook, reemplace {text} por el motivo de la terminación temprana. La dirección URL de HTTP POST tendrá un aspecto similar al siguiente:

POST https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635/terminate?reason=Because&taskHub=SampleHubVS&connection=Storage&code={systemKey}

Pasos siguientes

Este ejemplo ha demostrado cómo usar Durable Functions para supervisar el estado de un origen externo mediante temporizadores durables y lógica condicional. En el ejemplo siguiente se muestra cómo usar eventos externos y temporizadores duraderos para controlar la interacción humana.

Ejecución de la interacción humana de ejemplo