Implementing a Scheduler Via Couchbase Eventing (Part 1)

Esta é a primeira de uma série de várias partes para aproveitar o Serviço de eventos do Couchbase para executar várias tarefas agendadas em intervalos recorrentes específicos em um cron completamente dentro do banco de dados, sem necessidade de infraestrutura adicional, por meio de uma única Eventing Function de uso geral.

Nesta parte, vamos nos concentrar na execução de rotinas de usuário fixas, funções JavaScript definidas dentro de uma Eventing Function.

Mais adiante, em artigos subsequentes, estenderemos o cron como Eventing Function para agendar e executar instruções N1QL dinâmicas orientadas por banco de dados e, por fim, exploraremos o agendamento de funções JavaScript dinâmicas orientadas por banco de dados.

Histórico

O Couchbase Eventing Service fornece uma estrutura para escrever suas próprias rotinas, funções JavaScript simples, para processar alterações em documentos. Esse serviço fornece toda a infraestrutura necessária para criar funções escalonáveis e robustas baseadas na nuvem, permitindo que você se concentre no desenvolvimento de lógica comercial pura para interagir quase em tempo real com as alterações nos dados. Suas funções podem acessar o serviço de dados do Couchbase (KV), o serviço de consulta do Couchbase (N1QL) e os pontos de extremidade REST externos ao sistema Couchbase.

O JSON modelo de dados no Couchbase veio de JavaScriptPor isso, é natural que o Eventing Service exponha a capacidade de escrever código JavaScript para analisar e manipular documentos JSON em qualquer tipo de evento de alteração, incluindo inserções, atualizações, mesclagens e exclusões (denominadas mutações).

As Eventing Functions normalmente permitem que você implemente e execute fragmentos de código personalizados que reagem a milhares e até milhões de mutações por segundo em seus documentos. Vários casos de uso típicos são documentado para desenvolver Eventing Functions de alta velocidade e em escala que respondem a mutações de documentos do Couchbase.

Em vez disso, este artigo se concentrará em um caso de uso de velocidade muito baixa do Eventing Service, criando um crontab distribuído confiável "no banco de dados", permitindo que você execute funções JavaScript que interagem com os serviços do Couchbase em uma programação periódica regular.

Agendamento da lógica de negócios para execução em uma data ou hora específica

Cron, com o nome de "CronosA palavra grega "tempo" é um dos utilitários mais úteis em um sistema Linux. No Linux, o cron é acionado por um arquivo crontab (tabela cron), um arquivo de configuração que especifica os comandos do shell a serem executados periodicamente em uma determinada programação.

Uma desvantagem da execução cron é que ele não foi projetado para ser um serviço distribuído; ele é executado em uma única caixa e, portanto, apresenta um único ponto de falha. Se o sistema ficar off-line por várias horas, todas as tarefas agendadas serão perdidas.

Sim, há alguns distribuídos cron implementações como o Cloud Service do Google, as tarefas agendadas do AWS e o Azure Functions / Time Triggers. Mas as ofertas de cada fornecedor de nuvem têm seus próprios idiomas e não são diretamente portáteis.

Além disso, a metodologia de configuração e controle precisa ser protegida, por exemplo, se você controlar um sistema distribuído cron por meio de uma API REST sobre HTTP/S, você precisa levar isso em conta no seu plano de segurança.

Usar o próprio Couchbase para executar comandos periódicos

Com uma pequena quantidade de código e planejamento, você pode aproveitar o serviço Eventing do Couchbase para fornecer cron para suas operações ou manutenções programadas no banco de dados. A criação do agendador no banco de dados permite que você obtenha os seguintes benefícios:

Portabilidade entre provedores de nuvem: se você hospedar novamente o cluster do Couchbase, seu agendador não será afetado.
Capacidade de suporte: se você utilizar o Couchbase, terá um único fornecedor para fornecer suporte e outros serviços.
Distribuído, sem ponto único de falha e todos os serviços do Couchbase suportam réplicas distribuídas.
Execução garantida: sua tarefa é executada mesmo após a recuperação de uma falha de nó.

Programação do Couchbase, temporizadores, o molho secreto

Os temporizadores são construções do Couchbase Eventing Service por meio das quais os desenvolvedores podem especificar uma rotina (lógica de negócios) a ser acionada em um momento futuro. Usaremos essa funcionalidade para implementar um serviço puramente configurável do Couchbase crontab que lhe permite acionar tarefas repetitivas como parte de seus fluxos de trabalho, seja para executar uma simples consulta N1QL ou para criar um mecanismo de regras complexo.

Em todos os projetos subsequentes, limitaremos nossa cron implementações em uma resolução de 15 segundos ou mais. Temos essa limitação porque, embora os temporizadores sejam dimensionados para milhões e tenham garantia de disparo e execução, eles não têm precisão de relógio de parede e atualmente têm um atraso de estado estável limitado a menos de 14 segundos ^[1].

É claro que, se você precisar de um cronograma mais apertado, ou seja, menos de 15 segundos, deverá simplesmente processar a própria mutação na lógica Eventing sem usar uma construção de cronômetro para agendar uma chamada de volta no futuro.

No momento em que este texto foi escrito, a versão atual do Couchbase é a 6.5.1, que duas limitações que precisamos contornar ao criar um cron sistema.

Nas versões 5.5.x, 6.0.x e 6.5.x, uma função que é invocada por um retorno de chamada do temporizador não pode criar um novo temporizador de forma confiável (uma solução alternativa no espaço do usuário pode ser feita por meio de uma segunda função cooperativa).
Nas versões 6.5.x, a criação de temporizadores no futuro (como em uma hora ou mais) em um sistema ocioso pode resultar em um número crescente de operações de balde de metadados que podem, eventualmente, bloquear mutações para uma determinada função de evento (na versão 6.5.X, uma solução alternativa no espaço do usuário pode ser realizada por meio de uma segunda função cooperativa). A gravidade é regida por:
- O número de vBuckets que mantêm um timer ativo. Portanto, se houver apenas alguns cronômetros no futuro, o problema poderá não ser percebido ou se materializar. Esse é o caso de apenas algumas programações do cron, mas, para completar, caso você adicione a funcionalidade de data, coloquei uma correção para esse problema no código fornecido neste artigo.
- Se um timer de Eventing foi disparado recentemente em um vBucket (o que elimina o problema para o vBucket em questão com base em cada função). Portanto, os sistemas com muita atividade de temporizador de curto prazo não terão esse problema, mesmo que os temporizadores estejam programados para um futuro distante.

Felizmente, na versão 6.6.0, os dois problemas ou restrições acima foram eliminados e um agendador pode ser criado em uma única Eventing Function unificada e simples.

Pré-requisitos

Neste artigo, usaremos a versão mais recente do GA, ou seja, a versão 6.5.1 do Couchbase (talvez seja necessário fazer algumas alterações nas Eventing Functions descritas para versões anteriores do Couchbase). O exemplo deste artigo será executado com base no conjunto de dados de amostra de viagem fornecido com o servidor Couchbase.

DICA PRO: Somente para usuários avançados, Se você estiver familiarizado com o Couchbase Eventing e também com nossas ferramentas CLI / REST, poderá pular a maior parte deste blog e baixar um arquivo ZIP para configurar e executar rapidamente o sistema de agendamento apresentado abaixo. Clique com o botão direito do mouse no link a seguir e escolha Salvar link como para fazer o download do arquivo cron_impl_2func_CLI.zip, mova-o para um nó Eventing, extraia o arquivo ZIP e consulte o arquivo README.txt extraído.

No entanto, se você não estiver familiarizado com o Couchbase ou com o serviço Eventing, consulte o GET STARTED e um exemplo de Eventing, especificamente o seguinte:

Configure um servidor Couchbase 6.5.1 funcional de acordo com as instruções em Comece aqui!
Certifique-se de que você pode executar uma consulta N1QL no amostra de viagem de acordo com as instruções em Execute sua primeira consulta N1QL.
Entenda como implantar uma função básica de Eventing de acordo com as instruções do Arquivamento de documentos que também usa o amostra de viagem conjunto de dados.
Certifique-se de que você tenha o amostra de viagem na visualização de Compartimentos da interface do usuário.
Certifique-se de que você tenha um balde chamado metadados na visualização Buckets da interface do usuário, ele deve ter o tamanho mínimo de 200 MB.
Na visualização Buckets da interface do usuário, crie um bucket chamado dados cadastrais com o tamanho mínimo de 200 MB. Para obter etapas detalhadas sobre como criar buckets, consulte Criar um balde.
Conjunto allow_interbucket_recursion para verdadeiro a fim de permitir que duas (2) funções de Eventing alterem o mesmo 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 }'

Implementação #1, programação do tipo "cron" com código rígido

Para nossa primeira implementação, por exemplo, a Parte 1 da série, projetaremos uma estrutura de controle simples que é apenas um documento KV JSON e também duas (2) Eventing Functions que responderão e agirão de acordo com as informações da estrutura de controle.

Abaixo está um projeto de um documento JSON, ou estrutura de controle, que nos permitirá ter vários "eventos" agendados. Cada evento agendado terá seu próprio documento de controle com uma CHAVE exclusiva, como recurring_event::1, recurring_event::1, ... recurring_event::N. A própria estrutura JSON contém informações para "reconstituir a chave", pois nosso sistema de agendamento responderá a alterações ou atualizações (mutações) nos documentos de controle, como alternar o estado "ativo" para ativar ou desativar a ação ou alterar o campo "verbose", que controla a quantidade e o estilo do registro.

A seguir, um exemplo de documento de controle com KEY evento_recorrente::1 que executará a função JavaScript doCronActionA às 14:54 (2:30 pm) todos os dias.

Registro de controle JSON	Descrição
{
"type": "recurring_event",	A CHAVE será <>::<>
"id":1,
"hour":14,	A hora do dia 0-23, , 2X, *4X para acionar
"min":54,	O minuto na hora 0-59, , 2X, *4X para acionar
"action": "doCronActionA",	Função JavaScript a ser executada quando o cronômetro for acionado
"active":true,	Sinalizador para ativar ou desativar essa programação
"verbose": {	Controle de registro [OPCIONAL
"user_func":2,	Nível de registro para a lógica de ação: 0=nenhum, etc. etc.
"scheduler":3	Nível de registro para a lógica cron: 0=nenhum, etc. etc.
},
"dynamic": {	[DYNAMIC] controle e estatísticas do sistema
"estado": "braço",	"arm"\|"rearm"\|"pending" qualquer valor != "pending" inicia uma programação
"next_sched": 0,	Número de segundos desde a época até a próxima programação desejada
"prev_sched": 0,	Número de segundos desde a época para a programação anterior
"prev_etime": 0,	Número de segundos desde a época para a hora de execução real da programação anterior
"prev_delay": 0,	Número de segundos em que o cronômetro foi atrasado em relação à programação
"prev_atime": 0	Número de segundos usados pelo usuário 'action'
}
}

Como o Linux tradicional crontab você pode definir hour e min como números inteiros legais, e também pode definir hora para "*" para processar todas as horas ou definir min para "*" para processar todos os minutos.

Embora não apoiemos o crontab sintaxe, oferecemos suporte a duas configurações não padrão, como segue, se você definir ambos hora e min se você definir ambos como "*4X", executaremos e rearmaremos quatro (4) vezes por minuto e, se você definir ambos como "*2X", executaremos e rearmaremos duas (2) vezes por minuto. Abaixo está uma tabela de programações compatíveis com a descrição:

hora	min	Os valores podem ser números ou cadeias de caracteres
13	32	Corra às 13:32 (ou 13:32)
*	15	Executar a cada hora aos 15 minutos
8	12	Corra uma vez por dia às 8:32 (ou 8:32 da manhã)
*	*	Executar uma vez por minuto
*2X	*2X	Funciona duas vezes por minuto - requer que a hora e o minuto estejam definidos como "*2X"
*4X	*4X	Funciona quatro vezes por minuto - requer que a hora e o minuto estejam definidos como "*2X"

Eventualmente, usaremos o Query Workbench para inserir o cron documentos de controle, todos os quais devem ter uma CHAVE exclusiva de evento_recorrente::# para um horário de execução programado de 14:54 (14:54), para a ação doCronActionA, poderíamos usar a seguinte instrução N1QL.

Não se preocupe em executar nenhum comando N1QL neste momento, executaremos as instruções N1QL mais tarde, depois que tivermos criado e implementado nossa Eventing Function.

É possível criar um registro de controle (ou registros) no bucket amostra de viageme, em seguida, liste-o, arme-o, desarme-o, ajuste a programação que ele segue, altere o nível de verbosidade do registro em log ou exclua-o da seguinte forma:

Ação	Declaração N1QL
Criar uma programação	INSERT INTO `amostra de viagem` (KEY,VALUE) VALORES ( "recurring_event::1", { "type": "recurring_event", "id":1, "hour": "14″, "min": "54″, "action": "doCronActionA", "active":true } );
Criar um índice para consultar dados sem especificar chaves	CREATE primary INDEX on `dados cadastrais` ;
Mostrar todas as programações ordenadas por id	SELECT * FROM `dados cadastrais` WHERE type="recurring_event" order by id ;
Mostrar programação específica	SELECT * FROM `dados cadastrais` WHERE type="recurring_event" AND id=1 ;
Armar ou definir ativo	ATUALIZAÇÃO `dados cadastrais` SET active = true WHERE type="recurring_event" AND id=1 ;
Desarmar ou definir como inativo	ATUALIZAÇÃO `dados cadastrais` SET active = false WHERE type="recurring_event" AND id=1 ;
Ajuste do tempo de disparo	ATUALIZAÇÃO `dados cadastrais` SET hour = 11, min = 30 WHERE type="recurring_event" AND id=1 ;
Ajustar o registro da "ação"	ATUALIZAÇÃO `dados cadastrais` SET verbose.user_data = 0 WHERE type="recurring_event" AND id=1 ;
Ajustar o registro da lógica do agendador	ATUALIZAÇÃO `dados cadastrais` SET verbose.scheduler = 0 WHERE type="recurring_event" AND id=1 ;
Excluir a programação	DELETE FROM `dados cadastrais` WHERE type="recurring_event" AND id=1 ;

Suponha que tenhamos quatro (4) programações ativas. A execução do primeiro comando N1QL acima listará todas elas, por exemplo

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 ;

Retornaria algo como o seguinte resultado (exibição de tabela no Query Workbench):

ativo	ação	hora	id	min	programador	tipo	user_func
verdadeiro	"doCronActionA"	14	1	54	1	"recurring_event" (evento recorrente)	2
verdadeiro	"doCronActionB"	*	2	*	1	"recurring_event" (evento recorrente)	1
verdadeiro	"doCronActionC"	*2X	3	*2X	4	"recurring_event" (evento recorrente)	4
verdadeiro	"doCronActionD"	*	4	0	0	"recurring_event" (evento recorrente)	1

Na tabela acima, temos quatro ações: a primeira é executada uma vez por dia, a segunda é executada a cada minuto, a terceira a cada 30 segundos e a quarta é executada uma vez por hora. Em uma próxima edição desta série, adicionaremos o recurso "dia da semana".

O objeto aninhado do registro de controle JSON "detalhado", se não for fornecido, terá como padrão { "user_func":1, "scheduler":1 } indicando um nível de registro baixo ou conciso para a função de ação e também para a lógica de agendamento. Um valor de 0 suprimirá todas as mensagens de registro, ou seja, doCronActionD, enquanto valores maiores serão mais detalhados, ou seja, conforme definido em doCronActionC.

O objeto aninhado do registro de controle JSON "dinâmico" se normalmente nunca for fornecido e terá como padrão {"state": "arm", "next_sched": 0, "prev_sched": 0, "prev_etime": 0, "prev_delay": 0, "prev_atime": 0 } esse é um bloco de rascunho para a programação lógica de Eventing em execução e também fornece estatísticas úteis sobre os tempos de execução, portanto, deve ser tratado como somente leitura.

Neste ponto, temos um projeto de controle de alto nível, mas precisamos de lógica para processar nossas estruturas de controle. É aqui que o Eventing Service do Couchbase, especificamente uma Eventing Function, entra em ação.

As funções de eventos

Esse projeto requer duas (2) funções de eventos: uma função JavaScript principal "cron_impl_2func_651" e uma pequena função JavaScript auxiliar "cron_impl_2func_651_help". Discutiremos cada seção das funções JavaScript que compõem o código JavaScript combinado da implementação inicial, com quase 610 linhas (sendo que cerca de 44% das linhas são comentários e espaços em branco)

Não se preocupe em recortar e colar agoraSe você preferir, poderá usar o link para baixar (para importação) as duas Eventing Functions necessárias e todas as configurações necessárias em dois arquivos denominados "cron_impl_2func_651.json" "cron_impl_2func_651_help.json" e também, se preferir, as duas funções unificadas completas que podem ser recortadas e coladas diretamente.

Nossa função de evento principal "cron_impl_2func_651" será composta de nove (9) funções JavaScript

Três (3) funções de lógica comercial (duas das quais estão vazias).
- doCronActionA(doc) - um exemplo de ação de usuário N1QL a ser executado
- doCronActionB(doc) - um shell de ação do usuário vazio para experimentos
- doCronActionC(doc) - um shell de ação do usuário vazio para experimentos
Um (1) ponto de entrada para Eventing.
- OnUpdate(doc, meta) - o ponto de entrada padrão do Eventing para inserções ou atualizações
Um (1) cron analisador de sintaxe para gerar a próxima programação.
- getNextRecurringDate(hour_str, min_str) - lógica do cron para encontrar a próxima data agendada
Três (3) funções de suporte para verificar se a lógica comercial existe ou formatar resultados.
- verifyFunctionExistsViaEval(curDoc, id) - certifique-se de que temos uma função para executar
- toNumericFixed(number, precision) - formata um float em um estilo compacto
- toLocalISOTime(d) - formata uma data em um estilo compacto
Uma (1) função de retorno de chamada quando os cronômetros são executados.
- Callback(doc) - uma função de retorno de chamada para cronômetros programados

Nossa Helper Eventing Function "cron_impl_2func_651_help" será composta por uma (1) função JavaScript

Um (1) ponto de entrada para Eventing.
- OnUpdate(doc, meta) - o ponto de entrada padrão do Eventing para inserções ou atualizações

Nas seções seguintes, examinaremos cada uma das funções JavaScript acima.

Precisamos de uma função JavaScript, por exemplo, a lógica comercial a ser executada em uma programação periódica.

A primeira coisa que queremos é uma rotina ou função que tenha nossa lógica de negócios que será executada com base em nossas regras do crontab. Chamaremos o método JavaScript doCronActionA(doc)no entanto, ele pode ser chamado de qualquer coisa, por exemplo doPeriodicLedgerBalance(doc), Os únicos requisitos para nossas funções de "ação" que implementam nossa lógica comercial programada são os seguintes:

Tem um parâmetro: doc, um documento de controle, conforme descrito acima, do tipo="recurring_event".
O nome real do JavaScript corresponde ao campo "action" no documento de controle.
Devoluções verdadeiro sobre sucesso e falso em caso de falha
Utiliza doc.verbose.user_func para controlar o registro; se 0, ele é silencioso; se 1, ele emite uma única linha; se 2, ele emite qualquer informação de registro necessária para depurar a função etc. etc.

Escreveremos nossa função doCronActionA(doc)para executar uma consulta N1QL incorporada) para combinar contagens de companhias aéreas por país e, em seguida, criar um único documento KV de dados 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;

Em meu sistema de teste, um pequeno nó único nãoMDS (executando todos os serviços do Couchbase), o N1QL acima leva cerca de 20 ms. (para fins de clareza, fingir que é supercomplexo leva 10 segundos para ser concluído).

A ideia aqui é que o documento KV final calculado e resumido possa ser carregado rapidamente por 100 mil (ou um milhão) de mutações Eventing por segundo sem a sobrecarga adicional de comunicação com os nós do serviço Query e do processamento de instruções N1QL em cada mutação.

Deve ser óbvio que o objetivo dessa lógica comercial específica, doCronActionA(doc)O objetivo do cache semi-estático é criar um cache semi-estático que seja atualizado periodicamente em um cronograma.

Tudo o que estamos realmente fazendo (e é bastante rápido) é obter uma contagem de companhias aéreas por país a partir do conjunto de documentos de amostra de viagem. À medida que usamos o N1QL, criamos um documento e, por fim, o escrevemos no KV como um documento resumido. O ponto principal a ser destacado aqui é que não queremos repetir o mesmo trabalho para milhões de mutações cada, especialmente porque alguns cálculos podem levar 10 segundos de tempo de computação do serviço de consulta cada vez que iniciamos uma consulta N1QL incorporada a partir de uma função Eventing.

Abaixo, mostramos a função JavaScript que queremos executar uma vez por dia (ou talvez uma vez por hora, etc.). Observe que o nome da função corresponde ao nome no campo de ação da estrutura de controle. Para obter mais detalhes sobre a terminologia de Eventing e as construções de linguagem, consulte os documentos e exemplos do Couchbase em Serviço de eventos: 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;

}

A função acima simplesmente 1) consulta o bucket de amostra de viagens para extrair dados; nesse caso, a contagem de companhias aéreas para cada país; 2) cria um novo documento KV e uma nova chave e os grava no bucket de amostra de viagens para uso posterior.

Além disso, como parte deste exemplo, criamos um registro que responde a uma configuração de verbosidade numérica que a) registra uma única linha se o documento de controle tiver um valor para doc.verbose.user_func == 1 ou b) emite mais informações se o valor de doc.verbose.user_func >= 2.

Essa é uma estrutura genérica que pode executar um (1) cron ação ou até mesmo mil (1000) de cron ações. Por isso, forneci dois shells de função "vazios" adicionais - como já foi dito, eles poderiam ter qualquer nome.

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;

}

Essas funções acima, doCronActionB e doCronActionC, são triviais, pois simplesmente registram informações no log do aplicativo Eventing da função Eventing. Consulte Funções de registro em log para obter mais detalhes. É claro que você precisa de um documento de controle do tipo = "recurring_event" com active = true e uma ação como action = "doCronActionB" para realmente ativá-los e executá-los.

Precisamos de um ponto de entrada ou manipulador de eventos

A partir da versão 6.5, dois pontos de entrada ou manipuladores são compatíveis com o Eventing Service OnUpdate(doc, meta) e OnDelete(meta) estamos interessados apenas no OnUpdate(doc,meta) para este exemplo.

O OnUpdate(doc,meta) é chamado quando qualquer documento no bucket de origem é criado ou modificado (mutado) e filtra imediatamente os documentos que não interessam. ^[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);

}

A chave aqui é que o cron A lógica do nosso manipulador só se preocupa com os documentos que têm doc.type de "recurring_event" e também um doc.active de true. Além disso, neste exemplo, criamos um rastreamento para o cron lógica de manutenção que só é registrada no log do aplicativo se o documento de controle tiver um valor para doc.verbose >= 3.

Se você executar apenas algumas programações, poderá desativar o trabalho no espaço do usuário ou "Correção para operações de balde de crescimento 6.5.X" comentando quatro linhas de código no bloco OnUpdate acima para "cron_impl_2func_651" da seguinte forma:

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"] = {};

// }

Precisamos de código para contornar possíveis operações de aumento de balde para a versão 6.5.X

A partir da versão 6.5.X, precisamos de um "Correção para operações de balde de crescimento 6.5.X", o que acontece em sistemas ociosos com muitos cronômetros programados para o futuro. Esse código garante que um timer de Eventing tenha sido disparado recentemente em um vBucket (o que elimina o problema do vBucket em questão com base em cada função).

// 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);

}

Precisamos de um utilitário para calcular o próximo horário na programação

A próxima função getNextRecurringDate(hour, min) determinará um horário para executar a ação, conforme definido como parte de nossa programação. Essa não é uma implementação completa de cronEm vez disso, ele contém os principais recursos padrão para execução uma vez por dia, uma vez por hora, uma vez por minuto. Ele também contém uma sintaxe não padrão para permitir a execução duas vezes por minuto ou quatro vezes por minuto.

Conforme descrito anteriormente, a função getNextRecurringDate(hour, min) permite o seguinte (a tabela está duplicada abaixo), sendo que os dois últimos não são padrão.^[4]

hora	min	Os valores podem ser números ou cadeias de caracteres
13	32	Corra às 13:32 (ou 13:32)
*	15	Executar a cada hora aos 15 minutos
8	12	Corra uma vez por dia às 8:32 (ou 8:32 da manhã)
*	*	Executar uma vez por minuto
*2X	*2X	Funciona duas vezes por minuto - requer que a hora e o minuto estejam definidos como "*2X"
*4X	*4X	Funciona quatro vezes por minuto - requer que a hora e o minuto estejam definidos como "*2X"

Abaixo está uma implementação da lógica necessária para determinar o próximo momento para acionar um temporizador Eventing em nossa agenda, caso a lógica do usuário em nosso primeiro exemplo doCronActionA(doc) não for concluído em tempo hábil, por exemplo, excesso de tempo real, a próxima quantidade da programação será selecionada. Observe os temporizadores e suas funções principais. Portanto, se uma Eventing Function tiver um tempo limite de execução padrão de 60 segundos, se necessário, essa configuração poderá ser ajustada ou 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;

}

Precisamos de alguns pequenos serviços públicos

A função utilitária comum que simplesmente verifica se nosso JavaScript existe é usada por ambos OnUpdate(doc,meta), mostrado acima, e o cronômetro Callback(doc), mostrado mais tarde. Abaixo está verifyFunctionExistsViaEval(curDoc, id) que recebe dois argumentos: um documento de controle JSON e a KEY para esse documento.

Isso nos permite saber imediatamente, na implantação, se houve um problema com uma incompatibilidade de nomes entre o documento de registro de controle JSON e o nome real da função de lógica comercial no 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;

}

Observe que, se houver uma tentativa de executar uma função inexistente, o usuário final receberá um aviso no log do aplicativo cron_impl_2func_651.log para corrigir o problema.

2020-04-22T16:20:38.725-07:00 [INFO] "Avisar/Desabilitar (Sem Ação e Sem Rearmamento), porque a 'ação' necessária de doCronMyNewFunction(doc) não existe, o id é" "recurring_event::1"

Essa correção pode ser feita por meio de uma pausa/retomada que adiciona a função e, em seguida, ajusta o documento de controle com o ID ou KEY especificado (por meio de uma alternância ativa para falso e depois para verdadeiro) ou ajusta o documento de controle para apontar para uma função existente em seu manipulador.

Em seguida, o utilitário toNumericFixed(number, precision) permite apenas uma formatação compacta e agradável de floats para nossas mensagens 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 fim, o utilitário toLocalISOTime(d) permite apenas uma boa formatação compacta de datas para nossas mensagens 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);

}

Precisamos de uma chamada de retorno do temporizador para executar a lógica do usuário e rearmar o temporizador

A função JavaScript final em "cron_impl_2func_651" é o retorno de chamada do Timer, que é chamado quando o timer programado é acionado. A função de retorno de chamada deve ser uma função de nível superior que recebe um único argumento, o contexto.

Nesse caso, em nosso manipulador OnUpdate, fizemos referência a uma função JavaScript de Callback(doc) com um contexto de doc (nosso documento de controle do agendador ativo de type="recurring_event")

Na versão 6.6, podemos criar outro cronômetro dentro de um cronômetro, mas em todas as versões anteriores precisaremos acionar uma mutação para uma função "auxiliar" (evitamos cuidadosamente a recursão infinita). Na versão 6.6, a função auxiliar não é necessária e a lógica foi substancialmente simplificada.

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

}

Precisamos de uma função auxiliar para acionar uma nova mutação

Como antes da versão 6.6 (que ainda não foi lançada) não é possível criar um temporizador a partir de um retorno de chamada de um temporizador em execução, precisamos de uma segunda Eventing Function (juntamente com "allow_interbucket_recursion":true) para acionar uma mutação de modo que possamos gerar todos os nossos temporizadores no ponto de entrada OnUpdate(doc,meta) da Eventing Function principal. Fazemos isso da seguinte forma:

cron_impl_2func_651 OnUpdate(doc,meta) recebe uma mutação, programa um cronômetro
cron_impl_2func_651 Após um determinado período de tempo, quando o cronômetro amadurece, o Callback(doc) A rotina é executada, primeiro executa a ação do usuário desejada e, em seguida, cria uma mutação #1 no documento de controle (que não é vista pela função de criação para evitar recursão)
cron_impl_2func_651_help OnUpdate(doc,meta) recebe uma mutação, faz outra mutação #2 no documento de controle, o que aciona 1. acima em um ciclo sem fim.

Observe que, na versão 6.6 do Couchbase, não precisamos de uma função auxiliar, pois é permitido criar um timer de dentro de um timer em execução. Isso simplifica muito a lógica necessária para criar um cron sistema^[2].

A única função JavaScript em "cron_impl_2func_651_help" OnUpdate(doc,meta) é mostrado abaixo.

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;

}

A função auxiliar precisa de alguns utilitários

Esses utilitários fornecem um Correção para operações de balde de crescimento 6.5.X assegurando que um temporizador Eventing seja acionado em cada vBucket em tempo hábil.

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

}

Agora, vamos implementar as duas Eventing Functions

Revisamos muitos códigos e o design do agendador inicial, agora é hora de ver tudo funcionando em conjunto.

Lembre-se de que, neste exemplo, há três compartimentos amostra de viagem (um conjunto de dados padrão de amostra), metadados(o bucket de metadados é um bloco de notas para Eventing e pode ser compartilhado com outras funções de Eventing) e, por fim, o dados cadastrais (que contém nossas programações cron). O amostra de viagem tem um tamanho de 100 MB e os outros dois baldes metadados e dados cadastrais devem ter um tamanho de 200 MB e já existir de acordo com as instruções em "Pré-requisitos".

Verifique a configuração atual do seu bucket acessando o Console da Web do Couchbase > Buckets página:

Para implementar a função Eventing "cron_impl_2func_651", você pode seguir um destes dois métodos:

Complexidade básica, Método #1 Download/Importação
Complexidade média, método #2 Adicionar função manualmente, recortar e colar JavaScript

Método #1 Download/Importação

Importar a primeira função "cron_impl_2func_651"

Faça o download da primeira Eventing Function com todas as configurações necessárias, clique com o botão direito do mouse no link a seguir e selecione Salvar link como para fazer o download do arquivo cron_impl_2func_651.json em seu sistema de arquivos local.

Do Console da Web do Couchbase > Eventos página, clique em IMPORTAÇÃO, navegue até o arquivo cron_impl_2func_651.json, selecione-o e abra-o. A ADICIONAR FUNÇÃO é exibida.

No ADICIONAR FUNÇÃO para elementos individuais da função, forneça as informações abaixo. Observe o arquivo JSON cron_impl_2func_651.json pré-configurará todas as configurações corretamente para este exemplo:

Para o Balde de origem verifique se ele está definido como dados cadastrais.
Para o Balde de metadados verifique se ele está definido como metadados.
Verifique se cron_impl_2func_651 é o nome da função que você está criando no Nome da função caixa de texto.
[Etapa opcional] Digite o texto Um agendador do tipo cron - parte 1, no Descrição caixa de texto.
Para o Configurações use os valores padrão.
Para o Amarrações verifique se existem dois vínculos.
Para a vinculação, o "bucket alias" especifica cron_bkt como o "nome de alias" do bucket e selecione
dados cadastrais como o bucket associado, e o modo deve ser "leitura e gravação".
Para a vinculação, o "bucket alias" especifica ts_bkt como o "nome de alias" do bucket e selecione
amostra de viagem como o bucket associado, e o modo deve ser "leitura e gravação".
Suas configurações na caixa de diálogo devem ser parecidas com as seguintes:
Depois de verificar todas as informações necessárias na caixa de diálogo ADD FUNCTION, clique em Next: Add Code (Adicionar código). O código cron_impl_2func_651 é exibida (com o código JavaScript pré-carregado).
Para retornar à tela Eventing, clique no botão '< voltar para Eventing' (abaixo do editor) ou clique no link Eventos guia.

Importar a segunda função "cron_impl_2func_651_help"

Faça o download da segunda Eventing Function com todas as configurações necessárias, clique com o botão direito do mouse no link a seguir e selecione Salvar link como para fazer o download do arquivo cron_impl_2func_651_help.json em seu sistema de arquivos local.

Do Console da Web do Couchbase > Eventos página, clique em IMPORTAÇÃO, navegue até o arquivo cron_impl_2func_651_help.json, selecione-o e abra-o. A ADICIONAR FUNÇÃO é exibida.

No ADICIONAR FUNÇÃO para elementos individuais da função, forneça as informações abaixo. Observe o arquivo JSON cron_impl_2func_651_help.json pré-configurará todas as configurações corretamente para este exemplo:

Para o Balde de origem verifique se ele está definido como dados cadastrais.
Para o Balde de metadados verifique se ele está definido como metadados.
Verifique se cron_impl_2func_651_help é o nome da função que você está criando no Nome da função caixa de texto.
[Etapa opcional] Digite o texto Um auxiliar de agendamento do tipo cron - parte 1, no Descrição caixa de texto.
Para o Configurações use os valores padrão.
Para o Amarrações verifique se existe apenas uma ligação.
Para a vinculação, o "bucket alias" especifica cron_bkt como o "nome de alias" do bucket e selecione
dados cadastrais como o bucket associado, e o modo deve ser "leitura e gravação".
Suas configurações na caixa de diálogo devem ser parecidas com as seguintes:
Depois de verificar todas as informações necessárias na caixa de diálogo ADD FUNCTION, clique em Next: Add Code (Adicionar código). O código cron_impl_2func_651_help é exibida (com o código JavaScript pré-carregado).
Para retornar à tela Eventing, clique no botão '< voltar para Eventing' (abaixo do editor) ou clique no link Eventos guia.

Método #2 Adicionar manualmente a função, recortar e colar JavaScript

Criar manualmente "cron_impl_2func_651"

Para adicionar a primeira função Eventing da lista Console da Web do Couchbase > Eventos página, clique em ADICIONAR FUNÇÃOpara adicionar uma nova função. A função ADICIONAR FUNÇÃO é exibida.

No ADICIONAR FUNÇÃO para elementos de função individuais, forneça as informações abaixo:

Para o Balde de origem definido como dados cadastrais.
Para o Balde de metadados definido como metadados.
Fazer cron_impl_2func_651 é o nome da função que você está criando no Nome da função caixa de texto.
[Etapa opcional] Digite o texto Um agendador do tipo cron - parte 1, no Descrição caixa de texto.
Para o Configurações use os valores padrão.
Para o Amarrações crie dois vínculos:
Para a vinculação, o "bucket alias" especifica cron_bkt como o "nome de alias" do bucket e selecione
dados cadastrais como o bucket associado, e o modo deve ser "leitura e gravação".
Para a vinculação, o "bucket alias" especifica ts_bkt como o "nome de alias" do bucket e selecione
amostra de viagem como o bucket associado, e o modo deve ser "leitura e gravação".
Depois de definir suas configurações, sua caixa de diálogo deverá ter a seguinte aparência:
Depois de fornecer todas as informações necessárias no formulário ADICIONAR FUNÇÃO clique em Próximo: Adicionar código. O cron_impl_2func_651 é exibida. A caixa de diálogo cron_impl_2func_651 contém inicialmente um bloco de código de espaço reservado. Você substituirá seu cron_impl_2func_651 nesse bloco.

Copie o código-fonte JavaScript da Eventing Function a seguir (618 linhas) e cole-o no bloco de código de espaço reservado 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 */