Couchbase Eventing Handling Errors and Retries w/ Examples

En Servicio de eventos Couchbase le permite actuar rápidamente en caso de mutaciones (o cambios) en sus datos. Todas las acciones en Eventos se llevan a cabo mediante la ejecución de una lambda, un pequeño fragmento de lógica de negocio escrito en JavaScript.

Entre los casos de uso más comunes se incluyen el enriquecimiento de datos, el archivo de documentos y la integración con servicios REST externos. Más información aquí.

En el siguiente blog, describiremos cómo se pueden manejar los errores durante la ejecución del listador de eventos. Mediante el uso de un mecanismo de reintento nos aseguramos de que la acción prevista se lleva a cabo, incluso si el receptor de eventos falla durante la ejecución.

Ejemplo de solicitud

Como ejemplo, implementamos parte de una aplicación de comercio electrónico que almacena los pedidos de los clientes en una colección de Couchbase. En cuanto el estado de un pedido cambia a pagadoQueremos enviar una confirmación de pedido al cliente.

Para lograrlo, integramos un receptor de eventos de Couchbase con una aplicación externa servicio de correo electrónico. El receptor de eventos de Couchbase recogerá cualquier cambio en el documento del pedido, verificará que el pedido se ha pagado y, a continuación, llamará al servicio de correo electrónico para activar el mensaje de confirmación.

El servicio de correo electrónico es un microservicio independiente que proporciona un punto final REST. Usamos el soporte cURL integrado directamente en Couchbase Eventing para llamar al microservicio desde el receptor de eventos.

Flujo conceptual: A medida que los pedidos se actualizan en Couchbase, los eventos se disparan y son recogidos por un listener de eventos. A continuación, el receptor de eventos llama al servicio de correo electrónico externo.

El servicio de correo electrónico devuelve un error

El escenario descrito funciona muy bien si el servicio de correo electrónico está operativo. Sin embargo, ¿qué ocurre si el servicio de correo electrónico devuelve un error? Las peticiones del escuchador de eventos al servicio de correo electrónico fallarán y, por tanto, no se enviará ningún mensaje de confirmación al cliente. Dado que en el momento del fallo el evento de cambio de documento de Couchbase ya ha sido procesado, no se dispara ningún nuevo evento para el mismo documento a menos que haya otro cambio en él. Para asegurar que la confirmación es enviada necesitamos manejar el error e implementar un mecanismo de reintento. Haciendo esto podemos solucionar cualquier problema temporal del servicio externo y al mismo tiempo garantizar que la confirmación es enviada.

Hay diferentes maneras de enfocar esto. En mi ejemplo de abajo elijo crear una nueva colección llamada 'retry', que almacenará referencias a los documentos para los que falló la ejecución del listener de eventos.

El receptor de eventos recogerá los cambios en los documentos de pedido (paso #1) y luego llamará al servicio de correo electrónico (paso #2). Si la llamada al servicio de correo electrónico tiene éxito, el receptor de eventos actualiza el estado del mensaje de confirmación en el documento de pedido (paso #3). Sin embargo, en caso de fallo, se crea un documento de reintento y se coloca en la colección "reintento" (paso #3*).

Mantener una referencia a los documentos nos permite identificar todas las actualizaciones fallidas y nos permite volver a ejecutarlas más tarde. Esto podría ser mediante la intervención manual de un operador o el reintento automático utilizando temporizadores de eventos de Couchbase.

Iniciamos el proceso de reintento añadiendo un documento con un id de documento especificado a la colección de reintentos. Se crea un temporizador recurrente con un intervalo proporcionado.
En la ejecución del temporizador, todos los documentos más antiguos que un pequeño cuanto de tiempo en el reintentar se actualizan. Añadiendo un atributo como fireRetry = true a los documentos de reintento disparamos otro evento de actualización que es recogido por el escuchador de eventos para ejecutar el mecanismo de reintento. Esto nos da una mutación recursiva que enciende todos los documentos de la colección retry en paralelo. La función de reintento se ejecuta ahora utilizando todos los hilos de trabajo disponibles en paralelo.
Se activa un evento de actualización de documento para cada documento de reintento individualmente.
El documento de pedido correspondiente se recupera de la colección de entrada
Ahora se llama al servicio de correo electrónico.
Si la llamada al servicio de correo electrónico tiene éxito, el receptor de eventos actualiza el estado del mensaje de confirmación en el documento de pedido y elimina el documento de reintento.
En caso de fallo, el documento de reintento se actualiza y se coloca en la colección "reintento".

Revisión de códigos

Ahora que ya hemos establecido el diseño conceptual, echemos un vistazo al ejemplo de implementación:

Requisitos previos:

Couchbase 7 Enterprise Edition. Ejecuto Couchbase como un clúster de nodo único en Docker en mi máquina local. (https://docs.couchbase.com/server/current/install/getting-started-docker.html)
Para propósitos de desarrollo creamos un Cluster Couchbase de un solo nodo ejecutando los siguientes servicios:
- Índices, consultas, eventos y servicios de datos

Tenga en cuenta que no se recomienda la instalación en un único nodo para su uso en producción.

Preparación

Crear un cubo llamado pedidos
Cree dos colecciones en pedidos cubos _ámbito por defecto:
- entrada (contendrá todos los pedidos entrantes)
- reintentar (contendrá los documentos de reintento referidos a las órdenes que han fallado)

Crear cubo metadatos. Utilizaremos el ámbito _default y la colección _default. El bucket de metadatos se utiliza para los metadatos de Eventing.

Crear un índice en reintentar colección. El retry listener consultará cualquier documento contenido en la colección utilizando N1QL, por lo que es necesario que exista un índice para que se ejecute la consulta.

CREATE PRIMARY INDEX idx_default_primary ON orders._default.retry USING GSI;

1	CREAR PRIMARIO ÍNDICE idx_default_primary EN pedidos.Por defecto.reintentar USO DE GSI;

Documento de pedido de modelo de datos

Para esta aplicación de ejemplo, utilizamos un modelo de datos ligero para el documento de pedido, que sólo contiene los campos relevantes. Se omiten muchos otros campos que normalmente se esperarían en un documento de pedido.

{
  "email": "customer_email",
  "paymentStatus": "initiated",
  "confirmationEmailSent": false,
  "items": [
    {
      "name": "Swedish Meatballs 500g",
      "amount": 2,
      "unitPrice": 9.95
    }
  ]
}

{

"email": "correo_cliente",

"estado de pago": "iniciado",

"confirmationEmailSent": falso,

"artículos": [

{

"nombre": "Albóndigas suecas 500g",

"importe": 2,

"unitPrice": 9.95

}

]

}

Documento de reintento de modelo de datos

El documento de reintento contiene algunos atributos básicos, como el identificador del documento de pedido, un contador de intentos y una marca de tiempo. La dirección tipo no es necesario en nuestra aplicación, pero puede ser útil para determinar el tipo de notificación por correo electrónico en caso de que la aplicación se amplíe para enviar también actualizaciones de envío y entrega.

{
  "type": "confirmation",
  "docId": "order_140",
  "attempt": 1,
  "ts": 1632775908319
}

{

"tipo": "confirmación",

"docId": "pedido_140",

"intento": 1,

"ts": 1632775908319

}

Servicio de correo electrónico MOCK

Simularemos el Servicio de Correo Electrónico utilizando un simple script Python ejecutando un servidor web local. El script responderá aleatoriamente con HTTP 200 OK, o con HTTP 406 para indicar un fallo.

Actualice la dirección IP a la dirección IP de su máquina local en la línea 31servidor = ThreadedHTTPServer(('sustituir por su IP', 9080), Handler)
Inicie el script ejecutando: python http.py

Receptores de eventos

Ahora con todos los preparativos en su lugar podemos seguir adelante y añadir los dos oyentes de eventos:

evt_send_confirmation_email - proporciona la integración con el Servicio de Correo Electrónico
evt_send_confirmation_email_retry - contiene la lógica de reintento

Los oyentes están disponibles aquí: https://github.com/puhhma/cb_eventing_retry_sample
Importe los oyentes (json ) en el servicio Couchbase Eventing.

Ten en cuenta que, para que funcionen, debes seguir las convenciones de nomenclatura utilizadas en este artículo.

Consulte evt_send_confirmation_email oyente

Configuración del receptor de eventos:

El receptor de eventos está escuchando el evento entrada colección en el pedidos cubo.
En metadatos bucket se utiliza para almacenar los metadatos de los oyentes
Los alias de los cubos bkt_order_inbound y bkt_order_retry hacer referencia al entrada y reintentar colección en el pedir cubo
En curlServicioCorreoHost especifica el alias de URL para el mock EmailService. Por favor, asegúrese de actualizar con su dirección IP

// OnUpdate is invoked for all documents created/updated in the 'inbound' bucket
function OnUpdate(doc, meta) {
    // determine if document status is 'paid' & confirmation email was not previously sent
    if( doc.paymentStatus === "paid" && !doc.confirmationEmailSent ) {
        SendConfirmationMail(doc, meta.id);
    } else {
      if (debug_level > 1)
        log("Nothing to do for: " + meta.id);
    }
}

function SendConfirmationMail(doc, docId) {
  try {
      // build the request to the EmailService
      var request = {
          path: 'sendConfirmation',
          body: doc
      };
      //  perform the cURL request using the URL alias 'curlEmailServiceHost' from the settings
      var response = curl('POST', curlEmailServiceHost, request);
      if (response.status != 200) {
          // this did not work as expected
          if (debug_level > 1) {
              log("docId", docId, "cURL POST failed response.status:",response.status);
          }
          // create retry document referencing the documentId and store in 'retry' bucket
          bkt_order_retry[docId] = {
              "docId": docId,
              "attempt": 1,
              "ts": Date.now()
          }
      } else {
          if (debug_level > 5) {
              log("cURL POST success, sent",docId,"response.body:",response.body);
          }
          // update confirmationEmailSent status
          doc.confirmationEmailSent = true;
          bkt_order_inbound[docId] = doc;
      }
  } catch (e) {
      log("ERROR cURL request had an exception:",e)
  }
}

// OnUpdate se invoca para todos los documentos creados/actualizados en el bucket "de entrada

función OnUpdate(doc, meta) {

// determinar si el estado del documento es "pagado" y no se ha enviado previamente un correo electrónico de confirmación

si( doc.paymentStatus === "pagado" && !doc.confirmationEmailEnviado ) {

SendConfirmationMail(doc, meta.id);

} si no {

si (nivel_depuración > 1)

registro("Nada que hacer por: " + meta.id);

}

función SendConfirmationMail(doc, docId) {

pruebe {

// construir la petición al EmailService

var solicitar = {

ruta: enviarConfirmación,

cuerpo: doc

};

// realice la solicitud cURL utilizando el alias de URL "curlEmailServiceHost" de la configuración

var respuesta = rizo(POST, curlServicioCorreoHost, solicitar);

si (respuesta.estado != 200) {

// esto no funcionó como se esperaba

si (nivel_depuración > 1) {

registro("docId", docId, "cURL POST falló response.status:",respuesta.estado);

}

// crear documento de reintento con referencia al documentId y almacenarlo en el bucket 'retry

bkt_order_retry[docId] = {

"docId": docId,

"intento": 1,

"ts": Fecha.ahora()

}

} si no {

si (nivel_depuración > 5) {

registro("cURL POST exitoso, enviado",docId,"response.body:",respuesta.cuerpo);

}

// actualizar el estado de confirmationEmailSent

doc.confirmationEmailEnviado = verdadero;

bkt_order_inbound[docId] = doc;

}

} captura (e) {

registro("ERROR la petición cURL tuvo una excepción:",e)

}

Para más detalles, consulte los comentarios en línea
En OnUpdate se activa cuando se actualiza o se crea un documento de pedido
La solicitud se construye y la solicitud HTTP POST se envía al EmailService utilizando cURL.
Se evalúa el resultado. En caso de que la respuesta HTTP no sea satisfactoria, se construye un documento de reintento y se añade al archivo reintentar colección.

El receptor de eventos está escuchando el evento reintentar colección en el pedidos cubo.
En metadatos bucket se utiliza para almacenar los metadatos de los oyentes
Los alias de los cubos bkt_order_inbound y bkt_order_retry hacer referencia al entrada y reintentar en la cubeta de pedidos
En curlServicioCorreoHost especifica el alias de URL para el mock EmailService
En retryTimerIntervall especifica el intervalo del temporizador en segundos.

function OnUpdate(doc, meta) {
    if (meta.id === "allow_retrys") {
        // the timer is initialized by creating document with id = 'allow_retrys'
        CreateRetryTimer({"id": meta.id, "mode": "initial"});
    } else if (doc.fireRetry) {
      // process retry documents
      SendConfirmationMail(doc, meta.id);
    }
}

function CreateRetryTimer(context) {
    if (debug_level > 2) {
        log('From CreateRetryTimer: creating timer', context.mode, context.id);
    }
    // Create a timestamp 'retryTimerInterval' seconds (from the settings) from now
    var timerStartTime = new Date();
    // Get current time & add 'retryTimerInterval' sec. to it.
    timerStartTime.setSeconds(timerStartTime.getSeconds() + retryTimerInterval);
    // Create a document to use as out for our context
    createTimer(RetryTimerCallback, timerStartTime, context.id, context);
}

function RetryTimerCallback(context) {
    if (debug_level > 2) {
        log('From RetryTimerCallback: timer fired', context);
    }
    // rearm the timer ASAP, to ensure timer keeps running in the event
    // of later  errors or script timeouts in later "recurring work".
    CreateRetryTimer({ "id": context.id, "mode": "via_callback" });
    // Update all retry documents in the 'retry' bucket. Exclude the 'allow_retys' document
    // and any documents that were created more than 15 seconds ago, in order to avoid retry 'to early'.
    N1QL("UPDATE orders._default.retry SET fireRetry = true WHERE meta().id != 'allow_retrys' AND ts < DATE_ADD_MILLIS(NOW_MILLIS(), -15, 'second')");
}

function SendConfirmationMail(retryDoc, docId) {
  try {
      // resolve order document by id
      var doc = bkt_order_inbound[docId];
      // build the request
      var request = {
          path: 'sendConfirmation',
          body: doc
      };
      //  perform the cURL request using the URL alias from the settings
      var response = curl('POST', curlEmailServiceHost, request);
      if (response.status != 200) {
          // this did not work as expected
          if (debug_level > 1) {
              log("docId", docId, "cURL POST failed response.status:",response.status);
          }
        // increment attempt count in retry document
        retryDoc.attempt = ++retryDoc.attempt;
        // Set fireRetry = false, to avoid retry execution with this document change
        retryDoc.fireRetry = false;
        retryDoc.ts = Date.now();
        // update retry document
        bkt_order_retry[docId] = retryDoc;
      } else {
          if (debug_level > 5) {
              log("cURL POST success, sent",docId,"response.body:",response.body);
          }
          doc.confirmationEmailSent = true;
          bkt_order_inbound[docId] = doc;
          // delete the retry document
          delete bkt_order_retry[docId];
      }  } catch (e) {
      log("ERROR cURL request had an exception:",e)
  }
}

función OnUpdate(doc, meta) {

si (meta.id === "allow_retrys") {

// el temporizador se inicializa creando un documento con id = 'allow_retrys'

CreateRetryTimer({"id": meta.id, "mode": "inicial"});

} si no si (doc.fireRetry) {

// procesar documentos de reintento

SendConfirmationMail(doc, meta.id);

}

función CreateRetryTimer(contexto) {

si (nivel_depuración > 2) {

registro('De CreateRetryTimer: creando temporizador', contexto.modo, contexto.id);

}

// Crear un timestamp 'retryTimerInterval' segundos (de la configuración) a partir de ahora

var timerStartTime = nuevo Fecha();

// Obtén la hora actual y añádele un segundo de 'retryTimerInterval'.

timerStartTime.setSeconds(timerStartTime.getSeconds() + retryTimerInterval);

// Crear un documento para usar como out para nuestro contexto

crearTiempo(RetryTimerCallback, timerStartTime, contexto.id, contexto);

}

función RetryTimerCallback(contexto) {

si (nivel_depuración > 2) {

registro('De RetryTimerCallback: temporizador disparado', contexto);

}

// rearmar el temporizador lo antes posible, para asegurar que el temporizador sigue funcionando en el evento

// de errores posteriores o tiempos de espera del script en "trabajos recurrentes" posteriores.

CreateRetryTimer({ "id": contexto.id, "mode": "via_callback" });

// Actualiza todos los documentos de reintento en el bucket 'retry'. Excluir el documento 'allow_retys'.

// y cualquier documento que se haya creado hace más de 15 segundos, para evitar reintentos 'demasiado tempranos'.

N1QL("UPDATE orders._default.retry SET fireRetry = true WHERE meta().id != 'allow_retrys' AND ts < DATE_ADD_MILLIS(NOW_MILLIS(), -15, 'second')");

}

función SendConfirmationMail(retryDoc, docId) {

pruebe {

// resolver documento de pedido por id

var doc = bkt_order_inbound[docId];

// construir la solicitud

var solicitar = {

ruta: enviarConfirmación,

cuerpo: doc

};

// realizar la solicitud cURL utilizando el alias de URL de la configuración

var respuesta = rizo(POST, curlServicioCorreoHost, solicitar);

si (respuesta.estado != 200) {

// esto no funcionó como se esperaba

si (nivel_depuración > 1) {

registro("docId", docId, "cURL POST falló response.status:",respuesta.estado);

}

// incrementa el número de intentos en el documento de reintentos

retryDoc.intente = ++retryDoc.intente;

// Establecer fireRetry = false, para evitar reintentos de ejecución con este cambio de documento

retryDoc.fireRetry = falso;

retryDoc.ts = Fecha.ahora();

// actualizar documento de reintento

bkt_order_retry[docId] = retryDoc;

} si no {

si (nivel_depuración > 5) {

registro("cURL POST exitoso, enviado",docId,"response.body:",respuesta.cuerpo);

}

doc.confirmationEmailEnviado = verdadero;

bkt_order_inbound[docId] = doc;

// eliminar el documento de reintento

borrar bkt_order_retry[docId];

} } captura (e) {

registro("ERROR la petición cURL tuvo una excepción:",e)

}

El temporizador se inicia añadiendo un documento con el id allow_retrys a la reintentar colección
A continuación, el temporizador se inicializa y la función RetryTimerCallback asociada al temporizador
Una vez ejecutado el temporizador se llama a la función RetryTimerCallback
Antes de proceder con el mecanismo de reintento, se crea un nuevo temporizador como primer paso para garantizar que sigue funcionando en caso de errores posteriores.
Se utiliza una consulta N1QL para actualizar todos los reintentar documentos en el reintentar añadiendo un fireRetry al documento
Cada cambio de documento da lugar a un evento de actualización del documento y se ejecuta el mecanismo de reintento
El documento de pedido se resuelve a partir del entrada y el EmailService se llama a través de cURL
En caso de fallo, el reintentar y se actualiza el documento intente contador aumentado

Probar la aplicación de ejemplo

Ahora es el momento de probar finalmente la aplicación de ejemplo:

Asegúrese de que el mock EmailService está en funcionamiento
Inicie el evt_send_confirmation_email pero manteniendo el evt_send_confirmation_email_retry oyente se detuvo por ahora.
Cree un documento de pedido de muestra (véase el modelo de datos anterior) en la consola de Couchbase
En caso de respuesta positiva, el confirmationEmailEnviado se actualiza a true en el atributo pedir documento.
En caso de fallo, se crea un documento de reintento en la carpeta reintentar recogida. Dado que el EmailService responderá aleatoriamente con un error, repita el paso #3 hasta que se produzca un error.
Ahora que hemos capturado un error, vamos a iniciar el escuchador de eventos de reintento evt_send_confirmation_email_retry
Crea un documento con el id 'allow_retrys'. Esto inicializará el mecanismo de reintentos.
Al cabo de unos instantes, el oyente se activará y empezará a procesar los documentos de la base de datos. reintentar colección.
Por favor, observe que el atributo 'attempt' se incrementa con cada actualización fallida del Servicio de Correo Electrónico. En caso de éxito, el documento de pedido se actualiza y el atributo reintentar documento retirado del reintentar colección.

Dado que la respuesta del Email Service Mock es aleatoria, es posible que tenga que repetir los pasos anteriores para poder observar el comportamiento previsto.

Conclusión

En este artículo esbozo un mecanismo de reintento para manejar condiciones de error al integrar Couchbase Eventing con un servicio REST externo. Esta o soluciones similares se pueden utilizar para garantizar que las acciones previstas se lleven a cabo incluso si el servicio externo está temporalmente funcionando mal.

A la hora de considerar un mecanismo de reintentos hay que tener en cuenta varios factores, como el volumen de reintentos, los hilos de trabajo disponibles para Couchbase Eventing y las peticiones que el servicio externo puede gestionar.

Puedes encontrar más información sobre el funcionamiento interno de Couchbase Eventing aquí: https://docs.couchbase.com/server/current/eventing/eventing-overview.html

Muchas gracias a Jon Strabala (Principal Product Manager, Couchbase) por sus conocimientos técnicos y su ayuda con este artículo.

Marian Puhl, Ingeniera de soluciones

Platform

Self-Managed

Services

Capabilities

Why Couchbase?

Migrate to Capella

By Use Case

By Industry

By Application Need

Popular Docs

By Developer Role

Quickstart

Resource Center

About

Partnerships

Our Services

Partners: Register a Deal

Ready to register a deal with Couchbase?

Marriott

Todos los mensajes

Couchbase Eventing Manejo de Errores y Reintentos

Autor

Publicado por Marian Puhl, Ingeniera de soluciones

Dejar una respuesta Cancelar respuesta