From a127ea8060f0c5b5e199525585e6db9da092f1c3 Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Tue, 29 Nov 2022 16:00:34 +0000 Subject: [PATCH 1/9] loket to op sync with reasoning --- .../mandate-dispatching/config.js | 25 +++ .../delta-sync-dispatching.js | 126 ++++++++++++++ .../initial-sync-dispatching.js | 84 +++++++++ .../mandate-dispatching/util.js | 137 +++++++++++++++ config/reasoner/dl2op/main/mapping.n3 | 82 +++++++++ config/reasoner/dl2op/preprocess/mapping.n3 | 31 ++++ docker-compose.yml | 162 +++++++++++------- 7 files changed, 582 insertions(+), 65 deletions(-) create mode 100644 config/loket-consumer/mandate-dispatching/config.js create mode 100644 config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js create mode 100644 config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js create mode 100644 config/loket-consumer/mandate-dispatching/util.js create mode 100644 config/reasoner/dl2op/main/mapping.n3 create mode 100644 config/reasoner/dl2op/preprocess/mapping.n3 diff --git a/config/loket-consumer/mandate-dispatching/config.js b/config/loket-consumer/mandate-dispatching/config.js new file mode 100644 index 0000000..0c97ddd --- /dev/null +++ b/config/loket-consumer/mandate-dispatching/config.js @@ -0,0 +1,25 @@ +const BATCH_SIZE = parseInt(process.env.BATCH_SIZE) || 100; +const MU_CALL_SCOPE_ID_INITIAL_SYNC = process.env.MU_CALL_SCOPE_ID_INITIAL_SYNC || 'http://redpencil.data.gift/id/concept/muScope/deltas/consumer/initialSync'; +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 SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); +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/public`; + + +if(!process.env.FILE_SYNC_GRAPH) + throw `Expected 'FILE_SYNC_GRAPH' to be provided.`; +const FILE_SYNC_GRAPH = process.env.FILE_SYNC_GRAPH; + +module.exports = { + BATCH_SIZE, + MU_CALL_SCOPE_ID_INITIAL_SYNC, + BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, + DIRECT_DATABASE_ENDPOINT, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + INGEST_GRAPH, + FILE_SYNC_GRAPH +}; \ No newline at end of file diff --git a/config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js b/config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js new file mode 100644 index 0000000..9f936df --- /dev/null +++ b/config/loket-consumer/mandate-dispatching/delta-sync-dispatching.js @@ -0,0 +1,126 @@ +const { transformStatements, batchedDbUpdate, partition } = require('./util'); +const { BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, + DIRECT_DATABASE_ENDPOINT, + MU_CALL_SCOPE_ID_INITIAL_SYNC, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + INGEST_GRAPH, + FILE_SYNC_GRAPH +} = require('./config'); + +/** +* Dispatch the fetched information to a target graph. +* Note: will be ADDED to it's own graph. +* We take only care of adding them, not updating triples, this is a TODO +* @param { mu, muAuthSudo } lib - The provided libraries from the host service. +* @param { termObjectChangeSets: { deletes, inserts } } data - The fetched changes sets, which objects of serialized Terms +* [ { +* graph: "", +* subject: "", +* predicate: "", +* object: "^^" +* } +* ] +* @return {void} Nothing +*/ +async function dispatch(lib, data) { + const { mu, muAuthSudo, fetch } = lib; + const { termObjectChangeSets } = data; + + for (let { deletes, inserts } of termObjectChangeSets) { + const deletePartitions = partition(deletes, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`); + const deleteStatements = regularDeletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); + + if (deleteStatements.length) { + await transformStatements(fetch, deleteStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedStatements, + { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, + process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "DELETE" + ); + } + ) + } + + if (fileDeleteStatements.length) { + await transformStatements(fetch, fileDeleteStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedStatements, + { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, + process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "DELETE" + ); + } + ) + } + + const insertPartitions = partition(inserts, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`); + const fileInsertStatements = fileInserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); + + + if (regularInsertStatements.length) { + await transformStatements(fetch, regularInsertStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedStatements, + {}, + process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "INSERT" + ); + } + ) + } + + if (fileInsertStatements.length) { + await transformStatements(fetch, fileInsertStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedStatements, + {}, + process.env.MU_SPARQL_ENDPOINT, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "INSERT" + ); + } + ) + } + } +} + +module.exports = { + dispatch +}; \ No newline at end of file diff --git a/config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js b/config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js new file mode 100644 index 0000000..69bcdb4 --- /dev/null +++ b/config/loket-consumer/mandate-dispatching/initial-sync-dispatching.js @@ -0,0 +1,84 @@ +const { transformStatements, batchedDbUpdate, partition } = require('./util'); +const { BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, + DIRECT_DATABASE_ENDPOINT, + MU_CALL_SCOPE_ID_INITIAL_SYNC, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + INGEST_GRAPH, + FILE_SYNC_GRAPH +} = require('./config'); +const endpoint = BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES ? DIRECT_DATABASE_ENDPOINT : process.env.MU_SPARQL_ENDPOINT; + +/** +* Dispatch the fetched information to a target graph. +* @param { mu, muAuthSudo } lib - The provided libraries from the host service. +* @param { termObjects } data - The fetched quad information, which objects of serialized Terms +* [ { +* graph: "", +* subject: "", +* predicate: "", +* object: "^^" +* } +* ] +* @return {void} Nothing +*/ +async function dispatch(lib, data) { + const { mu, muAuthSudo, fetch } = lib; + const { termObjects } = data; + + const partitions = partition(termObjects, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`) + + await transformStatements(fetch, originalRegularInsertTriples).then( + transformedRegularInsertTriples => { + // console.log(transformedRegularInsertTriples); + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedRegularInsertTriples, + { 'mu-call-scope-id': MU_CALL_SCOPE_ID_INITIAL_SYNC }, + endpoint, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION + ) + } + ) + } + + if (fileInserts.length) { + originalFileInsertTriples = fileInserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`) + await transformStatements(fetch, originalFileInsertTriples).then( + transformedFileInsertTriples => { + // console.log(transformedFileInsertTriples); + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedFileInsertTriples, + { 'mu-call-scope-id': MU_CALL_SCOPE_ID_INITIAL_SYNC }, + endpoint, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION + ) + } + ) + } +} + +module.exports = { + dispatch +}; \ No newline at end of file diff --git a/config/loket-consumer/mandate-dispatching/util.js b/config/loket-consumer/mandate-dispatching/util.js new file mode 100644 index 0000000..496cda5 --- /dev/null +++ b/config/loket-consumer/mandate-dispatching/util.js @@ -0,0 +1,137 @@ +async function batchedDbUpdate(muUpdate, + graph, + triples, + extraHeaders, + endpoint, + batchSize, + maxAttempts, + sleepBetweenBatches = 1000, + sleepTimeOnFail = 1000, + operation = 'INSERT' +) { + + for (let i = 0; i < triples.length; i += batchSize) { + console.log(`Inserting triples in batch: ${i}-${i + batchSize}`); + + const batch = triples.slice(i, i + batchSize).join('\n'); + + const insertCall = async () => { + await muUpdate(` +${operation} DATA { +GRAPH <${graph}> { +${batch} +} +} +`, extraHeaders, endpoint); + }; + + await dbOperationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); + + console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); + await new Promise(r => setTimeout(r, sleepBetweenBatches)); + } +} + +async function dbOperationWithRetry(callback, + attempt, + maxAttempts, + sleepTimeOnFail) { + try { + return await callback(); + } + catch (e) { + console.log(`Operation failed for ${callback.toString()}, attempt: ${attempt} of ${maxAttempts}`); + console.log(`Error: ${e}`); + console.log(`Sleeping ${sleepTimeOnFail} ms`); + + if (attempt >= maxAttempts) { + console.log(`Max attempts reached for ${callback.toString()}, giving up`); + throw e; + } + + await new Promise(r => setTimeout(r, sleepTimeOnFail)); + return dbOperationWithRetry(callback, ++attempt, maxAttempts, sleepTimeOnFail); + } +} + +/** + * Splits an array into two parts, a part that passes and a part that fails a predicate function. + * Credits: https://github.com/benjay10 + * @public + * @function partition + * @param {Array} arr - Array to be partitioned + * @param {Function} fn - Function that accepts single argument: an element of the array, and should return a truthy or falsy value. + * @returns {Object} Object that contains keys passes and fails, each representing an array with elemets that pass or fail the predicate respectively + */ +function partition(arr, fn) { + let passes = [], fails = []; + arr.forEach((item) => (fn(item) ? passes : fails).push(item)); + return { passes, fails }; +} + + +/** + * Send triples to reasoning service for conversion + * + */ +function transformGraph(fetch, triples) { + return preProcess(fetch, triples) + .then(preprocessed => mainConversion(fetch, preprocessed)); +} + + +function preProcess(fetch, triples) { + let formdata = new URLSearchParams(); + formdata.append("data", triples); + + let requestOptions = { + method: 'POST', + body: formdata, + 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)); +} + +function mainConversion(fetch, triples) { + let formdata = new URLSearchParams(); + formdata.append("data", triples); + + let requestOptions = { + method: 'POST', + body: formdata, + redirect: 'follow' + }; + + 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)); +} + +function transformStatements(fetch, triples) { + return transformGraph(fetch, triples.join('\n')).then( + graph => { + statements = graph.replace(/\n{2,}/g, '').split('\n') + console.log(`CONVERSION: FROM ${triples.length} triples to ${statements.length}`) + return statements + } + ) +} + +module.exports = { + batchedDbUpdate, + partition, + transformStatements +}; \ No newline at end of file diff --git a/config/reasoner/dl2op/main/mapping.n3 b/config/reasoner/dl2op/main/mapping.n3 new file mode 100644 index 0000000..80a9c1a --- /dev/null +++ b/config/reasoner/dl2op/main/mapping.n3 @@ -0,0 +1,82 @@ +PREFIX : <#> +PREFIX besluit: +PREFIX e: +PREFIX ex: +PREFIX foaf: +PREFIX func: +PREFIX log: +PREFIX mandaat: +PREFIX org: +PREFIX persoon: +PREFIX string: +PREFIX xsd: + + +# First name +# persoon:gebruikteVoornaam => foaf:givenName (done in pre-processing) +# Alternatieve voornaam +# foaf:name => persoon:gebruikteVoornaam + +# Test Mock LLB +# foaf:name "Didier". +{ + { + ?s foaf:name ?o + } e:becomes { + ?s persoon:gebruikteVoornaam ?o + } +} => { + ?s persoon:gebruikteVoornaam ?o +}. + +# Test +# ex:s "2019-01-01T19:32:52+00:00"^^xsd:dateTime. +{ + ?s mandaat:start ?startDateTime. + # use backward arrow rule defined below + ?startDateTime :dateFromDateTime ?startDate. + { + ?s mandaat:start ?startDateTime + } e:becomes { + ?s mandaat:start ?startDate + }. +} => { + ?s mandaat:start ?startDate +}. + +# ex:s "2019-02-01T10:32:52+00:00"^^xsd:dateTime. +{ + ?s mandaat:einde ?eindeDateTime. + # use backward arrow rule defined below + ?eindeDateTime :dateFromDateTime ?eindeDate. + { + ?s mandaat:einde ?eindeDateTime + } e:becomes { + ?s mandaat:einde ?eindeDate + }. +} => { + ?s mandaat:einde ?eindeDate +}. + + +# Backward arrow rule is like a function call. +# Removes time and timezone info in the returned date +{ + ?dateTime :dateFromDateTime ?date +} <= { + (?dateTimeLiteral xsd:dateTime) log:dtlit ?dateTime. + (?dateTimeLiteral 0 11) func:substring ?dateLiteral. + (?dateLiteral xsd:date) log:dtlit ?date. +}. + + +# ex:orgA besluit:classificatie . +{ + { + ?s besluit:classificatie ?o + } e:becomes { + ?s org:classification ?o + } +} => { + ?s org:classification ?o +}. \ No newline at end of file diff --git a/config/reasoner/dl2op/preprocess/mapping.n3 b/config/reasoner/dl2op/preprocess/mapping.n3 new file mode 100644 index 0000000..0c6ae69 --- /dev/null +++ b/config/reasoner/dl2op/preprocess/mapping.n3 @@ -0,0 +1,31 @@ +PREFIX e: +PREFIX foaf: +PREFIX persoon: + + +# DL and OP both use persoon:gebruikteVoornaam but with different semantics (!) +# Hence the mapping is in two steps. +# The first step maps the DL persoon:gebruikteVoornaam to the OP counterpart +# and drops the DL persoon:gebruikteVoornaam statements with e:becomes +# A second reasoning will map DL foaf:name to OP persoon:gebruikteVoornaam +# Doing both at once would lead all DL foaf:names to incorrectly become foaf:givennames +# in OP + +# First name +# persoon:gebruikteVoornaam => foaf:givenName +# Alternatieve voornaam +# foaf:name => persoon:gebruikteVoornaam + +# Test Mock DL +# persoon:gebruikteVoornaam "Didier". + +{ + { + ?s persoon:gebruikteVoornaam ?o + } e:becomes { + ?s foaf:givenName ?o + } +} => { + ?s foaf:givenName ?o +}. + diff --git a/docker-compose.yml b/docker-compose.yml index 854f751..f6a11ad 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,7 +1,6 @@ -version: '3.4' +version: "3.4" -x-logging: - &default-logging +x-logging: &default-logging driver: "json-file" options: max-size: "10m" @@ -11,7 +10,7 @@ services: identifier: image: semtech/mu-identifier:1.9.1 environment: - DEFAULT_MU_AUTH_ALLOWED_GROUPS_HEADER: "[{\"variables\":[],\"name\":\"public\"}]" + DEFAULT_MU_AUTH_ALLOWED_GROUPS_HEADER: '[{"variables":[],"name":"public"}]' SESSION_COOKIE_SECURE: "on" labels: - "logging=true" @@ -201,34 +200,38 @@ services: - db:database restart: always logging: *default-logging -################################################################################ -# DELTA GENERAL -################################################################################ + reasoner: + image: bdevloed/mu-reasoner + volumes: + - ./config/reasoner:/config + ################################################################################ + # DELTA GENERAL + ################################################################################ delta-producer-dump-file-publisher: image: bdevloed/delta-producer-dump-file-publisher:0.8.0-beta volumes: - ./config/delta-producer/dump-file-publisher:/config - ./data/files:/share environment: - FILES_GRAPH: 'http://mu.semte.ch/graphs/public' # Note: this will be extracted in a config per job one day - DCAT_DATASET_GRAPH: 'http://mu.semte.ch/graphs/public' + FILES_GRAPH: "http://mu.semte.ch/graphs/public" # Note: this will be extracted in a config per job one day + DCAT_DATASET_GRAPH: "http://mu.semte.ch/graphs/public" links: - db:database restart: always logging: *default-logging -################################################################################ -# DELTA ADMINISTRATIVE-UNITS: START -################################################################################ + ################################################################################ + # DELTA ADMINISTRATIVE-UNITS: START + ################################################################################ delta-producer-bg-jobs-initiator-administrative-units: image: lblod/delta-producer-background-jobs-initiator:0.3.1 environment: JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" - DUMP_FILE_CREATION_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/administrative-units' - INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units' - HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units' - CRON_PATTERN_DUMP_JOB: '0 0 4 * * 6' # weekly on saturday - CRON_PATTERN_HEALING_JOB: '0 0 4 * * *' # everyday at 4 AM - START_INITIAL_SYNC: 'false' # prefer to disable it now. We want to have feeling with how the DB acts. + DUMP_FILE_CREATION_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/administrative-units" + INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units" + HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units" + CRON_PATTERN_DUMP_JOB: "0 0 4 * * 6" # weekly on saturday + CRON_PATTERN_HEALING_JOB: "0 0 4 * * *" # everyday at 4 AM + START_INITIAL_SYNC: "false" # prefer to disable it now. We want to have feeling with how the DB acts. labels: - "logging=true" links: @@ -241,12 +244,12 @@ services: MAX_BODY_SIZE: "50mb" JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" RELATIVE_FILE_PATH: "deltas/administrative-units" - PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/administrative-units' - HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units' - INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units' + PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/administrative-units" + HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/administrative-units" + INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/administrative-units" QUEUE_POLL_INTERVAL: 3000 USE_VIRTUOSO_FOR_EXPENSIVE_SELECTS: "true" # some queries are just too expensive - SKIP_MU_AUTH_INITIAL_SYNC: 'true' + SKIP_MU_AUTH_INITIAL_SYNC: "true" HEALING_PATCH_GRAPH_BATCH_SIZE: 500 volumes: - ./config/delta-producer/administrative-units:/config @@ -263,7 +266,7 @@ services: ERROR_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" RELATIVE_FILE_PATH: "deltas/administrative-units" PUBLISHER_URI: "http://data.lblod.info/services/delta-production-json-diff-file-manager/administrative-units" - PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/administrative-units' + PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/administrative-units" DELTA_INTERVAL_MS: 10000 PRETTY_PRINT_DIFF_JSON: "true" volumes: @@ -274,22 +277,22 @@ services: - db:database restart: always logging: *default-logging -################################################################################ -# DELTA ADMINISTRATIVE-UNITS: END -################################################################################ -################################################################################ -# DELTA ORGANIZATIONS: START -################################################################################ + ################################################################################ + # DELTA ADMINISTRATIVE-UNITS: END + ################################################################################ + ################################################################################ + # DELTA ORGANIZATIONS: START + ################################################################################ delta-producer-bg-jobs-initiator-organizations: image: lblod/delta-producer-background-jobs-initiator:0.3.1 environment: JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" - DUMP_FILE_CREATION_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/organizations' - INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations' - HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations' - CRON_PATTERN_DUMP_JOB: '42 * * * *' # At minute 42 every hour - CRON_PATTERN_HEALING_JOB: '0 0 4 * * *' # everyday at 4 AM - START_INITIAL_SYNC: 'false' # prefer to disable it now. We want to have feeling with how the DB acts. + DUMP_FILE_CREATION_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/organizations" + INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations" + HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations" + CRON_PATTERN_DUMP_JOB: "42 * * * *" # At minute 42 every hour + CRON_PATTERN_HEALING_JOB: "0 0 4 * * *" # everyday at 4 AM + START_INITIAL_SYNC: "false" # prefer to disable it now. We want to have feeling with how the DB acts. labels: - "logging=true" links: @@ -302,12 +305,12 @@ services: MAX_BODY_SIZE: "50mb" JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" RELATIVE_FILE_PATH: "deltas/organizations" - PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/organizations' - HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations' - INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations' + PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/organizations" + HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/organizations" + INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/organizations" QUEUE_POLL_INTERVAL: 3000 USE_VIRTUOSO_FOR_EXPENSIVE_SELECTS: "true" # some queries are just too expensive - SKIP_MU_AUTH_INITIAL_SYNC: 'true' + SKIP_MU_AUTH_INITIAL_SYNC: "true" HEALING_PATCH_GRAPH_BATCH_SIZE: 500 volumes: - ./config/delta-producer/organizations:/config @@ -324,7 +327,7 @@ services: ERROR_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" RELATIVE_FILE_PATH: "deltas/organizations" PUBLISHER_URI: "http://data.lblod.info/services/delta-production-json-diff-file-manager/organizations" - PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/organizations' + PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/organizations" DELTA_INTERVAL_MS: 10000 PRETTY_PRINT_DIFF_JSON: "true" volumes: @@ -335,24 +338,23 @@ services: - db:database restart: always logging: *default-logging -################################################################################ -# DELTA ORGANIZATIONS: END -################################################################################ + ################################################################################ + # DELTA ORGANIZATIONS: END + ################################################################################ - -################################################################################ -# DELTA PUBLIC INFORMATION: START -################################################################################ + ################################################################################ + # DELTA PUBLIC INFORMATION: START + ################################################################################ delta-producer-bg-jobs-initiator-public: image: lblod/delta-producer-background-jobs-initiator:0.3.1 environment: JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" - DUMP_FILE_CREATION_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/public' - INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public' - HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public' - CRON_PATTERN_DUMP_JOB: '11 */4 * * *' # At minute 11 every 4 hours - CRON_PATTERN_HEALING_JOB: '0 0 4 * * *' # everyday at 4 AM - START_INITIAL_SYNC: 'false' # prefer to disable it now. We want to have feeling with how the DB acts. + DUMP_FILE_CREATION_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/deltaDumpFileCreation/public" + INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public" + HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public" + CRON_PATTERN_DUMP_JOB: "11 */4 * * *" # At minute 11 every 4 hours + CRON_PATTERN_HEALING_JOB: "0 0 4 * * *" # everyday at 4 AM + START_INITIAL_SYNC: "false" # prefer to disable it now. We want to have feeling with how the DB acts. labels: - "logging=true" links: @@ -365,12 +367,12 @@ services: MAX_BODY_SIZE: "50mb" JOBS_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" RELATIVE_FILE_PATH: "deltas/public" - PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/public' - HEALING_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public' - INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: 'http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public' + PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/public" + HEALING_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/healingOperation/public" + INITIAL_PUBLICATION_GRAPH_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/initialPublicationGraphSyncing/public" QUEUE_POLL_INTERVAL: 3000 USE_VIRTUOSO_FOR_EXPENSIVE_SELECTS: "true" # some queries are just too expensive - SKIP_MU_AUTH_INITIAL_SYNC: 'true' + SKIP_MU_AUTH_INITIAL_SYNC: "true" HEALING_PATCH_GRAPH_BATCH_SIZE: 500 volumes: - ./config/delta-producer/public:/config @@ -387,7 +389,7 @@ services: ERROR_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" RELATIVE_FILE_PATH: "deltas/public" PUBLISHER_URI: "http://data.lblod.info/services/delta-production-json-diff-file-manager/public" - PUBLICATION_GRAPH: 'http://redpencil.data.gift/id/deltas/producer/public' + PUBLICATION_GRAPH: "http://redpencil.data.gift/id/deltas/producer/public" DELTA_INTERVAL_MS: 10000 PRETTY_PRINT_DIFF_JSON: "true" volumes: @@ -398,14 +400,44 @@ services: - db:database restart: always logging: *default-logging -################################################################################ -# DELTA PUBLIC INFORMATION: END -################################################################################ + ################################################################################ + # DELTA PUBLIC INFORMATION: END + ################################################################################ + ################################################################################ + # DELTA CONSUMERS + ################################################################################ + mandatendatabank-consumer: + image: bdevloed/delta-consumer:fetch + environment: + DCR_SERVICE_NAME: "mandatendatabank-consumer" + DCR_SYNC_BASE_URL: "https://loket.lblod.info" # replace with link to Loket API + DCR_SYNC_DATASET_SUBJECT: "http://data.lblod.info/datasets/delta-producer/dumps/MandatarissenCacheGraphDump" + DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/mandatarissen" + DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/mandatarissen" + DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/mandatendatabank-consumer" + # DISABLE_INITIAL_SYNC: "true" + BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" + DCR_SYNC_FILES_PATH: "/sync/mandatarissen/files" + FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" + KEEP_DELTA_FILES: "true" + volumes: + - ./config/loket-consumer/mandate-dispatching:/config/triples-dispatching/custom-dispatching + restart: always + labels: + - "logging=true" + links: + - db:database + - triplestore:virtuoso + logging: *default-logging -################################################################################ -# MU SEARCH - TODO USE THE CORRECT IMAGE ONCE INHERITANCE IS IN MU SEARCH -################################################################################ + ################################################################################ + # DELTA CONSUMERS: END + ################################################################################ + + ################################################################################ + # MU SEARCH - TODO USE THE CORRECT IMAGE ONCE INHERITANCE IS IN MU SEARCH + ################################################################################ search: image: nbittich/mu-search:wip-inheritance-alpha2 # Bump to fixed version when merged and release to the original repo links: @@ -418,7 +450,7 @@ services: volumes: - ./data/elasticsearch/:/usr/share/elasticsearch/data environment: - - discovery.type=single-node + - discovery.type=single-node restart: always error-report-service: From d2f49fef612ae78b59ffd6f047a0d80c7619236c Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Tue, 29 Nov 2022 17:28:59 +0000 Subject: [PATCH 2/9] specify ingest graph --- docker-compose.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/docker-compose.yml b/docker-compose.yml index f6a11ad..df09169 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -416,6 +416,7 @@ services: DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/mandatarissen" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/mandatarissen" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/mandatendatabank-consumer" + INGEST_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" # DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/mandatarissen/files" From e1fdde70102d84a6291d8935dde399df68dadc9a Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Wed, 30 Nov 2022 14:56:36 +0000 Subject: [PATCH 3/9] add leidinggevenden consumer --- .../leidinggevenden-dispatching/config.js | 25 ++++ .../delta-sync-dispatching.js | 126 ++++++++++++++++ .../initial-sync-dispatching.js | 84 +++++++++++ .../leidinggevenden-dispatching/util.js | 137 ++++++++++++++++++ docker-compose.yml | 33 ++++- 5 files changed, 404 insertions(+), 1 deletion(-) create mode 100644 config/loket-consumer/leidinggevenden-dispatching/config.js create mode 100644 config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js create mode 100644 config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js create mode 100644 config/loket-consumer/leidinggevenden-dispatching/util.js diff --git a/config/loket-consumer/leidinggevenden-dispatching/config.js b/config/loket-consumer/leidinggevenden-dispatching/config.js new file mode 100644 index 0000000..0c97ddd --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/config.js @@ -0,0 +1,25 @@ +const BATCH_SIZE = parseInt(process.env.BATCH_SIZE) || 100; +const MU_CALL_SCOPE_ID_INITIAL_SYNC = process.env.MU_CALL_SCOPE_ID_INITIAL_SYNC || 'http://redpencil.data.gift/id/concept/muScope/deltas/consumer/initialSync'; +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 SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); +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/public`; + + +if(!process.env.FILE_SYNC_GRAPH) + throw `Expected 'FILE_SYNC_GRAPH' to be provided.`; +const FILE_SYNC_GRAPH = process.env.FILE_SYNC_GRAPH; + +module.exports = { + BATCH_SIZE, + MU_CALL_SCOPE_ID_INITIAL_SYNC, + BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, + DIRECT_DATABASE_ENDPOINT, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + INGEST_GRAPH, + FILE_SYNC_GRAPH +}; \ No newline at end of file diff --git a/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js new file mode 100644 index 0000000..9f936df --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/delta-sync-dispatching.js @@ -0,0 +1,126 @@ +const { transformStatements, batchedDbUpdate, partition } = require('./util'); +const { BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, + DIRECT_DATABASE_ENDPOINT, + MU_CALL_SCOPE_ID_INITIAL_SYNC, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + INGEST_GRAPH, + FILE_SYNC_GRAPH +} = require('./config'); + +/** +* Dispatch the fetched information to a target graph. +* Note: will be ADDED to it's own graph. +* We take only care of adding them, not updating triples, this is a TODO +* @param { mu, muAuthSudo } lib - The provided libraries from the host service. +* @param { termObjectChangeSets: { deletes, inserts } } data - The fetched changes sets, which objects of serialized Terms +* [ { +* graph: "", +* subject: "", +* predicate: "", +* object: "^^" +* } +* ] +* @return {void} Nothing +*/ +async function dispatch(lib, data) { + const { mu, muAuthSudo, fetch } = lib; + const { termObjectChangeSets } = data; + + for (let { deletes, inserts } of termObjectChangeSets) { + const deletePartitions = partition(deletes, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`); + const deleteStatements = regularDeletes.map(o => `${o.subject} ${o.predicate} ${o.object}.`); + + if (deleteStatements.length) { + await transformStatements(fetch, deleteStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedStatements, + { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, + process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "DELETE" + ); + } + ) + } + + if (fileDeleteStatements.length) { + await transformStatements(fetch, fileDeleteStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedStatements, + { 'mu-call-scope-id': 'http://redpencil.data.gift/id/concept/muScope/deltas/write-for-dispatch' }, + process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "DELETE" + ); + } + ) + } + + const insertPartitions = partition(inserts, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`); + const fileInsertStatements = fileInserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`); + + + if (regularInsertStatements.length) { + await transformStatements(fetch, regularInsertStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedStatements, + {}, + process.env.MU_SPARQL_ENDPOINT, //Note: this is the default endpoint through auth + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "INSERT" + ); + } + ) + } + + if (fileInsertStatements.length) { + await transformStatements(fetch, fileInsertStatements).then( + transformedStatements => { + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedStatements, + {}, + process.env.MU_SPARQL_ENDPOINT, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + "INSERT" + ); + } + ) + } + } +} + +module.exports = { + dispatch +}; \ No newline at end of file diff --git a/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js b/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js new file mode 100644 index 0000000..69bcdb4 --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/initial-sync-dispatching.js @@ -0,0 +1,84 @@ +const { transformStatements, batchedDbUpdate, partition } = require('./util'); +const { BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES, + DIRECT_DATABASE_ENDPOINT, + MU_CALL_SCOPE_ID_INITIAL_SYNC, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION, + INGEST_GRAPH, + FILE_SYNC_GRAPH +} = require('./config'); +const endpoint = BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES ? DIRECT_DATABASE_ENDPOINT : process.env.MU_SPARQL_ENDPOINT; + +/** +* Dispatch the fetched information to a target graph. +* @param { mu, muAuthSudo } lib - The provided libraries from the host service. +* @param { termObjects } data - The fetched quad information, which objects of serialized Terms +* [ { +* graph: "", +* subject: "", +* predicate: "", +* object: "^^" +* } +* ] +* @return {void} Nothing +*/ +async function dispatch(lib, data) { + const { mu, muAuthSudo, fetch } = lib; + const { termObjects } = data; + + const partitions = partition(termObjects, o => o.subject.startsWith(' `${o.subject} ${o.predicate} ${o.object}.`) + + await transformStatements(fetch, originalRegularInsertTriples).then( + transformedRegularInsertTriples => { + // console.log(transformedRegularInsertTriples); + batchedDbUpdate( + muAuthSudo.updateSudo, + INGEST_GRAPH, + transformedRegularInsertTriples, + { 'mu-call-scope-id': MU_CALL_SCOPE_ID_INITIAL_SYNC }, + endpoint, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION + ) + } + ) + } + + if (fileInserts.length) { + originalFileInsertTriples = fileInserts.map(o => `${o.subject} ${o.predicate} ${o.object}.`) + await transformStatements(fetch, originalFileInsertTriples).then( + transformedFileInsertTriples => { + // console.log(transformedFileInsertTriples); + batchedDbUpdate( + muAuthSudo.updateSudo, + FILE_SYNC_GRAPH, + transformedFileInsertTriples, + { 'mu-call-scope-id': MU_CALL_SCOPE_ID_INITIAL_SYNC }, + endpoint, + BATCH_SIZE, + MAX_DB_RETRY_ATTEMPTS, + SLEEP_BETWEEN_BATCHES, + SLEEP_TIME_AFTER_FAILED_DB_OPERATION + ) + } + ) + } +} + +module.exports = { + dispatch +}; \ No newline at end of file diff --git a/config/loket-consumer/leidinggevenden-dispatching/util.js b/config/loket-consumer/leidinggevenden-dispatching/util.js new file mode 100644 index 0000000..496cda5 --- /dev/null +++ b/config/loket-consumer/leidinggevenden-dispatching/util.js @@ -0,0 +1,137 @@ +async function batchedDbUpdate(muUpdate, + graph, + triples, + extraHeaders, + endpoint, + batchSize, + maxAttempts, + sleepBetweenBatches = 1000, + sleepTimeOnFail = 1000, + operation = 'INSERT' +) { + + for (let i = 0; i < triples.length; i += batchSize) { + console.log(`Inserting triples in batch: ${i}-${i + batchSize}`); + + const batch = triples.slice(i, i + batchSize).join('\n'); + + const insertCall = async () => { + await muUpdate(` +${operation} DATA { +GRAPH <${graph}> { +${batch} +} +} +`, extraHeaders, endpoint); + }; + + await dbOperationWithRetry(insertCall, 0, maxAttempts, sleepTimeOnFail); + + console.log(`Sleeping before next query execution: ${sleepBetweenBatches}`); + await new Promise(r => setTimeout(r, sleepBetweenBatches)); + } +} + +async function dbOperationWithRetry(callback, + attempt, + maxAttempts, + sleepTimeOnFail) { + try { + return await callback(); + } + catch (e) { + console.log(`Operation failed for ${callback.toString()}, attempt: ${attempt} of ${maxAttempts}`); + console.log(`Error: ${e}`); + console.log(`Sleeping ${sleepTimeOnFail} ms`); + + if (attempt >= maxAttempts) { + console.log(`Max attempts reached for ${callback.toString()}, giving up`); + throw e; + } + + await new Promise(r => setTimeout(r, sleepTimeOnFail)); + return dbOperationWithRetry(callback, ++attempt, maxAttempts, sleepTimeOnFail); + } +} + +/** + * Splits an array into two parts, a part that passes and a part that fails a predicate function. + * Credits: https://github.com/benjay10 + * @public + * @function partition + * @param {Array} arr - Array to be partitioned + * @param {Function} fn - Function that accepts single argument: an element of the array, and should return a truthy or falsy value. + * @returns {Object} Object that contains keys passes and fails, each representing an array with elemets that pass or fail the predicate respectively + */ +function partition(arr, fn) { + let passes = [], fails = []; + arr.forEach((item) => (fn(item) ? passes : fails).push(item)); + return { passes, fails }; +} + + +/** + * Send triples to reasoning service for conversion + * + */ +function transformGraph(fetch, triples) { + return preProcess(fetch, triples) + .then(preprocessed => mainConversion(fetch, preprocessed)); +} + + +function preProcess(fetch, triples) { + let formdata = new URLSearchParams(); + formdata.append("data", triples); + + let requestOptions = { + method: 'POST', + body: formdata, + 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)); +} + +function mainConversion(fetch, triples) { + let formdata = new URLSearchParams(); + formdata.append("data", triples); + + let requestOptions = { + method: 'POST', + body: formdata, + redirect: 'follow' + }; + + 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)); +} + +function transformStatements(fetch, triples) { + return transformGraph(fetch, triples.join('\n')).then( + graph => { + statements = graph.replace(/\n{2,}/g, '').split('\n') + console.log(`CONVERSION: FROM ${triples.length} triples to ${statements.length}`) + return statements + } + ) +} + +module.exports = { + batchedDbUpdate, + partition, + transformStatements +}; \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index df09169..911ec3f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -407,7 +407,10 @@ services: ################################################################################ # DELTA CONSUMERS ################################################################################ - mandatendatabank-consumer: + ################################################################################ + # LOKET MANDATEN CONSUMERS + ################################################################################ + mandaten-consumer: image: bdevloed/delta-consumer:fetch environment: DCR_SERVICE_NAME: "mandatendatabank-consumer" @@ -432,6 +435,34 @@ services: - triplestore:virtuoso logging: *default-logging + ################################################################################ + # LOKET LEIDINGGEVENDEN CONSUMERS + ################################################################################ + leidinggevenden-consumer: + image: bdevloed/delta-consumer:fetch + environment: + DCR_SERVICE_NAME: "leidinggevenden-consumer" + DCR_SYNC_BASE_URL: "https://loket.lblod.info" # replace with link to Loket API + DCR_SYNC_DATASET_SUBJECT: "http://data.lblod.info/datasets/delta-producer/dumps/LeidinggevendenCacheGraphDump" + DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/leidinggevenden" + DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/leidinggevenden" + DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/leidinggevenden-consumer" + INGEST_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" + # DISABLE_INITIAL_SYNC: "true" + BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" + DCR_SYNC_FILES_PATH: "/sync/leidinggevenden/files" + FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" + KEEP_DELTA_FILES: "true" + volumes: + - ./config/loket-consumer/leidinggevenden-dispatching:/config/triples-dispatching/custom-dispatching + restart: always + labels: + - "logging=true" + links: + - db:database + - triplestore:virtuoso + logging: *default-logging + ################################################################################ # DELTA CONSUMERS: END ################################################################################ From e37df1387675ba538fa7a72020aac6515241175e Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Thu, 1 Dec 2022 13:52:35 +0000 Subject: [PATCH 4/9] use smarter default ingest graph --- config/loket-consumer/leidinggevenden-dispatching/config.js | 4 ++-- config/loket-consumer/mandate-dispatching/config.js | 2 +- docker-compose.yml | 2 -- 3 files changed, 3 insertions(+), 5 deletions(-) diff --git a/config/loket-consumer/leidinggevenden-dispatching/config.js b/config/loket-consumer/leidinggevenden-dispatching/config.js index 0c97ddd..526d237 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/config.js +++ b/config/loket-consumer/leidinggevenden-dispatching/config.js @@ -5,10 +5,10 @@ const DIRECT_DATABASE_ENDPOINT = process.env.DIRECT_DATABASE_ENDPOINT || 'http:/ const MAX_DB_RETRY_ATTEMPTS = parseInt(process.env.MAX_DB_RETRY_ATTEMPTS || 5); const SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); 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/public`; +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; diff --git a/config/loket-consumer/mandate-dispatching/config.js b/config/loket-consumer/mandate-dispatching/config.js index 0c97ddd..d3342b3 100644 --- a/config/loket-consumer/mandate-dispatching/config.js +++ b/config/loket-consumer/mandate-dispatching/config.js @@ -5,7 +5,7 @@ const DIRECT_DATABASE_ENDPOINT = process.env.DIRECT_DATABASE_ENDPOINT || 'http:/ const MAX_DB_RETRY_ATTEMPTS = parseInt(process.env.MAX_DB_RETRY_ATTEMPTS || 5); const SLEEP_BETWEEN_BATCHES = parseInt(process.env.SLEEP_BETWEEN_BATCHES || 1000); 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/public`; +const INGEST_GRAPH = process.env.INGEST_GRAPH || `http://mu.semte.ch/graphs/organisatieportaal`; if(!process.env.FILE_SYNC_GRAPH) diff --git a/docker-compose.yml b/docker-compose.yml index 911ec3f..2c8d764 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -419,7 +419,6 @@ services: DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/mandatarissen" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/mandatarissen" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/mandatendatabank-consumer" - INGEST_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" # DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/mandatarissen/files" @@ -447,7 +446,6 @@ services: DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/leidinggevenden" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/leidinggevenden" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/leidinggevenden-consumer" - INGEST_GRAPH: "http://mu.semte.ch/graphs/organisatieportaal" # DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/leidinggevenden/files" From 563817c5e614aca3a4cd5f0699dc71c34152442a Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Thu, 1 Dec 2022 14:28:25 +0000 Subject: [PATCH 5/9] renamed method to transformTriples --- config/loket-consumer/leidinggevenden-dispatching/util.js | 6 +++--- config/loket-consumer/mandate-dispatching/util.js | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/config/loket-consumer/leidinggevenden-dispatching/util.js b/config/loket-consumer/leidinggevenden-dispatching/util.js index 496cda5..db93194 100644 --- a/config/loket-consumer/leidinggevenden-dispatching/util.js +++ b/config/loket-consumer/leidinggevenden-dispatching/util.js @@ -74,9 +74,9 @@ function partition(arr, fn) { * Send triples to reasoning service for conversion * */ -function transformGraph(fetch, triples) { +function transformTriples(fetch, triples) { return preProcess(fetch, triples) - .then(preprocessed => mainConversion(fetch, preprocessed)); + .then(preprocessed => mainConversion(fetch, preprocessed)); } @@ -121,7 +121,7 @@ function mainConversion(fetch, triples) { } function transformStatements(fetch, triples) { - return transformGraph(fetch, triples.join('\n')).then( + return transformTriples(fetch, triples.join('\n')).then( graph => { statements = graph.replace(/\n{2,}/g, '').split('\n') console.log(`CONVERSION: FROM ${triples.length} triples to ${statements.length}`) diff --git a/config/loket-consumer/mandate-dispatching/util.js b/config/loket-consumer/mandate-dispatching/util.js index 496cda5..db93194 100644 --- a/config/loket-consumer/mandate-dispatching/util.js +++ b/config/loket-consumer/mandate-dispatching/util.js @@ -74,9 +74,9 @@ function partition(arr, fn) { * Send triples to reasoning service for conversion * */ -function transformGraph(fetch, triples) { +function transformTriples(fetch, triples) { return preProcess(fetch, triples) - .then(preprocessed => mainConversion(fetch, preprocessed)); + .then(preprocessed => mainConversion(fetch, preprocessed)); } @@ -121,7 +121,7 @@ function mainConversion(fetch, triples) { } function transformStatements(fetch, triples) { - return transformGraph(fetch, triples.join('\n')).then( + return transformTriples(fetch, triples.join('\n')).then( graph => { statements = graph.replace(/\n{2,}/g, '').split('\n') console.log(`CONVERSION: FROM ${triples.length} triples to ${statements.length}`) From 368513c0f3178fcc947aa4a86d4a94d2374fd294 Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Fri, 2 Dec 2022 13:33:03 +0000 Subject: [PATCH 6/9] disable initial sync - use override to start it --- docker-compose.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 2c8d764..fbbe220 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -419,7 +419,7 @@ services: DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/mandatarissen" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/mandatarissen" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/mandatendatabank-consumer" - # DISABLE_INITIAL_SYNC: "true" + DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/mandatarissen/files" FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" @@ -446,7 +446,7 @@ services: DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/leidinggevenden" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/leidinggevenden" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/leidinggevenden-consumer" - # DISABLE_INITIAL_SYNC: "true" + DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/leidinggevenden/files" FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" From 6591a5eaad0e12a22d8c9f31651c27c775802738 Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Fri, 2 Dec 2022 13:48:30 +0000 Subject: [PATCH 7/9] Add DRC_ prefix to disable initial sync --- docker-compose.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index fbbe220..d82b25c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -419,7 +419,7 @@ services: DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/mandatarissen" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/mandatarissen" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/mandatendatabank-consumer" - DISABLE_INITIAL_SYNC: "true" + DCR_DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/mandatarissen/files" FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" @@ -446,7 +446,7 @@ services: DCR_INITIAL_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/initialSync/leidinggevenden" DCR_DELTA_SYNC_JOB_OPERATION: "http://redpencil.data.gift/id/jobs/concept/JobOperation/deltas/consumer/deltaSync/leidinggevenden" DCR_JOB_CREATOR_URI: "http://data.lblod.info/services/id/leidinggevenden-consumer" - DISABLE_INITIAL_SYNC: "true" + DCR_DISABLE_INITIAL_SYNC: "true" BYPASS_MU_AUTH_FOR_EXPENSIVE_QUERIES: "true" DCR_SYNC_FILES_PATH: "/sync/leidinggevenden/files" FILE_SYNC_GRAPH: "http://mu.semte.ch/graphs/temp/original-physical-files-data" From 6516b3cb28ace11379b336b6374e415fe6558fe4 Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Mon, 5 Dec 2022 13:21:38 +0000 Subject: [PATCH 8/9] clarify comment --- config/reasoner/dl2op/main/mapping.n3 | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/config/reasoner/dl2op/main/mapping.n3 b/config/reasoner/dl2op/main/mapping.n3 index 80a9c1a..bf94c92 100644 --- a/config/reasoner/dl2op/main/mapping.n3 +++ b/config/reasoner/dl2op/main/mapping.n3 @@ -60,12 +60,22 @@ PREFIX xsd: # Backward arrow rule is like a function call. -# Removes time and timezone info in the returned date +# Removes time info in the returned date +# The user in DL selects a date via the UI. +# These dates are persisted as midnight dateTimes in UTC. +# Meaning that 1st of July is stored as +# "2022-06-30T22:00:00Z"^^xsd:dateTime +# and 1 December +# "2022-11-30T23:00:00Z"^^xsd:dateTime +# So, the following rule adds 2 hours to the datetime, and then takes the date part of the UTC date time +# to come to the date that the DL user selected (in CE(S)T) { ?dateTime :dateFromDateTime ?date } <= { (?dateTimeLiteral xsd:dateTime) log:dtlit ?dateTime. - (?dateTimeLiteral 0 11) func:substring ?dateLiteral. + (?dateTime "PT2H"^^xsd:dayTimeDuration) func:add-dayTimeDuration-to-dateTime ?localDateTime. + (?localDateTimeLiteral xsd:dateTime) log:dtlit ?localDateTime. + (?localDateTimeLiteral 0 11) func:substring ?dateLiteral. (?dateLiteral xsd:date) log:dtlit ?date. }. From f6e7f9257397a1cd35aeb7afd27fdfc255705c64 Mon Sep 17 00:00:00 2001 From: Boris De Vloed Date: Tue, 13 Dec 2022 09:36:34 +0000 Subject: [PATCH 9/9] 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) {