improve error handling reasaoning requests
This commit is contained in:
@@ -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
|
||||
};
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
};
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user