From f6e7f9257397a1cd35aeb7afd27fdfc255705c64 Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Tue, 13 Dec 2022 09:36:34 +0000 Subject: [PATCH] improve error handling reasaoning requests --- .../leidinggevenden-dispatching/config.js | 4 ++ .../leidinggevenden-dispatching/util.js | 38 +++++++++---------- .../mandate-dispatching/config.js | 6 ++- .../mandate-dispatching/util.js | 38 +++++++++---------- 4 files changed, 45 insertions(+), 41 deletions(-) diff --git a/config/loket-consumer/leidinggevenden-dispatching/config.js b/config/loket-consumer/leidinggevenden-dispatching/config.js index 526d237..e280733 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/config.js +++ b/config/loket-consumer/leidinggevenden-dispatching/config.js @@ -3,7 +3,9 @@ const MU_CALL_SCOPE_ID_INITIAL_SYNC = process.env.MU_CALL_SCOPE_ID_INITIAL_SYNC const BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES = process.env.BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES == 'true' ? true : false; const DIRECT_DATABASE_ENDPOINT = process.env.DIRECT_DATABASE_ENDPOINT || 'http://virtuoso:8890/sparql'; const MAX_DB_RETRY_ATTEMPTS = parseInt(process.env.MAX_DB_RETRY_ATTEMPTS || 5); +const MAX_REASONING_RETRY_ATTEMPTS = parseInt(process.env.MAX_REASONING_RETRY_ATTEMPTS || 5); const SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); +const SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION || 10000); const SLEEP_TIME_AFTER_FAILED_DB_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_DB_OPERATION || 60000); const INGEST_GRAPH = process.env.INGEST_GRAPH || `http://mu.semte.ch/graphs/organisatieportaal`; @@ -18,8 +20,10 @@ module.exports = { BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, DIRECT_DATABASE_ENDPOINT, MAX_DB_RETRY_ATTEMPTS, + MAX_REASONING_RETRY_ATTEMPTS, SLEEP_BETWEEN_BATCHES, SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION, INGEST_GRAPH, FILE_SYNC_GRAPH }; \ No newline at end of file diff --git a/config/loket-consumer/leidinggevenden-dispatching/util.js b/config/loket-consumer/leidinggevenden-dispatching/util.js index db93194..54bee47 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/util.js +++ b/config/loket-consumer/leidinggevenden-dispatching/util.js @@ -1,3 +1,8 @@ +const { + MAX_REASONING_RETRY_ATTEMPTS, + SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION +} = require('./config'); + async function batchedDbUpdate(muUpdate, graph, triples, @@ -25,19 +30,22 @@ ${batch} `, extraHeaders, endpoint); }; - await dbOperationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); + await operationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); await new Promise(r => setTimeout(r, sleepBetweenBatches)); } } -async function dbOperationWithRetry(callback, +async function operationWithRetry(callback, attempt, maxAttempts, sleepTimeOnFail) { try { - return await callback(); + if (typeof callback === "function") + return await callback(); + else // Catch error from promise - not how I would do it normally, but allows re use of existing code. + return await callback; } catch (e) { console.log(`Operation failed for ${callback.toString()}, attempt: ${attempt} of ${maxAttempts}`); @@ -50,7 +58,7 @@ async function dbOperationWithRetry(callback, } await new Promise(r => setTimeout(r, sleepTimeOnFail)); - return dbOperationWithRetry(callback, ++attempt, maxAttempts, sleepTimeOnFail); + return operationWithRetry(callback, ++attempt, maxAttempts, sleepTimeOnFail); } } @@ -75,8 +83,10 @@ function partition(arr, fn) { * */ function transformTriples(fetch, triples) { - return preProcess(fetch, triples) - .then(preprocessed => mainConversion(fetch, preprocessed)); + return operationWithRetry(preProcess(fetch, triples), 0, + MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION) + .then(preprocessed => operationWithRetry(mainConversion(fetch, preprocessed), 0, + MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION)); } @@ -90,15 +100,8 @@ function preProcess(fetch, triples) { redirect: 'follow' }; - // console.log('!!! Input\n' + triples) - return fetch("http://reasoner/reason/dl2op/preprocess", requestOptions) - .then(response => response.text()) - // .then(result => { - // console.log(`!!! PREPROCESS\n${result}`) - // return result - // }) - .catch(error => console.log('error', error)); + .then(response => response.text()); } function mainConversion(fetch, triples) { @@ -112,12 +115,7 @@ function mainConversion(fetch, triples) { }; return fetch("http://reasoner/reason/dl2op/main", requestOptions) - .then(response => response.text()) - // .then(result => { - // console.log(`!!! MAIN\n${result}`) - // return result - // }) - .catch(error => console.log('error', error)); + .then(response => response.text()); } function transformStatements(fetch, triples) { diff --git a/config/loket-consumer/mandate-dispatching/config.js b/config/loket-consumer/mandate-dispatching/config.js index d3342b3..e280733 100644 --- a/config/loket-consumer/mandate-dispatching/config.js +++ b/config/loket-consumer/mandate-dispatching/config.js @@ -3,12 +3,14 @@ const MU_CALL_SCOPE_ID_INITIAL_SYNC = process.env.MU_CALL_SCOPE_ID_INITIAL_SYNC const BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES = process.env.BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES == 'true' ? true : false; const DIRECT_DATABASE_ENDPOINT = process.env.DIRECT_DATABASE_ENDPOINT || 'http://virtuoso:8890/sparql'; const MAX_DB_RETRY_ATTEMPTS = parseInt(process.env.MAX_DB_RETRY_ATTEMPTS || 5); +const MAX_REASONING_RETRY_ATTEMPTS = parseInt(process.env.MAX_REASONING_RETRY_ATTEMPTS || 5); const SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); +const SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION || 10000); const SLEEP_TIME_AFTER_FAILED_DB_OPERATION = parseInt(process.env.SLEEP_TIME_AFTER_FAILED_DB_OPERATION || 60000); const INGEST_GRAPH = process.env.INGEST_GRAPH || `http://mu.semte.ch/graphs/organisatieportaal`; -if(!process.env.FILE_SYNC_GRAPH) +if (!process.env.FILE_SYNC_GRAPH) throw `Expected 'FILE_SYNC_GRAPH' to be provided.`; const FILE_SYNC_GRAPH = process.env.FILE_SYNC_GRAPH; @@ -18,8 +20,10 @@ module.exports = { BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, DIRECT_DATABASE_ENDPOINT, MAX_DB_RETRY_ATTEMPTS, + MAX_REASONING_RETRY_ATTEMPTS, SLEEP_BETWEEN_BATCHES, SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION, INGEST_GRAPH, FILE_SYNC_GRAPH }; \ No newline at end of file diff --git a/config/loket-consumer/mandate-dispatching/util.js b/config/loket-consumer/mandate-dispatching/util.js index db93194..54bee47 100644 --- a/config/loket-consumer/mandate-dispatching/util.js +++ b/config/loket-consumer/mandate-dispatching/util.js @@ -1,3 +1,8 @@ +const { + MAX_REASONING_RETRY_ATTEMPTS, + SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION +} = require('./config'); + async function batchedDbUpdate(muUpdate, graph, triples, @@ -25,19 +30,22 @@ ${batch} `, extraHeaders, endpoint); }; - await dbOperationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); + await operationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); await new Promise(r => setTimeout(r, sleepBetweenBatches)); } } -async function dbOperationWithRetry(callback, +async function operationWithRetry(callback, attempt, maxAttempts, sleepTimeOnFail) { try { - return await callback(); + if (typeof callback === "function") + return await callback(); + else // Catch error from promise - not how I would do it normally, but allows re use of existing code. + return await callback; } catch (e) { console.log(`Operation failed for ${callback.toString()}, attempt: ${attempt} of ${maxAttempts}`); @@ -50,7 +58,7 @@ async function dbOperationWithRetry(callback, } await new Promise(r => setTimeout(r, sleepTimeOnFail)); - return dbOperationWithRetry(callback, ++attempt, maxAttempts, sleepTimeOnFail); + return operationWithRetry(callback, ++attempt, maxAttempts, sleepTimeOnFail); } } @@ -75,8 +83,10 @@ function partition(arr, fn) { * */ function transformTriples(fetch, triples) { - return preProcess(fetch, triples) - .then(preprocessed => mainConversion(fetch, preprocessed)); + return operationWithRetry(preProcess(fetch, triples), 0, + MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION) + .then(preprocessed => operationWithRetry(mainConversion(fetch, preprocessed), 0, + MAX_REASONING_RETRY_ATTEMPTS, SLEEP_TIME_AFTER_FAILED_REASONING_OPERATION)); } @@ -90,15 +100,8 @@ function preProcess(fetch, triples) { redirect: 'follow' }; - // console.log('!!! Input\n' + triples) - return fetch("http://reasoner/reason/dl2op/preprocess", requestOptions) - .then(response => response.text()) - // .then(result => { - // console.log(`!!! PREPROCESS\n${result}`) - // return result - // }) - .catch(error => console.log('error', error)); + .then(response => response.text()); } function mainConversion(fetch, triples) { @@ -112,12 +115,7 @@ function mainConversion(fetch, triples) { }; return fetch("http://reasoner/reason/dl2op/main", requestOptions) - .then(response => response.text()) - // .then(result => { - // console.log(`!!! MAIN\n${result}`) - // return result - // }) - .catch(error => console.log('error', error)); + .then(response => response.text()); } function transformStatements(fetch, triples) {