¿Step Functions sin ASL? Bienvenidas las Lambda Durable Functions

Este artículo fue traducido usando IA.

Durante re:Invent, AWS anunció una nueva funcionalidad dentro de AWS Lambda llamada durable functions. Son las mismas funciones Lambda que todos amamos, pero te permiten ejecutar flujos de trabajo de múltiples pasos manteniendo checkpoints y estado. ¿Qué significa esto? Puedes ejecutar funcionalidad similar a lo que típicamente hemos usado AWS Step Functions. Pero en lugar de usar Amazon State Language, puedes usar código y dependencias familiares.

Términos y Conceptos

Repasemos algunos conceptos sobre las durable functions.

• Durable Function - Es una función Lambda regular que puede pausarse y reanudarse haciendo uso de un mecanismo de checkpoint y replay.
• Checkpoint - Cuando la función ejecuta un paso o necesita esperar un callback, agregará un checkpoint que persiste el estado actual y detiene su ejecución.
• Replay - Una vez que el flujo de trabajo está listo para reanudar la ejecución, retomará desde el último checkpoint en lugar de ejecutar todo el flujo de trabajo de nuevo.
• Durable Execution - El ciclo de vida completo de una durable function. Desde el momento en que se activa hasta que completa todos los pasos definidos y se cierra.
• Durable Context - El contexto que se proporciona al handler de Lambda. Este contiene los métodos para las operaciones durables.
• Durable Operations - Estas son las operaciones que nos permiten crear un flujo de trabajo dentro de una función Lambda. Cada paso tiene reintentos integrados y automáticamente crea checkpoints cuando se ejecuta. Repasemos cada una de las operaciones a un nivel alto.
  • Steps - Ejecutan lógica de negocio y se definen usando el operador context.step().
  • Wait States - Pausas planificadas que causan que la función deje de ejecutarse hasta que se reanuden. Esto puede usarse para esperar una cantidad específica de tiempo, un callback externo u otras condiciones específicas. Se definen usando context.wait(), context.waitForCallback() o context.waitForCondition().
  • Parallel - Habrá situaciones donde quieras optimizar por velocidad y ejecutar cosas concurrentemente para reducir el tiempo de ejecución. La operación parallel se usa para eso, y se define usando el operador context.parallel().
  • Iterations - Para procesar un arreglo de elementos en un loop, tendrás que usar la operación context.map().
  • Invoke other Lambda functions - Puedes invocar funciones Lambda externas desde dentro de un flujo de trabajo. Para hacer esto, llamarás al operador context.invoke().
  • Nested Operations - Para ejecutar operaciones como hijo de otra operación, llamarás context.runInChildContext().

Cómo crear una Lambda durable function

Esto es probablemente lo mejor de todo esto. La mayor parte de la configuración se hace de la misma manera que una función Lambda regular, ¡PORQUE ES EL MISMO RECURSO! Para habilitar un contexto durable para la función, necesitamos hacer un cambio en nuestra IaC y otro en el handler del código. Repasemos estos.

Configuración de Infraestructura

Para configurarlo en SAM, todo lo que necesitas hacer es establecer el objeto DurableConfig.

  ExampleFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: index.handler
      CodeUri: workflows/example/
      DurableConfig:
        ExecutionTimeout: 10
        RetentionPeriodInDays: 1

Repasemos estas dos propiedades en más detalle:

• ExecutionTimeout - Este es el tiempo total, en segundos, para una ejecución durable completa. Esto puede ser hasta un año completo.
• RetentionPeriodInDays - El número de días que el historial de ejecución durable se retiene después de que se cierra.

Configuración del Handler

Para hacerlo más simple para nosotros, AWS ha proporcionado un nuevo SDK llamado @aws/durable-execution-sdk-js. Este contiene todo lo que necesitarás para construir flujos de trabajo en una función Lambda. Pero ahora mismo, lo único que necesitamos para tener un contexto durable es envolver nuestro handler con la función withDurableExecution. ¡Sí, eso es todo!

import { withDurableExecution } from '@aws/durable-execution-sdk-js';

export const handler = withDurableExecution(async (event, context) => {
  // Tu código del handler
})

Con esas cosas en su lugar, ahora podemos definir el flujo de trabajo.

Operaciones Durables

Ahora quiero repasar las diferentes operaciones durables en detalle y mostrarlas en un ejemplo funcional. Si quieres seguir el tutorial, tengo todo definido y desplegable en esta ubicación de GitHub.

Step

En su nivel más básico, tenemos el step. Esto creará un checkpoint después de ejecutarse, y puedes definir cualquier lógica que quieras dentro del step. Recomiendo separar la lógica de negocio real fuera de la definición del flujo de trabajo para mantenerlo limpio y legible. Esto facilita seguirlo cuando hay problemas con tu flujo de trabajo, y puedes encontrar el step que falló.

  // Step 1: Operación step inicial - procesar datos de entrada
  const workItems = await context.step('processInputData', async () => {
    return processData(event.inputData);
  });

Humano en el loop

Cuando quieras esperar una acción humana, usarás la operación waitForCallback.

  // Step 2: Operación waitForCallback - pausar para evento externo
  // Esperar callback externo con manejo de timeout
  const callbackResult = await context.waitForCallback(
    "wait-for-external-callback",
    async (callbackId, ctx) => {
      // Enviar callback ID al sistema externo (simulado)
      ctx.logger.info(`Callback ID ${callbackId} enviado al sistema externo`);
      // En una implementación real, esto llamaría una API externa
      // await submitToExternalAPI(callbackId);
    },
    { timeout: { minutes: 60 } } // timeout de 1 hora
  );

Esto nos dará un callbackId que podemos usar para reanudar nuestro flujo de trabajo. Al probar, puedes reanudar el flujo de trabajo desde la Consola de AWS, pero en la vida real, necesitarás hacer esto programáticamente. Afortunadamente, el SDK proporciona el SendDurableExecutionCallbackSuccessCommand y SendDurableExecutionCallbackFailureCommand en el cliente de Lambda, que pueden usarse para reanudar el flujo de trabajo. También puedes especificar un timeout para que el flujo de trabajo no espere para siempre.

Una vez que se envía el comando, el flujo de trabajo se reanudará y continuará procesando.

Wait

Si alguna vez necesitas esperar una cantidad específica de tiempo, puedes usar la operación wait, que te permite especificar la cantidad de tiempo que quieres esperar.

  // Step 3: Operación wait simple - demostrar espera basada en tiempo
  await context.wait({ seconds: 5 }); // Esperar 5 segundos

Procesamiento Paralelo

Cuando tienes varias operaciones que pueden manejarse independientemente, podrías querer ejecutarlas en paralelo para reducir el tiempo total de procesamiento. Ahí es donde entra la operación parallel.

  // Step 4: Operaciones paralelas - procesar múltiples flujos de trabajo concurrentemente
  const parallelResults = await context.parallel([
    async (ctx) => ctx.step('parallelTask1', async () => {
      return await performDataValidation(workItemsCount);
    }),
    async (ctx) => ctx.step('parallelTask2', async () => {
      return await performDataEnrichment(workItemsCount);
    }),
    async (ctx) => ctx.step('parallelTask3', async () => {
      return await performQualityCheck();
    })
  ]);

Esto intentará ejecutar las tres operaciones en paralelo. Si te preocupa poner demasiada carga en otro servicio, puedes establecer límites de concurrency para que no todo se procese a la vez.

Iterar Arreglos

Siempre digo que la programación se trata de ifs y fors. Entonces, ¿cómo manejas la situación del for al trabajar con durable functions? ¡Tenemos el operador map!

  // Step 5: Operación map - iterar sobre una colección con checkpoints
  const mapResults = await context.map(workItems, async (ctx, item, index) => {
    return await ctx.step(`processItem-${index}`, async () => {
      const { processedItem, processingTime } = processWorkItem(item, index);

      // Simular tiempo de procesamiento basado en prioridad
      await new Promise(resolve => setTimeout(resolve, processingTime));

      return processedItem;
    });
  });

Podrías estarte preguntando, ¿Por qué no simplemente usar un loop for regular? La respuesta es CHECKPOINTS. Al usar la operación map, obtienes el beneficio de que Lambda maneje el checkpoint y el estado actual de dónde quedaron las cosas. Esto ayuda inmensamente si no procesó el arreglo completamente.

Wait for condition

También hay una operación wait especial que espera a que se cumpla una condición específica. Para esto, usarás la operación waitForCondition. Puedes pensar en esta operación como un do/while, donde seguirá iterando hasta que se cumpla la condición del while.

// Step 6: Wait for condition - consultar hasta que el sistema externo esté listo
  const conditionResult = await context.waitForCondition(
    async (state, ctx) => {
      const readinessCheck = await checkSystemReadiness();
      return {
        ...state,
        ready: readinessCheck.ready
      };
    },
    {
      initialState: {
        ready: false,
      },
      waitStrategy: (state) =>
        state.ready
          ? { shouldContinue: false }
          : { shouldContinue: true, delay: { seconds: 3 } }
    }
  );

Esto tiene dos parámetros:

1. WaitForConditionCheckFunc - Función que verificará el estado actual y retornará el estado actualizado.
2. WaitForConditionConfig - Contiene la configuración para el estado inicial y la estrategia de espera usada para determinar cuándo debe continuar y cuánto tiempo esperar antes de activar WaitForConditionCheckFunc de nuevo.

Invocar Función Lambda

Por más que trates de evitarlo, tendrás una razón para invocar una función Lambda separada que hace su propio procesamiento. Esto se hará usando la operación invoke. Todo lo que necesitas hacer es darle el ARN de la función y el payload esperado.

  // Step 7: Invocar otra función Lambda
  const invokePayload = createInvokePayload(workItems.length, context.executionId);
  const invokeResult = await context.invoke(
    'invoke-hello-world',
    process.env.HELLO_WORLD_FUNCTION_ARN,
    invokePayload
  );

Contextos hijo

La operación runInChildContext crea un contexto de ejecución aislado para un grupo de operaciones. Estos tienen su propio registro de checkpoints y pueden tener múltiples steps, waits y otras operaciones. Esto se trata como una unidad única para reintentos y recuperación. Usa estos cuando quieras organizar flujos de trabajo complejos, implementar sub-flujos de trabajo o aislar operaciones que deberían reintentarse juntas.

  // Step 8: Ejecutar operaciones en contexto hijo para aislamiento
  const childContextResult = await context.runInChildContext('isolated-operations', async (childCtx) => {
    // Estas operaciones se ejecutan en aislamiento con su propio registro de checkpoints
    const metadata = await childCtx.step('processMetadata', async () => {
      return await processMetadataInChild(context.executionId, workItems.length);
    });

    const validation = await childCtx.step('validateConfiguration', async () => {
      return await validateConfigurationInChild();
    });

    return {
      metadata,
      validation,
      childExecutionId: childCtx.executionId,
      completedAt: Date.now()
    };
  });

Invocación y Resolución de Problemas

Una vez que tu durable function está desplegada, querrás invocar la función. Bueno, sorpresa, sorpresa, esto se hace exactamente de la misma forma que invocas cualquier función Lambda, lo que significa que puedes usar todas tus fuentes de eventos regulares como API Gateway, EventBridge y SQS. Todo lo que estoy a punto de hacer puede hacerse usando el CLI, pero por razones de demo, usaré la consola para que podamos obtener mejores visualizaciones.

Cuando la Lambda se despliega con el DurableConfig verás una nueva pestaña de Durable executions en la consola de Lambda.

Para crear una nueva ejecución, invoqué la función usando la herramienta de testing estándar en la consola de Lambda.

Ahora, veamos los detalles de la ejecución.

Esta vista debería verse muy similar a lo que hemos visto en AWS Step Functions. No tiene el bonito diagrama de flujo de trabajo, sin embargo. Hay dos secciones aquí:

1. Durable operations - aquí veremos todas las operaciones que se han ejecutado. En nuestro ejemplo anterior, solo aparecen dos elementos porque tiene el paso de humano-en-el-loop y está esperando que alguien responda con el callback ID. Haremos eso en un segundo.
2. Event history - una lista más detallada de lo que ha sucedido. Aquí es donde puedes rastrear errores que ocurrieron en un paso específico y obtener una visión más amplia de todo lo que se ejecutó.

Ahora, déjame enviar el mensaje de éxito para el callback. Haré esto directamente desde la consola, como se muestra a continuación. Si buscas una forma de hacer esto programáticamente, he incluido un script aquí. Este script tomará el callback ID y enviará un mensaje de éxito para continuar el flujo.

Esto ahora continuará y ejecutará todos los pasos que hemos definido en nuestro flujo de trabajo.

Notarás que algunas operaciones tienen nombres explícitos y otras tienen IDs aleatorios. La razón es que no especifiqué explícitamente un nombre para todas las operaciones, así que generará uno aleatorio por nosotros.

Repasemos las operaciones.

• WaitForCallback - La duración fue 9 minutos y 43 segundos. Ese es el tiempo que tomó escribir algunos de los párrafos anteriores, y luego presionar el botón “Send success”.
• Wait - Como se esperaba, esto tomó los 5 segundos que habíamos configurado.
• Parallel - Esta operación agrupa todas las sub-operaciones bajo ella, haciéndolas más fáciles de rastrear.

• Map - Similar al step parallel, agrupará todas las iteraciones bajo ella.

• WaitForCondition - El énfasis especial aquí es que tiene un 2 en Retries. Esto significa que tuvo que iterar dos veces antes de que se cumpliera la condición.
• ChainedInvoke - No mucho que ver para este step. Simplemente invocó la función Lambda y continuó.
• RunInChildContext - Este también agrupa todas las operaciones hijo bajo él.

Esto te permite ver las diferentes operaciones y cómo se ven en la consola. Puedes obtener esta información usando el CLI o el SDK, y crear tus propias visualizaciones si es necesario.

Conclusión

¡Woah! ¡Eso fue mucho! Repasamos la configuración para habilitar y desplegar una durable function, así como cómo construir un flujo de trabajo que usa todas las operaciones durables actualmente soportadas.

El hecho de que ahora puedo crear flujos de trabajo de larga duración donde puedo incluir cualquier dependencia npm directamente sin tener que llamar explícitamente funciones Lambda externas es una gran victoria comparada con lo que hemos estado haciendo con Step Functions.

Voy a seguir experimentando con esto y haré una comparación lado a lado con AWS Step Functions. ¡Manténganse atentos!

¡Hasta la próxima!

Andres Moreno