Implementing a Scheduler Via Couchbase Eventing (Part 1)

Esta es la primera de una serie de varias partes para aprovechar la Servicio de eventos Couchbase para ejecutar múltiples tareas programadas a intervalos recurrentes específicos en un cron completamente dentro de la base de datos sin necesidad de infraestructura adicional a través de una única función de eventos de uso general.

En esta entrega, nos centraremos en la ejecución de rutinas fijas de usuario, funciones JavaScript definidas dentro de una función de eventos.

En artículos posteriores ampliaremos la cron como Eventing Function para programar y ejecutar sentencias N1QL dinámicas manejadas por la base de datos y finalmente exploraremos la programación de funciones JavaScript dinámicas manejadas por la base de datos.

Fondo

El servicio de eventos de Couchbase proporciona un marco para escribir tus propias rutinas, simples funciones JavaScript, para procesar los cambios en los documentos. Este servicio proporciona toda la infraestructura necesaria para crear funciones escalables y robustas basadas en la nube, permitiéndote centrarte en desarrollar pura lógica de negocio para interactuar casi en tiempo real con los cambios en tus datos. Tus funciones pueden acceder al servicio de datos de Couchbase (KV), al servicio de consultas de Couchbase (N1QL) y a puntos finales REST externos al sistema Couchbase.

En JSON modelo de datos de Couchbase procede de JavaScriptPor lo tanto, es natural que el servicio de eventos ofrezca la posibilidad de escribir código JavaScript para analizar y manipular documentos JSON en cualquier tipo de evento de cambio, incluyendo inserciones, actualizaciones, fusiones y eliminaciones (denominadas conjuntamente mutaciones).

Las funciones de eventos normalmente le permiten desplegar y ejecutar fragmentos de código personalizados para que reaccionen a miles e incluso millones de mutaciones por segundo en sus documentos. Algunos casos de uso típicos son documentado para desarrollar Funciones Eventing de alta velocidad a escala que respondan a mutaciones de documentos Couchbase.

Este artículo se centrará más bien en un caso de uso de muy baja velocidad del Servicio de Eventos construyendo un crontab distribuido fiable "en base de datos", permitiéndote ejecutar funciones JavaScript que interactúan con servicios Couchbase en un horario periódico regular.

Programación de la lógica empresarial para que se ejecute en una fecha u hora determinadas

Cronque lleva el nombre de "Chronos," la palabra griega para el tiempo es una de las utilidades más útiles en un sistema Linux. En Linux el cron está controlada por un archivo crontab (tabla cron), un archivo de configuración que especifica los comandos del shell que se ejecutarán periódicamente en un horario determinado.

Uno de los inconvenientes cron es que no está diseñado para ser un servicio distribuido; se ejecuta en una sola caja, por lo que presenta un único punto de fallo. Si el sistema está fuera de línea durante varias horas, se pierden todas las tareas programadas.

Sí, hay algunos distribuidos cron como Google's Cloud Service, AWS' Scheduled Tasks, y Azure Functions / Time Triggers. Pero las ofertas de cada proveedor de nube tienen sus propios modismos que no son directamente portables.

Además, es necesario asegurar la metodología de configuración y control, por ejemplo si se controla una cron a través de una API REST sobre HTTP/S, deberá tenerlo en cuenta en su plan de seguridad.

Utilizar el propio Couchbase para ejecutar comandos periódicos

Con una pequeña cantidad de código y planificación, puede aprovechar el servicio Eventing de Couchbase para proporcionar cron para las operaciones o el mantenimiento programados de la base de datos. La integración del programador en la base de datos permite obtener las siguientes ventajas:

Portabilidad a través de proveedores Cloud, si reubicas tu cluster de Couchbase tu planificador no se ve afectado.
Soporte, si utiliza Couchbase tiene un único proveedor para proporcionarle soporte y otros servicios.
Distribuido, no hay un único punto de fallo y todos los servicios de Couchbase admiten réplicas distribuidas.
Ejecución garantizada, su tarea se ejecuta incluso después de recuperarse de un fallo de nodo.

Programación de Couchbase, los temporizadores son la salsa secreta

Los temporizadores son construcciones del servicio de eventos de Couchbase mediante las cuales los desarrolladores pueden especificar una rutina (lógica de negocio) que se activará en un momento futuro. Utilizaremos esta funcionalidad para implementar un Couchbase configurable puro. crontab que le permite activar tareas repetitivas como parte de sus flujos de trabajo, tanto si necesita ejecutar una simple consulta N1QL como construir un complejo motor de reglas.

En todos los diseños posteriores limitaremos nuestra cron a una resolución de 15 segundos o más. Tenemos esta limitación porque, aunque los temporizadores se escalan a millones y se garantiza que se disparan y ejecutan, no son precisos como un reloj de pared y actualmente tienen un retardo de estado estacionario limitado a menos de 14 segundos. ^[1].

Por supuesto, si necesita una programación más ajustada, es decir, menos de 15 segundos, entonces simplemente debe procesar la mutación en sí en la lógica de eventos sin el uso de una construcción de temporizador para programar una llamada de vuelta en el futuro.

En el momento de escribir estas líneas, la versión actual de Couchbase es la 6.5.1 que dos limitaciones con las que debemos trabajar al hacer un robusto cron sistema.

En las versiones 5.5.x, 6.0.x y 6.5.x, una función invocada por una llamada de retorno de un temporizador no puede crear un nuevo temporizador de forma fiable (se puede solucionar mediante una segunda función cooperativa).
En las versiones 6.5.x, la creación de temporizadores en el futuro (como en una hora o más) en un sistema por lo demás inactivo puede dar lugar a un número creciente de operaciones de cubo de metadatos que eventualmente pueden bloquear las mutaciones para una función de Eventing dada (en 6.5.X se puede lograr una solución en el espacio de usuario a través de una segunda función cooperativa). La gravedad se rige por:
- El número de vBuckets que mantienen un temporizador activo. Por lo tanto, si solo hay unos pocos temporizadores en el futuro, es posible que el problema no se note o no se materialice. Este es el caso con sólo unos pocos cron horarios, pero para completar en caso de añadir la funcionalidad de la fecha puse en una solución para este problema para el código suministrado en este artículo.
- Si un temporizador de Eventing se ha disparado recientemente en un vBucket (lo que soluciona el problema para el vBucket en cuestión por función). Por lo tanto, los sistemas con mucha actividad de temporizadores a corto plazo no experimentarán este problema aunque los temporizadores estén programados para un futuro lejano.

Afortunadamente, en la versión 6.6.0 ambos problemas o restricciones se han eliminado y se puede crear un programador en una única y sencilla Función de Eventos unificada.

Requisitos previos

En este artículo utilizaremos la última versión de GA, es decir, la versión 6.5.1 de Couchbase (puede que tengas que hacer algunos cambios en las funciones de eventos descritas para versiones anteriores de Couchbase). El ejemplo de este artículo se ejecutará contra el conjunto de datos travel-sample que se entrega con el servidor Couchbase.

CONSEJO PROFESIONAL: Sólo para usuarios avanzados, si estás familiarizado con Couchbase Eventing y también con nuestras herramientas CLI / REST puedes saltarte la mayor parte de este blog y descargar un archivo ZIP para configurar y ejecutar rápidamente el sistema programador que se presenta a continuación. Haz clic con el botón derecho en el siguiente enlace y selecciona Guardar enlace como para descargar el archivo cron_impl_2func_CLI.zip, muévalo a un nodo Eventing, extraiga el archivo ZIP y consulte el archivo README.txt extraído.

Sin embargo, si usted no está familiarizado con Couchbase o el servicio Eventing por favor camine a través de GET STARTED y un ejemplo Eventing específicamente se refieren a los siguientes:

Configure un servidor Couchbase 6.5.1 que funcione según las instrucciones de Comience aquí
Asegúrese de que puede ejecutar una consulta N1QL en la base de datos viaje-muestra según las instrucciones de Ejecute su primera consulta N1QL.
Comprender cómo desplegar una función básica de Eventing según las instrucciones de la sección Archivo de documentos que también utiliza viaje-muestra conjunto de datos.
Asegúrese de tener la viaje-muestra en la vista de cubos de la interfaz de usuario.
Asegúrese de que un cubo llamado metadatos en la vista Buckets de la interfaz de usuario debe tener un tamaño mínimo de 200 MB.
En la vista Buckets de la interfaz de usuario, cree un bucket llamado crondata con un tamaño mínimo de 200 MB. Para obtener información detallada sobre cómo crear cubos, consulte Crear un cubo.
Establecer allow_interbucket_recursion a verdadero para permitir que dos (2) funciones de Eventing modifiquen el mismo documento KV ^[2].

Shell

curl -X POST -u "$CB_USERNAME:$CB_PASSWORD" 'https://localhost:8091/_p/event/api/v1/config' -d '{ "allow_interbucket_recursion":true }'

1

curl -X POST -u "$CB_USERNAME:$CB_PASSWORD" 'https://localhost:8091/_p/event/api/v1/config' -d '{ "allow_interbucket_recursion":true }'

Implementación #1, programación tipo "cron" codificada.

Para nuestra primera implementación, por ejemplo la Parte 1 de la serie, diseñaremos una estructura de control simple que es simplemente un documento KV JSON y también dos (2) Funciones Eventing que responderán y actuarán sobre la información en la estructura de control.

A continuación se muestra un diseño de un documento JSON, o estructura de control, que nos permitirá tener múltiples "eventos" programados. Cada evento programado tendrá su propio documento de control con una CLAVE única como recurring_event::1, recurring_event::1, ... recurring_event::N. La propia estructura JSON contiene información para "reconstituir la clave", ya que nuestro sistema de programación responderá a los cambios o actualizaciones (mutaciones) de los documentos de control, como alternar el estado "activo" para activar o desactivar la acción o cambiar el campo "verbose" que controla la cantidad y el estilo del registro.

A continuación se muestra un ejemplo de documento de control con CLAVE evento_recurrente::1 que ejecutará la función JavaScript doCronActionA a las 14:54 (14:30) todos los días.

Registro de control JSON	Descripción
{
"type": "recurring_event",	La CLAVE será <>::<>.
"id":1,
"hora":14,	La hora del día 0-23, , 2X, *4X para activar
"min":54,	El minuto en la hora 0-59, , 2X, *4X para activar
"acción": "doCronActionA",	Función JavaScript que se ejecutará cuando se active el temporizador
"activo":true,	Bandera para activar o desactivar este horario
"verbose": {	[OPCIONAL] control de registro
"user_func":2,	Nivel de registro para la lógica de acción : 0=ninguno, etc. etc.
"programador":3	Nivel de registro para la lógica cron : 0=ninguno, etc. etc.
},
"dinámico": {	control y estadísticas del sistema [DINÁMICO
"estado": "brazo",	"armar"\|"rearmar"\|"pendiente" cualquier valor != "pendiente" iniciar una programación
"next_sched": 0,	Número de segundos transcurridos desde la época hasta el siguiente horario deseado
"prev_sched": 0,	Número de segundos transcurridos desde la fecha de la programación anterior
"prev_etime": 0,	Número de segundos transcurridos desde la época para el tiempo de ejecución real del programa anterior
"prev_delay": 0,	Número de segundos que el temporizador se ha retrasado con respecto a la programación
"prev_atime": 0	Número de segundos que ha tardado el usuario 'acción'
}
}

Como Linux tradicional crontab puede establecer la hora y el minuto en números enteros legales, y también puede establecer hora a "*" para procesar todas las horas o ajuste min a "*" para procesar todos los minutos.

Aunque no apoyaremos la crontab sintaxis admitimos dos configuraciones no estándar, a saber, si fijar ambos hora y min a "*4X" ejecutaremos y rearmaremos cuatro (4) veces por minuto y si los configuras ambos a "*2X" ejecutaremos y rearmaremos dos (2) veces por minuto. A continuación se muestra una tabla de horarios soportados con su descripción:

hora	min	Los valores pueden ser números o cadenas
13	32	Corre a las 13:32 (o 13:32)
*	15	Se ejecuta cada hora a los 15 minutos
8	12	Corre una vez al día a las 8:32 (o 8:32 am)
*	*	Ejecutar una vez por minuto
*2X	*2X	Se ejecuta dos veces por minuto - requiere que tanto la hora como el minuto estén ajustados a "*2X".
*4X	*4X	Funciona cuatro veces por minuto - requiere que la hora y el minuto estén ajustados a "*2X".

Eventualmente utilizaremos el Query Workbench para insertar el cron documentos de control, todos los cuales deben tener una CLAVE única de evento_recurrente::# a una hora de ejecución programada de 14:54 ( 2:54 pm), para la acción doCronActionA, podríamos utilizar la siguiente sentencia N1QL.

No te preocupes por ejecutar ninguna sentencia N1QL en este momento, realizaremos las sentencias N1QL más tarde después de haber construido y desplegado nuestra Función de Eventos.

Puede crear un registro (o registros) de control en el cubo viaje-muestray luego listarlo, armarlo, desarmarlo, ajustar el horario que sigue, cambiar el nivel de verbosidad para el registro, o borrarlo de la siguiente manera:

Acción	Declaración N1QL
Crear un calendario	INSERTAR EN `viaje-muestra` (CLAVE,VALOR) VALORES ( "recurring_event::1", { "type": "recurring_event", "id":1, "hora": "14″, "min": "54″, "acción": "doCronActionA", "activo":true } );
Crear un índice para consultar datos sin especificar claves	CREAR un índice primario sobre `crondata` ;
Mostrar todos los horarios ordenados por id	SELECT * FROM `crondata` WHERE type="recurring_event" order by id ;
Mostrar horario específico	SELECT * FROM `crondata` WHERE type="recurring_event" AND id=1 ;
Armar o activar	ACTUALIZACIÓN `crondata` SET active = true WHERE type="recurring_event" AND id=1 ;
Desarmar o desactivar	ACTUALIZACIÓN `crondata` SET active = false WHERE type="recurring_event" AND id=1 ;
Ajustar el tiempo de activación	ACTUALIZACIÓN `crondata` SET hora = 11, min = 30 WHERE type="recurring_event" AND id=1 ;
Ajustar el registro de la "acción	ACTUALIZACIÓN `crondata` SET verbose.user_data = 0 WHERE type="recurring_event" AND id=1 ;
Ajustar el registro de la lógica del programador	ACTUALIZACIÓN `crondata` SET verbose.scheduler = 0 WHERE type="recurring_event" AND id=1 ;
Borrar el horario	DELETE FROM `crondata` WHERE type="recurring_event" AND id=1 ;

Supongamos que tenemos cuatro (4) horarios activos, la ejecución de la primera sentencia N1QL, por encima de la lista de todos ellos, por ejemplo.

SELECT active, action, hour, min, type, id, verbose.user_func, verbose.scheduler
FROM `crondata` where type="recurring_event" order by id ;

1 2	SELECT active, action, hour, min, type, id, verbose.user_func, verbose.scheduler FROM `crondata` where type="recurring_event" order by id ;

Devolvería algo como la siguiente salida (vista de tabla en el Query Workbench):

activo	acción	hora	id	min	programador	tipo	usuario_func
verdadero	"doCronActionA"	14	1	54	1	"evento_recurrente"	2
verdadero	"doCronActionB"	*	2	*	1	"evento_recurrente"	1
verdadero	"doCronActionC"	*2X	3	*2X	4	"evento_recurrente"	4
verdadero	"doCronActionD"	*	4	0	0	"evento_recurrente"	1

En la tabla anterior tenemos cuatro acciones: la primera se ejecuta una vez al día, la segunda cada minuto, la tercera cada 30 segundos y la cuarta cada hora. En una próxima entrega de esta serie añadiremos la función "día de la semana".

El objeto anidado "verbose"si no se suministra será por defecto { "user_func":1, "scheduler":1 } indicando un nivel de registro bajo o escueto para la función de acción y también para la lógica de programación. Un valor de 0 suprimirá todos los mensajes de registro, es decir, doCronActionD, mientras que valores mayores serán más verbosos, es decir, como se define en doCronActionC.

El objeto anidado "dinámico" si normalmente nunca se suministra y será por defecto { "state": "arm", "next_sched": 0, "prev_sched": 0, "prev_etime": 0, "prev_delay": 0, "prev_atime": 0 } este es un bloc de notas para la programación lógica de eventos en ejecución y también proporciona estadísticas útiles sobre los tiempos de ejecución, por lo que debe ser tratado como de sólo lectura.

En este punto tenemos un diseño de control de alto nivel, pero necesitamos lógica para procesar nuestras estructuras de control, aquí es donde el Servicio de Eventos de Couchbase, específicamente una Función de Eventos entra en juego.

Funciones del concurso

Este diseño requiere dos (2) funciones de Eventing: una función JavaScript principal "cron_impl_2func_651" y una pequeña función JavaScript de ayuda "cron_impl_2func_651_help". Discutiremos cada sección de las funciones JavaScript que comprende la implementación inicial combinada de código JavaScript de casi 610 líneas (con aproximadamente 44% de las líneas son comentarios y espacios en blanco)

No te preocupes por hacer un cortar y pegar en este momentoComo más tarde voy a proporcionar un enlace para descargar (para la importación) las dos funciones de eventos necesarios y todos los ajustes necesarios en dos archivos llamados "cron_impl_2func_651.json" "cron_impl_2func_651_help.json" y también si lo prefiere las dos funciones unificadas completo que se puede cortar y pegar directamente.

Nuestra función principal de eventos "cron_impl_2func_651" estará compuesta por nueve (9) funciones JavaScript

Tres (3) funciones de lógica empresarial (dos de las cuales están vacías).
- doCronActionA(doc) - una acción de usuario de ejemplo N1QL a ejecutar
- doCronActionB(doc) - un intérprete de comandos de acción de usuario vacío para experimentos
- doCronActionC(doc) - un intérprete de comandos de acción de usuario vacío para experimentos
Un (1) punto de entrada para concursos completos.
- OnUpdate(doc, meta) - el punto de entrada estándar de Eventing para Inserciones o Actualizaciones
Uno (1) cron analizador sintáctico para generar la siguiente programación.
- getNextRecurringDate(hour_str, min_str) - lógica cron para encontrar la siguiente Fecha programada
Tres (3) funciones de apoyo para comprobar que la lógica de negocio existe o formatear resultados.
- verifyFunctionExistsViaEval(curDoc, id) - asegurarse de que tenemos una función que ejecutar
- toNumericFixed(number, precision) - formatea un flotante a un estilo compacto
- toLocalISOTime(d) - formatea una fecha a un estilo compacto
Una (1) función callback cuando se ejecutan los temporizadores.
- Callback(doc) - una función de devolución de llamada para temporizadores programados

Nuestra función de ayuda para eventos "cron_impl_2func_651_help" estará compuesta por una (1) función JavaScript

Un (1) punto de entrada para concursos completos.
- OnUpdate(doc, meta) - el punto de entrada estándar de Eventing para Inserciones o Actualizaciones

En las secciones siguientes recorreremos cada una de las funciones JavaScript anteriores.

Necesitamos una función JavaScript, por ejemplo, la lógica de negocio para ejecutar en un horario periódico.

Lo primero que queremos es una rutina o función que contenga nuestra lógica de negocio que ejecutaremos basándonos en nuestras reglas crontab. Llamaremos al método JavaScript doCronActionA(doc)sin embargo, puede llamarse como se quiera, por ejemplo doPeriodicLedgerBalance(doc), los únicos requisitos para nuestras funciones de "acción" que implementan nuestra lógica de negocio programada son los siguientes:

Tiene un parámetro: doc, un documento de control como el descrito anteriormente de type="recurring_event".
El nombre real de JavaScript coincide con el campo "acción" del documento de control.
Devuelve verdadero sobre el éxito y falso sobre el fracaso
Utiliza doc.verbose.user_func para controlar el registro si 0 es silencioso, si 1 emite una sola línea, si 2 emite cualquier información de registro que se necesite para depurar la función, etc. etc..

Escribiremos nuestra función doCronActionA(doc)para ejecutar una consulta N1QL incrustada ) para combinar los recuentos de líneas aéreas por país y, a continuación, crear un único documento KV con los datos calculados.

SELECT country, count( * ) AS cnt FROM `travel-sample` WHERE `type` = 'airline' GROUP BY country;

1	SELECT country, count( * ) AS cnt FROM `travel-sample` WHERE `type` = 'airline' GROUP BY country;

En mi sistema de prueba un pequeño nodo único noMDS (ejecutando todos los servicios de Couchbase) el N1QL anterior tarda unos 20 ms. (para mayor claridad fingir que es supercomplejo tarda 10 segundos en completarse).

La idea aquí es que el documento KV final calculado y resumido pueda ser cargado rápidamente por 100K (o un millón) de mutaciones Eventing por segundo sin la sobrecarga adicional de comunicación con los nodos del servicio de Consulta y de procesar sentencias N1QL en cada mutación.

Debería ser obvio que el objetivo de esta lógica de negocio en particular, doCronActionA(doc)es crear una caché semiestática que se actualice periódicamente según un calendario.

Todo lo que estamos haciendo realmente (y es bastante rápido) es obtener un recuento de aerolíneas por país a partir del conjunto de documentos de la muestra de viajes. A medida que usamos N1QL construimos un documento y eventualmente lo escribimos en KV como un documento resumido. El punto clave a destacar aquí es que no queremos repetir el mismo trabajo para millones de mutaciones cada uno, sobre todo porque algunos cálculos podrían tomar 10 segundos de tiempo de cálculo del servicio de consulta cada vez que iniciamos una consulta N1QL incrustada desde una función Eventing.

A continuación mostramos la función JavaScript que queremos que se ejecute una vez al día (o quizás una vez cada hora, etc.). Observa que el nombre de la función coincide con el nombre del campo de acción de la estructura de control. Para más detalles sobre la terminología de Eventing y las construcciones del lenguaje, consulta los documentos y ejemplos de Couchbase en Servicio de Eventing: Fundamentos.

function doCronActionA(doc) {
  try {
    // Check that doc has desired values
    if (!doc.type || doc.type !== "recurring_event" || !doc.active || doc.active !== true) return;
    if (doc.verbose.user_func >= 1)
      log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

    // this is a 6.5 N1QL query (feature not available in GA prior to 6.5)
    // Create an embedded N1QL iterator by issuing a SELECT statement to get the
    // counts of airlines by country.  Make a new document and write it out to KV 

    // We will use the iterator to create a KV document representing the results of a
    // HARD lengthy embedded N1QL query and write it back to KV, the idea is to keep
    // a calculation up to date once a day such that it that can be read 'quickly' 
    // by other Eventing Functions, other Couchbase services or SDKs.   

    // Consider if we had 1 million docs in a minute do we really want to use N1QL
    // to recalculate something that is almost static for all 1 million documents, of 
    // course not, so we make an intermediate value that can be read into Eventing
    // and used via a single 'light weight' KV read.

    var q_iter = SELECT country,
      count( * ) cnt
    FROM `travel-sample`
    WHERE `type` = 'airline'
    GROUP BY country;

    // loop through the result set and update the map 'accumulate'
    var accumulate = {};
    var idx = 0;
    for (var val of q_iter) {
      if (doc.verbose.user_func >= 2)
        log(doc.action + ' N1QL idx ' + idx + ', country ' + val.country + " cnt " + val.cnt);
      accumulate[val.country] = val.cnt;
      idx++;
    }
    // close out embedded N1QL iterator
    q_iter.close();

    // Now let’s make a cached KV document representing a HARD length embedded N1QL
    // query and write it back to KV, we need a KEY and a type and id and then we 
    // upsert it into the `travel-sample` bucket.

    var cachedoc = {};
    cachedoc.type = "cron_cache";
    cachedoc.id = "airlines_by_country";
    cachedoc.date = new Date();
    cachedoc.data = accumulate;
    var ckey = cachedoc.type + '::' + cachedoc.id;
    ts_bkt[ckey] = cachedoc;
    if (doc.verbose.user_func >= 2) {
      log(doc.action + ' upsert to KV with KEY ' + ckey + ' cachedoc ', cachedoc);
    }
  } catch (e) {
    log(doc.action + ' Error exception:', e);
    return false;
  }
  return true;
}

function doCronActionA(doc) {

try {

// Check that doc has desired values

if (!doc.type || doc.type !== "recurring_event" || !doc.active || doc.active !== true) return;

if (doc.verbose.user_func >= 1)

log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

// this is a 6.5 N1QL query (feature not available in GA prior to 6.5)

// Create an embedded N1QL iterator by issuing a SELECT statement to get the

// counts of airlines by country. Make a new document and write it out to KV

// We will use the iterator to create a KV document representing the results of a

// HARD lengthy embedded N1QL query and write it back to KV, the idea is to keep

// a calculation up to date once a day such that it that can be read 'quickly'

// by other Eventing Functions, other Couchbase services or SDKs.

// Consider if we had 1 million docs in a minute do we really want to use N1QL

// to recalculate something that is almost static for all 1 million documents, of

// course not, so we make an intermediate value that can be read into Eventing

// and used via a single 'light weight' KV read.

var q_iter = SELECT country,

count( * ) cnt

FROM `travel-sample`

WHERE `type` = 'airline'

GROUP BY country;

// loop through the result set and update the map 'accumulate'

var accumulate = {};

var idx = 0;

for (var val of q_iter) {

if (doc.verbose.user_func >= 2)

log(doc.action + ' N1QL idx ' + idx + ', country ' + val.country + " cnt " + val.cnt);

accumulate[val.country] = val.cnt;

idx++;

}

// close out embedded N1QL iterator

q_iter.close();

// Now let’s make a cached KV document representing a HARD length embedded N1QL

// query and write it back to KV, we need a KEY and a type and id and then we

// upsert it into the `travel-sample` bucket.

var cachedoc = {};

cachedoc.type = "cron_cache";

cachedoc.id = "airlines_by_country";

cachedoc.date = new Date();

cachedoc.data = accumulate;

var ckey = cachedoc.type + '::' + cachedoc.id;

ts_bkt[ckey] = cachedoc;

if (doc.verbose.user_func >= 2) {

log(doc.action + ' upsert to KV with KEY ' + ckey + ' cachedoc ', cachedoc);

}

} catch (e) {

log(doc.action + ' Error exception:', e);

return false;

}

return true;

}

La función anterior simplemente 1) consulta el cubo de muestras de viajes para extraer datos, en este caso el recuento de aerolíneas para cada país, 2) crea un nuevo documento KV y una nueva clave y lo escribe en el cubo de muestras de viajes para su uso posterior.

Además, como parte de este ejemplo hemos construido un registro que responde a un ajuste numérico de verbosidad que a) registra una sola línea si el documento de control tiene un valor para doc.verbose.user_func == 1 o b) emite más información si el valor de doc.verbose.user_func >= 2.

Se trata de un marco genérico que puede ejecutar una (1) cron acción o incluso mil (1000) de cron acciones. Como tal, he proporcionado dos funciones adicionales "vacías" - como se señaló antes, podrían haber sido nombradas de cualquier manera.

function doCronActionB(doc) {
  try {
    // check that doc has desired values
    if (doc.type !== "recurring_event" || doc.active !== true) return;
    if (doc.verbose.user_func >= 1)
      log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

    // YOUR LOGIC HERE

  } catch (e) {
    log(doc.action + ' Error exception:', e);
    return false;
  }
  return true;
}

function doCronActionB(doc) {

try {

// check that doc has desired values

if (doc.type !== "recurring_event" || doc.active !== true) return;

if (doc.verbose.user_func >= 1)

log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

// YOUR LOGIC HERE

} catch (e) {

log(doc.action + ' Error exception:', e);

return false;

}

return true;

}

function doCronActionC(doc) {
  try {
    // check that doc has desired values
    if (doc.type !== "recurring_event" || doc.active !== true) return;
    if (doc.verbose.user_func >= 1)
      log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

    // YOUR LOGIC HERE

  } catch (e) {
    log(doc.action + ' Error exception:', e);
    return false;
  }
  return true;
}

function doCronActionC(doc) {

try {

// check that doc has desired values

if (doc.type !== "recurring_event" || doc.active !== true) return;

if (doc.verbose.user_func >= 1)

log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

// YOUR LOGIC HERE

} catch (e) {

log(doc.action + ' Error exception:', e);

return false;

}

return true;

}

Estas funciones doCronActionB y doCronActionC son triviales, ya que simplemente registran información en el registro de la aplicación de eventos de la función de eventos. Consulte Funciones de registro para más detalles. Por supuesto, necesita un documento de control de type="recurring_event" con active=true y una acción como action = "doCronActionB" para habilitarlos y ejecutarlos realmente.

Necesitamos un punto de entrada o manejador de eventos

A partir de la versión 6.5, el servicio de eventos admite dos puntos de entrada o controladores OnUpdate(doc, meta) y OnDelete(meta) sólo nos interesa el OnUpdate(doc,meta) para este ejemplo.

En OnUpdate(doc,meta) es llamado cuando se crea o modifica (muta) cualquier documento del bucket de origen y filtra inmediatamente los documentos que no interesan. ^[3]

function OnUpdate(doc, meta) {
  // fix for 6.5.X growing bucket ops
  if (doc.type === "_tmp_vbs") genNoopTimers(doc, meta, 30);
  if (!cron_bkt["fix_timer_scan_issue::1"]) {
      cron_bkt["fix_timer_scan_issue::1"] = {};
  }
    
  try {
    // Check if further analysis is needed we only trigger on an active recurring_event 
    if (doc.type !== "recurring_event" || doc.active !== true) return;

    var update_doc = false;
    if (!doc.dynamic) {
      // Add if missing doc.dynamic with defaults
      doc.dynamic = {
        "state": "arm",
        "next_sched": 0,
        "prev_sched": 0,
        "prev_etime": 0,
        "prev_delay": 0,
        "prev_atime": 0
      };
      // we need to update the document once we have the next schedule
      update_doc = true;
    }
    if (!doc.verbose) {
      // Add if missing doc.dynamic with defaults
      doc.verbose = {
        "user_func": 1,
        "scheduler": 1
      };
      // we need to update the document once we have the next schedule
      update_doc = true;
    }
    // Do not process dynamic.state pending
    if (!doc.dynamic || !doc.dynamic.state || doc.dynamic.state === "pending") return;

    var mid = doc.type + "::" + doc.id; // this is the same as meta.id or the KEY
    var hour = doc.hour;
    var min = doc.min;

    // Do an eval check the JavaScript function exists. The eval occurs in a common 
    // utility function shared with RecurringCallback
    if (!verifyFunctionExistsViaEval(doc, mid)) {
      // doc.action did not exist, we have already logged the issue
      return;
    }

    // Get the next valid execution time
    var date_timer = getNextRecurringDate(hour, min);
    var next_sched = Math.round(date_timer.getTime() / 1000);
    if (!update_doc && next_sched !== doc.dynamic.next_sched) {
      // the next_sched should be the same as the setting from the helper application, however
      // if we undeploy/deploy or pause/resume we might haver to reschedule to the next time slot
      log('OnUpdate U ' + mid + ' calculated next_sched !== doc.dynamic.next_sched, delta ' +
        (next_sched - doc.dynamic.next_sched) + ', reschedule');
      update_doc = true;
    }

    if (update_doc) {
      // this mutation is recursive and will be suppressed, we ensure we have a dynamic structure
      doc.dynamic.next_sched = next_sched;

      // rather then the call a function, to trap and retry if there is a resource issue
      // cron_bkt[mid] = doc;
      if (!tryBucketKvWriteWithLog('OnUpdate F', mid, doc)) {
        // Failed to write doc to cron_bkt[key] the error has been logged
        // and there is nothing more we can do.
        return;
      }
    }

    // Schedule an Eventing timer
    var timer_id = createTimer(Callback, date_timer, null, doc);
    if (doc.verbose.scheduler >= 1) {
      log('OnUpdate A ' + mid + ' rcv mutation (initial or rearm) schedule timer at ' +
        toLocalISOTime(date_timer));
    }
    if (doc.verbose.scheduler >= 2) {
      log('OnUpdate B ' + mid + ' recurring timer was created, timer_id ' + timer_id);
    }
  } catch (e) {
    log('OnUpdate E ' + meta.id + ', Error exception:', e);
  }
}

function OnUpdate(doc, meta) {

// fix for 6.5.X growing bucket ops

if (doc.type === "_tmp_vbs") genNoopTimers(doc, meta, 30);

if (!cron_bkt["fix_timer_scan_issue::1"]) {

cron_bkt["fix_timer_scan_issue::1"] = {};

}

try {

// Check if further analysis is needed we only trigger on an active recurring_event

if (doc.type !== "recurring_event" || doc.active !== true) return;

var update_doc = false;

if (!doc.dynamic) {

// Add if missing doc.dynamic with defaults

doc.dynamic = {

"state": "arm",

"next_sched": 0,

"prev_sched": 0,

"prev_etime": 0,

"prev_delay": 0,

"prev_atime": 0

};

// we need to update the document once we have the next schedule

update_doc = true;

}

if (!doc.verbose) {

// Add if missing doc.dynamic with defaults

doc.verbose = {

"user_func": 1,

"scheduler": 1

};

// we need to update the document once we have the next schedule

update_doc = true;

}

// Do not process dynamic.state pending

if (!doc.dynamic || !doc.dynamic.state || doc.dynamic.state === "pending") return;

var mid = doc.type + "::" + doc.id; // this is the same as meta.id or the KEY

var hour = doc.hour;

var min = doc.min;

// Do an eval check the JavaScript function exists. The eval occurs in a common

// utility function shared with RecurringCallback

if (!verifyFunctionExistsViaEval(doc, mid)) {

// doc.action did not exist, we have already logged the issue

return;

}

// Get the next valid execution time

var date_timer = getNextRecurringDate(hour, min);

var next_sched = Math.round(date_timer.getTime() / 1000);

if (!update_doc && next_sched !== doc.dynamic.next_sched) {

// the next_sched should be the same as the setting from the helper application, however

// if we undeploy/deploy or pause/resume we might haver to reschedule to the next time slot

log('OnUpdate U ' + mid + ' calculated next_sched !== doc.dynamic.next_sched, delta ' +

(next_sched - doc.dynamic.next_sched) + ', reschedule');

update_doc = true;

}

if (update_doc) {

// this mutation is recursive and will be suppressed, we ensure we have a dynamic structure

doc.dynamic.next_sched = next_sched;

// rather then the call a function, to trap and retry if there is a resource issue

// cron_bkt[mid] = doc;

if (!tryBucketKvWriteWithLog('OnUpdate F', mid, doc)) {

// Failed to write doc to cron_bkt[key] the error has been logged

// and there is nothing more we can do.

return;

}

// Schedule an Eventing timer

var timer_id = createTimer(Callback, date_timer, null, doc);

if (doc.verbose.scheduler >= 1) {

log('OnUpdate A ' + mid + ' rcv mutation (initial or rearm) schedule timer at ' +

toLocalISOTime(date_timer));

}

if (doc.verbose.scheduler >= 2) {

log('OnUpdate B ' + mid + ' recurring timer was created, timer_id ' + timer_id);

}

} catch (e) {

log('OnUpdate E ' + meta.id + ', Error exception:', e);

}

La clave aquí es que el cron en nuestro manejador sólo se preocupa por los documentos que tienen doc.type de "recurring_event y también un doc.active de true. Además, en este ejemplo hemos creado un rastreo para el método cron lógica de mantenimiento que sólo se registra en el registro de la aplicación si el documento de control tiene un valor para doc.verbose >= 3.

Si sólo ejecuta unas pocas programaciones, puede desactivar el trabajo en el espacio de usuario o "arreglo para 6.5.X cubo creciente ops" comentando cuatro líneas de código en el bloque OnUpdate anterior para "cron_impl_2func_651" como sigue:

function OnUpdate(doc, meta) {
  // fix for 6.5.X growing bucket ops
  // if (doc.type === "_tmp_vbs") genNoopTimers(doc, meta, 30);
  // if (!cron_bkt["fix_timer_scan_issue::1"]) {
  //     cron_bkt["fix_timer_scan_issue::1"] = {};
  // }

function OnUpdate(doc, meta) {

// fix for 6.5.X growing bucket ops

// if (doc.type === "_tmp_vbs") genNoopTimers(doc, meta, 30);

// if (!cron_bkt["fix_timer_scan_issue::1"]) {

// cron_bkt["fix_timer_scan_issue::1"] = {};

// }

Necesitamos código para solucionar posibles operaciones de crecimiento de cubos para 6.5.X

A partir de la versión 6.5.X necesitamos un "arreglo para 6.5.X cubo creciente ops"que ocurre en sistemas inactivos con muchos temporizadores programados en el futuro. Este código garantiza que un temporizador de Eventing se ha disparado recientemente en un vBucket (lo que soluciona el problema para el vBucket dado en función de cada función).

// FIXUP: ADDIN FUNCTON
function noopTimer(context) {
    // fix for 6.5.X growing bucket ops
    try {
        if (context.type === "_tmp_vbs" && context.vb === 0) { 
            // log("noopTimer timers firing, printing only for vBucket 0");
        }
    } catch (e) {
        log("OnUpdate Exception in callback noopTimer:", e);
    }
}

// FIXUP: ADDIN FUNCTON
function rearmTimer(context) {
    // fix for 6.5.X growing bucket ops
    try {
        if (context.type === "_tmp_vbs" && context.vb === 0) { 
            // Update/touch all docs in the helper_bucket the helper function will then
            // mutate all 1024 of type == vbs_seed (64 on MacOS) to create a recuring cycle.
            // log("noopTimer timer fired all 1024 vBuckets, logging only vb 0", context);
            
            // generate a mutation to re-arm the HELPER function: fix_scan_issue
            // which will in turn make new mutations for this Function
            var cur = cron_bkt[context.key];
            if (cur && cur.ts_millis === context.ts_millis) {
                // log("rearmTimer update fix_timer_scan_issue::1 in helper_bucket alias only for vBucket 0");
                var now = new Date();
                cron_bkt["fix_timer_scan_issue::1"] = { "last_update": now };
            } else {
                // NOOP we had multiple timer cycles, just let this one quietly stop.
            }
        }
    } catch (e) {
        log("OnUpdate Exception in callback rearmTimer:", e);
    }
}

// FIXUP: ADDIN FUNCTON
function genNoopTimers(doc, meta, seconds) {
    // fix for 6.5.X growing bucket ops
    try {
        // redundant but play it safe
        if (doc.type === "_tmp_vbs") {
            // Since we are using an different function a timer on all our vBuckets do immeadiately (can take up to 15 seconds)
            // If we used cross bucket recursion to rearm all the timers in a recurring fashion we would add a delay of at least 40 seconds.
            createTimer(noopTimer, new Date(), null, doc);
            if (doc.vb === 0) { 
                // Update/touch all docs in the helper_bucket the helper function will then
                // mutate all 1024 of type == vbs_seed (64 on MacOS) to create a recuring cycle.
                // log("noopTimer timer fired all 1024 vBuckets, logging only vb 0", context);
            
                // generate a mutation to re-arm the HELPER function: fix_scan_issue
                // which will in turn make new mutations for this Function
                
                // log("genNoopTimers make timer to rearm fix_timer_scan_issue::1");
                createTimer(rearmTimer, new Date(new Date().getTime() + seconds * 1000), null, doc);
            }
        }
    } catch (e) {
        log("OnUpdate Exception in genNoopTimers:", e);
    }
}

// FIXUP: ADDIN FUNCTON

function noopTimer(context) {

// fix for 6.5.X growing bucket ops

try {

if (context.type === "_tmp_vbs" && context.vb === 0) {

// log("noopTimer timers firing, printing only for vBucket 0");

}

} catch (e) {

log("OnUpdate Exception in callback noopTimer:", e);

}

// FIXUP: ADDIN FUNCTON

function rearmTimer(context) {

// fix for 6.5.X growing bucket ops

try {

if (context.type === "_tmp_vbs" && context.vb === 0) {

// Update/touch all docs in the helper_bucket the helper function will then

// mutate all 1024 of type == vbs_seed (64 on MacOS) to create a recuring cycle.

// log("noopTimer timer fired all 1024 vBuckets, logging only vb 0", context);

// generate a mutation to re-arm the HELPER function: fix_scan_issue

// which will in turn make new mutations for this Function

var cur = cron_bkt[context.key];

if (cur && cur.ts_millis === context.ts_millis) {

// log("rearmTimer update fix_timer_scan_issue::1 in helper_bucket alias only for vBucket 0");

var now = new Date();

cron_bkt["fix_timer_scan_issue::1"] = { "last_update": now };

} else {

// NOOP we had multiple timer cycles, just let this one quietly stop.

}

} catch (e) {

log("OnUpdate Exception in callback rearmTimer:", e);

}

// FIXUP: ADDIN FUNCTON

function genNoopTimers(doc, meta, seconds) {

// fix for 6.5.X growing bucket ops

try {

// redundant but play it safe

if (doc.type === "_tmp_vbs") {

// Since we are using an different function a timer on all our vBuckets do immeadiately (can take up to 15 seconds)

// If we used cross bucket recursion to rearm all the timers in a recurring fashion we would add a delay of at least 40 seconds.

createTimer(noopTimer, new Date(), null, doc);

if (doc.vb === 0) {

// Update/touch all docs in the helper_bucket the helper function will then

// mutate all 1024 of type == vbs_seed (64 on MacOS) to create a recuring cycle.

// log("noopTimer timer fired all 1024 vBuckets, logging only vb 0", context);

// generate a mutation to re-arm the HELPER function: fix_scan_issue

// which will in turn make new mutations for this Function

// log("genNoopTimers make timer to rearm fix_timer_scan_issue::1");

createTimer(rearmTimer, new Date(new Date().getTime() + seconds * 1000), null, doc);

}

} catch (e) {

log("OnUpdate Exception in genNoopTimers:", e);

}

Necesitamos una utilidad para calcular la siguiente hora en el horario

La siguiente función getNextRecurringDate(hour, min) determinará una hora para ejecutar la acción definida como parte de nuestra programación. Esto no es una implementación completa de cronEn lugar de ello, contiene las características estándar clave para ejecutarse una vez al día, una vez por hora y una vez por minuto. También contiene alguna sintaxis no estándar para proporcionar la capacidad de ejecutar dos veces por minuto o cuatro veces por minuto.

Como se ha descrito anteriormente, la función getNextRecurringDate(hora, min) permite lo siguiente (la tabla se duplica a continuación), siendo las dos últimas no estándar.^[4]

hora	min	Los valores pueden ser números o cadenas
13	32	Corre a las 13:32 (o 13:32)
*	15	Se ejecuta cada hora a los 15 minutos
8	12	Corre una vez al día a las 8:32 (o 8:32 am)
*	*	Ejecutar una vez por minuto
*2X	*2X	Se ejecuta dos veces por minuto - requiere que tanto la hora como el minuto estén ajustados a "*2X".
*4X	*4X	Funciona cuatro veces por minuto - requiere que la hora y el minuto estén ajustados a "*2X".

A continuación se muestra una implementación de la lógica necesaria para determinar la próxima vez que debe activarse un temporizador de Eventing en nuestro programa, en el caso de que la lógica de usuario de nuestro primer ejemplo doCronActionA(doc) no se completa a tiempo, por ejemplo, si se sobrepasa el tiempo real, se seleccionará el siguiente cuanto de la programación. Tenga en cuenta tanto los Temporizadores como sus Funciones Padre. Así, si una Función de Eventos tiene un tiempo de espera de ejecución por defecto de 60 segundos, si es necesario esta configuración puede ser ajustada o aumentada.

function getNextRecurringDate(hour_str, min_str) {
  // Note Javascript Dates are in milliseconds
  var date_now = new Date();
  var date_ret = new Date();
  var hour;
  var min;

  try {
    hour = parseInt(hour_str);
  } catch (e) {}
  try {
    min = parseInt(min_str);
  } catch (e) {}

  // Note, this is only a simplistic partial 'crontab' syntax with some slight extensions
  // it allows once a day, once an hour, once a minute.  It also contains some non-standard 
  // syntax to provide the ability to execute twice a minute or four times a minute.

  if (hour_str === '*4X' && min_str === '*4X') {
    // once every 15 seconds or four times a minute
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(15);
    while (date_ret.getTime() < date_now.getTime()) {
      date_ret.setSeconds(date_ret.getSeconds() + 15);
    }
    return date_ret;
  } else
  if (hour_str === '*2X' && min_str === '*2X') {
    // once every 30 seconds or twice a minute
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(30);
    while (date_ret.getTime() < date_now.getTime()) {
      date_ret.setSeconds(date_ret.getSeconds() + 30);
    }
    return date_ret;
  } else
  if (hour_str === '*' && min_str === '*') {
    // once a minute 
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(date_ret.getMinutes() + 1);
  } else
  if (hour_str !== '*' && isNaN(hour) === false && min_str === '*') {
    // once a minute only for a given hour
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(date_ret.getMinutes() + 1);
    if (date_ret.getTime() < date_now.getTime()) { date_ret.setHours(hour); } if (date_ret.getTime() > date_now.getTime()) {
      date_ret.setDate(date_ret.getDate() + 1);
      date_ret.setSeconds(0);
      date_ret.setMinutes(0);
      date_ret.setHours(hour);
    }
  } else
  if (hour_str === '*' && min_str !== '*' && isNaN(min) === false) {
    // once a hour at a given minute
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(min);
    // schedule for next hour
    date_ret.setHours(date_ret.getHours() + 1);
  } else
  if (isNaN(hour) === false && isNaN(min) === false) {
    // once a day for a given hour and a given minute 
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(min);
    date_ret.setHours(hour);
    if (date_ret.getTime() < date_now.getTime()) {
      // schedule for tomorrow
      date_ret.setDate(date_ret.getDate() + 1);
    }
  } else {
    log('getNextRecurringDate illegal input hour_str <' + hour_str + '> min_str <' + min_str + '>');
    throw new Error('getNextRecurringDate illegal input hour_str <' + hour_str + '> min_str <' + min_str + '>');
    return null;
  }
  return date_ret;
}

function getNextRecurringDate(hour_str, min_str) {

// Note Javascript Dates are in milliseconds

var date_now = new Date();

var date_ret = new Date();

var hour;

var min;

try {

hour = parseInt(hour_str);

} catch (e) {}

try {

min = parseInt(min_str);

} catch (e) {}

// Note, this is only a simplistic partial 'crontab' syntax with some slight extensions

// it allows once a day, once an hour, once a minute. It also contains some non-standard

// syntax to provide the ability to execute twice a minute or four times a minute.

if (hour_str === '*4X' && min_str === '*4X') {

// once every 15 seconds or four times a minute

date_ret.setMilliseconds(0);

date_ret.setSeconds(15);

while (date_ret.getTime() < date_now.getTime()) {

date_ret.setSeconds(date_ret.getSeconds() + 15);

}

return date_ret;

} else

if (hour_str === '*2X' && min_str === '*2X') {

// once every 30 seconds or twice a minute

date_ret.setMilliseconds(0);

date_ret.setSeconds(30);

while (date_ret.getTime() < date_now.getTime()) {

date_ret.setSeconds(date_ret.getSeconds() + 30);

}

return date_ret;

} else

if (hour_str === '*' && min_str === '*') {

// once a minute

date_ret.setMilliseconds(0);

date_ret.setSeconds(0);

date_ret.setMinutes(date_ret.getMinutes() + 1);

} else

if (hour_str !== '*' && isNaN(hour) === false && min_str === '*') {

// once a minute only for a given hour

date_ret.setMilliseconds(0);

date_ret.setSeconds(0);

date_ret.setMinutes(date_ret.getMinutes() + 1);

if (date_ret.getTime() < date_now.getTime()) { date_ret.setHours(hour); } if (date_ret.getTime() > date_now.getTime()) {

date_ret.setDate(date_ret.getDate() + 1);

date_ret.setSeconds(0);

date_ret.setMinutes(0);

date_ret.setHours(hour);

}

} else

if (hour_str === '*' && min_str !== '*' && isNaN(min) === false) {

// once a hour at a given minute

date_ret.setMilliseconds(0);

date_ret.setSeconds(0);

date_ret.setMinutes(min);

// schedule for next hour

date_ret.setHours(date_ret.getHours() + 1);

} else

if (isNaN(hour) === false && isNaN(min) === false) {

// once a day for a given hour and a given minute

date_ret.setMilliseconds(0);

date_ret.setSeconds(0);

date_ret.setMinutes(min);

date_ret.setHours(hour);

if (date_ret.getTime() < date_now.getTime()) {

// schedule for tomorrow

date_ret.setDate(date_ret.getDate() + 1);

}

} else {

log('getNextRecurringDate illegal input hour_str <' + hour_str + '> min_str <' + min_str + '>');

throw new Error('getNextRecurringDate illegal input hour_str <' + hour_str + '> min_str <' + min_str + '>');

return null;

}

return date_ret;

}

Necesitamos algunas pequeñas utilidades

La función de utilidad común que simplemente comprueba si nuestro JavaScript existe utilizado por ambos OnUpdate(doc,meta), mostrado arriba, y el temporizador Devolución de llamada(doc), que se muestra más adelante. A continuación verifyFunctionExistsViaEval(curDoc, id) que recibe dos argumentos: un documento de control JSON y la CLAVE de dicho documento.

Esto nos permite saber inmediatamente, en el despliegue, si hubo un problema con una falta de coincidencia de nombres entre el documento de registro de control JSON y el nombre real de la función de lógica de negocio en el código JavaScript.

function verifyFunctionExistsViaEval(curDoc, id) {
  var result = false;
  try {
    // check for function if missing this is invalid return result
    result = eval("typeof " + curDoc.action + " === 'function';");
    if (result === false) {
      if (curDoc.verbose.scheduler >= 1)
        log("Warn/Disable (No Action and No Re-Arm), because required 'action' of " +
          curDoc.action + "(doc) does not exist, id is", id);
      return result;
    }
  } catch (e) {
    log('verifyFunctionExistsViaEval Error exception:', e);
  }
  return result;
}

function verifyFunctionExistsViaEval(curDoc, id) {

var result = false;

try {

// check for function if missing this is invalid return result

result = eval("typeof " + curDoc.action + " === 'function';");

if (result === false) {

if (curDoc.verbose.scheduler >= 1)

log("Warn/Disable (No Action and No Re-Arm), because required 'action' of " +

curDoc.action + "(doc) does not exist, id is", id);

return result;

}

} catch (e) {

log('verifyFunctionExistsViaEval Error exception:', e);

}

return result;

}

Tenga en cuenta que si se intenta ejecutar una función inexistente, el usuario final recibirá una advertencia en el registro de la aplicación. cron_impl_2func_651.log para corregir el problema.

2020-04-22T16:20:38.725-07:00 [INFO] "Advertir/Desactivar (Sin Acción y Sin Rearmar), porque la 'acción' requerida de doCronMyNewFunction(doc) no existe, id es" "recurring_event::1"

Esta corrección se puede hacer a través de una Pausa/Reanudar añadiendo la función y luego ajustando el documento de control con el id especificado o CLAVE (a través de un toggle activo a falso y luego verdadero) -o- ajustando el documento de control para apuntar a una función existente en su manejador.

A continuación, la utilidad toNumericFixed(número, precisión) sólo permite un bonito formato compacto de flotantes para nuestros mensajes de registro.

function toNumericFixed(number, precision) {
  var multi = Math.pow(10, precision);
  return Math.round((number * multi).toFixed(precision + 1)) / multi;
}

function toNumericFixed(number, precision) {

var multi = Math.pow(10, precision);

return Math.round((number * multi).toFixed(precision + 1)) / multi;

}

Por último, la utilidad toLocalISOTime(d) sólo permite un formato compacto de fechas para nuestros mensajes de registro.

function toLocalISOTime(d) {
  var tzoffset = (new Date()).getTimezoneOffset() * 60000; //offset in milliseconds
  return (new Date(d.getTime() - tzoffset)).toISOString().slice(0, -1);
}

function toLocalISOTime(d) {

var tzoffset = (new Date()).getTimezoneOffset() * 60000; //offset in milliseconds

return (new Date(d.getTime() - tzoffset)).toISOString().slice(0, -1);

}

Necesitamos una llamada de retorno del temporizador para ejecutar la lógica de usuario y volver a activar el temporizador

La última función JavaScript de "cron_impl_2func_651" es la llamada de retorno del temporizador, a la que se llama cuando se activa el temporizador programado. La función de devolución de llamada debe ser una función de nivel superior que reciba un único argumento, el contexto.

En este caso en nuestro manejador OnUpdate referenciamos una función JavaScript de Devolución de llamada(doc) con un contexto de doc (nuestro documento de control del planificador activo de type="recurring_event")

En la versión 6.6 podemos crear otro temporizador dentro de un temporizador pero para todas las versiones anteriores necesitaremos lanzar una mutación a una función "helper" (evitamos cuidadosamente la recursión infinita). En la 6.6 la función helper no es necesaria y la lógica se ha simplificado sustancialmente.

function Callback(doc) {
  try {
    var fired_at = new Date();

    // Check if further analysis is needed we only trigger on a recurring_event that is active
    if (doc.type !== "recurring_event") return;
    // doc must have 'action', 'dynamic {}', verbose {}, dynamic.state
    if (!doc.action || !doc.dynamic || !doc.verbose || !doc.dynamic.state) return;
    // process any doc.dynamic.state BUT pending 
    if (doc.dynamic.state === "pending") return;

    // ==================
    // Check if still active

    // We make sure that in KV the 'doc' still exists and that it is still active if not just 
    // return thus skipping the action and not Re-arming the timer. Note `travel-sample` is 
    // aliased to the map 'cron_bkt

    var mid = doc.type + '::' + doc.id; // make our KEY
    var curDoc = null;
    try {
      // read the current version of doc from KV, e.g. curDoc
      curDoc = cron_bkt[mid];
    } catch (e) {} // needed for pre 6.5, note pure 6.5+ deployment returns null sans exception

    var reason = null;
    if (!curDoc || curDoc === null) {
      reason = "cron document is missing";
    } else
    if (!curDoc.active) {
      reason = "cron document has active = false";
    } else
    if (!curDoc.dynamic.state || curDoc.dynamic.state !== doc.dynamic.state) {
      reason = "cron document wrong dynamic.state expected " + doc.dynamic.state;
    } else
    if (crc64(doc) !== crc64(curDoc)) {
      reason = "cron document changed";
    }

    if (reason !== null) {
      if (!curDoc || curDoc === null || curDoc.verbose.scheduler >= 1) {
        log('Callback X ' + mid + " ignore/stop this timer's schedule because " + reason);
      }
      if (!curDoc || curDoc === null || curDoc.verbose.scheduler >= 4) {
        log('Callback Y ' + mid + ' timer doc', doc);
        log('Callback Z ' + mid + ' KV curDoc', curDoc);
      }
      return;
    }

    // ==================
    // Verify user routine exists and if so eval it 

    // Assume curDoc.action contains something like "doCronActionA" and we have a function in 
    // this handler like "doCronActionA(doc)". Below we use curDoc as the end user should be 
    // able to alter the eval'd JavaScript function.  We will execute two (2) evals.

    // First eval check the JavaScript function exists.  The eval occurs in a common 
    // utility function shared with Callback
    if (!verifyFunctionExistsViaEval(curDoc, mid)) {
      // curDoc.action did not exist, we have already logged the issue
      return;
    }

    // Second eval execute and process the user function we execute the defined function 
    // with an argument of curDoc
    var beg_act = new Date();
    var result = null;
    eval("result = " + curDoc.action + "(curDoc);");
    var end_act = new Date();
    var atime_ms = end_act.getTime() - beg_act.getTime();

    if (curDoc.verbose.scheduler >= 2)
      log('Callback R ' + mid + ' action took ' + toNumericFixed((atime_ms / 1000), 3) +
        ' sec., returned ' + result);

    // ==================
    // Calculate next time and mutate the control document for our our helper function
    // which will create another mutation such that OnUpdate of this function will pick
    // it up and generate the timer (avoids the MB-38554 issue).

    var hour = curDoc.hour;
    var min = curDoc.min;
    var date_timer = getNextRecurringDate(hour, min);

    curDoc.dynamic.prev_delay =
      toNumericFixed(((fired_at.getTime() / 1000) - curDoc.dynamic.next_sched), 3);
    curDoc.dynamic.prev_sched = curDoc.dynamic.next_sched;
    curDoc.dynamic.prev_etime = Math.round(fired_at.getTime() / 1000);
    curDoc.dynamic.prev_atime = toNumericFixed((atime_ms / 1000), 3);

    curDoc.dynamic.state = "pending";
    curDoc.dynamic.next_sched = Math.round(date_timer.getTime() / 1000);
    
    try {
      cron_bkt[mid] = curDoc;
    } catch (e) {
      log('Callback help: F ' + mid + ' FATAL could not update KV cron cycle ' + curDoc.action);
      return;
    }

    if (curDoc.verbose.scheduler >= 1) {
      log('Callback A ' + mid + ' gen mutation #1 to doc to force schedule rearm at ' +
        toLocalISOTime(date_timer));
    }
    if (curDoc.verbose.scheduler >= 2) {
      log('Callback B ' + mid + ' sched ' + curDoc.dynamic.prev_sched +
        ', actual ' + curDoc.dynamic.prev_etime +
        ', delay ' + curDoc.dynamic.prev_delay +
        ', took ' + curDoc.dynamic.prev_atime);
    }
    if (curDoc.verbose.scheduler >= 3) {
      log('Callback C ' + mid + ' curDoc', curDoc);
    }
  } catch (e) {
    var mid = doc.type + '::' + doc.id; // make our KEY
    log('Callback E ' + mid + ' Error exception:', e);
  }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

function Callback(doc) {

try {

var fired_at = new Date();

// Check if further analysis is needed we only trigger on a recurring_event that is active

if (doc.type !== "recurring_event") return;

// doc must have 'action', 'dynamic {}', verbose {}, dynamic.state

if (!doc.action || !doc.dynamic || !doc.verbose || !doc.dynamic.state) return;

// process any doc.dynamic.state BUT pending

if (doc.dynamic.state === "pending") return;

// ==================

// Check if still active

// We make sure that in KV the 'doc' still exists and that it is still active if not just

// return thus skipping the action and not Re-arming the timer. Note `travel-sample` is

// aliased to the map 'cron_bkt

var mid = doc.type + '::' + doc.id; // make our KEY

var curDoc = null;

try {

// read the current version of doc from KV, e.g. curDoc

curDoc = cron_bkt[mid];

} catch (e) {} // needed for pre 6.5, note pure 6.5+ deployment returns null sans exception

var reason = null;

if (!curDoc || curDoc === null) {

reason = "cron document is missing";

} else

if (!curDoc.active) {

reason = "cron document has active = false";

} else

if (!curDoc.dynamic.state || curDoc.dynamic.state !== doc.dynamic.state) {

reason = "cron document wrong dynamic.state expected " + doc.dynamic.state;

} else

if (crc64(doc) !== crc64(curDoc)) {

reason = "cron document changed";

}

if (reason !== null) {

if (!curDoc || curDoc === null || curDoc.verbose.scheduler >= 1) {

log('Callback X ' + mid + " ignore/stop this timer's schedule because " + reason);

}

if (!curDoc || curDoc === null || curDoc.verbose.scheduler >= 4) {

log('Callback Y ' + mid + ' timer doc', doc);

log('Callback Z ' + mid + ' KV curDoc', curDoc);

}

return;

}

// ==================

// Verify user routine exists and if so eval it

// Assume curDoc.action contains something like "doCronActionA" and we have a function in

// this handler like "doCronActionA(doc)". Below we use curDoc as the end user should be

// able to alter the eval'd JavaScript function. We will execute two (2) evals.

// First eval check the JavaScript function exists. The eval occurs in a common

// utility function shared with Callback

if (!verifyFunctionExistsViaEval(curDoc, mid)) {

// curDoc.action did not exist, we have already logged the issue

return;

}

// Second eval execute and process the user function we execute the defined function

// with an argument of curDoc

var beg_act = new Date();

var result = null;

eval("result = " + curDoc.action + "(curDoc);");

var end_act = new Date();

var atime_ms = end_act.getTime() - beg_act.getTime();

if (curDoc.verbose.scheduler >= 2)

log('Callback R ' + mid + ' action took ' + toNumericFixed((atime_ms / 1000), 3) +

' sec., returned ' + result);

// ==================

// Calculate next time and mutate the control document for our our helper function

// which will create another mutation such that OnUpdate of this function will pick

// it up and generate the timer (avoids the MB-38554 issue).

var hour = curDoc.hour;

var min = curDoc.min;

var date_timer = getNextRecurringDate(hour, min);

curDoc.dynamic.prev_delay =

toNumericFixed(((fired_at.getTime() / 1000) - curDoc.dynamic.next_sched), 3);

curDoc.dynamic.prev_sched = curDoc.dynamic.next_sched;

curDoc.dynamic.prev_etime = Math.round(fired_at.getTime() / 1000);

curDoc.dynamic.prev_atime = toNumericFixed((atime_ms / 1000), 3);

curDoc.dynamic.state = "pending";

curDoc.dynamic.next_sched = Math.round(date_timer.getTime() / 1000);

try {

cron_bkt[mid] = curDoc;

} catch (e) {

log('Callback help: F ' + mid + ' FATAL could not update KV cron cycle ' + curDoc.action);

return;

}

if (curDoc.verbose.scheduler >= 1) {

log('Callback A ' + mid + ' gen mutation #1 to doc to force schedule rearm at ' +

toLocalISOTime(date_timer));

}

if (curDoc.verbose.scheduler >= 2) {

log('Callback B ' + mid + ' sched ' + curDoc.dynamic.prev_sched +

', actual ' + curDoc.dynamic.prev_etime +

', delay ' + curDoc.dynamic.prev_delay +

', took ' + curDoc.dynamic.prev_atime);

}

if (curDoc.verbose.scheduler >= 3) {

log('Callback C ' + mid + ' curDoc', curDoc);

}

} catch (e) {

var mid = doc.type + '::' + doc.id; // make our KEY

log('Callback E ' + mid + ' Error exception:', e);

}

Necesitamos una función de ayuda para desencadenar una nueva mutación

Dado que antes de la versión 6.6 (que aún no se ha publicado) no se puede crear un temporizador desde la llamada de retorno de un temporizador en ejecución, necesitamos una segunda función de eventos (junto con "allow_interbucket_recursion":true) para activar una mutación de forma que podamos generar todos nuestros temporizadores en el punto de entrada OnUpdate(doc,meta) de la función de eventos principal. Hacemos esto de la siguiente manera:

cron_impl_2func_651 OnUpdate(doc,meta) recibe una mutación, programa un temporizador
cron_impl_2func_651 Después de una cantidad de tiempo cuando el temporizador madura el Devolución de llamada(doc) se ejecuta la rutina, primero ejecuta la acción de usuario deseada y luego crea una mutación #1 en el documento de control (que no es vista por la Función creadora para evitar la recursividad)
cron_impl_2func_651_help OnUpdate(doc,meta) recibe una mutación, hace otra mutación #2 en el documento de control esto desencadena 1. anterior en un ciclo sin fin.

Nota, en la versión 6.6 de Couchbase no necesitamos una función de ayuda en absoluto porque se te permite crear un temporizador desde dentro de un temporizador en ejecución. Esto simplifica enormemente la lógica necesaria para hacer un cron sistema^[2].

La única función JavaScript en "cron_impl_2func_651_help" OnUpdate(doc,meta) se muestra a continuación.

function OnUpdate(doc, meta) {
  // fix for 6.5.X growing bucket ops
  if (meta.id.startsWith("fix_timer_scan_issue:")) upsertOneDocPerBucket(doc, meta);
  
  try {
    // Check that doc has desired values
    if (!doc.type || doc.type !== "recurring_event" || !doc.active || doc.active != true) return;
    // doc must have 'action', 'dynamic {}', verbose {}, dynamic.state
    if (!doc.action || !doc.dynamic || !doc.verbose || !doc.dynamic.state) return;
    // Only process state pending this will only exist for a 'breif' time
    if (doc.dynamic.state !== "pending") return;

    var mid = doc.type + '::' + doc.id; // make our KEY
    var newdoc = null;
    try {
      // read the current version of doc from KV, e.g. curDoc
      newdoc = cron_bkt[mid];
    } catch (e) {} // needed for pre 6.5, note pure 6.5+ deployment returns null sans exception
    
    var reason = null;
    if (!newdoc || newdoc == null) {
      reason = "cron document is missing";
    } else
    if (!newdoc.active) {
      reason = "cron document has active = false";
    } else
    if (!newdoc.dynamic.state || newdoc.dynamic.state !== doc.dynamic.state) {
      reason = "cron document wrong dynamic.state expected " + doc.dynamic.state;
    } else
    if (crc64(doc) !== crc64(newdoc)) {
      reason = "cron document changed";
    }
    if (reason != null) {
      if (!newdoc || newdoc == null || newdoc.verbose.scheduler >= 1) {
        log('OnUpdate help: X stopping schedule because ' + reason + ',', newdoc)
        return;
      }
    }

    newdoc.dynamic.state = "rearm";
    // cron_bkt[mid] = newdoc;
    if (!tryBucketKvWriteWithLog('OnUpdate help: F', mid, newdoc)) {
      // Failed to write newdoc to cron_bkt[key] the error has been logged
      // and there is nothing more we can do.
      return;
    }

    if (newdoc.verbose.scheduler >= 1) {
      log('OnUpdate help: A ' + mid + ' mutation #2 to doc to force schedule rearm');
    }
    if (newdoc.verbose.scheduler >= 3) {
      log('OnUpdate help: B ' + mid + ',', newdoc);
    }
  } catch (e) {
    log('OnUpdate help: E ' + meta.id + ', Error exception:', e);
  }
}

function tryBucketKvWriteWithLog(tag, key, doc) {
  var success = false;
  var tries = 0;
  while (tries < 10) {
    tries++;
    try {
      // critical that the below succeeds, because if it doesn't the cron cycle will break
      cron_bkt[key] = doc;
      success = true;
      break;
    } catch (e) {
      log(tag + ' ' + key + ' WARN failed to update KV tries ' + tries, e);
    }
  }
  if (!success) {
    log(tag + ' ' + +key + ' FATAL could not update KV cron cycle, tried ' + tries + ', stoping ' + curDoc.action);
  }
  return success;
}

function OnUpdate(doc, meta) {

// fix for 6.5.X growing bucket ops

if (meta.id.startsWith("fix_timer_scan_issue:")) upsertOneDocPerBucket(doc, meta);

try {

// Check that doc has desired values

if (!doc.type || doc.type !== "recurring_event" || !doc.active || doc.active != true) return;

// doc must have 'action', 'dynamic {}', verbose {}, dynamic.state

if (!doc.action || !doc.dynamic || !doc.verbose || !doc.dynamic.state) return;

// Only process state pending this will only exist for a 'breif' time

if (doc.dynamic.state !== "pending") return;

var mid = doc.type + '::' + doc.id; // make our KEY

var newdoc = null;

try {

// read the current version of doc from KV, e.g. curDoc

newdoc = cron_bkt[mid];

} catch (e) {} // needed for pre 6.5, note pure 6.5+ deployment returns null sans exception

var reason = null;

if (!newdoc || newdoc == null) {

reason = "cron document is missing";

} else

if (!newdoc.active) {

reason = "cron document has active = false";

} else

if (!newdoc.dynamic.state || newdoc.dynamic.state !== doc.dynamic.state) {

reason = "cron document wrong dynamic.state expected " + doc.dynamic.state;

} else

if (crc64(doc) !== crc64(newdoc)) {

reason = "cron document changed";

}

if (reason != null) {

if (!newdoc || newdoc == null || newdoc.verbose.scheduler >= 1) {

log('OnUpdate help: X stopping schedule because ' + reason + ',', newdoc)

return;

}

newdoc.dynamic.state = "rearm";

// cron_bkt[mid] = newdoc;

if (!tryBucketKvWriteWithLog('OnUpdate help: F', mid, newdoc)) {

// Failed to write newdoc to cron_bkt[key] the error has been logged

// and there is nothing more we can do.

return;

}

if (newdoc.verbose.scheduler >= 1) {

log('OnUpdate help: A ' + mid + ' mutation #2 to doc to force schedule rearm');

}

if (newdoc.verbose.scheduler >= 3) {

log('OnUpdate help: B ' + mid + ',', newdoc);

}

} catch (e) {

log('OnUpdate help: E ' + meta.id + ', Error exception:', e);

}

function tryBucketKvWriteWithLog(tag, key, doc) {

var success = false;

var tries = 0;

while (tries < 10) {

tries++;

try {

// critical that the below succeeds, because if it doesn't the cron cycle will break

cron_bkt[key] = doc;

success = true;

break;

} catch (e) {

log(tag + ' ' + key + ' WARN failed to update KV tries ' + tries, e);

}

if (!success) {

log(tag + ' ' + +key + ' FATAL could not update KV cron cycle, tried ' + tries + ', stoping ' + curDoc.action);

}

return success;

}

La función helper necesita algunas utilidades

Estas utilidades proporcionan un arreglo para 6.5.X cubo creciente ops asegurándose de que se dispara un temporizador de eventos en cada vBucket en el momento oportuno.

// FIXUP: ADDIN FUNCTON
// fix for 6.5.X growing bucket ops
function upsertOneDocPerBucket(doc, meta) {

    var crcTable = makeCRC32Table();

    // make one doc per bucket
    var isVerbose = 0;
    var isMacOS = false; // would be nice if this was an exposed constant in Eventing
    var numvbs = 1024;   // default is linux/PC
    if (isMacOS) {
        numvbs = 64;
    }

    var beg = (new Date).getTime();
    var result = getKeysToCoverAllPartitions(crcTable, "_tmp_vbs:", numvbs);

    for (var vb=0; vb<numvbs; vb++) {
        // brute force to fit a key prefix into a vBucket
         var tst = result[vb];
        if (isVerbose > 1  || (isVerbose == 1) && (vb < 3 || vb > numvbs -4)) {
            log("KEY: " + tst);
        } else {
            if (vb == 5) console.log("\t*\n\t*\n\t*");
        }
        // update the items to trigger a mutation for our PRIMARY fucntion
        cron_bkt[tst] = { "type": "_tmp_vbs", "vb": vb, "ts_millis": beg, "key": tst };
    }
    var end = (new Date).getTime();
    log("seeding one doc to each vBucket in primary_bucket alias (took " + (end - beg) + " mililis)");
}

// FIXUP: ADDIN FUNCTON
// fix for 6.5.X growing bucket ops
function showHex(n) {
    return n.toString(16);
}

// FIXUP: ADDIN FUNCTON
// fix for 6.5.X growing bucket ops
function makeCRC32Table() {
    var crcTable = [];
    var c;
    for(var n =0; n < 256; n++){
        c = n;
        for(var k =0; k < 8; k++){
            c = ((c&1) ? (0xEDB88320 ^ (c >>> 1)) : (c >>> 1));
        }
        crcTable[n] = c;
    }
    return crcTable;
}

// FIXUP: ADDIN FUNCTON
// fix for 6.5.X growing bucket ops
function crc32(crcTable,str) {
    var crc = 0 ^ (-1);
    for (var i = 0; i < str.length; i++ ) {
        crc = (crc >>> 8) ^ crcTable[(crc ^ str.charCodeAt(i)) & 0xFF];
    }
    return (crc ^ (-1)) >>> 0;
}

// FIXUP: ADDIN FUNCTON
// fix for 6.5.X growing bucket ops
function getKeysToCoverAllPartitions(crcTable,keyPrefix,partitionCount) {
    var result = [];
    var remaining = partitionCount;
    for (var i = 0; remaining > 0; i++) {
      var key = keyPrefix + i;
      var rv = (crc32(crcTable,key) >> 16) & 0x7fff;
      var actualPartition = rv & partitionCount - 1;
      if (!result[actualPartition] || result[actualPartition] === undefined) {
        result[actualPartition] = key;
        remaining--;
      }
    }
    return result;
}

// FIXUP: ADDIN FUNCTON

// fix for 6.5.X growing bucket ops

function upsertOneDocPerBucket(doc, meta) {

var crcTable = makeCRC32Table();

// make one doc per bucket

var isVerbose = 0;

var isMacOS = false; // would be nice if this was an exposed constant in Eventing

var numvbs = 1024; // default is linux/PC

if (isMacOS) {

numvbs = 64;

}

var beg = (new Date).getTime();

var result = getKeysToCoverAllPartitions(crcTable, "_tmp_vbs:", numvbs);

for (var vb=0; vb<numvbs; vb++) {

// brute force to fit a key prefix into a vBucket

var tst = result[vb];

if (isVerbose > 1 || (isVerbose == 1) && (vb < 3 || vb > numvbs -4)) {

log("KEY: " + tst);

} else {

if (vb == 5) console.log("\t*\n\t*\n\t*");

}

// update the items to trigger a mutation for our PRIMARY fucntion

cron_bkt[tst] = { "type": "_tmp_vbs", "vb": vb, "ts_millis": beg, "key": tst };

}

var end = (new Date).getTime();

log("seeding one doc to each vBucket in primary_bucket alias (took " + (end - beg) + " mililis)");

}

// FIXUP: ADDIN FUNCTON

// fix for 6.5.X growing bucket ops

function showHex(n) {

return n.toString(16);

}

// FIXUP: ADDIN FUNCTON

// fix for 6.5.X growing bucket ops

function makeCRC32Table() {

var crcTable = [];

var c;

for(var n =0; n < 256; n++){

c = n;

for(var k =0; k < 8; k++){

c = ((c&1) ? (0xEDB88320 ^ (c >>> 1)) : (c >>> 1));

}

crcTable[n] = c;

}

return crcTable;

}

// FIXUP: ADDIN FUNCTON

// fix for 6.5.X growing bucket ops

function crc32(crcTable,str) {

var crc = 0 ^ (-1);

for (var i = 0; i < str.length; i++ ) {

crc = (crc >>> 8) ^ crcTable[(crc ^ str.charCodeAt(i)) & 0xFF];

}

return (crc ^ (-1)) >>> 0;

}

// FIXUP: ADDIN FUNCTON

// fix for 6.5.X growing bucket ops

function getKeysToCoverAllPartitions(crcTable,keyPrefix,partitionCount) {

var result = [];

var remaining = partitionCount;

for (var i = 0; remaining > 0; i++) {

var key = keyPrefix + i;

var rv = (crc32(crcTable,key) >> 16) & 0x7fff;

var actualPartition = rv & partitionCount - 1;

if (!result[actualPartition] || result[actualPartition] === undefined) {

result[actualPartition] = key;

remaining--;

}

return result;

}

Ahora vamos a desplegar las dos Funciones Eventing

Hemos revisado mucho código y el diseño del planificador inicial, ahora es el momento de ver cómo funciona todo junto.

Recuerde que en este ejemplo hay tres cubos muestra-viaje (un ejemplo de conjunto de datos por defecto), metadatos(el cubo de metadatos es un bloc de notas para Eventing y puede compartirse con otras funciones de Eventing) y, por último, la función crondata (que contiene nuestros cronogramas). La dirección viaje-muestra tiene un tamaño de 100 MB y los otros dos cubos metadatos y crondata deben tener un tamaño de 200 MB y existir ya según las instrucciones de "Requisitos previos".

Verifique la configuración actual de su cubo accediendo a la página Consola Web de Couchbase > Cubos página:

Para desplegar la función Eventing "cron_impl_2func_651"puede seguir uno de estos dos métodos:

Complejidad básica, Método #1 Descargar/Importar
Complejidad media, Método #2 Añadir función manualmente, Cortar y pegar JavaScript

Método #1 Descargar/Importar

Importar la 1ª función "cron_impl_2func_651"

Descargue la primera función Eventing con todos los ajustes necesarios, haga clic con el botón derecho del ratón en el siguiente enlace y seleccione Guardar enlace como para descargar el archivo cron_impl_2func_651.json en tu sistema de archivos local.

Desde el Consola Web de Couchbase > Eventos haga clic en IMPORTARnavega hasta el archivo cron_impl_2func_651.jsonselecciónelo y ábralo. La página AÑADIR FUNCIÓN aparece el cuadro de diálogo.

En el AÑADIR FUNCIÓN para cada uno de los elementos de la función, facilite la siguiente información. Tenga en cuenta el archivo JSON cron_impl_2func_651.json preconfigurará todos los ajustes correctamente para este ejemplo:

Para el Cubo de origen y compruebe que está en crondata.
Para el Cubo de metadatos y compruebe que está en metadatos.
Compruebe que cron_impl_2func_651 es el nombre de la función que está creando en el campo Nombre de la función cuadro de texto.
[Paso opcional] Introduzca el texto Un programador cron parte 1en el Descripción cuadro de texto.
Para el Ajustes utilice los valores por defecto.
Para el Fijaciones verifique que existen dos enlaces.
Para la vinculación, el "alias de cubo", especifica cron_bkt como "nombre alias" del cubo, y seleccione
crondata como cubo asociado, y el modo debe ser "lectura y escritura".
Para la vinculación, el "alias de cubo", especifica ts_bkt como "nombre alias" del cubo, y seleccione
viaje-muestra como cubo asociado, y el modo debe ser "lectura y escritura".
La configuración en el cuadro de diálogo debe ser como la siguiente:
Después de verificar toda la información requerida en el cuadro de diálogo AÑADIR FUNCIÓN, haga clic en Siguiente: Añadir Código. Aparecerá la página cron_impl_2func_651 (con el código JavaScript precargado).
Para volver a la pantalla de Eventos, pulse el botón '< volver a Concurso completo(debajo del editor) o haga clic en el enlace Eventos ficha.

Importar la 2ª Función "cron_impl_2func_651_help"

Descargue la segunda función Eventing con todos los ajustes necesarios, haga clic con el botón derecho del ratón en el siguiente enlace y seleccione Guardar enlace como para descargar el archivo cron_impl_2func_651_help.json en tu sistema de archivos local.

Desde el Consola Web de Couchbase > Eventos haga clic en IMPORTARnavega hasta el archivo cron_impl_2func_651_help.jsonselecciónelo y ábralo. La página AÑADIR FUNCIÓN aparece el cuadro de diálogo.

En el AÑADIR FUNCIÓN para cada uno de los elementos de la función, facilite la siguiente información. Tenga en cuenta el archivo JSON cron_impl_2func_651_help.json preconfigurará todos los ajustes correctamente para este ejemplo:

Para el Cubo de origen y compruebe que está en crondata.
Para el Cubo de metadatos y compruebe que está en metadatos.
Compruebe que cron_impl_2func_651_help es el nombre de la función que está creando en el campo Nombre de la función cuadro de texto.
[Paso opcional] Introduzca el texto Un ayudante cron como programador parte 1en el Descripción cuadro de texto.
Para el Ajustes utilice los valores por defecto.
Para el Fijaciones verifique que sólo existe un enlace.
Para la vinculación, el "alias de cubo", especifica cron_bkt como "nombre alias" del cubo, y seleccione
crondata como cubo asociado, y el modo debe ser "lectura y escritura".
La configuración en el cuadro de diálogo debe ser como la siguiente:
Después de verificar toda la información requerida en el cuadro de diálogo AÑADIR FUNCIÓN, haga clic en Siguiente: Añadir Código. Aparecerá la página cron_impl_2func_651_help (con el código JavaScript precargado).
Para volver a la pantalla de Eventos, pulse el botón '< volver a Concurso completo(debajo del editor) o haga clic en el enlace Eventos ficha.

Método #2 Añadir función manualmente, cortar y pegar JavaScript

Crear manualmente "cron_impl_2func_651"

Para añadir la primera función de Eventing desde Consola Web de Couchbase > Eventos haga clic en AÑADIR FUNCIÓNpara añadir una nueva función. La dirección AÑADIR FUNCIÓN aparece el cuadro de diálogo.

En el AÑADIR FUNCIÓN para cada elemento de la función, facilite la siguiente información:

Para el Cubo de origen seleccione crondata.
Para el Cubo de metadatos seleccione metadatos.
Visite cron_impl_2func_651 es el nombre de la función que está creando en el campo Nombre de la función cuadro de texto.
[Paso opcional] Introduzca el texto Un programador cron parte 1en el Descripción cuadro de texto.
Para el Ajustes utilice los valores por defecto.
Para el Fijaciones crear dos enlaces:
Para la vinculación, el "alias de cubo", especifica cron_bkt como "nombre alias" del cubo, y seleccione
crondata como cubo asociado, y el modo debe ser "lectura y escritura".
Para la vinculación, el "alias de cubo", especifica ts_bkt como "nombre alias" del cubo, y seleccione
viaje-muestra como cubo asociado, y el modo debe ser "lectura y escritura".
Una vez configurados los ajustes, el cuadro de diálogo debería tener este aspecto:
Tras facilitar toda la información requerida en el AÑADIR FUNCIÓN pulse Siguiente Añadir código. En cron_impl_2func_651 aparece el cuadro de diálogo. La página cron_impl_2func_651 contiene inicialmente un bloque de código. Usted sustituirá su cron_impl_2func_651 en este bloque.

Copie el siguiente código fuente JavaScript de Eventing Function (618 líneas) y péguelo en el bloque de código del marcador de posición de cron_impl_2func_651

/* 
Function "cron_impl_2func_651" also requires "cron_impl_2func_651_help"

Create a basic cron system using Eventing allows a recurring function to execute activity at a 
specified time every day, hour, min, 30 sec., and 15 sec.  We use a bucket called 'crondata' 
aliased to 'cron_bkt' which can hold one or more control documents of type = "recurring_event".

The following uses of timers do not work reliably in Couchbase versions 6.5 and 6.5.1
  a) scheduling an Eventing timer within a timer's callback 
  b) overwriting an existing timer by id
  
In addition the ability to cancel a timer does not exist in Couchbase versions 6.5 and 6.5.1
  
For this example, we supply one real user function that builds a recurring 'static' cache document
from bucket `travel-sample` via an N1QL query and save the result back to `travel-sample` via
the alais 'ts_bkt'.  This JavaScript function is doCronActionA(), we also provide two placeholders
doCronActionB() and doCronActionC() for additional experimentation.

Test Doc:
   {
        "type":"recurring_event",   // The KEY will be <<type>>::<<id>>
        "id":1,                     //
        "hour":14,                  // The hour of the day 0-23, *, *2X, *4X to trigger
        "min":54,                   // The minute in the hour 0-59, *, *2X, *4X to trigger
        "action":"doCronActionA",   // What function to run on the trigger
        "active":false,             // Flag to arm or disable this schedule
        "verbose" : {
          "user_func":2,            // Logging level for the action logic : 0=none, etc. etc.
          "scheduler":3             // Logging level for the cron logic   : 0=none, etc. etc.
        },
        "dynamic" : {
          "state":"arm",            // States "arm"|"rearm"|"pending" if any value but "pending" start a schedule
          "next_sched": 0,          // Number of seconds since epoch to next desired schedule
          "prev_sched": 0,          // Number of seconds since epoch for previous schedule
          "prev_etime": 0,          // Number of seconds since epoch for previous schedule actual exec time
          "prev_delay": 0,          // Number of seconds that the timer was delayed from the schedule
          "prev_atime": 0           // Number of seconds taken by the user 'action'
        }
    }
    
    INSERT INTO `crondata` (KEY,VALUE) VALUES ("recurring_event::1", 
    {
        "type":"recurring_event",
        "id":1,
        "hour":14,
        "min":54,
        "action":"doCronActionA",
    	"verbose" : {
          "user_func":2,
          "scheduler":3
        },
        "active":false,
	    "dynamic" : {
          "state": "arm",
          "next_sched": 0,
          "prev_sched": 0,
          "prev_etime": 0,
          "prev_delay": 0,
          "prev_atime": 0
	    }
    } 
    );

Note, you can omit verbose{} and dynamic{} as they will be auto-created by this main Eventing 
Function "cron_impl_2func_651". If verbose{} is missing the logging levels will default to 
verbose" : {  "user_func":1, "scheduler":1 }

    INSERT INTO `crondata` (KEY,VALUE) VALUES ("recurring_event::1", 
    {
        "type":"recurring_event",
        "id":1,
        "hour":14,
        "min":54,
        "action":"doCronActionA",
        "active":false
    } 
    );

N1QL : Make an index to query data without specifying keys
    CREATE primary INDEX on `crondata` ;

N1QL : Verify or inspect settings in schedule	        
    SELECT * FROM `crondata` WHERE type="recurring_event";

N1QL : Arm or set active	    
    UPDATE `crondata` SET active = true WHERE type="recurring_event" AND id=1 ;
    
N1QL : Disarm or set inactive	
    UPDATE `crondata` SET active = false WHERE type="recurring_event" AND id=1 ;
    
N1QL : Adjust time of trigger	
    UPDATE `crondata` SET hour = 11, min = 30 WHERE type="recurring_event" AND id=1 ;

N1QL : Adjust logging	        
    UPDATE `crondata` SET verbose.user_func = 1,  verbose.scheduler = 0 WHERE type="recurring_event" AND id=1 ;

N1QL : Delete the schedule	    
    DELETE FROM `crondata` WHERE type="recurring_event" AND id=1 ;
         
The action field is important it 'should' exist in this Eventing Function note it could be any 
JavaScript name e.g. MyFunc and you must implement like the example doCronActionA(doc) where
doc will be the currently active item of type = 'recurring_event' read from the alias bucket
‘cron_bkt’ when the timer is fired.  The action JavaScript function should return  either true
or false used for logging purposes.  If the action does not exist it is an error and a warning
is logged and the timer is disabled.

In Couchbase version 6.5+ to add a new cron like daily function just pause the active handler 
insert your new function doCronActionB(doc) {...} then Resume the eventing handler.  The nice 
thing is if a timer was to be fired will the function was paused it will NOT be lost, when you 
resume the function it will be processed at the next available time slot.  

Any change to a control structure will create a new recurring schedule or timer and cancel the 
current previous schedule this includes changing the verbosity level. The previous timer will 
continue to run however when executed it will do a Checksum on the current control structure 
from KV against it’s passed context and if different the Callback will ignore the old schedule.
This logic could be altered to process immediately if the schedule has expired search for the 
string "OnUpdate U" in the code below.
*/

// ==================
/* BEG USER FUNCTIONS TO RUN ONCE A DAY, HOUR, OR MINUTE - ANYTHING YOU WANT BELOW */
function doCronActionA(doc) {
  try {
    // Check that doc has desired values
    if (!doc.type || doc.type !== "recurring_event" || !doc.active || doc.active !== true) return;
    if (doc.verbose.user_func >= 1)
      log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

    // this is a 6.5 N1QL query (feature not available in GA prior to 6.5)
    // Create an embedded N1QL iterator by issuing a SELECT statement to get the
    // counts of airlines by country.  Make a new document and write it out to KV 

    // We will use the iterator to create a KV document representing the results of a
    // HARD lengthy embedded N1QL query and write it back to KV, the idea is to keep
    // a calculation up to date once a day such that it that can be read 'quickly' 
    // by other Eventing Functions, other Couchbase services or SDKs.   

    // Consider if we had 1 million docs in a minute do we really want to use N1QL
    // to recalculate something that is almost static for all 1 million documents, of 
    // course not, so we make an intermediate value that can be read into Eventing
    // and used via a single 'light weight' KV read.

    var q_iter = SELECT country,
      count( * ) cnt
    FROM `travel-sample`
    WHERE `type` = 'airline'
    GROUP BY country;

    // loop through the result set and update the map 'accumulate'
    var accumulate = {};
    var idx = 0;
    for (var val of q_iter) {
      if (doc.verbose.user_func >= 2)
        log(doc.action + ' N1QL idx ' + idx + ', country ' + val.country + " cnt " + val.cnt);
      accumulate[val.country] = val.cnt;
      idx++;
    }
    // close out embedded N1QL iterator
    q_iter.close();

    // Now let’s make a cached KV document representing a HARD length embedded N1QL
    // query and write it back to KV, we need a KEY and a type and id and then we 
    // upsert it into the `travel-sample` bucket.

    var cachedoc = {};
    cachedoc.type = "cron_cache";
    cachedoc.id = "airlines_by_country";
    cachedoc.date = new Date();
    cachedoc.data = accumulate;
    var ckey = cachedoc.type + '::' + cachedoc.id;
    ts_bkt[ckey] = cachedoc;
    if (doc.verbose.user_func >= 2) {
      log(doc.action + ' upsert to KV with KEY ' + ckey + ' cachedoc ', cachedoc);
    }
  } catch (e) {
    log(doc.action + ' Error exception:', e);
    return false;
  }
  return true;
}

function doCronActionB(doc) {
  try {
    // check that doc has desired values
    if (doc.type !== "recurring_event" || doc.active !== true) return;
    if (doc.verbose.user_func >= 1)
      log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

    // YOUR LOGIC HERE

  } catch (e) {
    log(doc.action + ' Error exception:', e);
    return false;
  }
  return true;
}

function doCronActionC(doc) {
  try {
    // check that doc has desired values
    if (doc.type !== "recurring_event" || doc.active !== true) return;
    if (doc.verbose.user_func >= 1)
      log(doc.action + ' user action controlled by ' + doc.type + '::' + doc.id);

    // YOUR LOGIC HERE

  } catch (e) {
    log(doc.action + ' Error exception:', e);
    return false;
  }
  return true;
}

/* END USER FUNCTIONS TO RUN ONCE A DAY, HOUR, OR MINUTE - ANYTHING YOU WANT ABOVE */
// ==================


// FIXUP: ADDIN FUNCTON
function noopTimer(context) {
    // fix for 6.5.X growing bucket ops
    try {
        if (context.type === "_tmp_vbs" && context.vb === 0) { 
            // log("noopTimer timers firing, printing only for vBucket 0");
        }
    } catch (e) {
        log("OnUpdate Exception in callback noopTimer:", e);
    }
}

// FIXUP: ADDIN FUNCTON
function rearmTimer(context) {
    // fix for 6.5.X growing bucket ops
    try {
        if (context.type === "_tmp_vbs" && context.vb === 0) { 
            // Update/touch all docs in the helper_bucket the helper function will then
            // mutate all 1024 of type == vbs_seed (64 on MacOS) to create a recuring cycle.
            // log("noopTimer timer fired all 1024 vBuckets, logging only vb 0", context);
            
            // generate a mutation to re-arm the HELPER function: fix_scan_issue
            // which will in turn make new mutations for this Function
            var cur = cron_bkt[context.key];
            if (cur && cur.ts_millis === context.ts_millis) {
                // log("rearmTimer update fix_timer_scan_issue::1 in helper_bucket alias only for vBucket 0");
                var now = new Date();
                cron_bkt["fix_timer_scan_issue::1"] = { "last_update": now };
            } else {
                // NOOP we had multiple timer cycles, just let this one quietly stop.
            }
        }
    } catch (e) {
        log("OnUpdate Exception in callback rearmTimer:", e);
    }
}

// FIXUP: ADDIN FUNCTON
function genNoopTimers(doc, meta, seconds) {
    // fix for 6.5.X growing bucket ops
    try {
        // redundant but play it safe
        if (doc.type === "_tmp_vbs") {
            // Since we are using an different function a timer on all our vBuckets do immeadiately (can take up to 15 seconds)
            // If we used cross bucket recursion to rearm all the timers in a recurring fashion we would add a delay of at least 40 seconds.
            createTimer(noopTimer, new Date(), null, doc);
            if (doc.vb === 0) { 
                // Update/touch all docs in the helper_bucket the helper function will then
                // mutate all 1024 of type == vbs_seed (64 on MacOS) to create a recuring cycle.
                // log("noopTimer timer fired all 1024 vBuckets, logging only vb 0", context);
            
                // generate a mutation to re-arm the HELPER function: fix_scan_issue
                // which will in turn make new mutations for this Function
                
                // log("genNoopTimers make timer to rearm fix_timer_scan_issue::1");
                createTimer(rearmTimer, new Date(new Date().getTime() + seconds * 1000), null, doc);
            }
        }
    } catch (e) {
        log("OnUpdate Exception in genNoopTimers:", e);
    }
}

function OnUpdate(doc, meta) {
  // fix for 6.5.X growing bucket ops
  if (doc.type === "_tmp_vbs") genNoopTimers(doc, meta, 30);
  if (!cron_bkt["fix_timer_scan_issue::1"]) {
      cron_bkt["fix_timer_scan_issue::1"] = {};
  }
    
  try {
    // Check if further analysis is needed we only trigger on an active recurring_event 
    if (doc.type !== "recurring_event" || doc.active !== true) return;

    var update_doc = false;
    if (!doc.dynamic) {
      // Add if missing doc.dynamic with defaults
      doc.dynamic = {
        "state": "arm",
        "next_sched": 0,
        "prev_sched": 0,
        "prev_etime": 0,
        "prev_delay": 0,
        "prev_atime": 0
      };
      // we need to update the document once we have the next schedule
      update_doc = true;
    }
    if (!doc.verbose) {
      // Add if missing doc.dynamic with defaults
      doc.verbose = {
        "user_func": 1,
        "scheduler": 1
      };
      // we need to update the document once we have the next schedule
      update_doc = true;
    }
    // Do not process dynamic.state pending
    if (!doc.dynamic || !doc.dynamic.state || doc.dynamic.state === "pending") return;

    var mid = doc.type + "::" + doc.id; // this is the same as meta.id or the KEY
    var hour = doc.hour;
    var min = doc.min;

    // Do an eval check the JavaScript function exists. The eval occurs in a common 
    // utility function shared with RecurringCallback
    if (!verifyFunctionExistsViaEval(doc, mid)) {
      // doc.action did not exist, we have already logged the issue
      return;
    }

    // Get the next valid execution time
    var date_timer = getNextRecurringDate(hour, min);
    var next_sched = Math.round(date_timer.getTime() / 1000);
    if (!update_doc && next_sched !== doc.dynamic.next_sched) {
      // the next_sched should be the same as the setting from the helper application, however
      // if we undeploy/deploy or pause/resume we might haver to reschedule to the next time slot
      log('OnUpdate U ' + mid + ' calculated next_sched !== doc.dynamic.next_sched, delta ' +
        (next_sched - doc.dynamic.next_sched) + ', reschedule');
      update_doc = true;
    }

    if (update_doc) {
      // this mutation is recursive and will be suppressed, we ensure we have a dynamic structure
      doc.dynamic.next_sched = next_sched;

      // rather then the call a function, to trap and retry if there is a resource issue
      // cron_bkt[mid] = doc;
      if (!tryBucketKvWriteWithLog('OnUpdate F', mid, doc)) {
        // Failed to write doc to cron_bkt[key] the error has been logged
        // and there is nothing more we can do.
        return;
      }
    }

    // Schedule an Eventing timer
    var timer_id = createTimer(Callback, date_timer, null, doc);
    if (doc.verbose.scheduler >= 1) {
      log('OnUpdate A ' + mid + ' rcv mutation (initial or rearm) schedule timer at ' +
        toLocalISOTime(date_timer));
    }
    if (doc.verbose.scheduler >= 2) {
      log('OnUpdate B ' + mid + ' recurring timer was created, timer_id ' + timer_id);
    }
  } catch (e) {
    log('OnUpdate E ' + meta.id + ', Error exception:', e);
  }
}

function getNextRecurringDate(hour_str, min_str) {
  // Note Javascript Dates are in milliseconds
  var date_now = new Date();
  var date_ret = new Date();
  var hour;
  var min;

  try {
    hour = parseInt(hour_str);
  } catch (e) {}
  try {
    min = parseInt(min_str);
  } catch (e) {}

  // Note, this is only a simplistic partial 'crontab' syntax with some slight extensions
  // it allows once a day, once an hour, once a minute.  It also contains some non-standard 
  // syntax to provide the ability to execute twice a minute or four times a minute.

  if (hour_str === '*4X' && min_str === '*4X') {
    // once every 15 seconds or four times a minute
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(15);
    while (date_ret.getTime() < date_now.getTime()) {
      date_ret.setSeconds(date_ret.getSeconds() + 15);
    }
    return date_ret;
  } else
  if (hour_str === '*2X' && min_str === '*2X') {
    // once every 30 seconds or twice a minute
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(30);
    while (date_ret.getTime() < date_now.getTime()) {
      date_ret.setSeconds(date_ret.getSeconds() + 30);
    }
    return date_ret;
  } else
  if (hour_str === '*' && min_str === '*') {
    // once a minute 
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(date_ret.getMinutes() + 1);
  } else
  if (hour_str !== '*' && isNaN(hour) === false && min_str === '*') {
    // once a minute only for a given hour
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(date_ret.getMinutes() + 1);
    if (date_ret.getTime() < date_now.getTime()) {
      date_ret.setHours(hour);
    }
    if (date_ret.getTime() > date_now.getTime()) {
      date_ret.setDate(date_ret.getDate() + 1);
      date_ret.setSeconds(0);
      date_ret.setMinutes(0);
      date_ret.setHours(hour);
    }
  } else
  if (hour_str === '*' && min_str !== '*' && isNaN(min) === false) {
    // once a hour at a given minute
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(min);
    // schedule for next hour
    date_ret.setHours(date_ret.getHours() + 1);
  } else
  if (isNaN(hour) === false && isNaN(min) === false) {
    // once a day for a given hour and a given minute 
    date_ret.setMilliseconds(0);
    date_ret.setSeconds(0);
    date_ret.setMinutes(min);
    date_ret.setHours(hour);
    if (date_ret.getTime() < date_now.getTime()) {
      // schedule for tomorrow
      date_ret.setDate(date_ret.getDate() + 1);
    }
  } else {
    log('getNextRecurringDate illegal input hour_str <' +
      hour_str + '> min_str <' + min_str + '>');
    throw new Error('getNextRecurringDate illegal input hour_str <' +
      hour_str + '> min_str <' + min_str + '>');
    return null;
  }
  return date_ret;
}

function verifyFunctionExistsViaEval(curDoc, id) {
  var result = false;
  try {
    // check for function if missing this is invalid return result
    result = eval("typeof " + curDoc.action + " === 'function';");
    if (result === false) {
      if (curDoc.verbose.scheduler >= 1)
        log("Warn/Disable (No Action and No Re-Arm), because required 'action' of " +
          curDoc.action + "(doc) does not exist, id is", id);
      return result;
    }
  } catch (e) {
    log('verifyFunctionExistsViaEval Error exception:', e);
  }
  return result;
}

function toNumericFixed(number, precision) {
  var multi = Math.pow(10, precision);
  return Math.round((number * multi).toFixed(precision + 1)) / multi;
}

function toLocalISOTime(d) {
  var tzoffset = (new Date()).getTimezoneOffset() * 60000; //offset in milliseconds
  return (new Date(d.getTime() - tzoffset)).toISOString().slice(0, -1);
}

function tryBucketKvWriteWithLog(tag, key, doc) {
  var success = false;
  var tries = 0;
  while (tries < 10) {
    tries++;
    try {
      // critical that the below succeeds, because if it doesn't the cron cycle will break
      cron_bkt[key] = doc;
      success = true;
      break;
    } catch (e) {
      log(tag + ' ' + key + ' WARN failed to update KV tries ' + tries, e);
    }
  }
  if (!success) {
    log(tag + ' ' + +key + ' FATAL could not update KV cron cycle, tried ' + tries + ', stoping ' + curDoc.action);
  }
  return success;
}

function Callback(doc) {
  try {
    var fired_at = new Date();

    // Check if further analysis is needed we only trigger on a recurring_event that is active
    if (doc.type !== "recurring_event") return;
    // doc must have 'action', 'dynamic {}', verbose {}, dynamic.state
    if (!doc.action || !doc.dynamic || !doc.verbose || !doc.dynamic.state) return;
    // process any doc.dynamic.state BUT pending 
    if (doc.dynamic.state === "pending") return;

    // ==================
    // Check if still active

    // We make sure that in KV the 'doc' still exists and that it is still active if not just 
    // return thus skipping the action and not Re-arming the timer. Note `travel-sample` is 
    // aliased to the map 'cron_bkt

    var mid = doc.type + '::' + doc.id; // make our KEY
    var curDoc = null;
    try {
      // read the current version of doc from KV, e.g. curDoc
      curDoc = cron_bkt[mid];
    } catch (e) {} // needed for pre 6.5, note pure 6.5+ deployment returns null sans exception

    var reason = null;
    if (!curDoc || curDoc === null) {
      reason = "cron document is missing";
    } else
    if (!curDoc.active) {
      reason = "cron document has active = false";
    } else
    if (!curDoc.dynamic.state || curDoc.dynamic.state !== doc.dynamic.state) {
      reason = "cron document wrong dynamic.state expected " + doc.dynamic.state;
    } else
    if (crc64(doc) !== crc64(curDoc)) {
      reason = "cron document changed";
    }

    if (reason !== null) {
      if (!curDoc || curDoc === null || curDoc.verbose.scheduler >= 1) {
        log('Callback X ' + mid + " ignore/stop this timer's schedule because " + reason);
      }
      if (!curDoc || curDoc === null || curDoc.verbose.scheduler >= 4) {
        log('Callback Y ' + mid + ' timer doc', doc);
        log('Callback Z ' + mid + ' KV curDoc', curDoc);
      }
      return;
    }

    // ==================
    // Verify user routine exists and if so eval it 

    // Assume curDoc.action contains something like "doCronActionA" and we have a function in 
    // this handler like "doCronActionA(doc)". Below we use curDoc as the end user should be 
    // able to alter the eval'd JavaScript function.  We will execute two (2) evals.

    // First eval check the JavaScript function exists.  The eval occurs in a common 
    // utility function shared with RecurringCallback
    if (!verifyFunctionExistsViaEval(curDoc, mid)) {
      // curDoc.action did not exist, we have already logged the issue
      return;
    }

    // Second eval execute and process the user function we execute the defined function 
    // with an argument of curDoc
    var beg_act = new Date();
    var result = null;
    eval("result = " + curDoc.action + "(curDoc);");
    var end_act = new Date();
    var atime_ms = end_act.getTime() - beg_act.getTime();

    if (curDoc.verbose.scheduler >= 2)
      log('Callback R ' + mid + ' action took ' + toNumericFixed((atime_ms / 1000), 3) +
        ' sec., returned ' + result);

    // ==================
    // Calculate next time and mutate the control document for our our helper function
    // which will create another mutation such that OnUpdate of this function will pick
    // it up and generate the timer (avoids the MB-38554 issue).

    var hour = curDoc.hour;
    var min = curDoc.min;
    var date_timer = getNextRecurringDate(hour, min);

    curDoc.dynamic.prev_delay =
      toNumericFixed(((fired_at.getTime() / 1000) - curDoc.dynamic.next_sched), 3);
    curDoc.dynamic.prev_sched = curDoc.dynamic.next_sched;
    curDoc.dynamic.prev_etime = Math.round(fired_at.getTime() / 1000);
    curDoc.dynamic.prev_atime = toNumericFixed((atime_ms / 1000), 3);

    curDoc.dynamic.state = "pending";
    curDoc.dynamic.next_sched = Math.round(date_timer.getTime() / 1000);

    // rather then the call a function, to trap and retry if there is a resource issue
    // cron_bkt[mid] = curDoc;
    if (!tryBucketKvWriteWithLog('Callback F', mid, curDoc)) {
      // Failed to write curDoc to cron_bkt[key] the error has been logged
      // and there is nothing more we can do.
      return;
    }

    if (curDoc.verbose.scheduler >= 1) {
      log('Callback A ' + mid + ' gen mutation #1 to doc to force schedule rearm at ' +
        toLocalISOTime(date_timer));
    }
    if (curDoc.verbose.scheduler >= 2) {
      log('Callback B ' + mid + ' sched ' + curDoc.dynamic.prev_sched +
        ', actual ' + curDoc.dynamic.prev_etime +
        ', delay ' + curDoc.dynamic.prev_delay +
        ', took ' + curDoc.dynamic.prev_atime);
    }
    if (curDoc.verbose.scheduler >= 3) {
      log('Callback C ' + mid + ' curDoc', curDoc);
    }
  } catch (e) {
    var mid = doc.type + '::' + doc.id; // make our KEY
    log('Callback E ' + mid + ' Error exception:', e);
  }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

Function "cron_impl_2func_651" also requires "cron_impl_2func_651_help"

Create a basic cron system using Eventing allows a recurring function to execute activity at a

specified time every day, hour, min, 30 sec., and 15 sec. We use a bucket called 'crondata'

aliased to 'cron_bkt' which can hold one or more control documents of type = "recurring_event".

The following uses of timers do not work reliably in Couchbase versions 6.5 and 6.5.1

a) scheduling an Eventing timer within a timer's callback

b) overwriting an existing timer by id

In addition the ability to cancel a timer does not exist in Couchbase versions 6.5 and 6.5.1

For this example, we supply one real user function that builds a recurring 'static' cache document

from bucket `travel-sample` via an N1QL query and save the result back to `travel-sample` via

the alais 'ts_bkt'. This JavaScript function is doCronActionA(), we also provide two placeholders

doCronActionB() and doCronActionC() for additional experimentation.

Test Doc:

{

"type":"recurring_event", // The KEY will be <<type>>::<<id>>

"id":1, //

"hour":14, // The hour of the day 0-23, *, *2X, *4X to trigger

"min":54, // The minute in the hour 0-59, *, *2X, *4X to trigger

"action":"doCronActionA", // What function to run on the trigger

"active":false, // Flag to arm or disable this schedule

"verbose" : {

"user_func":2, // Logging level for the action logic : 0=none, etc. etc.

"scheduler":3 // Logging level for the cron logic : 0=none, etc. etc.

"dynamic" : {

"state":"arm", // States "arm"|"rearm"|"pending" if any value but "pending" start a schedule

"next_sched": 0, // Number of seconds since epoch to next desired schedule

"prev_sched": 0, // Number of seconds since epoch for previous schedule

"prev_etime": 0, // Number of seconds since epoch for previous schedule actual exec time

"prev_delay": 0, // Number of seconds that the timer was delayed from the schedule

"prev_atime": 0 // Number of seconds taken by the user 'action'

}

INSERT INTO `crondata` (KEY,VALUE) VALUES ("recurring_event::1",

{

"type":"recurring_event",

"id":1,

"hour":14,

"min":54,

"action":"doCronActionA",

"verbose" : {

"user_func":2,

"scheduler":3

"active":false,

"dynamic" : {

"state": "arm",

"next_sched": 0,

"prev_sched": 0,

"prev_etime": 0,

"prev_delay": 0,

"prev_atime": 0

}

);

Note, you can omit verbose{} and dynamic{} as they will be auto-created by this main Eventing

Function "cron_impl_2func_651". If verbose{} is missing the logging levels will default to

verbose" : { "user_func":1, "scheduler":1 }

INSERT INTO `crondata` (KEY,VALUE) VALUES ("recurring_event::1",

{

"type":"recurring_event",

"id":1,

"hour":14,

"min":54,

"action":"doCronActionA",

"active":false

}

);

N1QL : Make an index to query data without specifying keys

CREATE primary INDEX on `crondata` ;

N1QL : Verify or inspect settings in schedule

SELECT * FROM `crondata` WHERE type="recurring_event";

N1QL : Arm or set active

UPDATE `crondata` SET active = true WHERE type="recurring_event" AND id=1 ;

N1QL : Disarm or set inactive

UPDATE `crondata` SET active = false WHERE type="recurring_event" AND id=1 ;

N1QL : Adjust time of trigger

UPDATE `crondata` SET hour = 11, min = 30 WHERE type="recurring_event" AND id=1 ;

N1QL : Adjust logging

UPDATE `crondata` SET verbose.user_func = 1, verbose.scheduler = 0 WHERE type="recurring_event" AND id=1 ;

N1QL : Delete the schedule

DELETE FROM `crondata` WHERE type="recurring_event" AND id=1 ;

The action field is important it 'should' exist in this Eventing Function note it could be any

JavaScript name e.g. MyFunc and you must implement like the example doCronActionA(doc) where

doc will be the currently active item of type = 'recurring_event' read from the alias bucket

‘cron_bkt’ when the timer is fired. The action JavaScript function should return either true

or false used for logging purposes. If the action does not exist it is an error and a warning

is logged and the timer is disabled.

In Couchbase version 6.5+ to add a new cron like daily function just pause the active handler

insert your new function doCronActionB(doc) {...} then Resume the eventing handler. The nice

thing is if a timer was to be fired will the function was paused it will NOT be lost, when you

resume the function it will be processed at the next available time slot.

Any change to a control structure will create a new recurring schedule or timer and cancel the

current previous schedule this includes changing the verbosity level. The previous timer will

continue to run however when executed it will do a Checksum on the current control structure

from KV against it’s passed context and if different the Callback will ignore the old schedule.

This logic could be altered to process immediately if the schedule has expired search for the

string "OnUpdate U" in the code below.

// ==================

/* BEG USER FUNCTIONS TO RUN ONCE A DAY, HOUR, OR MINUTE - ANYTHING YOU WANT BELOW */